Postgres and Redis

Postgres

ioxide.pg is a Postgres client that lives on a reactor's own io_uring ring. Connecting, authenticating and every query are ring operations, each result resumes the awaiting handler inline on the reactor's thread, and the queries a reactor's handlers send are pipelined over a few connections that reactor owns.

dotnet add package ioxide.pg

It depends on ioxide core alone and speaks the Postgres v3 protocol itself - no Npgsql, no ADO.NET.

A complete program#

An HTTP server that answers GET /users/{id} with a name from Postgres:

using System.Buffers.Text;
using System.Text;
using ioxide;
using ioxide.pg;
using ioxide.utils;

var config = new ServerConfig
{
    ReactorCount = Environment.ProcessorCount,
    Tcp = new TcpOptions { Port = 8080 },
};

var pgOptions = new PgOptions
{
    Host = "127.0.0.1",      // an IPv4 address: the client resolves no names
    Port = 5432,
    User = "app",
    Database = "app",
    Password = "secret",     // SCRAM-SHA-256; null when the server trusts the connection
    PoolSize = 4,            // connections per reactor
};

var threads = new Thread[config.ReactorCount];
for (int i = 0; i < threads.Length; i++)
{
    var reactor = new Reactor(i, config);

    // Runs on the reactor's own thread, so the pool's connections belong to this reactor's ring.
    reactor.OnStart = r => PgPool.Start(r, pgOptions);

    reactor.TcpHandle = async (r, conn) =>
    {
        PgPool pg = r.GetService<PgPool>();
        try
        {
            while (true)
            {
                RecvSnapshot snapshot = await conn.ReadAsync();

                long id = 0;
                while (conn.TryGetItem(snapshot, out SpscRecvRing.Item item))
                {
                    if (!item.HasBuffer) continue;
                    if (id == 0) id = UserId(item.AsSpan());   // read the request line before its buffer goes back
                    conn.ReturnBuffer(in item);
                }

                if (snapshot.IsClosed) return;    // the client has gone: nothing to answer

                string status, body;
                try
                {
                    // Parsed once per connection, then only bound and executed.
                    // The await resumes here, on this reactor's thread.
                    PgResult user = await pg.QueryAsync(
                        "SELECT name FROM (VALUES (1, 'ada'), (2, 'grace')) AS users(id, name) WHERE id = $1",
                        [PgParam.Int(id)]);

                    (status, body) = user.Value is { } name ? ("200 OK", name) : ("404 Not Found", "no such user");
                }
                catch (PgException e)
                {
                    (status, body) = ("500 Internal Server Error", e.Message);
                }

                conn.Write(Encoding.UTF8.GetBytes(
                    $"HTTP/1.1 {status}\r\nContent-Type: text/plain\r\nContent-Length: {Encoding.UTF8.GetByteCount(body)}\r\n\r\n{body}"));
                await conn.FlushAsync();
                conn.ResetRead();
            }
        }
        finally
        {
            conn.DecRef();
        }
    };

    threads[i] = new Thread(reactor.Run) { Name = $"reactor-{i}" };
    threads[i].Start();
}

foreach (Thread thread in threads) thread.Join();

// "GET /users/2 HTTP/1.1" -> 2: the number after the last '/' of the request target.
static long UserId(ReadOnlySpan<byte> request)
{
    int start = request.IndexOf((byte)' ') + 1;
    int length = start > 0 ? request[start..].IndexOf((byte)' ') : -1;
    if (length < 0) return 0;

    ReadOnlySpan<byte> target = request.Slice(start, length);
    return Utf8Parser.TryParse(target[(target.LastIndexOf((byte)'/') + 1)..], out long id, out _) ? id : 0;
}

Start a Postgres for it, run the program, and ask:

docker run --rm -d -p 5432:5432 -e POSTGRES_USER=app -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=app postgres:18

curl http://127.0.0.1:8080/users/1

