多处理apply_async()不适用于Ubuntu

Question

我在Mac OS X和Ubuntu 14.04上运行此代码作为CherryPy Web服务.通过multiprocessing在python3上使用我想以worker()异步方式启动静态方法Process Pool.

相同的代码在Mac OS X上运行完美,在Ubuntu 14.04 worker()中无法运行.即通过调试POST方法内部的代码,我可以看到每一行都被执行 - 来自

reqid = str(uuid.uuid4())

至

return handle_error(202, "Request ID: " + reqid)

在Ubuntu 14.04中启动相同的代码,它不运行该worker()方法,甚至不在方法print()的顶部(将被记录).

这是相关代码(我只省略了handle_error()方法):

import cherrypy
import json
from lib import get_parameters, handle_error
from multiprocessing import Pool
import os
from pymatbridge import Matlab
import requests
import shutil
import uuid
from xml.etree import ElementTree

class Schedule(object):
    exposed = True

    def __init__(self, mlab_path, pool):
        self.mlab_path = mlab_path
        self.pool = pool

    def POST(self, *paths, **params):

        if validate(cherrypy.request.headers):

            try:
                reqid = str(uuid.uuid4())
                path = os.path.join("results", reqid)
                os.makedirs(path)
                wargs = [(self.mlab_path, reqid)]
                self.pool.apply_async(Schedule.worker, wargs)

                return handle_error(202, "Request ID: " + reqid)
            except:
                return handle_error(500, "Internal Server Error")
        else:
            return handle_error(401, "Unauthorized")

    #### this is not executed ####
    @staticmethod
    def worker(args):

        mlab_path, reqid = args
        mlab = Matlab(executable=mlab_path)
        mlab.start()

        mlab.run_code("cd mlab")
        mlab.run_code("sched")
        a = mlab.get_variable("a")

        mlab.stop()

        return reqid

    ####

# to start the Web Service
if __name__ == "__main__":

    # start Web Service with some configuration
    global_conf = {
           "global":    {
                            "server.environment": "production",
                            "engine.autoreload.on": True,
                            "engine.autoreload.frequency": 5,
                            "server.socket_host": "0.0.0.0",
                            "log.screen": False,
                            "log.access_file": "site.log",
                            "log.error_file": "site.log",
                            "server.socket_port": 8084
                        }
    }
    cherrypy.config.update(global_conf)
    conf = {
        "/": {
            "request.dispatch": cherrypy.dispatch.MethodDispatcher(),
            "tools.encode.debug": True,
            "request.show_tracebacks": False
        }
    }

    pool = Pool(3)

    cherrypy.tree.mount(Schedule('matlab', pool), "/sched", conf)

    # activate signal handler
    if hasattr(cherrypy.engine, "signal_handler"):
        cherrypy.engine.signal_handler.subscribe()

    # start serving pages
    cherrypy.engine.start()
    cherrypy.engine.block()

Answer 1

你的逻辑是隐藏你的问题.该apply_async方法返回一个AsyncResult对象,该对象充当您刚刚安排的异步任务的处理程序.当您忽略计划任务的结果时,整个事情看起来像是"无声地失败".

如果您尝试从该任务获得结果,您将看到真正的问题.

handler = self.pool.apply_async(Schedule.worker, wargs)
handler.get()

... traceback here ...
cPickle.PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

简而言之,您必须确保传递给Pool的参数是Picklable.

如果它们所属的对象/类也是可选择的,则实例和类方法是Picklable.静态方法不可选,因为它们会松散与对象本身的关联,因此pickle库无法正确地序列化它们.

作为一般线,最好避免调度到multiprocessing.Pool与顶级定义函数不同的任何东西.