Fix bugs in Future; add benchmarks

Author: 1st1

examples/bench/client.py

@@ -0,0 +1,40 @@
+# Copied with minimal modifications from curio
+# https://github.com/dabeaz/curio
+
+from concurrent.futures import ProcessPoolExecutor
+
+from socket import *
+import time
+import sys
+
+if len(sys.argv) > 1:
+    MSGSIZE = int(sys.argv[1])
+else:
+    MSGSIZE = 1000
+
+msg = b'x'*MSGSIZE
+
+def run_test(n):
+    print('Sending', NMESSAGES, 'messages')
+    sock = socket(AF_INET, SOCK_STREAM)
+    sock.connect(('localhost', 25000))
+    while n > 0:
+        sock.sendall(msg)
+        nrecv = 0
+        while nrecv < MSGSIZE:
+            resp = sock.recv(MSGSIZE)
+            if not resp:
+                raise SystemExit()
+            nrecv += len(resp)
+        n -= 1
+
+N = 3
+NMESSAGES = 200000
+start = time.time()
+with ProcessPoolExecutor(max_workers=N) as e:
+    for _ in range(N):
+        e.submit(run_test, NMESSAGES)
+end = time.time()
+duration = end-start
+print(NMESSAGES*N,'in', duration)
+print(NMESSAGES*N/duration, 'requests/sec')

examples/bench/server.py

@@ -0,0 +1,51 @@
+import argparse
+import asyncio
+import uvloop
+
+from socket import *
+
+
+async def echo_server(loop, address):
+    sock = socket(AF_INET, SOCK_STREAM)
+    sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
+    sock.bind(address)
+    sock.listen(5)
+    sock.setblocking(False)
+    print('Server listening at', address)
+    with sock:
+        while True:
+             client, addr = await loop.sock_accept(sock)
+             print('Connection from', addr)
+             loop.create_task(echo_client(loop, client))
+
+
+async def echo_client(loop, client):
+    with client:
+         while True:
+             data = await loop.sock_recv(client, 10000)
+             if not data:
+                  break
+             await loop.sock_sendall(client, data)
+    print('Connection closed')
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--uvloop', default=False, action='store_true')
+    args = parser.parse_args()
+
+    if args.uvloop:
+        loop = uvloop.Loop()
+        print('using UVLoop')
+    else:
+        loop = asyncio.new_event_loop()
+        print('using asyncio loop')
+
+    asyncio.set_event_loop(loop)
+    loop.set_debug(False)
+
+    loop.create_task(echo_server(loop, ('', 25000)))
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()

setup.py

