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

Question

我有来自 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()

为什么是twisted，因为客户端/服务器已经用twisted编写了，我正在做一些小的改动。帮助表示赞赏。谢谢！

Answer 1

很难说出您的Unhandled error in Deferred 来自哪里，因为您的示例包含语法错误，但我会尝试使用我的直觉并重写您正在尝试做的事情:)。我已经做了一些 cmets，所以看看你的代码和这段代码有何不同。

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 为您提供了大量选项来在单个线程中高效地运行代码。