diff --git a/tests/test_udp.py b/tests/test_udp.py index e529b9e3..94018053 100644 --- a/tests/test_udp.py +++ b/tests/test_udp.py @@ -251,6 +251,100 @@ def connection_lost(self, exc): self.assertIn(tmp_file2, pr.addrs) + def _fill_send_queue(self, tr): + # Send until the OS refuses a datagram and the transport has to + # queue it. + for _ in range(10000): + if tr.get_write_buffer_size(): + return + tr.sendto(b'x' * 64) + self.fail('the OS never refused a datagram') + + def _close_with_queued_datagram(self, method, drain): + # Close the transport while a datagram is queued (the OS refused + # it with EAGAIN), optionally letting the peer make room for it + # before the loop runs again, and return the protocol events. + # Any call to the loop's exception handler fails the test. + if self.implementation == 'asyncio' and sys.platform == 'darwin': + # macOS refuses UNIX datagrams with ENOBUFS rather than + # EAGAIN; asyncio reports that to error_received() and drops + # the datagram instead of queueing it. + raise unittest.SkipTest( + 'asyncio does not queue datagrams refused with ENOBUFS') + + class Proto(asyncio.DatagramProtocol): + def __init__(self, loop): + self.events = [] + self.done = asyncio.Future(loop=loop) + + def connection_made(self, transport): + transport.set_write_buffer_limits(0) + + def pause_writing(self): + self.events.append('pause_writing') + + def resume_writing(self): + self.events.append('resume_writing') + + def error_received(self, exc): + self.events.append(('error_received', exc)) + + def connection_lost(self, exc): + self.events.append(('connection_lost', exc)) + self.done.set_result(None) + + async def run(peer, path): + with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock: + sock.connect(path) + sock.setblocking(False) + pr = Proto(self.loop) + tr, _ = await self.loop.create_datagram_endpoint( + lambda: pr, sock=sock) + + self._fill_send_queue(tr) + + getattr(tr, method)() + pr.events.append(method) + if drain: + try: + while True: + peer.recv(64) + except BlockingIOError: + pass + + await pr.done + return pr.events + + with tempfile.TemporaryDirectory() as tmp_dir: + path = os.path.join(tmp_dir, 'peer.sock') + # A UNIX datagram socket is used because UDP over loopback + # never refuses a datagram, so the sender's queue never fills. + with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as peer: + peer.bind(path) + peer.setblocking(False) + return self.loop.run_until_complete(run(peer, path)) + + def test_abort_with_queued_datagram(self): + # gh-771: the queued datagram is discarded silently. + self.assertEqual( + self._close_with_queued_datagram('abort', drain=False), + ['pause_writing', 'abort', ('connection_lost', None)]) + + def test_abort_with_queued_datagram_then_writable(self): + # The queued datagram can be sent before connection_lost() runs; + # the protocol must not hear about it after abort(). + self.assertEqual( + self._close_with_queued_datagram('abort', drain=True), + ['pause_writing', 'abort', ('connection_lost', None)]) + + def test_close_with_queued_datagram_then_writable(self): + # close() flushes the queue, so the protocol is resumed before + # connection_lost(). + self.assertEqual( + self._close_with_queued_datagram('close', drain=True), + ['pause_writing', 'close', 'resume_writing', + ('connection_lost', None)]) + def test_create_datagram_1(self): server_addr = ('127.0.0.1', 8888) client_addr = ('127.0.0.1', 0) @@ -351,6 +445,41 @@ def test_create_datagram_endpoint_reuse_address_warning(self): class Test_UV_UDP(_TestUDP, tb.UVTestCase): + def test_loop_close_with_queued_datagram(self): + # Closing the loop closes the transport's handle directly, which + # cancels the queued datagram; that must not be reported to the + # protocol or the exception handler. + events = [] + + class Proto(asyncio.DatagramProtocol): + def connection_made(self, transport): + transport.set_write_buffer_limits(0) + + def resume_writing(self): + events.append('resume_writing') + + def error_received(self, exc): + events.append(('error_received', exc)) + + with tempfile.TemporaryDirectory() as tmp_dir: + path = os.path.join(tmp_dir, 'peer.sock') + with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as peer: + peer.bind(path) + + with socket.socket( + socket.AF_UNIX, socket.SOCK_DGRAM) as sock: + sock.connect(path) + sock.setblocking(False) + tr, _ = self.loop.run_until_complete( + self.loop.create_datagram_endpoint(Proto, sock=sock)) + + self._fill_send_queue(tr) + + with self.assertWarnsRegex(ResourceWarning, 'unclosed'): + self.loop.close() + + self.assertEqual(events, []) + def test_create_datagram_endpoint_wrong_sock(self): sock = socket.socket(socket.AF_INET) with sock: diff --git a/uvloop/handles/udp.pxd b/uvloop/handles/udp.pxd index daa9a1be..e1302154 100644 --- a/uvloop/handles/udp.pxd +++ b/uvloop/handles/udp.pxd @@ -13,6 +13,8 @@ cdef class UDPTransport(UVBaseTransport): cdef open(self, int family, int sockfd) cdef _set_broadcast(self, bint on) + cdef _close(self) + cdef inline __receiving_started(self) cdef inline __receiving_stopped(self) diff --git a/uvloop/handles/udp.pyx b/uvloop/handles/udp.pyx index eac1bca5..61a2c699 100644 --- a/uvloop/handles/udp.pyx +++ b/uvloop/handles/udp.pyx @@ -171,6 +171,14 @@ cdef class UDPTransport(UVBaseTransport): else: self.__receiving_stopped() + cdef _close(self): + try: + # Drop the reference held while receiving, even when the + # handle is closed directly (e.g. by loop.close()). + self._stop_reading() + finally: + UVSocketHandle._close(self) + cdef inline __receiving_started(self): if self.__receiving: return @@ -267,6 +275,11 @@ cdef class UDPTransport(UVBaseTransport): run_in_context1(self.context, self._protocol.error_received, exc) cdef _on_sent(self, object exc, object context=None): + if self._conn_lost: + # abort() or a fatal error discarded the write buffer; like + # asyncio, don't report anything more to the protocol. + return + if exc is not None: if isinstance(exc, OSError): if context is None: @@ -400,6 +413,11 @@ cdef void __uv_udp_on_send( ctx.close() + if status == uv.UV_ECANCELED and udp._closed: + # The handle is being closed (e.g. by abort()) and libuv + # cancelled the queued send; the datagram is simply discarded. + return + if status < 0: exc = convert_error(status) print(exc)