feat(tcp): add backlog parameter and batch-drain accept loop (#617)

Author: azjezz

CHANGELOG.md

@@ -4,7 +4,9 @@
 
 ### features
 
-- feat(dict,vec): add filter_nonnull_by and map_nonnull - [#576](https://github.com/azjezz/psl/pull/576) by @Dima-369
+- feat(dict, vec): add filter_nonnull_by and map_nonnull - [#576](https://github.com/azjezz/psl/pull/576) by @Dima-369
+* feat(tcp): add `backlog` parameter to `TCP\listen()` for configuring the pending connection queue size - [#617](https://github.com/azjezz/psl/pull/617) - by @azjezz
+* feat(tcp): listener now drains the accept backlog in a loop for higher throughput - [#617](https://github.com/azjezz/psl/pull/617) - by @azjezz
 
 ### other
 

docs/content/networking/tcp.md

@@ -22,6 +22,12 @@ All TCP connectors implement `ConnectorInterface`, making them interchangeable a
 
 @example('networking/tcp-socket-pool.php')
 
+### Backlog
+
+`TCP\listen()` accepts a `backlog` parameter (default 512) to configure the OS-level queue of pending connections. A larger backlog helps high-connection-rate servers avoid dropped connections.
+
+@example('networking/tcp-backlog.php')
+
 ### Low-Level Socket
 
 `Socket` gives you fine-grained control over socket options before connecting or listening. Create a socket, configure it, then consume it:

docs/examples/networking/tcp-backlog.php

@@ -0,0 +1,13 @@
+<?php
+
+declare(strict_types=1);
+
+require_once __DIR__ . '/../../../vendor/autoload.php';
+
+use Psl\TCP;
+
+// Default backlog of 512
+$listener = TCP\listen('127.0.0.1', 8080);
+
+// High-throughput server with larger backlog
+$listener = TCP\listen('127.0.0.1', 8080, backlog: 4096);

src/Psl/TCP/Internal/Listener.php

@@ -10,9 +10,11 @@
 use Psl\TCP;
 use Revolt\EventLoop;
 
+use function error_clear_last;
 use function error_get_last;
 use function fclose;
 use function is_resource;
+use function str_contains;
 use function stream_socket_accept;
 
 /**
@@ -51,38 +53,42 @@ public function __construct(mixed $impl, int $idleConnections = self::DEFAULT_ID
         [$receiver, $sender] = Channel\bounded($idleConnections);
 
         $this->receiver = $receiver;
-        $this->watcher = EventLoop::onReadable(
-            $impl,
-            /**
-             * @param resource $resource
-             */
-            static function (string $watcher, mixed $resource) use ($sender): void {
-                try {
+        $this->watcher = EventLoop::onReadable($impl, static function (string $watcher, mixed $resource) use (
+            $sender,
+        ): void {
+            try {
+                while (true) {
+                    error_clear_last();
                     $sock = @stream_socket_accept($resource, timeout: 0.0);
-                    if (false !== $sock) {
+                    if ($sock !== false) {
                         $sender->send([true, new Stream($sock)]);
-
-                        return;
+                        continue;
                     }
 
                     // @codeCoverageIgnoreStart
-                    /** @var array{file: string, line: int, message: string, type: int} $err */
                     $err = error_get_last();
-                    $sender->send([
-                        false,
-                        new Network\Exception\RuntimeException(
-                            'Failed to accept incoming connection: ' . $err['message'],
-                            $err['type'],
-                        ),
-                    ]);
-                    // @codeCoverageIgnoreEnd
-                } catch (Channel\Exception\ClosedChannelException) {
-                    EventLoop::cancel($watcher);
+                    if ($err !== null && !str_contains($err['message'], 'Accept failed')) {
+                        // OS error (e.g., EMFILE, ENFILE, ENOBUFS)
+                        $sender->send([
+                            false,
+                            new Network\Exception\RuntimeException(
+                                'Failed to accept incoming connection: ' . $err['message'],
+                                $err['type'],
+                            ),
+                        ]);
 
-                    return;
+                        return;
+                    }
+
+                    // No more pending connections (EAGAIN / timeout with no backlog).
+                    break;
+                    // @codeCoverageIgnoreEnd
                 }
-            },
-        );
+            } catch (Channel\Exception\ClosedChannelException) {
+                EventLoop::cancel($watcher);
+                return;
+            }
+        });
     }
 
     #[Override]

src/Psl/TCP/listen.php

@@ -13,6 +13,7 @@
  * @param non-empty-string $host
  * @param int<0, max> $port
  * @param int<1, max> $idle_connections Maximum number of idle connections to buffer.
+ * @param int<1, max> $backlog Maximum length of the queue of pending connections.
  *
  * @throws Network\Exception\RuntimeException If failed to listen on given address.
  */
@@ -23,13 +24,15 @@ function listen(
     bool $reuse_address = false,
     bool $reuse_port = false,
     int $idle_connections = 256,
+    int $backlog = 512,
 ): ListenerInterface {
     $socket_context = ['socket' => [
         'ipv6_v6only' => true,
         'so_reuseaddr' => OS\is_windows() ? $reuse_port : $reuse_address,
         'so_reuseport' => $reuse_port,
         'so_broadcast' => false,
         'tcp_nodelay' => $no_delay,
+        'backlog' => $backlog,
     ]];
 
     $socket = Network\Internal\server_listen("tcp://{$host}:{$port}", $socket_context);

tests/unit/TCP/ServerTest.php

@@ -35,6 +35,53 @@ public function testGetLocalAddressOnStoppedListener(): void
         $listener->getLocalAddress();
     }
 
+    public function testListenWithCustomBacklog(): void
+    {
+        $listener = TCP\listen('127.0.0.1', 0, backlog: 128);
+        $address = $listener->getLocalAddress();
+
+        static::assertSame('127.0.0.1', $address->host);
+        static::assertGreaterThan(0, $address->port);
+
+        $client = TCP\connect('127.0.0.1', $address->port);
+        $server = $listener->accept();
+
+        $client->write('ping');
+        static::assertSame('ping', $server->read(4));
+
+        $client->close();
+        $server->close();
+        $listener->close();
+    }
+
+    public function testAcceptMultipleConnections(): void
+    {
+        $listener = TCP\listen('127.0.0.1', 0, no_delay: true, backlog: 64);
+        $address = $listener->getLocalAddress();
+
+        [$server1, $client1, $client2] = Async\concurrently([
+            $listener->accept(...),
+            static fn(): Network\StreamInterface => TCP\connect('127.0.0.1', $address->port),
+            static fn(): Network\StreamInterface => TCP\connect('127.0.0.1', $address->port),
+        ]);
+
+        $server2 = $listener->accept();
+
+        $client1->write('one');
+        $client2->write('two');
+
+        $msg1 = $server1->read(3);
+        $msg2 = $server2->read(3);
+
+        static::assertTrue($msg1 === 'one' && $msg2 === 'two' || $msg1 === 'two' && $msg2 === 'one');
+
+        $client1->close();
+        $client2->close();
+        $server1->close();
+        $server2->close();
+        $listener->close();
+    }
+
     public function testWaitsForPendingOperation(): void
     {
         $listener = TCP\listen('127.0.0.1');