记录多进程/多线程 python 脚本的死锁答案

【问题标题】：Deadlock with logging multiprocess/multithread python script记录多进程/多线程 python 脚本的死锁
【发布时间】：2014-07-01 11:54:07
【问题描述】：

我遇到了从以下脚本收集日志的问题。一旦我将SLEEP_TIME 设置为太“小”的值，LoggingThread 线程以某种方式阻塞了日志记录模块。记录请求时脚本冻结在action 函数中。如果SLEEP_TIME 约为 0.1，则脚本收集所有日志消息都符合我的预期。

我尝试关注this answer，但它并没有解决我的问题。

import multiprocessing
import threading
import logging
import time

SLEEP_TIME = 0.000001

logger = logging.getLogger()

ch = logging.StreamHandler()
ch.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(funcName)s(): %(message)s'))
ch.setLevel(logging.DEBUG)

logger.setLevel(logging.DEBUG)
logger.addHandler(ch)


class LoggingThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        while True:
            logger.debug('LoggingThread: {}'.format(self))
            time.sleep(SLEEP_TIME)


def action(i):
    logger.debug('action: {}'.format(i))


def do_parallel_job():

    processes = multiprocessing.cpu_count()
    pool = multiprocessing.Pool(processes=processes)
    for i in range(20):
        pool.apply_async(action, args=(i,))
    pool.close()
    pool.join()



if __name__ == '__main__':

    logger.debug('START')

    #
    # multithread part
    #
    for _ in range(10):
        lt = LoggingThread()
        lt.setDaemon(True)
        lt.start()

    #
    # multiprocess part
    #
    do_parallel_job()

    logger.debug('FINISH')

如何在多进程和多线程脚本中使用日志模块？

【问题讨论】：

我似乎无法重现您的问题。能否提供线程创建/启动代码？
只是为了确定：action() 的执行冻结（永远不会产生带有“actions:”的日志消息）。 LoggingThreads 永远做他们的工作。
死锁的概率取决于SLEEP_TIME的值。

标签： python multithreading logging multiprocess

【解决方案1】：

这可能是bug 6721。

这个问题在你有锁、线程和分叉的任何情况下都很常见。如果线程 1 有锁，而线程 2 调用 fork，则在分叉的进程中，将只有线程 2，并且锁将永远持有。在你的情况下，就是logging.StreamHandler.lock。

可以在here (permalink) 中找到针对logging 模块的修复程序。请注意，您还需要处理任何其他锁。

【讨论】：

【解决方案2】：

我最近在使用日志模块和 Pathos 多处理库时遇到了类似的问题。仍然不能 100% 确定，但似乎在我的情况下，问题可能是由于日志处理程序试图在不同进程中重用锁定对象造成的。

能够使用默认日志处理程序的简单包装器来修复它：

import threading
from collections import defaultdict
from multiprocessing import current_process

import colorlog


class ProcessSafeHandler(colorlog.StreamHandler):
    def __init__(self):
        super().__init__()

        self._locks = defaultdict(lambda: threading.RLock())

    def acquire(self):
        current_process_id = current_process().pid
        self._locks[current_process_id].acquire()

    def release(self):
        current_process_id = current_process().pid
        self._locks[current_process_id].release()

【讨论】：