Coverage for Lib/asyncio/transports.py: 88%
113 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-04-15 02:02 +0000
1"""Abstract Transport class."""
3__all__ = (
4 'BaseTransport', 'ReadTransport', 'WriteTransport',
5 'Transport', 'DatagramTransport', 'SubprocessTransport',
6)
9class BaseTransport:
10 """Base class for transports."""
12 __slots__ = ('_extra',)
14 def __init__(self, extra=None):
15 if extra is None:
16 extra = {}
17 self._extra = extra
19 def get_extra_info(self, name, default=None):
20 """Get optional transport information."""
21 return self._extra.get(name, default)
23 def is_closing(self):
24 """Return True if the transport is closing or closed."""
25 raise NotImplementedError
27 def close(self):
28 """Close the transport.
30 Buffered data will be flushed asynchronously. No more data
31 will be received. After all buffered data is flushed, the
32 protocol's connection_lost() method will (eventually) be
33 called with None as its argument.
34 """
35 raise NotImplementedError
37 def set_protocol(self, protocol):
38 """Set a new protocol."""
39 raise NotImplementedError
41 def get_protocol(self):
42 """Return the current protocol."""
43 raise NotImplementedError
46class ReadTransport(BaseTransport):
47 """Interface for read-only transports."""
49 __slots__ = ()
51 def is_reading(self):
52 """Return True if the transport is receiving."""
53 raise NotImplementedError
55 def pause_reading(self):
56 """Pause the receiving end.
58 No data will be passed to the protocol's data_received()
59 method until resume_reading() is called.
60 """
61 raise NotImplementedError
63 def resume_reading(self):
64 """Resume the receiving end.
66 Data received will once again be passed to the protocol's
67 data_received() method.
68 """
69 raise NotImplementedError
72class WriteTransport(BaseTransport):
73 """Interface for write-only transports."""
75 __slots__ = ()
77 def set_write_buffer_limits(self, high=None, low=None):
78 """Set the high- and low-water limits for write flow control.
80 These two values control when to call the protocol's
81 pause_writing() and resume_writing() methods. If specified,
82 the low-water limit must be less than or equal to the
83 high-water limit. Neither value can be negative.
85 The defaults are implementation-specific. If only the
86 high-water limit is given, the low-water limit defaults to an
87 implementation-specific value less than or equal to the
88 high-water limit. Setting high to zero forces low to zero as
89 well, and causes pause_writing() to be called whenever the
90 buffer becomes non-empty. Setting low to zero causes
91 resume_writing() to be called only once the buffer is empty.
92 Use of zero for either limit is generally sub-optimal as it
93 reduces opportunities for doing I/O and computation
94 concurrently.
95 """
96 raise NotImplementedError
98 def get_write_buffer_size(self):
99 """Return the current size of the write buffer."""
100 raise NotImplementedError
102 def get_write_buffer_limits(self):
103 """Get the high and low watermarks for write flow control.
104 Return a tuple (low, high) where low and high are
105 positive number of bytes."""
106 raise NotImplementedError
108 def write(self, data):
109 """Write some data bytes to the transport.
111 This does not block; it buffers the data and arranges for it
112 to be sent out asynchronously.
113 """
114 raise NotImplementedError
116 def writelines(self, list_of_data):
117 """Write a list (or any iterable) of data bytes to the transport.
119 The default implementation concatenates the arguments and
120 calls write() on the result.
121 """
122 data = b''.join(list_of_data)
123 self.write(data)
125 def write_eof(self):
126 """Close the write end after flushing buffered data.
128 (This is like typing ^D into a UNIX program reading from stdin.)
130 Data may still be received.
131 """
132 raise NotImplementedError
134 def can_write_eof(self):
135 """Return True if this transport supports write_eof(), False if not."""
136 raise NotImplementedError
138 def abort(self):
139 """Close the transport immediately.
141 Buffered data will be lost. No more data will be received.
142 The protocol's connection_lost() method will (eventually) be
143 called with None as its argument.
144 """
145 raise NotImplementedError
148class Transport(ReadTransport, WriteTransport):
149 """Interface representing a bidirectional transport.
151 There may be several implementations, but typically, the user does
152 not implement new transports; rather, the platform provides some
153 useful transports that are implemented using the platform's best
154 practices.
156 The user never instantiates a transport directly; they call a
157 utility function, passing it a protocol factory and other
158 information necessary to create the transport and protocol. (E.g.
159 EventLoop.create_connection() or EventLoop.create_server().)
161 The utility function will asynchronously create a transport and a
162 protocol and hook them up by calling the protocol's
163 connection_made() method, passing it the transport.
165 The implementation here raises NotImplemented for every method
166 except writelines(), which calls write() in a loop.
167 """
169 __slots__ = ()
172class DatagramTransport(BaseTransport):
173 """Interface for datagram (UDP) transports."""
175 __slots__ = ()
177 def sendto(self, data, addr=None):
178 """Send data to the transport.
180 This does not block; it buffers the data and arranges for it
181 to be sent out asynchronously.
182 addr is target socket address.
183 If addr is None use target address pointed on transport creation.
184 If data is an empty bytes object a zero-length datagram will be
185 sent.
186 """
187 raise NotImplementedError
189 def abort(self):
190 """Close the transport immediately.
192 Buffered data will be lost. No more data will be received.
193 The protocol's connection_lost() method will (eventually) be
194 called with None as its argument.
195 """
196 raise NotImplementedError
199class SubprocessTransport(BaseTransport):
201 __slots__ = ()
203 def get_pid(self):
204 """Get subprocess id."""
205 raise NotImplementedError
207 def get_returncode(self):
208 """Get subprocess returncode.
210 See also
211 http://docs.python.org/3/library/subprocess#subprocess.Popen.returncode
212 """
213 raise NotImplementedError
215 def get_pipe_transport(self, fd):
216 """Get transport for pipe with number fd."""
217 raise NotImplementedError
219 def send_signal(self, signal):
220 """Send signal to subprocess.
222 See also:
223 docs.python.org/3/library/subprocess#subprocess.Popen.send_signal
224 """
225 raise NotImplementedError
227 def terminate(self):
228 """Stop the subprocess.
230 Alias for close() method.
232 On Posix OSs the method sends SIGTERM to the subprocess.
233 On Windows the Win32 API function TerminateProcess()
234 is called to stop the subprocess.
236 See also:
237 http://docs.python.org/3/library/subprocess#subprocess.Popen.terminate
238 """
239 raise NotImplementedError
241 def kill(self):
242 """Kill the subprocess.
244 On Posix OSs the function sends SIGKILL to the subprocess.
245 On Windows kill() is an alias for terminate().
247 See also:
248 http://docs.python.org/3/library/subprocess#subprocess.Popen.kill
249 """
250 raise NotImplementedError
253class _FlowControlMixin(Transport):
254 """All the logic for (write) flow control in a mix-in base class.
256 The subclass must implement get_write_buffer_size(). It must call
257 _maybe_pause_protocol() whenever the write buffer size increases,
258 and _maybe_resume_protocol() whenever it decreases. It may also
259 override set_write_buffer_limits() (e.g. to specify different
260 defaults).
262 The subclass constructor must call super().__init__(extra). This
263 will call set_write_buffer_limits().
265 The user may call set_write_buffer_limits() and
266 get_write_buffer_size(), and their protocol's pause_writing() and
267 resume_writing() may be called.
268 """
270 __slots__ = ('_loop', '_protocol_paused', '_high_water', '_low_water')
272 def __init__(self, extra=None, loop=None):
273 super().__init__(extra)
274 assert loop is not None
275 self._loop = loop
276 self._protocol_paused = False
277 self._set_write_buffer_limits()
279 def _maybe_pause_protocol(self):
280 size = self.get_write_buffer_size()
281 if size <= self._high_water:
282 return
283 if not self._protocol_paused: 283 ↛ exitline 283 didn't return from function '_maybe_pause_protocol' because the condition on line 283 was always true
284 self._protocol_paused = True
285 try:
286 self._protocol.pause_writing()
287 except (SystemExit, KeyboardInterrupt):
288 raise
289 except BaseException as exc:
290 self._loop.call_exception_handler({
291 'message': 'protocol.pause_writing() failed',
292 'exception': exc,
293 'transport': self,
294 'protocol': self._protocol,
295 })
297 def _maybe_resume_protocol(self):
298 if (self._protocol_paused and
299 self.get_write_buffer_size() <= self._low_water):
300 self._protocol_paused = False
301 try:
302 self._protocol.resume_writing()
303 except (SystemExit, KeyboardInterrupt):
304 raise
305 except BaseException as exc:
306 self._loop.call_exception_handler({
307 'message': 'protocol.resume_writing() failed',
308 'exception': exc,
309 'transport': self,
310 'protocol': self._protocol,
311 })
313 def get_write_buffer_limits(self):
314 return (self._low_water, self._high_water)
316 def _set_write_buffer_limits(self, high=None, low=None):
317 if high is None:
318 if low is None: 318 ↛ 321line 318 didn't jump to line 321 because the condition on line 318 was always true
319 high = 64 * 1024
320 else:
321 high = 4 * low
322 if low is None:
323 low = high // 4
325 if not high >= low >= 0:
326 raise ValueError(
327 f'high ({high!r}) must be >= low ({low!r}) must be >= 0')
329 self._high_water = high
330 self._low_water = low
332 def set_write_buffer_limits(self, high=None, low=None):
333 self._set_write_buffer_limits(high=high, low=low)
334 self._maybe_pause_protocol()
336 def get_write_buffer_size(self):
337 raise NotImplementedError