[Twisted-Python] transport.write performance.

[zipxing]

tsvr.py
-----------------------------------------------------------
import sys, time, random, socket, traceback
from twisted.internet import epollreactor
epollreactor.install()
from twisted.internet import defer, reactor, task
from twisted.internet.protocol import Protocol, Factory
from protocol import TCPServerProtocol

def main():
        tcpprotocol = TCPServerProtocol
        factory = Factory()
        factory.protocol = tcpprotocol
        reactor.listenTCP(9976, factory)
        reactor.run()

if __name__ == '__main__':
        main()

protocol.py
--------------------------------------------------------- 
import socket
import datetime
import traceback
from twisted.protocols.basic import LineReceiver
from twisted.internet import protocol

class TCPServerProtocol(LineReceiver):
        req_count = 0
        req_time = datetime.datetime.now()

        def lineReceived(self, data):
                TCPServerProtocol.req_count+=1
                if TCPServerProtocol.req_count%10000==0:
                        ct = datetime.datetime.now()
                        dt = ct-TCPServerProtocol.req_time
                        pps = 10000/(dt.seconds+dt.microseconds/1000000.0)
                        TCPServerProtocol.req_time=ct
                        print('RPS='+str(pps))
                try:
                        #self.transport.write(data)
                        self.transport.getHandle().send(data)
                except:
                        traceback.print_exc()

tcli.py 
-----------------------------------------------------------------
import sys
import socket
import traceback
import time
import datetime

host = 'localhost'
port = 9976
loopcount = 300
sockcount = 5000
RPS = 4000

ss=[]
for x in xrange(sockcount):
        ss.append(socket.socket(socket.AF_INET, socket.SOCK_STREAM))
        ss[x].connect((host, port))
        ss[x].settimeout(120)

for x in xrange(10000000):
        st = datetime.datetime.now()
        for y in xrange(loopcount):
                try:
                        if ss[x%sockcount]!=None:
                                ss[x%sockcount].sendall('1234567890\r\n')
                                ss[x%sockcount].recv(1024)
                except:
                        print y
                        sys.exit()
        time.sleep(0.1)
        dt = (datetime.datetime.now()-st)
        plc = loopcount/(dt.seconds+dt.microseconds/1000000.0)
        print loopcount/(dt.seconds+dt.microseconds/1000000.0)
        #auto adjust RPS
        if plc<RPS:
                if RPS-plc>50:
                        loopcount+=10
        else:
                if plc-RPS>50:
                        loopcount-=10

echosvr.c 
----------------------------------------------------------------------
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <assert.h>

#include <event2/event.h>
#include <event2/bufferevent.h>

#define LISTEN_PORT 9976
#define LISTEN_BACKLOG 32

#ifdef FD_SETSIZE
#undef FD_SETSIZE
#endif
#define FD_SETSIZE 65536

void do_accept(evutil_socket_t listener, short event, void *arg);
void read_cb(struct bufferevent *bev, void *arg);
void error_cb(struct bufferevent *bev, short event, void *arg);
void write_cb(struct bufferevent *bev, void *arg);

int main(int argc, char *argv[])
{
    int ret;
    evutil_socket_t listener;
    listener = socket(AF_INET, SOCK_STREAM, 0);
    assert(listener> 0);
    evutil_make_listen_socket_reuseable(listener);

    struct sockaddr_in sin;
    sin.sin_family = AF_INET;
    sin.sin_addr.s_addr = 0;
    sin.sin_port = htons(LISTEN_PORT);

    if (bind(listener, (struct sockaddr *)&sin, sizeof(sin)) < 0) {
        perror("bind");
        return 1;
    }

    if (listen(listener, LISTEN_BACKLOG) < 0) {
        perror("listen");
        return 1;
    }

    printf ("Listening...\n");

    evutil_make_socket_nonblocking(listener);

    struct event_base *base = event_base_new();
    assert(base != NULL);
    struct event *listen_event;
    listen_event = event_new(base, listener, EV_READ|EV_PERSIST, do_accept, (void*)base);
    event_add(listen_event, NULL);
    event_base_dispatch(base);

    printf("The End.");
    return 0;
}

void do_accept(evutil_socket_t listener, short event, void *arg)
{
    struct event_base *base = (struct event_base *)arg;
    evutil_socket_t fd;
    struct sockaddr_in sin;
    socklen_t slen;
    fd = accept(listener, (struct sockaddr *)&sin, &slen);
    if (fd < 0) {
        perror("accept");
        return;
    }
    if (fd> FD_SETSIZE) {
        perror("fd> FD_SETSIZE\n");
        return;
    }

    printf("ACCEPT: fd = %u\n", fd);

    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
    bufferevent_setcb(bev, read_cb, NULL, error_cb, arg);
    bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST);
}

void read_cb(struct bufferevent *bev, void *arg)
{
#define MAX_LINE    256
    char line[MAX_LINE+1];
    int n;
    evutil_socket_t fd = bufferevent_getfd(bev);

    while (n = bufferevent_read(bev, line, MAX_LINE), n> 0) {
        line[n] = '\0';
        //printf("fd=%u, read line: %s\n", fd, line);

        bufferevent_write(bev, line, n);
    }
}

void write_cb(struct bufferevent *bev, void *arg) {}

void error_cb(struct bufferevent *bev, short event, void *arg)
{
    evutil_socket_t fd = bufferevent_getfd(bev);
    printf("fd = %u, ", fd);
    if (event & BEV_EVENT_TIMEOUT) {
        printf("Timed out\n"); //if bufferevent_set_timeouts() called
    }
    else if (event & BEV_EVENT_EOF) {
        printf("connection closed\n");
    }
    else if (event & BEV_EVENT_ERROR) {
        printf("some other error\n");
    }
    bufferevent_free(bev);
}