diff --git a/tests/test_tcp.py b/tests/test_tcp.py index ab104c6b..3731fd90 100644 --- a/tests/test_tcp.py +++ b/tests/test_tcp.py @@ -129,6 +129,8 @@ async def start_server(): srv_socks = srv.sockets self.assertTrue(srv_socks) + if self.has_start_serving(): + self.assertTrue(srv.is_serving()) addr = srv_socks[0].getsockname() @@ -147,6 +149,9 @@ async def start_server(): for srv_sock in srv_socks: self.assertEqual(srv_sock.fileno(), -1) + if self.has_start_serving(): + self.assertFalse(srv.is_serving()) + async def start_server_sock(): nonlocal CNT CNT = 0 @@ -167,6 +172,8 @@ async def start_server_sock(): srv_socks = srv.sockets self.assertTrue(srv_socks) + if self.has_start_serving(): + self.assertTrue(srv.is_serving()) tasks = [] for _ in range(TOTAL_CNT): @@ -183,6 +190,9 @@ async def start_server_sock(): for srv_sock in srv_socks: self.assertEqual(srv_sock.fileno(), -1) + if self.has_start_serving(): + self.assertFalse(srv.is_serving()) + self.loop.run_until_complete(start_server()) self.assertEqual(CNT, TOTAL_CNT) @@ -206,6 +216,8 @@ async def start_server_ephemeral_ports(): srv_socks = srv.sockets self.assertTrue(srv_socks) + if self.has_start_serving(): + self.assertTrue(srv.is_serving()) host, port = srv_socks[0].getsockname() self.assertNotEqual(0, port) @@ -217,6 +229,9 @@ async def start_server_ephemeral_ports(): for srv_sock in srv_socks: self.assertEqual(srv_sock.fileno(), -1) + if self.has_start_serving(): + self.assertFalse(srv.is_serving()) + self.loop.run_until_complete(start_server_ephemeral_ports()) def test_create_server_4(self): @@ -335,6 +350,60 @@ def test_create_server_8(self): self.loop.create_server( lambda: None, host='::', port=0, ssl_handshake_timeout=10)) + def test_create_server_9(self): + if not self.has_start_serving(): + raise unittest.SkipTest() + + async def handle_client(reader, writer): + pass + + async def start_server(): + srv = await asyncio.start_server( + handle_client, + '127.0.0.1', 0, + family=socket.AF_INET, + loop=self.loop, + start_serving=False) + + await srv.start_serving() + self.assertTrue(srv.is_serving()) + + # call start_serving again + await srv.start_serving() + self.assertTrue(srv.is_serving()) + + srv.close() + await srv.wait_closed() + self.assertFalse(srv.is_serving()) + + self.loop.run_until_complete(start_server()) + + def test_create_server_10(self): + if not self.has_start_serving(): + raise unittest.SkipTest() + + async def handle_client(reader, writer): + pass + + async def start_server(): + srv = await asyncio.start_server( + handle_client, + '127.0.0.1', 0, + family=socket.AF_INET, + loop=self.loop, + start_serving=False) + + fut = asyncio.ensure_future(srv.serve_forever(), loop=self.loop) + await asyncio.sleep(0, loop=self.loop) + self.assertTrue(srv.is_serving()) + + fut.cancel() + with self.assertRaises(asyncio.CancelledError): + await fut + self.assertFalse(srv.is_serving()) + + self.loop.run_until_complete(start_server()) + def test_create_connection_open_con_addr(self): async def client(addr): reader, writer = await asyncio.open_connection( diff --git a/tests/test_unix.py b/tests/test_unix.py index 0f3afff9..92ae1f1e 100644 --- a/tests/test_unix.py +++ b/tests/test_unix.py @@ -66,6 +66,8 @@ async def start_server(): try: srv_socks = srv.sockets self.assertTrue(srv_socks) + if self.has_start_serving(): + self.assertTrue(srv.is_serving()) tasks = [] for _ in range(TOTAL_CNT): @@ -83,6 +85,9 @@ async def start_server(): for srv_sock in srv_socks: self.assertEqual(srv_sock.fileno(), -1) + if self.has_start_serving(): + self.assertFalse(srv.is_serving()) + # asyncio doesn't cleanup the sock file self.assertTrue(os.path.exists(sock_name)) @@ -100,6 +105,8 @@ async def start_server_sock(start_server): try: srv_socks = srv.sockets self.assertTrue(srv_socks) + if self.has_start_serving(): + self.assertTrue(srv.is_serving()) tasks = [] for _ in range(TOTAL_CNT): @@ -117,6 +124,9 @@ async def start_server_sock(start_server): for srv_sock in srv_socks: self.assertEqual(srv_sock.fileno(), -1) + if self.has_start_serving(): + self.assertFalse(srv.is_serving()) + # asyncio doesn't cleanup the sock file self.assertTrue(os.path.exists(sock_name)) diff --git a/uvloop/_testbase.py b/uvloop/_testbase.py index e7580e22..3042c311 100644 --- a/uvloop/_testbase.py +++ b/uvloop/_testbase.py @@ -67,6 +67,10 @@ def new_loop(self): def mock_pattern(self, str): return MockPattern(str) + def has_start_serving(self): + return not (self.is_asyncio_loop() and + sys.version_info[:2] in [(3, 5), (3, 6)]) + def is_asyncio_loop(self): return type(self.loop).__module__.startswith('asyncio.') diff --git a/uvloop/handles/pipe.pxd b/uvloop/handles/pipe.pxd index a2b48229..f5dc1a93 100644 --- a/uvloop/handles/pipe.pxd +++ b/uvloop/handles/pipe.pxd @@ -4,6 +4,7 @@ cdef class UnixServer(UVStreamServer): @staticmethod cdef UnixServer new(Loop loop, object protocol_factory, Server server, + object backlog, object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout) diff --git a/uvloop/handles/pipe.pyx b/uvloop/handles/pipe.pyx index aa36d745..581554f4 100644 --- a/uvloop/handles/pipe.pyx +++ b/uvloop/handles/pipe.pyx @@ -38,13 +38,14 @@ cdef class UnixServer(UVStreamServer): @staticmethod cdef UnixServer new(Loop loop, object protocol_factory, Server server, + object backlog, object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout): cdef UnixServer handle handle = UnixServer.__new__(UnixServer) - handle._init(loop, protocol_factory, server, + handle._init(loop, protocol_factory, server, backlog, ssl, ssl_handshake_timeout, ssl_shutdown_timeout) __pipe_init_uv_handle(handle, loop) return handle diff --git a/uvloop/handles/streamserver.pxd b/uvloop/handles/streamserver.pxd index 019d022e..b2ab1887 100644 --- a/uvloop/handles/streamserver.pxd +++ b/uvloop/handles/streamserver.pxd @@ -1,5 +1,6 @@ cdef class UVStreamServer(UVSocketHandle): cdef: + int backlog object ssl object ssl_handshake_timeout object ssl_shutdown_timeout @@ -10,13 +11,15 @@ cdef class UVStreamServer(UVSocketHandle): # All "inline" methods are final cdef inline _init(self, Loop loop, object protocol_factory, - Server server, object ssl, + Server server, + object backlog, + object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout) cdef inline _mark_as_open(self) - cdef inline listen(self, backlog) + cdef inline listen(self) cdef inline _on_listen(self) cdef UVStream _make_new_transport(self, object protocol, object waiter) diff --git a/uvloop/handles/streamserver.pyx b/uvloop/handles/streamserver.pyx index 3a9a5b8b..5343d50d 100644 --- a/uvloop/handles/streamserver.pyx +++ b/uvloop/handles/streamserver.pyx @@ -10,10 +10,17 @@ cdef class UVStreamServer(UVSocketHandle): self.protocol_factory = None cdef inline _init(self, Loop loop, object protocol_factory, - Server server, object ssl, + Server server, + object backlog, + object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout): + if not isinstance(backlog, int): + # Don't allow floats + raise TypeError('integer argument expected, got {}'.format( + type(backlog).__name__)) + if ssl is not None: if not isinstance(ssl, ssl_SSLContext): raise TypeError( @@ -27,6 +34,7 @@ cdef class UVStreamServer(UVSocketHandle): raise ValueError( 'ssl_shutdown_timeout is only meaningful with ssl') + self.backlog = backlog self.ssl = ssl self.ssl_handshake_timeout = ssl_handshake_timeout self.ssl_shutdown_timeout = ssl_shutdown_timeout @@ -35,15 +43,10 @@ cdef class UVStreamServer(UVSocketHandle): self.protocol_factory = protocol_factory self._server = server - cdef inline listen(self, backlog): + cdef inline listen(self): cdef int err self._ensure_alive() - if not isinstance(backlog, int): - # Don't allow floats - raise TypeError('integer argument expected, got {}'.format( - type(backlog).__name__)) - if self.protocol_factory is None: raise RuntimeError('unable to listen(); no protocol_factory') @@ -51,7 +54,7 @@ cdef class UVStreamServer(UVSocketHandle): raise RuntimeError('unopened TCPServer') err = uv.uv_listen( self._handle, - backlog, + self.backlog, __uv_streamserver_on_listen) if err < 0: exc = convert_error(err) diff --git a/uvloop/handles/tcp.pxd b/uvloop/handles/tcp.pxd index ed886d8b..69284db3 100644 --- a/uvloop/handles/tcp.pxd +++ b/uvloop/handles/tcp.pxd @@ -3,7 +3,9 @@ cdef class TCPServer(UVStreamServer): @staticmethod cdef TCPServer new(Loop loop, object protocol_factory, Server server, - object ssl, unsigned int flags, + unsigned int flags, + object backlog, + object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout) diff --git a/uvloop/handles/tcp.pyx b/uvloop/handles/tcp.pyx index ee76e11a..363e0cd8 100644 --- a/uvloop/handles/tcp.pyx +++ b/uvloop/handles/tcp.pyx @@ -57,13 +57,15 @@ cdef class TCPServer(UVStreamServer): @staticmethod cdef TCPServer new(Loop loop, object protocol_factory, Server server, - object ssl, unsigned int flags, + unsigned int flags, + object backlog, + object ssl, object ssl_handshake_timeout, object ssl_shutdown_timeout): cdef TCPServer handle handle = TCPServer.__new__(TCPServer) - handle._init(loop, protocol_factory, server, + handle._init(loop, protocol_factory, server, backlog, ssl, ssl_handshake_timeout, ssl_shutdown_timeout) __tcp_init_uv_handle(handle, loop, flags) return handle diff --git a/uvloop/includes/stdlib.pxi b/uvloop/includes/stdlib.pxi index 2240b512..5ba53ffa 100644 --- a/uvloop/includes/stdlib.pxi +++ b/uvloop/includes/stdlib.pxi @@ -66,6 +66,8 @@ cdef gc_disable = gc.disable cdef iter_chain = itertools.chain cdef inspect_isgenerator = inspect.isgenerator +cdef int has_IPV6_V6ONLY = hasattr(socket, 'IPV6_V6ONLY') +cdef int IPV6_V6ONLY = getattr(socket, 'IPV6_V6ONLY', -1) cdef int has_SO_REUSEPORT = hasattr(socket, 'SO_REUSEPORT') cdef int SO_REUSEPORT = getattr(socket, 'SO_REUSEPORT', 0) cdef int SO_BROADCAST = getattr(socket, 'SO_BROADCAST') @@ -118,6 +120,7 @@ cdef sys_set_coroutine_wrapper = sys.set_coroutine_wrapper cdef sys_get_coroutine_wrapper = sys.get_coroutine_wrapper cdef sys_getframe = sys._getframe cdef sys_version_info = sys.version_info +cdef str sys_platform = sys.platform cdef ssl_SSLContext = ssl.SSLContext cdef ssl_MemoryBIO = ssl.MemoryBIO diff --git a/uvloop/loop.pxd b/uvloop/loop.pxd index 51f2947a..765e48ef 100644 --- a/uvloop/loop.pxd +++ b/uvloop/loop.pxd @@ -161,15 +161,6 @@ cdef class Loop: cdef _getnameinfo(self, system.sockaddr *addr, int flags) - cdef _create_server(self, system.sockaddr *addr, - object protocol_factory, - Server server, - object ssl, - bint reuse_port, - object backlog, - object ssl_handshake_timeout, - object ssl_shutdown_timeout) - cdef _track_transport(self, UVBaseTransport transport) cdef _fileobj_to_fd(self, fileobj) cdef _ensure_fd_no_transport(self, fd) diff --git a/uvloop/loop.pyx b/uvloop/loop.pyx index 6f7d8e3c..53071227 100644 --- a/uvloop/loop.pyx +++ b/uvloop/loop.pyx @@ -1084,48 +1084,6 @@ cdef class Loop: sys_set_coroutine_wrapper(None) self._coroutine_debug_set = False - cdef _create_server(self, system.sockaddr *addr, - object protocol_factory, - Server server, - object ssl, - bint reuse_port, - object backlog, - object ssl_handshake_timeout, - object ssl_shutdown_timeout): - cdef: - TCPServer tcp - int bind_flags - - tcp = TCPServer.new(self, protocol_factory, server, ssl, - addr.sa_family, - ssl_handshake_timeout, ssl_shutdown_timeout) - - if reuse_port: - self._sock_set_reuseport(tcp._fileno()) - - if addr.sa_family == uv.AF_INET6: - # Disable IPv4/IPv6 dual stack support (enabled by - # default on Linux) which makes a single socket - # listen on both address families. - bind_flags = uv.UV_TCP_IPV6ONLY - else: - bind_flags = 0 - - try: - tcp.bind(addr, bind_flags) - tcp.listen(backlog) - except OSError as err: - pyaddr = __convert_sockaddr_to_pyaddr(addr) - tcp._close() - raise OSError(err.errno, 'error while attempting ' - 'to bind on address %r: %s' - % (pyaddr, err.strerror.lower())) - except Exception: - tcp._close() - raise - - return tcp - def _get_backend_id(self): """This method is used by uvloop tests and is not part of the API.""" return uv.uv_backend_fd(self.uvloop) @@ -1584,10 +1542,11 @@ cdef class Loop: sock=None, backlog=100, ssl=None, - reuse_address=None, # ignored, libuv sets it + reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, - ssl_shutdown_timeout=None): + ssl_shutdown_timeout=None, + start_serving=True): """A coroutine which creates a TCP server bound to host and port. The return value is a Server object which can be used to stop @@ -1641,7 +1600,8 @@ cdef class Loop: raise ValueError( 'host/port and sock can not be specified at the same time') return await self.create_unix_server( - protocol_factory, sock=sock, backlog=backlog, ssl=ssl) + protocol_factory, sock=sock, backlog=backlog, ssl=ssl, + start_serving=start_serving) server = Server(self) @@ -1661,6 +1621,8 @@ cdef class Loop: raise ValueError( 'host/port and sock can not be specified at the same time') + if reuse_address is None: + reuse_address = os_name == 'posix' and sys_platform != 'cygwin' reuse_port = bool(reuse_port) if reuse_port and not has_SO_REUSEPORT: raise ValueError( @@ -1680,6 +1642,7 @@ cdef class Loop: infos = await aio_gather(*fs, loop=self) completed = False + sock = None try: for info in infos: addrinfo = (info).data @@ -1687,18 +1650,64 @@ cdef class Loop: if addrinfo.ai_family == uv.AF_UNSPEC: raise RuntimeError('AF_UNSPEC in DNS results') - tcp = self._create_server( - addrinfo.ai_addr, protocol_factory, server, - ssl, reuse_port, backlog, - ssl_handshake_timeout, ssl_shutdown_timeout) + try: + sock = socket_socket(addrinfo.ai_family, + addrinfo.ai_socktype, + addrinfo.ai_protocol) + except socket_error: + # Assume it's a bad family/type/protocol + # combination. + if self._debug: + aio_logger.warning( + 'create_server() failed to create ' + 'socket.socket(%r, %r, %r)', + addrinfo.ai_family, + addrinfo.ai_socktype, + addrinfo.ai_protocol, exc_info=True) + continue - server._add_server(tcp) + if reuse_address: + sock.setsockopt(uv.SOL_SOCKET, uv.SO_REUSEADDR, 1) + if reuse_port: + sock.setsockopt(uv.SOL_SOCKET, uv.SO_REUSEPORT, 1) + # Disable IPv4/IPv6 dual stack support (enabled by + # default on Linux) which makes a single socket + # listen on both address families. + if (addrinfo.ai_family == uv.AF_INET6 and + has_IPV6_V6ONLY): + sock.setsockopt(uv.IPPROTO_IPV6, IPV6_V6ONLY, 1) + + pyaddr = __convert_sockaddr_to_pyaddr(addrinfo.ai_addr) + try: + sock.bind(pyaddr) + except OSError as err: + raise OSError( + err.errno, 'error while attempting ' + 'to bind on address %r: %s' + % (pyaddr, err.strerror.lower())) from None + + tcp = TCPServer.new(self, protocol_factory, server, + uv.AF_UNSPEC, backlog, + ssl, ssl_handshake_timeout, + ssl_shutdown_timeout) + + try: + tcp._open(sock.fileno()) + except Exception: + tcp._close() + raise + + server._add_server(tcp) + sock.detach() + sock = None addrinfo = addrinfo.ai_next completed = True finally: if not completed: + if sock is not None: + sock.close() server.close() else: if sock is None: @@ -1711,13 +1720,13 @@ cdef class Loop: # we want Python socket object to notice that. sock.setblocking(False) - tcp = TCPServer.new(self, protocol_factory, server, ssl, - uv.AF_UNSPEC, - ssl_handshake_timeout, ssl_shutdown_timeout) + tcp = TCPServer.new(self, protocol_factory, server, + uv.AF_UNSPEC, backlog, + ssl, ssl_handshake_timeout, + ssl_shutdown_timeout) try: tcp._open(sock.fileno()) - tcp.listen(backlog) except Exception: tcp._close() raise @@ -1725,6 +1734,9 @@ cdef class Loop: tcp._attach_fileobj(sock) server._add_server(tcp) + if start_serving: + server._start_serving() + server._ref() return server @@ -1960,7 +1972,8 @@ cdef class Loop: async def create_unix_server(self, protocol_factory, path=None, *, backlog=100, sock=None, ssl=None, ssl_handshake_timeout=None, - ssl_shutdown_timeout=None): + ssl_shutdown_timeout=None, + start_serving=True): """A coroutine which creates a UNIX Domain Socket server. The return value is a Server object, which can be used to stop @@ -2068,8 +2081,8 @@ cdef class Loop: sock.setblocking(False) pipe = UnixServer.new( - self, protocol_factory, server, ssl, - ssl_handshake_timeout, ssl_shutdown_timeout) + self, protocol_factory, server, backlog, + ssl, ssl_handshake_timeout, ssl_shutdown_timeout) try: pipe._open(sock.fileno()) @@ -2078,14 +2091,12 @@ cdef class Loop: sock.close() raise - try: - pipe.listen(backlog) - except Exception: - pipe._close() - raise - pipe._attach_fileobj(sock) server._add_server(pipe) + + if start_serving: + server._start_serving() + return server @cython.iterable_coroutine diff --git a/uvloop/server.pxd b/uvloop/server.pxd index 1555202c..ef10f813 100644 --- a/uvloop/server.pxd +++ b/uvloop/server.pxd @@ -4,9 +4,12 @@ cdef class Server: list _waiters int _active_count Loop _loop + bint _serving + object _serving_forever_fut object __weakref__ cdef _add_server(self, UVStreamServer srv) + cdef _start_serving(self) cdef _wakeup(self) cdef _attach(self) diff --git a/uvloop/server.pyx b/uvloop/server.pyx index 5e0100af..1a0598a2 100644 --- a/uvloop/server.pyx +++ b/uvloop/server.pyx @@ -1,13 +1,25 @@ +import asyncio + + cdef class Server: def __cinit__(self, Loop loop): self._loop = loop self._servers = [] self._waiters = [] self._active_count = 0 + self._serving_forever_fut = None cdef _add_server(self, UVStreamServer srv): self._servers.append(srv) + cdef _start_serving(self): + if self._serving: + return + + self._serving = 1 + for server in self._servers: + (server).listen() + cdef _wakeup(self): cdef list waiters @@ -62,6 +74,7 @@ cdef class Server: try: servers = self._servers self._servers = None + self._serving = 0 for server in servers: (server)._close() @@ -71,6 +84,35 @@ cdef class Server: finally: self._unref() + def is_serving(self): + return self._serving + + @cython.iterable_coroutine + async def start_serving(self): + self._start_serving() + + @cython.iterable_coroutine + async def serve_forever(self): + if self._serving_forever_fut is not None: + raise RuntimeError( + f'server {self!r} is already being awaited on serve_forever()') + if self._servers is None: + raise RuntimeError(f'server {self!r} is closed') + + self._start_serving() + self._serving_forever_fut = self._loop.create_future() + + try: + await self._serving_forever_fut + except asyncio.CancelledError: + try: + self.close() + await self.wait_closed() + finally: + raise + finally: + self._serving_forever_fut = None + property sockets: def __get__(self): cdef list sockets = []