Optimizing Jenkins execution

class NewWorker implements WorkerInterface
{
    // ...    
    /**
     * @var \SplFixedArray<\Symfony\Component\Process\Process>
     */
    protected SplFixedArray $processes;
    // ...

    public function __construct(...)
    {
        // ...
        // can be configured in config and/or using environment variable QUEUE_ONE_WORKER_POOL_SIZE,
        // average recommended values are 5-10
        // defines how many PHP processes (`queue:task:start QUEUE-NAME`) allowed to run simultaneously
        // within NewWorker regardless of number of stores or queues
        $this->processes = new SplFixedArray($this->queueConfig->getQueueWorkerMaxProcesses());
    }

    public function start(string $command, array $options = []): void
    {
        // env var - QUEUE_WORKER_MAX_THRESHOLD_SECONDS
        // default is 60 seconds, 1 minute, it is safe to have it as 1 hour instead
        $maxThreshold = $this->queueConfig->getQueueWorkerMaxThreshold();

        // minimum interval after starting one process before executing another
        // config - QUEUE_WORKER_INTERVAL_MILLISECONDS, default is 1000 - 1s, recommended value = 100, 0.1s
        $delayIntervalMilliseconds = $this->queueConfig->getQueueWorkerInterval();

        // when false - there will be an exception thrown if the Worker can't read the system memory info
        // otherwise - memory info will be returned as 0, so the system will continue to work, but not launching processes
        // because it'll think there is no memory available
        // QUEUE_WORKER_IGNORE_MEM_READ_FAILURE, default = false
        $shouldIgnoreZeroMemory = $this->queueConfig->shouldIgnoreNotDetectedFreeMemory();

        $startTime = microtime(true);
        $lastStart = 0;
        $maxMemGrowthFactor = 0;

        while (microtime(true) - $startTime < $maxThreshold) {
            if (!$this->sysResManager->isEnoughResources($shouldIgnoreZeroMemory)) {
                // optional logging here
                continue;
            }

            $freeIndex = $this->removeFinishedProcesses();
            if ($freeIndex === null) {
                // any optional logging here for the case when there are no slots available
            } elseif ((microtime(true) - $lastStart) * 1000 > $delayIntervalMilliseconds) {
                $lastStart = microtime(true);
                $this->executeQueueProcessingStrategy($freeIndex);
            } else {
                // any optional logging for cooldown period
            }

            $ownMemGrowthFactor = $this->sysResManager->getOwnPeakMemoryGrowth();
            $maxMemGrowthFactor = max($ownMemGrowthFactor, $maxMemGrowthFactor);

            // QUEUE_WORKER_MEMORY_MAX_GROWTH_FACTOR, 50 by default
            // measures how much Worker own memory consumption increased after first iteration
            // when more than 50% - it is considered a memory leak and Worker will finish its operation
            // allowing Jenkins to run Worker again
            if ($ownMemGrowthFactor > $this->queueConfig->maxAllowedWorkerMemoryGrowthFactor()) {
                $this->logger->emergency(sprintf('Worker memory grew more than %d%%, probably a memory leak, exiting', $ownMemGrowthFactor));
                break;
            }
        }

        // to re-scan previously logged processes and update stats
        $this->removeFinishedProcesses();
        $this->processManager->flushIdleProcesses();

        // here you can have any summary logging/stats, similar as we have in the patch
    }

    // ...

    /**
     * Removes finished processes from the processes fixed array
     * Returns the first index of the array that is available for new processes
     *
     * @return int|null
     */
    protected function removeFinishedProcesses(): ?int
    {
        $freeIndex = -1;
        $runningProcCount = 0;

        foreach ($this->processes as $idx => $process) {
            if (!$process) {
                $freeIndex = $freeIndex >= 0 ? $freeIndex : $idx;
                continue;
            }

            if ($process->isRunning()) {
                $runningProcCount++;
                continue;
            }

            unset($this->processes[$idx]); // won't affect foreach
            $freeIndex = $freeIndex >= 0 ? $freeIndex : $idx;

            // any custom logging here
        }

        return $runningProcCount === $this->processes->count() ? null : $freeIndex;
    }

    // ...

    /**
     * Strategy defines which queue to return for processing,
     * it can have any other custom dependencies to make a decision.
     *
     * Strategy can be different, we can inject some smart strategy
     * which will delegate actual processing to another one depending on something, e.g. store operation times or time zones, etc.
     *
     * @param int $freeIndex
     *
     * @return void
     */
    protected function executeQueueProcessingStrategy(int $freeIndex): void
    {
        $queueTransfer = $this->queueProcessingStrategy->getNextQueue();
        if (!$queueTransfer) {
            // logging
            return;
        }

        $process = $this->processManager->triggerQueueProcess(
            sprintf(
                'APPLICATION_STORE=%s %s %s',
                $queueTransfer->getStoreName(),
                QueueWorkerConsole::QUEUE_RUNNER_COMMAND,
                $queueTransfer->getQueueName(),
            ),
            sprintf('%s.%s', $queueTransfer->getStoreName(), $queueTransfer->getQueueName()),
        );

        $this->processes[$freeIndex] = $process;
    }

    // ...

class SystemResourcesManager implements SystemResourcesManagerInterface
{
    // ...

