具有大量数据的多处理队列导致 _wait_for_tstate_lock

Question

当我通过multiprocessing.Queue 在Process 和Thread 之间传输大量数据时，threading._wait_for_tstate_lock 引发了异常。

我的最小工作示例首先看起来有点复杂 - 抱歉。我会解释。原始应用程序将大量（不那么重要）文件加载到 RAM 中。这是在一个单独的过程中完成的以节省资源。主 gui 线程不应冻结。

1. GUI 启动一个单独的Thread 以防止 gui 事件循环冻结。

2. 这个单独的Thread 然后启动一个Process，它应该可以完成工作。

a) 这个Thread 实例化了一个multiprocess.Queue（请注意这是一个multiprocessing 而不是threading！）

b) 这是给Process 用于将数据从Process 共享回Thread。

1. Process 做了一些工作（3 个步骤），.put() 将结果输入到multiprocessing.Queue。

2. 当Process 结束时Thread 再次接管并从Queue 收集数据，将其存储到自己的属性MyThread.result。

3. Thread 告诉 GUI 主循环/线程在有时间时调用回调函数。

4. 回调函数（MyWindow::callback_thread_finished()）从MyWindow.thread.result获取结果。

问题是如果放到Queue 的数据太大了，我不明白会发生什么——MyThread 永远不会结束。我必须通过 Strg+C 取消申请。

我从文档中得到了一些提示。但我的问题是我没有完全理解文档。但我有一种感觉，我的问题的关键可以在那里找到。 请参阅“Pipes and Queues”（Python 3.5 文档）中的两个红色方框。 这是完整的输出

MyWindow::do_start()
Running MyThread...
Running MyProcess...
MyProcess stoppd.
^CProcess MyProcess-1:
Exception ignored in: <module 'threading' from '/usr/lib/python3.5/threading.py'>
Traceback (most recent call last):
  File "/usr/lib/python3.5/threading.py", line 1288, in _shutdown
    t.join()
  File "/usr/lib/python3.5/threading.py", line 1054, in join
    self._wait_for_tstate_lock()
  File "/usr/lib/python3.5/threading.py", line 1070, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt
Traceback (most recent call last):
  File "/usr/lib/python3.5/multiprocessing/process.py", line 252, in _bootstrap
    util._exit_function()
  File "/usr/lib/python3.5/multiprocessing/util.py", line 314, in _exit_function
    _run_finalizers()
  File "/usr/lib/python3.5/multiprocessing/util.py", line 254, in _run_finalizers
    finalizer()
  File "/usr/lib/python3.5/multiprocessing/util.py", line 186, in __call__
    res = self._callback(*self._args, **self._kwargs)
  File "/usr/lib/python3.5/multiprocessing/queues.py", line 198, in _finalize_join
    thread.join()
  File "/usr/lib/python3.5/threading.py", line 1054, in join
    self._wait_for_tstate_lock()
  File "/usr/lib/python3.5/threading.py", line 1070, in _wait_for_tstate_lock
    elif lock.acquire(block, timeout):
KeyboardInterrupt

这是最小的工作示例

#!/usr/bin/env python3

import multiprocessing
import threading
import time
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from gi.repository import GLib


class MyThread (threading.Thread):
    """This thread just starts the process."""
    def __init__(self, callback):
        threading.Thread.__init__(self)
        self._callback = callback

    def run(self):
        print('Running MyThread...')
        self.result = []

        queue = multiprocessing.Queue()
        process = MyProcess(queue)
        process.start()
        process.join()

        while not queue.empty():
            process_result = queue.get()
            self.result.append(process_result)
        print('MyThread stoppd.')
        GLib.idle_add(self._callback)


class MyProcess (multiprocessing.Process):
    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue

    def run(self):
        print('Running MyProcess...')
        for i in range(3):
            self.queue.put((i, 'x'*102048))
        print('MyProcess stoppd.')

class MyWindow (Gtk.Window):
    def __init__(self):
        Gtk.Window.__init__(self)
        self.connect('destroy', Gtk.main_quit)
        GLib.timeout_add(2000, self.do_start)

    def do_start(self):
        print('MyWindow::do_start()')
        # The process need to be started from a separate thread
        # to prevent the main thread (which is the gui main loop)
        # from freezing while waiting for the process result.
        self.thread = MyThread(self.callback_thread_finished)
        self.thread.start()

    def callback_thread_finished(self):
        result = self.thread.result
        for r in result:
            print('{} {}...'.format(r[0], r[1][:10]))

if __name__ == '__main__':
    win = MyWindow()
    win.show_all()
    Gtk.main()

可能重复，但完全不同，IMO 对我的情况没有答案：Thread._wait_for_tstate_lock() never returns。

解决方法

使用 Manager 通过将第 22 行修改为 queue = multiprocessing.Manager().Queue() 来解决问题。但我不知道为什么。我提出这个问题的目的是了解背后的东西，而不仅仅是让我的代码工作。即使我真的不知道Manager() 是什么以及它是否有其他（导致问题的）含义。

Answer 1

根据您链接到的文档中的第二个警告框，当您在处理队列中的所有项目之前加入进程时，可能会出现死锁。因此，启动流程并立即加入它并然后处理队列中的项目是错误的步骤顺序。您必须启动该过程，然后接收项目，然后只有在收到所有项目后才能调用 join 方法。定义一些标记值来表示进程已完成通过队列发送数据。 None 例如，如果这不是您期望从流程中获得的常规值。

class MyThread(threading.Thread):
    """This thread just starts the process."""

    def __init__(self, callback):
        threading.Thread.__init__(self)
        self._callback = callback
        self.result = []

    def run(self):
        print('Running MyThread...')
        queue = multiprocessing.Queue()
        process = MyProcess(queue)
        process.start()
        while True:
            process_result = queue.get()
            if process_result is None:
                break
            self.result.append(process_result)
        process.join()
        print('MyThread stoppd.')
        GLib.idle_add(self._callback)


class MyProcess(multiprocessing.Process):

    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue

    def run(self):
        print('Running MyProcess...')
        for i in range(3):
            self.queue.put((i, 'x' * 102048))
        self.queue.put(None)
        print('MyProcess stoppd.')