amphp#64 finished

Ekstazi · Dec 14, 2018 · f17d827 · f17d827
1 parent 0e060da
commit f17d827
Show file tree

Hide file tree

Showing 14 changed files with 238 additions and 207 deletions.
diff --git a/lib/Context/Context.php b/lib/Context/Context.php
@@ -31,11 +31,4 @@ public function kill();
      * @throws \Amp\Parallel\Sync\PanicError If the context throws an uncaught exception.
      */
     public function join(): Promise;
-
-    /**
-     * Restarts the execution context.
-     * @param $force bool Whether to force restart or wait until finish first
-     * @return Promise<Context> new context instance
-     */
-    public function restart($force = false): Promise;
 }
diff --git a/lib/Context/Process.php b/lib/Context/Process.php
@@ -187,7 +187,9 @@ public function start(): Promise
             try {
                 $pid = yield $this->process->start();
 
-                $this->channel = yield $this->createChannel();
+                yield $this->process->getStdin()->write($this->hub->generateKey($pid, self::KEY_LENGTH));
+
+                $this->channel = yield $this->hub->accept($pid);
 
                 return $pid;
             } catch (\Throwable $exception) {
@@ -197,16 +199,6 @@ public function start(): Promise
         });
     }
 
-    private function createChannel(): Promise
-    {
-        return call(function () {
-            $pid = $this->process->getPid();
-            yield $this->process->getStdin()->write($this->hub->generateKey($pid, self::KEY_LENGTH));
-            return $this->hub->accept($pid);
-        });
-    }
-
-
     /**
      * {@inheritdoc}
      */
@@ -366,22 +358,4 @@ public function kill()
     {
         $this->process->kill();
     }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function restart($force = false): Promise
-    {
-        return call(function () use ($force) {
-            if ($force) {
-                $this->kill();
-            } else {
-                yield $this->join();
-            }
-            $instance = clone $this;
-            $instance->process = yield $this->process->restart($force);
-            $instance->channel = yield $instance->createChannel();
-            return $instance;
-        });
-    }
 }
diff --git a/lib/Context/Thread.php b/lib/Context/Thread.php
@@ -308,21 +308,4 @@ public function send($data): Promise
             return $result;
         });
     }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function restart($force = false): Promise
-    {
-        return call(function () use ($force) {
-            if ($force) {
-                $this->kill();
-            } else {
-                yield $this->join();
-            }
-            $instance = new static($this->function, ...$this->args);
-            yield $instance->start();
-            return $instance;
-        });
-    }
 }
diff --git a/lib/Worker/DefaultPool.php b/lib/Worker/DefaultPool.php
@@ -2,6 +2,7 @@
 
 namespace Amp\Parallel\Worker;
 
+use function Amp\call;
 use Amp\Parallel\Context\StatusError;
 use Amp\Promise;
 
