Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions Lib/asyncio/proactor_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -571,6 +571,15 @@ def _loop_reading(self, fut=None):
else:
self._read_fut = self._loop._proactor.recvfrom(self._sock,
self.max_size)
except ConnectionResetError as exc:
# WSARecvFrom() reports a stale ICMP port unreachable
# notification as a synchronous ConnectionResetError when the
# same socket was used to send to an address that is not
# listening. This is transient, so reschedule the read loop
# instead of leaving it dead.
self._protocol.error_received(exc)
if not self._closing:
self._loop.call_soon(self._loop_reading)
except OSError as exc:
self._protocol.error_received(exc)
except exceptions.CancelledError:
Expand Down
54 changes: 54 additions & 0 deletions Lib/test/test_asyncio/test_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -1583,6 +1583,60 @@ def create_socket():
transport_1.close()
transport_2.close()

def test_datagram_recvfrom_connection_reset_recovers(self):
# gh-127057: a UDP socket that sent a datagram to an address that
# wasn't listening can raise ConnectionResetError on a later
# receive. The transport must keep working afterwards.
loop = self.loop

class Protocol(asyncio.DatagramProtocol):
def connection_made(self, transport):
self.transport = transport
self.errors = []
self.received = []
self.datagram_received_event = loop.create_future()

def error_received(self, exc):
self.errors.append(exc)

def datagram_received(self, data, addr):
self.received.append(data)
if not self.datagram_received_event.done():
self.datagram_received_event.set_result(None)

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setblocking(False)
sock.bind(('127.0.0.1', 0))
addr = sock.getsockname()

# Bind and immediately close a second socket to get an address
# that is guaranteed not to be listening.
closed = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
closed.bind(('127.0.0.1', 0))
closed_addr = closed.getsockname()
closed.close()

# Trigger the error before the socket is wrapped in a transport,
# so that the first read raises synchronously.
sock.sendto(b'x', closed_addr)

transport, protocol = loop.run_until_complete(
loop.create_datagram_endpoint(Protocol, sock=sock))

transport.sendto(b'ping', addr)
loop.run_until_complete(asyncio.wait_for(
protocol.datagram_received_event, support.SHORT_TIMEOUT))
self.assertEqual(protocol.received, [b'ping'])

if sys.platform == 'win32':
# Other platforms don't report ICMP errors on an
# unconnected UDP socket.
self.assertEqual(len(protocol.errors), 1)
self.assertIsInstance(protocol.errors[0], ConnectionResetError)

transport.close()
test_utils.run_briefly(loop)

def test_internal_fds(self):
loop = self.create_event_loop()
if not isinstance(loop, selector_events.BaseSelectorEventLoop):
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Fix :class:`asyncio.ProactorEventLoop` UDP transports so that a
:exc:`ConnectionResetError` raised by ``WSARecvFrom`` no longer stops the
read loop.
2 changes: 1 addition & 1 deletion Objects/bytesobject.c
Original file line number Diff line number Diff line change
Expand Up @@ -3629,7 +3629,7 @@ byteswriter_resize(PyBytesWriter *writer, Py_ssize_t size, int resize)
return 0;
}

if (resize & writer->overallocate) {
if (resize && writer->overallocate) {
if (size <= (PY_SSIZE_T_MAX - size / OVERALLOCATE_FACTOR)) {
size += size / OVERALLOCATE_FACTOR;
}
Expand Down
Loading