Mpall May 2026

@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()

import argparse import logging import sys import time import subprocess import signal import threading from concurrent.futures import ProcessPoolExecutor, as_completed from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime import json import os Installation chmod +x mpall

app = Mpall(args) sys.exit(app.run()) if == " main ": main() README.md # mpall - Multi-Process All-in-One Launcher Run commands in parallel across multiple processes with logging, retries, timeouts, and aggregated output. Installation chmod +x mpall.py sudo ln -s $(pwd)/mpall.py /usr/local/bin/mpall </code></pre> <h2>Usage</h2> <pre><code class="language-bash">mpall -c "command placeholder" -r key=value -w 4 </code></pre> <h2>Features</h2> <ul> <li>✅ <strong>Parallel execution</strong> (configurable worker count)</li> <li>✅ <strong>Placeholder replacement</strong> (<code>var</code> in commands)</li> <li>✅ <strong>Retry on failure</strong> (per task)</li> <li>✅ <strong>Timeout protection</strong> (per command)</li> <li>✅ <strong>JSON output</strong> for programmatic analysis</li> <li>✅ <strong>Signal handling</strong> (graceful Ctrl+C)</li> <li>✅ <strong>Logging</strong> (file + console)</li> <li>✅ <strong>Environment variables</strong> injection</li> </ul> <h2>Examples</h2> <h3>Basic parallel runs</h3> <pre><code class="language-bash">mpall -c "echo Hello" -r {} -w 10 </code></pre> <h3>Variable replacement</h3> <pre><code class="language-bash">mpall -c "curl url" -r url=https://api.example.com,method=GET </code></pre> <h3>Multiple replacements</h3> <pre><code class="language-bash">mpall -c "process.py input output" \ -r input=file1.txt,output=out1.txt \ -r input=file2.txt,output=out2.txt </code></pre> <h3>From file</h3> <pre><code class="language-bash">cat > tasks.txt <<EOF input=in1.txt,output=out1.txt input=in2.txt,output=out2.txt EOF (configurable worker count)&lt

@dataclass class TaskResult: """Result of a single task execution.""" task_id: int args: Tuple success: bool stdout: str stderr: str exit_code: int duration: float retries: int timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) Retry on failure&lt

def run(self) -> int: """Main execution entry point.""" # Parse replacements replacements_list = self.parse_replacements() if not replacements_list: self.logger.error("No replacements provided") return 1 total_tasks = len(replacements_list) self.logger.info(f"Starting total_tasks tasks with self.args.workers workers") # Prepare environment env = os.environ.copy() if self.args.env: for env_var in self.args.env: if '=' in env_var: k, v = env_var.split('=', 1) env[k] = v # Execute tasks in parallel start_time = time.time() with ProcessPoolExecutor(max_workers=self.args.workers) as executor: futures = {} for idx, replacements in enumerate(replacements_list): if self.cancel: break future = executor.submit( worker, idx, self.args.command, replacements, self.args.timeout, self.args.retries, env ) futures[future] = idx # Collect results as they complete for future in as_completed(futures): if self.cancel: break result = future.result() self.results.append(result) self._log_result(result) total_duration = time.time() - start_time # Summary self._print_summary(total_tasks, total_duration) # Save results if requested if self.args.output_json: self._save_results_json() if self.args.output_summary: self._save_summary_text() # Return exit code (0 if all succeeded) return 0 if all(r.success for r in self.results) else 1

# Test 3: Save results ./mpall.py -c "sleep s && echo done" -r s=1 -r s=2 -w 2 -o test.json </code></pre> <p>This is a complete, production-ready feature with error handling, logging, and flexible input methods.</p>

```bash chmod +x mpall.py