マルチプロセスで単一ファイルにログする方法.Pool.apply_async

Question

multprocess.Pool.apply_asyncで動作する単一のファイルにログを取得できません。 Logging Cookbookからthisの例を適用しようとしていますが、multiprocessing.Processの場合にのみ機能します。ログキューをapply_asyncに渡すことは効果がないようです。 プールを使用して、同時スレッド数を簡単に管理できるようにしたいと考えています。

次のマルチプロセッシングプロセスの例は、メインプロセスからログメッセージを受け取っていないことを除いて、私にとってはうまくいきます。100個の大きなジョブがある場合はうまく動作しないと思います。 multiprocessing.Manager、multiprocessing.Queue、multiprocessing.get_logger、apply_async.getを使用して、私は多くのわずかな変化を試みた

def main_with_pool(): 
    start_time = time.time() 
    queue = multiprocessing.Queue(-1) 
    listener = multiprocessing.Process(target=listener_process, 
             args=(queue, listener_configurer)) 
    listener.start() 
    pool = multiprocessing.Pool(processes=3) 
    job_list = [np.random.randint(10)/2 for i in range(10)] 
    single_thread_time = np.sum(job_list) 
    for i, sleep_time in enumerate(job_list): 
     name = str(i) 
     pool.apply_async(worker_function, 
         args=(sleep_time, name, queue, worker_configurer)) 

    queue.put_nowait(None) 
    listener.join() 
    end_time = time.time() 
    print("Script execution time was {}s, but single-thread time was {}s".format(
     (end_time - start_time), 
     single_thread_time 
    )) 

if __name__ == "__main__": 
    main_with_pool() 

：

import logging 
import logging.handlers 
import numpy as np 
import time 
import multiprocessing 
import pandas as pd 
log_file = 'PATH_TO_FILE/log_file.log' 

def listener_configurer(): 
    root = logging.getLogger() 
    h = logging.FileHandler(log_file) 
    f = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s') 
    h.setFormatter(f) 
    root.addHandler(h) 

# This is the listener process top-level loop: wait for logging events 
# (LogRecords)on the queue and handle them, quit when you get a None for a 
# LogRecord. 
def listener_process(queue, configurer): 
    configurer() 
    while True: 
     try: 
      record = queue.get() 
      if record is None: # We send this as a sentinel to tell the listener to quit. 
       break 
      logger = logging.getLogger(record.name) 
      logger.handle(record) # No level or filter logic applied - just do it! 
     except Exception: 
      import sys, traceback 
      print('Whoops! Problem:', file=sys.stderr) 
      traceback.print_exc(file=sys.stderr) 


def worker_configurer(queue): 
    h = logging.handlers.QueueHandler(queue) # Just the one handler needed 
    root = logging.getLogger() 
    root.addHandler(h) 
    # send all messages, for demo; no other level or filter logic applied. 
    root.setLevel(logging.DEBUG) 


# This is the worker process top-level loop, which just logs ten events with 
# random intervening delays before terminating. 
# The print messages are just so you know it's doing something! 
def worker_function(sleep_time, name, queue, configurer): 
    configurer(queue) 
    start_message = 'Worker {} started and will now sleep for {}s'.format(name, sleep_time) 
    logging.info(start_message) 
    time.sleep(sleep_time) 
    success_message = 'Worker {} has finished sleeping for {}s'.format(name, sleep_time) 
    logging.info(success_message) 

def main_with_process(): 
    start_time = time.time() 
    single_thread_time = 0. 
    queue = multiprocessing.Queue(-1) 
    listener = multiprocessing.Process(target=listener_process, 
             args=(queue, listener_configurer)) 
    listener.start() 
    workers = [] 
    for i in range(10): 
     name = str(i) 
     sleep_time = np.random.randint(10)/2 
     single_thread_time += sleep_time 
     worker = multiprocessing.Process(target=worker_function, 
             args=(sleep_time, name, queue, worker_configurer)) 
     workers.append(worker) 
     worker.start() 
    for w in workers: 
     w.join() 
    queue.put_nowait(None) 
    listener.join() 
    end_time = time.time() 
    final_message = "Script execution time was {}s, but single-thread time was {}s".format(
     (end_time - start_time), 
     single_thread_time 
    ) 
    print(final_message) 

if __name__ == "__main__": 
    main_with_process() 

しかし、私は、次の適応が仕事を得ることができません（）、しかし、何も動かされていない。

私はこのための既製のソリューションがあると思います。代わりにセロリを試してみるべきですか？

ありがとう

Answer 1

あり絡み合っているここでは2つの別々の問題、実際には、次のとおりです。

• あなたは（あなたが直接起動する労働者に渡すことができますプールベースの関数の引数としてmultiprocessing.Queue()オブジェクトを渡すことはできませんが、それ以前の "それ以上の"ものではない）。
• Noneをリスナープロセスに送信する前に、すべての非同期ワーカーが完了するまで待つ必要があります。

  queue = multiprocessing.Queue(-1) 
  

  ：管理者が管理するQueue()インスタンスとして

  queue = multiprocessing.Manager().Queue(-1) 
  

  は通過することができ

は交換してください、最初のものを修正するには。

pool.close() 
pool.join() 
queue.put_nowait(None) 

以上の複合体を：：

getters = [] 
for i, sleep_time in enumerate(job_list): 
    name = str(i) 
    getters.append(
     pool.apply_async(worker_function, 
        args=(sleep_time, name, queue, worker_configurer)) 
    ) 
while len(getters): 
    getters.pop().get() 
# optionally, close and join pool here (generally a good idea anyway) 
queue.put_nowait(None) 

（あなたのいずれか、二を修正するなど、各非同期呼び出しからそれぞれの結果を収集し、またはプールを閉鎖し、それを待つために

put_nowaitをputの待機中のバージョンに置き換えて、無制限の長さのキューを使用しないことも検討してください。）

Answer 2

2つのキューを使用することを検討してください。最初のキューは、作業者のデータを格納する場所です。ジョブ完了後の各ワーカーは、結果を2番目のキューにプッシュします。この2番目のキューを消費して、ログをファイルに書き込みます。