TwistedWeb on multicore / multiprocessor

Posted 2021-03-27

人们在运行TwistedWeb服务器时使用了哪些技术来利用多个处理器/内核？有推荐的方法吗？

我基于twisted.web的Web服务在Amazon EC2实例上运行，该实例通常具有多个CPU核心（8、16），并且该服务正在执行的工作类型得益于额外的处理能力，所以我非常想使用它。

我知道可以在多个Twisted实例之前使用haproxy，squid或配置为反向代理的Web服务器。实际上，我们当前正在使用这样的设置，其中nginx充当在同一主机上运行但在不同端口上运行的多个上游twisted.web服务的反向代理。

这很好，但是我真正感兴趣的是一种解决方案，其中没有“前端”服务器，但是所有扭曲的进程都以某种方式绑定到同一套接字并接受请求。这样的事情甚至有可能...还是我疯了？操作系统是Linux（CentOS）。

谢谢。

安东。

答案

有多种方法可以支持Twisted应用程序的多进程操作。但是，一开始要回答的一个重要问题是您期望的并发模型是什么，以及您的应用程序如何处理共享状态。

在单个进程Twisted应用程序中，并发是所有协作的（在Twisted的异步I / O API的帮助下，并且共享状态可以保存在Python对象可以使用的任何地方。您的应用程序代码会运行，但要知道，直到放弃控制，其他任何东西都不会运行。另外，您的应用程序中想要访问某些共享状态的任何部分都可以很容易地做到这一点，因为该状态可能保存在易于访问的无聊的旧Python对象中。

