C++20 · optional C++23 · Apache-2.0

Actors that share nothing. I/O that never blocks.

qb is a C++20-first framework for concurrent and distributed systems on the actor model. An actor is thread-affine to one core, processes one event at a time, and talks only by message — so data races and deadlocks stop being possible rather than being avoided.

Install qb View on GitHub

greeter.cpp

#include <qb/main.h>
#include <qb/actor.h>
#include <qb/io.h>

struct GreetingEvent : qb::Event {
    qb::string<64> message;
    explicit GreetingEvent(const char *msg) : message(msg) {}
};

class GreeterActor : public qb::Actor {
public:
    qb::io::async::task<bool> onInit() final {
        registerEvent<GreetingEvent>(*this);  // subscribe
        push<GreetingEvent>(id(), "Hello");  // send to self
        co_return true;  // actor is ready
    }

    void on(const GreetingEvent &event) {
        qb::io::cout() << "Received: " << event.message << '\n';
        kill();  // work done; terminate this actor
    }
};

int main() {
    qb::Main engine;
    engine.addActor<GreeterActor>(0);  // run on core 0
    engine.start();  // start the engine
    engine.join();  // block until all actors stop
    return 0;
}

qb-core: Actor engine — scheduling, mailboxes, lifecycle
qb-io: Async runtime — libev loop, transports, coroutines
Standalone: qb-io is usable without the actor engine
3 platforms: Linux · macOS · Windows, on x86-64 and ARM64

The contract

Five rules decide whether qb code is correct.

Everything else is detail. They come from the framework's own code-generation guidance, and they are the fastest way to understand what qb guarantees and what it asks of you in return.

Time is std::chrono
Every timeout, TTL, interval and delay is a qb::duration — std::chrono::nanoseconds. It accepts finer-or-equal chrono literals and rejects bare integers at compile time. Subtracting a steady clock from a system clock does not compile.
qb::duration
Actors share nothing
An actor is thread-affine to one VirtualCore, processes one event at a time, and communicates only by events. No mutexes, no shared mutable state, and no actor is ever constructed outside a worker thread.
qb::VirtualCore
Sending is noexcept
push, send and broadcast are noexcept, so a throwing event constructor or an allocation failure calls std::terminate. Events are relocated with memcpy: no member may point into its own storage, which is why a by-value std::string is never a valid event member.
push / send / broadcast
A coroutine captures by value
The actor can be destroyed while the coroutine is suspended. After any suspension the only legal channel back is the context. Prefer Actor::spawn, bound to the actor's cancellation scope, over spawn_detached.
Actor::spawn
Register every event you handle
Call registerEvent for each event type inside onInit(), which is itself a coroutine returning a task. Returning false or throwing fails creation and yields an invalid ActorId.
registerEvent<T>(*this)

The engine

One worker thread per core, one mailbox each.

qb::Main launches a VirtualCore per engine core. Each worker owns its actors and drains a private lock-free mailbox. The only genuinely multi-threaded surface in the whole system is the message-passing layer between cores — which is why placement is a decision you make, not one the runtime hides.

Same-core delivery never leaves the thread. Only cross-core delivery touches the ring.

Same core

The event is appended to the local pipe and consumed in the same loop iteration, after the current handler returns. No mailbox, no inter-thread synchronisation.

Across cores

The event is published into the destination core's MPSC ring and dequeued on its next iteration. This is the only path that touches lock-free machinery.

Idle latency is a knob

A mailbox latency of 0 busy-spins for the lowest possible latency and occupies a core. Above 0, the consumer parks on a condition variable and a producer wakes it — less CPU at idle, up to one span of wake-up delay.

Messaging

Five primitives, five different contracts.

Actors communicate only by typed events. The primitives differ in ordering, in whether they allocate a new event or redirect the one you received, and in what the runtime may do under backpressure. When unsure, the answer is push.

Five primitives, five different contracts.
Primitive	Ordering	Event object	Notes
`push`	FIFO per source→dest	New	The default. Constructed at the back of the pipe; returns a mutable reference so the sender can finish populating it.
`send`	None	New	Constructed at the front. Requires a trivially destructible event; the engine may drop EventQOS0 under backpressure.
`reply`	Redirects one event	Reused	Sends the received event back to its source. The handler must take a non-const reference.
`forward`	Redirects one event	Reused	Redirects the received event to a new destination, preserving the original source.
`broadcast`	None	New	Fans out to every actor on every core via send, so it inherits send's lack of ordering.

In practice

A server, a client, a coroutine.

Network endpoints are actors with a mixin: I/O runs on the actor's own event loop and parsed messages arrive as ordinary handler calls. Nothing is registered globally, and nothing blocks a core.

TCP server

