ProcessManager.php

Go to the documentation of this file.

<?php
declare(strict_types=1);

namespace Magento\Indexer\Model;

class ProcessManager
{
    const THREADS_COUNT = 'MAGE_INDEXER_THREADS_COUNT';

    private $failInChildProcess = false;

    private $resource;

    private $registry;

    private $threadsCount;

    public function __construct(
        \Magento\Framework\App\ResourceConnection $resource,
        \Magento\Framework\Registry $registry = null,
        int $threadsCount = null
    ) {
        $this->resource = $resource;
        if (null === $registry) {
            $registry = \Magento\Framework\App\ObjectManager::getInstance()->get(
                \Magento\Framework\Registry::class
            );
        }
        $this->registry = $registry;
        $this->threadsCount = (int)$threadsCount;
    }

    public function execute($userFunctions)
    {
        if ($this->threadsCount > 1 && $this->isCanBeParalleled() && !$this->isSetupMode() && PHP_SAPI == 'cli') {
            $this->multiThreadsExecute($userFunctions);
        } else {
            $this->simpleThreadExecute($userFunctions);
        }
    }

    private function simpleThreadExecute($userFunctions)
    {
        foreach ($userFunctions as $userFunction) {
            call_user_func($userFunction);
        }
    }

    private function multiThreadsExecute($userFunctions)
    {
        $this->resource->closeConnection(null);
        $threadNumber = 0;
        foreach ($userFunctions as $userFunction) {
            $pid = pcntl_fork();
            if ($pid == -1) {
                throw new \RuntimeException('Unable to fork a new process');
            } elseif ($pid) {
                $this->executeParentProcess($threadNumber);
            } else {
                $this->startChildProcess($userFunction);
            }
        }
        while (pcntl_waitpid(0, $status) != -1) {
            //Waiting for the completion of child processes
        }

        if ($this->failInChildProcess) {
            throw new \RuntimeException('Fail in child process');
        }
    }

    private function isCanBeParalleled(): bool
    {
        return function_exists('pcntl_fork');
    }

    private function isSetupMode(): bool
    {
        return $this->registry->registry('setup-mode-enabled') ?: false;
    }

    private function startChildProcess(callable $userFunction)
    {
        $status = call_user_func($userFunction);
        $status = is_integer($status) ? $status : 0;
        exit($status);
    }

    private function executeParentProcess(int &$threadNumber)
    {
        $threadNumber++;
        if ($threadNumber >= $this->threadsCount) {
            pcntl_wait($status);
            if (pcntl_wexitstatus($status) !== 0) {
                $this->failInChildProcess = true;
            }
            $threadNumber--;
        }
    }
}