扭曲的 deferToThread 中 Deferred 中未处理的错误

Question

我有 client/server 来自 twisted 的代码示例。现在，我的要求是，当从客户端调用服务器时 - 服务器将调用推迟到一个线程，该线程实际上回复客户端并且服务器可以做其他事情。简而言之，假设客户端 C1 使用模板 Temp1 调用服务器 S1。服务器将其推迟到线程 T1。 T1 现在必须处理功能 A、B 和 C，最后返回给客户端 C1。下面是我的服务器代码。

我是 twisted 的新手，我收到错误：Deferred 中未处理的错误：

from twisted.internet import reactor, protocol, threads

def foo():
    time.sleep(5)
    print('Hello how are you!!!!')

def print_result():
    print('Processing done!!')

def onError():
    print('Error!!!!')

class Echo(protocol.Protocol):
    """This is just about the simplest possible protocol"""
    def process_func(self, data):
        print('hello i am in process_func!!!')
        self.transport.write(data)
        return foo()

    def onErrorfunc(self):
        onError()

    def onProcessDone(self):
        print_result()

    def dataReceived(self, data):
        "As soon as any data is received, write it back."
        # thr = threading.Thread(target=foo, args=(), kwargs={})
        # thr.start()
        d = threads.deferToThread(self.process_func, *data)
        d.addCallback(self.onProcessDone)
        d.addErrback(self.onErrorfunc)
        # do something else here
        # self.transport.write(data)

def main():
    """This runs the protocol on port 8000"""
    factory = protocol.ServerFactory()
    factory.protocol = Echo
    reactor.listenTCP(8000,factory)
    reactor.run()

# this only runs if the module was *not* imported
if __name__ == '__main__':
    main()

为什么要扭曲，因为 client/server 已经用扭曲的方式编写，我正在做一些小改动。感谢帮助。谢谢！

Answer 1

很难说出你的 Unhandled error in Deferred 的来源，因为你的例子已经结束运行并且有语法错误，但我会尝试使用我的直觉并重写你正在尝试做的事情:) .我发表了一些评论，所以请看一下您的代码和这段代码有何不同。

import time
from twisted.internet import reactor, protocol, threads

def foo():
    # this function didn't return anything in your example
    # now it returns a string
    time.sleep(5)
    return 'Hello how are you!!!!'

class Echo(protocol.Protocol):
    def process_func(self, data):
        # data is unused here, typically you would "do something" to data in a thread
        # remember data is a bytes type not string!
        print('hello i am in process_func!!!')
        return foo()

    def onErrorfunc(self, failure):
        print('Error: {0}'.format(failure.value))

    def onProcessDone(self, result):
        # result is the string returned from process_func()
        # if Python version >= 3 then transport.write arg must be bytes
        self.transport.write(result.encode('utf8'))
        print('Processing done!!')

    def dataReceived(self, data):
        d = threads.deferToThread(self.process_func, data)
        d.addCallback(self.onProcessDone)
        d.addErrback(self.onErrorfunc)

尽量不要在线程中使用 self.transport.write()，因为它是使用 Twisted reactor 安排的。而是在线程中的计算完成后在回调中运行它。线程应该只用于密集计算，因为 Twisted 为您提供了大量选项，可以在单个线程中高效地运行编码。

扭曲的 deferToThread 中 Deferred 中未处理的错误

Unhandled error in Deferred in twisted deferToThread

python

twisted

twisted.web

twisted.internet