Implementando Coroutines con Fibers en PHP 8.1+ para Sockets Asíncronos

PHP 8.1 introdujo Fibers, una herramienta poderosa para implementar concurrencia sin la sobrecarga de hilos reales. Podemos usarlas para crear coroutines, funciones que pueden ser pausadas y reanudadas, permitiendo un manejo asíncrono eficiente de operaciones de E/S, como la lectura y escritura de sockets.

Tradicionalmente, el manejo de sockets en PHP era bloqueante. Esto significa que una operación como socket_read detiene la ejecución hasta que haya datos disponibles. Con Fibers, podemos crear un bucle de eventos donde cada socket es manejado por una coroutine, permitiendo que otras tareas se ejecuten mientras se espera la disponibilidad de datos.

<?php

class SocketCoroutine
{
    private Fiber $fiber;
    private Socket $socket;

    public function __construct(Socket $socket)
    {
        $this->socket = $socket;
        $this->fiber = new Fiber($this->run(...));
    }

    public function start(): void
    {
        $this->fiber->start();
    }

    private function run(): void
    {
        while (true) {
            // Intenta leer del socket, pero sin bloquear
            $read = [$this->socket];
            $write = $except = null;
            $num_changed_streams = socket_select($read, $write, $except, 0, 100000); // Timeout de 0.1 segundos
            
            if ($num_changed_streams > 0) {
                $data = socket_read($this->socket, 1024);
                if ($data === false || $data === '') {
                    echo "Socket cerrado.\n";
                    socket_close($this->socket);
                    return;
                }
                echo "Recibido: " . $data;
            } else {
                // No hay datos disponibles, cede el control a otro Fiber
                Fiber::suspend();
            }
        }
    }

    public function resume(): void
    {
        if ($this->fiber->isSuspended()) {
            $this->fiber->resume();
        }
    }

    public function isFinished(): bool
    {
        return $this->fiber->isTerminated();
    }
}

// Ejemplo de uso:
$socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
socket_bind($socket, '127.0.0.1', 8080);
socket_listen($socket);
socket_set_nonblock($socket); //Importante para evitar bloqueos en el accept

$coroutines = [];
while (true) {
    $client_socket = socket_accept($socket);
    if ($client_socket !== false) {
        echo "Nueva conexión!\n";
        $coroutines[] = new SocketCoroutine($client_socket);
        $coroutines[count($coroutines)-1]->start();
    }

    foreach ($coroutines as $key => $coroutine) {
        if (!$coroutine->isFinished()) {
            $coroutine->resume();
        } else {
            unset($coroutines[$key]);
        }
    }

    usleep(10000); // Dormir 10ms para evitar consumo excesivo de CPU.
}
    

Este código crea una clase SocketCoroutine que envuelve un socket y una Fiber. La Fiber ejecuta un bucle que intenta leer del socket usando socket_select con un timeout. Si no hay datos disponibles, la Fiber se suspende utilizando Fiber::suspend(). El bucle principal se encarga de aceptar nuevas conexiones y reanudar las coroutines suspendidas.

Es crucial usar socket_set_nonblock() para evitar que socket_accept() bloquee. El socket_select() con timeout permite a la coroutine ceder el control si no hay datos disponibles, evitando el bloqueo del hilo principal.

Este enfoque permite manejar múltiples conexiones concurrentemente utilizando un solo hilo de ejecución, mejorando la eficiencia y reduciendo la sobrecarga en comparación con el uso de hilos tradicionales.