PHP协程初体验
By warezhou 2014.11.24
上次通过C扩展为PHP添加coroutine尝试失败之后，由于短期内啃下Zend可能性几乎为零，只能打语言原生能力的主意了。Google之后发现，PHP5.5引入了Generator和Coroutine新特性，于是才有了本文的诞生。
背景阅读《当C/C++后台开发遇上Coroutine》
http://km.oa.com/group/906/articles/show/165396
《一次失败的PHP扩展开发之旅》
http://km.oa.com/group/906/articles/show/208269
预备知识
function my_range($start, $end, $step = 1) { for ($i = $start; $i <p> 图 1 基于generator的range()实现</p> <p class="sycode"> </p><pre style="代码" class="precsshei">$range = my_range(1, 1000);var_dump($range);/* * object(Generator)#1 (0) { * } */var_dump($range instanceof Iterator);/* * bool(true) */图 2 my_range()的实现推测
由于接触PHP时日尚浅，并未深入语言实现细节，所以只能根据现象进行猜测，以下是我的一些个人理解：
细心的读者可能已经发现，截至目前，其实Generator已经实现了Coroutine的关键特性：中断执行、恢复执行。按照《当C/C++后台开发遇上Coroutine》的思路，借助“全局变量”一类语言设施进行信息传递，实现异步Server应该足够了。
其实相对于swapcontext族函数，Generator已经前进了一大步，具备了“返回数据”的能力，如果同时具备“发送数据”的能力，就再也不必通过那些蹩脚的手法绕路而行了。在PHP里面，通过Generator的send()接口（注意：不再是next()接口），可以完成“发送数据”的任务，从而实现了真正的“双向通信”。
function gen() { $ret = (yield 'yield1'); echo "[gen]", $ret, "\n"; $ret = (yield 'yield2'); echo "[gen]", $ret, "\n";}$gen = gen();$ret = $gen->current();echo "[main]", $ret, "\n";$ret = $gen->send("send1");echo "[main]", $ret, "\n";$ret = $gen->send("send2");echo "[main]", $ret, "\n";/* * [main]yield1 * [gen]send1 * [main]yield2 * [gen]send2 * [main] */图 3 Coroutine双向通信示例
作为C/C++系码农，发现“可重入”、“双向通信”能力之后，貌似没有更多奢求了，不过PHP还是比较慷慨，继续添加了Exception机制，“错误处理”机制得到进一步完善。
function gen() { $ret = (yield 'yield1'); echo "[gen]", $ret, "\n"; try { $ret = (yield 'yield2'); echo "[gen]", $ret, "\n"; } catch (Exception $ex) { echo "[gen][Exception]", $ex->getMessage(), "\n"; } echo "[gen]finish\n";}$gen = gen();$ret = $gen->current();echo "[main]", $ret, "\n";$ret = $gen->send("send1");echo "[main]", $ret, "\n";$ret = $gen->throw(new Exception("Test"));echo "[main]", $ret, "\n";/* * [main]yield1 * [gen]send1 * [main]yield2 * [gen][Exception]Test * [gen]finish * [main] */图 4 Coroutine错误处理示例
实战演习前面简单介绍了相关的语言设施，那么具体到实际项目中，到底应该如何运用呢？让我们继续《一次失败的PHP扩展开发之旅》描述的场景，借助上述特性实现那个美好的愿望：以同步方式书写异步代码！
<?phpclass AsyncServer { protected $handler; protected $socket; protected $tasks = []; public function __construct($handler) { $this->handler = $handler; $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$this->socket) { die(socket_strerror(socket_last_error())."\n"); } if (!socket_set_nonblock($this->socket)) { die(socket_strerror(socket_last_error())."\n"); } if(!socket_bind($this->socket, "0.0.0.0", 1234)) { die(socket_strerror(socket_last_error())."\n"); } } public function Run() { while (true) { $reads = array($this->socket); foreach ($this->tasks as list($socket)) { $reads[] = $socket; } $writes = NULL; $excepts= NULL; if (!socket_select($reads, $writes, $excepts, 0, 1000)) { continue; } foreach ($reads as $one) { $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port); if (!$len) { //echo "socket_recvfrom fail.\n"; continue; } if ($one == $this->socket) { //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n"; $handler = $this->handler; $coroutine = $handler($one, $data, $len, $ip, $port); $task = $coroutine->current(); //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n"; $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$socket) { //echo socket_strerror(socket_last_error())."\n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } if (!socket_set_nonblock($socket)) { //echo socket_strerror(socket_last_error())."\n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port); $this->tasks[$socket] = [$socket, $coroutine]; } else { //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port\n"; if (!isset($this->tasks[$one])) { //echo "no async_task found.\n"; } else { list($socket, $coroutine) = $this->tasks[$one]; unset($this->tasks[$one]); socket_close($socket); $coroutine->send(array($data, $len)); } } } } }}class AsyncTask { public $data; public $len; public $ip; public $port; public $timeout; public function __construct($data, $len, $ip, $port, $timeout) { $this->data = $data; $this->len = $len; $this->ip = $ip; $this->port = $port; $this->timeout = $timeout; }}function RequestHandler($socket, $req_buf, $req_len, $ip, $port) { //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf\n"; list($rsp_buf, $rsp_len) = (yield new AsyncTask($req_buf, $req_len, "127.0.0.1", 2345, 1000)); //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n"; socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);}$server = new AsyncServer(RequestHandler);$server->Run();?>代码解读：
第一版遗留问题：
<?phpclass AsyncServer { protected $handler; protected $socket; protected $tasks = []; protected $timers = []; public function __construct(callable $handler) { $this->handler = $handler; $this->socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$this->socket) { die(socket_strerror(socket_last_error())."\n"); } if (!socket_set_nonblock($this->socket)) { die(socket_strerror(socket_last_error())."\n"); } if(!socket_bind($this->socket, "0.0.0.0", 1234)) { die(socket_strerror(socket_last_error())."\n"); } } public function Run() { while (true) { $now = microtime(true) * 1000; foreach ($this->timers as $time => $sockets) { if ($time > $now) break; foreach ($sockets as $one) { list($socket, $coroutine) = $this->tasks[$one]; unset($this->tasks[$one]); socket_close($socket); $coroutine->throw(new Exception("Timeout")); } unset($this->timers[$time]); } $reads = array($this->socket); foreach ($this->tasks as list($socket)) { $reads[] = $socket; } $writes = NULL; $excepts= NULL; if (!socket_select($reads, $writes, $excepts, 0, 1000)) { continue; } foreach ($reads as $one) { $len = socket_recvfrom($one, $data, 65535, 0, $ip, $port); if (!$len) { //echo "socket_recvfrom fail.\n"; continue; } if ($one == $this->socket) { //echo "[Run]request recvfrom succ. data=$data ip=$ip port=$port\n"; $handler = $this->handler; $coroutine = $handler($one, $data, $len, $ip, $port); if (!$coroutine) { //echo "[Run]everything is done.\n"; continue; } $task = $coroutine->current(); //echo "[Run]AsyncTask recv. data=$task->data ip=$task->ip port=$task->port timeout=$task->timeout\n"; $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP); if(!$socket) { //echo socket_strerror(socket_last_error())."\n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } if (!socket_set_nonblock($socket)) { //echo socket_strerror(socket_last_error())."\n"; $coroutine->throw(new Exception(socket_strerror(socket_last_error()), socket_last_error())); continue; } socket_sendto($socket, $task->data, $task->len, 0, $task->ip, $task->port); $deadline = $now + $task->timeout; $this->tasks[$socket] = [$socket, $coroutine, $deadline]; $this->timers[$deadline][$socket] = $socket; } else { //echo "[Run]response recvfrom succ. data=$data ip=$ip port=$port\n"; list($socket, $coroutine, $deadline) = $this->tasks[$one]; unset($this->tasks[$one]); unset($this->timers[$deadline][$one]); socket_close($socket); $coroutine->send(array($data, $len)); } } } }}class AsyncTask { public $data; public $len; public $ip; public $port; public $timeout; public function __construct($data, $len, $ip, $port, $timeout) { $this->data = $data; $this->len = $len; $this->ip = $ip; $this->port = $port; $this->timeout = $timeout; }}function AsyncSendRecv($req_buf, $req_len, $ip, $port, $timeout) { return new AsyncTask($req_buf, $req_len, $ip, $port, $timeout);}function RequestHandler($socket, $req_buf, $req_len, $ip, $port) { //echo "[RequestHandler] before yield AsyncTask. REQ=$req_buf\n"; try { list($rsp_buf, $rsp_len) = (yield AsyncSendRecv($req_buf, $req_len, "127.0.0.1", 2345, 3000)); } catch (Exception $ex) { $rsp_buf = $ex->getMessage(); $rsp_len = strlen($rsp_buf); //echo "[Exception]$rsp_buf\n"; } //echo "[RequestHandler] after yield AsyncTask. RSP=$rsp_buf\n"; socket_sendto($socket, $rsp_buf, $rsp_len, 0, $ip, $port);}$server = new AsyncServer(RequestHandler);$server->Run();?>代码解读：
| 100Byte/REQ | 1000Byte/REQ | |
| async_svr_v1.php | 16000/s | 15000/s |
| async_svr_v2.php | 11000/s | 10000/s |