It answers ada. /users/3 answers 404, and an id too large for an int answers 500 with the server's error. Like the repository's samples, the program answers once per receive.

• One pool per reactor, made in OnStart. PgPool.Start opens the connections on the reactor's ring and registers the pool as a service, found again with GetService<PgPool>(). Use it only from that reactor's thread: it has no locks, by design.
• The await never leaves the thread. The query is written and read on the reactor's ring, and the handler resumes inline once the answer is complete.
• The id travels as a parameter, never spliced into the SQL. A server error is a PgException for that query alone; the connection stays in the pool.

The pool#

A query goes to the pool's next connection in turn, whatever that connection is doing. A connection carries many queries at once - written back to back, answered in order - so nothing is checked out or returned, and a query never waits for a free connection. Postgres runs one connection's queries one after another, though: PoolSize is how many of this reactor's queries the server works on at once, and a slow query delays every query queued behind it on its connection.

• Per reactor. The server sees PoolSize × ReactorCount connections: mind its max_connections, 100 by default.
• Warming up. Start begins all PoolSize connects at once; they complete once the reactor's loop runs. A query that comes first waits for a connection, and fails with pg pool: no connection to {host}:{port} after 10 attempts if none opens. Each failed open is logged to standard error: [pg] connect to {host}:{port} failed: {reason}.
• Healing. A broken connection is dropped when the rotation next reaches it, and a replacement starts opening. On every reactor tick (250 ms) the pool's sweep enforces CommandTimeoutMs and, when no open is under way, opens one connection toward PoolSize - 200 to 600 ms after a failed attempt, so a recovering server is not hammered by every reactor at once.
• Lifetime. A pool has no Dispose; it lives as long as its reactor. A second Start on a reactor replaces the first as the service, so for two databases keep the pools Start returns (services).

Options#

sealed record PgOptions, set with init.

Fixed, not options: each connection's 512 KiB send buffer, which one query - its SQL and parameters - must fit, and a cache of 256 prepared statements per connection.

Connecting and authentication#

A connection opens entirely on the ring: a TCP connect, a startup message carrying the user and the database and nothing else, then authentication. Two methods are supported: trust, and SCRAM-SHA-256 (RFC 7677, without channel binding) - the official Docker images' method once a password is set - with the server's signature checked too. SCRAM's key derivation, PBKDF2 over the iteration count the server asks for, is CPU work on the reactor thread: about 0.7 ms at Postgres's default 4096. A count above 1,000,000 - about 0.2 s of the reactor's time - fails the connect instead: SCRAM: the server asks for 2147483647 iterations, more than the 1000000 this client allows.

Any other method fails the connect with authentication method 3 not supported (trust or SCRAM-SHA-256) - 3 for a cleartext password, 5 for MD5. So does SCRAM without a Password, and the server's own refusals arrive as it sends them: FATAL 28P01: password authentication failed for user "app". There is no TLS - the client never asks for SSL - so keep the connection on loopback or a network you trust. Text goes both ways as UTF-8 and no client_encoding is set: use a UTF8 database.

Opening a connection - the connect, then the startup through authentication to the server's first ready signal - ends at CommandTimeoutMs. A server that never completes the connect, or accepts it and never answers, fails the open with 127.0.0.1:5432 did not accept the connection within 30000 ms or 127.0.0.1:5432 did not answer the startup within 30000 ms, and the pool tries again. The deadline is a timeout of its own on the ring, so it fires on time rather than at the next tick.

Queries and results#

QueryAsync(string sql) sends the SQL as a simple query: no parameters, any number of statements separated by semicolons, parsed and planned on every call. It returns a PgResult:

PgResult count = await pg.QueryAsync("SELECT count(*) FROM users");
// count.Value: "2"   count.Rows: 1   count.CommandTag: "SELECT 1"

