Add new benchmark type -- readline

echo_client

@@ -40,6 +40,8 @@ if __name__ == '__main__':
     parser = argparse.ArgumentParser()
     parser.add_argument('--msize', default=1000, type=int,
                         help='message size in bytes')
+    parser.add_argument('--mpr', default=1, type=int,
+                        help='messages per request')
     parser.add_argument('--duration', '-T', default=30, type=int,
                         help='duration of test in seconds')
     parser.add_argument('--concurrency', default=3, type=int,
@@ -62,8 +64,9 @@ if __name__ == '__main__':
         addr = tuple(addr)
 
     MSGSIZE = args.msize
+    msg = (b'x' * (MSGSIZE - 1) + b'\n') * args.mpr
 
-    msg = b'x' * MSGSIZE
+    REQSIZE = MSGSIZE * args.mpr
 
     timeout = args.timeout * 1000
 
@@ -85,8 +88,8 @@ if __name__ == '__main__':
             req_start = time.monotonic()
             sock.sendall(msg)
             nrecv = 0
-            while nrecv < MSGSIZE:
-                resp = sock.recv(MSGSIZE)
+            while nrecv < REQSIZE:
+                resp = sock.recv(REQSIZE)
                 if not resp:
                     raise SystemExit()
                 nrecv += len(resp)
@@ -98,6 +101,11 @@ if __name__ == '__main__':
             latency_stats[req_time] += 1
             n += 1
 
+        try:
+            sock.close()
+        except OSError:
+            pass
+
         return n, latency_stats, min_latency, max_latency
 
     N = args.concurrency

run_benchmarks

@@ -40,6 +40,7 @@ tcp_client = echo_client + ['--addr={}'.format(tcp_address)]
 unix_client = echo_client + ['--addr={}'.format(unix_address)]
 http_client = ['./http_client', '--output-format=json',
                '--addr={}'.format(tcp_address)]
+readline_client = tcp_client + ['--mpr=5']
 
 benchmarks = [{
     'name': 'tcpecho-gevent-sockets',
@@ -53,6 +54,12 @@ benchmarks = [{
     'server': python + ['/usr/src/servers/torecho.py'],
     'server_address': tcp_address,
     'client': tcp_client,
+}, {
+    'name': 'readline-tornado',
+    'title': 'Streams readline (tornado)',
+    'server': python + ['/usr/src/servers/torecho_readline.py'],
+    'server_address': tcp_address,
+    'client': readline_client,
 }, {
     'name': 'tcpecho-twisted-protocol',
     'title': 'TCP echo server (twisted)',
@@ -66,11 +73,11 @@ benchmarks = [{
     'server_address': tcp_address,
     'client': tcp_client,
 }, {
-    'name': 'tcpecho-curio-streams',
-    'title': 'TCP echo server (curio/streams)',
+    'name': 'readline-curio',
+    'title': 'Streams readline (curio)',
     'server': python + ['/usr/src/servers/curioecho_streams.py'],
     'server_address': tcp_address,
-    'client': tcp_client,
+    'client': readline_client,
 }, {
     'name': 'tcpecho-nodejs-streams',
     'title': 'TCP echo server (nodejs)',
@@ -92,13 +99,13 @@ benchmarks = [{
     'server_address': tcp_address,
     'client': tcp_client,
 }, {
-    'name': 'tcpecho-asyncio-streams',
-    'title': 'TCP echo server (asyncio/streams)',
+    'name': 'readline-asyncio',
+    'title': 'Streams readline (asyncio)',
     'server': python + ['/usr/src/servers/asyncioecho.py',
                         '--addr=0.0.0.0:25000',
                         '--streams'],
     'server_address': tcp_address,
-    'client': tcp_client,
+    'client': readline_client,
 }, {
     'name': 'tcpecho-asyncio-sockets',
     'title': 'TCP echo server (asyncio/sockets)',
@@ -131,13 +138,13 @@ benchmarks = [{
     'server_address': tcp_address,
     'client': tcp_client,
 }, {
-    'name': 'tcpecho-uvloop-streams',
-    'title': 'TCP echo server (uvloop/streams)',
+    'name': 'readline-uvloop',
+    'title': 'Streams readline (uvloop)',
     'server': python + ['/usr/src/servers/asyncioecho.py',
                         '--addr=0.0.0.0:25000',
                         '--streams', '--uvloop'],
     'server_address': tcp_address,
-    'client': tcp_client,
+    'client': readline_client,
 }, {
     'name': 'tcpecho-uvloop-sockets',
     'title': 'TCP echo server (uvloop/sockets)',

servers/asyncioecho.py

@@ -54,7 +54,7 @@ async def echo_client_streams(reader, writer):
     if PRINT:
         print('Connection from', sock.getpeername())
     while True:
-         data = await reader.read(102400)
+         data = await reader.readline()
          if not data:
              break
          writer.write(data)
@@ -133,10 +133,12 @@ if __name__ == '__main__':
         print('using asyncio/streams')
         if unix:
             coro = asyncio.start_unix_server(echo_client_streams,
-                                             addr, loop=loop)
+                                             addr, loop=loop,
+                                             limit=1024 * 1024)
         else:
             coro = asyncio.start_server(echo_client_streams,
-                                        *addr, loop=loop)
+                                        *addr, loop=loop,
+                                        limit=1024 * 1024)
         srv = loop.run_until_complete(coro)
     elif args.proto:
         if args.streams:

servers/curioecho_streams.py

@@ -9,7 +9,7 @@ async def echo_handler(client, addr):
         pass
     s = client.as_stream()
     while True:
-        data = await s.read(102400)
+        data = await s.readline()
         if not data:
             break
         await s.write(data)

servers/torecho_readline.py

+from tornado.ioloop import IOLoop
+from tornado.tcpserver import TCPServer
+
+
+class StreamHandler:
+    def __init__(self, stream):
+        self._stream = stream
+        stream.set_nodelay(True)
+        self._stream.read_until(b'\n', self._handle_read)
+
+    def _handle_read(self, data):
+        self._stream.write(data)
+        self._stream.read_until(b'\n', self._handle_read)
+
+
+class EchoServer(TCPServer):
+    def handle_stream(self, stream, address):
+        StreamHandler(stream)
+
+
+if __name__ == '__main__':
+    server = EchoServer()
+    server.bind(25000)
+    server.start(1)
+    IOLoop.instance().start()
+    IOLoop.instance().close()