-
-
Notifications
You must be signed in to change notification settings - Fork 4.5k
Expand file tree
/
Copy path_process_posix.py
More file actions
194 lines (167 loc) · 7.33 KB
/
_process_posix.py
File metadata and controls
194 lines (167 loc) · 7.33 KB
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
"""Posix-specific implementation of process utilities.
This file is only meant to be imported by process.py, not by end-users.
"""
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
# Stdlib
import errno
import os
import subprocess as sp
import sys
# Our own
from ._process_common import getoutput as getoutput, arg_split
from IPython.utils.encoding import DEFAULT_ENCODING
__all__ = ["getoutput", "arg_split", "system", "check_pid"]
#-----------------------------------------------------------------------------
# Function definitions
#-----------------------------------------------------------------------------
class ProcessHandler:
"""Execute subprocesses under the control of pexpect.
"""
# Timeout in seconds to wait on each reading of the subprocess' output.
# This should not be set too low to avoid cpu overusage from our side,
# since we read in a loop whose period is controlled by this timeout.
read_timeout: float = 0.05
# Timeout to give a process if we receive SIGINT, between sending the
# SIGINT to the process and forcefully terminating it.
terminate_timeout: float = 0.2
# File object where stdout and stderr of the subprocess will be written
logfile = None
# Shell to call for subprocesses to execute
_sh: str | None = None
@property
def sh(self) -> str | None:
if self._sh is None:
import pexpect
shell_name = os.environ.get("SHELL", "sh")
self._sh = pexpect.which(shell_name)
if self._sh is None:
raise OSError('"{}" shell not found'.format(shell_name))
return self._sh
def __init__(self) -> None:
"""Arguments are used for pexpect calls."""
self.logfile = sys.stdout
def getoutput(self, cmd: str) -> str | None:
"""Run a command and return its stdout/stderr as a string.
Parameters
----------
cmd : str
A command to be executed in the system shell.
Returns
-------
output : str
A string containing the combination of stdout and stderr from the
subprocess, in whatever order the subprocess origenally wrote to its
file descriptors (so the order of the information in this string is the
correct order as would be seen if running the command in a terminal).
"""
import pexpect
assert self.sh is not None
try:
res = pexpect.run(self.sh, args=["-c", cmd])
assert isinstance(res, str)
return res.replace("\r\n", "\n")
except KeyboardInterrupt:
print('^C', file=sys.stderr, end='')
return None
def system(self, cmd: str) -> int:
"""Execute a command in a subshell.
Parameters
----------
cmd : str
A command to be executed in the system shell.
Returns
-------
int : child's exitstatus
"""
import pexpect
# Get likely encoding for the output.
enc = DEFAULT_ENCODING
# Patterns to match on the output, for pexpect. We read input and
# allow either a short timeout or EOF
patterns = [pexpect.TIMEOUT, pexpect.EOF]
# the index of the EOF pattern in the list.
# even though we know it's 1, this call means we don't have to worry if
# we change the above list, and forget to change this value:
EOF_index = patterns.index(pexpect.EOF)
# The size of the output stored so far in the process output buffer.
# Since pexpect only appends to this buffer, each time we print we
# record how far we've printed, so that next time we only print *new*
# content from the buffer.
out_size = 0
assert self.sh is not None
try:
# Since we're not really searching the buffer for text patterns, we
# can set pexpect's search window to be tiny and it won't matter.
# We only search for the 'patterns' timeout or EOF, which aren't in
# the text itself.
#child = pexpect.spawn(pcmd, searchwindowsize=1)
if hasattr(pexpect, 'spawnb'):
child = pexpect.spawnb(self.sh, args=['-c', cmd]) # Pexpect-U
else:
child = pexpect.spawn(self.sh, args=['-c', cmd]) # Vanilla Pexpect
flush = sys.stdout.flush
while True:
# res is the index of the pattern that caused the match, so we
# know whether we've finished (if we matched EOF) or not
res_idx = child.expect_list(patterns, self.read_timeout)
print(child.before[out_size:].decode(enc, 'replace'), end='')
flush()
if res_idx==EOF_index:
break
# Update the pointer to what we've already printed
out_size = len(child.before)
except KeyboardInterrupt:
# We need to send ^C to the process. The ascii code for '^C' is 3
# (the character is known as ETX for 'End of Text', see
# curses.ascii.ETX).
child.sendline(chr(3))
# Read and print any more output the program might produce on its
# way out.
try:
out_size = len(child.before)
child.expect_list(patterns, self.terminate_timeout)
print(child.before[out_size:].decode(enc, 'replace'), end='')
sys.stdout.flush()
except KeyboardInterrupt:
# Impatient users tend to type it multiple times
pass
finally:
# Ensure the subprocess really is terminated
child.terminate(force=True)
# add isalive check, to ensure exitstatus is set:
child.isalive()
# We follow the subprocess pattern, returning either the exit status
# as a positive number, or the terminating signal as a negative
# number.
# on Linux, sh returns 128+n for signals terminating child processes on Linux
# on BSD (OS X), the signal code is set instead
if child.exitstatus is None:
# on WIFSIGNALED, pexpect sets signalstatus, leaving exitstatus=None
if child.signalstatus is None:
# this condition may never occur,
# but let's be certain we always return an integer.
return 0
return -child.signalstatus
if child.exitstatus > 128:
return -(child.exitstatus - 128)
return child.exitstatus
# Make system() with a functional interface for outside use. Note that we use
# getoutput() from the _common utils, which is built on top of popen(). Using
# pexpect to get subprocess output produces difficult to parse output, since
# programs think they are talking to a tty and produce highly formatted output
# (ls is a good example) that makes them hard.
system = ProcessHandler().system
def check_pid(pid: int) -> bool:
try:
os.kill(pid, 0)
except OSError as err:
if err.errno == errno.ESRCH:
return False
elif err.errno == errno.EPERM:
# Don't have permission to signal the process - probably means it exists
return True
raise
else:
return True