PHPシナジー実装の詳しい説明（コード付き）

$serv = stream_socket_server("tcp://0.0.0.0:8000", $errno, $errstr) or die("Create server failed"); while(1) { $conn = stream_socket_accept($serv); if (pcntl_fork() == 0) { $request = fread($conn); // do something // $response = "hello world"; fwrite($response); fclose($conn); exit(0); } }

$serv = stream_socket_server("tcp://0.0.0.0:8000", $errno, $errstr) or die("Create server failed"); for($i = 0; $i < 32; $i++) { if (pcntl_fork() == 0) { while(1) { $conn = stream_socket_accept($serv); if ($conn == false) continue; // do something $request = fread($conn); // $response = "hello world"; fwrite($response); fclose($conn); } exit(0); } }

interface Iterator extends Traversable { // 获取当前内部标量指向的元素的数据 public mixed current() // 获取当前标量 public scalar key() // 移动到下一个标量 public void next() // 重置标量 public void rewind() // 检查当前标量是否有效 public boolean valid() }

function range ($start, $end, $step = 1){ $ret = []; for ($i = $start; $i <= $end; $i += $step) { $ret[] = $i; } return $ret; }

class Xrange implements Iterator { protected $start; protected $limit; protected $step; protected $current; public function __construct($start, $limit, $step = 1){ $this->start = $start; $this->limit = $limit; $this->step = $step; } public function rewind(){ $this->current = $this->start; } public function next(){ $this->current += $this->step; } public function current(){ return $this->current; } public function key(){ return $this->current + 1; } public function valid(){ return $this->current <= $this->limit; } }

foreach (new Xrange(0, 9) as $key => $val) { echo $key, ' ', $val, "\n"; }

0 0 1 1 2 2 3 3 4 4 5 5 6 6 7 7 8 8 9 9

var_dump(new Xrange(0, 9));

object(Xrange)#1 (4) { ["start":protected]=> int(0) ["limit":protected]=> int(9) ["step":protected]=> int(1) ["current":protected]=>NULL }

// range $startMemory = memory_get_usage(); $arr = range(0, 500000); echo 'range(): ', memory_get_usage() - $startMemory, " bytes\n"; unset($arr); // xrange $startMemory = memory_get_usage(); $arr = new Xrange(0, 500000); echo 'xrange(): ', memory_get_usage() - $startMemory, " bytes\n";

xrange(): 624 bytes range(): 72194784 bytes

$query = (new \yii\db\Query)->from('user'); // yii\db\BatchQueryResult foreach ($query->batch() as $users) { // 每次循环得到多条 user 记录 }

/** * Starts a batch query. * * A batch query supports fetching data in batches, which can keep the memory usage under a limit. * This method will return a [[BatchQueryResult]] object which implements the [[\Iterator]] interface * and can be traversed to retrieve the data in batches. * * For example, * * * $query = (new Query)->from('user'); * foreach ($query->batch() as $rows) { * // $rows is an array of 10 or fewer rows from user table * } * * * @param integer $batchSize the number of records to be fetched in each batch. * @param Connection $db the database connection. If not set, the "db" application component will be used. * @return BatchQueryResult the batch query result. It implements the [[\Iterator]] interface * and can be traversed to retrieve the data in batches. */ public function batch($batchSize = 100, $db = null){ return Yii::createObject([ 'class' => BatchQueryResult::className(), 'query' => $this, 'batchSize' => $batchSize, 'db' => $db, 'each' => false, ]); }

class BatchQueryResult extends Object implements \Iterator { public $db; public $query; public $batchSize = 100; public $each = false; private $_dataReader; private $_batch; private $_value; private $_key; /** * Destructor. */ public function __destruct() { // make sure cursor is closed $this->reset(); } /** * Resets the batch query. * This method will clean up the existing batch query so that a new batch query can be performed. */ public function reset(){ if ($this->_dataReader !== null) { $this->_dataReader->close(); } $this->_dataReader = null; $this->_batch = null; $this->_value = null; $this->_key = null; } /* * * Resets the iterator to the initial state. * This method is required by the interface [[\Iterator]]. */ public function rewind() { $this->reset(); $this->next(); } /** * Moves the internal pointer to the next dataset. * This method is required by the interface [[\Iterator]]. */ public function next() { if ($this->_batch === null || !$this->each || $this->each && next($this->_batch) === false) { $this->_batch = $this->fetchData(); reset($this->_batch); } if ($this->each) { $this->_value = current($this->_batch); if ($this->query->indexBy !== null) { $this->_key = key($this->_batch); } elseif (key($this->_batch) !== null) { $this->_key++; } else { $this->_key = null; } } else { $this->_value = $this->_batch; $this->_key = $this->_key === null ? 0 : $this->_key + 1; } } /** * Fetches the next batch of data. * @return array the data fetched */ protected function fetchData() { // ... } /** * Returns the index of the current dataset. * This method is required by the interface [[\Iterator]]. * @return integer the index of the current row. */ public function key() { return $this->_key; } /** * Returns the current dataset. * This method is required by the interface [[\Iterator]]. * @return mixed the current dataset. */ public function current() { return $this->_value; } /** * Returns whether there is a valid dataset at the current position. * This method is required by the interface [[\Iterator]]. * @return boolean whether there is a valid dataset at the current position. */ public function valid() { return !empty($this->_batch); } }

function xrange($start, $limit, $step = 1) { for ($i = 0; $i < $limit; $i += $step) { yield $i + 1 => $i; } } foreach (xrange(0, 9) as $key => $val) { printf("%d %d \n", $key, $val); } // 输出 // 1 0 // 2 1 // 3 2 // 4 3 // 5 4 // 6 5 // 7 6 // 8 7 // 9 8

function printer() { while (true) { printf("receive: %s\n", yield); } } $printer = printer(); $printer->send('hello'); $printer->send('world'); // 输出 receive: hello receive: world

function printer() { $i = 0; while (true) { printf("receive: %s\n", (yield ++$i)); } } $printer = printer(); printf("%d\n", $printer->current()); $printer->send('hello'); printf("%d\n", $printer->current()); $printer->send('world'); printf("%d\n", $printer->current()); // 输出 1 receive: hello 2 receive: world 3

function gen() { $ret = (yield 'yield1'); var_dump($ret); $ret = (yield 'yield2'); var_dump($ret); } $gen = gen(); var_dump($gen->current()); // string(6) "yield1" var_dump($gen->send('ret1')); // string(4) "ret1" (第一个 var_dump) // string(6) "yield2" (继续执行到第二个 yield，吐出了返回值) var_dump($gen->send('ret2')); // string(4) "ret2" (第二个 var_dump) // NULL (var_dump 之后没有其他语句，所以这次 ->send() 的返回值为 null)

class Task { // 任务 IDprotected $taskId; // 协程对象protected $coroutine; // send() 值protected $sendVal = null; // 是否首次 yieldprotected $beforeFirstYield = true; public function __construct($taskId, Generator $coroutine) { $this->taskId = $taskId; $this->coroutine = $coroutine; } public function getTaskId() { return $this->taskId; } public function setSendValue($sendVal) { $this->sendVal = $sendVal; } public function run() { // 如之前提到的在send之前, 当迭代器被创建后第一次 yield 之前，一个 renwind() 方法会被隐式调用 // 所以实际上发生的应该类似: // $this->coroutine->rewind(); // $this->coroutine->send(); // 这样 renwind 的执行将会导致第一个 yield 被执行, 并且忽略了他的返回值. // 真正当我们调用 yield 的时候, 我们得到的是第二个yield的值，导致第一个yield的值被忽略。 // 所以这个加上一个是否第一次 yield 的判断来避免这个问题 if ($this->beforeFirstYield) { $this->beforeFirstYield = false; return $this->coroutine->current(); } else { $retval = $this->coroutine->send($this->sendVal); $this->sendVal = null; return $retval; } } public function isFinished() { return !$this->coroutine->valid(); } }

class Scheduler { protected $maxTaskId = 0; protected $taskMap = []; // taskId => task protected $taskQueue; public function __construct() { $this->taskQueue = new SplQueue(); } // （使用下一个空闲的任务id）创建一个新任务,然后把这个任务放入任务map数组里. 接着它通过把任务放入任务队列里来实现对任务的调度. 接着run()方法扫描任务队列, 运行任务.如果一个任务结束了, 那么它将从队列里删除, 否则它将在队列的末尾再次被调度。 public function newTask(Generator $coroutine) { $tid = ++$this->maxTaskId; $task = new Task($tid, $coroutine); $this->taskMap[$tid] = $task; $this->schedule($task); return $tid; } public function schedule(Task $task) { // 任务入队 $this->queue->enqueue($task); } public function run() { while (!$this->queue->isEmpty()) { // 任务出队 $task = $this->queue->dequeue(); $task->run(); if ($task->isFinished()) { unset($this->taskMap[$task->getTaskId()]); } else { $this->schedule($task); } } } }

function task1() { for ($i = 1; $i <= 10; ++$i) { echo "This is task 1 iteration $i.\n"; yield; } } function task2() { for ($i = 1; $i <= 5; ++$i){ echo "This is task 2 iteration $i.\n"; yield; } } $scheduler = new Scheduler; $scheduler->newTask(task1()); $scheduler->newTask(task2()); $scheduler->run();

This is task 1 iteration 1. This is task 2 iteration 1. This is task 1 iteration 2. This is task 2 iteration 2. This is task 1 iteration 3. This is task 2 iteration 3. This is task 1 iteration 4. This is task 2 iteration 4. This is task 1 iteration 5. This is task 2 iteration 5. This is task 1 iteration 6. This is task 1 iteration 7. This is task 1 iteration 8. This is task 1 iteration 9. This is task 1 iteration 10.

 task protected $queue; // resourceID => [socket, tasks] protected $waitingForRead = []; protected $waitingForWrite = []; public function __construct() { // SPL 队列 $this->queue = new SplQueue(); } public function newTask(Generator $coroutine) { $tid = ++$this->maxTaskId; $task = new Task($tid, $coroutine); $this->tasks[$tid] = $task; $this->schedule($task); return $tid; } public function schedule(Task $task) { // 任务入队 $this->queue->enqueue($task); } public function run() { while (!$this->queue->isEmpty()) { // 任务出队 $task = $this->queue->dequeue(); $task->run(); if ($task->isFinished()) { unset($this->tasks[$task->getTaskId()]); } else { $this->schedule($task); } } } public function waitForRead($socket, Task $task) { if (isset($this->waitingForRead[(int)$socket])) { $this->waitingForRead[(int)$socket][1][] = $task; } else { $this->waitingForRead[(int)$socket] = [$socket, [$task]]; } } public function waitForWrite($socket, Task $task) { if (isset($this->waitingForWrite[(int)$socket])) { $this->waitingForWrite[(int)$socket][1][] = $task; } else { $this->waitingForWrite[(int)$socket] = [$socket, [$task]]; } } /** * @param $timeout 0 represent */ protected function ioPoll($timeout) { $rSocks = []; foreach ($this->waitingForRead as list($socket)) { $rSocks[] = $socket; } $wSocks = []; foreach ($this->waitingForWrite as list($socket)) { $wSocks[] = $socket; } $eSocks = []; // $timeout 为 0 时, stream_select 为立即返回，为 null 时则会阻塞的等，见 http://php.net/manual/zh/function.stream-select.php if (!@stream_select($rSocks, $wSocks, $eSocks, $timeout)) { return; } foreach ($rSocks as $socket) { list(, $tasks) = $this->waitingForRead[(int)$socket]; unset($this->waitingForRead[(int)$socket]); foreach ($tasks as $task) { $this->schedule($task); } } foreach ($wSocks as $socket) { list(, $tasks) = $this->waitingForWrite[(int)$socket]; unset($this->waitingForWrite[(int)$socket]); foreach ($tasks as $task) { $this->schedule($task); } } } /** * 检查队列是否为空，若为空则挂起的执行 stream_select，否则检查完 IO 状态立即返回，详见 ioPoll() * 作为任务加入队列后，由于 while true，会被一直重复的加入任务队列，实现每次任务前检查 IO 状态 * @return Generator object for newTask * */ protected function ioPollTask() { while (true) { if ($this->taskQueue->isEmpty()) { $this->ioPoll(null); } else { $this->ioPoll(0); } yield; } } /** * $scheduler = new Scheduler; * $scheduler->newTask(Web Server Generator); * $scheduler->withIoPoll()->run(); * * 新建 Web Server 任务后先执行 withIoPoll() 将 ioPollTask() 作为任务入队 * * @return $this */ public function withIoPoll() { $this->newTask($this->ioPollTask()); return $this; } }