Releases

The tyhp/io Package

tyhp/io gives Tyhp programs non-blocking I/O on the tyhp/async event loop. It covers HTTP requests, external processes, local files, TCP, TLS, and Unix-domain sockets, Redis, PostgreSQL, MySQL/MariaDB, any PDO driver, DNS lookups, and a general way to move blocking work out of the way. While one fiber waits for a response, a query, or a child process, every other fiber keeps running.

Everything lives under the Tyhp\Io namespace. You write ordinary sequential code with await, and you combine independent operations with Promise::all() or Promise::batch() from tyhp/async. This page is the reference for the package API. The language features it relies on are covered in Async and Await, Structs, and Generics.

Overview

The Async naming convention

Every public method that returns a Promise ends in Async, for example HttpClient::getAsync(), File::linesAsync(), and DbConnection::queryAsync(). When you see the suffix, the call starts work and hands back a Promise<T>: await it inside async code to get the T, or collect several and await them together. Methods without the suffix (close(), isClosed(), header(), rowsAs(), Offload::usesWorkers(), and so on) return their value directly.

Methods that implement an interface contract keep the contract's name. OffloadJob::run() is the blocking method you write for Offload::runAsync(), and the iterator returned by File::linesAsync() uses the standard AsyncIterator methods next() and current(), which foreach (await ...) calls for you.

Note

The signatures on this page show what a caller receives. A method shown as returning Promise<T> may be declared async with return type T in the package source; the two are the same to the caller. Static methods are marked static.

What is in the package

Area Namespace Main types Needs
HTTP client Tyhp\Io\Http HttpClient, HttpRequest, HttpResponse ext-curl
Processes Tyhp\Io\Process Process, ProcessOptions, ProcessResult —
Files Tyhp\Io\File File —
Sockets Tyhp\Io\Socket Socket, SocketConnection ext-openssl for tls://
Redis Tyhp\Io\Redis RedisClient, RedisOptions —
Databases Tyhp\Io\Db DbConnection, DbResult, DbStatement, DbValue, PostgresConnection, MySqlConnection, MySqlOptions, PdoConnection, PdoOptions ext-pgsql, ext-mysqli, or ext-pdo plus a PDO driver
Blocking work Tyhp\Io Offload, OffloadJob tyhp/parallel (optional)
DNS Tyhp\Io\Dns Dns —
Errors Tyhp\Io\Exceptions IoException and its subclasses —

How each area waits

PHP does not offer non-blocking versions of every kind of I/O, so the package uses the best mechanism available for each area. This table tells you what still blocks the event loop.

Area How it waits What can still block the loop
HttpClient One curl multi handle per client, driven from the event loop Nothing during transfers
Process Non-blocking stdin/stdout/stderr pipes Nothing while the command runs
File Reads and writes in 256 KiB chunks and yields to other fibers between chunks Each single chunk, briefly
Socket / SocketConnection Non-blocking streams registered with the event loop Host-name resolution inside PHP's socket connect call
RedisClient A SocketConnection Same as sockets
PostgresConnection libpq's asynchronous connect and query API Nothing
MySqlConnection MYSQLI_ASYNC queries polled from the event loop The connection handshake
PdoConnection, Dns, Offload A tyhp/parallel worker process when available The whole call, when no worker is available

Installation

tyhp/io is a compiled runtime package. The compiler does not add it for you; require it explicitly:

composer require tyhp/io

