1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
"""Datagrams moved straight through the socket on Windows' event loop, applied
on import.
The node runs on the ProactorEventLoop, the one that can start ffmpeg there
(`platform.configure_event_loop`), and so do its connection processes. Its
datagram transport keeps one overlapped operation in each direction and starts
the next only from the completion callback of the one before, which runs on a
later turn of the loop. So a socket sends **one datagram per turn of the loop,
and receives one**, however many are waiting, and a turn lasts as long as
everything else the loop has to do: the other connections, reading the file,
the hub. The busier the node, the slower every connection, with cores to spare.
Measured on loopback, 1200-byte datagrams, with the loop otherwise busy 0.5 ms
a turn: 1,928 sent a second (2.3 MB/s), and 1,954 received, the kernel dropping
87 % of what arrived; straight through the socket, 88,000 sent, and every
datagram received at the 20,000 a second it was sent at. In the node holding the
connections itself, a browser's download over eight connections ran at 9.1 MB/s
with the node at 45 % of a core and an SCTP round trip of 745 ms on loopback:
that is the queue in front of each socket. In a connection process serving the
command-line client, 42 % of the time was `IocpProactor.sendto`. Linux never
paid for the sending half: the selector loop's transport tries the socket first
and queues only what it refuses.
Sending: a datagram goes out with a non-blocking `sendto` when nothing is queued
and no overlapped send is pending, so order is kept; anything else, the socket
refusing included, goes through the proactor's own queue as before.
Receiving: once an overlapped read completes, its datagram is delivered, then
whatever else the socket holds is read with non-blocking `recvfrom`, and only
then is the next overlapped read posted. Never while one is pending, which
could hand a later datagram to the completion port ahead of the ones read here.
The overlapped path turns the ICMP "port unreachable" a send to a silent
address brings back into an empty read; read directly it is a
ConnectionResetError, and it is skipped the same way.
No effect off Windows, where the class is never used.
"""
import sys
from asyncio import exceptions, proactor_events
_Transport = proactor_events._ProactorDatagramTransport
_proactor_sendto = _Transport.sendto
# Datagrams read straight from the socket after one completion, at most: a
# socket under a flood still lets the loop turn.
DRAIN_MAX = 256
def sendto(self, data, addr=None) -> None:
if (self._write_fut is None and not self._buffer and not self._conn_lost
and isinstance(data, (bytes, bytearray, memoryview)) and data
and (self._address is None or addr in (None, self._address))):
try:
if self._address is not None:
self._sock.send(data)
else:
self._sock.sendto(data, addr)
return
except (BlockingIOError, InterruptedError):
pass
except OSError as exc:
# What the overlapped path does with a failed send.
self._protocol.error_received(exc)
return
_proactor_sendto(self, data, addr)
def _deliver(self, data, addr) -> None:
"""To the protocol, and whatever it raises reported as a callback's would
be: in CPython's order the next read was posted first, and an exception
here must not end the reading either."""
try:
self._protocol.datagram_received(data, addr)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._loop.call_exception_handler({
"message": "Exception in datagram_received", "exception": exc,
"transport": self, "protocol": self._protocol})
def _drain(self) -> None:
for _ in range(DRAIN_MAX):
if self._conn_lost or self._closing:
return
try:
if self._address is not None:
data, addr = self._sock.recv(self.max_size), self._address
else:
data, addr = self._sock.recvfrom(self.max_size)
except (BlockingIOError, InterruptedError):
return
except ConnectionResetError:
continue
except OSError as exc:
# The overlapped read posted next meets it again, and reports it
# the proactor's way.
self._protocol.error_received(exc)
return
if data:
_deliver(self, data, addr)
def _loop_reading(self, fut=None) -> None:
"""CPython's own (3.12), but a completed read is delivered and the socket
drained before the next overlapped read is posted, not after."""
try:
if self._conn_lost:
return
assert self._read_fut is fut or (self._read_fut is None and self._closing)
self._read_fut = None
if fut is not None:
res = fut.result()
if self._closing:
return
data, addr = (res, self._address) if self._address is not None else res
if data:
_deliver(self, data, addr)
_drain(self)
if self._conn_lost:
return
if self._address is not None:
self._read_fut = self._loop._proactor.recv(self._sock, self.max_size)
else:
self._read_fut = self._loop._proactor.recvfrom(self._sock, self.max_size)
except OSError as exc:
self._protocol.error_received(exc)
except exceptions.CancelledError:
if not self._closing:
raise
else:
if self._read_fut is not None:
self._read_fut.add_done_callback(self._loop_reading)
if sys.platform == "win32":
_Transport.sendto = sendto
_Transport._loop_reading = _loop_reading
|