Implement Server and create_server

Author: 1st1

uvloop/handles/handle.pyx

@@ -38,13 +38,25 @@ cdef class UVHandle:
         if self._handle is NULL:
             return
 
+        if self._handle.loop is NULL:
+            # The handle wasn't initialized with "uv_{handle}_init"
+            self._closed = 1
+            PyMem_Free(self._handle)
+            raise RuntimeError(
+                '{} is open in __dealloc__ with loop set to NULL'
+                .format(self.__class__.__name__))
+
+        # When we're at this point, something went wrong.
+
         if self._closed == 1:
             # So _handle is not NULL and self._closed == 1?
             raise RuntimeError(
                 '{}.__dealloc__: _handle is NULL, _closed == 1'.format(
                     self.__class__.__name__))
 
-        # Let's close the handle.
+        # The handle is dealloced while open.  Let's try to close it.
+        # Situations when this is possible include unhandled exceptions,
+        # errors during Handle.__cinit__/__init__ etc.
         self._handle.data = <void*> __NOHANDLE__
         uv.uv_close(self._handle, __uv_close_handle_cb) # void; no errors
         self._handle = NULL

uvloop/handles/stream.pxd

@@ -2,8 +2,10 @@ cdef class UVStream(UVHandle):
     cdef:
         uv.uv_shutdown_t _shutdown_req
         bint __reading
+        object __cached_socket
 
     cdef _fileno(self)
+    cdef _get_socket(self)
 
     cdef _shutdown(self)
 

uvloop/handles/stream.pyx

@@ -69,6 +69,7 @@ cdef class _StreamWriteContext:
 cdef class UVStream(UVHandle):
     def __cinit__(self):
         self.__reading = 0
+        self.__cached_socket = None
 
     cdef _shutdown(self):
         cdef int err
@@ -186,9 +187,33 @@ cdef class UVStream(UVHandle):
 
         return fd
 
+    cdef _get_socket(self):
+        cdef:
+            int buf_len = sizeof(system.sockaddr_storage)
+            int err
+            system.sockaddr_storage buf
+
+        if self.__cached_socket is not None:
+            return self.__cached_socket
+
+        err = uv.uv_tcp_getsockname(<uv.uv_tcp_t*>self._handle,
+                                    <system.sockaddr*>&buf,
+                                    &buf_len)
+        if err < 0:
+            raise convert_error(err)
+
+        self.__cached_socket = socket_fromfd(self._fileno(),
+                                            buf.ss_family,
+                                            uv.SOCK_STREAM)
+
+        return self.__cached_socket
+
     cdef _close(self):
         self.__reading_stopped()
         UVHandle._close(self)
+        if self.__cached_socket is not None:
+            self.__cached_socket.close()
+            self.__cached_socket = None
 
     # Methods to override.
 

uvloop/handles/tcp.pxd

@@ -12,6 +12,7 @@ cdef class UVTcpStream(UVStream):
 cdef class UVTCPServer(UVTcpStream):
     cdef:
         object protocol_factory
+        Server host_server
 
     cdef _init(self)
     cdef _set_protocol_factory(self, object protocol_factory)
@@ -23,11 +24,13 @@ cdef class UVTCPServer(UVTcpStream):
     cdef _on_listen(self)
 
     @staticmethod
-    cdef UVTCPServer new(Loop loop, object protocol_factory)
+    cdef UVTCPServer new(Loop loop, object protocol_factory, Server server)
 
 
 cdef class UVServerTransport(UVTcpStream):
     cdef:
+        Server host_server
+
         bint eof
         bint reading
 
@@ -58,5 +61,7 @@ cdef class UVServerTransport(UVTcpStream):
     cdef _maybe_pause_protocol(self)
     cdef _maybe_resume_protocol(self)
 
+    cdef _call_connection_lost(self, exc)
+
     @staticmethod
-    cdef UVServerTransport new(Loop loop, object protocol)
+    cdef UVServerTransport new(Loop loop, object protocol, Server server)

uvloop/handles/tcp.pyx

@@ -42,14 +42,16 @@ cdef class UVTCPServer(UVTcpStream):
     cdef _init(self):
         UVTcpStream._init(<UVTcpStream>self)
         self.protocol_factory = None