    /**
     * Executed frequently in a loop within X minutes
     * We have a choice on what to do in case we failed to determine free memory (e.g. 0)
     *   A. consider we have NO free memory, so no processes will run
     *   B. consider it as a critical issue and throw an error
     * ...
     */
    public function enoughResources(bool $shouldIgnore = false): bool
    {
        $freeMemory = $this->getFreeMemory();
        if ($freeMemory === 0 && !$shouldIgnore) {
            throw new RuntimeException('Could not detect free memory and configured not to ignore that.');
        }

        // can be configured from config and/or environment variable - QUEUE_WORKER_FREE_MEMORY_BUFFER, in megabytes
        // default recommended value - 750 MB
        return $freeMemory > $this->queueConfig->getFreeMemoryBuffer();
    }

    /**
     * Read and parse system memory info
     */
    public function getFreeMemory(): int
    {
        $memory = $this->readSystemMemoryInfo();
        if (!preg_match_all('/(Mem\w+[l|e]):\s+(\d+)/msi', $memory, $matches, PREG_SET_ORDER)) {
            return 0;
        }

        $free = round((int)$matches[1][2] ?? 0) / 1024;
        $available = round((int)$matches[2][2] ?? 0) / 1024;

        return (int)max($free, $available);
    }

    /**
     * By how much own Worker memory consumption increased after first method invocation
     * @return int % of initial Worker consumption
     */
    public function getOwnPeakMemoryGrowth(): int
    {
        if (!$this->ownInitialMemoryConsumption) {
            $this->ownInitialMemoryConsumption = memory_get_peak_usage(true);
        }

        $diffNow = memory_get_peak_usage(true) - $this->ownInitialMemoryConsumption;

        return $diffNow <= 0 ? 0 : (int)round(100 * $diffNow / $this->ownInitialMemoryConsumption);
    }

    /**
     * @return string
     */
    private function readSystemMemoryInfo(): string
    {
        //
        $memoryReadProcessTimeout = $this->queueConfig->memoryReadProcessTimeout();
        $memory = @file_get_contents('/proc/meminfo') ?? '';

        return $memory ?? 0;
    }

class QueueScanner implements QueueScannerInterface
{
    // ...

    public function scanQueues(array $storeTransfers = [], int $emptyScanCooldownSeconds = 5): ArrayObject
    {
        // ...

        $sinceLastScan = microtime(true) - $this->lastScanAt;
        $lastEmptyScanTimeoutPassed = $this->lastScanWasEmpty && ($sinceLastScan > $emptyScanCooldownSeconds);

        if (!$this->lastScanWasEmpty || $lastEmptyScanTimeoutPassed) {
            $queueList = $this->directScanQueues($storeTransfers);

            $this->lastScanAt = microtime(true);
            $this->lastScanWasEmpty = $queueList->count() === 0;

            return $queueList;
        }

        return new ArrayObject();
    }

    /**
     * @param array<\Generated\Shared\Transfer\StoreTransfer> $storeTransfers
     *
     * @return \ArrayObject<\Generated\Shared\Transfer\QueueTransfer>
     */
    protected function directScanQueues(array $storeTransfers): ArrayObject
    {
        // ...
        $queuesPerStore = new ArrayObject();
        foreach ($storeTransfers as $storeTransfer) {
            foreach ($this->queueNames as $queueName) {

                $queueMessageCount = $this->mqClient->getQueueMetrics(
                    $queueName,
                    $storeTransfer->getName(),
                )['messageCount'] ?? 0;

                if ($queueMessageCount === 0) {
                    continue;
                }

                $queuesPerStore->append((new QueueTransfer())
                    ->setQueueName($queueName)
                    ->setStoreName($storeTransfer->getName())
                    ->setMsgCount($queueMessageCount)
                    ->setMsgToChunkSizeRatio(1), // default value
                );

                // ...
            }
        }

        return $queuesPerStore;
    }

    // ...

class ProcessManager extends SprykerProcessManager implements ProcessManagerInterface
{
    public function triggerQueueProcessForStore(string $storeCode, string $command, string $queue): Process
    {
        return $this->triggerQueueProcess($command, $this->getStoreBasedQueueName($storeCode, $queue));
    }

    public function getBusyProcessNumberForStore(string $storeCode, string $queueName): int
    {
        return $this->getBusyProcessNumber($this->getStoreBasedQueueName($storeCode, $queueName));
    }

    protected function getStoreBasedQueueName(string $storeCode, string $queueName): string
    {
        return sprintf('%s.%s', $storeCode, $queueName);
    }
}

class OrderedQueuesStrategy implements QueueProcessingStrategyInterface
{
    // ...

    /**
     * @param \Pyz\Zed\Queue\Business\QueueScannerInterface $queueScanner
     * @param \Psr\Log\LoggerInterface $logger
     */
    public function __construct(QueueScannerInterface $queueScanner, LoggerInterface $logger)
    {
        $this->queueScanner = $queueScanner;
        $this->currentIterator = new CountBasedIterator(new ArrayIterator());
    }

    /**
     * @return \Generated\Shared\Transfer\QueueTransfer|null
     */
    public function getNextQueue(): ?QueueTransfer
    {
        if (!$this->currentIterator->valid()) {
            $queuesPerStore = $this->getQueuesWithMessages();
            $this->currentIterator = new CountBasedIterator($queuesPerStore->getIterator());
        }

        /** @var \Generated\Shared\Transfer\QueueTransfer|null $queueTransfer */
        $queueTransfer = $this->currentIterator->current();
        $this->currentIterator->next();

        return $queueTransfer;
    }

    /**
     * @return \ArrayObject<int, \Generated\Shared\Transfer\QueueTransfer>
     */
    protected function getQueuesWithMessages(): ArrayObject
    {
        return $this->queueScanner->scanQueues();
    }