Don't report queued datagrams to the protocol after UDPTransport.abort()

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:

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)
 

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(<UVHandle>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)