Python：deferToThread XMLRPC 服务器 - Twisted - Cherrypy？

Question

这个问题与我在这里问的其他问题有关，主要是关于对内存中的大量数据进行排序。

基本上这就是我想要/拥有的：

Twisted XMLRPC 服务器正在运行。该服务器在内存中保存了几个 (32) Foo 类的实例。每个 Foo 类都包含一个列表栏（其中将包含数百万条记录）。有一个服务可以从数据库检索数据，并将其传递到 XMLRPC 服务器。数据基本上是一个字典，键对应于每个 Foo 实例，值是字典列表，如下所示：

data = {'foo1':[{'k1':'v1', 'k2':'v2'}, {'k1':'v1', 'k2':'v2'}], 'foo2':...}

然后向每个 Foo 实例传递与其键对应的值，并且 Foo.bar 字典被更新和排序。

class XMLRPCController(xmlrpc.XMLRPC):

    def __init__(self):
        ...
        self.foos = {'foo1':Foo(), 'foo2':Foo(), 'foo3':Foo()}
        ...

    def update(self, data):
        for k, v in data:
            threads.deferToThread(self.foos[k].processData, v)

    def getData(self, fookey):
        # return first 10 records of specified Foo.bar
        return self.foos[fookey].bar[0:10]

class Foo():

    def __init__(self):
        bar = []

    def processData(self, new_bar_data):
        for record in new_bar_data:
            # do processing, and add record, then sort
            # BUNCH OF PROCESSING CODE
            self.bar.sort(reverse=True)

问题是，当在 XMLRPCController 中调用包含大量记录（例如 100K 以上）的更新函数时，它会停止响应我的 getData 调用，直到所有 32 个 Foo 实例都完成 process_data 方法。我认为 deferToThread 会起作用，但我认为我误解了问题所在。

任何建议...我愿意使用其他东西，比如 Cherrypy，如果它支持这种必需的行为。

---

编辑

@Troy：这就是反应堆的设置方式

reactor.listenTCP(port_no, server.Site(XMLRPCController)
reactor.run()

就GIL而言，改变它是否是一个可行的选择 sys.setcheckinterval() 值变小，因此数据上的锁被释放以便可以读取？

Answer 1

让应用程序响应的最简单方法是将 CPU 密集型处理分解为较小的块，同时让扭曲的反应器在其间运行。例如，通过调用reactor.callLater(0, process_next_chunk) 前进到下一个块。有效地实现自己的协作多任务处理。

另一种方法是使用单独的进程来完成工作，然后您将受益于多个核心。看一下Ampoule： https://launchpad.net/ampoule 它提供了类似于deferToThread的API。

Answer 2

我不知道你的 processData 方法运行了多长时间，也不知道你如何设置你的扭曲反应器。 默认 ，扭曲的反应器有一个 0 到 10 个线程的线程池。您可能尝试将多达 32 个长时间运行的计算推迟到多达 10 个线程。这是次优的。

您还需要询问 GIL 在更新所有这些集合中扮演什么角色。

编辑：
在对程序进行任何重大更改（例如调用 sys.setcheckinterval()）之前，您可能应该使用探查器或 python 跟踪模块来运行它。这些应该告诉您您一直在使用哪些方法。如果没有正确的信息，您就无法做出正确的更改。