PgResult update = await pg.QueryAsync("UPDATE users SET active = false WHERE last_seen < now() - interval '1 year'");
// update.Rows: 0 - no rows came back. update.CommandTag: "UPDATE 17" - the rows it changed.

Every value is Postgres's text form - t and f for booleans, \x0102 for bytea, ISO dates - so parse what you need. There is no binary format and no typed decoding.

Parameters and prepared statements#

QueryAsync(string sql, ReadOnlySpan<PgParam> args) runs SQL with $1, $2, ... placeholders through the extended protocol. A parameter is PgParam.Int(long), PgParam.Text(string) or PgParam.Null. All of them travel as text, and the server converts each to the type it infers for its placeholder - so a date, a numeric or a uuid goes as Text, and a placeholder whose type the server cannot infer is text unless you cast it ($1::int).

PgResult inserted = await pg.QueryAsync(
    "INSERT INTO users (name, email, born) VALUES ($1, $2, $3) RETURNING id",
    [PgParam.Text(name), email is null ? PgParam.Null : PgParam.Text(email), PgParam.Text(born.ToString("yyyy-MM-dd"))]);

long id = long.Parse(inserted.Value!);

Statements are prepared for you. The first time a connection sees an SQL text, it sends a Parse that names the statement (s0, s1, ...) ahead of the Bind and Execute; from then on only the Bind and Execute. Each distinct SQL text is therefore parsed once per connection - once on each of the pool's connections, as queries rotate. Keep the text constant and pass the values as parameters: an SQL string built by interpolation is a new statement every time.

• A connection caches 256 statements. Past that, a new SQL text uses the unnamed statement and is parsed on every call.
• A statement whose Parse failed is forgotten, and one the server reports as gone or changed (SQLSTATE 26000, or 0A000 - "cached plan must not change result type" after a schema change) is dropped. That call fails; the next parses again. A statement whose Bind or Execute fails stays cached: its Parse succeeded, so the server holds it.
• A query takes at most 32 767 parameters.

Streaming rows#

For more than one value, pass a callback. It runs once per row as the row arrives, inline on the reactor, and nothing is collected: a result of any size flows through the connection's 64 KiB receive buffer. Only one of the two forms knows column names:

var names = new List<string>();
int rows = await pg.QueryRowsAsync("SELECT id, name FROM users ORDER BY id", row =>
{
    // The row points into the connection's receive buffer: copy out what you keep.
    names.Add(Encoding.UTF8.GetString(row.Field("name")));
});

var ids = new List<long>();
PgResult result = await pg.QueryAsync(
    "SELECT id, name FROM users WHERE id > $1 ORDER BY id", [PgParam.Int(after)],
    row => ids.Add(long.Parse(row.Field(0))));   // by index: this form has no column names
int rows = result.Rows;

PgRow is a ref struct over the receive buffer, valid only inside the callback. Field(int) returns a field's text bytes - empty for NULL, which IsNull(int) tells apart - and FieldCount counts them. Field(string), IsNull(string) and Ordinal(string) look a column up by name, case-sensitively; Postgres lowercases unquoted names. Columns holds each column's PgColumn: Name, TypeOid and the rest of its description. Every access walks the row from its first field, which matters only for very wide rows.

Errors and timeouts#

A query that fails throws a PgException. One the server reported carries its Severity and SqlState, and a message built from both, with the server's DETAIL and HINT appended:

try
{
    await pg.QueryAsync("INSERT INTO users (id, name) VALUES ($1, $2)", [PgParam.Int(1), PgParam.Text("ada")]);
}
catch (PgException e) when (e.SqlState == "23505")
{
    // unique_violation - e.Message: ERROR 23505: duplicate key value violates unique constraint
    // "users_pkey" DETAIL: Key (id)=(1) already exists. The connection is still in the pool.
}

