@@ -69,29 +69,39 @@ async def _start(self):
6969 self ._loop .call_soon (self ._protocol .connection_made , self )
7070 self ._loop .add_reader (self ._fileno , self ._read_ready )
7171
72+ # Maximum reads per callback to avoid starving other events
73+ _max_reads_per_call = 16
74+
7275 def _read_ready (self ):
73- """Called when data is available to read."""
76+ """Called when data is available to read.
77+
78+ Drains socket until EAGAIN with a budget to avoid starvation.
79+ """
7480 if self ._conn_lost :
7581 return
76- try :
77- data = self ._sock .recv (self .max_size )
78- except (BlockingIOError , InterruptedError ):
79- return
80- except Exception as exc :
81- self ._fatal_error (exc , 'Fatal read error' )
82- return
8382
84- if data :
85- self ._protocol .data_received (data )
86- else :
87- # Connection closed (EOF received)
88- self ._loop .remove_reader (self ._fileno )
89- keep_open = self ._protocol .eof_received ()
90- # If eof_received returns False/None, close the transport
91- if not keep_open :
92- self ._closing = True
93- self ._conn_lost += 1
94- self ._call_connection_lost (None )
83+ for _ in range (self ._max_reads_per_call ):
84+ try :
85+ data = self ._sock .recv (self .max_size )
86+ except (BlockingIOError , InterruptedError ):
87+ # EAGAIN - no more data available
88+ return
89+ except Exception as exc :
90+ self ._fatal_error (exc , 'Fatal read error' )
91+ return
92+
93+ if data :
94+ self ._protocol .data_received (data )
95+ else :
96+ # Connection closed (EOF received)
97+ self ._loop .remove_reader (self ._fileno )
98+ keep_open = self ._protocol .eof_received ()
99+ # If eof_received returns False/None, close the transport
100+ if not keep_open :
101+ self ._closing = True
102+ self ._conn_lost += 1
103+ self ._call_connection_lost (None )
104+ return
95105
96106 def write (self , data ):
97107 """Write data to the transport."""
@@ -122,30 +132,38 @@ def write(self, data):
122132
123133 self ._buffer .extend (data )
124134
135+ # Maximum writes per callback to avoid starving other events
136+ _max_writes_per_call = 16
137+
125138 def _write_ready_cb (self ):
126- """Called when socket is ready for writing."""
127- remaining = len (self ._buffer ) - self ._buffer_offset
128- if remaining <= 0 :
129- self ._loop .remove_writer (self ._fileno )
130- if self ._closing :
131- self ._call_connection_lost (None )
132- return
139+ """Called when socket is ready for writing.
140+
141+ Drains buffer until EAGAIN with a budget to avoid starvation.
142+ """
143+ for _ in range (self ._max_writes_per_call ):
144+ remaining = len (self ._buffer ) - self ._buffer_offset
145+ if remaining <= 0 :
146+ self ._loop .remove_writer (self ._fileno )
147+ if self ._closing :
148+ self ._call_connection_lost (None )
149+ return
133150
134- try :
135- # Use memoryview with offset for O(1) access to remaining data
136- data_view = memoryview (self ._buffer )[self ._buffer_offset :]
137- n = self ._sock .send (data_view )
138- except (BlockingIOError , InterruptedError ):
139- return
140- except Exception as exc :
141- self ._loop .remove_writer (self ._fileno )
142- self ._fatal_error (exc , 'Fatal write error' )
143- return
151+ try :
152+ # Use memoryview with offset for O(1) access to remaining data
153+ data_view = memoryview (self ._buffer )[self ._buffer_offset :]
154+ n = self ._sock .send (data_view )
155+ except (BlockingIOError , InterruptedError ):
156+ # EAGAIN - socket buffer full
157+ return
158+ except Exception as exc :
159+ self ._loop .remove_writer (self ._fileno )
160+ self ._fatal_error (exc , 'Fatal write error' )
161+ return
144162
145- if n :
146- self ._buffer_offset += n # O(1) offset update instead of O(n) deletion
163+ if n :
164+ self ._buffer_offset += n # O(1) offset update instead of O(n) deletion
147165
148- # Check if buffer is fully consumed
166+ # Check if buffer is fully consumed after budget exhausted
149167 if self ._buffer_offset >= len (self ._buffer ):
150168 # Reset buffer when fully consumed
151169 self ._buffer = self ._buffer_factory ()
@@ -258,6 +276,9 @@ class ErlangDatagramTransport(transports.DatagramTransport):
258276
259277 max_size = 256 * 1024 # 256 KB
260278
279+ # Maximum reads per callback to avoid starving other events
280+ _max_reads_per_call = 16
281+
261282 def __init__ (self , loop , sock , protocol , address = None , extra = None ):
262283 super ().__init__ (extra )
263284 self ._loop = loop
@@ -282,21 +303,27 @@ async def _start(self):
282303 self ._loop .add_reader (self ._fileno , self ._read_ready )
283304
284305 def _read_ready (self ):
285- """Called when data is available to read."""
306+ """Called when data is available to read.
307+
308+ Drains socket until EAGAIN with a budget to avoid starvation.
309+ """
286310 if self ._conn_lost :
287311 return
288- try :
289- data , addr = self ._sock .recvfrom (self .max_size )
290- except (BlockingIOError , InterruptedError ):
291- return
292- except OSError as exc :
293- self ._protocol .error_received (exc )
294- return
295- except Exception as exc :
296- self ._fatal_error (exc , 'Fatal read error on datagram transport' )
297- return
298312
299- self ._protocol .datagram_received (data , addr )
313+ for _ in range (self ._max_reads_per_call ):
314+ try :
315+ data , addr = self ._sock .recvfrom (self .max_size )
316+ except (BlockingIOError , InterruptedError ):
317+ # EAGAIN - no more data available
318+ return
319+ except OSError as exc :
320+ self ._protocol .error_received (exc )
321+ return
322+ except Exception as exc :
323+ self ._fatal_error (exc , 'Fatal read error on datagram transport' )
324+ return
325+
326+ self ._protocol .datagram_received (data , addr )
300327
301328 def sendto (self , data , addr = None ):
302329 """Send data to the transport."""
0 commit comments