[0.1.0] handle idempotent and non-idempotent query + simplify parameters

Author: Benjamin Calef

Model/ForkedArrayProcessor.php

@@ -33,22 +33,20 @@ public function __construct(
      * @param callable $callback
      * @param int $pageSize
      * @param int $maxChildrenProcess
-     * @param bool $isParallelize
      * @return void
      */
     public function process(
         array $array,
         callable $callback,
         int $pageSize = 1000,
-        int $maxChildrenProcess = 10,
-        bool $isParallelize = true
+        int $maxChildrenProcess = 10
     ): void {
         /** @var ArrayWrapper $itemProvider */
         $itemProvider = $this->arrayWrapperFactory->create([
             'items' => $array,
             'pageSize' => $pageSize
         ]);
 
-        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess, $isParallelize, false);
+        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess);
     }
 }

Model/ForkedCollectionProcessor.php

@@ -34,25 +34,24 @@ public function __construct(
      * @param callable $callback
      * @param int $pageSize
      * @param int $maxChildrenProcess
-     * @param bool $isParallelize
-     * @param bool $isFallBackEnabled
+     * @param bool $isIdempotent
      * @return void
      */
     public function process(
         Collection $collection,
         callable $callback,
         int $pageSize = 1000,
         int $maxChildrenProcess = 10,
-        bool $isParallelize = true,
-        bool $isFallBackEnabled = false
+        bool $isIdempotent = true
     ): void {
         /** @var CollectionWrapper $itemProvider */
         $itemProvider = $this->collectionWrapperFactory->create([
             'collection' => $collection,
-            'maxChildrenProcess' => $maxChildrenProcess,
             'pageSize' => $pageSize,
+            'maxChildrenProcess' => $maxChildrenProcess,
+            'isIdempotent' => $isIdempotent
         ]);
 
-        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess, $isParallelize, $isFallBackEnabled);
+        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess);
     }
 }

Model/ForkedSearchResultProcessor.php

@@ -32,8 +32,7 @@ public function __construct(
      * @param callable $callback
      * @param int $pageSize
      * @param int $maxChildrenProcess
-     * @param bool $isParallelize
-     * @param bool $isFallBackEnabled
+     * @param bool $isIdempotent
      * @return void
      */
     public function process(
@@ -42,8 +41,7 @@ public function process(
         callable $callback,
         int $pageSize = 1000,
         int $maxChildrenProcess = 10,
-        bool $isParallelize = true,
-        bool $isFallBackEnabled = false
+        bool $isIdempotent = true
     ): void {
         if (!method_exists($repository, 'getList')) {
             throw new InvalidArgumentException('The repository class must have a method called "getList"');
@@ -54,9 +52,10 @@ public function process(
             'searchCriteria' => $searchCriteria,
             'repository' => $repository,
             'pageSize' => $pageSize,
-            'maxChildrenProcess' => $maxChildrenProcess
+            'maxChildrenProcess' => $maxChildrenProcess,
+            'isIdempotent' => $isIdempotent
         ]);
 
-        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess, $isParallelize, $isFallBackEnabled);
+        $this->forkedProcessorRunner->run($itemProvider, $callback, $maxChildrenProcess);
     }
 }

Model/ItemProvider/ArrayWrapper.php

@@ -66,4 +66,12 @@ public function getItems(): array
 
         return array_slice($this->items, $offset, $this->pageSize);
     }
+
+    /**
+     * @inheirtDoc
+     */
+    public function isIdempotent(): bool
+    {
+        return true;
+    }
 }

Model/ItemProvider/CollectionWrapper.php

@@ -17,16 +17,25 @@ class CollectionWrapper implements ItemProviderInterface
     /** @var int */
     private $maxChildrenProcess;
 
+    /** @var bool */
+    private $isIdempotent;
+
     /**
      * @param Collection $collection
      * @param int $pageSize
      * @param int $maxChildrenProcess
+     * @param bool $isIdempotent
      */
-    public function __construct(Collection $collection, int $pageSize, int $maxChildrenProcess)
-    {
+    public function __construct(
+        Collection $collection,
+        int $pageSize,
+        int $maxChildrenProcess,
+        bool $isIdempotent
+    ) {
         $this->collection = $collection;
         $this->pageSize = $pageSize;
         $this->maxChildrenProcess = $maxChildrenProcess;
+        $this->isIdempotent = $maxChildrenProcess > 1 ? $isIdempotent : true;
     }
 
     /**
@@ -35,7 +44,7 @@ public function __construct(Collection $collection, int $pageSize, int $maxChild
     public function setCurrentPage(int $currentPage): void
     {
         $this->collection->setPageSize($this->getPageSize());
-        if ($this->maxChildrenProcess > 1) {
+        if (!$this->isIdempotent()) {
             $moduloPage = $currentPage % $this->maxChildrenProcess;
             $currentPage = $moduloPage === 0 ? $this->maxChildrenProcess : $moduloPage;
         }
@@ -79,4 +88,12 @@ public function getItems(): array
 
         return $this->collection->getItems();
     }
+
+    /**
+     * @inheirtDoc
+     */
+    public function isIdempotent(): bool
+    {
+        return $this->isIdempotent;
+    }
 }

