src/WorkerPool.php
protected function collectWorkerResults($sec = 0) {
		$this->respawnIfRequired();

		// dispatch signals
		pcntl_signal_dispatch();

		if (isset($this->workerProcesses) === FALSE) {
			throw new WorkerPoolException('There is no list of worker processes. Maybe you destroyed the worker pool?', 1401179881);
		}
		$result = SimpleSocket::select($this->workerProcesses->getSockets(), array(), array(), $sec);
		foreach ($result['read'] as $socket) {
			/** @var $socket SimpleSocket */
			$processId = $socket->annotation['pid'];
			$result = $socket->receive();

			$possibleArrayKeys = array('data', 'poolException', 'workerException');
			if (is_array($result) && count(($resultTypes = array_intersect(array_keys($result), $possibleArrayKeys))) === 1) {
				// If the result has the expected format, free the worker and store the result.
				// Otherwise, the worker may be abnormally terminated (fatal error, exit(), ...) and will
				// fall in the reapers arms.
				$this->workerProcesses->registerFreeProcessId($processId);
				$result['pid'] = $processId;
				$resultType = reset($resultTypes);
				// Do not store NULL
				if ($resultType !== 'data' || $result['data'] !== NULL) {
					array_push($this->results, $result);
				}
			}
		}
		// dispatch signals
		pcntl_signal_dispatch();

		$this->respawnIfRequired();
	}
protected function collectWorkerResults($sec = 0) {
		$this->respawnIfRequired();

		// dispatch signals
		pcntl_signal_dispatch();

		if (isset($this->workerProcesses) === FALSE) {
			throw new WorkerPoolException('There is no list of worker processes. Maybe you destroyed the worker pool?', 1401179881);
		}
		$result = SimpleSocket::select($this->workerProcesses->getSockets(), array(), array(), $sec);
		foreach ($result['read'] as $socket) {
			/** @var $socket SimpleSocket */
			$processId = $socket->annotation['pid'];
			$result = $socket->receive();

			$possibleArrayKeys = array('data', 'poolException', 'workerException');
			if (is_array($result) && count(($resultTypes = array_intersect(array_keys($result), $possibleArrayKeys))) === 1) {
				// If the result has the expected format, free the worker and store the result.
				// Otherwise, the worker may be abnormally terminated (fatal error, exit(), ...) and will
				// fall in the reapers arms.
				$this->workerProcesses->registerFreeProcessId($processId);
				$result['pid'] = $processId;
				$resultType = reset($resultTypes);
				// Do not store NULL
				if ($resultType !== 'data' || $result['data'] !== NULL) {
					array_push($this->results, $result);
				}
			}
		}
		// dispatch signals
		pcntl_signal_dispatch();

		$this->respawnIfRequired();
	}
protected function collectWorkerResults ( $ sec = 0 ) { $ this -> respawnIfRequired ( ) ; // dispatch signals pcntl_signal_dispatch ( ) ; if ( isset ( $ this -> workerProcesses ) === FALSE ) { throw new WorkerPoolException ( 'There is no list of worker processes. Maybe you destroyed the worker pool?' , 1401179881 ) ; } $ result = SimpleSocket :: select ( $ this -> workerProcesses -> getSockets ( ) , array ( ) , array ( ) , $ sec ) ; foreach ( $ result [ 'read' ] as $ socket ) { /** @var $socket SimpleSocket */ $ processId = $ socket -> annotation [ 'pid' ] ; $ result = $ socket -> receive ( ) ; $ possibleArrayKeys = array ( 'data' , 'poolException' , 'workerException' ) ; if ( is_array ( $ result ) && count ( ( $ resultTypes = array_intersect ( array_keys ( $ result ) , $ possibleArrayKeys ) ) ) === 1 ) { // If the result has the expected format, free the worker and store the result. // Otherwise, the worker may be abnormally terminated (fatal error, exit(), ...) and will // fall in the reapers arms. $ this -> workerProcesses -> registerFreeProcessId ( $ processId ) ; $ result [ 'pid' ] = $ processId ; $ resultType = reset ( $ resultTypes ) ; // Do not store NULL if ( $ resultType !== 'data' || $ result [ 'data' ] !== NULL ) { array_push ( $ this -> results , $ result ) ; } } } // dispatch signals pcntl_signal_dispatch ( ) ; $ this -> respawnIfRequired ( ) ; }
Collects the results form the workers and processes any pending signals
@param int $sec timeout to wait for new results from the workers
@throws WorkerPoolException
Collects the results form the workers and processes any pending signals