@@ -1583,6 +1583,79 @@ def create_socket():
15831583 transport_1 .close ()
15841584 transport_2 .close ()
15851585
1586+ def _test_datagram_write_error_resumes_paused_protocol (self , first , second ):
1587+ # See https://github.com/python/cpython/issues/156698: a
1588+ # datagram write error must not strand data left in the write
1589+ # buffer, nor leave a paused protocol paused forever.
1590+ loop = self .loop
1591+
1592+ class Protocol (asyncio .DatagramProtocol ):
1593+ def connection_made (self , transport ):
1594+ self .transport = transport
1595+ self .paused = False
1596+ self .resumed = False
1597+ self .errors = []
1598+ self .error_received_event = loop .create_future ()
1599+
1600+ def pause_writing (self ):
1601+ self .paused = True
1602+
1603+ def resume_writing (self ):
1604+ self .resumed = True
1605+
1606+ def error_received (self , exc ):
1607+ self .errors .append (exc )
1608+ if not self .error_received_event .done ():
1609+ self .error_received_event .set_result (None )
1610+
1611+ sock = socket .socket (socket .AF_INET , socket .SOCK_DGRAM )
1612+ sock .setblocking (False )
1613+ sock .bind (('127.0.0.1' , 0 ))
1614+ transport , protocol = loop .run_until_complete (
1615+ loop .create_datagram_endpoint (Protocol , sock = sock ))
1616+ addr = sock .getsockname ()
1617+
1618+ # A high water mark of 0 makes pausing deterministic whenever
1619+ # anything is left in the write buffer.
1620+ transport .set_write_buffer_limits (0 )
1621+
1622+ # The first sendto() may arm an in-flight write, so the second
1623+ # one can end up queued behind it; queuing is what trips
1624+ # pause_writing() at a high water mark of 0.
1625+ transport .sendto (first , addr )
1626+ transport .sendto (second , addr )
1627+
1628+ loop .run_until_complete (
1629+ asyncio .wait_for (protocol .error_received_event , 10 ))
1630+ self .assertTrue (protocol .errors )
1631+ self .assertIsInstance (protocol .errors [0 ], OSError )
1632+
1633+ # The write buffer must not be left stranded.
1634+ test_utils .run_until (
1635+ loop , lambda : transport .get_write_buffer_size () == 0 )
1636+
1637+ # A protocol that got paused must eventually be resumed too --
1638+ # without requiring an unsolicited extra sendto() to un-stick it.
1639+ if protocol .paused :
1640+ test_utils .run_until (loop , lambda : protocol .resumed )
1641+
1642+ transport .close ()
1643+ test_utils .run_briefly (loop )
1644+
1645+ def test_datagram_write_error_resumes_paused_protocol_in_flight (self ):
1646+ # oversized datagram fails while in flight; a normal datagram
1647+ # queued right behind it must not be stranded.
1648+ oversized = b'\x00 ' * 70000
1649+ self ._test_datagram_write_error_resumes_paused_protocol (
1650+ oversized , b'queued' )
1651+
1652+ def test_datagram_write_error_resumes_paused_protocol_from_callback (self ):
1653+ # oversized datagram fails once it reaches the front of the
1654+ # buffer; the protocol must not stay paused forever.
1655+ oversized = b'\x00 ' * 70000
1656+ self ._test_datagram_write_error_resumes_paused_protocol (
1657+ b'ok' , oversized )
1658+
15861659 def test_internal_fds (self ):
15871660 loop = self .create_event_loop ()
15881661 if not isinstance (loop , selector_events .BaseSelectorEventLoop ):
0 commit comments