MAGETWO-80789: [Indexer] Search Indexer is scoped & multi-threaded

Author: slopukhov

app/code/Magento/CatalogSearch/Model/Indexer/Fulltext.php

@@ -77,11 +77,6 @@ class Fulltext implements \Magento\Framework\Indexer\ActionInterface, \Magento\F
      */
     private $processManager;
 
-    /**
-     * @var int
-     */
-    private $threadsCount;
-
     /**
      * @param FullFactory $fullActionFactory
      * @param IndexerHandlerFactory $indexerHandlerFactory
@@ -93,7 +88,6 @@ class Fulltext implements \Magento\Framework\Indexer\ActionInterface, \Magento\F
      * @param IndexSwitcherInterface $indexSwitcher
      * @param Scope\State $indexScopeState
      * @param ProcessManager $processManager
-     * @param int $threadsCount
      */
     public function __construct(
         FullFactory $fullActionFactory,
@@ -105,8 +99,7 @@ public function __construct(
         array $data,
         IndexSwitcherInterface $indexSwitcher = null,
         State $indexScopeState = null,
-        ProcessManager $processManager = null,
-        int $threadsCount = 4 //Will be moved to the env.php
+        ProcessManager $processManager = null
     ) {
         $this->fullAction = $fullActionFactory->create(['data' => $data]);
         $this->indexerHandlerFactory = $indexerHandlerFactory;
@@ -127,7 +120,6 @@ public function __construct(
         $this->indexSwitcher = $indexSwitcher;
         $this->indexScopeState = $indexScopeState;
         $this->processManager = $processManager;
-        $this->threadsCount = $threadsCount;
     }
 
     /**
@@ -169,7 +161,7 @@ public function executeFull()
 
         }
 
-        $this->processManager->execute($userFunctions, $this->threadsCount);
+        $this->processManager->execute($userFunctions);
 
         $this->fulltextResource->resetSearchResults();
         $this->searchRequestConfig->reset();

app/code/Magento/Indexer/Model/ProcessManager.php

@@ -10,36 +10,46 @@
  */
 class ProcessManager
 {
+    /**
+     * Threads count environment variable name
+     */
+    const THREADS_COUNT = 'MAGE_INDEXER_THREADS_COUNT';
+
     /** @var bool */
     private $failInChildProcess = false;
 
     /** @var \Magento\Framework\App\ResourceConnection */
     private $resource;
 
+    /** @var int|null */
+    private $threadsCount;
+
     /**
      * @param \Magento\Framework\App\ResourceConnection $resource
+     * @param int|null $threadsCount
      */
     public function __construct(
-        \Magento\Framework\App\ResourceConnection $resource = null
+        \Magento\Framework\App\ResourceConnection $resource = null,
+        int $threadsCount = null
     ) {
         if (null === $resource) {
             $resource = \Magento\Framework\App\ObjectManager::getInstance()->get(
                 \Magento\Framework\App\ResourceConnection::class
             );
         }
         $this->resource = $resource;
+        $this->threadsCount = (int)$threadsCount;
     }
 
     /**
      * Execute user functions
      *
      * @param \Traversable $userFunctions
-     * @param int $threadsCount
      */
-    public function execute($userFunctions, $threadsCount)
+    public function execute($userFunctions)
     {
-        if ($threadsCount > 1 && $this->isCanBeParalleled()) {
-            $this->multiThreadsExecute($userFunctions, $threadsCount);
+        if ($this->threadsCount > 1 && $this->isCanBeParalleled()) {
+            $this->multiThreadsExecute($userFunctions);
         } else {
             $this->simpleThreadExecute($userFunctions);
         }
@@ -61,9 +71,8 @@ private function simpleThreadExecute($userFunctions)
      * Execute user functions in in multiThreads mode
      *
      * @param \Traversable $userFunctions
-     * @param int $threadsCount
      */
-    private function multiThreadsExecute($userFunctions, $threadsCount)
+    private function multiThreadsExecute($userFunctions)
     {
         $this->resource->closeConnection(null);
         $threadNumber = 0;
@@ -72,7 +81,7 @@ private function multiThreadsExecute($userFunctions, $threadsCount)
             if ($pid == -1) {
                 throw new \RuntimeException('Unable to fork a new process');
             } elseif ($pid) {
-                $this->executeParentProcess($threadNumber, $threadsCount);
+                $this->executeParentProcess($threadNumber);
             } else {
                 $this->startChildProcess($userFunction);
             }
@@ -111,12 +120,11 @@ private function startChildProcess($userFunction)
      * Execute parent process
      *
      * @param int $threadNumber
-     * @param int $threadsCount
      */
-    private function executeParentProcess(&$threadNumber, $threadsCount)
+    private function executeParentProcess(&$threadNumber)
     {
         $threadNumber++;
-        if ($threadNumber >= $threadsCount) {
+        if ($threadNumber >= $this->threadsCount) {
             pcntl_wait($status);
             if (pcntl_wexitstatus($status) !== 0) {
                 $this->failInChildProcess = true;

app/code/Magento/Indexer/etc/di.xml

@@ -42,7 +42,11 @@
         <plugin name="page-cache-indexer-reindex-clean-cache"
                 type="Magento\Indexer\Model\Processor\CleanCache" sortOrder="10"/>
     </type>
-
+    <type name="\Magento\Indexer\Model\ProcessManager">
+        <arguments>
+            <argument name="threadsCount" xsi:type="init_parameter">Magento\Indexer\Model\ProcessManager::THREADS_COUNT</argument>
+        </arguments>
+    </type>
     <type name="Magento\Framework\Console\CommandListInterface">
         <arguments>
             <argument name="commands" xsi:type="array">