D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
support
/
venv
/
lib
/
python3.13
/
site-packages
/
cproc
/
Filename :
process.py
back
Copy
"""cproc module process execution objects""" import signal import logging from multiprocessing import cpu_count from collections.abc import Callable import subprocess as s import rads.vz from .monitor import ProcMonitor class ProcLimit: """Contains load limits for cproc.Proc if used as the lim= arg Note: The grace argument could be useful for subprocesses which require a TCP session, such as rsync, to avoid a timeout caused by the process sleeping for too long. ex: ProcLimit(grace=(5, 1)) Args: value (int | float | None): maximum sever load. If set to None, it'll default to either the number of cores determined from multiprocessing.cpu_count(), or 1.0 if running on a VZ VPS grace (tuple[int, int] | None): If set, it should be a 2-item tuple of ints > 0. The first item is the number of seconds to sleep if the server is overloaded, and the second is how long to allow execution anyways once the sleep expires max_mem (int | None): If set, the process will be killed if it uses more than this amount of memory, in MiB mem_signal (Signal | int): kill signal to send the process if it goes over max_mem. Defaults to ``signal.SIGTERM`` mem_log_func (Callable | None): log function to call if a process is killed for being over memory. Defaults to ``logging.error`` """ __module__ = 'cproc' def __init__( self, value: int | float | None = None, grace: tuple[int, int] | None = None, max_mem: int | None = None, mem_signal: int | signal.Signals = signal.SIGTERM, mem_log_func: Callable | None = logging.error, ) -> None: # memory limits MB -> B self.max_mem = max_mem * 2**20 if max_mem else None self.mem_signal = mem_signal self.mem_log_func = mem_log_func if value is None: value = 1.0 if rads.vz.is_vps() else cpu_count() self.value = value self.grace = grace class Proc(s.Popen): """Custom subprocess.Popen subclass which allows automatic pausing and resuming based on server load and has no defaults for encoding, stdout, and stderr. Warning: Using None for stdout or stderr can be dangerous if your program runs as a daemon under systemd. Systemd can kill the subprocess with SIGPIPE. It's better to set the outputs to Proc.DEVNULL if you don't need them. Args: args: A string, or a sequence of program arguments lim (int, float, ProcLimit | None): max load limit. This can be a static limit if you set it to a float or int, or for more flexibility, use a ``ProcLimit`` instance shell (bool): If true, the command will be executed through the shell (unsafe) encoding (str | None): text encoding for subprocess output and input, such as "UTF-8". Set this to None to use no encoding (as bytes) stdout (int | None): specifies the executed program's standard output. Set to Proc.DEVNULL to not capture output. or Proc.PIPE to capture. Use None to print to your shell, but see Warning above stderr (int | None): specifies the executed program's error output. Set to Proc.DEVNULL to not capture output. or Proc.PIPE to capture. Use None to print to your shell, but see Warning above """ __module__ = 'cproc' # static variables for convenience; access them from Proc instead of having # to import each from subprocess too PIPE = s.PIPE DEVNULL = s.DEVNULL STDOUT = s.STDOUT TimeoutExpired = s.TimeoutExpired CalledProcessError = s.CalledProcessError CompletedProcess = s.CompletedProcess def __init__( self, args, *, lim: int | float | ProcLimit | None, shell: bool = False, encoding: str | None, stdout: int | None, stderr: int | None, **kwargs, ): if lim is None: self.lim = None elif isinstance(lim, ProcLimit): self.lim = lim else: self.lim = ProcLimit(lim) super().__init__( args, shell=shell, encoding=encoding, stdout=stdout, stderr=stderr, **kwargs, ) if self.lim is not None: self.mon = ProcMonitor(self) # pylint: disable=redefined-builtin # input is a builtin, but also what vanilla subprocess calls it def complete( self, input: bytes | str | None = None, timeout: int | float | None = None, check: bool = False, ): """Wait for a process to complete and return a CompletedProcess instance Args: input (bytes | str | None): text to send to stdin. stdin should have already been set to PIPE before sending anything to it timeout (int | float | None): raise Proc.TimeoutExpired if the process takes longer than this number of seconds to execute. Defaults to None check (bool): if True, raise Proc.CalledProcessError if the process exits with a non-zero exit code. The raised object will contain the .returncode attribute. Defaults to False Raises: Proc.CalledProcessError: program exited with a non-zero exit code and check=True was specified Proc.TimeoutExpired: program took too long to execute and timeout= was specified Returns: Proc.CompletedProcess: contains .args, .returncode, .stdout, and .stdin """ try: stdout, stderr = self.communicate(input=input, timeout=timeout) except: # Includes KeyboardInterrupt; communicate handled that self.kill() self.wait() raise # communicate() already called self.wait() if check and self.returncode: raise Proc.CalledProcessError( self.returncode, self.args, output=stdout, stderr=stderr ) return Proc.CompletedProcess(self.args, self.returncode, stdout, stderr) @staticmethod def run( args, *, lim, shell: bool = False, encoding: str | None, input: bytes | str | None = None, capture_output: bool = False, timeout: int | float | None = None, check: bool = False, **kwargs, ) -> s.CompletedProcess: """Run command with arguments and return a CompletedProcess instance Args: args: A string, or a sequence of program arguments lim: max load limit. This can be a float, an int, an object with a .value property (such as a multiprocessing.Value or cproc.ProcLimit object), or an object with .__int__() defined to allow casting to an int shell (bool): If true, the command will be executed through the shell (unsafe) encoding (str | None): text encoding for subprocess output and input, such as "UTF-8". Set this to None to use no encoding (as bytes) input (bytes | str | None): text to send to stdin. Do not use the stdin= kwarg if you use input= capture_output (bool, optional): Sets stdout and stderr to Proc.PIPE if True. Do not also manually set stdout and stderr if this is set to True timeout (int | float | None): raise Proc.TimeoutExpired if the process takes longer than this number of seconds to execute. Defaults to None check (bool): if True, raise Proc.CalledProcessError if the process exits with a non-zero exit code. The raised object will contain the .returncode attribute. Defaults to False Raises: FileNotFoundError: requested program to execute does not exist Proc.CalledProcessError: program exited with a non-zero exit code and check=True was specified Proc.TimeoutExpired: program took too long to execute and timeout= was specified Returns: Proc.CompletedProcess: contains .args, .returncode, .stdout, and .stdin """ if input is not None: if 'stdin' in kwargs: raise TypeError('stdin and input are mutually exclusive') kwargs['stdin'] = Proc.PIPE if capture_output: if 'stdout' in kwargs or 'stderr' in kwargs: raise TypeError( 'stdout and stderr cannot be manually ' 'specified if capture_output=True' ) stdout_num = Proc.PIPE stderr_num = Proc.PIPE else: try: stdout_num = kwargs.pop('stdout') stderr_num = kwargs.pop('stderr') except KeyError as key_exc: raise TypeError( 'stdout and stderr kwargs are required ' 'if capture_output=False' ) from key_exc return Proc( args, lim=lim, shell=shell, stdout=stdout_num, stderr=stderr_num, encoding=encoding, **kwargs, ).complete(input=input, timeout=timeout, check=check)