An acceptor armed on the actor's own core; each connection becomes a session actor.

#include <qb/actor.h>
#include <qb/io/async.h>
#include <qb/io/uri.h>
#include "MyClientSession.h"   // a use<MyClientSession>::tcp::client<MyServer>

class MyServer : public qb::Actor,
                 public qb::io::use<MyServer>::tcp::server<MyClientSession> {
public:
    explicit MyServer(qb::io::uri listen_at) : _listen_at(std::move(listen_at)) {}

    qb::io::async::task<bool> onInit() override {
        // transport().listen(uri) returns 0 on success (non-zero is an error).
        if (this->transport().listen(_listen_at) != 0) {
            qb::io::cerr() << "listen failed on " << _listen_at.source() << "\n";
            co_return false;
        }
        this->start();          // arm the acceptor on this core's event loop
        co_return true;
    }

private:
    qb::io::uri _listen_at;
};

TCP client

A protocol frames the byte stream; disconnection schedules a guarded retry.

class ClientActor : public qb::Actor,
                    public qb::io::use<ClientActor>::tcp::client<> {
public:
    // The framer reads this alias; the base constructor activates it automatically.
    using Protocol = chat::ChatProtocol<ClientActor>;

    qb::io::async::task<bool> onInit() override {
        connect();
        co_return true;
    }

    // A parsed message arrives as an ordinary handler call.
    void on(const chat::Message &msg) {
        qb::io::cout() << "server: " << msg.payload << "\n";
    }

    // Connection loss: reset state and schedule a delayed retry.
    void on(qb::io::async::event::disconnected const &) {
        _connected = false;
        if (_should_reconnect)
            // The timer holds no claim on this actor — guard re-entry.
            qb::io::async::callback([this] { if (is_alive()) connect(); }, RECONNECT_DELAY);
    }
};

Coroutine

Captured by value, bound to the actor's cancellation scope, talking back through its context.

qb::io::async::task<bool> onInit() final {
    registerEvent<PriceTick>(*this);

    // spawn() binds the coroutine to this actor's cancellation scope: if the actor dies
    // while the coroutine is suspended, the scope cancels it.
    spawn([id = id(), symbol = _symbol](auto ctx) -> qb::io::async::task<void> {
        // Everything the coroutine needs was captured by value before the first co_await.
        for (;;) {
            co_await qb::io::async::sleep(std::chrono::seconds(1));

            // After a suspension the context is the only legal channel back to the actor.
            ctx.push_to<PriceTick>(id, symbol, ctx.time());
        }
    });                 // no trailing () — that would destroy the closure before it starts

    co_return true;
}

Events

Fixed-capacity members, because events are relocated with memcpy.

struct UpdateConfig : qb::Event {
    qb::string<64>  key;      // fixed capacity: events are relocated with memcpy,
    qb::string<256> value;    // so no member may point into its own storage
    int             priority;

    UpdateConfig(const char *k, const char *v, int p)
        : key(k), value(v), priority(p) {}
};

void Producer::on(const TickEvent &) {
    // push constructs the event in place at the back of the destination pipe and hands
    // back a reference, so the sender can finish populating it.
    auto &cmd = push<UpdateConfig>(_target, "timeout_ms", "500", 1);
    cmd.priority = 2;
}

qb-io

The runtime works without the actors.

qb-io is a single-threaded, libev-backed event loop with non-blocking transports, a message-framing protocol layer and a native C++20 coroutine layer. qb-core is built on it, and qb-io can be taken alone. Application protocols like HTTP live in modules, not here.

Transports

TCP
UDP
SSL/TLS
QUIC

Protocols

text
binary8/16/32
JSON
MessagePack
AProtocol

Coroutines

task<T>
channels
generators
when_all
retry

Utilities

crypto & JWT
compression
URI
UUID
lock-free queues

Standalone

The runtime on its own

Every sample here is a plain executable that drives its own loop: no actor, no engine. That is the claim the section makes, so it is the one worth showing.

Event loop

A delayed callback owns and frees its own watcher; an inactivity deadline is a CRTP mixin, not a timer you arm by hand.

#include <qb/io/async.h>
#include <qb/system/time.h>

using namespace std::chrono_literals;

int main() {
    qb::io::async::init();  // one event loop per thread

    // A delayed callback arms a one-shot watcher that frees itself after it fires.
    // Reschedule from inside it and you have a periodic task with no timer object
    // to own. A thread-local freelist backs the allocation, so steady callback
    // traffic performs no malloc.
    std::function<void()> tick = [&] {
        std::cout << "tick at " << qb::to_iso8601(qb::wall_now()) << '\n';
        qb::io::async::callback([&] { tick(); }, 1s);
    };
    qb::io::async::callback([&] { tick(); }, 1s);

    qb::io::async::run();  // drive the loop
}

