Reads pipes, executes callback.

@param bool $blocking Whether to use blocking calls or not
@param bool $close    Whether to close file handles or not
181212
private function readPipes(bool $blocking, bool $close)
    {
        $result = $this->processPipes->readAndWrite($blocking, $close);

        $callback = $this->callback;
        foreach ($result as $type => $data) {
            if (3 !== $type) {
                $callback(self::STDOUT === $type ? self::OUT : self::ERR, $data);
            } elseif (!isset($this->fallbackStatus['signaled'])) {
                $this->fallbackStatus['exitcode'] = (int) $data;
            }
        }
    }
399670
public function acquire($blocking = true)
    {
        if ($this->locked) {
            throw new \RuntimeException('already lock by yourself');
        }

        if ($blocking === false) {
            if (version_compare(PHP_VERSION, '5.6.0') < 0) {
                throw new \RuntimeException('php version is at least 5.6.0 for param blocking');
            }
            if (!sem_acquire($this->lock_id, true)) {
                return false;
            }
            $this->locked = true;

            return true;
        }

        if (!sem_acquire($this->lock_id)) {
            return false;
        }
        $this->locked = true;

        return true;
    }
426135
private function exec($interactive = false, $blocking = self::NON_BLOCKING)
    {
        $this->validate();

        $this->resource = proc_open($this->command->serialize(),
                                    $this->descriptorspec,
                                    $this->pipes,
                                    $this->cwd,
                                    null);

        if ($this->resource === false) {
            throw new ProcessException('Call to [proc_open] failed.', $this);
        }

        $this->start = microtime(true);

        $this->running = true;

        if (!$interactive && isset($this->pipes[static::STDIN]) && is_resource($this->pipes[static::STDIN])) {
            fclose($this->pipes[static::STDIN]);
        }

        $this->setStreamBlocking($blocking);
    }
669447
public function set_blocking()
    {
        if (!$this->socket || !is_resource($this->socket)) return false;
        Debug::Log(Debug::DEBUG, 'SOCKET: blocking mode: ON');
        stream_set_blocking($this->socket, true);
        return true;
    }
958625
private function finishAvgBlock($block)
    {
        if (!array_key_exists($block, $this->avgs)) {
            throw new \Exception('Average block '.$block.' not defined');
        }
        $this->avgs[$block]['avg'] = $this->avgs[$block]['total'] / $this->avgs[$block]['count'];

        return $this->avgs[$block];
    }
435321
protected function readBuffer($n)
    {
        $n = min($n, strlen($this->buffer) - $this->offset);
        if ($n === 0) {
            // substr('', 0, 0) => false, which screws up read loops that are
            // expecting non-blocking reads to return ''. This avoids that edge
            // case when the buffer is empty/used up.
            return '';
        }
        $block = substr($this->buffer, $this->offset, $n);
        $this->offset += $n;
        return $block;
    }
812828
public function closeWriteAndReadAll()
    {
        $this->closeWrite();
        $response = $this->read();
        while (!is_null($response)) {
            yield $response;
            $response = $this->read();
        }
    }
445389
public function wait($block = true, $sleep = 100000)
    {
        while (true) {
            if ($this->isRunning() === false) {
                return;
            }
            if (!$block) {
                break;
            }
            usleep($sleep);
        }
    }
307321
private void closeStreams(final Process process) throws IOException {
    IOException caught = null;

    try {
      process.getOutputStream().close();
    }
    catch (IOException e) {
      if (e.getMessage().equals("Stream closed")) {
        /**
         * OutputStream's contract for the close() method: If the stream is already closed then invoking this method has no effect.
         *
         * When a UNIXProcess exits ProcessPipeOutputStream automatically closes its target FileOutputStream and replaces it with NullOutputStream.
         * However the ProcessPipeOutputStream doesn't close itself at that moment.
         * As ProcessPipeOutputStream extends BufferedOutputStream extends FilterOutputStream closing it flushes the buffer first.
         * In Java 7 closing FilterOutputStream ignores any exception thrown by the target OutputStream. Since Java 8 these exceptions are now thrown.
         *
         * So since Java 8 after UNIXProcess detects the exit and there's something in the output buffer closing this stream throws IOException
         * with message "Stream closed" from NullOutputStream.
         */
        log.trace("Failed to close process output stream:", e);
      }
      else {
        log.error("Failed to close process output stream:", e);
        caught = add(caught, e);
      }
    }

    try {
      process.getInputStream().close();
    }
    catch (IOException e) {
      log.error("Failed to close process input stream:", e);
      caught = add(caught, e);
    }

    try {
      process.getErrorStream().close();
    }
    catch (IOException e) {
      log.error("Failed to close process error stream:", e);
      caught = add(caught, e);
    }

    if (caught != null) {
      throw caught;
    }
  }
