stream_select でイベントループ

README.md

stream_select でイベントループ

php eventloop.php

nc 127.0.0.1 9001

eventloop.php

<?php
/**
 * Minimal event loop based on stream_select().
 *
 * Features:
 * - Read watchers (onReadable)
 * - Write watchers (onWritable)
 * - One-shot timers (addTimer)
 * - Periodic timers (addPeriodicTimer)
 * - cancel($id), stop()
 *
 * Notes:
 * - Uses stream_select() => select(2) semantics (FD limit / O(n) scan).
 * - Works best for CLI long-running processes.
 */
final class SelectLoop
{
    /** @var array<int, array{stream: resource, cb: callable}> */
    private array $read = [];

    /** @var array<int, array{stream: resource, cb: callable}> */
    private array $write = [];

    /**
     * @var array<int, array{
     *   at: float, interval: float, periodic: bool, cb: callable
     * }>
     */
    private array $timers = [];

    private int $nextId = 1;
    private bool $running = false;

    // ---------- Watchers ----------

    /**
     * Register a callback invoked when $stream becomes readable.
     * @param resource $stream
     */
    public function onReadable($stream, callable $cb): int
    {
        $this->assertStream($stream);
        $id = $this->nextId++;
        stream_set_blocking($stream, false);
        $this->read[$id] = ['stream' => $stream, 'cb' => $cb];
        return $id;
    }

    /**
     * Register a callback invoked when $stream becomes writable.
     * @param resource $stream
     */
    public function onWritable($stream, callable $cb): int
    {
        $this->assertStream($stream);
        $id = $this->nextId++;
        stream_set_blocking($stream, false);
        $this->write[$id] = ['stream' => $stream, 'cb' => $cb];
        return $id;
    }

    // ---------- Timers ----------

    /** One-shot timer */
    public function addTimer(float $afterSeconds, callable $cb): int
    {
        return $this->addTimerInternal($afterSeconds, 0.0, false, $cb);
    }

    /** Periodic timer */
    public function addPeriodicTimer(float $intervalSeconds, callable $cb): int
    {
        return $this->addTimerInternal($intervalSeconds, $intervalSeconds, true, $cb);
    }

    private function addTimerInternal(float $after, float $interval, bool $periodic, callable $cb): int
    {
        if ($after < 0) {
            throw new InvalidArgumentException('Timer delay must be >= 0');
        }
        $id = $this->nextId++;
        $now = microtime(true);
        $this->timers[$id] = [
            'at' => $now + $after,
            'interval' => $interval,
            'periodic' => $periodic,
            'cb' => $cb,
        ];
        return $id;
    }

    // ---------- Control ----------

    public function cancel(int $id): void
    {
        unset($this->read[$id], $this->write[$id], $this->timers[$id]);
    }

    public function stop(): void
    {
        $this->running = false;
    }

    public function run(): void
    {
        $this->running = true;

        while ($this->running) {
            // 1) Run due timers first (fairness: timers aren't starved)
            $this->runDueTimers();

            // 2) Build read/write arrays for stream_select()
            $r = [];
            foreach ($this->read as $w) {
                $r[] = $w['stream'];
            }

            $wArr = [];
            foreach ($this->write as $w) {
                $wArr[] = $w['stream'];
            }

            $e = null; // we don't use except set

            // 3) Compute timeout until next timer (or block)
            [$sec, $usec] = $this->computeSelectTimeout();

            // If no watchers and no timers -> nothing to do; stop.
            if (empty($r) && empty($wArr) && empty($this->timers)) {
                $this->running = false;
                break;
            }

            // 4) Wait for I/O readiness (or timeout for timers)
            // stream_select requires arrays passed by reference and will modify them.
            $num = @stream_select($r, $wArr, $e, $sec, $usec);

            // On error (e.g., interrupted by signal), just continue.
            if ($num === false) {
                // Optionally: check error_get_last()
                continue;
            }

            // 5) Dispatch readable callbacks
            if ($num > 0) {
                if (!empty($r)) {
                    $this->dispatchReady($r, $this->read);
                }
                if (!empty($wArr)) {
                    $this->dispatchReady($wArr, $this->write);
                }
            }
        }
    }

