update: update some proc task logic

Author: inhere

README.md

@@ -37,7 +37,7 @@ ProcTasks::new(['procNum' => 2])
     ->onWorkersCreated(fn($pid, $info) => println("master [PID:$pid] - All task process started,", 'Workers info', $info))
     ->onWorkerStart(fn($pid, $id) => println("worker#$id started, pid is", $pid))
     ->onWorkerExit(fn($pid, $id) => println("worker#$id exited, pid is", $pid))
-    ->onWorkersExited(fn($pid) => println("master [PID:$pid] - all workers exited, tasks run completed"))
+    ->onCompleted(fn($pid) => println("master [PID:$pid] - all workers exited, tasks run completed"))
     ->setTasks([
         ['task1'], // one task
         ['task2'],

example/proc-tasks.php

@@ -7,6 +7,7 @@
 // run: php example/proc-tasks.php
 // ProcTasks::new() // will auto get cpu num as proc num
 ProcTasks::new(['procNum' => 3])
+    ->setProcName("procTasks")
     ->setTaskHandler(function (array $task, array $ctx) {
         $pid = $ctx['pid'];
         println("worker#{$ctx['id']} [PID:$pid] - handle task, task data", $task);
@@ -16,7 +17,7 @@
     ->onWorkersCreated(fn($pid, $info) => println("master [PID:$pid] - All task process started,", 'Workers info', $info))
     ->onWorkerStart(fn($pid, $id) => println("worker#$id started, pid is", $pid))
     ->onWorkerExit(fn($pid, $id) => println("worker#$id exited, pid is", $pid))
-    ->onWorkersExited(fn($pid) => println("master [PID:$pid] - all workers exited, tasks run completed"))
+    ->onCompleted(fn($pid) => println("master [PID:$pid] - all workers exited, tasks run completed"))
     ->setTasks([
         ['task1'], // one task
         ['task2'],

src/Cmd/AbstractCmdBuilder.php

@@ -10,6 +10,7 @@
 namespace Toolkit\Sys\Cmd;
 
 use RuntimeException;
+use Toolkit\Stdlib\Helper\Assert;
 use Toolkit\Sys\Exec;
 use function trim;
 
@@ -239,6 +240,7 @@ public function getWorkDir(): string
      */
     public function setWorkDir(string $workDir): static
     {
+        Assert::isDir($workDir, "workdir is not exists. path: $workDir");
         $this->workDir = $workDir;
         return $this;
     }

src/Proc/ProcTasks.php

@@ -2,13 +2,16 @@
 
 namespace Toolkit\Sys\Proc;
 
+use Exception;
 use JetBrains\PhpStorm\NoReturn;
+use Throwable;
 use Toolkit\Stdlib\Helper\Assert;
 use Toolkit\Stdlib\Num;
 use Toolkit\Stdlib\Obj\Traits\AutoConfigTrait;
 use Toolkit\Sys\Sys;
 use function array_chunk;
 use function count;
+use function println;
 use function time;
 
 /**
@@ -40,12 +43,14 @@ class ProcTasks
     private $taskHandler;
 
     /**
-     * hooks on error
-     * - param#1 is PID.
+     * hooks on run task error.
+     * - param#1 is Exception.
+     * - param#2 is PID.
+     * - return bool. TRUE - stop handle task, FALSE - continue handle next.
      *
-     * @var callable(int): void
+     * @var callable(Exception, int): bool
      */
-    private $onError;
+    private $errorHandler;
 
     /**
      * hooks on before create workers, in parent.
@@ -297,13 +302,38 @@ protected function doRunTasks(array $tasks, int $pid, int $id): void
         // exec task handler
         $handler = $this->taskHandler;
         foreach ($tasks as $task) {
-            $handler($task, $info);
+            try {
+                $handler($task, $info);
+            } catch (Throwable $e) {
+                if ($this->handleTaskError($e, $pid)) {
+                    break;
+                }
+            }
         }
 
         // exit worker
         exit(0);
     }
 
+    /**
+     * @param Throwable $e
+     * @param int $pid
+     *
+     * @return bool TRUE - stop continue handle. FALSE - continue next.
+     */
+    protected function handleTaskError(Throwable $e, int $pid): bool
+    {
+        // call hooks
+        if ($fn = $this->errorHandler) {
+            $stop = $fn($e, $pid);
+
+            return (bool)$stop;
+        }
+
+        println("[PID:$pid] ERROR: handle task -", $e->getMessage());
+        return false;
+    }
+
     /**
      * In parent.
      *
@@ -328,7 +358,6 @@ protected function handleWorkerExit(int $pid, int $exitCode, int $status): void
 
             $fn($pid, $info['id'], $info);
         }
-
     }
 
     /**
@@ -384,15 +413,15 @@ public function onWorkerExit(callable $listener): self
     }
 
     /**
-     * On all worker process exited, in parent.
+     * On run task error
      *
-     * @param callable $listener
+     * @param callable(Exception, int): void $errorHandler
      *
-     * @return $this
+     * @return ProcTasks
      */
-    public function onCompleted(callable $listener): self
+    public function onTaskError(callable $errorHandler): self
     {
-        $this->workersExitedFn = $listener;
+        $this->errorHandler = $errorHandler;
         return $this;
     }
 
@@ -403,7 +432,7 @@ public function onCompleted(callable $listener): self
      *
      * @return $this
      */
-    public function onWorkersExited(callable $listener): self
+    public function onCompleted(callable $listener): self
     {
         $this->workersExitedFn = $listener;
         return $this;
@@ -417,6 +446,14 @@ public function getProcNum(): int
         return $this->procNum;
     }
 
+    /**
+     * @return int
+     */
+    public function getTaskNum(): int
+    {
+        return count($this->tasks);
+    }
+
     /**
      * @param int $procNum
      *
@@ -468,21 +505,18 @@ public function setProcName(string $procName): self
     }
 
     /**
-     * @param callable $onError
-     *
-     * @return ProcTasks
+     * @return array
      */
-    public function setOnError(callable $onError): self
+    public function getProcesses(): array
     {
-        $this->onError = $onError;
-        return $this;
+        return $this->processes;
     }
 
     /**
-     * @return array
+     * @return int
      */
-    public function getProcesses(): array
+    public function getMasterPid(): int
     {
-        return $this->processes;
+        return $this->masterPid;
     }
 }