Version pins follow the same 80N.X.Y scheme as the other compiled tyhp/* helpers, where 80N is the PHP version your project targets (output.phpVersion) and X.Y is the package's own version (currently 0.1). A project targeting PHP 8.4 therefore pins "tyhp/io": "804.0.1" or "804.0.*". See Composer Runtime Packages for how these versions work. The package requires PHP 8.2 or later, tyhp/core, and tyhp/async.

PHP extensions

The package installs without any optional extension. Each component checks for the extension it needs when you first use it and fails with IoUnavailableException when it is missing:

Component Extension
HttpClient ext-curl
PostgresConnection ext-pgsql
MySqlConnection ext-mysqli
PdoConnection ext-pdo and the driver for your DSN (for example pdo_sqlite, pdo_mysql, pdo_pgsql)

tls:// sockets and RedisOptions::$tls use PHP's OpenSSL stream support.

The package's public API uses only its own types and builtins, so you do not need any tyhpdef/php-ext-* package to call it. Add one only when your own code calls that extension directly, for example tyhpdef/php-ext-pdo to write \PDO::ATTR_TIMEOUT in PdoOptions::$attributes.

Optional: tyhp/parallel

tyhp/parallel is not required. When it is installed and \Tyhp\Parallel::isAvailable() returns true, Offload, PdoConnection, and Dns run their blocking work in worker processes and the event loop stays free. Without it, the same calls run in your process and block the loop until they finish. Your code does not change either way. See With and without tyhp/parallel below.

composer require tyhp/parallel

Getting started

An application entry script can use await at the top level; the compiler starts the event loop for it. Put the work in an async function and await it:

<?tyhp

use Tyhp\Io\File\File;
use Tyhp\Io\Http\HttpClient;

type Release = struct {
    string $tag_name = '';
    string $published_at = '';
};

async function mainAsync(): void {
    HttpClient $github = new HttpClient(['User-Agent' => 'release-watch/1.0']);

    Release $release = await $github->getJsonAsync<Release>(
        'https://api.github.com/repos/composer/composer/releases/latest'
    );

    await File::writeAsync(__DIR__ . '/latest-release.txt', $release->tag_name . ' ' . $release->published_at . "\n");
    echo 'Latest Composer release: ' . $release->tag_name . "\n";
}

await mainAsync();

getJsonAsync<Release>() sends the request, requires a 2xx status, and decodes the JSON body into the Release struct. File::writeAsync() writes the file without holding up other fibers. If any step fails, the exception is thrown at its await, exactly like a synchronous call.

From synchronous code, such as a framework controller, a console command, or a library function, run a promise to completion with the event loop:

<?tyhp

namespace App\Console;

use Tyhp\EventLoop;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;

final class StatusCommand {
    public function execute(string $url): int {
        HttpResponse $response = EventLoop::getInstance()->run(HttpClient::shared()->getAsync($url));
        echo $url . ' answered ' . $response->status . "\n";
        return $response->isSuccess() ? 0 : 1;
    }
}

EventLoop::run() blocks until the promise settles and returns its value, or throws its error. To run several awaits in one go, wrap them in an async closure and pass it to Promise::run(), which returns the closure's settled value (the type argument is inferred from the closure):

<?tyhp

namespace App\Console;

use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;
use Tyhp\Promise;

function statusOf(string $url): int {
    HttpResponse $response = Promise::run(async fn(): HttpResponse => await HttpClient::shared()->getAsync($url));
    return $response->status;
}

Concepts

Awaiting and running work concurrently

Awaiting one call after another runs them in sequence. To run independent operations at the same time, start them all and await the combined promise:

<?tyhp

namespace App\Dashboard;

use Tyhp\Io\Db\DbResult;
use Tyhp\Io\Db\PostgresConnection;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;
use Tyhp\Promise;

async function loadDashboardAsync(HttpClient $api, PostgresConnection $db, int $userId): string {
    // All three start immediately; await waits for the slowest one.
    Promise<HttpResponse> $profile = $api->getAsync('https://api.example.com/users/' . $userId);
    Promise<HttpResponse> $notices = $api->getAsync('https://api.example.com/users/' . $userId . '/notices');
    Promise<DbResult> $orders = $db->queryAsync('SELECT id, total FROM orders WHERE user_id = $1', [$userId]);

    HttpResponse $profileResponse = await $profile;
    HttpResponse $noticesResponse = await $notices;
    DbResult $orderRows = await $orders;

    return $profileResponse->body . $noticesResponse->body . $orderRows->count();
}

Promise::all() collects many promises of the same type, and Promise::batch() limits how many run at once, which matters when you fan out over hundreds of URLs, files, or commands. Both come from tyhp/async; see Promise Combinators.

Concurrency has limits inside some components:

  • One HttpClient runs any number of requests at once.
  • One RedisClient pipelines commands from every fiber over its single connection.
  • One DbConnection runs one statement at a time, in call order. Open several connections to run queries in parallel.
  • One SocketConnection allows one pending read at a time; writes may overlap and are sent in call order.

Cancellation and timeouts

Most Async methods take an optional trailing ?CancellationToken $token. Create the token with a CancellationTokenSource from tyhp/async, optionally with an automatic deadline in milliseconds. When the token is cancelled, the operation stops and its promise rejects with Tyhp\Exceptions\OperationCancelledException. A token that is already cancelled when you make the call rejects right away.

<?tyhp

namespace App\Deadline;

use Tyhp\CancellationTokenSource;
use Tyhp\Exceptions\OperationCancelledException;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;
use Tyhp\Promise;

async function fetchBothAsync(HttpClient $client): ?array<HttpResponse> {
    // One deadline for the whole operation: cancelled automatically after 3 seconds.
    $cts := new CancellationTokenSource(3000);
    try {
        array<HttpResponse> $responses = await Promise::all([
            $client->getAsync('https://a.example.com/', [], $cts->getToken()),
            $client->getAsync('https://b.example.com/', [], $cts->getToken()),
        ]);
        return $responses;
    } catch (OperationCancelledException $e) {
        return null;
    }
}

What cancellation does depends on the component:

Component On cancellation
HttpClient Aborts the transfer
Process Kills the command (signal 9), then rejects
File Stops at the next 256 KiB chunk; an interrupted write or copy leaves a partial file
Socket::connectAsync() Closes the half-open connection
SocketConnection reads and writes Stops waiting; data already buffered stays buffered
RedisClient::connectAsync() Stops connecting (commands themselves take no token)
PostgresConnection Sends a cancel request for the running statement, waits for the server to stop it
MySqlConnection Runs KILL QUERY on a second, short-lived connection
PdoConnection, Dns, Offload With a worker: stops that worker. Without: checked only before the work starts

Timeouts are separate settings that reject with Tyhp\Exceptions\TimeoutException (which has getTimeoutMs()): HttpRequest::$timeoutMs and $connectTimeoutMs, ProcessOptions::$timeoutMs, the $timeoutMs argument of Socket::connectAsync(), RedisOptions::$connectTimeoutMs, and MySqlOptions::$connectTimeoutSeconds. For anything without its own timeout, wrap the promise in Promise::timeout() or pass a token from new CancellationTokenSource($ms).

Errors

All package-specific failures extend Tyhp\Io\Exceptions\IoException, which extends \RuntimeException. The subclasses tell you which component failed and carry extra data: HttpStatusException::getResponse(), ProcessFailedException::getResult(), and DbException::getSqlState(). The full list is in Exceptions.

A few errors come from outside the package and do not extend IoException:

Exception When
Tyhp\Exceptions\TimeoutException (tyhp/async) A configured timeout elapsed
Tyhp\Exceptions\OperationCancelledException (tyhp/async) The token was cancelled
Tyhp\Exceptions\IncompatibleTypeException (tyhp/core) Decoded JSON or a database row did not match the requested struct
\JsonException Invalid JSON in json(), getJsonAsync(), readJsonAsync(), or an unencodable postJsonAsync() payload
\InvalidArgumentException Invalid input: unsupported URL or address scheme, a header containing a line break, an empty command, an unknown hash algorithm, a placeholder count mismatch, an empty PDO DSN
\LogicException A second read started on a SocketConnection while one is pending
Tyhp\Exceptions\ParallelException (tyhp/parallel) An Offload job threw inside a worker process

Failures are reported by rejecting the returned promise, so they are thrown where you await. Two calls throw immediately instead: HttpClient::postJsonAsync() throws \JsonException before returning when the payload cannot be encoded, and the PdoConnection constructor throws \InvalidArgumentException for an empty DSN.

With and without tyhp/parallel

Offload::runAsync() sends work to tyhp/parallel when two things are true: the Tyhp\Parallel class exists (the package is installed) and Parallel::isAvailable() returns true. That is the case in a CLI process with ext-pcntl and ext-posix, or in any process (such as php-fpm) whose TYHP_PARALLEL_SOCKET environment variable names a running worker service. Offload::usesWorkers() reports the result. PdoConnection and Dns go through Offload, so they follow the same rule.

With workers Without workers
Where the job runs A worker process This process
Event loop while it runs Free; other fibers keep running Blocked until the job returns
A job that throws Rejects with ParallelException (the original class is in getRemoteClass()) Rejects with the original exception
Cancellation Stops the worker running the job Checked once, before the job starts
Job requirements Serializable job object and result; the worker must be able to autoload the job class None beyond OffloadJob
PdoConnection connections One connection per worker for each DSN, user, and password One cached connection in this process

In a CLI program that never calls Parallel::start(), the first offloaded job starts a pool of four worker processes. Those workers are forked from your process, so start the pool before you open long-lived connections. See The tyhp/parallel Package.

HTTP client (Tyhp\Io\Http)

HttpClient sends HTTP and HTTPS requests through ext-curl. Every request made through one client shares a curl multi handle, so any number of requests run at the same time on the event loop. Every response status resolves normally, including 4xx and 5xx; call ensureSuccess() (or use getJsonAsync()) when a non-2xx status should be an error.

Only http:// and https:// URLs are accepted, including on redirects. Compressed responses are decoded automatically, so HttpResponse::$body is always the plain body.

HttpClient

<?tyhp
namespace Tyhp\Io\Http;

final class HttpClient { /* ... */ }

