如何使扭曲延迟获得功能结果? [英] How to make twisted defer get function result?

问题描述

我想在 multiprocessing 中做点什么，我想推迟得到结果，像这样:

I want to do something in multiprocessing, and I want to defer get result, something like this:

from multiprocessing import Pool
from twisted.internet import defer
import time

def f(x):
    time.sleep(0.5)
    print(x)
    return x*x

pool = Pool(processes=4)              # start 4 worker processes
def get_result(i):
    res = pool.apply_async(f, (i, )) # do work in process pool
    return defer.Deferred(res.get()) # now, I want to make process do something else, so it should not be blocked

def main():
    from twisted.internet import reactor

    @defer.inlineCallbacks
    def _run():
        for i in range(4):
            yield get_result(i)

        reactor.stop()
    reactor.callLater(1, _run)
    reactor.run()


if __name__ == '__main__':
    main()

推荐答案

Pool.apply_async() 具有 callback arg，您可以利用它来启动 Deferred .要注意的是(记住这一点绝对至关重要):Pool-callback-function将在另一个线程中执行！！因此，将结果应用于 Deferred 时，必须调用 reactor.callFromThread ，以便回调链与 reactor 出现在同一线程中.否则，将导致在没有上下文的其他线程中执行回调.这是一个稍作修改的示例:

Pool.apply_async() has a callback arg that you can leverage to start the callback chain in the Deferred. The catch (which is absolutely crucial to remember) is that the Pool-callback-function will be executed in another thread! Therefore, you must call reactor.callFromThread when applying the result to the Deferred so that the callback chain occurs in the same thread as the reactor. Failure to do this will result in callbacks being executed in a different thread that the reactor has no context into. Here is a slightly modified example:

from functools import partial
from multiprocessing import Pool
import threading
import time

from twisted.internet import defer, reactor


def f(x):
    time.sleep(5)
    return x*x

def get_result(pool, i):
    deferred = defer.Deferred()   # create a Deferred that will eventually provide a result
    _set_result = partial(set_result, deferred=deferred)  # pass the Deferred to the apply_async callback
    pool.apply_async(f, args=(i,), callback=_set_result)  # execute in a separate process, supply callback fn
    return deferred

def set_result(result, deferred):
    """
    Set the result in the deferred
    """
    print('Thread ID: %d, Setting result %d' % (threading.get_ident(), result))
    reactor.callFromThread(deferred.callback, result)   # execute the Deferred callback chain from the reactor thread

def display_result(result):
    """
    Just display the result
    """
    print('Thread ID: %d, Display %d' % (threading.get_ident(), result))

def kill_reactor(null):
    print('Thread ID: %d, Stopping reactor' % threading.get_ident())
    reactor.stop()

def main():
    print('Thread ID: %d, Main' % threading.get_ident())
    pool = Pool(processes=4)
    d = get_result(pool, 3)
    d.addCallback(display_result)
    d.addCallback(kill_reactor)

    reactor.run()

main()


#---------- OUTPUT ----------#
# Thread ID: 803872, Main
# Thread ID: 533632, Setting result 9
# Thread ID: 803872, Display 9
# Thread ID: 803872, Stopping reactor

我已经打印了线程ID，以便您可以看到 set_result()确实是在另一个线程(进程?)中而不是在主线程(即反应堆线程)中调用的.在此示例中省略 reactor.callFromThread(deferred.callback，result)将导致回调在 reactor.stop()无法正常工作且Twisted抛出的线程中执行呕吐(回声)无处不在！考虑在 reactor.spawnProcess 中使用这将限制您(或我自己)否则会犯的错误.和往常一样，如果您可以执行自己想做的任何事情，请在一个线程中忽略 multiprocessing 或 threading ，我建议您改为这样做

I've printed the thread id so that you can see that set_result() is indeed called in another thread (process?) and not in the main thread, ie reactor thread. Ommiting reactor.callFromThread(deferred.callback, result) in this example will cause the callbacks to executed in a thread that reactor.stop() will not work and Twisted throws up vomit (tracebacks) everywhere! Consider using at reactor.spawnProcess as this will limit mistakes that you (or myself) would make otherwise. And as always, if you can do whatever it is you're tyring to do, in a single thread and omit multiprocessing or threading, I'd suggest you do that instead.

这篇关于如何使扭曲延迟获得功能结果?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！