关于工作刷新的 python 多处理池通知

Question

我正在使用 Python 2.7 的 multiprocessing.Pool 来管理 3 个工作进程的池。每个工作线程都相当复杂，并且某些第三方代码中可能存在资源泄漏，导致连续运行 6-8 小时后出现问题。因此，我想使用 maxtasksperchild 定期刷新工作人员。

我还希望每个工作人员写入自己单独的日志文件。如果没有 maxtasksperchild，我会使用共享的 multiprocessing.Value 为每个工作线程分配一个整数(0、1 或 2)，然后使用该整数来命名日志文件。

使用 maxtasksperchild 我想在工作人员完成后重用日志文件。因此，如果整个过程运行一个月，我只需要三个日志文件，而不是为每个生成的工作人员提供一个日志文件。

如果我可以传递回调(例如，与当前支持的初始化程序一起使用的终结器)，这将很简单。如果没有它，我就看不到一种强大而简单的方法来做到这一点。

Answer 1

据我所知，这是没有记录的，但是 multiprocessing 有一个 Finalizer 类，“它支持使用弱引用的对象终结”。您可以使用它在初始化程序中注册终结器。

不过，在这种情况下，我认为 multiprocessing.Value 不是一个有用的同步选择。多个工作人员可以同时退出，从而表明哪些文件整数是空闲的，这比(锁定的)计数器可以提供的更多。

我建议使用多个裸 multiprocessing.Lock，每个文件一个，而不是:

from multiprocessing import Pool, Lock, current_process
from multiprocessing.util import Finalize


def f(n):
    global fileno
    for _ in range(int(n)):  # xrange for Python 2
        pass
    return fileno


def init_fileno(file_locks):
    for i, lock in enumerate(file_locks):
        if lock.acquire(False):  # non-blocking attempt
            globals()['fileno'] = i
            print("{} using fileno: {}".format(current_process().name, i))
            Finalize(lock, lock.release, exitpriority=15)
            break


if __name__ == '__main__':

    n_proc = 3
    file_locks = [Lock() for _ in range(n_proc)]

    pool = Pool(
        n_proc, initializer=init_fileno, initargs=(file_locks,),
        maxtasksperchild=2
    )

    print(pool.map(func=f, iterable=[50e6] * 18))
    pool.close()
    pool.join()
    # all locks should be available if all finalizers did run
    assert all(lock.acquire(False) for lock in file_locks)

输出:

ForkPoolWorker-1 using fileno: 0
ForkPoolWorker-2 using fileno: 1
ForkPoolWorker-3 using fileno: 2
ForkPoolWorker-4 using fileno: 0
ForkPoolWorker-5 using fileno: 1
ForkPoolWorker-6 using fileno: 2
[0, 0, 1, 1, 2, 2, 0, 0, 1, 1, 2, 2, 0, 0, 1, 1, 2, 2]

Process finished with exit code 0

请注意，使用 Python 3，您无法可靠地使用 Pool 的上下文管理器，而不是使用上面所示的旧方法。池的上下文管理器(不幸的是)调用 terminate()，这可能会在终结器有机会运行之前终止工作进程。