Coverage for Lib/asyncio/base_subprocess.py: 84%
222 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 03:29 +0000
1import collections
2import subprocess
3import warnings
4import os
5import signal
6import sys
8from . import protocols
9from . import transports
10from .log import logger
13class BaseSubprocessTransport(transports.SubprocessTransport):
15 def __init__(self, loop, protocol, args, shell,
16 stdin, stdout, stderr, bufsize,
17 waiter=None, extra=None, **kwargs):
18 super().__init__(extra)
19 self._closed = False
20 self._protocol = protocol
21 self._loop = loop
22 self._proc = None
23 self._pid = None
24 self._returncode = None
25 self._exit_waiters = set()
26 self._pending_calls = collections.deque()
27 self._pipes = {}
28 self._finished = False
30 if stdin == subprocess.PIPE:
31 self._pipes[0] = None
32 if stdout == subprocess.PIPE:
33 self._pipes[1] = None
34 if stderr == subprocess.PIPE:
35 self._pipes[2] = None
37 # Create the child process: set the _proc attribute
38 try:
39 self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
40 stderr=stderr, bufsize=bufsize, **kwargs)
41 except:
42 self.close()
43 raise
45 self._pid = self._proc.pid
46 self._extra['subprocess'] = self._proc
48 if self._loop.get_debug(): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true
49 if isinstance(args, (bytes, str)):
50 program = args
51 else:
52 program = args[0]
53 logger.debug('process %r created: pid %s',
54 program, self._pid)
56 self._loop.create_task(self._connect_pipes(waiter))
58 def __repr__(self):
59 info = [self.__class__.__name__]
60 if self._closed: 60 ↛ 61line 60 didn't jump to line 61 because the condition on line 60 was never true
61 info.append('closed')
62 if self._pid is not None:
63 info.append(f'pid={self._pid}')
64 if self._returncode is not None:
65 info.append(f'returncode={self._returncode}')
66 elif self._pid is not None:
67 info.append('running')
68 else:
69 info.append('not started')
71 stdin = self._pipes.get(0)
72 if stdin is not None: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 info.append(f'stdin={stdin.pipe}')
75 stdout = self._pipes.get(1)
76 stderr = self._pipes.get(2)
77 if stdout is not None and stderr is stdout: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true
78 info.append(f'stdout=stderr={stdout.pipe}')
79 else:
80 if stdout is not None: 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 info.append(f'stdout={stdout.pipe}')
82 if stderr is not None: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 info.append(f'stderr={stderr.pipe}')
85 return '<{}>'.format(' '.join(info))
87 def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
88 raise NotImplementedError
90 def set_protocol(self, protocol):
91 self._protocol = protocol
93 def get_protocol(self):
94 return self._protocol
96 def is_closing(self):
97 return self._closed
99 def close(self):
100 if self._closed:
101 return
102 self._closed = True
104 for proto in self._pipes.values():
105 if proto is None:
106 continue
107 # See gh-114177
108 # skip closing the pipe if loop is already closed
109 # this can happen e.g. when loop is closed immediately after
110 # process is killed
111 if self._loop and not self._loop.is_closed():
112 proto.pipe.close()
114 if (self._proc is not None and
115 # has the child process finished?
116 self._returncode is None and
117 # the child process has finished, but the
118 # transport hasn't been notified yet?
119 self._proc.poll() is None):
121 if self._loop.get_debug(): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 logger.warning('Close running child process: kill %r', self)
124 try:
125 self._proc.kill()
126 except (ProcessLookupError, PermissionError):
127 # the process may have already exited or may be running setuid
128 pass
130 # Don't clear the _proc reference yet: _post_init() may still run
132 def __del__(self, _warn=warnings.warn):
133 if not self._closed: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
135 self.close()
137 def get_pid(self):
138 return self._pid
140 def get_returncode(self):
141 return self._returncode
143 def get_pipe_transport(self, fd):
144 if fd in self._pipes:
145 return self._pipes[fd].pipe
146 else:
147 return None
149 def _check_proc(self):
150 if self._proc is None:
151 raise ProcessLookupError()
153 if sys.platform == 'win32': 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 def send_signal(self, signal):
155 self._check_proc()
156 self._proc.send_signal(signal)
158 def terminate(self):
159 self._check_proc()
160 self._proc.terminate()
162 def kill(self):
163 self._check_proc()
164 self._proc.kill()
165 else:
166 def send_signal(self, signal):
167 self._check_proc()
168 if self._returncode is not None: 168 ↛ 170line 168 didn't jump to line 170 because the condition on line 168 was never true
169 # The process already exited
170 return
171 try:
172 os.kill(self._proc.pid, signal)
173 except ProcessLookupError:
174 pass
176 def terminate(self):
177 self.send_signal(signal.SIGTERM)
179 def kill(self):
180 self.send_signal(signal.SIGKILL)
182 async def _connect_pipes(self, waiter):
183 try:
184 proc = self._proc
185 loop = self._loop
187 if proc.stdin is not None:
188 _, pipe = await loop.connect_write_pipe(
189 lambda: WriteSubprocessPipeProto(self, 0),
190 proc.stdin)
191 self._pipes[0] = pipe
193 if proc.stdout is not None:
194 _, pipe = await loop.connect_read_pipe(
195 lambda: ReadSubprocessPipeProto(self, 1),
196 proc.stdout)
197 self._pipes[1] = pipe
199 if proc.stderr is not None:
200 _, pipe = await loop.connect_read_pipe(
201 lambda: ReadSubprocessPipeProto(self, 2),
202 proc.stderr)
203 self._pipes[2] = pipe
205 assert self._pending_calls is not None
207 loop.call_soon(self._protocol.connection_made, self)
208 for callback, data in self._pending_calls:
209 loop.call_soon(callback, *data)
210 self._pending_calls = None
211 except (SystemExit, KeyboardInterrupt):
212 raise
213 except BaseException as exc:
214 # Close any pipes that were already connected before the
215 # error/cancellation to avoid leaking file descriptors.
216 for proto in self._pipes.values():
217 if proto is not None: 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 proto.pipe.close()
219 for raw_pipe in (proc.stdin, proc.stdout, proc.stderr):
220 if raw_pipe is not None:
221 raw_pipe.close()
222 if waiter is not None and not waiter.cancelled(): 222 ↛ exitline 222 didn't return from function '_connect_pipes' because the condition on line 222 was always true
223 waiter.set_exception(exc)
224 else:
225 if waiter is not None and not waiter.cancelled():
226 waiter.set_result(None)
228 def _call(self, cb, *data):
229 if self._pending_calls is not None:
230 self._pending_calls.append((cb, data))
231 else:
232 self._loop.call_soon(cb, *data)
234 def _pipe_connection_lost(self, fd, exc):
235 self._call(self._protocol.pipe_connection_lost, fd, exc)
236 self._try_finish()
238 def _pipe_data_received(self, fd, data):
239 self._call(self._protocol.pipe_data_received, fd, data)
241 def _process_exited(self, returncode):
242 assert returncode is not None, returncode
243 assert self._returncode is None, self._returncode
244 if self._loop.get_debug(): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 logger.info('%r exited with return code %r', self, returncode)
246 self._returncode = returncode
248 if self._proc.returncode is None:
249 # asyncio uses a child watcher: copy the status into the Popen
250 # object. On Python 3.6, it is required to avoid a ResourceWarning.
251 self._proc.returncode = returncode
252 self._call(self._protocol.process_exited)
254 self._try_finish()
256 # gh-119710: Wake up futures waiting for wait() as soon as the process
257 # exits.
258 for waiter in self._exit_waiters:
259 if not waiter.done(): 259 ↛ 258line 259 didn't jump to line 258 because the condition on line 259 was always true
260 waiter.set_result(returncode)
261 self._exit_waiters = None
263 async def _wait(self):
264 """Wait until the process exit and return the process return code.
266 This method is a coroutine."""
267 if self._returncode is not None:
268 return self._returncode
270 waiter = self._loop.create_future()
271 self._exit_waiters.add(waiter)
272 try:
273 return await waiter
274 finally:
275 if self._exit_waiters is not None:
276 self._exit_waiters.discard(waiter)
278 def _try_finish(self):
279 assert not self._finished
280 if self._returncode is None:
281 return
283 if all(p is not None and p.disconnected
284 for p in self._pipes.values()):
285 self._finished = True
286 self._call(self._call_connection_lost, None)
288 def _call_connection_lost(self, exc):
289 try:
290 self._protocol.connection_lost(exc)
291 finally:
292 self._loop = None
293 self._proc = None
294 self._protocol = None
297class WriteSubprocessPipeProto(protocols.BaseProtocol):
299 def __init__(self, proc, fd):
300 self.proc = proc
301 self.fd = fd
302 self.pipe = None
303 self.disconnected = False
305 def connection_made(self, transport):
306 self.pipe = transport
308 def __repr__(self):
309 return f'<{self.__class__.__name__} fd={self.fd} pipe={self.pipe!r}>'
311 def connection_lost(self, exc):
312 self.disconnected = True
313 self.proc._pipe_connection_lost(self.fd, exc)
314 self.proc = None
316 def pause_writing(self):
317 self.proc._protocol.pause_writing()
319 def resume_writing(self):
320 self.proc._protocol.resume_writing()
323class ReadSubprocessPipeProto(WriteSubprocessPipeProto,
324 protocols.Protocol):
326 def data_received(self, data):
327 self.proc._pipe_data_received(self.fd, data)