Matplotlib和多处理RuntimeError

Question

我正在尝试一起使用多处理和matplotlib.

我正在创建一个标准Pool,添加工作apply_async并使用apply_async回调函数更新GUI ,该函数在Pool的父进程上运行(我验证了这一点os.getpid()).示例:

from pylab import *
from numpy import *
from numpy.random import random
from multiprocessing import Pool

# Output image
global out_all
out_all = zeros((256, 256))

# Only does something to in_image, doesn't access anything else
def do_work(in_image):
    for x in xrange(100000):
        out_image = in_image[::-1, ::-1]
    return out_image

# Update the output image and display if needed
def do_update(out_image):
    global out_all
    print ("Updating")
    out_all += out_image
    clf()
    imshow(out_all)
    show()

# Input images (close enough to what I do as well)
work = [random((256, 256)) for f in range(20)]

# Don't block when showing something
ion()

# Do the work
print "Starting pool"
pool = Pool()
for o in work:
    pool.apply_async(do_work, [o], callback=do_update).get()
pool.close()
pool.join()
print "Stopping pool"

# Block
ioff()
show()
print "Done"

处理本身工作正常,过程真的被破坏了pool.join(),但Matplotlib(和我猜)我一旦尝试做某事就会抱怨,甚至只是退出程序:

Traceback (most recent call last):
  File "test_thread.py", line 27, in <module>
    show()
  File "/usr/lib/pymodules/python2.7/matplotlib/pyplot.py", line 139, in show
    _show(*args, **kw)
  File "/usr/lib/pymodules/python2.7/matplotlib/backend_bases.py", line 83, in __call__
    manager.show()
  File "/usr/lib/pymodules/python2.7/matplotlib/backends/backend_tkagg.py", line 444, in show
    self.canvas.draw_idle()
  File "/usr/lib/pymodules/python2.7/matplotlib/backends/backend_tkagg.py", line 258, in draw_idle
    self._idle_callback = self._tkcanvas.after_idle(idle_draw)
  File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 512, in after_idle
    return self.after('idle', func, *args)
  File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 504, in after
    name = self._register(callit)
  File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 1101, in _register
    self.tk.createcommand(name, f)
RuntimeError: main thread is not in main loop
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/usr/lib/python2.7/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/usr/lib/pymodules/python2.7/matplotlib/_pylab_helpers.py", line 82, in destroy_all
    manager.destroy()
  File "/usr/lib/pymodules/python2.7/matplotlib/backends/backend_tkagg.py", line 452, in destroy
    self.canvas._tkcanvas.after_cancel(self.canvas._idle_callback)
  File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 519, in after_cancel
    data = self.tk.call('after', 'info', id)
RuntimeError: main thread is not in main loop
Error in sys.exitfunc:
Traceback (most recent call last):
  File "/usr/lib/python2.7/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/usr/lib/pymodules/python2.7/matplotlib/_pylab_helpers.py", line 82, in destroy_all
    manager.destroy()
  File "/usr/lib/pymodules/python2.7/matplotlib/backends/backend_tkagg.py", line 452, in destroy
    self.canvas._tkcanvas.after_cancel(self.canvas._idle_callback)
  File "/usr/lib/python2.7/lib-tk/Tkinter.py", line 519, in after_cancel
    data = self.tk.call('after', 'info', id)
RuntimeError: main thread is not in main loop

我的第一个想法是传统知识背景在每一个都是重复的fork(),它在某个过程中以某种方式干扰了TK循环,但我在工人中没有做任何与传统知识相关的事情.有任何想法吗？

Answer 1

错误消息引用Tkinter.所以看起来你正在使用TkAgg后端.后面的代码是特定于TkAgg/Tkinter的.特别是电话

win.after(100, animate)

利用特定于Tkinter的after方法.对GtkAgg/PyGtk有一个类似的调用,对于其他后端也是如此.但我只想强调下面是TkAgg/Tkinter特有的.

Tkinter旨在在单个线程中运行.也就是说,所有Tkinter GUI调用都应该来自单个线程(通常不一定是主线程).

Pool的apply_async回调方法_handle_results在主进程中的一个单独的()线程中运行.因为imshow()从Pool的_handle_results线程show()调用并在主线程中调用,所以Tkinter抱怨

RuntimeError: main thread is not in main loop

apply_async在这种情况下我没有看到使用回调的方法.

取而代之的是,我们所能做的就是安排do_work把out_image一个multiprocessing.Queue()(我称之为out_queue在下面的代码).然后我们将主进程的主线程轮询此队列的项目,并在它们从队列中出来时显示它们.此轮询在animate下面的函数中完成.

plt.ion()仅用于交互式会话.虽然有时可以编写一些似乎可以使用的小脚本,但plt.ion()如果您拒绝plt.ion()在脚本中使用而是编写尊重GUI框架事件循环的代码,那么您将获得更好的结果和更清晰的GUI .

虽然可能修复你的脚本和使用plt.ion(),因为这不是编写matplotlib脚本的推荐方法,让我们看看我们是否可以避免这样做.

plt.show()告诉Tkinter运行它的事件循环.请注意,一旦进行此调用,将绘制GUI窗口,您可以单击按钮,放大和缩小等.

不知何故,我们需要将一个函数注入到这个事件循环中,由事件循环定期运行,并与可能发生的所有其他GUI事件协同工作.我们希望此函数检查我们的任何工作子进程是否为我们输出,如果有,则更新imshow映像.

使用TkAgg/Tkinter,注入这样一个函数的方法是

win = fig.canvas.manager.window
win.after(100, animate)

这将告诉Tkinter animate在(大约)100ms过去之后运行该函数(一次).由于我们希望函数animate定期运行,我们只是坚持另一个

win.after(100, animate)

打电话到最后animate.

import matplotlib as mpl
mpl.use('TkAgg')
import matplotlib.pyplot as plt
import numpy as np
import multiprocessing as mp
import logging
import Queue
logger = mp.log_to_stderr(logging.INFO)

# Only does something to in_image, doesn't access anything else
def do_work(in_image):
    logger.info('Processing in_image')
    for x in xrange(100000):
        out_image = in_image[::-1, ::-1]
    out_queue.put(out_image)

# Update the output image and display if needed
out_all = np.zeros((256, 256))


def pool_initializer(out_queue_):
    # Setup out_queue as a global variable *in the worker subprocesses*
    global out_queue
    out_queue = out_queue_


def animate():
    global out_all
    try:
        out_image = out_queue.get_nowait()
    except Queue.Empty:
        pass
    else:
        logger.info("Updating")
        out_all += out_image
        im.set_data(out_all)
        fig.canvas.draw()  # redraw the canvas
    win.after(100, animate)

if __name__ == '__main__':
    out_queue = mp.Queue()
    logger.info("Starting pool")
    pool = mp.Pool(initializer=pool_initializer, initargs=(out_queue, ))
    work = [np.random.random((256, 256)) for f in range(20)]
    for o in work:
        pool.apply_async(do_work, [o])
    pool.close()

    fig, ax = plt.subplots()
    win = fig.canvas.manager.window
    # Output image
    im = plt.imshow(out_all, vmin=0, vmax=1)

    # Register a function to be run once
    win.after(100, animate)
    plt.show()
    logger.info("Done")