• A server error fails its query alone. The client reads on to the server's next ready signal, so the connection and the queries behind it carry on.
• A broken connection fails everything in flight on it, with one message and null SqlState: recv: server closed the connection, send failed: errno 32, backend message 'D' of 104857604 bytes exceeds 67108864 bytes. The pool replaces it. Nothing is retried, so after a server restart the first query on each old connection fails.
• A query that does not fit fails before it is sent: query exceeds send buffer (524288 bytes), or pipelined send buffer full when the queries still waiting to go out fill the buffer.

CommandTimeoutMs is checked by the sweep on every tick. A connection whose oldest query in flight was submitted longer ago than that is torn down: everything in flight on it fails with pg command timed out after 30000 ms (3 in flight on 127.0.0.1:5432, oldest sent ~30120 ms ago), and a replacement opens. It fires at the first tick past the deadline, up to 250 ms late, and counts time spent queued behind a slow query; a long report or migration needs a larger value, or 0. Postgres is not asked to cancel the query: the client sends no cancel request.

Transactions#

The pool spreads one handler's queries over its connections, with other handlers' queries between them, so a BEGIN, COMMIT or SET sent through it alone acts on whichever session it lands on. Two ways to run a transaction:

• One simple query. The statements travel as one message and run as one transaction - implicitly so, even without BEGIN. Simple queries take no parameters, so this is for SQL you write yourself:

  await pg.QueryAsync(
      "BEGIN; UPDATE accounts SET balance = balance - 10 WHERE id = 1; " +
      "UPDATE accounts SET balance = balance + 10 WHERE id = 2; COMMIT");

• A connection of your own. PgConnection.ConnectAsync(reactor, options) opens one outside any pool, with the same query methods and an IsBroken flag. It costs a connect and an authentication, has no command timeout and is not replaced if it breaks; use it from one handler at a time:

  PgConnection tx = await PgConnection.ConnectAsync(r, pgOptions);
  try
  {
      await tx.QueryAsync("BEGIN");
      PgResult stock = await tx.QueryAsync("SELECT stock FROM items WHERE id = $1 FOR UPDATE", [PgParam.Int(itemId)]);
      if (int.Parse(stock.Value!) > 0)
      {
          await tx.QueryAsync("UPDATE items SET stock = stock - 1 WHERE id = $1", [PgParam.Int(itemId)]);
      }
      await tx.QueryAsync("COMMIT");
  }
  finally
  {
      tx.Dispose();   // closing the connection rolls back a transaction COMMIT did not end
  }

How a query runs#

QueryAsync writes the query's messages into the connection's send buffer, queues a waiter, and starts the connection's sender and reader if they are idle; it returns before anything is sent. The sender sends everything staged in one RingSocket send - a submission on the reactor's ring that the loop hands the kernel with its next io_uring_enter - and queries staged meanwhile go out together in the next. The reader receives into the connection's buffer, walks the messages, hands rows to the waiter at the front of the queue, and completes it at the server's ready signal. That waiter is an IValueTaskSource that neither defers its continuation nor posts it to the reactor's synchronization context, so the handler resumes right there, inside the reader, before the next message is parsed (completions resume inline). No thread pool and no lock on the way: a connection belongs to its reactor's thread.

Measured#

The repository's benchmark run of 9 August 2026 (bench/results/20260809T181559Z.json, commit 1f2e52c, 2 reactors, wrk with 64 connections, one machine) has the Clients/Pg sample answering every request with SELECT 42 over 4 connections per reactor: 337k requests per second at 5.97 us of the server's CPU per request. Tcp/Raw, the baseline with no I/O beyond the socket, used 2.30 us in the same run. The CPU is the sample process's alone; Postgres's own is not counted.

Limits#

• No TLS, no names, no IPv6; trust and SCRAM-SHA-256 only.
• Text format only, and parameters are Int, Text or Null.
• No COPY, no LISTEN/NOTIFY, no cancel requests; a query returns all its rows. No retry, and no way to stop a pool.

Where to go next#