<commit_before><?php

namespace Amp\MultiProcess;

use Amp\Reactor,
    Amp\ReactorFactory;

class Dispatcher {
    
    const CALL        = 1;
    const CALL_RESULT = 2;
    const CALL_ERROR  = 3;
    const MAX_PROCEDURE_LENGTH = 255;
    const MAX_CALL_ID = 2147483647;
    
    private $reactor;
    private $workerSessionFactory;
    private $autoWriteSubscription;
    private $autoTimeoutSubscription;
    
    private $workerSessions = [];
    private $pendingCallCounts = [];
    private $workerSubscriptions = [];
    private $workerCallMap = [];
    private $callWorkerMap = [];
    private $callFrames = [];
    private $availableWorkers = [];
    private $timeoutSchedule = [];
    private $onResultCallbacks = [];
    private $partialResults = [];
    private $writableWorkers;
    
    private $poolSize;
    private $workerCmd;
    private $workerCwd;
    private $lastCallId = 0;
    private $pendingCalls = 0;
    private $callTimeout = 30;
    private $readTimeout = 60;
    private $autoWriteInterval = 0.025;
    private $timeoutCheckInterval = 1;
    private $writeErrorsTo = STDERR;
    private $isStarted = FALSE;
    private $chrCallCode;
    
    function __construct(Reactor $reactor, $workerCmd, $poolSize = 1, WorkerSessionFactory $wsf = NULL, CallResultFactory $crf = NULL) {
        $this->reactor = $reactor;
        $this->workerCmd = $workerCmd;
        $this->poolSize = (is_int($poolSize) && $poolSize > 0) ? $poolSize : 1;
        $this->workerSessionFactory = $wsf ?: new WorkerSessionFactory;
        $this->callResultFactory = $crf ?: new CallResultFactory;
        
        $this->writableWorkers = new \SplObjectStorage;
        $this->chrCallCode = chr(self::CALL);
    }
    
    function setCallTimeout($seconds) {
        $this->callTimeout = filter_var($seconds, FILTER_VALIDATE_FLOAT, ['options' => [
            'min_range' => 0,
            'default' => 30
        ]]);
    }
    
    function setGranularity($bytes) {
        $this->workerSessionFactory->setGranularity($bytes);
    }
    
    function setWorkerCwd($directory) {
        $this->workerCwd = $directory;
    }
    
    /**
     * Hydrate the worker process pool
     * 
     * Repeated calls to start have no effect after the first invocation.
     * 
     * @param int $poolSize The number of worker processes to maintain in the pool
     * @param string $cmd   The executable worker process command
     * @param string $cwd   An optional current working directory for worker processes. If not
     *                      specified workers will have the same cwd as the main process.
     * 
     * @return void
     */
    function start() {
        if ($this->isStarted) {
            return;
        }
        
        for ($i=0; $i < $this->poolSize; $i++) {
            $this->spawnWorkerSession();
        }
        
        $this->autoWriteSubscription = $this->reactor->repeat(function() {
            foreach ($this->writableWorkers as $workerSession) {
                if ($this->write($workerSession)) {
                    $this->writableWorkers->detach($workerSession);
                }
            }
        }, $this->autoWriteInterval);
        
        if ($this->callTimeout) {
            $this->autoTimeoutSubscription = $this->reactor->repeat(function() {
                $this->autoTimeout();
            }, $this->timeoutCheckInterval);
        }
        
        $this->isStarted = TRUE;
    }
    
    private function spawnWorkerSession() {
        $workerSession = $this->workerSessionFactory->__invoke(
            $this->workerCmd,
            $this->writeErrorsTo,
            $this->workerCwd
        );
        
        $readSubscription = $this->reactor->onReadable(
            $workerSession->getReadPipe(),
            function($pipe, $trigger) use ($workerSession) { $this->read($workerSession, $trigger); },
            $this->readTimeout
        );
        
        $workerId = spl_object_hash($workerSession);
        $this->workerSessions[$workerId] = $workerSession;
        $this->pendingCallCounts[$workerId] = 0;
        $this->workerSubscriptions[$workerId] = $readSubscription;
    }
    
