<commit_before><?php declare(strict_types=1);

namespace Novuso\Common\Adapter\Process;

use Novuso\Common\Application\Process\Exception\ProcessException;
use Novuso\Common\Application\Process\Process;
use Novuso\Common\Application\Process\ProcessErrorBehavior;
use Novuso\Common\Application\Process\ProcessRunnerInterface;
use Novuso\System\Collection\Api\QueueInterface;
use Novuso\System\Collection\LinkedQueue;
use Novuso\System\Exception\DomainException;
use Psr\Log\LoggerInterface;
use Symfony\Component\Process\Exception\ProcessFailedException;
use Symfony\Component\Process\Process as SymfonyProcess;
use Throwable;

/**
 * SymfonyProcessRunner is a Symfony process runner adapter
 *
 * @copyright Copyright (c) 2017, Novuso. <http://novuso.com>
 * @license   http://opensource.org/licenses/MIT The MIT License
 * @author    <NAME> <<EMAIL>>
 */
class SymfonyProcessRunner implements ProcessRunnerInterface
{
    /**
     * Logger
     *
     * @var LoggerInterface
     */
    private $logger;

    /**
     * Max concurrent processes
     *
     * @var int
     */
    private $maxConcurrent;

    /**
     * Delay used for usleep
     *
     * @var int
     */
    private $delay;

    /**
     * Process queue
     *
     * @var QueueInterface
     */
    private $queue;

    /**
     * Currently running processes
     *
     * @var array
     */
    private $processes;

    /**
     * Constructs SymfonyProcessRunner
     *
     * @param LoggerInterface $logger        The logger service
     * @param int             $maxConcurrent The max concurrent processes or 0
     *                                       for no limit
     * @param int             $delay         The number of microseconds to
     *                                       delay between process checks
     *
     * @throws DomainException When delay is not a natural number
     */
    public function __construct(LoggerInterface $logger, int $maxConcurrent = 1, int $delay = 1000)
    {
        if ($delay < 1) {
            $message = sprintf('%s expects delay to be a natural number', __METHOD__);
            throw new DomainException($message);
        }

        $this->logger = $logger;
        $this->maxConcurrent = $maxConcurrent;
        $this->delay = $delay;
        $this->queue = LinkedQueue::of(Process::class);
        $this->processes = [];
    }

    /**
     * Destructs SymfonyProcessRunner
     */
    public function __destruct()
    {
        $this->stop();
    }

    /**
     * {@inheritdoc}
     */
    public function attach(Process $process): void
    {
        $this->queue->enqueue($process);
    }

    /**
     * {@inheritdoc}
     */
    public function clear(): void
    {
        $this->queue = LinkedQueue::of(Process::class);
        $this->processes = [];
    }

    /**
     * {@inheritdoc}
     */
    public function run(?ProcessErrorBehavior $errorBehavior = null): void
    {
        if ($errorBehavior === null) {
            $errorBehavior = ProcessErrorBehavior::EXCEPTION();
        }

        while (!$this->queue->isEmpty()) {
            $this->init($errorBehavior);
            $this->tick($errorBehavior);
        }

        while (count($this->processes)) {
            $this->tick($errorBehavior);
        }

        $this->clear();
    }

    /**
     * Starts a process if possible
     *
     * @param ProcessErrorBehavior $errorBehavior The process error behavior
     *
     * @return void
     *
     * @throws ProcessException When an error occurs, depending on behavior
     */
    private function init(ProcessErrorBehavior $errorBehavior): void
    {
        if ($this->maxConcurrent !== 0 && count($this->processes) >= $this->maxConcurrent) {
            return;
        }

        try {
            /** @var Process $process */
            $process = $this->queue->dequeue();
            $symfonyProcess = $this->exchangeProcess($process);

            $this->startProcess($symfonyProcess, $process->stdout(), $process->stderr());

            $pid = $symfonyProcess->getPid();
            $this->processes[$pid] = [
                'iteration' => 1,
                'original'  => $process,
                'process'   => $symfonyProcess
            ];

            $this->logProcessStarted($symfonyProcess);
        } catch (Throwable $e) {
            if ($errorBehavior->value() === ProcessErrorBehavior::EXCEPTION) {
                throw new ProcessException($e->getMessage(), $e->getCode(), $e);
            }
        }
    }

