p-queue-work allow to invalidate all items instead of limit based ?

This is some batch porting of http://cgit.drupalcode.org/acquia_purge/tree/acquia_purge.drush.inc?h=7.... . But in batch way it looks like lots of code duplicacy as in batch you cant pass objects as param due to serialization problem.

Also as discussed with @nielsvm @adam.weingarten, we should try to spawn new processes (or child processes), So I am trying to use the http://php.net/manual/en/function.proc-open.php but struggling with 2 problems - a) as command argument needs the command name but different systems have different drush path so I think we can't directly use the 'drush p-queue-work'.

I was also checking the pcntl_fork() but not sure if we need to spawn child process or completely a separate process.

This patch contains something with proc_open().
2 Things I am unable to get is -
.

• How to close process if some exception occurs as proc_terminate() requires process resource.
• Get the path of drush script as each system might have different

This is with pcntl_fork() sample. Not in loop.

function purge_drush_p_queue_work($format, QueueServiceInterface $purgeQueue, ProcessorsServiceInterface $purgeProcessors, PurgersServiceInterface $purgePurgers) {
  // Retrieve our queuer object and fail when it is not returned.
  if (!($processor = $purgeProcessors->get('drush_purge_queue_work'))) {
    return drush_set_error('NOT_AUTHORIZED', dt('Not authorized, processor missing!'));
  }

  // Check if claims item from queue, as then no need to spawn process.
  $claims = $purgeQueue->claim();
  //print_r(count($purgeQueue->numberOfItems()));
  //print_r(count($claims));
  //die;
  if (!$claims) {
    return dt('There were no items to be claimed from the queue.');
  }

  $child_processes = [];

  
    // Fork the process.
    $pid = pcntl_fork();

    if ($pid) {
      // Parent process.

      if ($pid < 0) {
        // Unable to fork process, handle error.
        //continue;
      }
      else {
        // Add child PID to tracker childrens.
        $child_processes[$pid] = $pid;
      }
    }
    else {
      // Child process.

      // Acquire lock to prevent race condition process.
      $lock = \Drupal::lock();

      if ($lock->acquire('mymodule_long_operation')) {
        // Claim items from the queue.
        if (!($claims = $purgeQueue->claim())) {
          exit(-1);
          return dt('There were no items to be claimed from the queue.');
        }
        // Attempt the cache invalidation and deal with errors.
        try {
          $purgePurgers->invalidate($processor, $claims);
        }
        catch (DiagnosticsException $e) {
          exit(-1);
          return drush_set_error($e->getMessage());
        }
        catch (CapacityException $e) {
          exit(-1);
          return drush_set_error($e->getMessage());
        }
        catch (LockException $e) {
          exit(-1);
          return drush_set_error($e->getMessage());
        }
        finally {
          $purgeQueue->handleResults($claims);
        }

        // Release the lock.
        $lock->release('mymodule_long_operation');
        // Stop further processing of child process.
        exit(0);
      }
    }

    // Wait for the child processes to exit.
    if (!empty($child_processes)) {
      foreach ($child_processes as $pid) {
        $status = pcntl_waitpid($pid, $status);
      }
    }
   

  //return dt('Processed @number objects...', ['@number' => count($claims)]);
}