@ -1,6 +1,7 @@
""" Miscellaneous utility functions and classes. """
from __future__ import annotations
import abc
import errno
import fcntl
import importlib . util
@ -41,6 +42,7 @@ from .io import (
from . thread import (
mutex ,
WrappedThread ,
)
from . constants import (
@ -254,18 +256,37 @@ def get_available_python_versions(): # type: () -> t.Dict[str, str]
def raw_command (
cmd , # type: t.Iterable[str]
capture = False , # type: bool
capture , # type: bool
env = None , # type: t.Optional[t.Dict[str, str]]
data = None , # type: t.Optional[str]
cwd = None , # type: t.Optional[str]
explain = False , # type: bool
stdin = None , # type: t.Optional[t.Union[t.IO[bytes], int]]
stdout = None , # type: t.Optional[t.Union[t.IO[bytes], int]]
interactive = False , # type: bool
cmd_verbosity = 1 , # type: int
str_errors = ' strict ' , # type: str
error_callback = None , # type: t.Optional[t.Callable[[SubprocessError], None]]
) : # type: (...) -> t.Tuple[t.Optional[str], t.Optional[str]]
""" Run the specified command and return stdout and stderr as a tuple. """
if capture and interactive :
raise InternalError ( ' Cannot combine capture=True with interactive=True. ' )
if data and interactive :
raise InternalError ( ' Cannot combine data with interactive=True. ' )
if stdin and interactive :
raise InternalError ( ' Cannot combine stdin with interactive=True. ' )
if stdout and interactive :
raise InternalError ( ' Cannot combine stdout with interactive=True. ' )
if stdin and data :
raise InternalError ( ' Cannot combine stdin with data. ' )
if stdout and not capture :
raise InternalError ( ' Redirection of stdout requires capture=True to avoid redirection of stderr to stdout. ' )
if not cwd :
cwd = os . getcwd ( )
@ -276,7 +297,30 @@ def raw_command(
escaped_cmd = ' ' . join ( shlex . quote ( c ) for c in cmd )
display . info ( ' Run command: %s ' % escaped_cmd , verbosity = cmd_verbosity , truncate = True )
if capture :
description = ' Run '
elif interactive :
description = ' Interactive '
else :
description = ' Stream '
description + = ' command '
with_types = [ ]
if data :
with_types . append ( ' data ' )
if stdin :
with_types . append ( ' stdin ' )
if stdout :
with_types . append ( ' stdout ' )
if with_types :
description + = f ' with { " / " . join ( with_types ) } '
display . info ( f ' { description } : { escaped_cmd } ' , verbosity = cmd_verbosity , truncate = True )
display . info ( ' Working directory: %s ' % cwd , verbosity = 2 )
program = find_executable ( cmd [ 0 ] , cwd = cwd , path = env [ ' PATH ' ] , required = ' warning ' )
@ -294,17 +338,23 @@ def raw_command(
if stdin is not None :
data = None
communicate = True
elif data is not None :
stdin = subprocess . PIPE
communicate = True
if stdout :
communicate = True
if capture :
elif interactive :
pass # allow the subprocess access to our stdin
else :
stdin = subprocess . DEVNULL
if not interactive :
# When not running interactively, send subprocess stdout/stderr through a pipe.
# This isolates the stdout/stderr of the subprocess from the current process, and also hides the current TTY from it, if any.
# This prevents subprocesses from sharing stdout/stderr with the current process or each other.
# Doing so allows subprocesses to safely make changes to their file handles, such as making them non-blocking (ssh does this).
# This also maintains consistency between local testing and CI systems, which typically do not provide a TTY.
# To maintain output ordering, a single pipe is used for both stdout/stderr when not capturing output.
stdout = stdout or subprocess . PIPE
stderr = subprocess . PIPE
stderr = subprocess . PIPE if capture else subprocess . STDOUT
communicate = True
else :
stderr = None
@ -324,7 +374,7 @@ def raw_command(
if communicate :
data_bytes = to_optional_bytes ( data )
stdout_bytes , stderr_bytes = process. communicate ( data_bytes )
stdout_bytes , stderr_bytes = communicate_with_process( process , data_bytes , stdout == subprocess . PIPE , stderr == subprocess . PIPE , capture = capture )
stdout_text = to_optional_text ( stdout_bytes , str_errors ) or u ' '
stderr_text = to_optional_text ( stderr_bytes , str_errors ) or u ' '
else :
@ -347,6 +397,114 @@ def raw_command(
raise SubprocessError ( cmd , status , stdout_text , stderr_text , runtime , error_callback )
def communicate_with_process ( process : subprocess . Popen , stdin : t . Optional [ bytes ] , stdout : bool , stderr : bool , capture : bool ) - > t . Tuple [ bytes , bytes ] :
""" Communicate with the specified process, handling stdin/stdout/stderr as requested. """
threads : t . List [ WrappedThread ] = [ ]
reader : t . Type [ ReaderThread ]
if capture :
reader = CaptureThread
else :
reader = OutputThread
if stdin is not None :
threads . append ( WriterThread ( process . stdin , stdin ) )
if stdout :
stdout_reader = reader ( process . stdout )
threads . append ( stdout_reader )
else :
stdout_reader = None
if stderr :
stderr_reader = reader ( process . stderr )
threads . append ( stderr_reader )
else :
stderr_reader = None
for thread in threads :
thread . start ( )
for thread in threads :
try :
thread . wait_for_result ( )
except Exception as ex : # pylint: disable=broad-except
display . error ( str ( ex ) )
if isinstance ( stdout_reader , ReaderThread ) :
stdout_bytes = b ' ' . join ( stdout_reader . lines )
else :
stdout_bytes = b ' '
if isinstance ( stderr_reader , ReaderThread ) :
stderr_bytes = b ' ' . join ( stderr_reader . lines )
else :
stderr_bytes = b ' '
process . wait ( )
return stdout_bytes , stderr_bytes
class WriterThread ( WrappedThread ) :
""" Thread to write data to stdin of a subprocess. """
def __init__ ( self , handle : t . IO [ bytes ] , data : bytes ) - > None :
super ( ) . __init__ ( self . _run )
self . handle = handle
self . data = data
def _run ( self ) - > None :
""" Workload to run on a thread. """
try :
self . handle . write ( self . data )
self . handle . flush ( )
finally :
self . handle . close ( )
class ReaderThread ( WrappedThread , metaclass = abc . ABCMeta ) :
""" Thread to read stdout from a subprocess. """
def __init__ ( self , handle : t . IO [ bytes ] ) - > None :
super ( ) . __init__ ( self . _run )
self . handle = handle
self . lines = [ ] # type: t.List[bytes]
@abc.abstractmethod
def _run ( self ) - > None :
""" Workload to run on a thread. """
class CaptureThread ( ReaderThread ) :
""" Thread to capture stdout from a subprocess into a buffer. """
def _run ( self ) - > None :
""" Workload to run on a thread. """
src = self . handle
dst = self . lines
try :
for line in src :
dst . append ( line )
finally :
src . close ( )
class OutputThread ( ReaderThread ) :
""" Thread to pass stdout from a subprocess to stdout. """
def _run ( self ) - > None :
""" Workload to run on a thread. """
src = self . handle
dst = sys . stdout . buffer
try :
for line in src :
dst . write ( line )
dst . flush ( )
finally :
src . close ( )
def common_environment ( ) :
""" Common environment used for executing all programs. """
env = dict (
@ -654,12 +812,15 @@ class MissingEnvironmentVariable(ApplicationError):
self . name = name
def retry ( func , ex_type = SubprocessError , sleep = 10 , attempts = 10 :
def retry ( func , ex_type = SubprocessError , sleep = 10 , attempts = 10 , warn = True ):
""" Retry the specified function on failure. """
for dummy in range ( 1 , attempts ) :
try :
return func ( )
except ex_type :
except ex_type as ex :
if warn :
display . warning ( str ( ex ) )
time . sleep ( sleep )
return func ( )
Matt Clay
2 years ago
committed by