Subprocess Scheduler

SubprocessScheduler.py

import typing
import subprocess
import time
import logging


class ProcessConfig(typing.NamedTuple):
    cmd: typing.Any  # cmd to run
    stdin: typing.IO = None
    stdout: typing.IO = None
    stderr: typing.IO = None
    shell: bool = False
    timeout: int = 120
    callback: typing.Callable[
        [subprocess.Popen, "ProcessConfig"], None] = lambda proc, config: ...


class SubprocessScheduler:
    def __init__(
        self,
        configs: typing.List[ProcessConfig],
        processes: int = 1,
        logger: logging.Logger = logging.getLogger(),
    ) -> None:
        self.configs = configs
        self.processes = processes
        self.process_pool = {}
        self.logger = logger

    def run(self) -> None:
        configs = self.configs.copy()
        counter = 0
        while len(configs) or len(self.process_pool):
            if len(configs) and len(self.process_pool) < self.processes:
                counter += 1
                config = configs.pop(0)
                proc = subprocess.Popen(
                    config.cmd,
                    stdin=config.stdin,
                    stdout=config.stdout,
                    stderr=config.stderr,
                    shell=config.shell)
                self.process_pool[proc] = (config, time.time())
                self.logger.info(
                    f'running {counter} of {len(self.configs)}: {config, proc}')
            else:
                for proc, (config, start_time) in list(self.process_pool.items()):
                    if proc.poll() is not None:
                        del self.process_pool[proc]
                        config.callback(proc, config)
                    elif time.time() - start_time > config.timeout:
                        del self.process_pool[proc]
                        proc.kill()
                        self.logger.warn(f'killed timeout: {config, proc}')


def cmdCallback(proc: subprocess.Popen, config: ProcessConfig):
    outs, errs = proc.communicate()
    print(outs)


cmd = ProcessConfig('sleep 2 && pwd', stdout=subprocess.PIPE,
                    shell=True, callback=cmdCallback)
cmd_list = [cmd] * 9


SubprocessScheduler(cmd_list, 3).run()