labix.org/mgo 连接池泄漏问题

问题复现

labix.org/mgo是golang常用的mongo driver,笔者的项目中重度依赖,不过项目年久失修,已经不维护。所以结论是用官方包

最近在使用中,我发现了一个问题,服务对mongo的长连接一直缓慢增长,形似mongo连接泄漏。

查看了下mgo源码,发现mgo内部维护了连接池,而默认连接池大小socketsPerServer是4096,连接池的连接新增是惰性的,不会在初始化时建立所有连接,而是在有新请求且当前无剩余可用连接时,建立新连接,使用结束后就放入连接池中供以后使用,期待连接池所维护的最多连接数是4096个。连接池大小限制代码如下,请注意mongoServer.unusedSockets(空闲的连接)和mongoServer.liveSockets(生效过的连接)的含义:

// 连接池大小:
var socketsPerServer = 4096
// 连接池大小设置:
func SetPoolLimitPerServer(limit int) {
	socketsPerServer = limit
}
// 连接池管理结构
type mongoServer struct {
  ...
	unusedSockets []*mongoSocket // 空闲的连接
	liveSockets   []*mongoSocket // 生效过的连接,这里记录所有建立的连接,当连接使用结束会复制一份进unusedSockets,但是不会从liveSockets删除
	... 
}

如上代码,socketsPerServer是可以通过方法SetPoolLimitPerServer改写的,于是我在下面的代码中,OpenDB里调用SetPoolLimitPerServer将连接池上限改成了20,但是通过并发测试发现,连接池的限制并不准确,实际运行下来最后建立的连接依然会超过20。

主要代码如下,我插入200条数据,为了防止数据库压力,通过笔者写的goworker以稳定的并发90写入,其他会进行排队。

最后通过mgo提供的GetStats()接口(或者在服务端通过db.serverStatus()也能看到连接数量connections.current),能够看到池里的连接为SocketsAlive 59个,远远超过了20,这是为什么呢?

// 测试mgo连接泄露问题
func main() {
	db := OpenDB(xl, fmgo.Config{Config: mgo3.Config{Host: "127.0.0.1:27017", DB: "test_black", Coll: "connectiontest", Mode: "strong", SyncTimeoutInS: 5}})
	defer db.Close()
	mgo.SetStats(true)
	// 初始化 goworker,并发不超过90
	worker := goworker.New(goworker.WorkerConfig{
		ConcurrencyNum: 90,
	})
	// 总共插入200条数据
	for i := 0; i < 200; i++ {
		var j = i
		worker.Add(func() {
			db.Insert(mgoDBInfo{Name: strconv.Itoa(j) + "_name"})
		})
	}
	worker.IsDone()
	stats := mgo.GetStats()
	fmt.Printf("%+v", stats)
 // 输出:{Clusters:0 MasterConns:59 SlaveConns:0 SentOps:459 ReceivedOps:259 ReceivedDocs:259 SocketsAlive:59 SocketsInUse:0 SocketRefs:0}
	time.Sleep(100 * time.Second)
}

问题定位

问题出现在当有新的请求时,mgo 的连接池管理逻辑,当需要新建一个连接时,并发情况下会有问题,精简代码如下

func (server *mongoServer) AcquireSocket(limit int, timeout time.Duration) (socket *mongoSocket, abended bool, err error) {
 	for {
  	server.Lock()
    n := len(server.unusedSockets)
    // 判断当前正在使用的连接,是否到连接池上限,如果到了上限就退出等待
		if len(server.liveSockets)-n >= limit {
			server.Unlock()
			return nil, false, errSocketLimit
		}
   
    if n > 0{
      // 拿一个可用的 unusedSockets
    } else {
    	server.Unlock()
      // bug here
      // 这里 unlock->connect->lock,是因为如果不unlock,新建连接时间太长,导致阻塞所有并发AcquireSocket的请求
      // 这样做在并发时就会有一个bug,如果连接池大小是20,而同时并发的进入30个AcquireSocket,所有的请求都会走到下面的Connect(),并正常的拿到连接,加入liveSockets,返回成功,导致连接池里有30个连接
      socket, err = server.Connect()
      if err == nil {
				server.Lock()
				if server.closed {
          server.UnLock()
          return nil, abended, errServerClosed
        }
				server.liveSockets = append(server.liveSockets, socket)
				server.Unlock()
      }
    }
    return
  }
  panic("unreachable")
}

问题解决

从代码逻辑看,mgo对连接池的理解和一般的理解不同:

  • 一般而言,连接池大小,就代表有多少个长连接维持在池里,也就是 正在使用的连接+未使用的连接 <= 连接池大小
  • mgo的连接池管理里,从上面的方法里用if len(server.liveSockets)-n >= limit判断连接池是否已经满可以看出, mgo认为当前正在用的连接数 <= 连接池大小 就可以了,没有使用的连接不应该受连接池大小控制。(注意,第一部分已经强调,liveSockets是历史建立的所有连接数量,不是正在使用socket的数量)

很明显,mgo的定义是不对的,这样会导致真实建立的连接(正在使用的+池里未使用的)>连接池限制。

针对这一点,我们在拿到新连接并加锁后,判断一下当前建立的连接是否已经超限,超限就关闭当前连接并等待,解决这个问题。

这个解决方案,会放弃新建的连接,对资源是有一定的浪费的,因为毕竟新建连接是耗时的。但是一旦建立后,就导致了连接泄漏,所以是不得已而为之。

func (server *mongoServer) AcquireSocket(limit int, timeout time.Duration) (socket *mongoSocket, abended bool, err error) {
     socket, err = server.Connect()
      if err == nil {
				server.Lock()
				if server.closed {
          server.UnLock()
          return nil, abended, errServerClosed
        }
        // fix bug start
        // +1 是要算上当前新建的这个连接
        if limit > 0 && len(server.liveSockets)-n+1 > limit {
					server.Unlock()
					socket.Release()
					socket.Close()
					return nil, false, errSocketLimit
				}
        // fix bug end
				server.liveSockets = append(server.liveSockets, socket)
				server.Unlock()
      }
}

https://juejin.im/post/5ed4f995f265da76cd47ee45

「点点赞赏,手留余香」

    还没有人赞赏,快来当第一个赞赏的人吧!
0 条回复 A 作者 M 管理员
    所有的伟大,都源于一个勇敢的开始!
欢迎您,新朋友,感谢参与互动!欢迎您 {{author}},您在本站有{{commentsCount}}条评论