rerun 方式改为 refork 方式

Author: ueaner

README.md

@@ -57,6 +57,9 @@ Soli Process
 
 master 进程 fork worker 进程，在 PHP 7.1+ 下 master 进程还会接收终止信号传递给 worker 进程执行退出。
 
+## 自动补充退出的进程
+
+如程序出现异常或其他情况导致进程退出，会自动补充进程，且保持对应的 $worker->id 不变。
 
 ## 函数列表
 

src/Process.php

@@ -43,26 +43,31 @@ public function start()
         $this->setProcName();
 
         for ($id = 1; $id <= $this->count; $id++) {
-            $pid = $this->fork();
+            $this->forkWorker($id);
+        }
 
-            if ($pid > 0) {
-                static::$workers[$id] = $pid;
-            } else {
-                static::$isMaster = false;
-                $this->id = $id;
-                $this->workerPid = posix_getpid();
+        $this->waitAll();
+    }
 
-                $this->setProcName();
+    protected function forkWorker($id)
+    {
+        $pid = $this->fork();
 
-                register_shutdown_function([$this, 'handleShutdown']);
+        if ($pid > 0) {
+            static::$workers[$id] = $pid;
+        } else {
+            static::$isMaster = false;
+            $this->id = $id;
+            $this->workerPid = posix_getpid();
 
-                // Run worker.
-                $this->run();
-                exit(0);
-            }
-        }
+            $this->setProcName();
 
-        $this->waitAll();
+            register_shutdown_function([$this, 'handleShutdown']);
+
+            // Run worker.
+            $this->run();
+            exit(0);
+        }
     }
 
     protected function daemonize()
@@ -123,10 +128,13 @@ protected function waitAll()
         while (($pid = pcntl_wait($status, WUNTRACED)) != -1) {
             if ($pid > 0) {
                 $id = array_search($pid, static::$workers);
-                if ($id !== false) {
-                    unset(static::$workers[$id]);
+                if ($id === false) {
+                    continue;
                 }
-                $this->log("[worker.$id $pid] exited with status $status");
+                $this->log("[worker:$id $pid] exited with status $status");
+                unset(static::$workers[$id]);
+                // refork
+                $this->forkWorker($id);
             }
         }
 
@@ -150,15 +158,14 @@ protected function run()
             // Run job.
             call_user_func($this->job, $this);
         } catch (\Throwable $e) {
-            $this->log("[worker.{$this->id} {$this->workerPid}] " . $e);
-            // rerun
-            $this->run();
+            $this->log("[worker:{$this->id} {$this->workerPid}] " . $e);
+            exit(250);
         }
     }
 
     public function handleShutdown()
     {
-        $errmsg = "[worker.{$this->id} {$this->workerPid}] process terminated";
+        $errmsg = "[worker:{$this->id} {$this->workerPid}] process terminated";
         // Handle last error.
         $error = error_get_last();
         if ($error) {
@@ -223,7 +230,7 @@ public function signalShutdownHandler($signo)
                 unset(static::$workers[$id]);
             }
         } else {
-            $this->log("[worker.{$this->id} {$this->workerPid}] " . $msg);
+            $this->log("[worker:{$this->id} {$this->workerPid}] " . $msg);
 
             // Worker process exit.
             exit(0);