HttpRequest

HttpRequest is a struct describing one request for sendAsync(). Build it with new HttpRequest() with [...] and set only the fields you need.

<?tyhp
namespace Tyhp\Io\Http;

type HttpRequest = struct {
    string $url = '';
    string $method = 'GET';
    array<string, string> $headers = [];
    string $body = '';
    int $timeoutMs = 30000;
    int $connectTimeoutMs = 10000;
    bool $followRedirects = true;
    int $maxRedirects = 5;
    bool $verifyTls = true;
};
Field Type Default Meaning
url string '' Absolute http:// or https:// URL. Required.
method string 'GET' HTTP method, case-insensitive.
headers array<string, string> [] Header name ⇒ value. Merged over the client's default headers.
body string '' Request body. A non-empty body is sent with any method except HEAD.
timeoutMs int 30000 Limit for the whole transfer in milliseconds. 0 means no limit.
connectTimeoutMs int 10000 Limit for establishing the connection in milliseconds. 0 uses curl's built-in default.
followRedirects bool true Follow 3xx responses. When false, the redirect response itself resolves.
maxRedirects int 5 Most redirects to follow before failing with HttpException.
verifyTls bool true Verify the server certificate and host name. Turn off only for local testing.

HttpResponse

HttpResponse is the completed response. After redirects, headers and url describe the final response.

HTTP examples

Calling a JSON API with default headers and typed responses:

<?tyhp

namespace App\Billing;

use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;

type Invoice = struct {
    string $id = '';
    int $amount_due = 0;
    string $status = '';
};

final class BillingApi {
    private HttpClient $http;

    public function __construct(private readonly string $baseUrl, string $apiKey) {
        $this->http = new HttpClient([
            'Authorization' => 'Bearer ' . $apiKey,
            'User-Agent' => 'shop-backend/2.3',
        ]);
    }

    public async function invoiceAsync(string $id): Invoice {
        return await $this->http->getJsonAsync<Invoice>($this->baseUrl . '/invoices/' . \rawurlencode($id));
    }

    public async function createInvoiceAsync(string $customerId, int $amountCents): Invoice {
        HttpResponse $response = await $this->http->postJsonAsync(
            $this->baseUrl . '/invoices',
            ['customer' => $customerId, 'amount' => $amountCents],
            ['Idempotency-Key' => \bin2hex(\random_bytes(16))]
        );
        return $response->ensureSuccess()->json<Invoice>();
    }
}

Building a custom request and telling the failure modes apart:

<?tyhp

namespace App\Sync;

use Tyhp\Exceptions\TimeoutException;
use Tyhp\Io\Exceptions\HttpException;
use Tyhp\Io\Exceptions\HttpStatusException;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpRequest;
use Tyhp\Io\Http\HttpResponse;

async function putDocumentAsync(HttpClient $client, string $url, string $json): bool {
    HttpRequest $request = new HttpRequest() with [
        url => $url,
        method => 'PUT',
        headers => ['Content-Type' => 'application/json'],
        body => $json,
        timeoutMs => 5000,
        followRedirects => false,
    ];

    try {
        HttpResponse $response = await $client->sendAsync($request);
        $response->ensureSuccess();
        return true;
    } catch (HttpStatusException $e) {
        // The server answered, but not with 2xx.
        HttpResponse $failed = $e->getResponse();
        \error_log('PUT ' . $url . ' returned ' . $failed->status . ': ' . $failed->body);
        return false;
    } catch (TimeoutException $e) {
        \error_log('PUT ' . $url . ' timed out after ' . $e->getTimeoutMs() . 'ms');
        return false;
    } catch (HttpException $e) {
        // No response at all: DNS, connect, TLS, or transfer failure.
        \error_log('PUT ' . $url . ' failed (curl error ' . $e->getCode() . '): ' . $e->getMessage());
        return false;
    }
}

HttpStatusException extends HttpException, so catch it first when you handle both.

Processes (Tyhp\Io\Process)

Process runs external commands without blocking the event loop. The command is an argument list passed straight to the program, with no shell in between, so arguments never need quoting and cannot be interpreted as shell syntax. stdout and stderr are collected in full and returned when the command exits.

Process

<?tyhp
namespace Tyhp\Io\Process;

final class Process { /* ... */ }