+        self.host_server = None
 
     @staticmethod
-    cdef UVTCPServer new(Loop loop, object protocol_factory):
+    cdef UVTCPServer new(Loop loop, object protocol_factory, Server server):
         cdef UVTCPServer handle
         handle = UVTCPServer.__new__(UVTCPServer)
         handle._set_loop(loop)
         handle._init()
         handle._set_protocol_factory(protocol_factory)
+        handle.host_server = server
         return handle
 
     cdef _set_protocol_factory(self, object protocol_factory):
@@ -88,7 +90,7 @@ cdef class UVTCPServer(UVTcpStream):
 
     cdef _on_listen(self):
         protocol = self.protocol_factory()
-        client = UVServerTransport.new(self._loop, protocol)
+        client = UVServerTransport.new(self._loop, protocol, self.host_server)
         client._accept(<UVStream>self)
 
 
@@ -112,12 +114,15 @@ cdef class UVServerTransport(UVTcpStream):
         self.low_water = FLOW_CONTROL_LOW_WATER
 
     @staticmethod
-    cdef UVServerTransport new(Loop loop, object protocol):
+    cdef UVServerTransport new(Loop loop, object protocol, Server server):
         cdef UVServerTransport handle
         handle = UVServerTransport.__new__(UVServerTransport)
         handle._set_loop(loop)
         handle._init()
         handle._set_protocol(protocol)
+        handle.host_server = server
+        if server is not None:
+            (<Server>server)._attach()
         return handle
 
     cdef _set_protocol(self, object protocol):
@@ -210,15 +215,37 @@ cdef class UVServerTransport(UVTcpStream):
                     'protocol': self.protocol,
                 })
 
+    cdef _call_connection_lost(self, exc):
+        try:
+            if self.protocol is not None:
+                self.protocol.connection_lost(exc)
+        finally:
+            self.protocol = None
+            self.protocol_data_received = None
+
+            self._close()
+
+            server = self.host_server
+            if server is not None:
+                server._detach()
+                self.host_server = None
+
+    cdef _report_callback_error(self, exc):
+        self._call_connection_lost(exc)
+        super()._report_callback_error(exc)
+
+    cdef _raise_fatal_error(self, exc):
+        self._call_connection_lost(exc)
+        super()._raise_fatal_error(exc)
+
     # Public API
 
     property _closing:
         def __get__(self):
-            # "self._closing" refers to "UVHandle._closing" cdef
-            return (<UVHandle>self)._closing or (<UVHandle>self)._closed
+            return (<UVHandle>self)._closed
 
     def is_closing(self):
-        return (<UVHandle>self)._closing or (<UVHandle>self)._closed
+        return (<UVHandle>self)._closed
 
     def write(self, object buf):
         self._write(buf, None)
@@ -279,23 +306,8 @@ cdef class UVServerTransport(UVTcpStream):
         return (self._low_water, self._high_water)
 
     def get_extra_info(self, name, default=None):
-        cdef:
-            int buf_len = sizeof(system.sockaddr_storage)
-            int err
-            system.sockaddr_storage buf
-
         if name == 'socket':
-            err = uv.uv_tcp_getsockname(<uv.uv_tcp_t*>self._handle,
-                                        <system.sockaddr*>&buf,
-                                        &buf_len)
-            if err < 0:
-                self._close()
-                raise convert_error(err)
-
-            return socket_fromfd(self._fileno(),
-                                 buf.ss_family,
-                                 uv.SOCK_STREAM)
-
+            return self._get_socket()
         return default
 
     def abort(self):

uvloop/includes/__init__.py

@@ -3,6 +3,7 @@
 import collections
 import concurrent.futures
 import functools
+import itertools
 import os
 import socket
 import sys

uvloop/includes/stdlib.pxi

@@ -2,6 +2,7 @@ import asyncio, asyncio.log, asyncio.base_events
 import collections
 import concurrent.futures
 import functools
+import itertools
 import os
 import socket
 import sys
@@ -29,6 +30,8 @@ cdef cc_ThreadPoolExecutor = concurrent.futures.ThreadPoolExecutor
 
 cdef ft_partial = functools.partial
 