@@ -51,7 +51,8 @@ def build_extensions(self):
             "uvloop.futures",
             sources = [
                 "uvloop/futures.c",
-            ]
+            ],
+            extra_compile_args=['-O2']
         ),
 
         Extension(

tests/test_futures.py

@@ -24,7 +24,7 @@ def test_cancel(self):
         self.assertFalse(f.cancel())
 
     def test_initial_state(self):
-        f = CFuture(self.loop)
+        f = CFuture(loop=self.loop)
         self.assertFalse(f.cancelled())
         self.assertFalse(f.done())
         f.cancel()

uvloop/consts.pxi

@@ -1,5 +1,9 @@
 DEF DEBUG = 0
 
+# XXX: To use Future implemented in C, you need a modified
+# version of asyncio.
+DEF USE_C_FUTURE = 0
+
 DEF UV_STREAM_RECV_BUF_SIZE = 65536
 
 DEF FLOW_CONTROL_HIGH_WATER = 65536

uvloop/futures.c

@@ -3,7 +3,7 @@
 #include "Python.h"
 #include "structmember.h"
 
-static struct Locals { // XXX
+static struct Locals { // XXX!!!
     PyObject* is_error;
     PyObject* ce_error;
 } locals;
@@ -67,10 +67,12 @@ _schedule_callbacks(FutureObj *fut) {
 static PyObject *
 FutureObj_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
 {
+    static char *kwlist[] = {"loop", NULL};
+
     PyObject *loop;
     FutureObj *fut;
 
-    if (!PyArg_UnpackTuple(args, "Future", 1, 1, &loop))
+    if (!PyArg_ParseTupleAndKeywords(args, kwds, "|O", kwlist, &loop))
         return NULL;
 
     fut = PyObject_GC_New(FutureObj, type);
@@ -255,6 +257,13 @@ FutureObj_iternext(FutureObj *fut)
     return NULL;
 }
 
+static PyObject *
+FutureObj_send(FutureObj *fut, PyObject *res) {
+    PyErr_Format(PyExc_RuntimeError,
+                 "future.send() was called; unpatched asyncio");
+    return NULL;
+}
+
 static PyObject *
 FutureObj_add_done_callback(FutureObj *fut, PyObject *arg)
 {
@@ -345,6 +354,8 @@ static PyMethodDef FutureType_methods[] = {
     {"done", (PyCFunction)FutureObj_done, METH_NOARGS, NULL},
     {"result", (PyCFunction)FutureObj_result, METH_NOARGS, NULL},
     {"exception", (PyCFunction)FutureObj_exception, METH_NOARGS, NULL},
+
+    {"send", (PyCFunction)FutureObj_send, METH_O, NULL}, // XXX
     {NULL, NULL}        /* Sentinel */
 };
 
@@ -425,6 +436,7 @@ futures_exec(PyObject *module) {
         return -1;
     }
 
+    // XXX!!!
     PyObject *asyncio = PyImport_ImportModule("asyncio");
     if (asyncio == NULL) {
         return -1;

uvloop/loop.pyx

@@ -24,6 +24,13 @@ include "consts.pxi"
 include "stdlib.pxi"
 
 
+cdef Future
+IF USE_C_FUTURE:
+    Future = c_Future
+ELSE:
+    Future = aio_Future
+
+
 class LoopError(Exception):
     pass
 
@@ -279,7 +286,7 @@ cdef class Loop:
                       int proto, int flags,
                       int unpack):
 
-        fut = aio_Future(loop=self)
+        fut = Future(loop=self)
 
         def callback(result):
             if AddrInfo.isinstance(result):
@@ -655,27 +662,27 @@ cdef class Loop:
         return result
 
     def sock_recv(self, sock, n):
-        fut = CFuture(self)
+        fut = Future(loop=self)
         self._sock_recv(fut, False, sock, n)
         return fut
 
     def sock_sendall(self, sock, data):
-        fut = CFuture(self)
+        fut = Future(loop=self)
         if data:
             self._sock_sendall(fut, False, sock, data)
         else:
             fut.set_result(None)
         return fut
 
     def sock_accept(self, sock):
-        fut = aio_Future(loop=self)
+        fut = Future(loop=self)
         self._sock_accept(fut, False, sock)
         return fut
 
     def sock_connect(self, sock, address):
         if self._debug and sock.gettimeout() != 0:
             raise ValueError("the socket must be non-blocking")
-        fut = aio_Future(loop=self)
+        fut = Future(loop=self)
         try:
             if self._debug:
                 aio__check_resolved_address(sock, address)

uvloop/stdlib.pxi

@@ -7,6 +7,11 @@ import socket
 import sys
 
 
+from . import futures
+# asyncio.Future.register(futures.Future)
+cdef c_Future = futures.Future
+
+
 cdef aio_CancelledError = asyncio.CancelledError
 cdef aio_TimeoutError = asyncio.TimeoutError
 cdef aio_Future = asyncio.Future
@@ -36,4 +41,4 @@ cdef str sys_platform = sys.platform
 
 
 # Cython doesn't clean-up imported objects properly in Py3 mode.
-del asyncio, concurrent, collections, functools, socket, os, sys
+del asyncio, concurrent, collections, futures, functools, socket, os, sys