FutureTask与fork

在多线程代码中我们经常会遇到这种模型，将一个耗时任务， new一个新的Thread或者通常放到线程池后台执行，当前线程执行另外任务，之后通过某个api接口阻塞获取后台任务结果。

Java童鞋应该对这个概念非常熟悉——JDK给予直接支持的Future。

同样的模型我们可以利用channel对多个协程进行同步来实现，代码很简单:

  1. <?php
  3. go(function() {
  4.     $start = microtime(true);
  6.     $ch = chan();
  8.     // 开启一个新的协程，异步执行耗时任务
  9.     spawn(function() use($ch) {
  10.         yield delay(1000);
  11.         yield $ch->send(42); // 通知+传送结果
  12.     });
  14.     yield delay(500);
  15.     $r = (yield $ch->recv()); // 阻塞等待结果
  16.     echo $r; // 42
  18.     // 我们这里两个耗时任务并发执行，总耗时约1000ms
  19.     echo "cost ", microtime(true) - $start, "\n";
  20. });

事实上我们也很容易把Future模型移植过来：

  1. <?php
  2. final class FutureTask
  3. {
  4.     const PENDING = 1;
  5.     const DONE = 2;
  6.     private $cc;
  8.     public $state;
  9.     public $result;
  10.     public $ex;
  12.     public function __construct(\Generator $gen)
  13.     {
  14.         $this->state = self::PENDING;
  16.         $asyncTask = new AsyncTask($gen);
  17.         $asyncTask->begin(function($r, $ex = null)  {
  18.             $this->state = self::DONE;
  19.             if ($cc = $this->cc) {
  20.                 // 有cc，说明有call get方法挂起协程，在此处唤醒
  21.                 $cc($r, $ex);
  22.             } else {
  23.                 // 无挂起，暂存执行结果
  24.                 $this->result = $r;
  25.                 $this->ex = $ex;
  26.             }
  27.         });
  28.     }
  30.     public function get()
  31.     {
  32.         return callcc(function($cc) use($timeout) {
  33.             if ($this->state === self::DONE) {
  34.                 // 获取结果时，任务已经完成，同步返回结果
  35.                 // 这里也可以考虑用defer实现，异步返回结果，首先先释放php栈，降低内存使用
  36.                 $cc($this->result, $this->ex);
  37.             } else {
  38.                 // 获取结果时未完成，保存$cc，挂起等待
  39.                 $this->cc = $cc;
  40.             }
  41.         });
  42.     }
  43. }
  46. // helper
  47. function fork($task, ...$args)
  48. {
  49.     $task = await($task); // 将task转换为生成器
  50.     yield new FutureTask($task);
  51. }

还是刚才那个例子, 我们改写为FutureTask版本：

  1. <?php
  2. go(function() {
  3.     $start = microtime(true);
  5.     // fork 子协程执行耗时任务
  6.     /** @var $future FutureTask */
  7.     $future = (yield fork(function() {
  8.         yield delay(1000);
  9.         yield 42;
  10.     }));
  12.     yield delay(500);
  14.     // 阻塞等待结果
  15.     $r = (yield $future->get());
  16.     echo $r; // 42
  18.     // 总耗时仍旧只有1000ms
  19.     echo "cost ", microtime(true) - $start, "\n";
  20. });

