chore(io/tcp/unix): improve performance, and closed handle/server condition

docs/component/filesystem.md

@@ -16,7 +16,7 @@
 
 #### `Functions`
 
-- [append_file](./../../src/Psl/Filesystem/append_file.php#L21)
+- [append_file](./../../src/Psl/Filesystem/append_file.php#L23)
 - [canonicalize](./../../src/Psl/Filesystem/canonicalize.php#L15)
 - [change_group](./../../src/Psl/Filesystem/change_group.php#L20)
 - [change_owner](./../../src/Psl/Filesystem/change_owner.php#L20)
@@ -51,6 +51,6 @@
 - [read_directory](./../../src/Psl/Filesystem/read_directory.php#L19)
 - [read_file](./../../src/Psl/Filesystem/read_file.php#L23)
 - [read_symbolic_link](./../../src/Psl/Filesystem/read_symbolic_link.php#L21)
-- [write_file](./../../src/Psl/Filesystem/write_file.php#L21)
+- [write_file](./../../src/Psl/Filesystem/write_file.php#L23)
 
 

docs/component/tcp.md

@@ -21,7 +21,7 @@
 #### `Classes`
 
 - [ConnectOptions](./../../src/Psl/TCP/ConnectOptions.php#L7)
-- [Server](./../../src/Psl/TCP/Server.php#L14)
+- [Server](./../../src/Psl/TCP/Server.php#L17)
 - [ServerOptions](./../../src/Psl/TCP/ServerOptions.php#L9)
 
 

docs/component/unix.md

@@ -20,6 +20,6 @@
 
 #### `Classes`
 
-- [Server](./../../src/Psl/Unix/Server.php#L12)
+- [Server](./../../src/Psl/Unix/Server.php#L15)
 
 

examples/tcp/basic-http-server.php

@@ -6,29 +6,52 @@
 
 use Psl\Async;
 use Psl\IO;
+use Psl\Network\Exception\AlreadyStoppedException;
 use Psl\Str;
 use Psl\TCP;
+use Throwable;
 
 require __DIR__ . '/../../vendor/autoload.php';
 
-Async\main(static function (): never {
-    $server = TCP\Server::create('localhost');
+Async\main(static function (): int {
+    $output = IO\output_handle();
+    $server = TCP\Server::create('localhost', 3030);
 
-    IO\output_handle()->writeAll("Server is listening on http://localhost:" . $server->getLocalAddress()->port . "\n");
+    $output->writeAll("Server is listening on http://localhost:3030\n");
 
-    while (true) {
-        $connection = $server->nextConnection();
-        $request = $connection->read();
+    Async\Scheduler::defer(static function () use ($server, $output) {
+        Async\await_signal(SIGINT);
 
-        IO\output_handle()->writeAll(Str\split($request, "\n")[0] . "\n");
+        $output->writeAll("\nGoodbye 👋\n");
 
-        $connection->writeAll("HTTP/1.1 200 OK\n");
-        $connection->writeAll("Server: PSL\n");
-        $connection->writeAll("Connection: close\n");
-        $connection->writeAll("Content-Type: text/plain\n\n");
+        $server->stopListening();
+    });
 
-        $connection->writeAll("Hello, World!");
+    try {
+        while (true) {
+            $connection = $server->nextConnection();
 
-        $connection->close();
+            Async\Scheduler::defer(static function() use ($connection, $output) {
+                try {
+                    $request = $connection->read();
+
+                    $output->writeAll("[" . round(memory_get_peak_usage(true) / 1024 / 1024, 1) . "MiB] " . Str\split($request, "\n")[0] . "\n");
+
+                    $connection->writeAll("HTTP/1.1 200 OK\n");
+
+                    $connection->writeAll("Server: PSL\n");
+                    $connection->writeAll("Connection: close\n");
+                    $connection->writeAll("Content-Type: text/plain\n\n");
+
+                    $connection->writeAll("Hello, World!");
+                    $connection->close();
+                } catch (Throwable) {
+                }
+            });
+        }
+    } catch (AlreadyStoppedException) {
+        // server stopped.
     }
+
+    return 0;
 });

src/Psl/Filesystem/append_file.php

@@ -9,6 +9,8 @@
 use Psl\IO;
 use Psl\Str;
 
+use function clearstatcache;
+
 /**
  * Append $content to $file.
  *
@@ -20,6 +22,8 @@
  */
 function append_file(string $file, string $content): void
 {
+    clearstatcache();
+
     try {
         $handle = File\open_write_only($file, File\WriteMode::APPEND);
         $lock = $handle->lock(File\LockType::EXCLUSIVE);
@@ -28,6 +32,8 @@ function append_file(string $file, string $content): void
 
         $lock->release();
         $handle->close();
+
+        clearstatcache();
     } catch (File\Exception\ExceptionInterface | IO\Exception\ExceptionInterface $previous) {
         // @codeCoverageIgnoreStart
         throw new Exception\RuntimeException(Str\format(

src/Psl/Filesystem/write_file.php

@@ -9,6 +9,8 @@
 use Psl\IO;
 use Psl\Str;
 
+use function clearstatcache;
+
 /**
  * Write $content to $file.
  *
@@ -20,6 +22,8 @@
  */
 function write_file(string $file, string $content): void
 {
+    clearstatcache();
+
     try {
         if (namespace\is_file($file)) {
             $mode = File\WriteMode::TRUNCATE;
@@ -34,6 +38,8 @@ function write_file(string $file, string $content): void
 
         $lock->release();
         $handle->close();
+
+        clearstatcache();
     } catch (File\Exception\ExceptionInterface | IO\Exception\ExceptionInterface $previous) {
         // @codeCoverageIgnoreStart
         throw new Exception\RuntimeException(Str\format(

src/Psl/IO/Internal/ResourceHandle.php

@@ -5,11 +5,12 @@
 namespace Psl\IO\Internal;
 
 use Psl;
-use Psl\Async;
 use Psl\Exception\InvariantViolationException;
 use Psl\IO;
 use Psl\IO\Exception;
 use Psl\Type;
+use Revolt\EventLoop;
+use Revolt\EventLoop\Suspension;
 
 use function error_get_last;
 use function fclose;
@@ -37,7 +38,7 @@ class ResourceHandle implements IO\CloseSeekReadWriteHandleInterface
     use IO\ReadHandleConvenienceMethodsTrait;
     use IO\WriteHandleConvenienceMethodsTrait;
 
-    public const DEFAULT_READ_BUFFER_SIZE = 65536;
+    public const DEFAULT_READ_BUFFER_SIZE = 4096;
     public const MAXIMUM_READ_BUFFER_SIZE = 786432;
 
     /**
@@ -49,6 +50,12 @@ class ResourceHandle implements IO\CloseSeekReadWriteHandleInterface
 
     private bool $blocks;
 
+    private string $readWatcher = '';
+    private string $writeWatcher = '';
+
+    private ?Suspension $readSuspension = null;
+    private ?Suspension $writeSuspension = null;
+
     /**
      * @param resource|object $resource
      *
@@ -75,7 +82,7 @@ public function __construct(mixed $resource, bool $read, bool $write, bool $seek
         $meta = stream_get_meta_data($resource);
         $this->blocks = ($meta['wrapper_type'] ?? '') === 'plainfile';
         if ($seek) {
-            $seekable = (bool) $meta['seekable'];
+            $seekable = (bool)$meta['seekable'];
 
             Psl\invariant($seekable, 'Handle is not seekable.');
         }
@@ -84,17 +91,32 @@ public function __construct(mixed $resource, bool $read, bool $write, bool $seek
             $readable = str_contains($meta['mode'], 'r') || str_contains($meta['mode'], '+');
 
             Psl\invariant($readable, 'Handle is not readable.');
+
+            $suspension = &$this->readSuspension;
+            $this->readWatcher = EventLoop::onReadable($resource, static function () use (&$suspension) {
+                /** @var Suspension|null $suspension */
+                $suspension?->resume(null);
+            });
+
+            EventLoop::disable($this->readWatcher);
         }
 
         if ($write) {
             $writable = str_contains($meta['mode'], 'x')
                 || str_contains($meta['mode'], 'w')
                 || str_contains($meta['mode'], 'c')
                 || str_contains($meta['mode'], 'a')
-                || str_contains($meta['mode'], '+')
-            ;
+                || str_contains($meta['mode'], '+');
 
             Psl\invariant($writable, 'Handle is not writeable.');
+
+            $suspension = &$this->writeSuspension;
+            $this->writeWatcher = EventLoop::onWritable($resource, static function () use (&$suspension) {
+                /** @var Suspension|null $suspension */
+                $suspension?->resume(null);
+            });
+
+            EventLoop::disable($this->writeWatcher);
         }
 
         $this->useSingleRead = $meta["stream_type"] === "udp_socket" || $meta["stream_type"] === "STDIO";
@@ -119,18 +141,32 @@ public function write(string $bytes, ?float $timeout = null): int
 
         $bytes = substr($bytes, $written);
 
-        try {
-            /**
-             * @psalm-suppress PossiblyInvalidArgument
-             * @psalm-suppress PossiblyNullArgument - If null, writeImmediately would have thrown.
-             */
-            Async\await_writable($this->resource, timeout: $timeout);
-        } catch (Async\Exception\TimeoutException) {
-            throw new Exception\TimeoutException('reached timeout while the handle is still not writable.');
-        } catch (Async\Exception\ResourceClosedException) {
-            $this->resource = null;
+        $this->writeSuspension = EventLoop::createSuspension();
+        $suspension = &$this->writeSuspension;
+        /** @psalm-suppress MissingThrowsDocblock */
+        EventLoop::enable($this->writeWatcher);
+        $delay_watcher = null;
+        if (null !== $timeout) {
+            $delay_watcher = EventLoop::delay(
+                $timeout,
+                static function () use (&$suspension) {
+                    /** @var Suspension|null $suspension */
+                    $suspension?->throw(
+                        new Exception\TimeoutException('reached timeout while the handle is still not writable.')
+                    );
+                }
+            );
+        }
 
-            throw new Exception\AlreadyClosedException('Handle has already been closed.');
+        try {
+            /** @var Suspension $suspension */
+            $suspension->suspend();
+        } finally {
+            $suspension = null;
+            EventLoop::disable($this->writeWatcher);
+            if (null !== $delay_watcher) {
+                EventLoop::cancel($delay_watcher);
+            }
         }
 
         return $written + $this->writeImmediately($bytes);
@@ -144,6 +180,10 @@ public function write(string $bytes, ?float $timeout = null): int
      */
     public function writeImmediately(string $bytes): int
     {
+        if (null !== $this->writeSuspension) {
+            throw new Exception\RuntimeException('Pending operation.');
+        }
+
         if (null === $this->resource) {
             throw new Exception\AlreadyClosedException('Handle has already been closed.');
         }
@@ -217,18 +257,32 @@ public function read(?int $max_bytes = null, ?float $timeout = null): string
             return $chunk;
         }
 
-        try {
-            /**
-             * @psalm-suppress PossiblyInvalidArgument
-             * @psalm-suppress PossiblyNullArgument - If null, writeImmediately would have thrown.
-             */
-            Async\await_readable($this->resource, timeout: $timeout);
-        } catch (Async\Exception\TimeoutException) {
-            throw new Exception\TimeoutException('reached timeout while the handle is still not readable.');
-        } catch (Async\Exception\ResourceClosedException) {
-            $this->resource = null;
+        $this->readSuspension = EventLoop::createSuspension();
+        $suspension = &$this->readSuspension;
+        /** @psalm-suppress MissingThrowsDocblock */
+        EventLoop::enable($this->readWatcher);
+        $delay_watcher = null;
+        if (null !== $timeout) {
+            $delay_watcher = EventLoop::delay(
+                $timeout,
+                static function () use (&$suspension) {
+                    /** @var Suspension|null $suspension */
+                    $suspension?->throw(
+                        new Exception\TimeoutException('reached timeout while the handle is still not readable.')
+                    );
+                }
+            );
+        }
 
-            throw new Exception\AlreadyClosedException('Handle has already been closed.');
+        try {
+            /** @var Suspension $suspension */
+            $suspension->suspend();
+        } finally {
+            $suspension = null;
+            EventLoop::disable($this->readWatcher);
+            if (null !== $delay_watcher) {
+                EventLoop::cancel($delay_watcher);
+            }
         }
 
         return $this->readImmediately($max_bytes);
@@ -241,6 +295,10 @@ public function read(?int $max_bytes = null, ?float $timeout = null): string
      */
     public function readImmediately(?int $max_bytes = null): string
     {
+        if (null !== $this->readSuspension) {
+            throw new Exception\RuntimeException('Pending operation.');
+        }
+
         if (null === $this->resource) {
             throw new Exception\AlreadyClosedException('Handle has already been closed.');
         }
@@ -285,5 +343,8 @@ public function close(): void
 
             throw new Exception\RuntimeException($error['message'] ?? 'unknown error.');
         }
+
+        $this->readSuspension?->throw(throw new Exception\AlreadyClosedException('Handle has already been closed.'));
+        $this->writeSuspension?->throw(throw new Exception\AlreadyClosedException('Handle has already been closed.'));
     }
 }

src/Psl/Internal/Loader.php

@@ -474,7 +474,6 @@ final class Loader
         'Psl\Network\Internal\get_sock_name',
         'Psl\Network\Internal\socket_connect',
         'Psl\Network\Internal\server_listen',
-        'Psl\Network\Internal\socket_accept',
         'Psl\TCP\connect',
         'Psl\Unix\connect',
     ];