all与parallel

Any表示多个异步回调，任一回调完成则任务完成，All表示等待所有回调均执行完成才算任务完成，二者相同点是IO部分并发执行;

  1. <?php
  3. class All implements Async
  4. {
  5.     public $parent;
  6.     public $tasks;
  7.     public $continuation;
  9.     public $n;
  10.     public $results;
  11.     public $done;
  13.     public function __construct(array $tasks, AsyncTask $parent = null)
  14.     {
  15.         $this->tasks = $tasks;
  16.         $this->parent = $parent;
  17.         $this->n = count($tasks);
  18.         assert($this->n > 0);
  19.         $this->results = [];
  20.     }
  22.     public function begin(callable $continuation = null)
  23.     {
  24.         $this->continuation = $continuation;
  25.         foreach ($this->tasks as $id => $task) {
  26.             (new AsyncTask($task, $this->parent))->begin($this->continuation($id));
  27.         };
  28.     }
  30.     private function continuation($id)
  31.     {
  32.         return function($r, $ex = null) use($id) {
  33.             if ($this->done) {
  34.                 return;
  35.             }
  37.             // 任一回调发生异常，终止任务
  38.             if ($ex) {
  39.                 $this->done = true;
  40.                 $k = $this->continuation;
  41.                 $k(null, $ex);
  42.                 return;
  43.             }
  45.             $this->results[$id] = $r;
  46.             if (--$this->n === 0) {
  47.                 // 所有回调完成，终止任务
  48.                 $this->done = true;
  49.                 if ($this->continuation) {
  50.                     $k = $this->continuation;
  51.                     $k($this->results);
  52.                 }
  53.             }
  54.         };
  55.     }
  56. }
  59. function all(array $tasks)
  60. {
  61.     $tasks = array_map(__NAMESPACE__ . "\\await", $tasks);
  63.     return new Syscall(function(AsyncTask $parent) use($tasks) {
  64.         if (empty($tasks)) {
  65.             return null;
  66.         } else {
  67.             return new All($tasks, $parent);
  68.         }
  69.     });
  70. }
  1. <?php
  2. spawn(function() {
  3.     $ex = null;
  4.     try {
  5.         $r = (yield all([
  6.             async_dns_lookup("www.bing.com", 100),
  7.             async_dns_lookup("www.so.com", 100),
  8.             async_dns_lookup("www.baidu.com", 100),
  9.         ]));
  10.         var_dump($r);
  12. /*
  13. array(3) {
  14.   [0]=>
  15.   string(14) "202.89.233.103"
  16.   [1]=>
  17.   string(14) "125.88.193.243"
  18.   [2]=>
  19.   string(15) "115.239.211.112"
  20. }
  21. */
  22.     } catch (\Exception $ex) {
  23.         echo $ex;
  24.     }
  25. });

我们这里实现了与Promise.all相同语义的接口，或者更复杂一些，我们也可以实现批量任务以chunk方式进行作业的接口，留待读者自己完成;

至此， 我们已经拥有了 spawn callcc race all timeout.