为什么一条异常的连接会出现在 connect pool 中,而且总会拿到这条连接 ?

在本系列第篇文章末尾提出了这样一个问题 『 为什么一条异常的连接会出现在 connect pool 中,而且总会拿到这条连接 』,本文将继续深入这个问题,找到其根本原因。

问题分析

弄清这个问题,要深入 happybase 的代码看一下具体实现逻辑。

happybase 的连接池要求必须要使用 context manager,以下是其通过 context manager 获取/归还 Connection 的源码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@contextlib.contextmanager
def connection(self, timeout=None):
# 获取一个 Connection 对象
connection = getattr(self._thread_connections, 'current', None)
return_after_use = False
if connection is None
return_after_use = True
connection = self._acquire_connection(timeout)
with self._lock:
self._thread_connections.current = connection
try:
# 打开连接,并返回一个可用的连接给contextmanager
connection.open()
yield connection
except (TException, socket.error):
# 捕获 Trhift 和 socket 的异常和错误,则需要 refresh thrift client,保证最后归还的 Connection 的 thrift client 是可用的。
logger.info("Replacing tainted pool connection")
connection._refresh_thrift_client()
connection.open()
raise
finally:
# 最终会归还这条连接
if return_after_use:
del self._thread_connections.current
self._return_connection(connection)

实现逻辑大概可以归纳为:

  • 在 pool 初始化的时候构建一个包含若干 Connection 的 queue 以及一个 lock。
  • 当向连接池请求连接时,open 一个 Connection 并返回,在这个过程中会 catch TException 以及 socket.error。
  • 如果 catch 到以上 error/exception 则会刷新 Connection 对象的thrift_client,保证在退出 context manager 时返回给队列的 Connection 是可用的。
  • 退出 context manager 前将 Connection 返还。

从以上逻辑可以发现,happybase 是通过 context manager 保证 Connection 在退出时是正常的,而我们的场景中出现了 socket.error 却并没有被 catch 住,说明有可能是错误发生在 context manager 之外,回到代码:

1
2
3
4
5
6
7
8
9
def recent_events_v1(start, end, table=None, filter_str=None, limit=2000):
with get_connetion_pool().connection() as conn:
if table is not None:
t = conn.table(table)
else:
t = conn.table(TABLE)
start_row = 'ARTICLE' + str(start * 1000000)
end_row = 'ARTICLE' + str(end * 1000000)
return t.scan(row_start=start_row, row_stop=end_row, filter=filter_str, limit=limit)

从代码可以发现,再使用连接池的过程中,退出 context manager 前直接 return table.scan() ,而 scan 方法会创建一个 scanner ,最终返回一个 generator,到这里基本可以说是水落石出了!

问题原因

因为 generator 的特性,在退出 context manager 前并没有发生真正的查询,所以这时返回给 connent pool 的 Connection 仍然是没有问题的。只用在真正遍历这个 generator 时才会发生数据查询,而这个过程肯定在 context manager 之外,所以此时如果出现 error 则不会再有类似的 catch 逻辑去保证这条 Connection 在发生异常时去刷新 thrift_client,最终导致这条已经归还给 pool 的 Connection 失效了。

同时这里还会存在另一个问题,在并发场景下,遍历 generator 发生查询时可能这个 Connection 已经被分配给其他线程使用了,导致这个 Connection 同时被两个线程所有,出现一些难以预测的问题。

解决问题

既然发现了问题的根因,解决起来就比较简单了,只要保证所有使用 Connection 的逻辑都发生在 context manager 内就好,所以这里可以把遍历generator 的逻辑放在 context manager 内,最终返回一个 list 对象而不是 generator,代码如下:

1
2
3
4
5
6
7
8
9
10
def recent_events_v4(start, end, table=None, filter_str=None, limit=2000):
with get_connetion_pool().connection() as conn:
if table is not None:
t = conn.table(table)
else:
t = conn.table(TABLE)
start_row = 'ARTICLE' + str(start * 1000000)
end_row = 'ARTICLE' + str(end * 1000000)
result_generator = t.scan(row_start=start_row, row_stop=end_row, filter=filter_str, limit=limit)
return [l for l in result_generator]

从下图可以看出,修正后的代码在运行的过程中出现 TTransportException后,不会在出现 Broken pipe

不会在出现 Broken pipe

继续思考

上面分析了『为什么一条异常的连接会出现在 connect pool 中』,那 『而且总会拿到这条连接 』 又是为什么?

仔细观察 happybase 获取 connection 的前半部分逻辑可以发现,其是优先从self._thread_connections 获取链接对象,当获取不到时才通过 self._acquire_connection 从 pool 中取。

这个 self._thread_connections 是个什么东西?这是一个 thread local 变量,即线程自有的局部变量,其他线程不可访问,happybase 源码:

1
self._thread_connections = threading.local()

某个线程第一次请求获取 Connection 时,通过这个 thread local 变量,把分配给它的 Connection 记录下来,当下次这个线程再请求时,则优先把这个 thread local 变量记录的 Connection 返回给线程。

因为我们是单线程场景,所以每次返回给主进程的都是同一个有问题的 Connection 对象,这就解释了 『总会拿到问题链接』 这个问题。

小结

综上,在使用连接池的场景中,注意类似 scan 这种具有延时行的操作,一定要放在 context mananer 内,才能保证连接池内的连接可用~