尝试递归比较列表时带有多处理模块的 RuntimeError

Question

我正在生成一个列表，其中包含随机生成的 0 和 1 的子列表，然后尝试将每个列表与每个其他列表进行比较，以有效地确定它们的相似性。

我知道我的代码适用于单个进程（即不涉及multiprocessing，但一旦我开始涉及multiprocessing.Pool() 或multiprocessing.Process()，一切都会开始中断。

我想比较单个进程与多个进程所需的时间。我已经用threading 尝试过这个，但一个进程实际上最终花费的时间更少，可能是由于全局解释器锁。

这是我的代码：

import difflib
import secrets
import timeit
import multiprocessing
import numpy

random_lists = [[secrets.randbelow(2) for _ in range(500)] for _ in range(500)]
random_lists_split = numpy.array_split(numpy.array(random_lists), 5)


def get_similarity_value(lists_to_check, sublists_to_check) -> list:
    ratios = []
    matcher = difflib.SequenceMatcher()
    for sublist_major in sublists_to_check:
        try:
            sublist_major = sublist_major.tolist()
        except AttributeError:
            pass
        for sublist_minor in lists_to_check:
            if sublist_major == sublist_minor or [lists_to_check.index(sublist_major), lists_to_check.index(sublist_minor)] in [ratios[i][1] for i in range(len(ratios))] or [lists_to_check.index(sublist_minor), lists_to_check.index(sublist_major)] in [ratios[i][1] for i in range(len(ratios))]:  # or lists_to_check.index(sublist_major.tolist()) > lists_to_check.index(sublist_minor):
                pass
            else:
                matcher.set_seqs(sublist_major, sublist_minor)
                ratios.append([matcher.ratio(), sorted([lists_to_check.index(sublist_major), lists_to_check.index(sublist_minor)])])
    return ratios


def start():
    test = multiprocessing.Pool(4)
    data = [(random_lists, random_lists_split[i]) for i in range(len(random_lists_split))]
    print(test.map(get_similarity_value, data))


statement = timeit.Timer(start)
print(statement.timeit(1))

statement2 = timeit.Timer(lambda: get_similarity_value(random_lists, random_lists))
print(statement2.timeit(1))

这是错误：

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 105, in spawn_main
    exitcode = _main(fd)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 114, in _main
    prepare(preparation_data)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 225, in prepare
    _fixup_main_from_path(data['init_main_from_path'])
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 277, in _fixup_main_from_path
    run_name="__mp_main__")
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\runpy.py", line 263, in run_path
    pkg_name=pkg_name, script_name=fname)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\runpy.py", line 96, in _run_module_code
    mod_name, mod_spec, pkg_name, script_name)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "timings.py", line 38, in <module>
    print(statement.timeit(1))
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\timeit.py", line 178, in timeit
    timing = self.inner(it, self.timer)
  File "<timeit-src>", line 6, in inner
  File "timings.py", line 32, in start
    test = multiprocessing.Pool(4)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\context.py", line 119, in Pool
    context=self.get_context())
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\pool.py", line 174, in __init__
    self._repopulate_pool()
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\pool.py", line 239, in _repopulate_pool
    w.start()
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\process.py", line 105, in start
    self._popen = self._Popen(self)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\context.py", line 322, in _Popen
    return Popen(process_obj)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\popen_spawn_win32.py", line 33, in __init__
    prep_data = spawn.get_preparation_data(process_obj._name)
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 143, in get_preparation_data
    _check_not_importing_main()
  File "C:\ProgramData\Anaconda3\envs\Computing Coursework\lib\multiprocessing\spawn.py", line 136, in _check_not_importing_main
is not going to be frozen to produce an executable.''')
RuntimeError: 
    An attempt has been made to start a new process before the
    current process has finished its bootstrapping phase.

    This probably means that you are not using fork to start your
    child processes and you have forgotten to use the proper idiom
    in the main module:

        if __name__ == '__main__':
            freeze_support()
            ...

    The "freeze_support()" line can be omitted if the program
    is not going to be frozen to produce an executable.

注意我试过使用multiprocessing.freeze_support()，但它会导致同样的错误。代码似乎也试图无限期地运行，因为错误一遍又一遍地出现。

谢谢！

Answer 1

问题在于，您的顶级代码（包括创建子 Process 的代码）没有受到保护，无法在子进程中运行。

正如the docs 解释的那样：如果您没有使用fork 启动方法（并且由于您使用的是Windows，所以您没有使用）：

确保新的 Python 解释器可以安全地导入主模块，而不会导致意外的副作用（例如启动新进程）。

事实上，它与该警告后面的示例几乎相同。您正在推出一整群孩子而不是仅仅一个，但这是同样的问题。池中的每个孩子都尝试启动一个新池，幸运的是，multiprocessing 发现发生了一些不好的事情并以RuntimeError 失败，而不是指数级生成进程，直到 Windows 拒绝再生成或它的调度程序刚刚崩溃下来。

正如文档所说：

应该使用if __name__ == '__main__' 来保护程序的“入口点”：

在你的情况下，这意味着这部分：

if __name__ == '__main__':
    statement = timeit.Timer(start)
    print(statement.timeit(1))

    statement2 = timeit.Timer(lambda: get_similarity_value(random_lists, random_lists))
    print(statement2.timeit(1))