[当您有多个进程时，即使它们都在运行基于Twisted的应用程序，也有两种并发形式。一个与前面的情况相同-在特定过程中，并发是协作的。但是，您有一种新的类型，其中正在运行多个进程。您平台的进程调度程序可能会随时在这些进程之间切换执行，并且您对此几乎没有控制权（也几乎看不到它何时发生）。它甚至可能安排您的两个进程同时在不同的内核上运行（这甚至可能是您所希望的）。这意味着您无法保证一致性，因为一个进程不知道何时会出现第二个进程并尝试在某些共享状态下运行。这导致了另一个需要考虑的重要领域，即如何在流程之间实际共享状态。

与单一流程模型不同，您不再拥有任何方便且易于访问的位置来存储所有代码可以到达的状态。如果将其放在一个进程中，则该进程中的所有代码都可以像普通的Python对象一样轻松地访问它，但是在您的任何其他进程中运行的任何代码都不再可以轻松访问它。您可能需要找到一个RPC系统，以使您的进程相互通信。或者，您可以设计流程划分，以便每个流程仅接收需要存储在该流程中的状态的请求。这样的示例可能是带有会话的网站，其中有关用户的所有状态都存储在他们的会话中，并且他们的会话由cookie标识。前端进程可以接收Web请求，检查cookie，查找哪个后端进程负责该会话，然后将请求转发到该后端进程。这种方案意味着后端通常不需要进行通信（只要您的Web应用程序足够简单，即只要用户彼此之间不交互或对共享数据进行操作即可）。

请注意，在该示例中，预分叉模型不合适。前端进程必须专门拥有侦听端口，以便它可以在后端进程处理所有传入请求之前对其进行检查。

当然，有许多类型的应用程序，还有许多其他用于管理状态的模型。选择正确的模型进行多处理需要首先了解哪种并发对您的应用程序有意义，以及如何管理应用程序的状态。

话说回来，对于非常新的Twisted版本（目前尚未发布），在多个进程之间共享侦听TCP端口非常容易。这是一个代码段，演示了您可以使用一些新API来完成此操作的一种方法：

from os import environ
from sys import argv, executable
from socket import AF_INET

from twisted.internet import reactor
from twisted.web.server import Site
from twisted.web.static import File

def main(fd=None):
    root = File("/var/www")
    factory = Site(root)

    if fd is None:
        # Create a new listening port and several other processes to help out.                                                                     
        port = reactor.listenTCP(8080, factory)
        for i in range(3):
            reactor.spawnProcess(
                    None, executable, [executable, __file__, str(port.fileno())],
                childFDs={0: 0, 1: 1, 2: 2, port.fileno(): port.fileno()},
                env=environ)
    else:
        # Another process created the port, just start listening on it.                                                                            
        port = reactor.adoptStreamPort(fd, AF_INET, factory)

    reactor.run()


if __name__ == '__main__':
    if len(argv) == 1:
        main()
    else:
        main(int(argv[1]))

使用旧版本，有时可以不使用fork共享端口。但是，这很容易出错，在某些平台上会失败，并且不是使用Twisted的受支持方式：

from os import fork

from twisted.internet import reactor
from twisted.web.server import Site
from twisted.web.static import File

def main():
    root = File("/var/www")
    factory = Site(root)

    # Create a new listening port
    port = reactor.listenTCP(8080, factory)

    # Create a few more processes to also service that port
    for i in range(3):
        if fork() == 0:
            # Proceed immediately onward in the children.
            # The parent will continue the for loop.
            break

    reactor.run()


if __name__ == '__main__':
    main()

这是由于fork的正常行为而起作用的，在该行为中，新创建的进程（子进程）继承了原始进程（父进程）的所有内存和文件描述符。由于进程是隔离的，所以这两个进程不会相互干扰，至少就它们正在执行的Python代码而言不会。由于文件描述符是继承的，因此父级或任何子级都可以接受端口上的连接。

由于转发HTTP请求是一项简单的任务，因此我怀疑您会注意到使用这些技术中的任何一种都能大大提高性能。前者比代理要好一些，因为它简化了部署并更轻松地用于非HTTP应用程序。后者可能是更多的责任，不值得接受。

另一答案

IMO推荐的使用方式就像您已经使用haproxy（或其他负载平衡器）一样，如果配置正确，则瓶颈不应成为负载平衡器。此外，您将需要一种haproxy提供的故障转移方法，以防您的进程之一发生故障。

不可能将多个进程绑定到同一个TCP套接字，但是可以使用UDP。

另一答案

如果您也希望通过HTTPS提供Web内容，这是在@ Jean-Paul的代码片段之上要做的事情。

from twisted.internet.ssl import PrivateCertificate
from twisted.protocols.tls import TLSMemoryBIOFactory

'''
Original snippet goes here
..........
...............
'''

privateCert = PrivateCertificate.loadPEM(open('./server.cer').read() + open('./server.key').read())
tlsFactory = TLSMemoryBIOFactory(privateCert.options(), False, factory)
reactor.adoptStreamPort(fd, AF_INET, tlsFactory)

通过使用fd，您将提供HTTP或HTTPS，但不能同时提供。如果希望两者都包含，请在父进程上添加listenSSL，并在生成子进程时将从ssl端口获得的ssl fd作为第二个参数。

完整的狙击手在这里：

from os import environ
from sys import argv, executable
from socket import AF_INET

from twisted.internet import reactor
from twisted.web.server import Site
from twisted.web.static import File

from twisted.internet import reactor, ssl
from twisted.internet.ssl import PrivateCertificate
from twisted.protocols.tls import TLSMemoryBIOFactory

def main(fd=None, fd_ssl=None):
    root = File("/var/www")
    factory = Site(root)

    spawned = []
    if fd is None:
        # Create a new listening port and several other processes to help out.                                                                     
        port = reactor.listenTCP(8080, factory)
        port_ssl = reactor.listenSSL(8443, factory, ssl.DefaultOpenSSLContextFactory('./server.key', './server.cer'))
        for i in range(3):
            child = reactor.spawnProcess(
                None, executable, [executable, __file__, str(port.fileno()), str(port_ssl.fileno())],
                childFDs={0: 0, 1: 1, 2: 2, port.fileno(): port.fileno(), port_ssl.fileno(): port_ssl.fileno()},
                env=environ)
            spawned.append(child)
    else:
        # Another process created the port, just start listening on it.                                                                            
        port = reactor.adoptStreamPort(fd, AF_INET, factory)
        cer = open('./server.cer')
        key = open('./server.key')
        pem_data = cer.read() + key.read()
        cer.close()
        pem.close()
        privateCert = PrivateCertificate.loadPEM(pem_data )
        tlsFactory = TLSMemoryBIOFactory(privateCert.options(), False, factory)
        reactor.adoptStreamPort(fd_ssl, AF_INET, tlsFactory)

    reactor.run()

    for p in spawned:
        p.signalProcess('INT')


if __name__ == '__main__':
    if len(argv) == 1:
        main()
    else:
        main(int(argv[1:]))