PHP 生成器、Fiber 与协程调度：从 yield 到协作式并发

final class Countdown implements Iterator
{
    private int $current;
    public function __construct(private int $start) { $this->current = $start; }
    public function current(): mixed { return $this->current; }
    public function key(): mixed { return $this->start - $this->current; }
    public function next(): void { $this->current--; }
    public function rewind(): void { $this->current = $this->start; }
    public function valid(): bool { return $this->current > 0; }
}

function countdown(int $start): Generator
{
    for ($i = $start; $i > 0; $i--) { yield $i; }
}

function demo(): Generator
{
    yield 'a';              // key 自动从 0 递增
    yield 'x' => 'b';       // 指定 key
    yield;                  // 产出 null
    yield from [1, 2, 3];   // 委托给另一个可迭代对象
}

$gen = (function (): Generator {
    while (true) {
        $line = yield;              // 挂起，等待外部 send
        echo "收到: {$line}\n";
    }
})();
$gen->current();                    // 启动到第一个 yield
$gen->send('hello');                // 恢复，$line = 'hello'

$nums = range(1, 10_000_000);       // 数组：约 300+ MB 一次性进内存

function nums(int $max): Generator
{
    for ($i = 1; $i <= $max; $i++) { yield $i; }   // 生成器：一次只产一个
}

function fibonacci(): Generator
{
    [$a, $b] = [0, 1];
    while (true) { yield $a; [$a, $b] = [$b, $a + $b]; }
}

function accumulator(): Generator
{
    $total = 0;
    while (true) {
        $x = yield $total;          // 产出 total，同时接收下一个 x
        if ($x === null) break;
        $total += $x;
    }
    return $total;
}
$acc = accumulator();
$acc->current();                    // 启动 → 产出 0
$acc->send(10);                     // 产出 10
$acc->send(5);                      // 产出 15

$g = (function (): Generator {
    try { yield 1; } catch (RuntimeException $e) { echo "被注入: {$e->getMessage()}"; }
})();
$g->current();
$g->throw(new RuntimeException('boom'));   // 在暂停点抛出

function readLines(string $path): Generator
{
    $fh = fopen($path, 'rb');
    try {
        while (($line = fgets($fh)) !== false) {
            yield rtrim($line, "\r\n");
        }
    } finally {
        fclose($fh);                 // 生成器被提前 break 时也会执行
    }
}

foreach (readLines('/var/log/huge.log') as $line) {
    if (str_contains($line, 'ERROR')) { echo $line, PHP_EOL; }
}

foreach (User::where('active', 1)->cursor() as $user) {
    $user->syncToCrm();              // 百万行也不爆内存
}

$fiber = new Fiber(function (string $name): string {
    echo "开始 {$name}\n";
    $paused = Fiber::suspend('暂停中');   // 挂起，把值交给 resume 方
    echo "恢复，收到 {$paused}\n";
    return '完成';
});

$ret = $fiber->start('A');        // 打印「开始 A」，返回 '暂停中'
$fiber->resume('go');             // 打印「恢复，收到 go」
echo $fiber->getReturn();         // 完成

final class Scheduler
{
    private SplQueue $tasks;
    public function __construct() { $this->tasks = new SplQueue(); }

    public function run(): void
    {
        while (!$this->tasks->isEmpty()) {
            [$fiber, $value] = $this->tasks->dequeue();
            $ret = $fiber->isStarted() ? $fiber->resume($value) : $fiber->start($value);
            if ($ret !== null) { $this->tasks->enqueue([$fiber, null]); }
        }
    }
}

$fiber = new Fiber(function () {
    $body = Fiber::suspend();          // 等 HTTP 响应
    echo strlen($body);
});
$http->get('https://example.com', fn ($body) => $fiber->resume($body));
$fiber->start();

final class RoundRobin
{
    /** @var Fiber[] */
    private array $fibers = [];

    public function spawn(callable $task): void { $this->fibers[] = new Fiber($task); }

    public function run(): void
    {
        while ($this->fibers) {
            foreach ($this->fibers as $i => $fiber) {
                if (!$fiber->isStarted())          { $fiber->start(); }
                elseif (!$fiber->isTerminated())   { $fiber->resume(); }
                if ($fiber->isTerminated())        { unset($this->fibers[$i]); }
            }
            $this->fibers = array_values($this->fibers);
        }
    }
}

$s = new RoundRobin();
$s->spawn(function () { for ($i = 0; $i < 3; $i++) { echo "A{$i} "; Fiber::suspend(); } });
$s->spawn(function () { for ($i = 0; $i < 3; $i++) { echo "B{$i} "; Fiber::suspend(); } });
$s->run();   // A0 B0 A1 B1 A2 B2