Use react pcntl extension to get rid of cleanup

cursedcoder · Apr 6, 2016 · 4baa105 · 4baa105
1 parent e02736c
commit 4baa105
Show file tree

Hide file tree

Showing 9 changed files with 75 additions and 25 deletions.
diff --git a/EProcess/Adapter/BaseAdapter.php b/EProcess/Adapter/BaseAdapter.php
@@ -2,6 +2,7 @@
 
 namespace EProcess\Adapter;
 
+use EProcess\Terminator;
 use React\EventLoop\LoopInterface;
 
 abstract class BaseAdapter
@@ -15,6 +16,11 @@ public function __construct(LoopInterface $loop)
         $this->node = uniqid('thread_');
     }
 
+    public function getUnixSocketFile()
+    {
+        return sprintf('%s/%s.sock', EPROCESS_SOCKET_DIR, $this->node);
+    }
+
     protected function createUnixSocket()
     {
         if (!defined('EPROCESS_SOCKET_DIR')) {
@@ -29,16 +35,7 @@ protected function createUnixSocket()
             throw new \RuntimeException(sprintf('Cannot write to "%s".', EPROCESS_SOCKET_DIR));
         }
 
-        $unixFile = sprintf('%s/%s.sock', EPROCESS_SOCKET_DIR, $this->node);
-        $unix = sprintf('unix://%s', $unixFile);
-
-        $cleanup = function () use ($unixFile) {
-            $this->loop->stop();
-            @unlink($unixFile);
-        };
-
-        register_shutdown_function($cleanup);
-        pcntl_signal(SIGINT, $cleanup);
+        $unix = sprintf('unix://%s', $this->getUnixSocketFile());
 
         return $unix;
     }

diff --git a/EProcess/Adapter/ChildProcess.php b/EProcess/Adapter/ChildProcess.php
@@ -72,7 +72,7 @@ public function create($class, array $data = [])
         $this->process = new Process(sprintf('exec %s %s', $php, realpath($file)));
         $this->process->start($this->loop);
 
-        $this->loop->addTimer(3, function() use ($file) {
+        $this->loop->addTimer(1, function() use ($file) {
             unlink($file);
         });
 

diff --git a/EProcess/Application/Application.php b/EProcess/Application/Application.php
@@ -4,10 +4,12 @@
 
 use EProcess\Behaviour\UniversalSerializer;
 use EProcess\Behaviour\Workable;
+use EProcess\Message;
+use EProcess\Messenger;
+use EProcess\Worker;
 use Evenement\EventEmitterTrait;
+use MKraemer\ReactPCNTL\PCNTL;
 use React\EventLoop\LoopInterface;
-use EProcess\Messenger;
-use EProcess\Message;
 
 abstract class Application
 {
@@ -20,6 +22,30 @@ abstract class Application
     private $loop;
     private $messenger;
     private $data;
+    private $pcntl;
+    private $workers = [];
+
+    public function addWorker(Worker $worker)
+    {
+        $this->workers[] = $worker;
+    }
+
+    public function cleanWorkers()
+    {
+        foreach ($this->workers as $worker) {
+            $worker->emit('shutdown');
+            unlink($worker->adapter()->getUnixSocketFile());
+        }
+    }
+
+    public function pcntl(PCNTL $pcntl = null)
+    {
+        if ($pcntl) {
+            $this->pcntl = $pcntl;
+        }
+
+        return $this->pcntl;
+    }
 
     public function loop(LoopInterface $loop = null)
     {
@@ -33,7 +59,7 @@ public function loop(LoopInterface $loop = null)
     public function messenger(Messenger $messenger = null)
     {
         if ($messenger) {
-            $messenger->on('message', function(Message $message) {
+            $messenger->on('message', function (Message $message) {
                 $this->emitterEmit($message->getEvent(), [$message->getContent()]);
             });
 
@@ -52,7 +78,7 @@ public function data(array $data = null)
         return $this->data;
     }
 
-    public function emit($event, $data)
+    public function emit($event, $data = '')
     {
         $this->messenger->emit($event, $data);
     }

diff --git a/EProcess/Application/ApplicationFactory.php b/EProcess/Application/ApplicationFactory.php
@@ -2,6 +2,8 @@
 
 namespace EProcess\Application;
 
+use EProcess\Worker;
+use MKraemer\ReactPCNTL\PCNTL;
 use React\EventLoop\Factory;
 
 class ApplicationFactory
@@ -29,6 +31,22 @@ public static function create($fqcn)
         $application = new $fqcn();
         $application->loop($loop);
 
+        $shutdown = function() use ($application) {
+            $application->loop()->stop();
+            $application->cleanWorkers();
+        };
+
+        $pcntl = new PCNTL($loop);
+        $pcntl->on(SIGINT, $shutdown);
+
+        $application->on('shutdown', $shutdown);
+
+        $application->pcntl($pcntl);
+
+        $application->on('worker.created', function(Worker $worker) use ($application) {
+            $application->addWorker($worker);
+        });
+
         return $application;
     }
 }
diff --git a/EProcess/Behaviour/Workable.php b/EProcess/Behaviour/Workable.php
@@ -8,7 +8,16 @@ trait Workable
 {
     public function createWorker($fqcn, array $data = [])
     {
-        return new Worker($this->loop(), $fqcn, extension_loaded('pthreads') ? 'pthreads' : 'child_process', $data);
+        $worker = new Worker(
+            $this->loop(),
+            $fqcn,
+            extension_loaded('pthreads') ? 'pthreads' : 'child_process',
+            $data
+        );
+
+        $this->emitterEmit('worker.created', [$worker]);
+
+        return $worker;
     }
 
     abstract public function loop();

diff --git a/EProcess/Messenger.php b/EProcess/Messenger.php
@@ -37,7 +37,7 @@ public function __construct($connection)
         }
     }
 
-    public function emit($event, $data)
+    public function emit($event, $data = [])
     {
         $this->connection->send((string) new Message($event, $this->serialize($data)));
     }

diff --git a/EProcess/Worker.php b/EProcess/Worker.php
@@ -38,10 +38,6 @@ public function __construct(LoopInterface $loop, $class, $adapter = null, array
         $this->messenger()->on('initialized', function() {
             $this->initialized = true;
         });
-
-        register_shutdown_function(function() {
-            $this->kill();
-        });
     }
 
     public function kill()
@@ -54,7 +50,12 @@ public function messenger()
         return $this->messenger;
     }
 
-    public function emit($event, $data)
+    public function adapter()
+    {
+        return $this->adapter;
+    }
+
+    public function emit($event, $data = [])
     {
         if ($this->initialized) {
             $this->messenger->emit($event, $data);

diff --git a/composer.json b/composer.json
@@ -17,7 +17,8 @@
     "jms/serializer": "^1.1",
     "doctrine/collections": "~1.3",
     "concerto/comms": "~0.8",
-    "symfony/process": "^3.0"
+    "symfony/process": "^3.0",
+    "mkraemer/react-pcntl": "^2.0"
   },
   "license": "MIT",
   "require-dev": {

diff --git a/examples/autoload.php b/examples/autoload.php
@@ -1,7 +1,5 @@
 <?php
 
-declare(ticks = 1);
-
 define('EPROCESS_AUTOLOAD', __FILE__);
 define('EPROCESS_SOCKET_DIR', __DIR__ . '/../tmp/');