@@ -69,7 +70,9 @@ public function __construct(int $maxSize = self::DEFAULT_MAX_SIZE, WorkerFactory
 
         $this->push = static function (Worker $worker) use ($workers, $idleWorkers, $busyQueue) {
             \assert($workers->contains($worker), "The provided worker was not part of this queue");
-
+			if(!$workers->contains($worker)) {
+				$workers->attach($worker, 1);
+			}
             if (($workers[$worker] -= 1) === 0) {
                 // Worker is completely idle, remove from busy queue and add to idle queue.
                 foreach ($busyQueue as $key => $busy) {
@@ -253,7 +256,18 @@ private function pull(): Worker
         return $worker;
     }
 
+	/**
+	 * {@inheritdoc}
+	 */
     public function restart($force = false): Promise
     {
+		return call(function () use($force) {
+			if($force) {
+				$this->kill();
+			} else {
+				yield $this->shutdown();
+			}
+			return new static($this->maxSize, $this->factory);
+		});
     }
 }
diff --git a/lib/Worker/Internal/PooledWorker.php b/lib/Worker/Internal/PooledWorker.php
@@ -2,6 +2,7 @@
 
 namespace Amp\Parallel\Worker\Internal;
 
+use function Amp\call;
 use Amp\Parallel\Worker\Task;
 use Amp\Parallel\Worker\Worker;
 use Amp\Promise;
@@ -75,6 +76,9 @@ public function kill()
 
     public function restart($force = false): Promise
     {
-        return $this->worker->restart($force);
+        return call(function() use($force){
+			$worker = yield $this->worker->restart($force);
+			return new static($worker, $this->push);
+		});
     }
 }
diff --git a/lib/Worker/Internal/WorkerProcess.php b/lib/Worker/Internal/WorkerProcess.php
@@ -38,8 +38,14 @@ public function start(): Promise
         return call(function () {
             $result = yield $this->process->start();
 
+            $stdout = $this->process->getStdout();
+            $stdout->unreference();
 
-            $this->redirectOutput();
+            $stderr = $this->process->getStderr();
+            $stderr->unreference();
+
+            ByteStream\pipe($stdout, ByteStream\getStdout());
+            ByteStream\pipe($stderr, ByteStream\getStderr());
 
             return $result;
         });
@@ -54,31 +60,4 @@ public function join(): Promise
     {
         return $this->process->join();
     }
-
-    public function restart($force = false): Promise
-    {
-        return call(function () use ($force) {
-            if ($force) {
-                $this->kill();
-            } else {
-                yield $this->join();
-            }
-            $instance = clone $this;
-            $instance->process = yield $this->process->restart();
-            $instance->redirectOutput();
-            return $instance;
-        });
-    }
-
-    private function redirectOutput()
-    {
-        $stdout = $this->process->getStdout();
-        $stdout->unreference();
-
-        $stderr = $this->process->getStderr();
-        $stderr->unreference();
-
-        ByteStream\pipe($stdout, ByteStream\getStdout());
-        ByteStream\pipe($stderr, ByteStream\getStderr());
-    }
 }
diff --git a/lib/Worker/TaskWorker.php b/lib/Worker/TaskWorker.php
@@ -160,8 +160,13 @@ public function restart($force = false): Promise
                 yield $this->shutdown();
             }
 
-			$context = yield $this->context->restart($force);
-			return new static($context);
+			return $this->createInstance();
         });
     }
+
+	/**
+	 * Create new instance of worker
+	 * @return Worker
+	 */
+    abstract protected function createInstance(): Worker;
 }
diff --git a/lib/Worker/WorkerProcess.php b/lib/Worker/WorkerProcess.php
@@ -8,8 +8,20 @@
 final class WorkerProcess extends TaskWorker
 {
     const SCRIPT_PATH = __DIR__ . "/Internal/worker-process.php";
+	/**
+	 * @var string
+	 */
+	private $envClassName;
+	/**
+	 * @var array|mixed[]
+	 */
+	private $env;
+	/**
+	 * @var null|string
+	 */
+	private $binary;
 
-    /**
+	/**
      * @param string $envClassName Name of class implementing \Amp\Parallel\Worker\Environment to instigate.
      *     Defaults to \Amp\Parallel\Worker\BasicEnvironment.
      * @param mixed[] $env Array of environment variables to pass to the worker. Empty array inherits from the current
@@ -20,11 +32,22 @@ final class WorkerProcess extends TaskWorker
      */
     public function __construct(string $envClassName = BasicEnvironment::class, array $env = [], string $binary = null)
     {
-        $script = [
-            self::SCRIPT_PATH,
+		$this->envClassName = $envClassName;
+		$this->env = $env;
+		$this->binary = $binary;
+
+		$script = [
+			self::SCRIPT_PATH,
             $envClassName,
-        ];
+		];
 
         parent::__construct(new Internal\WorkerProcess($script, $env, $binary));
-    }
+	}
+
+	protected function createInstance(): Worker
+	{
+		return new static($this->envClassName, $this->env, $this->binary);
+	}
+
+
 }