246260
protected function lock(string $key, Closure $callback, int $lockType = LOCK_EX) {
		$filename = $this->generateFilename($key);
		$type     = is_file($filename) ? 'rb+' : 'wb+';
		if ($handle = fopen($filename, $type)) {
			if (flock($handle, $lockType)) {
				$callback($handle);
				flock($handle, LOCK_UN);
				fclose($handle);
				return true;
			}
		}
		return false;
	}
719000
private static function debugHasKnownBlockType(array $block): bool
    {
        $result = false;

        if (array_key_exists(BlockSerializer::TYPE, $block) === true) {
            $type = $block[BlockSerializer::TYPE];

            $result =
                $type === BlockSerializer::TYPE__PROCEDURE ||
                $type === BlockSerializer::TYPE__AND_EXPRESSION ||
                $type === BlockSerializer::TYPE__OR_EXPRESSION ||
                $type === BlockSerializer::TYPE__IF_EXPRESSION;
        }

        return $result;
    }
272608
public function findABlock($slug)
    {
        if (isset($this->blocks[$slug])) {
            return $this->blocks[$slug];
        }

        return false;
    }
1118436
public function wait($block = false, $interval = 100)
    {
        do {
            if ($this->isFinished()) {
                return;
            }
            parent::wait(false);
            if ($this->aliveCount() < $this->max) {
                foreach ($this->processes as $process) {
                    if ($process->isStarted()) continue;
                    $process->start();
                    if ($this->aliveCount() >= $this->max) break;
                }
            }
            $block ? usleep($interval) : null;
        } while ($block);
    }
1278376
public static function executeStarts(
        iterable $indexes,
        array $blocks,
        ContextStorageInterface $context,
        ErrorAggregatorInterface $errors
    ): bool {
        $allOk = true;

        foreach ($indexes as $index) {
            $context->setCurrentBlockId($index);
            $block      = $blocks[$index];
            $errorsInfo = static::executeProcedureStart($block, $context);
            if (empty($errorsInfo) === false) {
                static::addBlockErrors($errorsInfo, $context, $errors);
                $allOk = false;
            }
        }

        return $allOk;
    }
842750
public function stop($block)
    {
        if ($this->enabled) {
            if (!isset($this->blocks[$block])) {
                throw new \Exception('Block '.$block.' has not been started!');
            }

            $this->blocks[$block]['stop'] = microtime(true);
            $this->blocks[$block]['stop-line'] = $this->getLineNumber();
        }
    }
380080
public static function nblock($lock, \Closure $callback, $timeout = 120) {
        try {
            $redis = RedisClient::getRedis();
            if ($redis) {
                $cnt = $redis->incr($lock);
                if ($cnt != 1) {
                    return false;
                }
                $redis->setTimeout($lock, $timeout);
                try {
                    return $callback($redis);
                } catch (\Exception $e) {
                    log_warn($e->getMessage(), 'redis_lock');
                } finally {
                    $redis->decr($lock);
                }
            }
        } catch (\Exception $e) {
            log_warn($e->getMessage(), 'redis_lock');
        }

        return false;
    }
310112
public function block($type = null, $extra = null)
    {
        $this->addContext('blocked', true);

        if (isset($type)) {
            $this->registerEvent($type, $extra);
        }

        if (is_callable($this->responseCodeHandler)) {
            $responseCodeHandler = $this->responseCodeHandler;
            $responseCodeHandler(403, 'Forbidden');
        }
        else {
            $this->error('No response code handler available.');
        }

        if (is_callable($this->exitHandler)) {
            $exitHandler = $this->exitHandler;
            $exitHandler();
        }
        else {
            // $this->error('No exit callable set. PHP exit construct will be used.');
            exit;
        }
    }