Simple Symfony3 command Thread Pool

ThreadPoolCommand.php

<?php

namespace AppBundle\Command;

use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Process\Process;

/**
 * basic thread pool to execute a fixed amount of worker commands
 */
class ThreadPoolCommand extends Command
{
    const INPUT_THREADS = 'threads';
    const INPUT_COMMAND = 'exec';
    const TICK = 500000;
    const MAX_THREADS = 20;

    /**
     * setup help
     */
    protected function configure()
    {
        $this
            ->setName('thread:pool')
            ->addOption(self::INPUT_THREADS, 't', InputOption::VALUE_OPTIONAL, 'number of threads to work (default is one)', 1)
            ->addArgument(self::INPUT_COMMAND, InputArgument::REQUIRED, 'command to run (to include parameters add quotes)')
        ;
    }

    /**
     * execute pool
     *
     * @param InputInterface $input
     * @param OutputInterface $output
     * @return int
     */
    public function execute(InputInterface $input, OutputInterface $output)
    {
        $threads = $input->getOption(self::INPUT_THREADS);
        $command = $this->buildCommand($input);

        if (empty($threads) || $threads < 1 || $threads > self::MAX_THREADS) {
            $output->writeln('pool[master] <error>invalid thread count</error>');
            return 1;
        }

        /** @var Process[] $pool */
        $pool = array_fill(0, $threads, null);
        $logger = $this->buildLoggers($threads);

        $output->writeln('pool[master] start with <info>'.$threads.'</info> workers');

        while (true) {
            for ($i = 0; $i < $threads; $i ++) {
                if ($pool[$i] instanceof Process && ($pool[$i]->isRunning() || !$pool[$i]->isStarted())) {
                    continue;
                }

                if ($pool[$i] instanceof Process) {
                    $output->writeln(sprintf(
                        'pool[%d] %s failed',
                        $i,
                        $command
                    ));
                }

                $output->writeln(sprintf(
                    'pool[%d] %s started',
                    $i,
                    $command
                ));

                $pool[$i] = new Process($command);
                // stop output caching and stream it to strerr and stdout
                $pool[$i]->disableOutput();
                $pool[$i]->start($logger[$i]);
            }

            usleep(self::TICK);
        }

        $output->writeln('pool[master] <error>error with pool</error>');
        return 0;
    }

    /**
     * build prefixed loggers
     * @param int $threads
     *
     * @return callable[]
     */
    private function buildLoggers($threads)
    {
        $logger = [];

        for ($i = 0; $i < $threads; $i++) {
            $prefix = 'pool['.$i.'] ';
            $method = <<<PHP
if (\Symfony\Component\Process\Process::ERR === \$type) {
    fwrite(STDERR, '$prefix'.str_replace(PHP_EOL, PHP_EOL.'$prefix', trim(\$buffer)).PHP_EOL);
} else {
    fwrite(STDOUT, '$prefix'.str_replace(PHP_EOL, PHP_EOL.'$prefix', trim(\$buffer)).PHP_EOL);
}
PHP;

            $logger[] = create_function('$type, $buffer', $method);
        }

        return $logger;
    }

    /**
     * build execution path, also append custom php ini for heroku
     *
     * @param InputInterface $input
     * @return string
     */
    private function buildCommand(InputInterface $input)
    {
        if ($path = php_ini_loaded_file()) {
            $ini = '--php-ini '.$path;
        } else {
            $ini = '';
        }

        $command = $input->getArgument(self::INPUT_COMMAND);

        return sprintf(
            'php %s bin/console %s',
            $ini,
            $command
        );
    }
}