Model/ItemProvider/ItemProviderInterface.php

@@ -31,4 +31,9 @@ public function getTotalPages(): int;
      * @return array
      */
     public function getItems(): array;
+
+    /**
+     * @return bool
+     */
+    public function isIdemPotent(): bool;
 }

Model/ItemProvider/SearchResultWrapper.php

@@ -21,22 +21,28 @@ class SearchResultWrapper implements ItemProviderInterface
     /** @var int */
     private $maxChildrenProcess;
 
+    /** @var bool */
+    private $isIdempotent;
+
     /**
      * @param SearchCriteria $searchCriteria
      * @param $repository
      * @param int $pageSize
      * @param int $maxChildrenProcess
+     * @param bool $isIdempotent
      */
     public function __construct(
         SearchCriteria $searchCriteria,
         $repository,
         int $pageSize,
-        int $maxChildrenProcess
+        int $maxChildrenProcess,
+        bool $isIdempotent
     ) {
         $this->searchCriteria = $searchCriteria;
         $this->repository = $repository;
         $this->pageSize = $pageSize;
         $this->maxChildrenProcess = $maxChildrenProcess;
+        $this->isIdempotent = $maxChildrenProcess > 1 ? $isIdempotent : true;
     }
 
     /**
@@ -45,7 +51,7 @@ public function __construct(
     public function setCurrentPage(int $currentPage): void
     {
         $this->searchCriteria->setPageSize($this->getPageSize());
-        if ($this->maxChildrenProcess > 1) {
+        if (!$this->isIdempotent()) {
             $moduloPage = $currentPage % $this->maxChildrenProcess;
             $currentPage = $moduloPage === 0 ? $this->maxChildrenProcess : $moduloPage;
         }
@@ -87,6 +93,14 @@ public function getItems(): array
         return $this->getSearchResults()->getItems();
     }
 
+    /**
+     * @inheirtDoc
+     */
+    public function isIdempotent(): bool
+    {
+        return $this->isIdempotent;
+    }
+
     /**
      * @return SearchResultsInterface
      */

Model/Processor/ForkedProcessor.php

@@ -22,9 +22,6 @@ class ForkedProcessor
     /** @var int */
     private $maxChildrenProcess;
 
-    /** @var bool */
-    private $isFallBackEnabled;
-
     /** @var bool */
     private $running = true;
 
@@ -33,22 +30,17 @@ class ForkedProcessor
      * @param ItemProviderInterface $itemProvider
      * @param callable $callback
      * @param int $maxChildrenProcess
-     * @param bool $isParallelize
-     * @param bool $isFallBackEnabled
      */
     public function __construct(
         LoggerInterface $logger,
         ItemProviderInterface $itemProvider,
         callable $callback,
-        int $maxChildrenProcess = 10,
-        bool $isParallelize = true,
-        bool $isFallBackEnabled = false
+        int $maxChildrenProcess = 10
     ) {
         $this->logger = $logger;
         $this->itemProvider = $itemProvider;
         $this->callback = $callback;
-        $this->maxChildrenProcess = $isParallelize ? $maxChildrenProcess : 1;
-        $this->isFallBackEnabled = $isFallBackEnabled;
+        $this->maxChildrenProcess = $maxChildrenProcess;
         pcntl_signal(SIGINT, [$this, 'handleSigInt']);
     }
 
@@ -157,7 +149,7 @@ private function handleMultipleChildProcesses(): void
         }
 
         // Fallback based on database query