    /**
     * Asynchronously execute a procedure and invoke the $onResult callback when complete
     * 
     * @param callable $onResult The callback to process the async execution's CallResult
     * @param string $procedure  The function to execute asynchronously
     * @param string $workload   The data to pass as an argument to the procedure
     * 
     * @return string Returns the task's call ID
     */
    function call(callable $onResult, $procedure, $workload = NULL) {
        if (!is_string($procedure)) {
            throw new \InvalidArgumentException;
        } elseif (($procLen = strlen($procedure)) > self::MAX_PROCEDURE_LENGTH) {
            throw new \RangeException(
                'Procedure name exceeds maximum allowable length (' .
                self::MAX_PROCEDURE_LENGTH . '): ' . $procedure
            );
        }
        
        if (($callId = ++$this->lastCallId) == self::MAX_CALL_ID) {
            $this->lastCallId = 0;
        }<commit_msg>Dispatcher::call now throws if the dispatcher has not been started yet<commit_after><?php

namespace Amp\MultiProcess;

use Amp\Reactor,
    Amp\ReactorFactory;

class Dispatcher {
    
    const CALL        = 1;
    const CALL_RESULT = 2;
    const CALL_ERROR  = 3;
    const MAX_PROCEDURE_LENGTH = 255;
    const MAX_CALL_ID = 2147483647;
    
    private $reactor;
    private $workerSessionFactory;
    private $autoWriteSubscription;
    private $autoTimeoutSubscription;
    
    private $workerSessions = [];
    private $pendingCallCounts = [];
    private $workerSubscriptions = [];
    private $workerCallMap = [];
    private $callWorkerMap = [];
    private $callFrames = [];
    private $availableWorkers = [];
    private $timeoutSchedule = [];
    private $onResultCallbacks = [];
    private $partialResults = [];
    private $writableWorkers;
    
    private $poolSize;
    private $workerCmd;
    private $workerCwd;
    private $lastCallId = 0;
    private $pendingCalls = 0;
    private $callTimeout = 30;
    private $readTimeout = 60;
    private $autoWriteInterval = 0.025;
    private $timeoutCheckInterval = 1;
    private $writeErrorsTo = STDERR;
    private $isStarted = FALSE;
    private $chrCallCode;
    
    function __construct(Reactor $reactor, $workerCmd, $poolSize = 1, WorkerSessionFactory $wsf = NULL, CallResultFactory $crf = NULL) {
        $this->reactor = $reactor;
        $this->workerCmd = $workerCmd;
        $this->poolSize = (is_int($poolSize) && $poolSize > 0) ? $poolSize : 1;
        $this->workerSessionFactory = $wsf ?: new WorkerSessionFactory;
        $this->callResultFactory = $crf ?: new CallResultFactory;
        
        $this->writableWorkers = new \SplObjectStorage;
        $this->chrCallCode = chr(self::CALL);
    }
    
    function setCallTimeout($seconds) {
        $this->callTimeout = filter_var($seconds, FILTER_VALIDATE_FLOAT, ['options' => [
            'min_range' => 0,
            'default' => 30
        ]]);
    }
    
    function setGranularity($bytes) {
        $this->workerSessionFactory->setGranularity($bytes);
    }
    
    function setWorkerCwd($directory) {
        $this->workerCwd = $directory;
    }
    
    /**
     * Hydrate the worker process pool
     * 
     * Repeated calls to start have no effect after the first invocation.
     * 
     * @param int $poolSize The number of worker processes to maintain in the pool
     * @param string $cmd   The executable worker process command
     * @param string $cwd   An optional current working directory for worker processes. If not
     *                      specified workers will have the same cwd as the main process.
     * 
     * @return void
     */
    function start() {
        if ($this->isStarted) {
            return;
        }
        
        for ($i=0; $i < $this->poolSize; $i++) {
            $this->spawnWorkerSession();
        }
        
        $this->autoWriteSubscription = $this->reactor->repeat(function() {
            foreach ($this->writableWorkers as $workerSession) {
                if ($this->write($workerSession)) {
                    $this->writableWorkers->detach($workerSession);
                }
            }
        }, $this->autoWriteInterval);
        
        if ($this->callTimeout) {
            $this->autoTimeoutSubscription = $this->reactor->repeat(function() {
                $this->autoTimeout();
            }, $this->timeoutCheckInterval);
        }
        
        $this->isStarted = TRUE;
    }
    
    private function spawnWorkerSession() {
        $workerSession = $this->workerSessionFactory->__invoke(
            $this->workerCmd,
            $this->writeErrorsTo,
            $this->workerCwd
        );
        
        $readSubscription = $this->reactor->onReadable(
            $workerSession->getReadPipe(),
            function($pipe, $trigger) use ($workerSession) { $this->read($workerSession, $trigger); },
            $this->readTimeout
        );
        
        $workerId = spl_object_hash($workerSession);
        $this->workerSessions[$workerId] = $workerSession;
        $this->pendingCallCounts[$workerId] = 0;
        $this->workerSubscriptions[$workerId] = $readSubscription;
    }
    
    /**
     * Asynchronously execute a procedure and invoke the $onResult callback when complete
     * 
     * @param callable $onResult The callback to process the async execution's CallResult
     * @param string $procedure  The function to execute asynchronously
     * @param string $workload   The data to pass as an argument to the procedure
     * 
     * @return string Returns the task's call ID
     */
    function call(callable $onResult, $procedure, $workload = NULL) {
        if (!$this->isStarted) {
            throw new \RuntimeException(
                __CLASS__ . '::start() must be invoked before making calls'
            );
        }
        
        if (!is_string($procedure)) {
            throw new \InvalidArgumentException;
        } elseif (($procLen = strlen($procedure)) > self::MAX_PROCEDURE_LENGTH) {
            throw new \RangeException(
                'Procedure name exceeds maximum allowable length (' .
                self::MAX_PROCEDURE_LENGTH . '): ' . $procedure
            );
        }
        
        if (($callId = ++$this->lastCallId) == self::MAX_CALL_ID) {
            $this->lastCallId = 0;
        }
{'fraction_non_alphanumeric': 0.11267471712465532, 'fraction_numerical': 0.006846058762004374, 'mean_word_length': 2.244293645897594, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 102, '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': '4023147', 'n_tokens_mistral': 3016, 'n_tokens_neox': 2909, 'n_words': 697}