多处理 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

简而言之，您必须确保传递给池的参数是Picklable。

如果实例和类方法所属的对象/类也是可挑选的，则它们也是可挑选的。静态方法是不可腌制的，因为它们失去了与对象本身的关联，因此腌制库无法正确序列化它们。

一般来说，最好避免调度到multiprocessing.Pool 任何不同于顶级定义的函数。

Answer 2

要使用 Cherrypy 运行后台任务，最好使用异步任务队列管理器，例如 Celery 或 RQ。这些服务非常易于安装和运行，您的任务将在一个完全独立的进程中运行，如果您因为负载增加而需要扩展，这将非常简单。

您有一个简单的 Cherrypy 示例here。

Answer 3

我解决了将方法从 @staticmethod 更改为 @classmethod 的问题。现在作业在ProcessPool 内运行。正如here 所解释的那样，我发现类方法在这种情况下更有用。

谢谢。