+cdef iter_chain = itertools.chain
+
 cdef int has_AF_INET6 = hasattr(socket, 'AF_INET6')
 cdef int has_SO_REUSEPORT = hasattr(socket, 'SO_REUSEPORT')
 cdef int has_IPPROTO_IPV6 = hasattr(socket, 'IPPROTO_IPV6')
@@ -66,5 +69,5 @@ cdef long MAIN_THREAD_ID = <long>threading.main_thread().ident
 
 
 # Cython doesn't clean-up imported objects properly in Py3 mode.
-del asyncio, concurrent, collections, \
-    functools, socket, os, sys, threading
+del asyncio, concurrent, collections
+del functools, itertools, socket, os, sys, threading

uvloop/loop.pxd

@@ -120,4 +120,6 @@ include "handles/tcp.pxd"
 
 include "request.pxd"
 
+include "server.pxd"
+
 include "os_signal.pxd"

uvloop/loop.pyx

@@ -662,21 +662,66 @@ cdef class Loop:
         return self._getaddrinfo(host, port, family, type, proto, flags, 1)
 
     @aio_coroutine
-    async def create_server(self, protocol_factory, str host, int port):
-        addrinfo = await self._getaddrinfo(host, port, 0, 0, 0, 0, 0)
-        if not AddrInfo.isinstance(addrinfo):
-            raise RuntimeError('unvalid loop._getaddeinfo() result')
+    async def create_server(self, protocol_factory, str host, int port,
+                            *,
+                            int family=uv.AF_UNSPEC,
+                            int flags=uv.AI_PASSIVE,
+                            sock=None,
+                            int backlog=100,
+                            ssl=None,            # not implemented
+                            reuse_address=None,  # ignored, libuv sets it
+                            reuse_port=None):    # ignored
 
-        cdef system.addrinfo *ai = (<AddrInfo>addrinfo).data
-        if ai is NULL:
-            raise RuntimeError('loop._getaddeinfo() result is NULL')
+        cdef:
+            UVTCPServer tcp
+            system.addrinfo *addrinfo
+            Server server = Server(self)
+
+        if ssl is not None:
+            raise NotImplementedError('SSL is not yet supported')
+
+        if host is not None or port is not None:
+            if sock is not None:
+                raise ValueError(
+                    'host/port and sock can not be specified at the same time')
+
+            if host == '':
+                hosts = [None]
+            elif (isinstance(host, str) or not isinstance(host, col_Iterable)):
+                hosts = [host]
+            else:
+                hosts = host
+
+            fs = [self._getaddrinfo(host, port, family,
+                                    uv.SOCK_STREAM, 0, flags,
+                                    0) for host in hosts]
 
-        cdef UVTCPServer srv = UVTCPServer.new(self, protocol_factory)
+            infos = await aio_gather(*fs, loop=self)
+
+            completed = False
+            try:
+                for info in infos:
+                    addrinfo = (<AddrInfo>info).data
+                    while addrinfo != NULL:
+                        tcp = UVTCPServer.new(self, protocol_factory, server)
+                        tcp.bind(addrinfo.ai_addr)
+                        tcp.listen(backlog)
+
+                        server._add_server(tcp)
+
+                        addrinfo = addrinfo.ai_next
+
+                completed = True
+            finally:
+                if not completed:
+                    server.close()
+        else:
+            tcp = UVTCPServer.new(self, protocol_factory, server)
+            tcp.open(sock.fileno())
+            tcp.listen(backlog)
+            server._add_server(tcp)
 
-        srv.bind(ai.ai_addr)
-        srv.listen()
-        self._servers.add(srv) # XXX
-        return srv
+        return server
 
     def default_exception_handler(self, context):
         message = context.get('message')

uvloop/server.pxd

@@ -0,0 +1,12 @@
+cdef class Server:
+    cdef:
+        list _servers
+        list _waiters
+        int _active_count
+        Loop _loop
+
+
+    cdef _add_server(self, UVTCPServer srv)
+    cdef _wakeup(self)
+    cdef _attach(self)
+    cdef _detach(self)