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

vlambda
2020-12-21

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

一

问题重现

labix.org/mgo(https://github.com/go-mgo/mgo)是golang常用的mongo driver，笔者的项目中重度依赖，不过项目年久失修，已经不维护。所以最终结论是用官方包（https://github.com/mongodb/mongo-go-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（https://github.com/jiangz222/gowork）以稳定的并发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() }}