formatter = logging.Formatter( '%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' )
## Quick Test
parser.add_argument( "-o", "--output-json", help="Save detailed results to JSON file" ) formatter = logging
self.logger.info("=" * 50) self.logger.info("SUMMARY") self.logger.info(f"Total tasks: total_tasks") self.logger.info(f"Succeeded: succeeded") self.logger.info(f"Failed: failed") self.logger.info(f"Total duration: total_duration:.2fs")
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 as_completed from typing import List
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
@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()) Tuple from dataclasses import dataclass
if failed > 0: self.logger.info("\nFailed tasks:") for r in self.results: if not r.success: self.logger.info(f" Task r.task_id: r.stderr[:200]")
# 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 )
mpall [options] command file1 file2 file3 ...