-        if ($this->isFallBackEnabled) {
+        if (!$this->itemProvider->isIdemPotent()) {
             $size = $this->itemProvider->getSize();
             $this->logger->info('Missing items from original query collection', ['total_items' => $size]);
             if ($size !== 0) {

Model/Processor/ForkedProcessorRunner.php

@@ -22,24 +22,18 @@ public function __construct(
      * @param ItemProviderInterface $itemProvider
      * @param callable $callback
      * @param int $maxChildrenProcess
-     * @param bool $isParallelize
-     * @param bool $isFallBackEnabled
      * @return void
      */
     public function run(
         ItemProviderInterface $itemProvider,
         callable $callback,
-        int $maxChildrenProcess,
-        bool $isParallelize,
-        bool $isFallBackEnabled
+        int $maxChildrenProcess
     ): void {
         /** @var $forkedProcessor ForkedProcessor */
         $forkedProcessor = $this->forkedProcessorFactory->create([
             'itemProvider' => $itemProvider,
             'callback' => $callback,
             'maxChildrenProcess' => $maxChildrenProcess,
-            'isParallelize' => $isParallelize,
-            'isFallBackEnabled' => $isFallBackEnabled
         ]);
 
         $forkedProcessor->process();

README.md

@@ -51,9 +51,9 @@ class MyAwesomeClass
         $searchCriteria,
         $productRepository,
         $callback,
-        $pageSize = 100,
+        $pageSize = 1000,
         $maxChildrenProcess = 10,
-        $isParallelize = true
+        $isIdempotent = true
     );
 }
 ```
@@ -86,9 +86,9 @@ class MyAwesomeClass
         $searchCriteria,
         $productRepository,
         $callback,
-        $pageSize = 100,
+        $pageSize = 1000,
         $maxChildrenProcess = 10,
-        $isParallelize = true
+        $isIdempotent = true
     );
 }
 ```
@@ -119,8 +119,7 @@ class MyAwesomeClass
         $array,
         $callback,
         $pageSize = 2,
-        $maxChildrenProcess = 2,
-        $isParallelize = true
+        $maxChildrenProcess = 2
     );
 }
 ```
@@ -166,16 +165,18 @@ function specified by the user on each item of that page.
 
 - `$maxChildrenProcess`: This parameter is used to set the maximum number of child
   processes that can be run simultaneously. This is used to control the number of threads
-  that will be used by the multi-threading process.
-
-- `$isParallelize`: This parameter is used to set whether the multi-threading process should run
-  in parallel or sequentially. If set to true, the process will run in parallel,
-  if set to false, the process will run sequentially. Running task sequentially may be useful when
-  you want to keep the sort order of your items.
-
-- `$isFallBackEnabled`: This parameter is set to false by default and can be used for `ForkedSearchResultProcessor` 
-  or `ForkedCollectionProcessor` it can be useful only if your initial query has a flag
-  that will be processed in your callback method.
+  that will be used by the multi-threading process. If set to 1, by definition you will have no parallelization, 
+  the parent process will wait the child process to finish before creating another one.
+
+- `$isIdempotent`: This parameter is a flag set to `true` by default and can be used for `ForkedSearchResultProcessor` 
+  or `ForkedCollectionProcessor` when your `$maxChildrenProcess` is greater than one.
+  While fetching data from database with `ForkedSearchResult` and `ForkedCollectionProcessor` you may change values
+  queried: by modifying items on columns queried you will change the nature of the initial collection query and at the end, 
+  the OFFSET limit in the query will be invalid because the native pagination system expect the pagination to be
+  processed by only one process. To avoid that, set `$isIdempotent` to `false`.<br>
+  E.G.: In your collection query, you request all products `disabled`, in your callback method you `enable` and save 
+  them in database, then in this particular case you are modifying the column that you request in your collection,
+  your query is not idempotent.
 
 ### Memory Limit
 This module allows to bypass the limitation of the memory limit, because the memory
@@ -188,28 +189,6 @@ system and adjust the parameters accordingly.
 ### Limitations
 This module uses `pcntl_fork()` function which is not available on Windows.
 
-There is limitation for Multi-threading pagination on `ForkedSearchResultProcessor` and `ForkedCollectionProcessor`:<br>
-Multi-threading pagination works only if your initial query has a flag that will be excluded 
-from results on each pagination iteration.
-
-For example:<br>
-You select all orders without a `gift_cards` value in your query, if you set a `gift_cards` value on your callback
-method and save it in database, then multi-threading and fallback system will work. 
-Otherwise, children will go for the same items because your query doesn't flag a specific column modified 
-by your callback function.<br>
-<b>So currently you need a flag to be sure to not proceed the same items</b>.
-
-Workaround:<br>
-You can create your own flag by using a column like modified_at in your collection/searchCriteria query 
-to request items with a modified_at value lower than the current timestamp, then you just have to update 
-the modified_at column in your callback method and set the current timestamp. 
-This way your query will fetch fresh items on each iteration.
-
-I will provide support to handle this soon, sorry for the inconvenient.
-Until then if you want to process data, I recommend usage of single thread solution, 
-or the `ForkedArrayProcessor` which is not impacted by pagination system.
-
-
 ### Conclusion
 This module provides a useful tool for running commands or processing collections
 and search criteria in a multi-threaded way, making it a great solution for improving

composer.json

@@ -2,7 +2,7 @@
     "name": "zepgram/module-multi-threading",
     "description": "This module is a powerful tool for developers who want to process large data sets in a short amount of time",
     "type": "magento2-module",
-    "version": "0.0.3",
+    "version": "0.1.0",
     "authors": [
         {
             "name": "Benjamin Calef",