<commit_before><?php
namespace FutureProcess;

use React\Promise\FulfilledPromise;
use React\Promise\PromiseInterface;

/**
 * @author <NAME> <<EMAIL>>
 */
class FutureProcess
{
    const STATUS_QUEUED   = 0;
    const STATUS_RUNNING  = 1;
    const STATUS_EXITED   = 2;
    const STATUS_ABORTED  = 3;
    const STATUS_UNKNOWN  = 4;
    
    private static $defaultOptions;
    
    private $promise;
    private $options;
    private $futureExitCode;
    private $queueSlot;
    private $status;
    private $resource;
    private $startTime;
    private $pid;
    private $pipes;
    private $result;
    
    public function __construct(
        $command,
        array $options,
        FutureValue $futureExitCode,
        FutureValue $queueSlot = null
    ) {
        $options = $this->prepareOptions($options);
        
        $this->options = $options;
        $this->futureExitCode = $futureExitCode;
        $this->pipes = new Pipes($options['io']);
        
        $startFn = $this->getStartFn();
        
        if ($this->queueSlot = $queueSlot) {
            $this->status = self::STATUS_QUEUED;
            $that = $this;
            $this->promise = $queueSlot->then(
                function () use ($startFn, $command, $options, $that) {
                    $startFn($command, $options);
                    return $that;
                },
                function (\Exception $e) use ($that) {
                    $that->abort($e);
                    throw $e;
                }
            );
        } else {
            $startFn($command, $options);
            $this->promise = new FulfilledPromise($this);
        }
    }
    
    /**
     * @return int
     */
    public function getPid()
    {
        $this->wait();
        
        return $this->pid;
    }
    
    /**
     * @param int $descriptor
     * @return null|resource
     */
    public function getPipe($descriptor)
    {
        $this->wait();
            
        return $this->pipes->getResource($descriptor);
    }
    
    /**
     * @param int $descriptor
     * @param string $data
     */
    public function writeToBuffer($descriptor, $data)
    {
        $this->pipes->writeToBuffer($descriptor, $data);
    }
    
    /**
     * @param int $descriptor
     * @return string
     */
    public function readFromBuffer($descriptor)
    {
        $this->wait();

        return $this->pipes->readFromBuffer($descriptor);
    }
    
    /**
     * @param bool $refresh OPTIONAL
     * @return int One of the status constants defined in this class
     */
    public function getStatus($refresh = true)
    {
        if ($refresh && $this->status === self::STATUS_RUNNING) {
            $this->refreshStatus();
        }
        
        return $this->status;
    }
    
    /**
     * @return FutureResult
     */
    public function getResult()
    {
        if (is_null($this->result)) {
            $this->result = new FutureResult($this->pipes, $this->futureExitCode);
        }
        
        return $this->result;
    }
    
    /**
     * @param \Exception|null $error
     * @param int|null $signal If null is passed, no signal will be sent to the process
     */
    public function abort(\Exception $error = null, $signal = 15)
    {
        if ($this->status === self::STATUS_RUNNING) {
            if (null !== $signal) {
                proc_terminate($this->resource, $signal);
            }
        } elseif ($this->status === self::STATUS_QUEUED) {
            if ($error) {
                $this->queueSlot->reject($error);
            } else {
                $this->queueSlot->resolve();
            }
        } else {
            return;
        }
        
        $this->doExit(self::STATUS_ABORTED, $error);
    }
    
    /**
     * Wait for the process to start
     * 
     * @param double $timeout OPTIONAL
     * @return static
     */
    public function wait($timeout = null)
    {
        if ($this->queueSlot) {
            $this->queueSlot->wait($timeout);
        }
        
        return $this;
    }
    
    /**
     * @return PromiseInterface
     */
    public function promise()
    {
        return $this->promise;
    }
    
    /**
     * @param callable $onFulfilled
     * @param callable $onError
     * @param callable $onProgress
     * @return PromiseInterface
     */
    public function then($onFulfilled = null, $onError = null, $onProgress = null)
    {
        return $this->promise->then($onFulfilled, $onError, $onProgress);
    }
    
    private function refreshStatus()
    {
        if (false === $status = proc_get_status($this->resource)) {
            $this->doExit(self::STATUS_UNKNOWN, new \RuntimeException('An unknown error occurred.'));
        } elseif (!$status['running']) {
            $exitCode = (-1 == $status['exitcode'] ? null : $status['exitcode']);
            $this->doExit(self::STATUS_EXITED, $exitCode);
        } elseif (!$this->hasExceededTimeLimit()) {
            $this->pipes->drainBuffers();
        }
        
        return $status;
    }
    
    private function hasExceededTimeLimit()
    {
        if ($this->options['timeout'] && microtime(true) > $this->startTime + $this->options['timeout']) {
            $this->abort(
                $this->options['timeout_error'],
                $this->options['timeout_signal']
            );
            
            return true;
        }
        
        return false;
    }
    
