快捷導(dǎo)航

Python進(jìn)程池log死鎖問題分析及解決

更新時間：2024年01月08日 08:49:38 作者：Java之美

最近線上運行的一個python任務(wù)負(fù)責(zé)處理一批數(shù)據(jù),為提高處理效率,使用了python進(jìn)程池,并會打印log,本文給大家分析了Python進(jìn)程池log死鎖問題以及解決方法,需要的朋友可以參考下

背景

最近線上運行的一個python任務(wù)負(fù)責(zé)處理一批數(shù)據(jù)，為提高處理效率，使用了python進(jìn)程池，并會打印log。最近發(fā)現(xiàn)，任務(wù)時常會出現(xiàn)夯住的情況，當(dāng)查看現(xiàn)場時發(fā)現(xiàn)，夯住時通常會有幾個子進(jìn)程打印了相關(guān)錯誤日志，然后整個任務(wù)就停滯在那里了。

原因

夯住的原因正是由于一行不起眼的log導(dǎo)致，簡而言之，Python的logging模塊在寫文件模式下，是不支持多進(jìn)程的，強行使用可能會導(dǎo)致死鎖。

問題復(fù)現(xiàn)

可以用下面的代碼來描述我們遇到的問題

 import logging
 from threading import Thread
 from queue import Queue
 from logging.handlers import QueueListener, QueueHandler
 from multiprocessing import Pool
 ?
 def setup_logging():
     # log的時候會寫到一個隊列里，然后有一個單獨的線程從這個隊列里去獲取日志信息并寫到文件里
     _log_queue = Queue()
     QueueListener(
         _log_queue, logging.FileHandler("out.log")).start()
     logging.getLogger().addHandler(QueueHandler(_log_queue))
 ?
     # 父進(jìn)程里起一個單獨的線程來寫日志
     def write_logs():
         while True:
             logging.info("hello, I just did something")
     Thread(target=write_logs).start()
 ?
 def runs_in_subprocess():
     print("About to log...")
     logging.info("hello, I did something")
     print("...logged")
 ?
 if __name__ == '__main__':
     setup_logging()
 ?
     # 讓一個進(jìn)程池在死循環(huán)里執(zhí)行，增加觸發(fā)死鎖的幾率
     while True:
         with Pool() as pool:
             pool.apply(runs_in_subprocess)

我們在linux上執(zhí)行該代碼：

 About to log...
 ...logged
 About to log...
 ...logged
 About to log...

發(fā)現(xiàn)程序輸出幾行之后就卡住了。

問題出在了哪里

python的進(jìn)程池是基于fork實現(xiàn)的，當(dāng)我們只使用fork()創(chuàng)建子進(jìn)程而不是用execve()來替換進(jìn)程上下時，需要注意一個問題：fork()出來的子進(jìn)程會和父進(jìn)程共享內(nèi)存空間，除了父進(jìn)程所擁有的線程。

對于代碼

 from threading import Thread, enumerate
 from os import fork
 from time import sleep
 ?
 # Start a thread:
 Thread(target=lambda: sleep(60)).start()
 ?
 if fork():
     print("The parent process has {} threads".format(
         len(enumerate())))
 else:
     print("The child process has {} threads".format(
         len(enumerate())))

輸出:

The parent process has 2 threads
The child process has 1 threads

可以發(fā)現(xiàn)，父進(jìn)程中的子線程并沒有被fork到子進(jìn)程中，而這正是導(dǎo)致死鎖的原因:

當(dāng)父進(jìn)程中的線程要向隊列中寫log時，它需要獲取鎖
如果恰好在獲取鎖后進(jìn)行了fork操作，那這個鎖也會被帶到子進(jìn)程中，同時這個鎖的狀態(tài)是占用中
這時候子進(jìn)程要寫日志的話，也需要獲取鎖，但是由于鎖是占用狀態(tài)，導(dǎo)致永遠(yuǎn)也無法獲取，至此，死鎖產(chǎn)生。