// An inactivity deadline is a CRTP mixin, not a timer you arm and cancel by hand.
class IdleHandler : public qb::io::async::with_timeout<IdleHandler> {
public:
    explicit IdleHandler(qb::duration timeout = 100ms) : with_timeout(timeout) {}

    void on(qb::io::async::event::timer const &) {
        // The deadline passed: close the connection, fail the operation.
    }

    void on_activity() {
        updateTimeout();  // records now as the last activity; no re-arm per byte
    }
};

TCP server

A server, its sessions and a framer, in thirty lines. The Protocol alias is what turns a byte stream into on(message).

#include <qb/io.h>
#include <qb/io/async.h>
#include <qb/io/protocol/text.h>

// One session per connected client. The Protocol alias is the framer: the base
// reads it, cuts the stream into messages, and calls on() with each one.
class Session : public qb::io::use<Session>::tcp::client<class EchoServer> {
public:
    using Protocol = qb::protocol::text::command<Session>;  // newline-framed text

    explicit Session(IOServer &server) : client(server) {}

    void on(Protocol::message &&msg) {
        *this << "echo: " << msg.text << Protocol::end;  // reply through the buffer
    }
};

class EchoServer : public qb::io::use<EchoServer>::tcp::server<Session> {
public:
    void on(IOSession &) {}  // called once per new session
};

int main() {
    qb::io::async::init();

    EchoServer server;
    server.transport().listen_v4(8888);  // bind + listen
    server.start();                      // begin accepting

    while (true)
        qb::io::async::run(EVRUN_NOWAIT);
}

Coroutines

The task type, sleep, scatter-gather, first-to-finish with the losers reclaimed, a deadline wrapper, and channels between coroutines.

#include <qb/io/async/coroutine.h>

using namespace qb::io::async;
using namespace std::chrono_literals;

task<int> compute_value(int base) {
    co_await sleep(100ms);  // suspends; the loop runs other work meanwhile
    co_return base * 2;
}

task<void> root() {
    // Scatter-gather. Heterogeneous tasks come back as a tuple.
    auto [value, name] = co_await when_all(compute_value(21), fetch_name());

    // First to finish wins, and the losers are reclaimed — the scheduler tears
    // each one down, so nothing keeps running in the background.
    auto winner = co_await when_any(fast(), slow(), backup());
    int v = winner.get<int>();

    // A deadline is a wrapper around a task, not a flag on every call.
    try {
        auto data = co_await coro_with_timeout(fetch_data(), 2s);
    } catch (qb::io::async::timeout_error const &) {
        // The deadline elapsed. The inner task finishes on its own; its result is dropped.
    }

    // Channels hand values between coroutines on the one thread. Capacity 0 is
    // a rendezvous: the send and the recv meet with no buffering.
    channel<int> ch(16);
    co_await ch.send(42);
    std::optional<int> got = co_await ch.recv_for(200ms);  // empty on timeout
}

int main() {
    coro_scheduler().spawn(root());  // the scheduler is the listener's
    run_for(2s);
}

Custom protocol

Three functions: is a whole message here, take it, reset. The framework removes the bytes after onMessage returns.

// A protocol answers one question — "is a whole message in the buffer yet?" —
// and then hands it over. The framework removes the bytes for you.
template <typename IO_>
class custom_protocol : public qb::io::async::AProtocol<IO_> {
    static constexpr std::size_t HEADER_SIZE = sizeof(MessageHeader);
    bool          _reading_header = true;
    MessageHeader _header{};

public:
    using message = custom_message;  // what on() receives

    explicit custom_protocol(IO_ &io) noexcept : qb::io::async::AProtocol<IO_>(io) {}

    std::size_t getMessageSize() noexcept final {
        auto &buffer = this->_io.in();

        if (_reading_header) {
            if (buffer.size() < HEADER_SIZE)
                return 0;  // not enough bytes yet; called again on the next read
            std::memcpy(&_header, buffer.cbegin(), HEADER_SIZE);
            if (_header.magic != MAGIC || _header.version != VERSION) {
                this->not_ok();  // unrecoverable: the connection is dropped
                return 0;
            }
            _reading_header = false;
        }

        const std::size_t total = HEADER_SIZE + _header.length;
        return buffer.size() < total ? 0 : total;
    }

    void onMessage(std::size_t size) noexcept final {
        const char *base = this->_io.in().cbegin();

        message msg;
        msg.type = _header.type;
        msg.id   = _header.id;
        if (_header.length > 0)
            msg.payload.assign(base + HEADER_SIZE, _header.length);

        this->_io.on(std::move(msg));  // dispatch to the session handler
        _reading_header = true;
    }

