PHP协程初体验

admin PHP 暂无评论 1,290 浏览

上次通过C扩展为PHP添加coroutine尝试失败之后，由于短期内啃下Zend可能性几乎为零，只能打语言原生能力的主意了。Google之后发现，PHP5.5引入了Generator和Coroutine新特性，于是才有了本文的诞生。

背景阅读

《当C/C++后台开发遇上Coroutine》

http://blog.csdn.net/cszhouwei/article/details/14230529

《一次失败的PHP扩展开发之旅》

http://blog.csdn.net/cszhouwei/article/details/41290673

预备知识

Generator

[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. function my_range($start, $end, $step = 1) {
  2.     for ($i = $start; $i <= $end; $i += $step) {
  3.         yield $i;
  4.     }
  5. }
  7. foreach (my_range(1, 1000) as $num) {
  8.     echo $num, "\n";
  9. }
  10. /*
  11.  * 1
  12.  * 2
  13.  * ...
  14.  * 1000
  15.  */

图 1 基于generator的range()实现

 

[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. $range = my_range(1, 1000);
  3. var_dump($range);
  4. /*
  5.  * object(Generator)#1 (0) {
  6.  * }
  7.  */
  9. var_dump($range instanceof Iterator);
  10. /*
  11.  * bool(true)
  12.  */

 

图 2 my_range()的实现推测

由于接触PHP时日尚浅，并未深入语言实现细节，所以只能根据现象进行猜测，以下是我的一些个人理解：

  • 包含yield关键字的函数比较特殊，返回值是一个Generator对象，此时函数内语句尚未真正执行
  • Generator对象是Iterator接口实例，可以通过rewind()、current()、next()、valid()系列接口进行操纵
  • Generator可以视为一种“可中断”的函数，而yield构成了一系列的“中断点”
  • Generator类似于车间生产的流水线，每次需要用产品的时候才从那里取一个，然后这个流水线就停在那里等待下一次取操作

Coroutine

细心的读者可能已经发现，截至目前，其实Generator已经实现了Coroutine的关键特性：中断执行、恢复执行。按照《当C/C++后台开发遇上Coroutine》的思路，借助“全局变量”一类语言设施进行信息传递，实现异步Server应该足够了。

其实相对于swapcontext族函数，Generator已经前进了一大步，具备了“返回数据”的能力，如果同时具备“发送数据”的能力，就再也不必通过那些蹩脚的手法绕路而行了。在PHP里面，通过Generator的send()接口（注意：不再是next()接口），可以完成“发送数据”的任务，从而实现了真正的“双向通信”。

 

[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. function gen() {
  2.     $ret = (yield 'yield1');
  3.     echo "[gen]", $ret, "\n";
  4.     $ret = (yield 'yield2');
  5.     echo "[gen]", $ret, "\n";
  6. }
  8. $gen = gen();
  9. $ret = $gen->current();
  10. echo "[main]", $ret, "\n";
  11. $ret = $gen->send("send1");
  12. echo "[main]", $ret, "\n";
  13. $ret = $gen->send("send2");
  14. echo "[main]", $ret, "\n";
  16. /*
  17.  * [main]yield1
  18.  * [gen]send1
  19.  * [main]yield2
  20.  * [gen]send2
  21.  * [main]
  22.  */

 

图 3 Coroutine双向通信示例

作为C/C++系码农，发现“可重入”、“双向通信”能力之后，貌似没有更多奢求了，不过PHP还是比较慷慨，继续添加了Exception机制，“错误处理”机制得到进一步完善。

 

[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. function gen() {
  2.     $ret = (yield 'yield1');
  3.     echo "[gen]", $ret, "\n";
  4.     try {
  5.         $ret = (yield 'yield2');
  6.         echo "[gen]", $ret, "\n";
  7.     } catch (Exception $ex) {
  8.         echo "[gen][Exception]", $ex->getMessage(), "\n";
  9.     }
  10.     echo "[gen]finish\n";
  11. }
  13. $gen = gen();
  14. $ret = $gen->current();
  15. echo "[main]", $ret, "\n";
  16. $ret = $gen->send("send1");
  17. echo "[main]", $ret, "\n";
  18. $ret = $gen->throw(new Exception("Test"));
  19. echo "[main]", $ret, "\n";
  21. /*
  22.  * [main]yield1
  23.  * [gen]send1
  24.  * [main]yield2
  25.  * [gen][Exception]Test
  26.  * [gen]finish
  27.  * [main]
  28.  */

 

图 4 Coroutine错误处理示例

实战演习

前面简单介绍了相关的语言设施，那么具体到实际项目中，到底应该如何运用呢？让我们继续《一次失败的PHP扩展开发之旅》描述的场景，借助上述特性实现那个美好的愿望：以同步方式书写异步代码！

第一版初稿

 

[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. <?php
  3. class AsyncServer {
  4.     protected $handler;
  5.     protected $socket;
  6.     protected $tasks = [];
  8.     public function __construct($handler) {
  9.         $this->handler = $handler;
  11.         $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
  12.         if(!$this->socket) {
  13.             die(socket_strerror(socket_last_error())."\n");
  14.         }
  15.         if (!socket_set_nonblock($this->socket)) {
  16.             die(socket_strerror(socket_last_error())."\n");
  17.         }
  18.         if(!socket_bind($this->socket, "0.0.0.0", 1234)) {
  19.             die(socket_strerror(socket_last_error())."\n");
  20.         }
  21.     }
  23.     public function Run() {
  24.         while (true) {
  25.             $reads = array($this->socket);
  26.             foreach ($this->tasks as list($socket)) {
  27.                 $reads[] = $socket;
  28.             }
  29.             $writes = NULL;
  30.             $excepts= NULL;
  31.             if (!socket_select($reads, $writes, $excepts, 0, 1000)) {
  32.                 continue;
  33.             }
  35.             foreach ($reads as $one) {
  36.                 $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);
  37.                 if (!$len) {
  38.                     //echo "socket_recvfrom fail.\n";
  39.                     continue;
  40.                 }
  41.                 if ($one == $this->socket) {
  42.                     //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n";
  43.                     $handler = $this->handler;
  44.                     $coroutine = $handler($one, $data, $len, $ip, $port);
  45.                     $task = $coroutine->current();
  46.                     //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n";
  47.                     $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
  48.                     if(!$socket) {
  49.                         //echo socket_strerror(socket_last_error())."\n";
  50.                         $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
  51.                         continue;
  52.                     }
  53.                     if (!socket_set_nonblock($socket)) {
  54.                         //echo socket_strerror(socket_last_error())."\n";
  55.                         $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
  56.                         continue;
  57.                     }
  58.                     socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);
  59.                     $this->tasks[$socket] = [$socket, $coroutine];
  60.                 } else {
  61.                     //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port\n";
  62.                     if (!isset($this->tasks[$one])) {
  63.                         //echo "no async_task found.\n";
  64.                     } else {
  65.                         list($socket, $coroutine) = $this->tasks[$one];
  66.                         unset($this->tasks[$one]);
  67.                         socket_close($socket);
  68.                         $coroutine->send(array($data, $len));
  69.                     }
  70.                 }
  71.             }
  72.         }
  73.     }
  74. }
  76. class AsyncTask {
  77.     public $data;
  78.     public $len;
  79.     public $ip;
  80.     public $port;
  81.     public $timeout;
  83.     public function __construct($data, $len, $ip, $port, $timeout) {
  84.         $this->data = $data;
  85.         $this->len = $len;
  86.         $this->ip = $ip;
  87.         $this->port = $port;
  88.         $this->timeout = $timeout;
  89.     }
  90. }
  92. function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {
  93.     //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf\n";
  94.     list($rsp_buf, $rsp_len) = (yield new AsyncTask($req_buf, $req_len, "127.0.0.1", 2345, 1000));
  95.     //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n";
  96.     socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);
  97. }
  99. $server = new AsyncServer(RequestHandler);
  100. $server->Run();
  102. ?>

 

代码解读：

  • 为了便于说明问题，这里所有底层通讯基于UDP，省略了TCP的connect等繁琐细节
  • AsyncServer为底层框架类，封装了网络通讯细节以及协程切换细节，通过socket进行coroutine绑定
  • RequestHandler为业务处理函数，通过yield new AsyncTask()实现异步网络交互

第二版完善

第一版遗留问题：

  • 异步网络交互的timeout未实现，仅预留了接口参数
  • yield new AsyncTask()调用方式不够自然，略感别扭
[php][/php] view plaincopy在CODE上查看代码片派生到我的代码片

  1. <?php
  3. class AsyncServer {
  4.     protected $handler;
  5.     protected $socket;
  6.     protected $tasks = [];
  7.     protected $timers = [];
  9.     public function __construct(callable $handler) {
  10.         $this->handler = $handler;
  12.         $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
  13.         if(!$this->socket) {
  14.             die(socket_strerror(socket_last_error())."\n");
  15.         }
  16.         if (!socket_set_nonblock($this->socket)) {
  17.             die(socket_strerror(socket_last_error())."\n");
  18.         }
  19.         if(!socket_bind($this->socket, "0.0.0.0", 1234)) {
  20.             die(socket_strerror(socket_last_error())."\n");
  21.         }
  22.     }
  24.     public function Run() {
  25.         while (true) {
  26.             $now = microtime(true) * 1000;
  27.             foreach ($this->timers as $time => $sockets) {
  28.                 if ($time > $now) break;
  29.                 foreach ($sockets as $one) {
  30.                     list($socket, $coroutine) = $this->tasks[$one];
  31.                     unset($this->tasks[$one]);
  32.                     socket_close($socket);
  33.                     $coroutine->throw(new Exception("Timeout"));
  34.                 }
  35.                 unset($this->timers[$time]);
  36.             }
  38.             $reads = array($this->socket);
  39.             foreach ($this->tasks as list($socket)) {
  40.                 $reads[] = $socket;
  41.             }
  42.             $writes = NULL;
  43.             $excepts= NULL;
  44.             if (!socket_select($reads, $writes, $excepts, 0, 1000)) {
  45.                 continue;
  46.             }
  48.             foreach ($reads as $one) {
  49.                 $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port);
  50.                 if (!$len) {
  51.                     //echo "socket_recvfrom fail.\n";
  52.                     continue;
  53.                 }
  54.                 if ($one == $this->socket) {
  55.                     //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n";
  56.                     $handler = $this->handler;
  57.                     $coroutine = $handler($one, $data, $len, $ip, $port);
  58.                     if (!$coroutine) {
  59.                         //echo "[Run]everything is done.\n";
  60.                         continue;
  61.                     }
  62.                     $task = $coroutine->current();
  63.                     //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n";
  64.                     $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
  65.                     if(!$socket) {
  66.                         //echo socket_strerror(socket_last_error())."\n";
  67.                         $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
  68.                         continue;
  69.                     }
  70.                     if (!socket_set_nonblock($socket)) {
  71.                         //echo socket_strerror(socket_last_error())."\n";
  72.                         $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error()));
  73.                         continue;
  74.                     }
  75.                     socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port);
  76.                     $deadline = $now + $task->timeout;
  77.                     $this->tasks[$socket] = [$socket, $coroutine, $deadline];
  78.                     $this->timers[$deadline][$socket] = $socket;
  79.                 } else {
  80.                     //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port\n";
  81.                     list($socket, $coroutine, $deadline) = $this->tasks[$one];
  82.                     unset($this->tasks[$one]);
  83.                     unset($this->timers[$deadline][$one]);
  84.                     socket_close($socket);
  85.                     $coroutine->send(array($data, $len));
  86.                 }
  87.             }
  88.         }
  89.     }
  90. }
  92. class AsyncTask {
  93.     public $data;
  94.     public $len;
  95.     public $ip;
  96.     public $port;
  97.     public $timeout;
  99.     public function __construct($data, $len, $ip, $port, $timeout) {
  100.         $this->data = $data;
  101.         $this->len = $len;
  102.         $this->ip = $ip;
  103.         $this->port = $port;
  104.         $this->timeout = $timeout;
  105.     }
  106. }
  108. function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) {
  109.     return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout);
  110. }
  112. function RequestHandler($socket, $req_buf, $req_len, $ip, $port) {
  113.     //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf\n";
  114.     try {
  115.         list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000));
  116.     } catch (Exception $ex) {
  117.         $rsp_buf = $ex->getMessage();
  118.         $rsp_len = strlen($rsp_buf);
  119.         //echo "[Exception]$rsp_buf\n";
  120.     }
  121.     //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n";
  122.     socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);
  123. }
  125. $server = new AsyncServer(RequestHandler);
  126. $server->Run();
  128. ?>

代码解读：

  • 借助PHP内置array能力，实现简单的“超时管理”，以毫秒为精度作为时间分片
  • 封装AsyncSendRecv接口，调用形如yield AsyncSendRecv()，更加自然
  • 添加Exception作为错误处理机制，添加ret_code亦可，仅为展示之用

性能测试

测试环境

测试数据

100Byte/REQ 1000Byte/REQ
async_svr_v1.php 16000/s 15000/s
async_svr_v2.php 11000/s 10000/s

展望未来

  • 有兴趣的PHPer可以基于该思路进行底层框架封装，对于常见阻塞操作进行封装，比如：connect、send、recv、sleep ...
  • 本人接触PHP时日尚浅，很多用法非最优，高手可有针对性优化，性能应该可以继续提高
  • 目前基于socket进行coroutine绑定，如果基于TCP通信，每次connect/close，开销过大，需要考虑实现连接池
  • python等语言也有类似的语言设施，有兴趣的读者可以自行研究

相关博文

PHP协程初体验
赏
标签: