マルチプロセスでのログ(python, 引数にlogger)

狙い

関数multiは関数oneを並列的に実行する。
multiはloggerを引数として受け取る。
main.pyでloggerを作成し，それをmultiに渡すことで，multiやoneのログをmain.pyで定義したhandlerで制御できる。
関数multi内でhandlerまで定義してしまう例が多くみられたので引数としてloggerを受け取る例を残す。

コード

multi外で，loggerの作成とhandlerの作成，追加をする。

multiの外

# 内容は略している
logger = logging.getLogger()
h = logging.handler()
logger.addHandler(h)
multi(logger)

multi内ではin_loggerがマルチプロセスに対するキューを扱う。
キューをmultiの外で定義したloggerのhandlerたちに渡すことでhandlerの設定に応じたログが得られる。

関数multi

listener = logging.handlers.QueueListener(que, *logger.handlers, respect_handler_level=True)

以下のコードでは随所にloggerのアドレスを示すようにしている。
executorで渡されたin_loggerのアドレスがmultiの定義のすぐ後に定義されたin_loggerと違うことに注目。

import logging, logging.handlers
import concurrent.futures, multiprocessing
from pathlib import Path

def multi(logger):
    # キューを受け取るlogger
    # getLoggerの引数をloggerと同じにするとid(in_logger) = id(logger)になり問題が起きる。
    # in_logger = loggerならどうなるかみてみよう
    in_logger = logging.getLogger(__name__) 
    in_logger.setLevel(logging.DEBUG)
    logger.debug(f'passed logger:{id(logger)}')
    logger.debug(f'in_logger: {id(in_logger)}')
    
    with multiprocessing.Manager() as manager:
        que = manager.Queue(-1)
        qh = logging.handlers.QueueHandler(que)
        qh.setLevel(60) # qhは受け取りだけ
        in_logger.addHandler(qh)
        listener = logging.handlers.QueueListener(que, *logger.handlers, respect_handler_level=True)
        listener.start()
        
        # 関数oneのためだけ。demo用
        items = list(range(1000))
        try:
            max_workers = 6
            with concurrent.futures.ProcessPoolExecutor(
                    max_workers=max_workers,
                    initializer=log_initializer, initargs=(que,),
                ) as executor:
                for item in items:
                    executor.submit(func_wrapper,item,in_logger)            
        except:
            logger.exception()
        finally:
            listener.stop()
    
    
    return

def log_initializer(que):
    # 各 子プロセスで一度だけ実行される。
    # oneを実行するそれぞれのプロセスでloggerとqueueHandlerを作成
    p_logger = logging.getLogger()
    p_logger.setLevel(logging.DEBUG)
    qh = logging.handlers.QueueHandler(que) # NOTSET (0)
    qh.setLevel(logging.DEBUG)
    p_logger.addHandler(qh)
    p_logger.info('queue handler added')
    p_logger.info
    return

def func_wrapper(item, logger):
    logger.debug(f'logger in func_wrapper: {id(logger)}')
    one(item, logger)

def one(item, logger):
    logger.debug('item: %d', item)
    pass

if __name__ == '__main__':
    ## logger
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    # stream
    sh = logging.StreamHandler()
    shf = logging.Formatter('%(asctime)s - %(levelname)8s -%(processName)s- %(module)20s - %(funcName)20s: %(message)s')
    sh.setFormatter(shf)
    sh.setLevel(logging.DEBUG)
    logger.addHandler(sh)
    # file
    logFileName = Path(__file__).parent.joinpath('log.log') # パス
    fh = logging.FileHandler(logFileName, encoding='utf-8', mode='w')
    fhf = logging.Formatter('%(asctime)s - %(levelname)8s -%(processName)20s - %(module)20s - %(funcName)20s: %(message)s')
    fh.setFormatter(fhf)
    fh.setLevel(logging.DEBUG)
    logger.addHandler(fh)
    logger.debug(f'first logger:  {id(logger)}')
    # run
    multi(logger)

一応，main.pyで実行してうまくいくかを確認。

main.py

import multi
import logging

if __name__ == '__main__':
    m_logger = logging.getLogger(__name__)
    m_logger.setLevel(logging.DEBUG)
    sh = logging.StreamHandler()
    m_logger.addHandler(sh)
    m_logger.debug(f'm_logger: {id(m_logger)}')
    multi(logger=m_logger)