    void reset() noexcept final {
        _reading_header = true;
        _header = {};
    }
};

TLS

The same server one alias deeper. The handler, the protocol and the send operator do not change.

#include <qb/io/async.h>
#include <qb/io/protocol/text.h>
#include <qb/io/tcp/ssl/socket.h>  // qb::io::ssl::create_server_context

using namespace qb::io;

class SecureServer;

// Compare with the plain TCP session: the only change is ssl:: in the alias.
// The handler, the protocol and the send operator are identical.
class SecureSession : public use<SecureSession>::tcp::ssl::client<SecureServer> {
public:
    using Protocol = qb::protocol::text::command<SecureSession>;

    explicit SecureSession(IOServer &server) : client(server) {}

    void on(Protocol::message &&msg) {
        *this << msg.text << Protocol::end;  // echo back, still encrypted
    }
};

class SecureServer : public use<SecureServer>::tcp::ssl::server<SecureSession> {
public:
    void on(IOSession &) {}  // a new secure session was established
};

void run_server(std::filesystem::path const &cert, std::filesystem::path const &key) {
    async::init();

    SecureServer server;
    // The listener takes ownership of the SSL_CTX and frees it on destruction.
    server.transport().init(ssl::create_server_context(TLS_server_method(), cert, key));
    server.transport().listen_v4(64384);
    server.start();

    while (true)
        async::run(EVRUN_ONCE);
}

Utilities

Hashing, AEAD, JWT, compression, URI and UUID: the pieces a network service needs before it can start.

#include <qb/io/crypto.h>
#include <qb/io/crypto_jwt.h>
#include <qb/io/compression.h>
#include <qb/io/uri.h>

// Hashing and encoding
std::string digest = qb::crypto::sha256("payload");  // 64-char lowercase hex
std::string b64    = qb::crypto::base64_encode(data.data(), data.size());

// Authenticated encryption. An empty plaintext means the GCM tag did not verify —
// treat it as an authentication failure, never as "decrypted to nothing".
using C  = qb::crypto;
auto key = C::generate_key(C::SymmetricAlgorithm::AES_256_GCM);
auto iv  = C::generate_iv(C::SymmetricAlgorithm::AES_256_GCM);
auto ct  = C::encrypt(plaintext, key, iv, C::SymmetricAlgorithm::AES_256_GCM, aad);
auto pt  = C::decrypt(ct, key, iv, C::SymmetricAlgorithm::AES_256_GCM, aad);

// JSON Web Tokens
qb::jwt::CreateOptions opt;
opt.algorithm = qb::jwt::Algorithm::HS256;
opt.key       = "my-256-bit-secret";

std::string token = qb::jwt::create_token(claims, "issuer", "user-12345", "audience",
                                          std::chrono::seconds(3600),
                                          std::chrono::seconds(0), "token-id", opt);
qb::jwt::ValidationResult result = qb::jwt::verify(token, vopt);

// Compression, one-shot or streaming
std::string gz = qb::gzip::compress(original.c_str(), original.size());

// RFC 3986 URIs
qb::io::uri u{"https://www.example.com:8080/a/b?query1=value1#frag"};
auto host             = u.host();            // "www.example.com"
const std::string &v1 = u.query("query1");   // "value1"

// UUID
qb::uuid id = qb::generate_random_uuid();  // version 4

Modules

The protocols ship as modules, not as core.

Each is its own repository, added as a submodule and discovered by CMake. They are compiled libraries built on qb-io, so they run on the same event loop as your actors and never block a core.

qbm-http

Stable

Verified against qb 3.0.0

HTTP servers and clients as actors: a fluent router with path parameters, groups and controllers, a middleware pipeline, request validation, cookies and multipart, and both callback and coroutine client APIs.

HTTP/1.1
HTTP/2
HTTP/3
WebSocket
routing
middleware
validation
JWT

HTTP/1.1, routing, middleware and validation are always built. HTTPS, HTTP/2, WebSocket and JWT need an SSL build; HTTP/3 additionally needs QUIC and libnghttp3. The gates are real #ifdef boundaries, so what your code can reach is what was compiled.

Umbrella header: <qbm/http/http.h>

View repository →

qbm-pgsql

Stable

Verified against qb 3.0.0

The PostgreSQL v3 wire protocol spoken directly over a qb-io socket. The handshake, authentication, the simple and extended query protocols, type encoding and LISTEN/NOTIFY are all implemented on the event loop.

no libpq
SCRAM-SHA-256
prepared statements
transactions
savepoints
LISTEN/NOTIFY
typed results

timestamptz round-trips through qb::wall_time at integer-microsecond precision. TLS is available on an SSL build. The client does not require actors — a plain executable that drives the loop works.

Umbrella header: <qbm/pgsql/pgsql.h>

