Source code for xonsh.procs.pipelines

"""Command pipeline tools."""
import io
import os
import re
import signal
import subprocess
import sys
import threading
import time

import xonsh.jobs as xj
import xonsh.lazyasd as xl
import xonsh.platform as xp
import xonsh.tools as xt
from xonsh.built_ins import XSH
from xonsh.procs.readers import ConsoleParallelReader, NonBlockingFDReader, safe_fdclose


@xl.lazyobject
def STDOUT_CAPTURE_KINDS():
    return frozenset(["stdout", "object"])


@xl.lazyobject
def RE_HIDDEN_BYTES():
    return re.compile(b"(\001.*?\002)")


@xl.lazyobject
def RE_VT100_ESCAPE():
    return re.compile(b"(\x9B|\x1B\\[)[0-?]*[ -\\/]*[@-~]")


@xl.lazyobject
def RE_HIDE_ESCAPE():
    return re.compile(
        b"(" + RE_HIDDEN_BYTES.pattern + b"|" + RE_VT100_ESCAPE.pattern + b")"
    )


@xl.lazyobject
def SIGNAL_MESSAGES():
    sm = {
        signal.SIGABRT: "Aborted",
        signal.SIGFPE: "Floating point exception",
        signal.SIGILL: "Illegal instructions",
        signal.SIGTERM: "Terminated",
        signal.SIGSEGV: "Segmentation fault",
    }
    if xp.ON_POSIX:
        sm.update(
            {
                signal.SIGQUIT: "Quit",
                signal.SIGHUP: "Hangup",
                signal.SIGKILL: "Killed",
                signal.SIGTSTP: "Stopped",
            }
        )
    return sm


