Python 多处理日志记录错误 - TypeError：需要一个整数（获取类型 NoneType）

Question

我有一个我一直在从事的 python3 (3.4.5) 项目，它利用 multiprocessing.Pool 通过 4 名工人运行大约 50 多个工作。我有一个带有logging.handlers.QueueListener 的单独进程设置，因此我可以通过与multiprocessing.Manager() 一起使用的Queue 将全局内容记录到单个文件中。所以基本上流程是这样的

1. 主程序启动
2. 通过multiprocessing.Manager()创建Queue
3. 使用QueueListener 启动专用日志记录进程，监听我刚刚为全局日志创建的Queue。 （我也试过这个，只是使用主程序的一个线程，结果相同。）
4. 创建一个multiprocessing.Pool 来处理单个作业，将之前创建的Queue 和必要的配置信息传递给它们以运行和设置它们的日志记录（有一个全局日志，以及每个作业的单个日志，其中包含更详细的信息）。作业以map_async 开始。
5. 等待所有作业处理完毕，然后执行一些最后步骤并进行清理。

不过，我在某些作业中不断收到间歇性错误，通常其中 1 个作业有错误（每次都不同），但偶尔也会有 2 个相同的错误或零。据我所知，导致错误的不是作业中的代码，而是multiprocessing 或logging 设置中的某些内容。这是我遇到的错误的一个示例：

--- Logging error ---
Traceback (most recent call last):
  File "/usr/lib64/python3.4/logging/handlers.py", line 1347, in emit
    self.enqueue(self.prepare(record))
  File "/usr/lib64/python3.4/logging/handlers.py", line 1313, in enqueue
    self.queue.put_nowait(record)
  File "<string>", line 2, in put_nowait
  File "/usr/lib64/python3.4/multiprocessing/managers.py", line 731, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/usr/lib64/python3.4/multiprocessing/connection.py", line 206, in send
    self._send_bytes(ForkingPickler.dumps(obj))
  File "/usr/lib64/python3.4/multiprocessing/connection.py", line 413, in _send_bytes
    self._send(chunk)
  File "/usr/lib64/python3.4/multiprocessing/connection.py", line 369, in _send
    n = write(self._handle, buf)
TypeError: an integer is required (got type NoneType)
Call stack:
  File "./sampling__test__py.py", line 100, in <module>
    run_pool     = multiprocessing.Pool(4)
  File "/usr/lib64/python3.4/multiprocessing/context.py", line 118, in Pool
    context=self.get_context())
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 168, in __init__
    self._repopulate_pool()
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 233, in _repopulate_pool
    w.start()
  File "/usr/lib64/python3.4/multiprocessing/process.py", line 105, in start
    self._popen = self._Popen(self)
  File "/usr/lib64/python3.4/multiprocessing/context.py", line 267, in _Popen
    return Popen(process_obj)
  File "/usr/lib64/python3.4/multiprocessing/popen_fork.py", line 21, in __init__
    self._launch(process_obj)
  File "/usr/lib64/python3.4/multiprocessing/popen_fork.py", line 77, in _launch
    code = process_obj._bootstrap()
  File "/usr/lib64/python3.4/multiprocessing/process.py", line 254, in _bootstrap
    self.run()
  File "/usr/lib64/python3.4/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/usr/lib64/python3.4/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))
  File "/home/username/value_check.py", line 338, in value_check
    global_logger.info("SplitTime: {str_timeDelta} -- COMPLETED: {Check_Name} --- Total Txn Count: {var_Total_Txn_Count} --- Criteria Txn Count: {var_Criteria_Txn_Count} --- Threshold: {Threshold} --- Low_Vol Threshold: {LowVolThresh}".format(str_timeDelta = timeDelta(datetime.now() - YAML_Config['start_time']), **YAML_Config))
Message: 'SplitTime: 00:01:05,031 -- COMPLETED: ALPHA_CHECK --- Total Txn Count: 1234--- Criteria Txn Count: 0 --- Threshold: 10 --- Low_Vol Threshold: 0'
Arguments: None

代码中的错误指的是我的代码中的日志记录对象，但即使我在调用周围放置try/except 逻辑，它也没有做任何事情，错误似乎发生在上游。我还尝试将记录的内容从格式化的字符串更改为简单的字符串，但无济于事。似乎在某个地方，个别工作要么失去与Queue 的联系，要么Queue 中的某些东西失败并导致问题。

有什么想法吗？我一直在努力获得一个更新版本的 Python 可用，这将有很多原因（特别是 f 字符串），但我不知道这是否能解决这个问题，而且我已经用完了故障排除思路。

Answer 1

即使我在调用周围加上 try/except 逻辑，它也没有做任何事情。

这很可能是因为，如果日志包遇到与日志本身有关的异常，它会打印回溯，但不会引发异常本身。这在logging.Handler.handleError 的文档字符串中有更全面的解释。

一个起点是通过设置：

logging.raiseExceptions = True

如果模块级属性 raiseExceptions 为 False，异常会被忽略。

如果这没有帮助，您可以在.emit() 的代码中调用import pdb; pdb.set_trace()；类似：

def emit(self, record):
    try:
        msg = self.format(record)
        stream = self.stream
        stream.write(msg)
        stream.write(self.terminator)
        self.flush()
    except Exception as e:
        import pdb; pdb.set_trace()  # < ---
        self.handleError(record)

其中record 将是一个LogRecord 实例。通常当我看到一个日志记录错误弹出时，这是因为我为给定的格式字符串使用了错误数量的 args，但检查 record 对象应该希望能告诉你更多信息。

最后，从调用堆栈中，您的日志记录调用本身：

global_logger.info(
    "SplitTime: {str_timeDelta} -- "
    "COMPLETED: {Check_Name} --- "
    "Total Txn Count: {var_Total_Txn_Count} --- "
    "Criteria Txn Count: {var_Criteria_Txn_Count} --- "
    "Threshold: {Threshold} --- "
    "Low_Vol Threshold: {LowVolThresh}".format(
    str_timeDelta = timeDelta(datetime.now() - YAML_Config['start_time']), **YAML_Config))

很难准确说出最终引发异常的原因是什么，因为字符串似乎已完全格式化。 （虽然我们看不到YAML_Config。）

不管怎样，一个建议是：您可以利用日志记录的“惰性”字符串格式，而不是像现在这样使用str.format()。 str.format() 调用将尽快得到评估，而如果您将 kwargs 传递给 global_logger.info()，日志记录包将等待评估它们，直到必须进行评估。