View repository →

qbm-redis

Stable

Verified against qb 3.0.0

A native RESP2 and RESP3 parser over a qb-io socket. Every command group is a mixin on the client: strings, hashes, sets, sorted sets, streams, scripting, ACL and cluster.

no hiredis
RESP2 / RESP3
full command surface
pipelining
pub/sub
streams
scripting
cluster

Auto-reconnect uses exponential backoff, and a command deadline drops a connection whose reply never arrives. Like qbm-pgsql, it runs without actors if you drive the loop yourself.

Umbrella header: <qbm/redis/redis.h>

View repository →

In practice

What each one actually looks like

Every sample below is taken from the module’s own documentation, which is generated against its headers, so what you read here is what compiles.

qbm-http

One router, four transports. A handler written for HTTP/1.1 serves HTTP/2 and HTTP/3 unchanged, because every server feeds the same Router.

Server

An actor that is also a server. Routes go on router(), compile() builds the trie once, and each handler completes its own context.

#include <qbm/http/http.h>
#include <qb/main.h>

class ApiServer : public qb::Actor, public qb::http::Server<> {
public:
    qb::io::async::task<bool> onInit() override {
        router().get("/", [](auto ctx) {
            ctx->response().body() = "Hello from qbm-http";
            ctx->complete();  // a handler that never completes hangs until the timeout
        });

        router().get("/users/:id", [](auto ctx) {
            ctx->response().status() = qb::http::status::OK;
            ctx->response().body()   = "user " + ctx->path_param("id");
            ctx->complete();
        });

        router().compile();  // required after the last route, before serving

        if (listen({"tcp://0.0.0.0:8080"})) {
            start();
            co_return true;
        }
        co_return false;
    }
};

Client

The one-shot verbs return an awaiter. run_sync drives the same call from code that is not a coroutine.

#include <qbm/http/http.h>

// Inside a coroutine the call reads top to bottom: request out, response back,
// error handling is a plain if. No callback, no extra thread, no second loop.
qb::io::async::task<void> fetch() {
    qb::http::Request req(qb::io::uri("http://api.example.com/data"));
    req.add_header("Accept", "application/json");

    using namespace std::chrono_literals;
    auto reply = co_await qb::http::GET(std::move(req), 5s);  // timeout is a qb::duration

    if (reply.response.status() == qb::http::status::OK)
        handle(reply.response.body().as<std::string_view>());
}

// The same call from ordinary code: run_sync drives the loop until it resolves.
int main() {
    qb::io::async::init();

    auto reply = qb::http::run_sync(
        qb::http::GET(qb::http::Request{qb::io::uri("http://api.example.com/data")}));

    std::cout << reply.response.body().as<std::string_view>() << '\n';
}

HTTP/2

TLS-only with ALPN h2. The client multiplexes over one connection and connects on first use.

#include <qbm/http/http.h>  // pulls <qbm/http/2/http2.h> under QB_HAS_SSL

// Server. Routing is identical to HTTP/1.1 — the HTTP/2 session feeds the same
// Router, so groups, controllers, middleware and validation work unchanged.
auto server = qb::http2::make_server();

server->router().get("/hello", [](auto ctx) {
    ctx->response().body() = "Hello over HTTP/2";
    ctx->complete();
});
server->router().compile();

// listen() builds the TLS context, sets ALPN to {"h2","http/1.1"}, and binds.
server->listen({"https://0.0.0.0:8443"}, "cert.pem", "key.pem");
server->start();

// Client. Requests multiplex over one connection, and it connects lazily —
// push_request establishes the connection on first use.
auto response = qb::http::run_sync([]() -> qb::io::async::task<qb::http::Response> {
    auto client = qb::http2::make_client("https://api.example.com");
    co_return co_await client->push_request(
        qb::http::Request{qb::io::uri("https://api.example.com/data")});
}());

HTTP/3

QUIC transport, ALPN h3. Adds a cancellable request id and a cap on the reassembled body.

#include <qbm/http/http.h>

#ifdef QBM_HTTP_HAS_HTTP3  // set only when SSL, QUIC and libnghttp3 are all present

// Server. The transport is UDP/QUIC and ALPN is locked to {"h3"}; the router,
// the context and the handlers are the ones you already wrote.
auto server = qb::http3::make_server();

server->router().get("/ping", [](auto ctx) {
    ctx->response().set_header("x-protocol", "HTTP/3");
    ctx->response().body() = "pong-h3";
    ctx->complete();
});
server->router().compile();
server->listen(qb::io::uri("https://0.0.0.0:4433"), "cert.pem", "key.pem");