ProcessOptions

<?tyhp
namespace Tyhp\Io\Process;

type ProcessOptions = struct {
    ?string $cwd;
    ?array<string, string> $env;
    string $input = '';
    int $timeoutMs = 0;
};
Field Type Default Meaning
cwd ?string null Working directory for the command. null uses this process's working directory.
env ?array<string, string> null The command's complete environment. null inherits this process's environment. A non-null array replaces the environment, so include variables such as PATH or HOME if the command needs them.
input string '' Written to the command's stdin, which is then closed. With '', stdin is closed immediately.
timeoutMs int 0 Kill the command and reject with TimeoutException after this many milliseconds. 0 means no limit.

ProcessResult

<?tyhp
namespace Tyhp\Io\Process;

type ProcessResult = struct {
    array<int, string> $command = [];
    int $exitCode = 0;
    int $signal = 0;
    string $stdout = '';
    string $stderr = '';
};
Field Type Meaning
command array<int, string> The command that ran.
exitCode int The exit code. When a signal ended the command, this is 128 + signal, as a shell reports it.
signal int The number of the signal that ended the command, or 0 when it exited normally.
stdout string Everything the command wrote to stdout.
stderr string Everything the command wrote to stderr.

Process examples

Running a command in a directory, with a timeout, and treating failure as an error:

<?tyhp

namespace App\Git;

use Tyhp\Io\Process\Process;
use Tyhp\Io\Process\ProcessOptions;
use Tyhp\Io\Process\ProcessResult;

async function currentCommitAsync(string $repoDir): string {
    ProcessResult $result = await Process::mustRunAsync(
        ['git', 'rev-parse', 'HEAD'],
        new ProcessOptions() with [cwd => $repoDir, timeoutMs => 5000]
    );
    return \trim($result->stdout);
}

Feeding stdin and distinguishing the ways a command can fail:

<?tyhp

namespace App\Json;

use Tyhp\Exceptions\TimeoutException;
use Tyhp\Io\Exceptions\IoException;
use Tyhp\Io\Exceptions\ProcessFailedException;
use Tyhp\Io\Process\Process;
use Tyhp\Io\Process\ProcessOptions;
use Tyhp\Io\Process\ProcessResult;

async function prettyPrintAsync(string $json): string {
    try {
        ProcessResult $result = await Process::mustRunAsync(
            ['jq', '.'],
            new ProcessOptions() with [input => $json, timeoutMs => 2000]
        );
        return $result->stdout;
    } catch (ProcessFailedException $e) {
        ProcessResult $failed = $e->getResult();
        throw new \RuntimeException('jq exited with ' . $failed->exitCode . ': ' . $failed->stderr, 0, $e);
    } catch (TimeoutException $e) {
        throw new \RuntimeException('jq did not finish within 2 seconds.', 0, $e);
    } catch (IoException $e) {
        // ProcessFailedException is also an IoException, so it is caught above first.
        throw new \RuntimeException('Could not start jq: ' . $e->getMessage(), 0, $e);
    }
}

Running many commands with bounded concurrency:

<?tyhp

namespace App\Lint;

use Tyhp\Io\Process\Process;
use Tyhp\Io\Process\ProcessResult;
use Tyhp\Promise;

async function lintFilesAsync(array<int, string> $files): int {
    // At most four `php -l` processes at a time.
    array<int, ProcessResult> $results = await Promise::batch(
        $files,
        fn(string $file): Promise<ProcessResult> => Process::runAsync([\PHP_BINARY, '-l', $file]),
        4
    );

    int $failures = 0;
    foreach ($results as $result) {
        if ($result->exitCode !== 0) {
            $failures++;
            echo $result->stdout . $result->stderr;
        }
    }
    return $failures;
}

Files (Tyhp\Io\File)

PHP has no non-blocking file I/O, so File reads and writes in 256 KiB chunks and lets other fibers, timers, and sockets run between chunks. Each chunk still blocks briefly, but a large file no longer freezes your server while it is read or written. A cancelled token stops the operation at the next chunk; a write or copy stopped that way leaves a partial file.

Every method fails with IoException when a file cannot be opened, read, or written; the message includes the path and PHP's reason.

File

<?tyhp
namespace Tyhp\Io\File;

final class File { /* ... */ }

File examples

Loading a typed configuration file:

<?tyhp

namespace App\Config;

use Tyhp\Io\File\File;

type AppConfig = struct {
    string $name = '';
    bool $debug = false;
    array<string, string> $database = [];
};

async function loadConfigAsync(string $path): AppConfig {
    return await File::readJsonAsync<AppConfig>($path);
}

Scanning a large log file line by line, with cancellation:

<?tyhp

namespace App\Logs;

use Tyhp\CancellationToken;
use Tyhp\Io\File\File;

async function countErrorsAsync(string $logPath, ?CancellationToken $token = null): int {
    int $errors = 0;
    foreach (await File::linesAsync($logPath) as string $line) {
        $token?->throwIfCancellationRequested();
        if (\str_contains($line, ' ERROR ')) {
            $errors++;
        }
    }
    return $errors;
}

Backing up a file under a content-addressed name and recording it:

<?tyhp

namespace App\Backup;

use Tyhp\Io\File\File;

async function backupAsync(string $path): string {
    string $digest = await File::hashAsync($path);
    string $target = $path . '.' . \substr($digest, 0, 12) . '.bak';
    await File::copyAsync($path, $target);
    await File::appendAsync($path . '.log', \date('c') . ' backed up to ' . $target . "\n");
    return $target;
}

Sockets (Tyhp\Io\Socket)

Socket::connectAsync() opens a client connection over TCP, TLS, or a Unix-domain socket and returns a SocketConnection whose reads and writes wait on the event loop. Use it to speak line-based or framed protocols that have no dedicated client in this package.

Socket

SocketConnection

A connected, non-blocking stream. You get instances from Socket::connectAsync(); there is no public constructor.