    // ---------- Internals ----------

    private function assertStream($stream): void
    {
        if (!is_resource($stream)) {
            throw new InvalidArgumentException('Expected stream resource');
        }
    }

    private function dispatchReady(array $readyStreams, array $watchers): void
    {
        // We need to call callbacks for matching watcher streams.
        // Minimal O(n*m) approach; for performance, index by (int)$stream.
        foreach ($watchers as $id => $w) {
            $s = $w['stream'];
            foreach ($readyStreams as $ready) {
                if ($ready === $s) {
                    ($w['cb'])($s, $id, $this);
                    break;
                }
            }
        }
    }

    private function runDueTimers(): void
    {
        if (empty($this->timers)) {
            return;
        }

        $now = microtime(true);

        // Collect due timers first to avoid mutation issues during callbacks.
        $due = [];
        foreach ($this->timers as $id => $t) {
            if ($t['at'] <= $now) {
                $due[$id] = $t;
            }
        }

        foreach ($due as $id => $t) {
            // Timer might have been cancelled by another callback already
            if (!isset($this->timers[$id])) {
                continue;
            }

            // Execute callback
            ($t['cb'])($id, $this);

            // Reschedule periodic timers
            if ($t['periodic'] && isset($this->timers[$id])) {
                $now2 = microtime(true);
                $nextAt = $now2 + $t['interval'];
                $this->timers[$id]['at'] = $nextAt;
            } else {
                unset($this->timers[$id]);
            }
        }
    }

    /**
     * Return [sec, usec] for stream_select timeout.
     * - If timers exist, timeout = time until earliest timer (>=0).
     * - If no timers, block "forever": sec = null (but stream_select needs int|null).
     *   We pass null to block indefinitely when there is at least one watcher.
     */
    private function computeSelectTimeout(): array
    {
        if (empty($this->timers)) {
            // No timers => block indefinitely (until I/O) if watchers exist.
            return [null, null];
        }

        $now = microtime(true);
        $nextAt = null;

        foreach ($this->timers as $t) {
            if ($nextAt === null || $t['at'] < $nextAt) {
                $nextAt = $t['at'];
            }
        }

        $delta = max(0.0, ($nextAt ?? $now) - $now);

        $sec = (int)floor($delta);
        $usec = (int)floor(($delta - $sec) * 1_000_000);

        // stream_select expects usec 0..999999
        if ($usec >= 1_000_000) {
            $sec += intdiv($usec, 1_000_000);
            $usec = $usec % 1_000_000;
        }

        return [$sec, $usec];
    }
}

/* ------------------------------
 * Demo: TCP echo server
 * ------------------------------ */

$loop = new SelectLoop();

$server = stream_socket_server("tcp://127.0.0.1:9001", $errno, $errstr);
if (!$server) {
    fwrite(STDERR, "server create failed: $errstr ($errno)\n");
    exit(1);
}
stream_set_blocking($server, false);

echo "Listening on 127.0.0.1:9001\n";

$clients = [];

// Accept new connections
$loop->onReadable($server, function ($srv) use (&$clients, $loop) {
    $conn = @stream_socket_accept($srv, 0);
    if (!$conn) {
        return;
    }
    stream_set_blocking($conn, false);

    $id = $loop->onReadable($conn, function ($c) use (&$clients, $loop) {
        $data = @fread($c, 8192);
        if ($data === '' || $data === false) {
            // EOF or error => close
            $cid = (int)$c;
            fclose($c);
            if (isset($clients[$cid])) {
                $loop->cancel($clients[$cid]);
                unset($clients[$cid]);
            }
            return;
        }

        // Echo back (write immediately; if you need backpressure, register onWritable)
        @fwrite($c, $data);
    });

    $clients[(int)$conn] = $id;
    echo "client connected\n";
});

// Periodic timer
$loop->addPeriodicTimer(5.0, function () {
    echo "[timer] still alive: " . date('H:i:s') . "\n";
});

// Stop after 60 seconds (one-shot)
$loop->addTimer(60.0, function ($id, SelectLoop $loop) {
    echo "Stopping loop (60s elapsed)\n";
    $loop->stop();
});

$loop->run();
echo "bye\n";