postgresql - 在查询 PostgreSQL 数据库的黑盒函数上使用 multiprocessing.Pool 时为 "InterfaceError: connection already closed"

Question

我得到了一个 Python (2.7) 函数，它接受 3 个字符串作为参数，并返回一个字典列表。由于项目的性质，我无法更改函数，它非常复杂，调用其他几个非标准 Python 模块并使用 psychopg2 查询 PostgreSQL 数据库。我认为是 Postgres 的功能导致了我的问题。

我想使用 multiprocessing 模块来加速调用函数数百次。我已经编写了一个“帮助程序”函数，以便我可以在我的函数中使用 multiprocessing.Pool(它只需要一个参数):

from function_script import function

def function_helper(args):
    return function(*args)

我的主要代码是这样的:

from helper_script import function_helper
from multiprocessing import Pool

argument_a = ['a0', 'a1', ..., 'a99']
argument_b = ['b0', 'b1', ..., 'b99']
argument_c = ['c0', 'c1', ..., 'c99']

input = zip(argument_a, argument_b, argument_c)
p = Pool(4)
results = p.map(function_helper, input)

print results

我期待的是字典列表的列表，但是我收到以下错误:

Traceback (most recent call last):
  File "/local/python/2.7/lib/python2.7/site-packages/variantValidator/variantValidator.py", line 898, in validator
    vr.validate(input_parses)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 33, in validate
    return self._ivr.validate(var, strict) and self._evr.validate(var, strict)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 69, in validate
    (res, msg) = self._ref_is_valid(var)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 89, in _ref_is_valid
    var_x = self.vm.c_to_n(var) if var.type == "c" else var
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/variantmapper.py", line 223, in c_to_n
    tm = self._fetch_TranscriptMapper(tx_ac=var_c.ac, alt_ac=var_c.ac, alt_aln_method="transcript")
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/decorators/lru_cache.py", line 176, in wrapper
    result = user_function(*args, **kwds)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/variantmapper.py", line 372, in _fetch_TranscriptMapper
    self.hdp, tx_ac=tx_ac, alt_ac=alt_ac, alt_aln_method=alt_aln_method)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/transcriptmapper.py", line 69, in __init__
    self.tx_identity_info = hdp.get_tx_identity_info(self.tx_ac)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/decorators/lru_cache.py", line 176, in wrapper
    result = user_function(*args, **kwds)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 353, in get_tx_identity_info
    rows = self._fetchall(self._queries['tx_identity_info'], [tx_ac])
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 216, in _fetchall
    with self._get_cursor() as cur:
  File "/local/python/2.7/lib/python2.7/contextlib.py", line 17, in __enter__
    return self.gen.next()
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 529, in _get_cursor
    cur.execute("set search_path = " + self.url.schema + ";")
  File "/local/python/2.7/lib/python2.7/site-packages/psycopg2/extras.py", line 144, in execute
    return super(DictCursor, self).execute(query, vars)
DatabaseError: SSL error: decryption failed or bad record mac

和:

Traceback (most recent call last):
  File "/local/python/2.7/lib/python2.7/site-packages/variantValidator/variantValidator.py", line 898, in validator
    vr.validate(input_parses)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 33, in validate
    return self._ivr.validate(var, strict) and self._evr.validate(var, strict)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 69, in validate
    (res, msg) = self._ref_is_valid(var)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/validator.py", line 89, in _ref_is_valid
    var_x = self.vm.c_to_n(var) if var.type == "c" else var
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/variantmapper.py", line 223, in c_to_n
    tm = self._fetch_TranscriptMapper(tx_ac=var_c.ac, alt_ac=var_c.ac, alt_aln_method="transcript")
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/decorators/lru_cache.py", line 176, in wrapper
    result = user_function(*args, **kwds)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/variantmapper.py", line 372, in _fetch_TranscriptMapper
    self.hdp, tx_ac=tx_ac, alt_ac=alt_ac, alt_aln_method=alt_aln_method)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/transcriptmapper.py", line 69, in __init__
    self.tx_identity_info = hdp.get_tx_identity_info(self.tx_ac)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/decorators/lru_cache.py", line 176, in wrapper
    result = user_function(*args, **kwds)
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 353, in get_tx_identity_info
    rows = self._fetchall(self._queries['tx_identity_info'], [tx_ac])
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 216, in _fetchall
    with self._get_cursor() as cur:
  File "/local/python/2.7/lib/python2.7/contextlib.py", line 17, in __enter__
    return self.gen.next()
  File "/local/python/2.7/lib/python2.7/site-packages/hgvs/dataproviders/uta.py", line 526, in _get_cursor
    conn.autocommit = True
InterfaceError: connection already closed

有人知道什么可能导致 Pool 函数出现这样的行为，而在我尝试过的其他示例中使用它似乎很简单吗？如果这还不够继续下去，谁能告诉我如何找到问题的根源(这是我第一次使用别人的代码)？或者，还有其他方法可以使用多处理模块调用该函数数百次吗？

谢谢

Answer 1

我认为可能发生的情况是您的连接对象被所有工作人员使用，当一名工作人员完成其所有任务时它关闭连接，同时其他工作人员仍在工作并且连接关闭所以当其中一个工作人员工作人员尝试使用已经关闭的数据库。