如何解決

使用多進(jìn)程共享隊列

出現(xiàn)上述死鎖的原因之一在于在fork子進(jìn)程的時候，把隊列和鎖的狀態(tài)都給fork過來了，那要避免死鎖，一種方案就是使用進(jìn)程共享的隊列。

 import logging
 import multiprocessing
 from logging.handlers import QueueListener
 from time import sleep
 ?
 ?
 def listener_configurer():
     root = logging.getLogger()
     h = logging.handlers.RotatingFileHandler('out.log', 'a', 300, 10)
     f = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s')
     h.setFormatter(f)
     root.addHandler(h)
 ?
 # 從隊列獲取元素，并寫日志
 def listener_process(queue, configurer):
     configurer()
     while False:
         try:
             record = queue.get()
             if record is None:  
                 break
             logger = logging.getLogger(record.name)
             logger.handle(record) 
         except Exception:
             import sys, traceback
             print('Whoops! Problem:', file=sys.stderr)
             traceback.print_exc(file=sys.stderr)
 ?
 # 業(yè)務(wù)進(jìn)程的日志配置，使用queueHandler, 將要寫的日志塞入隊列
 def worker_configurer(queue):
     h = logging.handlers.QueueHandler(queue)  
     root = logging.getLogger()
     root.addHandler(h)
     root.setLevel(logging.DEBUG)
 ?
 ?
 def runs_in_subprocess(queue, configurer):
     configurer(queue)
     print("About to log...")
     logging.debug("hello, I did something: %s", multiprocessing.current_process().name)
     print("...logged, %s",queue.qsize())
 ?
 ?
 if __name__ == '__main__':
     queue = multiprocessing.Queue(-1)
     listener = multiprocessing.Process(target=listener_process,
                                        args=(queue, listener_configurer))
     listener.start()
     
     #父進(jìn)程也持續(xù)寫日志
     worker_configurer(queue)
     def write_logs():
         while True:
             logging.debug("in main process, I just did something")
     Thread(target=write_logs).start()
 ?
     while True:
         multiprocessing.Process(target=runs_in_subprocess,
                        args=(queue, worker_configurer)).start()
         sleep(2)
 ?

在上面代碼中，我們設(shè)置了一個進(jìn)程間共享的隊列，將每個子進(jìn)程的寫日志操作轉(zhuǎn)換為向隊列添加元素，然后由單獨的另一個進(jìn)程將日志寫入文件。和文章開始處的問題代碼相比，雖然都使用了隊列，但此處用的是進(jìn)程共享隊列，不會隨著fork子進(jìn)程而出現(xiàn)多個拷貝，更不會出現(xiàn)給子進(jìn)程拷貝了一個已經(jīng)占用了的鎖的情況。

spawn

出現(xiàn)死鎖的另外一層原因是我們只進(jìn)行了fork, 但是沒有進(jìn)行execve, 即子進(jìn)程仍然和父進(jìn)程享有同樣的內(nèi)存空間導(dǎo)致，因此另一種解決方法是在fork后緊跟著執(zhí)行execve調(diào)用，對應(yīng)于python中的spawn操作，修改后的代碼如下：

 if __name__ == '__main__':
     setup_logging()
 ?
     while True:
         # 使用spawn類型的啟動
         with get_context("spawn").Pool() as pool:
             pool.apply(runs_in_subprocess)

使用spawn方法時，父進(jìn)程會啟動一個新的 Python 解釋器進(jìn)程。子進(jìn)程將只繼承那些運行進(jìn)程對象的 run()方法所必須的資源，來自父進(jìn)程的非必需文件描述符和句柄將不會被繼承，因此使用此方法啟動進(jìn)程會比較慢，但是安全。

以上就是Python進(jìn)程池log死鎖問題分析及解決的詳細(xì)內(nèi)容，更多關(guān)于Python進(jìn)程池log死鎖的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: