python - 多处理 apply_async() 在 Ubuntu 上不起作用

Question

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

相同的代码在 Mac OS X 上完美运行，在 Ubuntu 14.04 中 worker() 无法运行。 IE。通过调试 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 调度到与顶级定义函数不同的任何内容。