Parallel Laravel Package

Getting Started

Minimal Steps

1. Installation:

   composer require amphp/parallel
   

   Ensure PHP 8.1+ is used. For threads, PHP 8.2+ with ZTS and ext-parallel is required.

2. First Use Case: Offload a blocking task (e.g., file I/O, CPU-heavy operations) to a worker:

   use Amp\Parallel\Worker;
   use Amp\Parallel\Worker\Task;
   
   class MyTask implements Task {
       public function run(\Amp\Sync\Channel $channel, \Amp\Cancellation $cancellation): string {
           return file_get_contents('https://example.com'); // Blocking call
       }
   }
   
   $worker = Worker::createWorker();
   $execution = $worker->submit(new MyTask());
   $result = $execution->await();
   

3. Key Files:

   • Review src/ for core classes (Worker, Task, WorkerPool).
   • Check tests/ for integration patterns (e.g., cancellation, IPC).

Implementation Patterns

Core Workflows

1. Task Submission:

   • Use Worker::submit() or WorkerPool::submit() for parallel execution.
   • Example: Batch-processing user uploads:

     $pool = new WorkerPool(4); // 4 concurrent workers
     foreach ($uploads as $upload) {
         $pool->submit(new ProcessUploadTask($upload));
     }
     

2. Worker Pools:

   • Reuse workers for efficiency:

     $pool = new WorkerPool(8, new ProcessContextFactory());
     $pool->submit(new HeavyTask());
     

   • Access global pool via Worker\workerPool().

3. IPC Patterns:

   • Structured messaging with Channel:

     $context = contextFactory()->start(__DIR__.'/child.php');
     $context->send(['command' => 'fetch', 'url' => 'https://api.example.com']);
     $response = $context->receive();
     

4. Cancellation:

   • Propagate cancellation to workers:

     $cancellation = new Cancellation();
     $execution = $worker->submit(new MyTask(), $cancellation);
     $cancellation->cancel(); // Triggers CancellationException in worker
     

Integration Tips

• Laravel Integration:

  • Use WorkerPool in Laravel’s app() container:

    $this->app->singleton(WorkerPool::class, fn() => new WorkerPool(4));
    

  • Offload queue jobs to workers:

    class ParallelJob implements ShouldQueue {
        public function handle() {
            $worker = app(WorkerPool::class)->getWorker();
            $worker->submit(new ProcessDataTask($this->data));
        }
    }
    

• Error Handling:

  • Wrap submissions in try-catch:

    try {
        $result = $execution->await();
    } catch (WorkerException $e) {
        \Log::error('Worker failed:', ['error' => $e->getMessage()]);
    }
    

• Resource Sharing:

  • Use AtomicCache for thread-safe shared state:

    class SharedTask implements Task {
        private static AtomicCache $cache;
    
        public function run(Channel $channel, Cancellation $cancellation) {
            self::$cache ??= new AtomicCache();
            return self::$cache->getOrSet('key', fn() => computeExpensiveValue());
        }
    }
    

Gotchas and Tips

Pitfalls

1. Serialization:

   • Tasks must be serializable. Avoid closures or non-serializable objects (e.g., resources, DOM nodes).
   • Fix: Use static classes or serialize()-compatible data.

2. Blocking the Event Loop:

   • Workers run blocking code, but the main process must remain non-blocking. Avoid sleep() or synchronous HTTP calls in the parent.

3. Thread Limitations:

   • Threads (via ext-parallel) are faster but have PHP’s GIL limitations. Prefer processes for CPU-bound tasks.

4. Memory Leaks:

   • Unclosed Channel or Worker instances can leak resources. Use finally blocks:

     $worker = Worker::createWorker();
     try {
         $result = $worker->submit(...)->await();
     } finally {
         $worker->close(); // Critical!
     }
     

5. Global State:

   • Shared state (e.g., $_SESSION) is not shared between workers. Use Cache or databases instead.

Debugging

• Worker Logs:

  • Redirect stderr to a file in the worker script:

    // In child.php
    file_put_contents('worker.log', print_r($data, true));
    

  • Use Amp\Log\ConsoleLogger for structured logs.

• Timeouts:

  • Add timeouts to Execution:

    $execution->await(new TimeoutException(5)); // 5-second timeout
    

• Common Exceptions:

  • WorkerException: Worker process failed.
  • CancelledException: Task was cancelled.
  • SerializationException: Invalid task data.

Extension Points

1. Custom Context Factories:

   • Override DefaultContextFactory to customize worker bootstrapping:

     class CustomContextFactory extends ProcessContextFactory {
         public function start(string $script): Context {
             $context = parent::start($script);
             $context->send(['env' => 'custom']); // Inject config
             return $context;
         }
     }
     

2. Task Middleware:

   • Wrap tasks in middleware for logging/metrics:

     class LoggingTask implements Task {
         public function __construct(private Task $task) {}
     
         public function run(Channel $channel, Cancellation $cancellation) {
             \Log::info('Task started', ['task' => get_class($this->task)]);
             return $this->task->run($channel, $cancellation);
         }
     }
     

3. Dynamic Worker Pools:

   • Scale pools based on load:

     $pool = new WorkerPool(2);
     if ($queue->count() > 100) {
         $pool->resize(8); // Dynamically add workers
     }
     

4. Progress Tracking:

   • Use Execution::getProgress() for long-running tasks:

     $execution = $worker->submit(new LongTask());
     while (!$execution->isComplete()) {
         $progress = $execution->getProgress();
         \Log::info("Progress: {$progress}%");
         Amp\delay(1000); // Poll every second
     }