ยปCore Development>Code coverage>Lib/asyncio/base_subprocess.py

Python code coverage for Lib/asyncio/base_subprocess.py

#countcontent
1n/aimport collections
2n/aimport subprocess
3n/aimport warnings
4n/a
5n/afrom . import compat
6n/afrom . import protocols
7n/afrom . import transports
8n/afrom .coroutines import coroutine
9n/afrom .log import logger
10n/a
11n/a
12n/aclass BaseSubprocessTransport(transports.SubprocessTransport):
13n/a
14n/a def __init__(self, loop, protocol, args, shell,
15n/a stdin, stdout, stderr, bufsize,
16n/a waiter=None, extra=None, **kwargs):
17n/a super().__init__(extra)
18n/a self._closed = False
19n/a self._protocol = protocol
20n/a self._loop = loop
21n/a self._proc = None
22n/a self._pid = None
23n/a self._returncode = None
24n/a self._exit_waiters = []
25n/a self._pending_calls = collections.deque()
26n/a self._pipes = {}
27n/a self._finished = False
28n/a
29n/a if stdin == subprocess.PIPE:
30n/a self._pipes[0] = None
31n/a if stdout == subprocess.PIPE:
32n/a self._pipes[1] = None
33n/a if stderr == subprocess.PIPE:
34n/a self._pipes[2] = None
35n/a
36n/a # Create the child process: set the _proc attribute
37n/a try:
38n/a self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
39n/a stderr=stderr, bufsize=bufsize, **kwargs)
40n/a except:
41n/a self.close()
42n/a raise
43n/a
44n/a self._pid = self._proc.pid
45n/a self._extra['subprocess'] = self._proc
46n/a
47n/a if self._loop.get_debug():
48n/a if isinstance(args, (bytes, str)):
49n/a program = args
50n/a else:
51n/a program = args[0]
52n/a logger.debug('process %r created: pid %s',
53n/a program, self._pid)
54n/a
55n/a self._loop.create_task(self._connect_pipes(waiter))
56n/a
57n/a def __repr__(self):
58n/a info = [self.__class__.__name__]
59n/a if self._closed:
60n/a info.append('closed')
61n/a if self._pid is not None:
62n/a info.append('pid=%s' % self._pid)
63n/a if self._returncode is not None:
64n/a info.append('returncode=%s' % self._returncode)
65n/a elif self._pid is not None:
66n/a info.append('running')
67n/a else:
68n/a info.append('not started')
69n/a
70n/a stdin = self._pipes.get(0)
71n/a if stdin is not None:
72n/a info.append('stdin=%s' % stdin.pipe)
73n/a
74n/a stdout = self._pipes.get(1)
75n/a stderr = self._pipes.get(2)
76n/a if stdout is not None and stderr is stdout:
77n/a info.append('stdout=stderr=%s' % stdout.pipe)
78n/a else:
79n/a if stdout is not None:
80n/a info.append('stdout=%s' % stdout.pipe)
81n/a if stderr is not None:
82n/a info.append('stderr=%s' % stderr.pipe)
83n/a
84n/a return '<%s>' % ' '.join(info)
85n/a
86n/a def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
87n/a raise NotImplementedError
88n/a
89n/a def set_protocol(self, protocol):
90n/a self._protocol = protocol
91n/a
92n/a def get_protocol(self):
93n/a return self._protocol
94n/a
95n/a def is_closing(self):
96n/a return self._closed
97n/a
98n/a def close(self):
99n/a if self._closed:
100n/a return
101n/a self._closed = True
102n/a
103n/a for proto in self._pipes.values():
104n/a if proto is None:
105n/a continue
106n/a proto.pipe.close()
107n/a
108n/a if (self._proc is not None
109n/a # the child process finished?
110n/a and self._returncode is None
111n/a # the child process finished but the transport was not notified yet?
112n/a and self._proc.poll() is None
113n/a ):
114n/a if self._loop.get_debug():
115n/a logger.warning('Close running child process: kill %r', self)
116n/a
117n/a try:
118n/a self._proc.kill()
119n/a except ProcessLookupError:
120n/a pass
121n/a
122n/a # Don't clear the _proc reference yet: _post_init() may still run
123n/a
124n/a # On Python 3.3 and older, objects with a destructor part of a reference
125n/a # cycle are never destroyed. It's not more the case on Python 3.4 thanks
126n/a # to the PEP 442.
127n/a if compat.PY34:
128n/a def __del__(self):
129n/a if not self._closed:
130n/a warnings.warn("unclosed transport %r" % self, ResourceWarning,
131n/a source=self)
132n/a self.close()
133n/a
134n/a def get_pid(self):
135n/a return self._pid
136n/a
137n/a def get_returncode(self):
138n/a return self._returncode
139n/a
140n/a def get_pipe_transport(self, fd):
141n/a if fd in self._pipes:
142n/a return self._pipes[fd].pipe
143n/a else:
144n/a return None
145n/a
146n/a def _check_proc(self):
147n/a if self._proc is None:
148n/a raise ProcessLookupError()
149n/a
150n/a def send_signal(self, signal):
151n/a self._check_proc()
152n/a self._proc.send_signal(signal)
153n/a
154n/a def terminate(self):
155n/a self._check_proc()
156n/a self._proc.terminate()
157n/a
158n/a def kill(self):
159n/a self._check_proc()
160n/a self._proc.kill()
161n/a
162n/a @coroutine
163n/a def _connect_pipes(self, waiter):
164n/a try:
165n/a proc = self._proc
166n/a loop = self._loop
167n/a
168n/a if proc.stdin is not None:
169n/a _, pipe = yield from loop.connect_write_pipe(
170n/a lambda: WriteSubprocessPipeProto(self, 0),
171n/a proc.stdin)
172n/a self._pipes[0] = pipe
173n/a
174n/a if proc.stdout is not None:
175n/a _, pipe = yield from loop.connect_read_pipe(
176n/a lambda: ReadSubprocessPipeProto(self, 1),
177n/a proc.stdout)
178n/a self._pipes[1] = pipe
179n/a
180n/a if proc.stderr is not None:
181n/a _, pipe = yield from loop.connect_read_pipe(
182n/a lambda: ReadSubprocessPipeProto(self, 2),
183n/a proc.stderr)
184n/a self._pipes[2] = pipe
185n/a
186n/a assert self._pending_calls is not None
187n/a
188n/a loop.call_soon(self._protocol.connection_made, self)
189n/a for callback, data in self._pending_calls:
190n/a loop.call_soon(callback, *data)
191n/a self._pending_calls = None
192n/a except Exception as exc:
193n/a if waiter is not None and not waiter.cancelled():
194n/a waiter.set_exception(exc)
195n/a else:
196n/a if waiter is not None and not waiter.cancelled():
197n/a waiter.set_result(None)
198n/a
199n/a def _call(self, cb, *data):
200n/a if self._pending_calls is not None:
201n/a self._pending_calls.append((cb, data))
202n/a else:
203n/a self._loop.call_soon(cb, *data)
204n/a
205n/a def _pipe_connection_lost(self, fd, exc):
206n/a self._call(self._protocol.pipe_connection_lost, fd, exc)
207n/a self._try_finish()
208n/a
209n/a def _pipe_data_received(self, fd, data):
210n/a self._call(self._protocol.pipe_data_received, fd, data)
211n/a
212n/a def _process_exited(self, returncode):
213n/a assert returncode is not None, returncode
214n/a assert self._returncode is None, self._returncode
215n/a if self._loop.get_debug():
216n/a logger.info('%r exited with return code %r',
217n/a self, returncode)
218n/a self._returncode = returncode
219n/a if self._proc.returncode is None:
220n/a # asyncio uses a child watcher: copy the status into the Popen
221n/a # object. On Python 3.6, it is required to avoid a ResourceWarning.
222n/a self._proc.returncode = returncode
223n/a self._call(self._protocol.process_exited)
224n/a self._try_finish()
225n/a
226n/a # wake up futures waiting for wait()
227n/a for waiter in self._exit_waiters:
228n/a if not waiter.cancelled():
229n/a waiter.set_result(returncode)
230n/a self._exit_waiters = None
231n/a
232n/a @coroutine
233n/a def _wait(self):
234n/a """Wait until the process exit and return the process return code.
235n/a
236n/a This method is a coroutine."""
237n/a if self._returncode is not None:
238n/a return self._returncode
239n/a
240n/a waiter = self._loop.create_future()
241n/a self._exit_waiters.append(waiter)
242n/a return (yield from waiter)
243n/a
244n/a def _try_finish(self):
245n/a assert not self._finished
246n/a if self._returncode is None:
247n/a return
248n/a if all(p is not None and p.disconnected
249n/a for p in self._pipes.values()):
250n/a self._finished = True
251n/a self._call(self._call_connection_lost, None)
252n/a
253n/a def _call_connection_lost(self, exc):
254n/a try:
255n/a self._protocol.connection_lost(exc)
256n/a finally:
257n/a self._loop = None
258n/a self._proc = None
259n/a self._protocol = None
260n/a
261n/a
262n/aclass WriteSubprocessPipeProto(protocols.BaseProtocol):
263n/a
264n/a def __init__(self, proc, fd):
265n/a self.proc = proc
266n/a self.fd = fd
267n/a self.pipe = None
268n/a self.disconnected = False
269n/a
270n/a def connection_made(self, transport):
271n/a self.pipe = transport
272n/a
273n/a def __repr__(self):
274n/a return ('<%s fd=%s pipe=%r>'
275n/a % (self.__class__.__name__, self.fd, self.pipe))
276n/a
277n/a def connection_lost(self, exc):
278n/a self.disconnected = True
279n/a self.proc._pipe_connection_lost(self.fd, exc)
280n/a self.proc = None
281n/a
282n/a def pause_writing(self):
283n/a self.proc._protocol.pause_writing()
284n/a
285n/a def resume_writing(self):
286n/a self.proc._protocol.resume_writing()
287n/a
288n/a
289n/aclass ReadSubprocessPipeProto(WriteSubprocessPipeProto,
290n/a protocols.Protocol):
291n/a
292n/a def data_received(self, data):
293n/a self.proc._pipe_data_received(self.fd, data)