// Client. Same shape as HTTP/2, plus two things only this one has: a request id
// you can cancel, and a cap on the reassembled body (64 MiB by default).
auto client = qb::http3::make_client("https://127.0.0.1:4433");
auto id = client->push_request_with_id(qb::http::Request{qb::io::uri("/slow")},
                                       [](qb::http::Response res) { handle(res); });
client->cancel_request(id);  // completes with CLIENT_CLOSED_REQUEST

#endif

Dual stack

h2 and h3 cannot share a socket, so the facade runs both and registers each route on both routers.

// HTTP/2 is TCP/TLS and HTTP/3 is UDP/QUIC, so they cannot share a socket.
// The facade registers every route on both routers and compiles both.
auto server = qb::http::make_dual_stack_server();

server->router().get("/shared", [](auto ctx) {
    // Same handler for both protocols; ask which one served this request.
    const bool over_h3 = ctx->request().major_version == 3;
    ctx->response().set_header("x-version", over_h3 ? "h3" : "h2");
    ctx->response().body() = over_h3 ? "shared-over-h3" : "shared-over-h2";
    ctx->complete();
});
server->router().compile();  // compiles BOTH routers

// Two URIs — TCP/TLS first, then UDP/QUIC — and one certificate pair. The bind
// is all or nothing: if one stack comes up and the other fails, the one that
// came up is rolled back, so you never listen on half a server.
server->listen(qb::io::uri("https://0.0.0.0:443"),
               qb::io::uri("https://0.0.0.0:443"),
               "cert.pem", "key.pem");

Choosing a client

The three persistent clients share the connect and push_request shape. These are the rows that decide which one you want.

Choosing a client
Capability	http1	http2	http3
Compile gate	none	QB_HAS_SSL	QBM_HTTP_HAS_HTTP3
Transport (ALPN)	TCP / TLS	TLS only (h2)	QUIC (h3)
Base URI scheme	http:// or https://	https:// only	https:// only
Concurrency	one active, rest queued	multiplexed	multiplexed

qbm-pgsql

The wire protocol, not a wrapper around libpq. Every call has a coroutine form and a callback form under the same method name.

Query

Parameters bind by position with deduced types, and rows decode into tuples, so structured bindings work directly.

#include <qbm/pgsql/pgsql.h>

using namespace qb::pg;

qb::io::async::task<void> read(tcp::database &db, int user_id) {
    // One-shot parameterized query: no statement name, no OID list. The
    // parameter types are deduced from the C++ arguments, and it runs through
    // the unnamed prepared statement, so it does not pollute the cache.
    auto r = co_await db.query("SELECT id, name FROM users WHERE id = $1", user_id);
    if (!r) co_return;  // Reply<T> carries either the result or a typed db_error

    // A row decodes straight into a tuple, so a structured binding just works.
    for (auto [id, name] : r.result().all<int, std::string>())
        qb::io::cout() << id << ' ' << name << '\n';

    // rows<>() is the same thing lazily — no intermediate vector.
    for (auto [id, name] : r.result().rows<int, std::string>())
        use(id, name);

    // First row, or nothing. std::optional<T> in the tuple reads a nullable column.
    if (auto one = r.result().one<int, std::optional<std::string>>()) {
        auto [id, maybe_name] = *one;
    }
}

Prepared

The extended-query protocol with a client-side LRU, so a named statement costs no round trip after the first.

// Prepare once, execute many. The client keeps a prepared-statement LRU, so the
// name resolves without a round trip after the first use.
auto pr = co_await db.prepare("by_email",
                              "SELECT id, name FROM users WHERE email = $1",
                              qb::pg::type_oid_sequence{qb::pg::oid::text});
if (!pr.ok()) co_return;

auto r = co_await db.execute("by_email", qb::pg::params{email});

// The callback form is the same call with handlers, for code that cannot be a
// coroutine. Both forms drain on the same event loop.
db.execute("by_email", qb::pg::params{email},
           [](qb::pg::transaction &, qb::pg::results rs) { use(rs); },
           [](qb::pg::error::db_error const &e) { report(e); });

Transaction

with_transaction wraps BEGIN and COMMIT; throwing transaction_abort from the body rolls back and keeps the error.

using namespace qb::pg;

qb::io::async::task<void> transfer(tcp::database &db) {
    // BEGIN, run the body, COMMIT — or ROLLBACK when the body throws.
    auto tx = co_await with_transaction(db, [](transaction &tr)
                                                -> qb::io::async::task<void> {
        auto debit = co_await tr.execute(
            "UPDATE account SET balance = balance - $1 WHERE id = $2", 100, 1);
        if (!debit)
            throw transaction_abort{debit.error()};  // rolls back, keeps the error

        auto credit = co_await tr.execute(
            "UPDATE account SET balance = balance + $1 WHERE id = $2", 100, 2);
        if (!credit)
            throw transaction_abort{credit.error()};
    });

    if (!tx.ok())
        qb::io::cerr() << "rolled back: " << tx.error().what() << '\n';

    // Savepoints nest inside a block, and a statement timeout is a qb::duration
    // armed on the next BEGIN.
    db.set_timeout(std::chrono::seconds{5});
}

