fiber と stream_select でイベントループ

README.md

fiber と stream_select でイベントループ

php -S localhost:3000

php loop.php

loop.php

<?php
declare(strict_types=1);

/**
 * 超ミニマルな Fiber + stream_select ベースのイベントループ。
 * - addTask(fn() => ...) で Fiber タスクを登録
 * - awaitReadable($stream), awaitWritable($stream) でI/O待ち（Fiber::suspend）
 */
final class Loop
{
    /** @var SplQueue<Fiber> */
    private SplQueue $ready;

    /** @var array<int, array{stream: resource, fiber: Fiber}> */
    private array $readWait = [];

    /** @var array<int, array{stream: resource, fiber: Fiber}> */
    private array $writeWait = [];

    public function __construct()
    {
        $this->ready = new SplQueue();
    }

    public function addTask(callable $fn): void
    {
        $fiber = new Fiber(function () use ($fn) {
            $fn($this);
        });
        $this->schedule($fiber);
    }

    private function schedule(Fiber $fiber, mixed $value = null): void
    {
        // 再開時に値を渡したい場合は resume の引数へ
        // ここではキューに (fiber, value) を入れても良いが、最小化のため単純化
        $this->ready->enqueue([$fiber, $value]);
    }

    public function run(): void
    {
        while (!$this->ready->isEmpty() || $this->readWait || $this->writeWait) {
            // まず ready キューを回す（CPU側の進行）
            while (!$this->ready->isEmpty()) {
                [$fiber, $value] = $this->ready->dequeue();

                if ($fiber->isTerminated()) {
                    continue;
                }

                try {
                    if (!$fiber->isStarted()) {
                        $fiber->start();
                    } else {
                        $fiber->resume($value);
                    }
                } catch (Throwable $e) {
                    // タスク例外はここで握り潰さずにログ/伝播方針を決める
                    fwrite(STDERR, "Task error: {$e}\n");
                }
            }

            // ready が空なら I/O を待つ
            if ($this->ready->isEmpty()) {
                $r = [];
                foreach ($this->readWait as $w) $r[] = $w['stream'];
                $w = [];
                foreach ($this->writeWait as $ww) $w[] = $ww['stream'];
                $e = null;

                if (!$r && !$w) {
                    // 監視対象が無いのにここに来ることは基本ないが保険
                    break;
                }

                // タイムアウトは用途で調整（null=無期限）
                $n = @stream_select($r, $w, $e, null);

                if ($n === false) {
                    // EINTR 等の扱いは必要ならここで
                    continue;
                }

                // 読み可能になった stream を待っている Fiber を再開
                foreach ($r as $rs) {
                    $id = (int)$rs;
                    if (!isset($this->readWait[$id])) continue;
                    $fiber = $this->readWait[$id]['fiber'];
                    unset($this->readWait[$id]);
                    $this->schedule($fiber, $rs);
                }

                // 書き可能になった stream を待っている Fiber を再開
                foreach ($w as $ws) {
                    $id = (int)$ws;
                    if (!isset($this->writeWait[$id])) continue;
                    $fiber = $this->writeWait[$id]['fiber'];
                    unset($this->writeWait[$id]);
                    $this->schedule($fiber, $ws);
                }
            }
        }
    }

    /** 読み可能になるまで待つ（Fiber内から呼ぶ） */
    public function awaitReadable($stream): void
    {
        $id = (int)$stream;
        stream_set_blocking($stream, false);

        $this->readWait[$id] = ['stream' => $stream, 'fiber' => Fiber::getCurrent()];
        Fiber::suspend();
    }

    /** 書き可能になるまで待つ（Fiber内から呼ぶ） */
    public function awaitWritable($stream): void
    {
        $id = (int)$stream;
        stream_set_blocking($stream, false);

        $this->writeWait[$id] = ['stream' => $stream, 'fiber' => Fiber::getCurrent()];
        Fiber::suspend();
    }
}

/* ---------------------------
   デモ: 非同期っぽい TCP fetch
---------------------------- */

function async_http_get(Loop $loop, string $host, int $port, string $path): string
{
    $addr = "tcp://{$host}:{$port}";
    $errno = 0; $errstr = '';
    $stream = stream_socket_client(
        $addr,
        $errno,
        $errstr,
        5,
        STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT
    );
    if ($stream === false) {
        throw new RuntimeException("connect failed: {$errstr} ({$errno})");
    }

    // connect完了待ち（書き可能=connectが進むことが多い）
    $loop->awaitWritable($stream);

    $req =
        "GET {$path} HTTP/1.1\r\n" .
        "Host: {$host}\r\n" .
        "Connection: close\r\n" .
        "\r\n";

    $off = 0;
    $len = strlen($req);
    while ($off < $len) {
        $loop->awaitWritable($stream);
        $n = fwrite($stream, substr($req, $off));
        if ($n === false) {
            throw new RuntimeException("write failed");
        }
        $off += $n;
    }

    $buf = '';
    while (!feof($stream)) {
        $loop->awaitReadable($stream);
        $chunk = fread($stream, 8192);
        if ($chunk === false) {
            throw new RuntimeException("read failed");
        }
        if ($chunk === '') {
            // 一時的に読めないだけの可能性もあるのでループ継続
            continue;
        }
        $buf .= $chunk;
    }

    fclose($stream);
    return $buf;
}

/* ---------------------------
   実行
---------------------------- */

$loop = new Loop();

$loop->addTask(function (Loop $loop) {
    $res = async_http_get($loop, 'example.com', 3000, '/');
    echo "=== example.com response (first 300 bytes) ===\n";
    echo substr($res, 0, 300), "\n\n";
});

$loop->addTask(function (Loop $loop) {
    // “別タスク”が同時進行している雰囲気を出す
    // I/O待ちが無いので、このままだと占有する → 適宜 suspend するAPIを別途用意しても良い
    for ($i = 1; $i <= 3; $i++) {
        echo "[tick] {$i}\n";
        // 簡易的に「自発的に譲る」
        Fiber::suspend();
    }
});

$loop->run();