Merge pull request #20 from hschimpf/2.x-dev

To release

README.md

@@ -265,6 +265,15 @@ foreach (Scheduler::getTasks() as $task) {
 }
 ```
 
+### Specifying the No. of CPU Cores
+You can control the maximum percentage or number of CPU cores to use by calling the following methods:
+```php
+use HDSSolutions\Console\Parallel\Scheduler;
+
+Scheduler::setMaxCpuCountUsage(2);        // Use at max two CPU cores
+Scheduler::setMaxCpuPercentageUsage(0.5); // Use at max 50% of the total of CPU cores
+```
+
 ### ProgressBar
 
 #### Requirements
@@ -320,10 +329,14 @@ final class ExampleWorker extends ParallelWorker {
 2. [Parallel\Runtime::run() Task Characteristics](https://www.php.net/manual/en/parallel-runtime.run.php#refsect1-parallel-runtime.run-closure-characteristics)
 
 # Security Vulnerabilities
-If you encounter any security related issue, feel free to raise a ticket on the issue tracker.
+If you encounter any security-related issues, please feel free to raise a ticket on the issue tracker.
+
+# Contributing
+Contributions are welcome! If you find any issues or would like to add new features or improvements, please feel free to submit a pull request.
 
-# Contributors
+## Contributors
 - [Hermann D. Schimpf](https://hds-solutions.net)
 
 # Licence
-GPL-3.0 Please see [License File](LICENSE) for more information.
+This library is open-source software licensed under the [MIT License](LICENSE).
+Please see the [License File](LICENSE) for more information.

src/Internals/Commands/Runner/SetMaxCpuUsage.php

@@ -0,0 +1,16 @@
+<?php declare(strict_types=1);
+
+namespace HDSSolutions\Console\Parallel\Internals\Commands\Runner;
+
+use HDSSolutions\Console\Parallel\Internals\Commands\ParallelCommandMessage;
+
+/**
+ * Message sent to {@see Runner} to execute {@see Runner::setMaxCpuCountUsage()} | {@see Runner::setMaxCpuPercentageUsage()} action
+ */
+final class SetMaxCpuUsage extends ParallelCommandMessage {
+
+    public function __construct(int $max, bool $percentage = false) {
+        parent::__construct(sprintf('set_max_cpu_%s_usage', $percentage ? 'percentage' : 'count'), [ $max ]);
+    }
+
+}

src/Internals/ProgressBarWorker.php

@@ -62,7 +62,7 @@ protected function progressBarAction(string $action, array $args): void {
 
         if ($action === 'advance') {
             // count processed item
-            $this->items[ time() ] = ($this->items[ time() ] ?? 0) + 1;
+            $this->items[ time() ] = ($this->items[ time() ] ?? 0) + (int) array_shift($args);
             // update ProgressBar items per second report
             $this->progressBar->setMessage($this->getItemsPerSecond(), 'items_per_second');
         }

src/Internals/Runner.php

@@ -40,6 +40,14 @@ private function getMaxCpuUsage(): int {
         return $this->max_cpu_count ??= (isset($_SERVER['PARALLEL_MAX_COUNT']) ? (int) $_SERVER['PARALLEL_MAX_COUNT'] : cpu_count( (float) ($_SERVER['PARALLEL_MAX_PERCENT'] ?? 1.0) ));
     }
 
+    protected function setMaxCpuCountUsage(int $count): int {
+        return $this->send($this->max_cpu_count = $count);
+    }
+
+    protected function setMaxCpuPercentageUsage(float $percentage): int {
+        return $this->send($this->max_cpu_count = max(1, cpu_count($percentage)));
+    }
+
     protected function getRegisteredWorker(string $worker): RegisteredWorker | false {
         if ( !array_key_exists($worker, $this->workers_hashmap)) {
             return $this->send(false);

src/Scheduler.php

@@ -53,6 +53,44 @@ private static function instance(): self {
         return self::$instance ??= new self();
     }
 
+    /**
+     * Sets the maximum number of CPU cores to use in parallel
+     *
+     * @param  int  $count  No. of CPU cores. (minimum: 1)
+     *
+     * @return int No. of assigned CPU cores
+     */
+    public static function setMaxCpuCountUsage(int $count): int {
+        $message = new Commands\Runner\SetMaxCpuUsage(max(1, $count));
+
+        if (PARALLEL_EXT_LOADED) {
+            self::instance()->send($message);
+
+            return self::instance()->recv();
+        }
+
+        return self::instance()->runner->processMessage($message);
+    }
+
+    /**
+     * Sets the maximum percentage of cpu cores to use in parallel
+     *
+     * @param  float  $percentage  Percentage of CPU cores. (between 0 and 1.0)
+     *
+     * @return int No. of assigned CPU cores
+     */
+    public static function setMaxCpuPercentageUsage(float $percentage): int {
+        $message = new Commands\Runner\SetMaxCpuUsage(max(0, min($percentage, 1.0)), percentage: true);
+
+        if (PARALLEL_EXT_LOADED) {
+            self::instance()->send($message);
+
+            return self::instance()->recv();
+        }
+
+        return self::instance()->runner->processMessage($message);
+    }
+
     /**
      * Register a worker class to process tasks
      *

tests/ParallelTest.php

@@ -209,4 +209,25 @@ public function testThatChannelsDontOverlap(): void {
         Scheduler::removeAllTasks();
     }
 
+    /**
+     * @depends testThatParallelExtensionIsAvailable
+     */
+    public function testThatCpuUsageCanBeControlled(): void {
+        Scheduler::setMaxCpuCountUsage(1);
+        Scheduler::using(static fn() => usleep(250_000));
+
+        $start = time();
+        foreach (range(1, 5) as $task) {
+            try { Scheduler::runTask($task);
+            } catch (Throwable) {
+                Scheduler::stop();
+            }
+        }
+
+        Scheduler::awaitTasksCompletion();
+        Scheduler::removeAllTasks();
+
+        $this->assertGreaterThanOrEqual(1, time() - $start);
+    }
+
 }