Large results

A server-side cursor and the COPY paths, both in constant memory whatever the result size.

// query() buffers the whole result set. query_stream runs the statement through a
// server-side CURSOR and holds exactly one batch, whatever the result size.
std::uint64_t total = 0;
co_await db.query_stream("SELECT amount FROM ledger", 1000,
                         [&](auto row) { total += row[0].template as<long>(); });

// COPY is PostgreSQL's bulk path, and copy_out streams it out in constant memory:
// each CopyData chunk reaches the sink as it arrives.
std::ofstream f("users.csv");
co_await db.copy_out("COPY users TO STDOUT (FORMAT csv)",
                     [&](std::string_view chunk) { f.write(chunk.data(), chunk.size()); });

// Bulk in: return std::nullopt from the source to finish.
co_await db.copy_in("COPY t (id, v) FROM STDIN",
                    [&]() -> std::optional<std::string> { return next_line(); });

LISTEN / NOTIFY

The database pushes; the handler runs on the I/O thread when the notification arrives. Nothing polls.

// Publisher: NOTIFY is a statement like any other.
co_await db.notify("jobs", R"({"id":42})");

// Subscriber: register the handler, then LISTEN. The handler runs on the I/O
// thread when a NotificationResponse arrives — there is nothing to poll.
db.on_incoming_notify([](qb::pg::notification &&n) {
    qb::io::cout() << n.channel << ": " << n.payload
                   << " (backend " << n.server_backend_pid << ")\n";
});

(void)co_await db.listen("jobs");
(void)co_await db.unlisten("jobs");

qbm-redis

A native RESP parser and the full command surface, each group a mixin on the client. Redis errors arrive as a false reply, never as an exception.

Basics

Connect and run a command. The reply type follows the command: an optional for GET, a count for DEL.

#include <qbm/redis/redis.h>
#include <qb/io/async.h>

int main() {
    qb::io::async::init();

    qb::redis::tcp::client redis{qb::io::uri{"tcp://localhost:6379"}};
    if (!qb::io::async::run_sync(redis.connect()))
        return 1;

    // SET returns Reply<status>; operator bool is true on success.
    auto set_r = qb::io::async::run_sync(redis.set("greeting", "Hello Redis!"));
    if (!set_r)
        return 1;

    // GET returns Reply<std::optional<std::string>> — the optional is the NULL,
    // and a Redis error is ok() == false, never an exception.
    auto value = qb::io::async::run_sync(redis.get("greeting")).value_or("");
    qb::io::cout() << "Retrieved: " << value << '\n';

    qb::io::async::run_sync(redis.del("greeting"));  // Reply<long long>
    return 0;
}

Data types

Hashes, sorted sets and streams, decoded into real types. There is an escape hatch for a command that is not wrapped yet.

// Hash
co_await redis.hset("user:1", "email", "ada@example.com");
auto email = co_await redis.hget("user:1", "email");  // Reply<optional<string>>

// Sorted set — a leaderboard, then a blocking pop with a real timeout
std::vector<qb::redis::score_member> members = {
    {10.0, "member1"}, {20.0, "member2"}, {30.0, "member3"}};
co_await redis.zadd("board", members);

auto top = co_await redis.bzpopmin({"board"}, std::chrono::seconds{5});
if (top.ok() && top.result().has_value()) {
    auto const &[key, member, score] = *top.result();
}

// Stream — XADD gives back a typed id, XRANGE gives back decoded entries
std::vector<std::pair<std::string, std::string>> entries = {
    {"field1", "value1"}, {"field2", "value2"}};
auto id = co_await redis.xadd("mystream", entries);  // Reply<stream_id>

auto range = co_await redis.xrange("mystream", "-", "+");
for (auto const &entry : range.result())
    qb::io::cout() << entry.id.to_string() << " has "
                   << entry.fields.size() << " fields\n";

// Not wrapped yet? Name the command and pick its decode type.
auto n = co_await redis.command<long long>("EVAL", "return 1", "0");

Transaction

WATCH, MULTI, EXEC — optimistic concurrency that aborts rather than applying a decision taken on stale data.

// WATCH marks the key. If anyone else writes it before EXEC, the whole batch is
// aborted instead of applying a decision taken on stale data.
auto w = co_await redis.watch("balance");
if (!w.ok()) co_return;

auto current = (co_await redis.get("balance")).value_or("0");
auto next = recompute(current);