再进一步，我们扩充FutureTask的状态，阻塞获取结果加入超时选项：

  1. <?php
  3. final class FutureTask
  4. {
  5.     const PENDING = 1;
  6.     const DONE = 2;
  7.     const TIMEOUT = 3;
  9.     private $timerId;
  10.     private $cc;
  12.     public $state;
  13.     public $result;
  14.     public $ex;
  16.     // 我们这里加入新参数，用来链接futureTask与caller父任务
  17.     // 这样的好处比如可以共享父子任务上下文
  18.     public function __construct(\Generator $gen, AsyncTask $parent = null)
  19.     {
  20.         $this->state = self::PENDING;
  22.         if ($parent) {
  23.             $asyncTask = new AsyncTask($gen, $parent);
  24.         } else {
  25.             $asyncTask = new AsyncTask($gen);
  26.         }
  28.         $asyncTask->begin(function($r, $ex = null)  {
  29.             // PENDING or TIMEOUT
  30.             if ($this->state === self::TIMEOUT) {
  31.                 return;
  32.             }
  34.             // PENDING -> DONE
  35.             $this->state = self::DONE;
  37.             if ($cc = $this->cc) {
  38.                 if ($this->timerId) {
  39.                     swoole_timer_clear($this->timerId);
  40.                 }
  41.                 $cc($r, $ex);
  42.             } else {
  43.                 $this->result = $r;
  44.                 $this->ex = $ex;
  45.             }
  46.         });
  47.     }
  49.     // 这里超时时间0为永远阻塞，
  50.     // 否则超时未获取到结果，将向父任务传递超时异常
  51.     public function get($timeout = 0)
  52.     {
  53.         return callcc(function($cc) use($timeout) {
  54.             // PENDING or DONE
  55.             if ($this->state === self::DONE) {
  56.                 $cc($this->result, $this->ex);
  57.             } else {
  58.                 // 获取结果时未完成，保存$cc，开启定时器(如果需要)，挂起等待
  59.                 $this->cc = $cc;
  60.                 $this->getResultTimeout($timeout);
  61.             }
  62.         });
  63.     }
  65.     private function getResultTimeout($timeout)
  66.     {
  67.         if (!$timeout) {
  68.             return;
  69.         }
  71.         $this->timerId = swoole_timer_after($timeout, function() {
  72.             assert($this->state === self::PENDING);
  73.             $this->state = self::TIMEOUT;
  74.             $cc = $this->cc;
  75.             $cc(null, new AsyncTimeoutException());
  76.         });
  77.     }
  78. }

因为引入parentTask参数，需要将父任务隐式传参，而我们执行通过Syscall与执行当前生成器的父任务交互，所以我们重写fork辅助函数，改用Syscall实现：

  1. <?php
  3. /**
  4.  * @param $task
  5.  * @return Syscall
  6.  */
  7. function fork($task)
  8. {
  9.     $task = await($task);
  10.     return new Syscall(function(AsyncTask $parent) use($task) {
  11.         return new FutureTask($task, $parent);
  12.     });
  13. }

下面看一些关于超时的示例：

  1. <?php
  3. go(function() {
  4.     $start = microtime(true);
  6.     /** @var $future FutureTask */
  7.     $future = (yield fork(function() {
  8.         yield delay(500);
  9.         yield 42;
  10.     }));
  12.     // 阻塞等待超时，捕获到超时异常
  13.     try {
  14.         $r = (yield $future->get(100));
  15.         var_dump($r);
  16.     } catch (\Exception $ex) {
  17.         echo "get result timeout\n";
  18.     }
  20.     yield delay(1000);
  22.     // 因为我们只等待子任务100ms，我们的总耗时只有 1100ms
  23.     echo "cost ", microtime(true) - $start, "\n";
  24. });
  26. go(function() {
  27.     $start = microtime(true);
  29.     /** @var $future FutureTask */
  30.     $future = (yield fork(function() {
  31.         yield delay(500);
  32.         yield 42;
  33.         throw new \Exception();
  34.     }));
  36.     yield delay(1000);
  38.     // 子任务500ms前发生异常，已经处于完成状态
  39.     // 我们调用get会当即引发异常
  40.     try {
  41.         $r = (yield $future->get());
  42.         var_dump($r);
  43.     } catch (\Exception $ex) {
  44.         echo "something wrong in child task\n";
  45.     }
  47.     // 因为耗时任务并发执行，这里总耗时仅1000ms
  48.     echo "cost ", microtime(true) - $start, "\n";
  49. });