    private static function prepareOptions(array $options)
    {
        return array_merge(
            self::getDefaultOptions(),
            $options<commit_msg>Replace UNKNOWN status with ERROR<commit_after><?php
namespace FutureProcess;

use React\Promise\FulfilledPromise;
use React\Promise\PromiseInterface;

/**
 * @author <NAME> <<EMAIL>>
 */
class FutureProcess
{
    const STATUS_QUEUED   = 0;
    const STATUS_RUNNING  = 1;
    const STATUS_EXITED   = 2;
    const STATUS_ABORTED  = 3;
    const STATUS_ERROR    = 4;
    
    private static $defaultOptions;
    
    private $promise;
    private $options;
    private $futureExitCode;
    private $queueSlot;
    private $status;
    private $resource;
    private $startTime;
    private $pid;
    private $pipes;
    private $result;
    
    public function __construct(
        $command,
        array $options,
        FutureValue $futureExitCode,
        FutureValue $queueSlot = null
    ) {
        $options = $this->prepareOptions($options);
        
        $this->options = $options;
        $this->futureExitCode = $futureExitCode;
        $this->pipes = new Pipes($options['io']);
        
        $startFn = $this->getStartFn();
        
        if ($this->queueSlot = $queueSlot) {
            $this->status = self::STATUS_QUEUED;
            $that = $this;
            $this->promise = $queueSlot->then(
                function () use ($startFn, $command, $options, $that) {
                    $startFn($command, $options);
                    return $that;
                },
                function (\Exception $e) use ($that) {
                    $that->abort($e);
                    throw $e;
                }
            );
        } else {
            $startFn($command, $options);
            $this->promise = new FulfilledPromise($this);
        }
    }
    
    /**
     * @return int
     */
    public function getPid()
    {
        $this->wait();
        
        return $this->pid;
    }
    
    /**
     * @param int $descriptor
     * @return null|resource
     */
    public function getPipe($descriptor)
    {
        $this->wait();
            
        return $this->pipes->getResource($descriptor);
    }
    
    /**
     * @param int $descriptor
     * @param string $data
     */
    public function writeToBuffer($descriptor, $data)
    {
        $this->pipes->writeToBuffer($descriptor, $data);
    }
    
    /**
     * @param int $descriptor
     * @return string
     */
    public function readFromBuffer($descriptor)
    {
        $this->wait();

        return $this->pipes->readFromBuffer($descriptor);
    }
    
    /**
     * @param bool $refresh OPTIONAL
     * @return int One of the status constants defined in this class
     */
    public function getStatus($refresh = true)
    {
        if ($refresh && $this->status === self::STATUS_RUNNING) {
            $this->refreshStatus();
        }
        
        return $this->status;
    }
    
    /**
     * @return FutureResult
     */
    public function getResult()
    {
        if (is_null($this->result)) {
            $this->result = new FutureResult($this->pipes, $this->futureExitCode);
        }
        
        return $this->result;
    }
    
    /**
     * @param \Exception|null $error
     * @param int|null $signal If null is passed, no signal will be sent to the process
     */
    public function abort(\Exception $error = null, $signal = 15)
    {
        if ($this->status === self::STATUS_RUNNING) {
            if (null !== $signal) {
                proc_terminate($this->resource, $signal);
            }
        } elseif ($this->status === self::STATUS_QUEUED) {
            if ($error) {
                $this->queueSlot->reject($error);
            } else {
                $this->queueSlot->resolve();
            }
        } else {
            return;
        }
        
        $this->doExit(self::STATUS_ABORTED, $error);
    }
    
    /**
     * Wait for the process to start
     * 
     * @param double $timeout OPTIONAL
     * @return static
     */
    public function wait($timeout = null)
    {
        if ($this->queueSlot) {
            $this->queueSlot->wait($timeout);
        }
        
        return $this;
    }
    
    /**
     * @return PromiseInterface
     */
    public function promise()
    {
        return $this->promise;
    }
    
    /**
     * @param callable $onFulfilled
     * @param callable $onError
     * @param callable $onProgress
     * @return PromiseInterface
     */
    public function then($onFulfilled = null, $onError = null, $onProgress = null)
    {
        return $this->promise->then($onFulfilled, $onError, $onProgress);
    }
    
    private function refreshStatus()
    {
        if (false === $status = proc_get_status($this->resource)) {
            $this->doExit(self::STATUS_ERROR, new \RuntimeException('An unknown error occurred.'));
        } elseif (!$status['running']) {
            $exitCode = (-1 == $status['exitcode'] ? null : $status['exitcode']);
            $this->doExit(self::STATUS_EXITED, $exitCode);
        } elseif (!$this->hasExceededTimeLimit()) {
            $this->pipes->drainBuffers();
        }
        
        return $status;
    }
    
    private function hasExceededTimeLimit()
    {
        if ($this->options['timeout'] && microtime(true) > $this->startTime + $this->options['timeout']) {
            $this->abort(
                $this->options['timeout_error'],
                $this->options['timeout_signal']
            );
            
            return true;
        }
        
        return false;
    }
    
    private static function prepareOptions(array $options)
    {
        return array_merge(
            self::getDefaultOptions(),
            $options
{'fraction_non_alphanumeric': 0.1323450616181461, 'fraction_numerical': 0.0014288265761743168, 'mean_word_length': 1.70965400435519, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1319167', 'n_tokens_mistral': 3199, 'n_tokens_neox': 3175, 'n_words': 669}