python - 使用并行 python 时记录

Question

我正在使用并行 python 多次执行一个大函数 (executePipeline)。此函数还使用了多处理(使用 multiprocessing 模块)。
我在使用并行 python 模块在我的控制台上正确显示日志消息时遇到了一些问题。当我不使用它时，日志消息会很好地显示。

这是它的工作原理。我有一台服务器，每次收到来自客户端的请求时，它都会调用一个工作人员:

job = self.server.job_server.submit(func = executeWorker, args = (config, ) )

每当有来自客户端的新请求时，这个函数就会从一个新线程执行。然后 worker 调用函数 executePipeline，它使用多处理执行不同的进程。

服务器 SocketServer.TCPServer 我使用线程。我使用根记录器在我的服务器中设置了一个记录器，如下所示:

self.logger = logging.getLogger()
self.logger.setLevel(logging.INFO)
self.logger.addHandler(logging.StreamHandler() 
self.job_server = pp.Server(ncpus = 8) # for test
self.jobs = []

当我运行我的服务器时，我只能从 executePipeline 中获取日志记录，而不能从子进程中获取。此外，我仅在作业结束时而不是在运行时获取执行管道的日志记录。

这里还有 worker 代码。 “Executing pipeline with worker number” 很好地显示在我的终端中

'''
Setup logging
'''

logger = logging.getLogger()
logger.setLevel(logging.INFO)  

# worker name
publicIP = socket.gethostbyname(socket.gethostname()) 
pid = os.getpid()
workerID = unicode(str(publicIP) + ":" + str(pid))

logger.info( "Executing pipeline with worker {}".format(workerID))
res = executePipeline(config)    
markedScore = res["marker.score"]
markedDetails = res["marker.detail"]
results = {'marker.detail' : markedDetails , 'marker.score' : markedScore } 

return results

是否有一种好方法也能让日志记录正常工作并查看我的 executePipeline 函数的子进程发回了什么？

感谢您的帮助!

罗曼佐

Answer 1

当我尝试编写将结果写入共享字典的并行化测试时，我遇到了类似的问题。 multiprocessing.Manager 就是答案:

# create shared results dictionary
manager = multiprocessing.Manager()
result_dict = manager.dict({})

因此您可以简单地将日志从进程发布到该通用字典，然后对其进行处理。

或使用 LOG = multiprocessing.get_logger()，如下所述:https://docs.python.org/2/library/multiprocessing.html在这里:How should I log while using multiprocessing in Python?