diff --git a/lib/Worker/WorkerThread.php b/lib/Worker/WorkerThread.php
@@ -11,29 +11,43 @@
  */
 final class WorkerThread extends TaskWorker
 {
-    /**
+	/**
+	 * @var string
+	 */
+	private $envClassName;
+
+	/**
      * @param string $envClassName Name of class implementing \Amp\Parallel\Worker\Environment to instigate.
      *     Defaults to \Amp\Parallel\Worker\BasicEnvironment.
      */
     public function __construct(string $envClassName = BasicEnvironment::class)
     {
+		$this->envClassName = $envClassName;
+
         parent::__construct(new Thread(function (Channel $channel, string $className): Promise {
-            if (!\class_exists($className)) {
-                throw new \Error(\sprintf("Invalid environment class name '%s'", $className));
-            }
+			if (!\class_exists($className)) {
+				throw new \Error(\sprintf("Invalid environment class name '%s'", $className));
+			}
 
-            if (!\is_subclass_of($className, Environment::class)) {
-                throw new \Error(\sprintf("The class '%s' does not implement '%s'", $className, Environment::class));
-            }
+			if (!\is_subclass_of($className, Environment::class)) {
+				throw new \Error(\sprintf("The class '%s' does not implement '%s'", $className, Environment::class));
+			}
 
-            $environment = new $className;
+			$environment = new $className;
 
-            if (!\defined("AMP_WORKER")) {
-                \define("AMP_WORKER", "amp-worker");
-            }
+			if (!\defined("AMP_WORKER")) {
+				\define("AMP_WORKER", "amp-worker");
+			}
 
-            $runner = new TaskRunner($channel, $environment);
-            return $runner->run();
+			$runner = new TaskRunner($channel, $environment);
+			return $runner->run();
         }, $envClassName));
-    }
+	}
+
+	protected function createInstance(): Worker
+	{
+		return new static($this->envClassName);
+	}
+
+
 }
diff --git a/test/Context/AbstractContextTest.php b/test/Context/AbstractContextTest.php
@@ -308,42 +308,4 @@ public function testExitingContextOnSend()
             yield $context->send(\str_pad("", 1024 * 1024, "-"));
         });
     }
-
-    public function testRestart()
-    {
-        $this->assertRunTimeGreaterThan(function () {
-            Loop::run(function () {
-                $context = $original = $this->createContext(function () {
-                    \usleep(100000);
-                });
-                $this->assertFalse($context->isRunning());
-                yield $context->start();
-
-                for ($i = 0; $i <= 1; $i++) {
-                    $this->assertTrue($context->isRunning());
-                    $context = yield $context->restart();
-                    $this->assertNotSame($context, $original);
-                }
-            });
-        }, 200);
-    }
-
-    public function testForceRestart()
-    {
-        $this->assertRunTimeLessThan(function () {
-            Loop::run(function () {
-                $context = $original = $this->createContext(function () {
-                    \usleep(100000);
-                });
-                $this->assertFalse($context->isRunning());
-                yield $context->start();
-
-                for ($i = 0; $i <= 1; $i++) {
-                    $this->assertTrue($context->isRunning());
-
-                    $context = yield $context->restart(true);
-                }
-            });
-        }, 200);
-    }
 }
diff --git a/test/Context/ProcessTest.php b/test/Context/ProcessTest.php
@@ -84,41 +84,4 @@ public function testParseError()
             \var_dump(yield $process->join());
         });
     }
-
-
-    public function testRestart()
-    {
-        $this->assertRunTimeGreaterThan(function () {
-            Loop::run(function () {
-                $context = $original = new Process(__DIR__ . "/wait-process.php");
-                $this->assertFalse($context->isRunning());
-                yield $context->start();
-
-                for ($i = 0; $i <= 1; $i++) {
-                    $this->assertTrue($context->isRunning());
-
-                    $context = yield $context->restart();
-
-                    $this->assertNotSame($context, $original);
-                }
-            });
-        }, 2000);
-    }
-
-    public function testForceRestart()
-    {
-        $this->assertRunTimeLessThan(function () {
-            Loop::run(function () {
-                $context = new Process(__DIR__ . "/wait-process.php");
-                $this->assertFalse($context->isRunning());
-                yield $context->start();
-
-                for ($i = 0; $i <= 1; $i++) {
-                    $this->assertTrue($context->isRunning());
-
-                    yield $context->restart(true);
-                }
-            });
-        }, 2000);
-    }
 }