Console Parallelization Laravel Package

Getting Started

Minimal Setup

1. Installation: Add the package via Composer:

   composer require webmozarts/console-parallelization
   

2. Extend ParallelCommand: Create a command extending ParallelCommand or using the Parallelization trait:

   use Webmozarts\Console\Parallelization\ParallelCommand;
   
   class ImportUsersCommand extends ParallelCommand
   {
       protected static $defaultName = 'import:users';
   
       protected function fetchItems(): iterable
       {
           // Return string items (e.g., JSON-encoded data)
           yield '{"id": 1, "name": "John"}';
           yield '{"id": 2, "name": "Jane"}';
       }
   
       protected function runSingleCommand(string $item): void
       {
           $data = json_decode($item);
           // Process $data (e.g., save to DB)
       }
   
       protected function getItemName(?int $count): string
       {
           return $count ? 'users' : 'user';
       }
   }
   

3. Run the Command:

   • Single Process: php artisan import:users --main-process
   • Parallel: php artisan import:users --processes=4 (default: auto-detects CPU cores).

Implementation Patterns

Core Workflow

1. Fetch Items: Implement fetchItems() to return an iterable of string items (e.g., JSON, CSV lines).

   • Tip: Use generators for large datasets to avoid memory issues.

   protected function fetchItems(): iterable
   {
       $file = fopen('users.csv', 'r');
       while (($line = fgets($file)) !== false) {
           yield trim($line);
       }
       fclose($file);
   }
   

2. Process Items: Override runSingleCommand() to handle each item.

   protected function runSingleCommand(string $item): void
   {
       $user = json_decode($item);
       User::create($user->name, $user->email);
   }
   

3. Batch Processing: Use hooks for batch-level operations (e.g., DB flushes).

   protected function runAfterBatch(array $items): void
   {
       DB::commit(); // Flush transactions
   }
   

4. Configuration: Customize segment/batch sizes via configureParallelExecutableFactory().

   protected function configureParallelExecutableFactory(
       ParallelExecutorFactory $factory,
       InputInterface $input,
       OutputInterface $output
   ): ParallelExecutorFactory {
       return $factory
           ->withSegmentSize(1000) // Process 1000 items per child process
           ->withBatchSize(100);   // Group 100 items per batch
   }
   

Integration Tips

• Symfony Services: Use subscribed services or fetch services directly from the container to avoid stale instances after errors.

  protected function runSingleCommand(string $item): void
  {
      $userRepository = $this->getContainer()->get(UserRepository::class);
      // ...
  }
  

• Error Handling: Leverage the default ResetServiceErrorHandler to reset the container on failures (avoids Doctrine EM issues).
• Logging: Customize logging via createLogger() for distributed process logs.

  protected function createLogger(): LoggerInterface
  {
      return new Logger('parallel_import');
  }
  

Gotchas and Tips

Pitfalls

1. Item Format:

   • Items must be strings and cannot contain newlines (STDIN delimiter).
   • Fix: Use json_encode() or base64_encode() for complex data.

   yield base64_encode(json_encode($item));
   

2. Non-Rewindable Generators:

   • fetchItems() must return a rewindable iterable (e.g., array or Iterator).
   • Fix: Convert generators to arrays or use Iterator:

   protected function fetchItems(): array
   {
       return iterator_to_array($this->generateItems());
   }
   

3. Service Staleness:

   • Avoid injecting services via constructor if using parallelization (container resets on errors).
   • Fix: Fetch services from the container in methods:

   $this->getContainer()->get(MyService::class);
   

4. Working Directory:

   • Child processes inherit the main process’s working directory. Use absolute paths or chdir() in hooks.

5. Memory Leaks:

   • Large segments/batches may cause memory issues. Monitor with --processes=1 first.

Debugging

• Child Process Isolation: Use --child to test child process logic:

  php artisan import:users --child < item.txt
  

• Logging: Add debug logs in hooks to trace execution:

  protected function runBeforeBatch(): void
  {
      $this->getOutput()->writeln('Starting batch...');
  }
  

• Process Limits: On Unix, check ulimit -u for max processes. Adjust --processes accordingly.

Extension Points

1. Custom PHP Executable:

   ->withPhpExecutable(['php82', '-n']) // Use PHP 8.2 with no config
   

2. Hooks for Pre/Post Work:

   protected function runBeforeFirstCommand(): void
   {
       $this->getOutput()->writeln('Preparing resources...');
   }
   

3. Dynamic Segment Sizes:

   protected function getSegmentSize(): int
   {
       return $this->getItemCount() / $this->getProcessCount();
   }
   

Performance Tips

• Batch Size: Start with 50 (default) and adjust based on memory usage.
• Segment Size: Larger segments reduce process overhead but increase memory usage.
• Process Count: Use --processes=N where N is ≤ CPU cores (e.g., N=4 for 8-core machines).

```markdown
### Laravel-Specific Notes
1. **Artisan Integration**:
   - Register commands in `app/Console/Kernel.php`:
   ```php
   protected $commands = [
       Commands\ImportUsersCommand::class,
   ];

2. Service Container:
   • Use app() instead of $this->getContainer():

   $userRepository = app(UserRepository::class);
   

3. Event Dispatching:
   • Avoid dispatching events in parallel commands (race conditions). Use queues instead.
4. Testing:
   • Mock ParallelCommand by overriding fetchItems() with a small dataset:

   $command = new ImportUsersCommand();
   $command->fetchItems = fn() => ['item1', 'item2'];