    /**
     * Performs running checks on processes
     *
     * @param ProcessErrorBehavior $errorBehavior The process error behavior
     *
     * @return void
     *
     * @throws ProcessException When an error occurs, depending on behavior
     */
    private function tick(ProcessErrorBehavior $errorBehavior): void
    {
        usleep($this->delay);

        try {
            foreach ($this->processes as $pid => $processData) {
                /** @var SymfonyProcess $process */
                $process = $processData['process'];
                $process->checkTimeout();

                if ($process->isRunning()) {
                    continue;
                }

                // remove from current process before
                // checking for success
                unset($this->processes[$pid]);<commit_msg>comment<commit_after><?php declare(strict_types=1);

namespace Novuso\Common\Adapter\Process;

use Novuso\Common\Application\Process\Exception\ProcessException;
use Novuso\Common\Application\Process\Process;
use Novuso\Common\Application\Process\ProcessErrorBehavior;
use Novuso\Common\Application\Process\ProcessRunnerInterface;
use Novuso\System\Collection\Api\QueueInterface;
use Novuso\System\Collection\LinkedQueue;
use Novuso\System\Exception\DomainException;
use Psr\Log\LoggerInterface;
use Symfony\Component\Process\Exception\ProcessFailedException;
use Symfony\Component\Process\Process as SymfonyProcess;
use Throwable;

/**
 * SymfonyProcessRunner is a Symfony process runner adapter
 *
 * @copyright Copyright (c) 2017, Novuso. <http://novuso.com>
 * @license   http://opensource.org/licenses/MIT The MIT License
 * @author    <NAME> <<EMAIL>>
 */
class SymfonyProcessRunner implements ProcessRunnerInterface
{
    /**
     * Logger
     *
     * @var LoggerInterface
     */
    private $logger;

    /**
     * Max concurrent processes
     *
     * @var int
     */
    private $maxConcurrent;

    /**
     * Delay used for usleep
     *
     * @var int
     */
    private $delay;

    /**
     * Process queue
     *
     * @var QueueInterface
     */
    private $queue;

    /**
     * Currently running processes
     *
     * @var array
     */
    private $processes;

    /**
     * Constructs SymfonyProcessRunner
     *
     * @param LoggerInterface $logger        The logger service
     * @param int             $maxConcurrent The max concurrent processes or 0
     *                                       for no limit
     * @param int             $delay         The number of microseconds to
     *                                       delay between process checks
     *
     * @throws DomainException When delay is not a natural number
     */
    public function __construct(LoggerInterface $logger, int $maxConcurrent = 1, int $delay = 1000)
    {
        if ($delay < 1) {
            $message = sprintf('%s expects delay to be a natural number', __METHOD__);
            throw new DomainException($message);
        }

        $this->logger = $logger;
        $this->maxConcurrent = $maxConcurrent;
        $this->delay = $delay;
        $this->queue = LinkedQueue::of(Process::class);
        $this->processes = [];
    }

    /**
     * Destructs SymfonyProcessRunner
     */
    public function __destruct()
    {
        $this->stop();
    }

    /**
     * {@inheritdoc}
     */
    public function attach(Process $process): void
    {
        $this->queue->enqueue($process);
    }

    /**
     * {@inheritdoc}
     */
    public function clear(): void
    {
        $this->queue = LinkedQueue::of(Process::class);
        $this->processes = [];
    }

    /**
     * {@inheritdoc}
     */
    public function run(?ProcessErrorBehavior $errorBehavior = null): void
    {
        if ($errorBehavior === null) {
            $errorBehavior = ProcessErrorBehavior::EXCEPTION();
        }

        while (!$this->queue->isEmpty()) {
            $this->init($errorBehavior);
            $this->tick($errorBehavior);
        }

        while (count($this->processes)) {
            $this->tick($errorBehavior);
        }

        $this->clear();
    }

    /**
     * Starts a process if possible
     *
     * @param ProcessErrorBehavior $errorBehavior The process error behavior
     *
     * @return void
     *
     * @throws ProcessException When an error occurs, depending on behavior
     */
    private function init(ProcessErrorBehavior $errorBehavior): void
    {
        if ($this->maxConcurrent !== 0 && count($this->processes) >= $this->maxConcurrent) {
            return;
        }

        try {
            /** @var Process $process */
            $process = $this->queue->dequeue();
            $symfonyProcess = $this->exchangeProcess($process);

            $this->startProcess($symfonyProcess, $process->stdout(), $process->stderr());

            $pid = $symfonyProcess->getPid();
            $this->processes[$pid] = [
                'iteration' => 1,
                'original'  => $process,
                'process'   => $symfonyProcess
            ];

            $this->logProcessStarted($symfonyProcess);
        } catch (Throwable $e) {
            if ($errorBehavior->value() === ProcessErrorBehavior::EXCEPTION) {
                throw new ProcessException($e->getMessage(), $e->getCode(), $e);
            }
        }
    }

    /**
     * Performs running checks on processes
     *
     * @param ProcessErrorBehavior $errorBehavior The process error behavior
     *
     * @return void
     *
     * @throws ProcessException When an error occurs, depending on behavior
     */
    private function tick(ProcessErrorBehavior $errorBehavior): void
    {
        usleep($this->delay);

        try {
            foreach ($this->processes as $pid => $processData) {
                /** @var SymfonyProcess $process */
                $process = $processData['process'];
                $process->checkTimeout();

                if ($process->isRunning()) {
                    continue;
                }

                // remove current process before
                // checking for success
                unset($this->processes[$pid]);
{'fraction_non_alphanumeric': 0.12073342736248237, 'fraction_numerical': 0.0026328161730136343, 'mean_word_length': 1.9809417040358743, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 95, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5433371', 'n_tokens_mistral': 2852, 'n_tokens_neox': 2745, 'n_words': 642}