python - 类型错误 : init() missing 1 required positional argument: 'message' using Multiprocessing

Question

我正在使用多处理池运行一段代码。该代码适用于一个数据集，但在另一个数据集上失败。显然问题是由数据驱动的 - 话虽如此，我不清楚从哪里开始进行故障排除，因为我收到的错误如下。任何关于起点的提示都是最有帮助的。两组数据都是使用相同的代码准备的——所以我不希望有什么不同——但我在这里。

另请参阅 Robert 的评论 - 我们在操作系统和 Python 3.6 版(我有 3.4，他有 3.6)和完全不同的数据集方面存在差异。然而错误与 python 代码中的行完全相同。

我的怀疑:

1. 每个内核都有内存限制。

2. 有一段时间后进程开始收集 - 发现进程没有结束并放弃。

   线程 Thread-9 中的异常:

   追溯(最近的调用最后):

   文件“C:\Program Files\Python\WinPython-64bit-3.4.4.4Qt5\python-3.4.4.amd64\lib\threading.py”，第 911 行，在 _bootstrap_inner self.run()

   文件“C:\Program Files\Python\WinPython-64bit-3.4.4.4Qt5\python-3.4.4.amd64\lib\threading.py”，第 859 行，运行中 self._target(*self._args, **self._kwargs)

   文件“C:\Program Files\Python\WinPython-64bit-3.4.4.4Qt5\python-3.4.4.amd64\lib\multiprocessing\pool.py”，第 429 行，在 _handle_results 中 任务=获取()

   文件“C:\Program Files\Python\WinPython-64bit-3.4.4.4Qt5\python-3.4.4.amd64\lib\multiprocessing\connection.py”，第 251 行，在 recv 中 返回 ForkingPickler.loads(buf.getbuffer())

   TypeError:init() 缺少 1 个必需的位置参数:'message'

Answer 1

我认为问题在于 langdetect 在这里悄悄声明了一个隐藏的全局检测器工厂 https://github.com/Mimino666/langdetect/blob/master/langdetect/detector_factory.py#L120 :

def init_factory():
    global _factory
    if _factory is None:
        _factory = DetectorFactory()
        _factory.load_profile(PROFILES_DIRECTORY)

def detect(text):
    init_factory()
    detector = _factory.create()
    detector.append(text)
    return detector.detect()


def detect_langs(text):
    init_factory()
    detector = _factory.create()
    detector.append(text)
    return detector.get_probabilities()

根据我的经验，这种事情可能会导致多进程出现问题，因为它与多进程尝试跨进程共享内存中的资源以及管理工作进程和主进程中的 namespace 的方式发生冲突，尽管在这种情况下是确切的机制对我来说是一个黑盒子。我通过向我的池初始化函数添加对 init_factory 函数的调用来修复它:

from langdetect.detector_factory import init_factory
def worker_init_corpus(stops_in):
    global sess
    global stops
    sess = requests.Session()
    sess.mount("http://", HTTPAdapter(max_retries=10))
    stops = stops_in
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    init_factory()

仅供引用:“sess”逻辑是为每个工作人员提供一个用于请求的 HTTP 连接池，以解决在将该模块与多处理池一起使用时出现的类似问题。如果你不这样做，工作人员将通过父进程备份所有的 http 通信，因为默认情况下隐藏的全局 http 连接池就在那里，然后一切都非常缓慢。这是我遇到的问题之一，让我怀疑这里有类似的原因。

此外，为了进一步减少潜在的混淆:stops 用于为映射函数提供我正在使用的停用词列表。 signal 调用是强制池在遇到用户中断 (ctrl-c) 时正常退出。否则，他们经常会成为孤儿，并在父进程死亡后继续前进。

然后我的池是这样初始化的:

self.pool = mp.Pool(mp.cpu_count()-2, worker_init_corpus, (self.stops,))

我还将对 detect 的调用包装在 try/catch LangDetectExeception block 中:

try:
    posting_out["lang"] = detect(posting_out["job_description"])
except LangDetectException:
    posting_out["lang"] = "none"

但这并不能自行解决问题。非常有信心初始化是解决方法。