twisted + pypy 优雅与性能并存

原文链接 

最近迷上了Twisted。

本来是比较喜欢用gevent的，不过gevent不能和pypy很好地结合，在对性能要求比较高的地方有点慢，能明显感觉到建立连接的时候有卡顿。

后来嘛，试验了一下evenlet，因为其epoll部分没有依赖C的库或代码，可以在pypy下跑，但是在pypy下表现平平，估计原因是greenlet，做得太通用了，肯定就以性能为代价了，记得它的实现原理是拷贝栈和恢复栈。

再后来嘛，Twisted上阵了，做hello world测试的时候，如果用CPython驱动它，性能确实比gevent差很多，大概只有gevent的三分之二。上了pypy后直接飙升到gevent的两倍，和erlang的效率差不多。

再后来嘛，用twisted草草实现了一个垃圾级别的sock5代理服务器，用pypy来驱动的话，在1MB/s的传输速率下，CPU占用率和我用erlang写的版本基本一样。如果用CPython占用就高很多。

erlang也研究过一段时间，因为它对字符串处理的支持薄弱，所以暂时放弃了，不知到有木有大神对erlang的字符串处理有相关经验？如果能重拾erlang也不错的。

上个twisted实现的垃圾级别的sock5代理服务器代码。pipe部分实现得不好，没法作流量控制，以后再研究。

import struct, socket
from twisted.internet import reactor
from twisted.names import client as names_client
from twisted.internet.protocol import ServerFactory, ClientFactory, Protocol
from twisted.internet.endpoints import TCP4ServerEndpoint

class TCPPipe(Protocol):
    def __init__(self, client):
        self.client = client

    def connectionMade(self):
        try:
            self.client.fsm.send(self)
        except StopIteration:
            pass

    def dataReceived(self, data):
        self.client.transport.write(data)

    def connectionLost(self, reason):
        self.client.transport.loseConnection()


class TcpPipeFactory(ClientFactory):
    def __init__(self, client):
        self.client = client

    def buildProtocol(self, addr):
        return TCPPipe(self.client)

class Socks(Protocol):

    def __init__(self):
        self.fsm = self.state_version("")
        self.fsm.next()

    def connectionMade(self):
        pass

    def connectionLost(self, reason):
        print reason

    def dataReceived(self, data):
        try:
            self.fsm.send(data)
        except StopIteration:
            pass
        except:
            self.transport.loseConnection()

    def state_version(self, buff):
        while len(buff) < 1:
            buff += yield
        version, buff = ord(buff[0]), buff[1:]
        if version == 5:
            self.fsm = self.state_sock5(buff)
            self.fsm.next()
        else:
            self.transport.abortConnection()

    def state_sock5(self, buff):
        while len(buff) < 1:
            buff += yield
        ncommands, buff = ord(buff[0]), buff[1:]

        while len(buff) < ncommands:
            buff += yield
        commands, buff = buff[:ncommands], buff[ncommands:]
        if '\x00' not in commands:
            self.transport.write('\x05\x01\xFF')
            return
        self.transport.write('\x05\x00')

        while len(buff) < 4:
            buff += yield
        data, buff = buff[:4], buff[4:]
        if data[0] != '0x05' and data[2] != '\x00':
            self.transport.write('\x05\x07\x00')
            return
        command = data[1]
        atype = data[3]
        if atype == '\x01':
            while len(buff) < 6:
                buff += yield
            data, buff = buff[:6], buff[6:]
            target_IP, target_Port = struct.unpack('!4sH', data)
            target_IP = socket.inet_ntoa(target_IP)
        elif atype == '\x03':
            while len(buff) < 1:
                buff += yield
            hostnameLen, buff = ord(buff[:1]), buff[1:]
            while len(buff) < hostnameLen+2:
                buff += yield
            data, buff = buff[:hostnameLen+2], buff[hostnameLen+2:]
            target_hostname, target_Port = struct.unpack('!%dsH' % hostnameLen, data)
            self.transport.stopReading()
            d = names_client.getHostByName(target_hostname)
            d.addCallbacks(lambda ip: self.fsm.send(ip), lambda err: self.transport.abortConnection())
            target_IP = yield
            self.transport.startReading()
        else:
            self.transport.write('\x05\x08\x00')
            return
        if command == '\x01':
            print (target_IP, target_Port)
            self.transport.stopReading()
            reactor.connectTCP(target_IP, target_Port, TcpPipeFactory(self))
            remote = yield
            self.transport.startReading()
            remote.transport.write(buff)
            peer = remote.transport.getPeer()
            reply = '\x05\x00\x00\x01' + socket.inet_aton(peer.host) + struct.pack('!H', peer.port)
            self.transport.write(reply)
            self.remote = remote
            self.fsm = self.state_5_pipe(buff)
            self.fsm.next()
        else:
            client_socket.sendall('\x05\x07\x00')
            return

    def state_5_pipe(self, buff):
        remote = self.remote

        while True:
            buff = yield
            remote.transport.write(buff)


    def connectionLost(self, reason):
        if hasattr(self, "remote"):
            self.remote.transport.loseConnection()

class SocksFactory(ServerFactory):

    def buildProtocol(self, addr):
        return Socks()

endpoint = TCP4ServerEndpoint(reactor, 8888, backlog=10000)
endpoint.listen(SocksFactory())
reactor.run()