Configurable decoder buffer size

Author: ondrejmirtes

conf/config.neon

@@ -46,6 +46,7 @@ parameters:
 		processTimeout: 60.0
 		maximumNumberOfProcesses: 32
 		minimumNumberOfJobsPerProcess: 2
+		buffer: 134217728 # 128 MB
 	polluteScopeWithLoopInitialAssignments: true
 	polluteScopeWithAlwaysIterableForeach: true
 	polluteCatchScopeWithTryAssignments: false
@@ -169,7 +170,8 @@ parametersSchema:
 		jobSize: int(),
 		processTimeout: float(),
 		maximumNumberOfProcesses: int(),
-		minimumNumberOfJobsPerProcess: int()
+		minimumNumberOfJobsPerProcess: int(),
+		buffer: int()
 	])
 	polluteScopeWithLoopInitialAssignments: bool()
 	polluteScopeWithAlwaysIterableForeach: bool()
@@ -416,6 +418,7 @@ services:
 		arguments:
 			internalErrorsCountLimit: %internalErrorsCountLimit%
 			processTimeout: %parallel.processTimeout%
+			decoderBufferSize: %parallel.buffer%
 
 	-
 		class: PHPStan\Parallel\Scheduler

src/Command/WorkerCommand.php

@@ -115,7 +115,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 		$tcpConector = new TcpConnector($loop);
 		$tcpConector->connect(sprintf('127.0.0.1:%d', $port))->then(function (ConnectionInterface $connection) use ($container, $identifier, $analysedFiles): void {
 			$out = new Encoder($connection);
-			$in = new Decoder($connection, true, 512, 0, 4 * 1024 * 1024);
+			$in = new Decoder($connection, true, 512, 0, $container->getParameter('parallel')['buffer']);
 			$out->write(['action' => 'hello', 'identifier' => $identifier]);
 			$this->runWorker($container, $out, $in, $analysedFiles);
 		});

src/Parallel/ParallelAnalyser.php

@@ -30,15 +30,20 @@ class ParallelAnalyser
 	/** @var ProcessPool */
 	private $processPool;
 
+	/** @var int */
+	private $decoderBufferSize;
+
 	public function __construct(
 		IgnoredErrorHelper $ignoredErrorHelper,
 		int $internalErrorsCountLimit,
-		float $processTimeout
+		float $processTimeout,
+		int $decoderBufferSize
 	)
 	{
 		$this->ignoredErrorHelper = $ignoredErrorHelper;
 		$this->internalErrorsCountLimit = $internalErrorsCountLimit;
 		$this->processTimeout = $processTimeout;
+		$this->decoderBufferSize = $decoderBufferSize;
 	}
 
 	/**
@@ -78,7 +83,7 @@ public function analyse(
 		$server = new \React\Socket\TcpServer('127.0.0.1:0', $loop);
 		$this->processPool = new ProcessPool($server);
 		$server->on('connection', function (ConnectionInterface $connection) use (&$jobs): void {
-			$decoder = new Decoder($connection, true, 512, 0, 4 * 1024 * 1024);
+			$decoder = new Decoder($connection, true, 512, 0, $this->decoderBufferSize);
 			$encoder = new Encoder($connection);
 			$decoder->on('data', function (array $data) use (&$jobs, $decoder, $encoder): void {
 				if ($data['action'] !== 'hello') {