Python多进程使用Logging官方cookbook示例触发死锁问题

问题现象 #

多进程场景下按照官方推荐方案使用QueueHandler+QueueListener实现日志时，若日志监听线程先于子进程启动，会触发死锁，和“队列消费者应先于生产者启动”的常规认知矛盾。

复现代码 #

以下为调整启动顺序的官方Logging Cookbook示例，仅将日志监听线程的启动位置从子进程启动后移至子进程启动前：

import logging
import logging.config
import logging.handlers
from multiprocessing import Process, Queue
import random
import threading
import time

def logger_thread(q):
    while True:
        record = q.get()
        if record is None:
            break
        logger = logging.getLogger(record.name)
        logger.handle(record)


def worker_process(q):
    qh = logging.handlers.QueueHandler(q)
    root = logging.getLogger()
    root.setLevel(logging.DEBUG)
    root.addHandler(qh)
    levels = [logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR,
              logging.CRITICAL]
    loggers = ['foo', 'foo.bar', 'foo.bar.baz',
               'spam', 'spam.ham', 'spam.ham.eggs']
    for i in range(100):
        lvl = random.choice(levels)
        logger = logging.getLogger(random.choice(loggers))
        logger.log(lvl, 'Message no. %d', i)

if __name__ == '__main__':
    q = Queue()
    d = {
        'version': 1,
        'formatters': {
            'detailed': {
                'class': 'logging.Formatter',
                'format': '%(asctime)s %(name)-15s %(levelname)-8s %(processName)-10s %(message)s'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'DEBUG',
            },
            'file': {
                'class': 'logging.FileHandler',
                'filename': 'mplog.log',
                'mode': 'w',
                'formatter': 'detailed',
            },
            'foofile': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-foo.log',
                'mode': 'w',
                'formatter': 'detailed',
                'level': 'DEBUG'
            },
            'errors': {
                'class': 'logging.FileHandler',
                'filename': 'mplog-errors.log',
                'mode': 'w',
                'level': 'DEBUG',
                'formatter': 'detailed',
            },
        },
        'loggers': {
            'foo': {
                'handlers': ['foofile']
            }
        },
        'root': {
            'level': 'DEBUG',
            'handlers': ['console', 'file', 'errors']
        },
    }
    logging.config.dictConfig(d)
    lp = threading.Thread(target=logger_thread, args=(q,))

    lp.start() # 日志线程在子进程启动前启动
    
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1), args=(q,))
        workers.append(wp)
        wp.start()
    
    # 官方示例中日志线程在此处启动
    #lp.start()

    for wp in workers:
        wp.join()
    q.put(None)
    lp.join()

初步排查 #

移除Logging相关逻辑，简化为「独立线程从队列读取数据写入文件、子进程作为生产者向队列投递数据」的场景后，死锁完全消失，初步判断问题不在多进程Queue本身，而在Logging库实现。简化版无死锁代码如下：

from multiprocessing import Process, Queue
import threading
import time

def queue_listener_thread(q):
    with open('test.log', 'w') as f:
        while True:
            record = q.get()
            if record is None:
                break
            time.sleep(0.001)
            f.write(f'{str(record)}\n')


def worker_process(q):
    time.sleep(0.1)
    for i in range(100):
        time.sleep(0.001)
        q.put_nowait(i)


if __name__ == '__main__':
    q = Queue()
    lp = threading.Thread(target=queue_listener_thread, args=(q,))
    lp.start()
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1), args=(q,))
        workers.append(wp)
        wp.start()

    for wp in workers:
        wp.join()
    q.put(None)
    lp.join()

运行环境 #

• 业务场景：日志模块在应用启动时完成初始化，后续按需启动、销毁子进程，为生产环境普遍使用模式
• 系统环境：Fedora Linux 35
• Python版本：3.10.5

根因定位过程 #

通过gdb抓取死锁时的Python调用栈如下：

(gdb) py-bt
Traceback (most recent call first):
  File "/usr/lib64/python3.10/logging/__init__.py", line 1084, in flush
    self.stream.flush()
  File "/usr/lib64/python3.10/logging/__init__.py", line 1104, in emit
    self.flush()
  File "/usr/lib64/python3.10/logging/__init__.py", line 1218, in emit
    StreamHandler.emit(self, record)

最初误判问题由文件IO触发，实际阻塞点为StreamHandler对应的标准输出打印逻辑。将简化版代码中的写文件逻辑替换为直接print输出到屏幕后，死锁可稳定复现，复现代码如下：

from multiprocessing import Process, Queue
import threading

def queue_listener_thread(q):
    while True:
        record = q.get()
        if record is None:
            break
        print(f'{str(record)}')


def worker_process(q):
    for i in range(100):
        q.put_nowait(i)


if __name__ == '__main__':
    q = Queue()
    lp = threading.Thread(target=queue_listener_thread, args=(q,))
    lp.start()
    workers = []
    for i in range(5):
        wp = Process(target=worker_process, name='worker %d' % (i + 1), args=(q,))
        workers.append(wp)
        wp.start()

    for wp in workers:
        wp.join()
    q.put(None)
    lp.join()

内容的提问来源于stack exchange，提问作者MegaMax