python - 扭曲延迟 : client connects to a server, 异步单线程

Question

我们有一个异步单线程twisted/python 进程。有一个X服务器监听X端口。还有另一个服务器Y服务器监听Y端口。Y服务器也是X服务器的客户端(执行Y请求涉及将请求传递给X服务器) .

它们都应在一个线程中异步执行。它应该像下面这样工作:

• Y-服务器收到请求
• Y-服务器创建 X-客户端:打开与 X-服务器的连接并发送内部请求
• 同时(!!!)X服务器接受内部请求
• X-server 完成其工作并返回内部响应
• X 客户端(属于 Y 服务器)接受内部响应并将其作为 Y 响应进一步发送

我试图实现这样的事情，但我失败了，可能是因为没有使用延迟。据我了解 deferred，他们的工作是将上面的序列分成更小的 block ，以便这些部分可以由 X 和 Y 同时完成。

我需要的是了解这种沟通如何运作的方案。伪代码应该做...

<小时/>

下面是我失败的尝试的简短代码。

主要服务类由协议(protocol)/工厂类组成:

class PyCached(protocol.Protocol):
    def __init__(self, factory, verbose):
        self.factory = factory
        self.verbose = verbose
    def dataReceived(self, data):
        log.msg(data)
        if self.verbose:
            print 'received: %s' % (data,)
        request = json.loads(data)
        if self.verbose:
            print 'request: %s' % (request,)
        command = "handle_%s" % (request.pop('command'),)
        if self.verbose:
            print 'command: %s\n' % (command,)
        result = getattr(self.factory, command)(**request)
        self.transport.write(result + "\n")

class PyCachedFactory(protocol.Factory):
    def __init__(self, verbose=False):
        self.clear()
        self.start_time = datetime.now()
        self.verbose = verbose
        log.msg('starts on %s, verbose=%s' % (self.start_time, self.verbose))

    # many many more commands performed by factory

还有http访问服务器:

from twisted.web.resource import Resource
from twisted.python import log
from twisted.web.server import Site
from client import PyCachedClient

class PyCachedCommand(Resource):
    isLeaf = True

    def getServiceClient(self):
        client = PyCachedClient()
        client.connect(*self.service_address)
        return client

    def render_GET(self, request):
        '''
        Renders service status as plain text.
        '''
        log.msg('GET')
        request.setHeader('Content-Type', 'text/plain')
        try:
            client = self.getServiceClient()
            status = client.status()
            client.close()
            return "PyCached is up since %0.2f seconds" % (status['uptime'],)
        except:
            return "PyCached is down."

    def render_POST(self, request):
        '''
        Executes pycached request ad returns the response.
        '''
        log.msg('POST %s' % (str(request.args)))
        client = self.getServiceClient()
        kwargs = {k: v[0] for k,v in request.args.iteritems()}
        command_name = kwargs.pop('command')
        command = getattr(client, command_name)
        result = str(command(**kwargs))
        client.close()
        request.setHeader('Content-Type', 'text/plain')
        return result

class PyCachedSite(Site):
    '''
    Performs all operations for PyCached HTTP access.
    '''
    def __init__(self, service_address, **kwargs):
        resource = PyCachedCommand()
        resource.service_address = service_address
        Site.__init__(self, resource, **kwargs)

http 使用主服务客户端，它是通过简单的套接字实现的 - 这可能就是问题所在，因为这些客户端套接字调用是阻塞的:

class PyCachedClient(object):
    def __init__(self):
        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    def connect(self, host, port):
        try:
            self.s.connect((host, port))
        except socket.error:
            raise RuntimeError('Something went wrong with PyCached.')

    def close(self):
        self.s.close()

    def _receive(self):
        received = self.s.recv(1024)
        decoded = json.loads(received.rstrip('\n'))
        return decoded['value'] if decoded.has_key('value') else None

    def _send(self, command, options={}):
        request = {'command': command}
        request.update(options)
        self.s.sendall(json.dumps(request))

    def version(self):
        self._send('version')
        return self._receive()

    # many many more commands similar to version

最后，所有内容都由 twind/TAC 文件运行 - 因此它驻留在单个线程中:

from twisted.application import internet, service
from server.service import PyCachedFactory
from server.http import PyCachedSite

application = service.Application('pycached')
# pycached core service
pycachedService = internet.TCPServer(8001, PyCachedFactory())
pycachedService.setServiceParent(application)
# pycached http access
addr = ('localhost', 8001)
pycachedHttp = internet.TCPServer(8002, PyCachedSite(addr))
pycachedHttp.setServiceParent(application)

当我telnet 8001(主要服务)时，例如{"command":"version"}，一切正常。但是当我询问 http 时，一切都会阻塞，因为客户端套接字被阻塞并且主服务永远无法响应。

Answer 1

Deferred只是一个管理回调的工具。 X-client 必须提供某种方法来在收到结果时添加回调，并且您应该从此回调发送 Y-response。而 Deferred 只是一个实现细节。