問題
deadlockのようなものが発生し、プログラムが終了しません。
公式サイトを参考にして作った、multiprocessing
を用いた並列処理中のログをサーバースレッドを立てて一括管理するプログラムで起こりました。
Ctrl+Cによるエラーメッセージ
Exception ignored in: <module 'threading' from '/Users/username/miniforge3/envs/LoggingServer/lib/python3.9/threading.py'>
Traceback (most recent call last):
File "/Users/username/miniforge3/envs/LoggingServer/lib/python3.9/threading.py", line 1470, in _shutdown
lock.acquire()
KeyboardInterrupt:
環境
macOS Monterey arm64
Python 3.9.10 | packaged by conda-forge
標準モジュールのみを使用しています。
ソースコード
import sys, logging
import socketserver, struct, pickle
from typing import *
import logging.handlers
import threading
import time
SERVER_LOG_NAME = "LoggingServer"
class LogRecordStreamHandler(socketserver.StreamRequestHandler):
"""Read the LogRecord binary and process it."""
def handle(self):
"""make the LogRecord object from binary and process it."""
while True:
chunk = self.connection.recv(4)
if len(chunk) < 4:
break
slen = struct.unpack(">L",chunk)[0]
chunk = self.connection.recv(slen)
while len(chunk) < slen:
chunk = chunk + self.connection.recv(slen - len(chunk))
obj = self.unPickle(chunk)
record = logging.makeLogRecord(obj)
self.handleLogRecord(record)
def unPickle(self, data:bytes) -> Any:
return pickle.loads(data)
def handleLogRecord(self, record:logging.LogRecord) -> None:
"""process the LogRecord object."""
name = self.server.logname
logger = logging.getLogger(name)
logger.propagate = True
logger.handle(record)
class LoggingServer(socketserver.ThreadingTCPServer):
"""The SocketServer which receive Logs."""
allow_reuse_address = True
_server_thread:threading.Thread = None
def __init__(self,host='localhost',port=logging.handlers.DEFAULT_TCP_LOGGING_PORT,
handler=LogRecordStreamHandler):
super().__init__((host, port), handler)
self.timeout = 1
self.logname = SERVER_LOG_NAME
self.__shutdown = False
self.logger = logging.getLogger(SERVER_LOG_NAME)
def start(self):
"""Start server as daemon thread."""
self._server_thread = threading.Thread(target=self.serve_until_stopped)
self._server_thread.daemon = True
self.__shutdown = False
self._server_thread.start()
self.logger.info("About starting LoggingServer...")
def shutdown(self) -> None:
self.__shutdown = True
def serve_until_stopped(self):
import select
while not self.__shutdown:
rd, wr, ex = select.select([self.socket.fileno()], [], [], self.timeout)
if rd:
self.handle_request()
self.logger.info("LoggingServer stopped.")
def set_root_logger_components():
"""Add stdout to root logger"""
sh = logging.StreamHandler(sys.stdout)
root = logging.getLogger()
root.setLevel(logging.DEBUG)
root.addHandler(sh)
def get_socket_logger(
name, level:int = logging.NOTSET, host="localhost",
port = logging.handlers.DEFAULT_TCP_LOGGING_PORT
) -> logging.Logger:
assert name != SERVER_LOG_NAME # Avoid logging roop
logger = logging.getLogger(name)
logger.setLevel(level)
socket_handler = logging.handlers.SocketHandler(host, port)
logger.addHandler(socket_handler)
logger.propagate = False # Socket logger
return logger
if __name__ == "__main__":
set_root_logger_components()
server = LoggingServer()
server.start()
logger = get_socket_logger("SocketLogger")
logger.info("aaa")
time.sleep(0.1)
server.shutdown()
ソースコード実行後のコンソールアウトプット
終了しません。
About starting LoggingServer...
aaa
LoggingServer stopped.
試したこと
97行目のlogger.info("aaa")
コメントアウトすると正常に終了します。
lib/python3.9/threading.py
の1470行目lock.acquire
をコメントアウトすると終了しますが、良い方法ではありません。
ThreadingTCPServer
ではなく、TCPServer
を使用した場合は問題なく実行されます。
なぜ、どこで、どのように問題が発生しているのか全くわかりません。よろしくお願いいたします。