Mpall May 2026

parser.add_argument( "-t", "--timeout", type=int, default=60, help="Timeout per command in seconds (default: 60)" )

parser.add_argument( "--log-file", help="Save log to file" ) parser

# All retries failed return TaskResult( task_id=task_id, args=tuple(replacements.items()), success=False, stdout="", stderr=stderr.strip() if 'stderr' in locals() else "Unknown error", exit_code=exit_code if 'exit_code' in locals() else -1, duration=time.time() - start_time, retries=retries ) class Mpall: """Main application class.""" duration=time.time() - start_time

@staticmethod def run(cmd: List[str], timeout: int, env: Optional[Dict] = None) -> Tuple[int, str, str]: """Run command, return (exit_code, stdout, stderr).""" try: result = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, env=env if env else os.environ.copy(), text=True ) return result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired: return -1, "", f"Command timed out after timeout seconds" except Exception as e: return -2, "", str(e) def worker(task_id: int, args_template: List[str], replacements: Dict[str, str], timeout: int, retries: int, env: Dict) -> TaskResult: """ Worker function executed in a separate process. Builds command by replacing placeholders and runs it. """ start_time = time.time() env: Optional[Dict] = None) -&gt

# Test 2: With retries (simulate failure) ./mpall.py -c "sh -c 'exit code'" -r code=0 -r code=1 --retries 2