One read may be pending at a time. Data that arrives beyond what a read returns stays in an internal buffer for the next read, so you can mix readLineAsync(), readExactlyAsync(), and readAsync() freely. Writes may overlap; they are sent in the order you called writeAsync().

Socket examples

A request/response exchange over a line protocol (here, memcached's version command):

<?tyhp

namespace App\Memcached;

use Tyhp\Io\Exceptions\IoException;
use Tyhp\Io\Socket\Socket;
use Tyhp\Io\Socket\SocketConnection;

async function memcachedVersionAsync(string $host, int $port = 11211): string {
    SocketConnection $conn = await Socket::connectAsync('tcp://' . $host . ':' . $port, 2000);
    try {
        await $conn->writeAsync("version\r\n");
        ?string $line = await $conn->readLineAsync();
        if ($line === null || !\str_starts_with($line, 'VERSION ')) {
            throw new IoException('Unexpected reply: ' . ($line ?? '(connection closed)'));
        }
        return \substr($line, 8);
    } finally {
        $conn->close();
    }
}

Reading length-prefixed frames (a 4-byte big-endian length followed by the payload):

<?tyhp

namespace App\Frames;

use Tyhp\Io\Socket\SocketConnection;

async function readFrameAsync(SocketConnection $conn): string {
    string $header = await $conn->readExactlyAsync(4);
    int $length = (\ord($header[0]) << 24) | (\ord($header[1]) << 16) | (\ord($header[2]) << 8) | \ord($header[3]);
    return await $conn->readExactlyAsync($length);
}

async function readUntilClosedAsync(SocketConnection $conn): string {
    string $all = '';
    string $chunk = await $conn->readAsync();
    while ($chunk !== '') {
        $all .= $chunk;
        $chunk = await $conn->readAsync();
    }
    return $all;
}

Redis (Tyhp\Io\Redis)

RedisClient speaks the RESP2 protocol over one connection. Commands from concurrent fibers are pipelined: each is written as soon as you call it, without waiting for earlier replies, and each promise resolves with its own reply in order. A single client can therefore serve a whole application.

An error reply from Redis (for example WRONGTYPE or ERR value is not an integer) rejects only that command with RedisException; the connection stays usable. A transport failure, such as the server closing the connection, rejects every pending command and closes the client.

RedisOptions

<?tyhp
namespace Tyhp\Io\Redis;

type RedisOptions = struct {
    string $host = '127.0.0.1';
    int $port = 6379;
    bool $tls = false;
    ?string $username;
    ?string $password;
    int $database = 0;
    int $connectTimeoutMs = 10000;
};
Field Type Default Meaning
host string '127.0.0.1' Server host name or IP address.
port int 6379 Server port.
tls bool false Connect with TLS and verify the server certificate against host.
username ?string null ACL user name. Used only together with password.
password ?string null When set, the client sends AUTH password, or AUTH username password when username is also set.
database int 0 When not 0, the client sends SELECT database after connecting.
connectTimeoutMs int 10000 Limit for opening the connection, in milliseconds. 0 means no limit.

RedisClient

Redis examples

Using the typed helpers and falling back to commandAsync() for anything else:

<?tyhp

namespace App\Cache;

use Tyhp\Io\Exceptions\RedisException;
use Tyhp\Io\Redis\RedisClient;
use Tyhp\Io\Redis\RedisOptions;

async function sessionDemoAsync(string $password): void {
    RedisClient $redis = await RedisClient::connectAsync(new RedisOptions() with [
        host => 'cache.internal',
        password => $password,
        database => 2,
    ]);
    try {
        await $redis->hSetAsync('session:abc', ['user' => '42', 'role' => 'admin']);
        await $redis->expireAsync('session:abc', 1800);

        array<string, string> $session = await $redis->hGetAllAsync('session:abc');
        echo 'User ' . ($session['user'] ?? '?') . "\n";

        mixed $ttl = await $redis->commandAsync('TTL', 'session:abc');
        if (\is_int($ttl)) {
            echo 'Expires in ' . $ttl . " seconds\n";
        }

        try {
            await $redis->incrAsync('session:abc');
        } catch (RedisException $e) {
            // WRONGTYPE: the connection is still usable afterwards.
            echo 'Redis refused: ' . $e->getMessage() . "\n";
        }
    } finally {
        $redis->close();
    }
}

A fixed-window rate limiter:

<?tyhp

namespace App\RateLimit;

use Tyhp\Io\Redis\RedisClient;

async function allowRequestAsync(RedisClient $redis, string $clientId, int $limitPerMinute): bool {
    string $key = 'rate:' . $clientId . ':' . \intdiv(\time(), 60);
    int $count = await $redis->incrAsync($key);
    if ($count === 1) {
        await $redis->expireAsync($key, 120);
    }
    return $count <= $limitPerMinute;
}

Databases (Tyhp\Io\Db)

The database layer has one interface, DbConnection, and three implementations:

Class Driver Placeholders Non-blocking?
PostgresConnection ext-pgsql, libpq's asynchronous API $1, $2, … Connect and statements
MySqlConnection ext-mysqli with MYSQLI_ASYNC ? Statements; the connection handshake blocks
PdoConnection Any PDO driver, run through Offload ? Only with tyhp/parallel workers

All three share these rules:

  • Statements on one connection run one at a time, in the order you call them. Calls made while a statement is running queue up. Open more connections to run queries concurrently.
  • Each queryAsync() runs one SQL statement. Use transactionAsync() to run several together.
  • Parameters are positional DbValues: int, float, string, bool, or null.
  • Failures reject with DbException, whose getSqlState() returns the five-character SQLSTATE (for example 23505 for a unique violation in PostgreSQL) when the driver reports one.

DbConnection

<?tyhp
namespace Tyhp\Io\Db;

interface DbConnection { /* ... */ }

Type your services against DbConnection when they do not care which driver is behind it. The placeholder syntax still differs between drivers.

DbValue

DbStatement

<?tyhp
namespace Tyhp\Io\Db;

type DbStatement = struct {
    string $sql = '';
    array<int, DbValue> $params = [];
};

One statement for transactionAsync(). params use the connection's placeholder style.

Field Type Default Meaning
sql string '' One SQL statement.
params array<int, DbValue> [] Positional parameter values.

DbResult

The rows and counts from one statement.

PostgresConnection

PostgreSQL over ext-pgsql's asynchronous libpq API. Connecting, sending statements, and waiting for results all run on the event loop. Placeholders are $1, $2, …

Parameters are sent as text and PostgreSQL infers their types from the statement. Where it cannot, such as a bare SELECT $1, add a cast like $1::int. A bool parameter is sent as t/f, and non-finite floats as NaN, Infinity, or -Infinity.

MySqlConnection

MySQL and MariaDB over ext-mysqli with MYSQLI_ASYNC queries. The connection handshake blocks the event loop; statements do not.

Placeholders are ?. They are replaced on the client with escaped literals before the statement is sent: strings are escaped and quoted, bool becomes 1 or 0, and null becomes NULL. A ? inside a quoted string, a backtick-quoted identifier, or a comment (#, -- , /* */) is not a placeholder.

MySqlOptions

<?tyhp
namespace Tyhp\Io\Db;

type MySqlOptions = struct {
    string $host = '127.0.0.1';
    int $port = 3306;
    string $user = 'root';
    string $password = '';
    string $database = '';
    string $charset = 'utf8mb4';
    ?string $socket;
    int $connectTimeoutSeconds = 10;
};
Field Type Default Meaning
host string '127.0.0.1' Server host name or IP address.
port int 3306 Server port.
user string 'root' User name.
password string '' Password.
database string '' Default database. '' selects none.
charset string 'utf8mb4' Connection character set.
socket ?string null Unix socket path. mysqli uses it only when host is localhost; any other host connects over TCP and ignores it.
connectTimeoutSeconds int 10 Connection timeout in seconds.

PdoConnection

Any PDO driver, with each call run through Offload. With tyhp/parallel workers available, statements run in a worker process that keeps one PDO connection per DSN, user name, and password, so the event loop stays free. Without workers, they run in this process and block the loop.

Placeholders are ?, bound in order: null as PDO::PARAM_NULL, bool as PDO::PARAM_BOOL, int as PDO::PARAM_INT, and anything else as a string. PDO::ATTR_ERRMODE is always PDO::ERRMODE_EXCEPTION, and every \PDOException is reported as DbException with the SQLSTATE from PDO's error info.

PdoOptions

<?tyhp
namespace Tyhp\Io\Db;

type PdoOptions = struct {
    string $dsn = '';
    ?string $username;
    ?string $password;
    array<int, int|string|bool> $attributes = [];
};
Field Type Default Meaning
dsn string '' PDO data source name, for example 'sqlite:/var/lib/app/app.sqlite' or 'mysql:host=db;dbname=shop'. Required.
username ?string null User name passed to \PDO.
password ?string null Password passed to \PDO.
attributes array<int, int|string|bool> [] Driver attributes passed to the \PDO constructor, keyed by \PDO::ATTR_* constant.

Database examples

A repository that maps rows to structs, with a transaction:

<?tyhp

namespace App\Orders;

use Tyhp\Io\Db\DbConnection;
use Tyhp\Io\Db\DbResult;
use Tyhp\Io\Db\DbStatement;
use Tyhp\Io\Exceptions\DbException;

type OrderRow = struct {
    int $id = 0;
    string $status = '';
    ?string $note;
};

final class OrderRepository {
    public function __construct(private readonly DbConnection $db) {}

    public async function openOrdersAsync(int $customerId): array<int, OrderRow> {
        DbResult $result = await $this->db->queryAsync(
            'SELECT id, status, note FROM orders WHERE customer_id = $1 AND status <> $2 ORDER BY id',
            [$customerId, 'closed']
        );
        return $result->rowsAs<OrderRow>();
    }

    public async function createAsync(int $customerId, string $note): int {
        DbResult $result = await $this->db->queryAsync(
            'INSERT INTO orders (customer_id, status, note) VALUES ($1, $2, $3) RETURNING id',
            [$customerId, 'open', $note]
        );
        mixed $id = $result->scalar();
        return \is_int($id) ? $id : throw new \UnexpectedValueException('No id returned.');
    }

    public async function transferAsync(int $fromAccount, int $toAccount, int $cents): void {
        try {
            array<int, DbResult> $results = await $this->db->transactionAsync([
                new DbStatement() with [
                    sql => 'UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1',
                    params => [$cents, $fromAccount],
                ],
                new DbStatement() with [
                    sql => 'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
                    params => [$cents, $toAccount],
                ],
            ]);
        } catch (DbException $e) {
            if ($e->getSqlState() === '40001') {
                throw new \RuntimeException('Serialization failure; retry the transfer.', 0, $e);
            }
            throw $e;
        }
        if ($results[0]->affectedRows !== 1) {
            throw new \DomainException('Insufficient funds.');
        }
    }
}

Note that the insufficient-funds check runs after the transaction has committed both updates; in a real system, enforce the rule inside the database (for example with a CHECK (balance >= 0) constraint) so a violation rolls the transaction back.

Running queries in parallel on separate connections:

<?tyhp

namespace App\Reports;

use Tyhp\Io\Db\DbResult;
use Tyhp\Io\Db\PostgresConnection;
use Tyhp\Promise;

async function countsAsync(string $dsn): string {
    PostgresConnection $a = await PostgresConnection::connectAsync($dsn);
    PostgresConnection $b = await PostgresConnection::connectAsync($dsn);
    try {
        array<DbResult> $results = await Promise::all([
            $a->queryAsync('SELECT count(*)::int FROM orders'),
            $b->queryAsync('SELECT count(*)::int FROM customers'),
        ]);
        return 'orders=' . (string) $results[0]->scalar() . ' customers=' . (string) $results[1]->scalar();
    } finally {
        $a->close();
        $b->close();
    }
}

MySQL and PDO use ? placeholders and report generated IDs through lastInsertId:

<?tyhp

namespace App\Catalog;

use Tyhp\Io\Db\DbResult;
use Tyhp\Io\Db\MySqlConnection;
use Tyhp\Io\Db\MySqlOptions;
use Tyhp\Io\Db\PdoConnection;
use Tyhp\Io\Db\PdoOptions;

async function addProductAsync(string $name, float $price): ?string {
    MySqlConnection $db = await MySqlConnection::connectAsync(new MySqlOptions() with [
        host => 'db.internal',
        user => 'shop',
        password => 'secret',
        database => 'shop',
    ]);
    try {
        DbResult $insert = await $db->queryAsync(
            'INSERT INTO products (name, price, active) VALUES (?, ?, ?)',
            [$name, $price, true]
        );
        return $insert->lastInsertId;
    } finally {
        $db->close();
    }
}

async function recordVisitAsync(string $path): void {
    PdoConnection $db = new PdoConnection(new PdoOptions() with [dsn => 'sqlite:/var/lib/app/stats.sqlite']);
    await $db->queryAsync('CREATE TABLE IF NOT EXISTS visits (path TEXT, at INTEGER)');
    await $db->queryAsync('INSERT INTO visits (path, at) VALUES (?, ?)', [$path, \time()]);
}

Offloading blocking work (Tyhp\Io)

Some work blocks no matter what: CPU-heavy computation, a third-party SDK that uses blocking I/O, or a PHP extension without an asynchronous API. Offload runs that work in a tyhp/parallel worker process when one is available, so the event loop keeps serving other fibers. Without workers it runs the job in your process, so the same code works everywhere and only the blocking behavior changes. See With and without tyhp/parallel.

OffloadJob

Offload

Offload example

Hashing passwords without stalling other requests:

<?tyhp

namespace App\Auth;

use Tyhp\Exceptions\ParallelException;
use Tyhp\Io\Offload;
use Tyhp\Io\OffloadJob;

final class HashPasswordJob implements OffloadJob<string> {
    public function __construct(private readonly string $password) {}

    public function run(): string {
        return \password_hash($this->password, \PASSWORD_BCRYPT, ['cost' => 12]);
    }
}

async function hashPasswordAsync(string $password): string {
    try {
        return await Offload::runAsync(new HashPasswordJob($password));
    } catch (ParallelException $e) {
        \error_log('Hashing failed in worker (' . $e->getRemoteClass() . '): ' . $e->getMessage());
        throw $e;
    }
}

DNS (Tyhp\Io\Dns)

Dns

<?tyhp

namespace App\Net;

use Tyhp\Io\Dns\Dns;

async function printAddressesAsync(string $host): void {
    array<int, string> $addresses = await Dns::resolveAsync($host);
    foreach ($addresses as $address) {
        echo $host . ' => ' . $address . "\n";
    }
}

Exceptions (Tyhp\Io\Exceptions)

\RuntimeException
└── IoException
    ├── IoUnavailableException
    ├── HttpException
    │   └── HttpStatusException
    ├── ProcessFailedException
    ├── DbException
    └── RedisException

Catch IoException to handle any failure from this package in one place. Remember that timeouts, cancellation, JSON errors, and struct mismatches use the exception types listed under Errors, which are not IoExceptions.

IoException

IoUnavailableException

HttpException

HttpStatusException

ProcessFailedException

DbException

RedisException

Worked examples

This command reads URLs from a file, checks them with at most eight requests in flight, stops everything after two minutes, and writes a report of broken links.

<?tyhp

namespace App\LinkCheck;

use Tyhp\CancellationToken;
use Tyhp\CancellationTokenSource;
use Tyhp\Exceptions\TimeoutException;
use Tyhp\Io\Exceptions\HttpException;
use Tyhp\Io\File\File;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpRequest;
use Tyhp\Io\Http\HttpResponse;
use Tyhp\Promise;

type LinkStatus = struct {
    string $url = '';
    int $status = 0;
    string $error = '';
};

async function checkLinkAsync(HttpClient $client, string $url, CancellationToken $token): LinkStatus {
    HttpRequest $request = new HttpRequest() with [url => $url, method => 'HEAD', timeoutMs => 10_000];
    try {
        HttpResponse $response = await $client->sendAsync($request, $token);
        return new LinkStatus() with [url => $url, status => $response->status];
    } catch (TimeoutException $e) {
        return new LinkStatus() with [url => $url, error => 'timed out'];
    } catch (HttpException $e) {
        return new LinkStatus() with [url => $url, error => $e->getMessage()];
    } catch (\InvalidArgumentException $e) {
        return new LinkStatus() with [url => $url, error => 'not an http(s) URL'];
    }
}

async function checkLinksAsync(string $listPath, string $reportPath): int {
    array<int, string> $urls = [];
    foreach (await File::linesAsync($listPath) as string $line) {
        string $url = \trim($line);
        if ($url !== '' && !\str_starts_with($url, '#')) {
            $urls[] = $url;
        }
    }

    HttpClient $client = new HttpClient(['User-Agent' => 'link-checker/1.0']);
    $cts := new CancellationTokenSource(120_000);
    CancellationToken $token = $cts->getToken();

    array<int, LinkStatus> $results = await Promise::batch(
        $urls,
        fn(string $url): Promise<LinkStatus> => checkLinkAsync($client, $url, $token),
        8
    );

    string $report = '';
    int $broken = 0;
    foreach ($results as $result) {
        if ($result->error !== '' || $result->status >= 400) {
            $broken++;
            $report .= $result->url . "\t" . ($result->error !== '' ? $result->error : (string) $result->status) . "\n";
        }
    }
    await File::writeAsync($reportPath, $report);
    return $broken;
}

If the two-minute deadline passes, the requests still in flight reject with OperationCancelledException, which checkLinkAsync() does not catch, so checkLinksAsync() fails with it as well.

A cache-aside repository with Redis and PostgreSQL

Reads go to Redis first and fall back to PostgreSQL. Cache problems are logged and never fail the request, because the database remains the source of truth.

<?tyhp

namespace App\Products;

use Tyhp\Io\Db\DbResult;
use Tyhp\Io\Db\PostgresConnection;
use Tyhp\Io\Exceptions\RedisException;
use Tyhp\Io\Redis\RedisClient;
use Tyhp\Json;

type Product = struct {
    int $id = 0;
    string $name = '';
    int $price_cents = 0;
};

final class ProductRepository {
    public function __construct(
        private readonly PostgresConnection $db,
        private readonly RedisClient $cache
    ) {}

    public async function findAsync(int $id): ?Product {
        string $key = 'product:' . $id;
        try {
            ?string $cached = await $this->cache->getAsync($key);
            if ($cached !== null) {
                return Json::decode<Product>($cached);
            }
        } catch (RedisException $e) {
            \error_log('cache read failed: ' . $e->getMessage());
        }

        DbResult $result = await $this->db->queryAsync(
            'SELECT id, name, price_cents FROM products WHERE id = $1',
            [$id]
        );
        ?Product $product = $result->firstAs<Product>();
        if ($product !== null) {
            try {
                await $this->cache->setAsync($key, \json_encode($product, \JSON_THROW_ON_ERROR), 300_000);
            } catch (RedisException $e) {
                \error_log('cache write failed: ' . $e->getMessage());
            }
        }
        return $product;
    }
}

Because price_cents is declared int, the column must be an integer type (int4, int8, …) for firstAs<Product>() to succeed; a numeric column would arrive as a string.

A deploy script

An application entry script that updates a checkout, installs PHP and JavaScript dependencies at the same time, and checks a health endpoint. Any failing step stops the script with a message that includes the command's stderr.

<?tyhp

use Tyhp\Io\Exceptions\ProcessFailedException;
use Tyhp\Io\Http\HttpClient;
use Tyhp\Io\Http\HttpResponse;
use Tyhp\Io\Process\Process;
use Tyhp\Io\Process\ProcessOptions;
use Tyhp\Io\Process\ProcessResult;
use Tyhp\Promise;

async function deployAsync(string $repoDir, string $healthUrl): int {
    ProcessOptions $inRepo = new ProcessOptions() with [cwd => $repoDir, timeoutMs => 300_000];
    try {
        await Process::mustRunAsync(['git', 'pull', '--ff-only'], $inRepo);

        array<ProcessResult> $steps = await Promise::all([
            Process::mustRunAsync(['composer', 'install', '--no-dev', '--no-interaction'], $inRepo),
            Process::mustRunAsync(['npm', 'ci'], $inRepo),
        ]);
        foreach ($steps as $step) {
            echo \implode(' ', $step->command) . " finished\n";
        }
    } catch (ProcessFailedException $e) {
        echo 'Deploy step failed: ' . $e->getMessage() . "\n";
        return 1;
    }

    HttpResponse $health = await HttpClient::shared()->getAsync($healthUrl);
    if (!$health->isSuccess()) {
        echo 'Health check returned ' . $health->status . "\n";
        return 1;
    }
    echo "Deploy complete\n";
    return 0;
}

exit(await deployAsync('/srv/app', 'https://app.example.com/health'));

Gotchas

Connections run one statement at a time

A DbConnection queues statements, so Promise::all() over several queries on one connection runs them one after another. Open a connection per concurrent query (or keep a small pool of connections) when you need parallel database work.

Catching `IoException` does not catch timeouts or cancellation

TimeoutException and OperationCancelledException come from tyhp/async and extend \RuntimeException directly. Catch them separately, or catch \Throwable at the top of a request.

Everything a `RedisClient` does shares one connection

Commands from every fiber are pipelined on the same connection, so connection state is shared. SELECT, MULTI / EXEC, and WATCH sent through commandAsync() affect, or are interleaved with, commands from other fibers. A blocking command such as BLPOP holds up every command queued behind it. Use a separate client for those.

Redis commands have no timeout or token

Only connectAsync() takes a timeout and a token. To give a command a deadline, wrap it: await Promise::timeout($redis->getAsync($key), 500). The command still completes on the server, and its reply is discarded.

`PdoConnection` statements may run in different workers

Do not split a transaction across queryAsync() calls (BEGIN, statements, COMMIT) or rely on session state between calls. Use transactionAsync(). Connections are cached per DSN, user name, and password, so two PdoConnection objects with the same values share a connection, and the attributes of the first one to connect apply.

Without `tyhp/parallel`, PDO, DNS, and `Offload` block

They still work, but the event loop waits for each call. In a server that handles many requests in one process, install tyhp/parallel or prefer PostgresConnection / MySqlConnection, which do not need workers.

Fork the worker pool before opening connections

In a CLI program, the tyhp/parallel pool is forked from your process the first time Offload, PdoConnection, or Dns needs a worker. Call \Tyhp\Parallel::start() at startup, before opening database, Redis, or socket connections, so the workers do not inherit them.

HTTP error statuses resolve

getAsync(), postAsync(), postJsonAsync(), and sendAsync() resolve for 404 and 500 responses. Check isSuccess() or call ensureSuccess(). Only getJsonAsync() checks the status for you.

`ProcessOptions::$env` replaces the environment

Setting env gives the command exactly those variables. Start from \getenv() and add to it when the command needs the rest of your environment.

Entire command output is held in memory

ProcessResult contains the complete stdout and stderr. For commands that produce very large output, redirect it to a file with the command's own options and read that file with File::linesAsync().

Reuse clients

Create one HttpClient per API and one RedisClient per Redis server for your whole process, and share them between fibers. Both are built for concurrent use.

The struct-typed helpers need struct types

getJsonAsync<T>(), json<T>(), readJsonAsync<T>(), rowsAs<T>(), and firstAs<T>() accept a struct type for T. When an API returns a top-level JSON array, read the body with HttpResponse::$body and decode it yourself.