[docs]def safe_readlines(handle, hint=-1): """Attempts to read lines without throwing an error.""" try: lines = handle.readlines(hint) except OSError: lines = [] return lines
[docs]def safe_readable(handle): """Attempts to find if the handle is readable without throwing an error.""" try: status = handle.readable() except (OSError, ValueError): status = False return status
[docs]def update_process_group(pipeline_group, background): if not xp.ON_POSIX: return False env = XSH.env if not env.get("XONSH_INTERACTIVE"): return False if background: return True return xj.give_terminal_to(pipeline_group)
[docs]class CommandPipeline: """Represents a subprocess-mode command pipeline.""" attrnames = ( "stdin", "stdout", "stderr", "pid", "returncode", "args", "alias", "stdin_redirect", "stdout_redirect", "stderr_redirect", "timestamps", "executed_cmd", "input", "output", "errors", ) nonblocking = (io.BytesIO, NonBlockingFDReader, ConsoleParallelReader) def __init__(self, specs): """ Parameters ---------- specs : list of SubprocSpec Process specifications Attributes ---------- spec : SubprocSpec The last specification in specs proc : Popen-like The process in procs ended : bool Boolean for if the command has stopped executing. input : str A string of the standard input. output : str A string of the standard output. errors : str A string of the standard error. lines : list of str The output lines starttime : floats or None Pipeline start timestamp. """ self.starttime = None self.ended = False self.procs = [] self.specs = specs self.spec = specs[-1] self.captured = specs[-1].captured self.input = self._output = self.errors = self.endtime = None self._closed_handle_cache = {} self.lines = [] self._raw_output = self._raw_error = b"" self._stderr_prefix = self._stderr_postfix = None self.term_pgid = None background = self.spec.background pipeline_group = None if xp.ON_POSIX and not xt.on_main_thread(): # If we are inside a ProcProxyThread, then run commands in the same # process group as xonsh. This fixes case 2 of issue #4277, where # the terminal is given to a command inside the ProcProxyThread, # taking the terminal away from the `less` command, causing `less` # to stop. pipeline_group = os.getpgid(0) for spec in specs: if self.starttime is None: self.starttime = time.time() try: proc = spec.run(pipeline_group=pipeline_group) except Exception: xt.print_exception() self._return_terminal() self.proc = None return if ( proc.pid and pipeline_group is None and not spec.is_proxy and self.captured != "object" ): pipeline_group = proc.pid if update_process_group(pipeline_group, background): self.term_pgid = pipeline_group self.procs.append(proc) self.proc = self.procs[-1] def __repr__(self): s = self.__class__.__name__ + "(\n " s += ",\n ".join(a + "=" + repr(getattr(self, a)) for a in self.attrnames) s += "\n)" return s def __bool__(self): return self.returncode == 0 def __int__(self): return self.returncode def __hash__(self): return hash(self.returncode) def __str__(self): self.end() return self.output def __len__(self): return len(self.procs) def __eq__(self, other): if isinstance(other, int): return self.returncode == other elif isinstance(other, str): return str(self) == other raise Exception( f"CommandPipeline doesn't support comparing with {type(other)}." ) def __iter__(self): """Iterates through stdout and returns the lines, converting to strings and universal newlines if needed. """ if self.ended: yield from iter(self.lines) else: yield from self.tee_stdout()
[docs] def iterraw(self): """Iterates through the last stdout, and returns the lines exactly as found. """ # get appropriate handles spec = self.spec proc = self.proc if proc is None: return timeout = XSH.env.get("XONSH_PROC_FREQUENCY") # get the correct stdout stdout = proc.stdout if ( stdout is None or spec.stdout is None or not safe_readable(stdout) ) and spec.captured_stdout is not None: stdout = spec.captured_stdout if hasattr(stdout, "buffer"): stdout = stdout.buffer if stdout is not None and not isinstance(stdout, self.nonblocking): stdout = NonBlockingFDReader(stdout.fileno(), timeout=timeout) if ( not stdout or self.captured == "stdout" or not safe_readable(stdout) or not spec.threadable ): # we get here if the process is not threadable or the # class is the real Popen PrevProcCloser(pipeline=self) task = xj.wait_for_active_job() if task is None or task["status"] != "stopped": proc.wait() self._endtime() if self.captured == "object": self.end(tee_output=False) elif self.captured == "hiddenobject" and stdout: b = stdout.read() lines = b.splitlines(keepends=True) yield from lines self.end(tee_output=False) elif self.captured == "stdout": b = stdout.read() s = self._decode_uninew(b, universal_newlines=True) self.lines = s.splitlines(keepends=True) return # get the correct stderr stderr = proc.stderr if ( stderr is None or spec.stderr is None or not safe_readable(stderr) ) and spec.captured_stderr is not None: stderr = spec.captured_stderr if hasattr(stderr, "buffer"): stderr = stderr.buffer if stderr is not None and not isinstance(stderr, self.nonblocking): stderr = NonBlockingFDReader(stderr.fileno(), timeout=timeout) # read from process while it is running check_prev_done = len(self.procs) == 1 prev_end_time = None i = j = cnt = 1 while proc.poll() is None: if getattr(proc, "suspended", False): return elif getattr(proc, "in_alt_mode", False): time.sleep(0.1) # probably not leaving any time soon continue elif not check_prev_done: # In the case of pipelines with more than one command # we should give the commands a little time # to start up fully. This is particularly true for # GNU Parallel, which has a long startup time. pass elif self._prev_procs_done(): self._close_prev_procs() proc.prevs_are_closed = True break stdout_lines = safe_readlines(stdout, 1024) i = len(stdout_lines) if i != 0: yield from stdout_lines stderr_lines = safe_readlines(stderr, 1024) j = len(stderr_lines) if j != 0: self.stream_stderr(stderr_lines) if not check_prev_done: # if we are piping... if stdout_lines or stderr_lines: # see if we have some output. check_prev_done = True elif prev_end_time is None: # or see if we already know that the next-to-last # proc in the pipeline has ended. if self._prev_procs_done(): # if it has, record the time prev_end_time = time.time() elif time.time() - prev_end_time >= 0.1: # if we still don't have any output, even though the # next-to-last proc has finished, wait a bit to make # sure we have fully started up, etc. check_prev_done = True # this is for CPU usage if i + j == 0: cnt = min(cnt + 1, 1000) else: cnt = 1 time.sleep(timeout * cnt) # read from process now that it is over yield from safe_readlines(stdout) self.stream_stderr(safe_readlines(stderr)) proc.wait() self._endtime() yield from safe_readlines(stdout) self.stream_stderr(safe_readlines(stderr)) if self.captured == "object": self.end(tee_output=False)
[docs] def itercheck(self): """Iterates through the command lines and throws an error if the returncode is non-zero. """ yield from self if self.returncode: # I included self, as providing access to stderr and other details # useful when instance isn't assigned to a variable in the shell. raise xt.XonshCalledProcessError( self.returncode, self.executed_cmd, self.stdout, self.stderr, self )
[docs] def tee_stdout(self): """Writes the process stdout to the output variable, line-by-line, and yields each line. This may optionally accept lines (in bytes) to iterate over, in which case it does not call iterraw(). """ env = XSH.env enc = env.get("XONSH_ENCODING") err = env.get("XONSH_ENCODING_ERRORS") lines = self.lines raw_out_lines = [] stream = self.captured not in STDOUT_CAPTURE_KINDS if stream and not self.spec.stdout: stream = False stdout_has_buffer = hasattr(sys.stdout, "buffer") nl = b"\n" cr = b"\r" crnl = b"\r\n" for line in self.iterraw(): # write to stdout line ASAP, if needed if stream: if stdout_has_buffer: sys.stdout.buffer.write(line) else: sys.stdout.write(line.decode(encoding=enc, errors=err)) sys.stdout.flush() # save the raw bytes raw_out_lines.append(line) # do some munging of the line before we return it if line.endswith(crnl): line = line[:-2] + nl elif line.endswith(cr): line = line[:-1] + nl line = RE_HIDE_ESCAPE.sub(b"", line) line = line.decode(encoding=enc, errors=err) # tee it up! lines.append(line) yield line # using join is more efficient than concatenating in a loop self._raw_output = b"".join(raw_out_lines)
[docs] def stream_stderr(self, lines): """Streams lines to sys.stderr and the errors attribute.""" if not lines: return env = XSH.env enc = env.get("XONSH_ENCODING") err = env.get("XONSH_ENCODING_ERRORS") b = b"".join(lines) if self.stderr_prefix: b = self.stderr_prefix + b if self.stderr_postfix: b += self.stderr_postfix stderr_has_buffer = hasattr(sys.stderr, "buffer") show_stderr = self.captured != "object" or env.get( "XONSH_SUBPROC_CAPTURED_PRINT_STDERR", True ) if show_stderr: # write bytes to std stream if stderr_has_buffer: sys.stderr.buffer.write(b) else: sys.stderr.write(b.decode(encoding=enc, errors=err)) sys.stderr.flush() # save the raw bytes self._raw_error = b # do some munging of the line before we save it to the attr b = b.replace(b"\r\n", b"\n").replace(b"\r", b"\n") b = RE_HIDE_ESCAPE.sub(b"", b) env = XSH.env s = b.decode( encoding=env.get("XONSH_ENCODING"), errors=env.get("XONSH_ENCODING_ERRORS") ) # set the errors if self.errors is None: self.errors = s else: self.errors += s
def _decode_uninew(self, b, universal_newlines=None): """Decode bytes into a str and apply universal newlines as needed.""" if not b: return "" if isinstance(b, (bytes, bytearray)): env = XSH.env s = b.decode( encoding=env.get("XONSH_ENCODING"), errors=env.get("XONSH_ENCODING_ERRORS"), ) else: s = b if universal_newlines or self.spec.universal_newlines: s = s.replace("\r\n", "\n").replace("\r", "\n") return s # # Ending methods #
[docs] def end(self, tee_output=True): """ End the pipeline, return the controlling terminal if needed. Main things done in self._end(). """ if self.ended: return self._end(tee_output=tee_output) self._return_terminal()
def _end(self, tee_output): """Waits for the command to complete and then runs any closing and cleanup procedures that need to be run. """ if tee_output: for _ in self.tee_stdout(): pass self._endtime() # since we are driven by getting output, input may not be available # until the command has completed. self._set_input() self._close_prev_procs() self._close_proc() self._check_signal() self._apply_to_history() self.ended = True self._raise_subproc_error() def _return_terminal(self): if xp.ON_WINDOWS or not xp.ON_POSIX: return pgid = os.getpgid(0) if self.term_pgid is None or pgid == self.term_pgid: return if xj.give_terminal_to(pgid): # if gave term succeed self.term_pgid = pgid if XSH.shell is not None: # restoring sanity could probably be called whenever we return # control to the shell. But it only seems to matter after a # ^Z event. This *has* to be called after we give the terminal # back to the shell. XSH.shell.shell.restore_tty_sanity()
[docs] def resume(self, job, tee_output=True): self.ended = False if xj.give_terminal_to(job["pgrp"]): self.term_pgid = job["pgrp"] xj._continue(job) self.end(tee_output=tee_output)
def _endtime(self): """Sets the closing timestamp if it hasn't been already.""" if self.endtime is None: self.endtime = time.time() def _safe_close(self, handle): safe_fdclose(handle, cache=self._closed_handle_cache) def _prev_procs_done(self): """Boolean for if all previous processes have completed. If there is only a single process in the pipeline, this returns False. """ any_running = False for s, p in zip(self.specs[:-1], self.procs[:-1]): if p.poll() is None: any_running = True continue self._safe_close(s.stdin) self._safe_close(s.stdout) self._safe_close(s.stderr) if p is None: continue self._safe_close(p.stdin) self._safe_close(p.stdout) self._safe_close(p.stderr) return False if any_running else (len(self) > 1) def _close_prev_procs(self): """Closes all but the last proc's stdout.""" for s, p in zip(self.specs[:-1], self.procs[:-1]): self._safe_close(s.stdin) self._safe_close(s.stdout) self._safe_close(s.stderr) if p is None: continue self._safe_close(p.stdin) self._safe_close(p.stdout) self._safe_close(p.stderr) def _close_proc(self): """Closes last proc's stdout.""" s = self.spec p = self.proc self._safe_close(s.stdin) self._safe_close(s.stdout) self._safe_close(s.stderr) self._safe_close(s.captured_stdout) self._safe_close(s.captured_stderr) if p is None: return self._safe_close(p.stdin) self._safe_close(p.stdout) self._safe_close(p.stderr) def _set_input(self): """Sets the input variable.""" if self.proc is None: return stdin = self.proc.stdin if ( stdin is None or isinstance(stdin, int) or stdin.closed or not stdin.seekable() or not safe_readable(stdin) ): input = b"" else: stdin.seek(0) input = stdin.read() self.input = self._decode_uninew(input) def _check_signal(self): """Checks if a signal was received and issues a message.""" proc_signal = getattr(self.proc, "signal", None) if proc_signal is None: return sig, core = proc_signal sig_str = SIGNAL_MESSAGES.get(sig) if sig_str: if core: sig_str += " (core dumped)" print(sig_str, file=sys.stderr) if self.errors is not None: self.errors += sig_str + "\n" def _apply_to_history(self): """Applies the results to the current history object.""" hist = XSH.history if hist is not None: hist.last_cmd_rtn = 1 if self.proc is None else self.proc.returncode def _raise_subproc_error(self): """Raises a subprocess error, if we are supposed to.""" spec = self.spec rtn = self.returncode if rtn is not None and rtn != 0 and XSH.env.get("RAISE_SUBPROC_ERROR"): try: raise subprocess.CalledProcessError(rtn, spec.args, output=self.output) finally: # this is need to get a working terminal in interactive mode self._return_terminal() # # Properties # @property def stdin(self): """Process stdin.""" return self.proc.stdin @property def stdout(self): """Process stdout.""" return self.proc.stdout @property def stderr(self): """Process stderr.""" return self.proc.stderr @property def inp(self): """Creates normalized input string from args.""" return " ".join(self.args) @property def output(self): """Non-blocking, lazy access to output""" if self.ended: if self._output is None: self._output = "".join(self.lines) return self._output else: return "".join(self.lines) @property def out(self): """Output value as a str.""" self.end() return self.output @property def err(self): """Error messages as a string.""" self.end() return self.errors @property def raw_out(self): """Output as raw bytes.""" self.end() return self._raw_output @property def raw_err(self): """Errors as raw bytes.""" self.end() return self._raw_error @property def pid(self): """Process identifier.""" return self.proc.pid @property def returncode(self): """Process return code, waits until command is completed.""" self.end() if self.proc is None: return 1 return self.proc.returncode @property def args(self): """Arguments to the process.""" return self.spec.args @property def rtn(self): """Alias to return code.""" return self.returncode @property def alias(self): """Alias the process used.""" return self.spec.alias @property def stdin_redirect(self): """Redirection used for stdin.""" stdin = self.spec.stdin name = getattr(stdin, "name", "<stdin>") mode = getattr(stdin, "mode", "r") return [name, mode] @property def stdout_redirect(self): """Redirection used for stdout.""" stdout = self.spec.stdout name = getattr(stdout, "name", "<stdout>") mode = getattr(stdout, "mode", "a") return [name, mode] @property def stderr_redirect(self): """Redirection used for stderr.""" stderr = self.spec.stderr name = getattr(stderr, "name", "<stderr>") mode = getattr(stderr, "mode", "r") return [name, mode] @property def timestamps(self): """The start and end time stamps.""" return [self.starttime, self.endtime] @property def executed_cmd(self): """The resolve and executed command.""" return self.spec.cmd @property def stderr_prefix(self): """Prefix to print in front of stderr, as bytes.""" p = self._stderr_prefix if p is None: env = XSH.env t = env.get("XONSH_STDERR_PREFIX") s = xt.format_std_prepost(t, env=env) p = s.encode( encoding=env.get("XONSH_ENCODING"), errors=env.get("XONSH_ENCODING_ERRORS"), ) self._stderr_prefix = p return p @property def stderr_postfix(self): """Postfix to print after stderr, as bytes.""" p = self._stderr_postfix if p is None: env = XSH.env t = env.get("XONSH_STDERR_POSTFIX") s = xt.format_std_prepost(t, env=env) p = s.encode( encoding=env.get("XONSH_ENCODING"), errors=env.get("XONSH_ENCODING_ERRORS"), ) self._stderr_postfix = p return p
[docs]class HiddenCommandPipeline(CommandPipeline): def __repr__(self): return ""
[docs]def resume_process(p): """Sends SIGCONT to a process if possible.""" can_send_signal = ( hasattr(p, "send_signal") and xp.ON_POSIX and not xp.ON_MSYS and not xp.ON_CYGWIN ) if can_send_signal: try: p.send_signal(signal.SIGCONT) except PermissionError: pass
[docs]class PrevProcCloser(threading.Thread): """Previous process closer thread for pipelines whose last command is itself unthreadable. This makes sure that the pipeline is driven forward and does not deadlock. """ def __init__(self, pipeline): """ Parameters ---------- pipeline : CommandPipeline The pipeline whose prev procs we should close. """ self.pipeline = pipeline super().__init__() self.daemon = True self.start()
[docs] def run(self): """Runs the closing algorithm.""" pipeline = self.pipeline check_prev_done = len(pipeline.procs) == 1 if check_prev_done: return proc = pipeline.proc prev_end_time = None timeout = XSH.env.get("XONSH_PROC_FREQUENCY") sleeptime = min(timeout * 1000, 0.1) while proc.poll() is None: if not check_prev_done: # In the case of pipelines with more than one command # we should give the commands a little time # to start up fully. This is particularly true for # GNU Parallel, which has a long startup time. pass elif pipeline._prev_procs_done(): pipeline._close_prev_procs() proc.prevs_are_closed = True break if not check_prev_done: # if we are piping... if prev_end_time is None: # or see if we already know that the next-to-last # proc in the pipeline has ended. if pipeline._prev_procs_done(): # if it has, record the time prev_end_time = time.time() elif time.time() - prev_end_time >= 0.1: # if we still don't have any output, even though the # next-to-last proc has finished, wait a bit to make # sure we have fully started up, etc. check_prev_done = True # this is for CPU usage time.sleep(sleeptime)