Merge pull request #2 from icicleio/no-context-interface

Remove ContextInterface & rename forks and threads

examples/fork.php

@@ -2,12 +2,12 @@
 <?php
 require dirname(__DIR__).'/vendor/autoload.php';
 
-use Icicle\Concurrent\Forking\ForkContext;
+use Icicle\Concurrent\Forking\Fork;
 use Icicle\Coroutine;
 use Icicle\Loop;
 
 Coroutine\create(function () {
-    $context = new ForkContext(function () {
+    $context = Fork::spawn(function () {
         print "Child sleeping for 4 seconds...\n";
         sleep(4);
 
@@ -19,8 +19,6 @@ Coroutine\create(function () {
         yield 42;
     });
 
-    $context->start();
-
     $timer = Loop\periodic(1, function () use ($context) {
         static $i;
         $i = $i ? ++$i : 1;

examples/thread.php

@@ -2,7 +2,7 @@
 <?php
 require dirname(__DIR__).'/vendor/autoload.php';
 
-use Icicle\Concurrent\Threading\ThreadContext;
+use Icicle\Concurrent\Threading\Thread;
 use Icicle\Coroutine;
 use Icicle\Loop;
 
@@ -13,8 +13,10 @@ $timer = Loop\periodic(1, function () {
 });
 
 Coroutine\create(function () {
+    $foo = 1;
+
     // Create a new child thread that does some blocking stuff.
-    $context = new ThreadContext(function () {
+    $context = Thread::spawn(function () use ($foo) {
         printf("\$this: %s\n", get_class($this));
 
         printf("Received the following from parent: %s\n", (yield $this->receive()));
@@ -36,9 +38,6 @@ Coroutine\create(function () {
         yield 42;
     });
 
-    // Run the thread and wait asynchronously for it to finish.
-    $context->start();
-
     yield $context->send('Start data');
 
     $lock = (yield $context->acquire());

src/ContextInterface.php

@@ -1,37 +0,0 @@
-<?php
-namespace Icicle\Concurrent;
-
-/**
- * Interface for all types of execution contexts.
- */
-interface ContextInterface extends ChannelInterface
-{
-    /**
-     * Checks if the context is running.
-     *
-     * @return bool True if the context is running, otherwise false.
-     */
-    public function isRunning();
-
-    /**
-     * Starts the context execution.
-     */
-    public function start();
-
-    /**
-     * Immediately kills the context.
-     */
-    public function kill();
-
-    /**
-     * @coroutine
-     *
-     * Gets a promise that resolves when the context ends and joins with the
-     * parent context.
-     *
-     * @return \Generator
-     *
-     * @resolve mixed Resolved with the return or resolution value of the context once it has completed execution.
-     */
-    public function join();
-}

src/Forking/Fork.php

@@ -1,7 +1,7 @@
 <?php
 namespace Icicle\Concurrent\Forking;
 
-use Icicle\Concurrent\ContextInterface;
+use Icicle\Concurrent\ChannelInterface as ContextChannelInterface;
 use Icicle\Concurrent\Exception\ForkException;
 use Icicle\Concurrent\Exception\InvalidArgumentError;
 use Icicle\Concurrent\Exception\SynchronizationError;
@@ -16,7 +16,7 @@ use Icicle\Loop;
 /**
  * Implements a UNIX-compatible context using forked processes.
  */
-class ForkContext implements ContextInterface
+class Fork implements ContextChannelInterface
 {
     /**
      * @var \Icicle\Concurrent\Sync\Channel A channel for communicating with the child.
@@ -38,6 +38,20 @@ class ForkContext implements ContextInterface
      */
     private $synchronized;
 
+    /**
+     * Spawns a new forked process and runs it.
+     *
+     * @param callable $function A callable to invoke in the process.
+     *
+     * @return Thread The process object that was spawned.
+     */
+    public static function spawn(callable $function /* , ...$args */)
+    {
+        $thread = new static($function);
+        $thread->start();
+        return $thread;
+    }
+
     public function __construct(callable $function /* , ...$args */)
     {
         $this->function = $function;
@@ -57,7 +71,9 @@ class ForkContext implements ContextInterface
     }
 
     /**
-     * {@inheritdoc}
+     * Checks if the context is running.
+     *
+     * @return bool True if the context is running, otherwise false.
      */
     public function isRunning()
     {
@@ -65,7 +81,7 @@ class ForkContext implements ContextInterface
     }
 
     /**
-     * {@inheritdoc}
+     * Starts the context execution.
      */
     public function start()
     {
@@ -161,7 +177,7 @@ class ForkContext implements ContextInterface
     }
 
     /**
-     * {@inheritdoc}
+     * Immediately kills the context.
      */
     public function kill()
     {
@@ -176,7 +192,14 @@ class ForkContext implements ContextInterface
     }
 
     /**
-     * {@inheritdoc}
+     * @coroutine
+     *
+     * Gets a promise that resolves when the context ends and joins with the
+     * parent context.
+     *
+     * @return \Generator
+     *
+     * @resolve mixed Resolved with the return or resolution value of the context once it has completed execution.
      */
     public function join()
     {

src/Threading/InternalThread.php

@@ -0,0 +1,146 @@
+<?php
+namespace Icicle\Concurrent\Threading;
+
+use Icicle\Concurrent\Sync\Channel;
+use Icicle\Concurrent\Sync\ChannelInterface;
+use Icicle\Concurrent\Sync\ExitFailure;
+use Icicle\Concurrent\Sync\ExitSuccess;
+use Icicle\Coroutine\Coroutine;
+use Icicle\Loop;
+
+/**
+ * An internal thread that executes a given function concurrently.
+ *
+ * @internal
+ */
+class InternalThread extends \Thread
+{
+    /**
+     * @var callable The function to execute in the thread.
+     */
+    private $function;
+
+    /**
+     * @var
+     */
+    private $args;
+
+    /**
+     * @var resource
+     */
+    private $socket;
+
+    /**
+     * @var bool
+     */
+    private $lock = true;
+
+    /**
+     * Creates a new thread object.
+     *
+     * @param resource $socket   IPC communication socket.
+     * @param callable $function The function to execute in the thread.
+     * @param mixed[]  $args     Arguments to pass to the function.
+     */
+    public function __construct($socket, callable $function, array $args = [])
+    {
+        $this->function = $function;
+        $this->args = $args;
+        $this->socket = $socket;
+    }
+
+    /**
+     * Runs the thread code and the initialized function.
+     */
+    public function run()
+    {
+        /* First thing we need to do is re-initialize the class autoloader. If
+         * we don't do this first, any object of a class that was loaded after
+         * the thread started will just be garbage data and unserializable
+         * values (like resources) will be lost. This happens even with
+         * thread-safe objects.
+         */
+        foreach (get_declared_classes() as $className) {
+            if (strpos($className, 'ComposerAutoloaderInit') === 0) {
+                // Calling getLoader() will register the class loader for us
+                $className::getLoader();
+                break;
+            }
+        }
+
+        // Erase the old event loop inherited from the parent thread and create
+        // a new one.
+        Loop\loop(Loop\create());
+
+        // At this point, the thread environment has been prepared so begin using the thread.
+        $channel = new Channel($this->socket);
+
+        $coroutine = new Coroutine($this->execute($channel));
+        $coroutine->done();
+
+        Loop\run();
+    }
+
+    /**
+     * Attempts to obtain the lock. Returns true if the lock was obtained.
+     *
+     * @return bool
+     */
+    public function tsl()
+    {
+        if (!$this->lock) {
+            return false;
+        }
+
+        $this->lock();
+
+        try {
+            if ($this->lock) {
+                $this->lock = false;
+                return true;
+            }
+            return false;
+        } finally {
+            $this->unlock();
+        }
+    }
+
+    /**
+     * Releases the lock.
+     */
+    public function release()
+    {
+        $this->lock = true;
+    }
+
+    /**
+     * @coroutine
+     *
+     * @param \Icicle\Concurrent\Sync\ChannelInterface $channel
+     *
+     * @return \Generator
+     *
+     * @resolve int
+     */
+    private function execute(ChannelInterface $channel)
+    {
+        $executor = new ThreadExecutor($this, $channel);
+
+        try {
+            $function = $this->function;
+            if ($function instanceof \Closure) {
+                $function = $function->bindTo($executor, ThreadExecutor::class);
+            }
+
+            $result = new ExitSuccess(yield call_user_func_array($function, $this->args));
+        } catch (\Exception $exception) {
+            $result = new ExitFailure($exception);
+        }
+
+        try {
+            yield $channel->send($result);
+        } finally {
+            $channel->close();
+        }
+    }
+}

src/Threading/Thread.php

@@ -1,146 +1,173 @@
 <?php
 namespace Icicle\Concurrent\Threading;
 
+use Icicle\Concurrent\ChannelInterface;
+use Icicle\Concurrent\Exception\InvalidArgumentError;
+use Icicle\Concurrent\Exception\SynchronizationError;
 use Icicle\Concurrent\Sync\Channel;
-use Icicle\Concurrent\Sync\ChannelInterface;
-use Icicle\Concurrent\Sync\ExitFailure;
-use Icicle\Concurrent\Sync\ExitSuccess;
-use Icicle\Coroutine\Coroutine;
-use Icicle\Loop;
+use Icicle\Concurrent\Sync\ExitStatusInterface;
+use Icicle\Concurrent\Sync\Lock;
+use Icicle\Coroutine;
 
 /**
- * An internal thread that executes a given function concurrently.
+ * Implements an execution context using native multi-threading.
  *
- * @internal
+ * The thread context is not itself threaded. A local instance of the context is
+ * maintained both in the context that creates the thread and in the thread
+ * itself.
  */
-class Thread extends \Thread
+class Thread implements ChannelInterface
 {
     /**
-     * @var callable The function to execute in the thread.
+     * @var \Icicle\Concurrent\Threading\InternalThread An internal thread instance.
      */
-    private $function;
+    private $thread;
 
     /**
-     * @var
+     * @var \Icicle\Concurrent\Sync\Channel A channel for communicating with the thread.
      */
-    private $args;
+    private $channel;
 
     /**
-     * @var resource
-     */
-    private $socket;
-
-    /**
-     * @var bool
-     */
-    private $lock = true;
-
-    /**
-     * Creates a new thread object.
+     * Spawns a new thread and runs it.
      *
-     * @param resource $socket   IPC communication socket.
-     * @param callable $function The function to execute in the thread.
-     * @param mixed[]  $args     Arguments to pass to the function.
-     */
-    public function __construct($socket, callable $function, array $args = [])
-    {
-        $this->function = $function;
-        $this->args = $args;
-        $this->socket = $socket;
-    }
-
-    /**
-     * Runs the thread code and the initialized function.
-     */
-    public function run()
-    {
-        /* First thing we need to do is re-initialize the class autoloader. If
-         * we don't do this first, any object of a class that was loaded after
-         * the thread started will just be garbage data and unserializable
-         * values (like resources) will be lost. This happens even with
-         * thread-safe objects.
-         */
-        foreach (get_declared_classes() as $className) {
-            if (strpos($className, 'ComposerAutoloaderInit') === 0) {
-                // Calling getLoader() will register the class loader for us
-                $className::getLoader();
-                break;
-            }
-        }
-
-        // Erase the old event loop inherited from the parent thread and create
-        // a new one.
-        Loop\loop(Loop\create());
-
-        // At this point, the thread environment has been prepared so begin using the thread.
-        $channel = new Channel($this->socket);
-
-        $coroutine = new Coroutine($this->execute($channel));
-        $coroutine->done();
-
-        Loop\run();
-    }
-
-    /**
-     * Attempts to obtain the lock. Returns true if the lock was obtained.
+     * @param callable $function A callable to invoke in the thread.
      *
-     * @return bool
+     * @return Thread The thread object that was spawned.
      */
-    public function tsl()
+    public static function spawn(callable $function /* , ...$args */)
     {
-        if (!$this->lock) {
-            return false;
-        }
-
-        $this->lock();
-
-        try {
-            if ($this->lock) {
-                $this->lock = false;
-                return true;
-            }
-            return false;
-        } finally {
-            $this->unlock();
-        }
+        $class  = new \ReflectionClass(__CLASS__);
+        $thread = $class->newInstanceArgs(func_get_args());
+        $thread->start();
+        return $thread;
     }
 
     /**
-     * Releases the lock.
+     * Creates a new thread context from a thread.
+     *
+     * @param callable $function
      */
-    public function release()
+    public function __construct(callable $function /* , ...$args */)
     {
-        $this->lock = true;
+        $args = array_slice(func_get_args(), 1);
+
+        list($channel, $socket) = Channel::createSocketPair();
+
+        $this->channel = new Channel($channel);
+        $this->thread = new InternalThread($socket, $function, $args);
+    }
+
+    /**
+     * Checks if the context is running.
+     *
+     * @return bool True if the context is running, otherwise false.
+     */
+    public function isRunning()
+    {
+        return $this->thread->isRunning();
+    }
+
+    /**
+     * Starts the context execution.
+     */
+    public function start()
+    {
+        if ($this->isRunning()) {
+            throw new SynchronizationError('The thread has already been started.');
+        }
+
+        $this->thread->start(PTHREADS_INHERIT_ALL);
+    }
+
+    /**
+     * Immediately kills the context.
+     */
+    public function kill()
+    {
+        $this->channel->close();
+        $this->thread->kill();
     }
 
     /**
      * @coroutine
      *
-     * @param \Icicle\Concurrent\Sync\ChannelInterface $channel
+     * Gets a promise that resolves when the context ends and joins with the
+     * parent context.
      *
      * @return \Generator
      *
-     * @resolve int
+     * @resolve mixed Resolved with the return or resolution value of the context once it has completed execution.
      */
-    private function execute(ChannelInterface $channel)
+    public function join()
     {
-        $executor = new ThreadExecutor($this, $channel);
+        if (!$this->isRunning()) {
+            throw new SynchronizationError('The thread has not been started or has already finished.');
+        }
 
         try {
-            $function = $this->function;
-            if ($function instanceof \Closure) {
-                $function = $function->bindTo($executor, ThreadExecutor::class);
+            $response = (yield $this->channel->receive());
+
+            if (!$response instanceof ExitStatusInterface) {
+                throw new SynchronizationError('Did not receive an exit status from thread.');
             }
 
-            $result = new ExitSuccess(yield call_user_func_array($function, $this->args));
-        } catch (\Exception $exception) {
-            $result = new ExitFailure($exception);
-        }
-
-        try {
-            yield $channel->send($result);
+            yield $response->getResult();
         } finally {
-            $channel->close();
+            $this->channel->close();
+            $this->thread->join();
         }
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function receive()
+    {
+        if (!$this->isRunning()) {
+            throw new SynchronizationError('The thread has not been started or has already finished.');
+        }
+
+        $data = (yield $this->channel->receive());
+
+        if ($data instanceof ExitStatusInterface) {
+            $data = $data->getResult();
+            throw new SynchronizationError(sprintf(
+                'Thread unexpectedly exited with result of type: %s',
+                is_object($data) ? get_class($data) : gettype($data)
+            ));
+        }
+
+        yield $data;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function send($data)
+    {
+        if (!$this->isRunning()) {
+            throw new SynchronizationError('The thread has not been started or has already finished.');
+        }
+
+        if ($data instanceof ExitStatusInterface) {
+            throw new InvalidArgumentError('Cannot send exit status objects.');
+        }
+
+        return $this->channel->send($data);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function acquire()
+    {
+        while (!$this->thread->tsl()) {
+            yield Coroutine\sleep(0.01);
+        }
+
+        yield new Lock(function () {
+            $this->thread->release();
+        });
+    }
 }

src/Threading/ThreadContext.php

@@ -1,164 +0,0 @@
-<?php
-namespace Icicle\Concurrent\Threading;
-
-use Icicle\Concurrent\ContextInterface;
-use Icicle\Concurrent\Exception\InvalidArgumentError;
-use Icicle\Concurrent\Exception\SynchronizationError;
-use Icicle\Concurrent\Sync\Channel;
-use Icicle\Concurrent\Sync\ExitStatusInterface;
-use Icicle\Concurrent\Sync\Lock;
-use Icicle\Coroutine;
-
-/**
- * Implements an execution context using native multi-threading.
- *
- * The thread context is not itself threaded. A local instance of the context is
- * maintained both in the context that creates the thread and in the thread
- * itself.
- */
-class ThreadContext implements ContextInterface
-{
-    /**
-     * @var \Icicle\Concurrent\Threading\Thread A thread instance.
-     */
-    private $thread;
-
-    /**
-     * @var \Icicle\Concurrent\Sync\Channel A channel for communicating with the thread.
-     */
-    private $channel;
-
-    /**
-     * Spawns a new thread and runs it.
-     *
-     * @param callable $function A callable to invoke in the thread.
-     *
-     * @return ThreadContext The thread object that was spawned.
-     */
-    public static function spawn(callable $function /* , ...$args */)
-    {
-        $class  = new \ReflectionClass(__CLASS__);
-        $thread = $class->newInstanceArgs(func_get_args());
-        $thread->start();
-        return $thread;
-    }
-
-    /**
-     * Creates a new thread context from a thread.
-     *
-     * @param callable $function
-     */
-    public function __construct(callable $function /* , ...$args */)
-    {
-        $args = array_slice(func_get_args(), 1);
-
-        list($channel, $socket) = Channel::createSocketPair();
-
-        $this->channel = new Channel($channel);
-        $this->thread = new Thread($socket, $function, $args);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function isRunning()
-    {
-        return $this->thread->isRunning();
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function start()
-    {
-        if ($this->isRunning()) {
-            throw new SynchronizationError('The thread has already been started.');
-        }
-
-        $this->thread->start(PTHREADS_INHERIT_ALL);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function kill()
-    {
-        $this->channel->close();
-        $this->thread->kill();
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function join()
-    {
-        if (!$this->isRunning()) {
-            throw new SynchronizationError('The thread has not been started or has already finished.');
-        }
-
-        try {
-            $response = (yield $this->channel->receive());
-
-            if (!$response instanceof ExitStatusInterface) {
-                throw new SynchronizationError('Did not receive an exit status from thread.');
-            }
-
-            yield $response->getResult();
-        } finally {
-            $this->channel->close();
-            $this->thread->join();
-        }
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function receive()
-    {
-        if (!$this->isRunning()) {
-            throw new SynchronizationError('The thread has not been started or has already finished.');
-        }
-
-        $data = (yield $this->channel->receive());
-
-        if ($data instanceof ExitStatusInterface) {
-            $data = $data->getResult();
-            throw new SynchronizationError(sprintf(
-                'Thread unexpectedly exited with result of type: %s',
-                is_object($data) ? get_class($data) : gettype($data)
-            ));
-        }
-
-        yield $data;
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function send($data)
-    {
-        if (!$this->isRunning()) {
-            throw new SynchronizationError('The thread has not been started or has already finished.');
-        }
-
-        if ($data instanceof ExitStatusInterface) {
-            throw new InvalidArgumentError('Cannot send exit status objects.');
-        }
-
-        return $this->channel->send($data);
-    }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function acquire()
-    {
-        while (!$this->thread->tsl()) {
-            yield Coroutine\sleep(0.01);
-        }
-
-        yield new Lock(function () {
-            $this->thread->release();
-        });
-    }
-}

src/Threading/ThreadExecutor.php

@@ -11,7 +11,7 @@ use Icicle\Coroutine;
 class ThreadExecutor implements ExecutorInterface
 {
     /**
-     * @var \Icicle\Concurrent\Threading\Thread
+     * @var \Icicle\Concurrent\Threading\InternalThread
      */
     private $thread;
 
@@ -21,10 +21,10 @@ class ThreadExecutor implements ExecutorInterface
     private $channel;
 
     /**
-     * @param \Icicle\Concurrent\Threading\Thread
+     * @param \Icicle\Concurrent\Threading\InternalThread
      * @param \Icicle\Concurrent\Sync\ChannelInterface $channel
      */
-    public function __construct(Thread $thread, ChannelInterface $channel)
+    public function __construct(InternalThread $thread, ChannelInterface $channel)
     {
         $this->thread = $thread;
         $this->channel = $channel;
@@ -63,4 +63,4 @@ class ThreadExecutor implements ExecutorInterface
             $this->thread->release();
         });
     }
-}
+}

src/Worker/WorkerThread.php

@@ -1,7 +1,7 @@
 <?php
 namespace Icicle\Concurrent\Worker;
 
-use Icicle\Concurrent\Threading\ThreadContext;
+use Icicle\Concurrent\Threading\Thread;
 use Icicle\Coroutine\Coroutine;
 
 /**
@@ -19,7 +19,7 @@ class WorkerThread implements WorkerInterface
      */
     public function __construct()
     {
-        $this->thread = new ThreadContext(function () {
+        $this->thread = new Thread(function () {
             while (true) {
                 print "Waiting for task...\n";
                 $task = (yield $this->receive());