co_await redis.multi();               // is_in_multi() is now true
(void)co_await redis.set("balance", next);  // queued server-side, acked QUEUED

auto ex = co_await redis.exec<std::string>();  // Reply<std::vector<std::string>>
if (!ex.ok()) {
    // Someone changed "balance" after WATCH. Nothing was applied — retry the block.
}

Pub/sub

A consumer for messages and a separate client to publish, because publish() is not on the consumer.

#include <qbm/redis/redis.h>
#include <qb/io/async.h>

qb::io::async::task<void> chat() {
    // Subscriber: a callback consumer routes messages out-of-band.
    qb::redis::tcp::cb_consumer consumer{
        qb::io::uri{"tcp://localhost:6379"},
        [](qb::redis::message &&msg) {
            qb::io::cout() << '[' << msg.channel << "] " << msg.payload << '\n';
        }};

    co_await consumer.connect();
    co_await consumer.hello(3);  // RESP3, for native push frames
    co_await consumer.subscribe("alerts");
    co_await consumer.psubscribe("user:*:updates");

    // Publisher: a separate client — publish() is not on the consumer.
    qb::redis::tcp::client publisher{qb::io::uri{"tcp://localhost:6379"}};
    co_await publisher.connect();
    co_await publisher.publish("alerts", "Server restarted");
}

Adding one to a project

Two steps. The CMake helper discovers every module under the directory you point it at, so the second step is the same whichever module you added.

Add the repository

git submodule add https://github.com/isndev/qbm-http qbm/http

Discover and link

qb_load_modules("${CMAKE_CURRENT_SOURCE_DIR}/qbm")
target_link_libraries(my_app PRIVATE qbm::http)

Writing your own

A qbm module is a contract rather than a private arrangement: the scaffold that produced these three is a script you can run. It generates the target, the qbm:: namespace, a tiered test suite and a superbuild root that can run those tests.

Scaffold a module

curl -fsSL https://raw.githubusercontent.com/isndev/qb/main/script/qb-new-module.sh \
  | bash /dev/stdin mymodule

Install

CMake 3.24, and dependencies fetch themselves.

A generator writes a project already wired to qb. Or clone the framework and run its test suite.

Scaffold a project

curl -fsSL https://raw.githubusercontent.com/isndev/qb/main/script/qb-new-project.sh \
  | bash /dev/stdin MyProject
cd MyProject

Configure and build

cmake -DCMAKE_BUILD_TYPE=Release -B build
cmake --build build --parallel

Build from source

git clone --recursive https://github.com/isndev/qb.git
cd qb
cmake -DCMAKE_BUILD_TYPE=Release -B build
cmake --build build --parallel
ctest --test-dir build --output-on-failure

Opt into what you deploy

cmake -B build \
  -DCMAKE_BUILD_TYPE=Release \
  -DQB_WITH_SSL=ON \
  -DQB_WITH_COMPRESSION=ON \
  -DQB_WITH_QUIC=ON \
  -DQB_BUILD_TESTS=ON

Build options

Every capability beyond the core is a flag, so a build carries only what it uses.

QB_WITH_SSL: SSL/TLS transports
QB_WITH_COMPRESSION: Payload compression
QB_WITH_QUIC: QUIC and HTTP/3
QB_WITH_LOGGING: Structured logging
QB_BUILD_TESTS: Framework test suite
QB_BUILD_BENCHMARKS: Google Benchmark suites

Platform support

Linux — GCC and Clang, libstdc++CI enabled
macOS — Apple Clang, libc++CI enabled
Windows — MSVCSupported, CI disabled

x86-64 and ARM64 on all three.

Documentation

Seven sections, from first principles to production.

The guide is written in order: what the model is, then each library on its own, then the two together, then the task-oriented material and the reference.

Browse the full guide

Contact

Talk to the maintainers.

Questions about adopting qb, a module you need, or a production deployment. Bugs and feature requests belong in GitHub issues.

Open an issue

Actors that share nothing. I/O that never blocks.

Five rules decide whether qb code is correct.

Time is std::chrono

Actors share nothing

Sending is noexcept

A coroutine captures by value

Register every event you handle

One worker thread per core, one mailbox each.

Same core

Across cores

Idle latency is a knob

Five primitives, five different contracts.

A server, a client, a coroutine.

The runtime works without the actors.

Transports

Protocols

Coroutines

Utilities

The runtime on its own

The protocols ship as modules, not as core.

qbm-http

Provides

qbm-pgsql

Provides

qbm-redis

Provides

What each one actually looks like

qbm-http

Choosing a client

qbm-pgsql

qbm-redis

Adding one to a project

Writing your own

CMake 3.24, and dependencies fetch themselves.

Build options

Platform support

Seven sections, from first principles to production.

Talk to the maintainers.