Zig

Each file is a program of its own, an executable in your build.zig; these two are the repository's examples/zig, a project that builds them. Start the two in either order: fastipc.streamConnect waits up to its timeout for a server to create the stream and accept. The client sends the message ping, and the server answers in upper case: the exchange every example on this site makes, so either program works with the other half in any language (every pair of languages is tested).

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    // rings of 1 MiB each way
    const listener = try fastipc.streamCreate(
        init.gpa,
        "my_channel",
        1 << 20,
    );
    defer listener.close();
    // waits for a client
    const conn = try listener.accept(.none);
    defer conn.close();

    var buf: [256]u8 = undefined;
    // .none: no timeout
    const len = try conn.recv(&buf, .none);
    const reply = std.ascii.upperString(&buf, buf[0..len]);
    try conn.send(reply, .none);
}

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    const conn = try fastipc.streamConnect(
        init.gpa,
        "my_channel",
        five_s,
    );
    defer conn.close();
    try conn.send("ping", .none);
    var buf: [256]u8 = undefined;
    const len = try conn.recv(&buf, .none);
    std.debug.print("{s}\n", .{buf[0..len]}); // PING
}

send takes a message of any size and sends it in pieces of up to maxPiece() bytes when it is longer; its timeout bounds the wait for room for the first piece, and a started message completes unless a cancel or the peer's end stops it, which drops it whole. recv fills your buffer and returns the length; recvAlloc returns a new slice that you free. This server echoes every message back until its client leaves, which ends its next receive with error.Disconnected.

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    // rings of 1 MiB each way: a power of two, 1 KiB to 2 GiB
    const listener = try fastipc.streamCreate(
        init.gpa,
        "demo",
        1 << 20,
    );
    defer listener.close();
    // waits for a client
    const conn = try listener.accept(.none);
    defer conn.close();
    while (true) {
        // a message of any size, in memory from gpa
        const message = conn.recvAlloc(
            init.gpa,
            .none,
        ) catch |err| switch (err) {
            error.Disconnected => break, // the client left
            else => return err,
        };
        defer init.gpa.free(message);
        try conn.send(message, .none); // echo it
    }
    std.debug.print("client gone\n", .{});
}

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    const conn = try fastipc.streamConnect(
        init.gpa,
        "demo",
        five_s,
    );
    defer conn.close();
    var buf: [64]u8 = undefined;
    for ([_][]const u8{ "hello", "shared", "memory" }) |word| {
        try conn.send(word, .none);
        const len = try conn.recv(&buf, five_s);
        std.debug.print("{s}\n", .{buf[0..len]});
    }
}

Each error has the name of the C result code it stands for (the codes). The four creates (streamCreate, broadcastCreate, queueCreate, serviceCreate) return fastipc.CreateError, accept returns fastipc.AcceptError, streamConnect and the joins return fastipc.Error, a queue's and a service's waitForMembers return fastipc.WaitError, and the data calls (and a publisher's waitForMembers) return fastipc.DataError; recvAlloc and call add std.mem.Allocator.Error.

const Timeout = std.Io.Timeout;
const forever: Timeout = .none;
const no_wait: Timeout = .{ .duration = .{
    .raw = .zero,
    .clock = .awake,
} };
const five_s: Timeout = .{ .duration = .{
    .raw = .fromSeconds(5),
    .clock = .awake,
} };
const this_frame: Timeout = .{ .deadline = .fromNow(io, .{
    .raw = .fromMilliseconds(16),
    .clock = .awake,
}) };

const hundred_ms: std.Io.Timeout = .{ .duration = .{
    .raw = .fromMilliseconds(100),
    .clock = .awake,
} };
var buf: [4096]u8 = undefined;
if (conn.recv(&buf, hundred_ms)) |len| {
    handle(buf[0..len]);
} else |err| switch (err) {
    error.Timeout => {}, // nothing yet
    else => return err, // error.Disconnected: the peer is gone
}

The data calls (send, recv, recvAlloc, acquire, recvAcquire, call) wait in whole milliseconds and round a timeout up to the next one.

acquire returns room in the ring as a []u8; write the message there and commit the bytes you wrote, at least one. recvAcquire returns the next message where it lies, as a []const u8 that stays valid until release, the next receive or close. maxPiece() is the ring's capacity less 64 bytes. A longer message goes through send, and recvAcquire returns error.TooLarge for a message of several pieces, which stays queued for recv.

const out = try conn.acquire(64, .none); // room in the ring
const text = try std.mem.print(out, "tick {d}", .{tick});
try conn.commit(text.len); // publishes text.len bytes

// the next message, read-only, in the ring
const message = try conn.recvAcquire(.none);
handle(message);
conn.release(); // frees its room

A broadcast sends every message of one publisher to each of its subscribers, in any processes of the user; each message is stored once, in memory every subscriber maps. Messages are of one piece, up to a quarter of the capacity. fastipc.broadcastCreate returns the publisher's connection; the publisher's process admits the subscribers on a task of its own, so no thread of yours waits for them. The publisher waits for the first one (publisher.waitForMembers) before it sends tick 1 to tick 3; publisher.close() ends the broadcast and frees the name. The subscriber joins gated (fastipc.broadcastJoin with .gated), so it misses nothing (the publisher waits for it), and receives until error.Disconnected, which comes after the last message. A lossy subscriber (.lossy) never holds the publisher back: one that falls a lap behind gets error.Overrun, and conn.lostMessages() says how many it lost. A gated one that holds the publisher back for the broadcast's max_stall_ms is unsubscribed (error.Unsubscribed).

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    // 64 KiB of messages, up to 8 subscribers;
    // one that holds the publisher back 5 s is dropped
    const publisher = try fastipc.broadcastCreate(
        init.gpa,
        "my_broadcast",
        1 << 16,
        8,
        5000,
    );
    defer publisher.close(); // ends the broadcast

    // waits for the first subscriber
    _ = try publisher.waitForMembers(1, .none);
    for (1..4) |i| {
        var buf: [16]u8 = undefined;
        const tick = try std.mem.print(&buf, "tick {d}", .{i});
        try publisher.send(tick, .none);
    }
}

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    // gated: misses nothing (the publisher waits for it)
    const subscriber = try fastipc.broadcastJoin(
        init.gpa,
        "my_broadcast",
        .gated,
        five_s,
    );
    defer subscriber.close();
    var buf: [256]u8 = undefined;
    while (true) {
        const len = subscriber.recv(&buf, .none) catch |err| switch (err) {
            error.Disconnected => return, // the publisher closed
            else => return err,
        };
        std.debug.print("{s}\n", .{buf[0..len]}); // tick 1, ...
    }
}

A work queue takes messages from up to 63 producers and hands each to one of up to 64 consumers, in any processes of the user. Each producer has a ring of its own; a consumer takes a message of any producer, and the queue promises no order. Messages are of one piece, up to a quarter of the capacity, and wait in the queue until a consumer takes them. Each message goes to one consumer, and again, to another, only if that consumer's process ends before acknowledging it (delivery, below). Here the producer creates the queue (fastipc.queueCreate: a Queue) and joins it as its producer with fastipc.queueJoin, as any process does: the owner's process admits every joiner on a task of its own, so the thread that created the queue may join it. It sends job 1 to job 3, waits for the consumer (queue.waitForMembers) and closes the queue, which ends it: the consumer receives what is left, then error.Disconnected. Other processes join with fastipc.queueJoin (.producer, .consumer); a member's close never ends the queue.

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    // 4 KiB of messages per producer, 1 producer and 1 consumer:
    // this process owns the queue and is its producer
    const queue = try fastipc.queueCreate(
        init.gpa,
        "my_queue",
        4096,
        1,
        1,
    );
    defer queue.close(); // ends the queue
    const producer = try fastipc.queueJoin(
        init.gpa,
        "my_queue",
        .producer,
        five_s,
    );
    defer producer.close();
    for (1..4) |i| {
        var buf: [16]u8 = undefined;
        const job = try std.mem.print(&buf, "job {d}", .{i});
        try producer.send(job, .none); // waits in the queue for a consumer
    }
    _ = try queue.waitForMembers(.consumer, 1, .none); // waits for the consumer to join
}

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    const consumer = try fastipc.queueJoin(
        init.gpa,
        "my_queue",
        .consumer,
        five_s,
    );
    defer consumer.close();
    var buf: [256]u8 = undefined;
    while (true) {
        const len = consumer.recv(&buf, .none) catch |err| switch (err) {
            error.Disconnected => return, // the queue ended
            else => return err,
        };
        std.debug.print("{s}\n", .{buf[0..len]}); // job 1, ...
    }
}

Delivery. Each message goes to one consumer. It goes again, to another consumer, only if the process of the consumer that took it ends before acknowledging it, and when a consumer acknowledges is its own choice, not the queue's. A message received in place is acknowledged at its release (or the consumer's next receive, or its close).

• Choose by the work, per consumer. Work that may safely run twice (rendering, a computation, re-indexing, converting a file) acknowledges after it; work whose effects must never repeat (a payment, an email, a motor command) acknowledges before it.
• Not covered: a consumer that hangs while its process runs (there are no timeouts, by design); the owner's process, with which the queue ends; a reboot. A consumer holds at most one unacknowledged message at a time, the one received in place.
• A message that goes again comes in no set order, after messages its producer sent later: the queue promises no order anyway.
• A message held in place holds back its producer's room until its release: a consumer that acknowledges after a long job keeps that message's room, and so the room of the producer's messages after it, until then.
• A message that crashes every consumer that takes it, acknowledged after the work, goes to each consumer in turn, with no limit: acknowledge before the work where that can happen.

A service takes requests from up to 64 clients and hands each to one of up to 64 servers, in any processes of the user; the server that took a request replies to it, and a client receives its replies in the order of its requests. Here the server creates the service (fastipc.serviceCreate: a Service) and joins it as its server with fastipc.serviceJoin (.server), answers three requests with their uppercase and closes the service, which ends it; the client joins (.client) and calls ping 1 to ping 3: call sends a request and returns its reply, in memory from the allocator, and the client prints each. A request whose server ended before it replied gets error.Abandoned.

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    // 4 KiB per client, requests and replies of up to 256 bytes, 1 client
    // and 1 server: this process owns the service and is its server
    const service = try fastipc.serviceCreate(
        init.gpa,
        "my_service",
        4096,
        256,
        1,
        1,
    );
    defer service.close(); // ends the service
    const server = try fastipc.serviceJoin(
        init.gpa,
        "my_service",
        .server,
        five_s,
    );
    defer server.close();
    var buf: [256]u8 = undefined;
    for (0..3) |_| { // three requests, then the end
        const len = try server.recv(&buf, .none); // the next request of any client
        const reply = std.ascii.upperString(&buf, buf[0..len]);
        try server.send(reply, .none); // goes to its client, never waits
    }
}

const std = @import("std");
const fastipc = @import("fastipc");

pub fn main(init: std.process.Init) !void {
    const five_s: std.Io.Timeout = .{ .duration = .{
        .raw = .fromSeconds(5),
        .clock = .awake,
    } };
    const client = try fastipc.serviceJoin(
        init.gpa,
        "my_service",
        .client,
        five_s,
    );
    defer client.close();
    var buf: [16]u8 = undefined;
    for (1..4) |i| {
        const request = try std.mem.print(&buf, "ping {d}", .{i});
        // sends the request and returns its reply, which the caller
        // frees; error.Abandoned if the server ended before it replied
        const reply = try client.call(init.gpa, request, .none);
        defer init.gpa.free(reply);
        std.debug.print("{s}\n", .{reply}); // PING 1, ...
    }
}

The calls. A service's connections are Conns, and what each call does depends on the role. Requests and replies are of one piece, 1 to max_message bytes, the owner's choice at serviceCreate (1 to a quarter of the capacity); maxPiece() returns it on both roles.

• Replies come back in the order of the requests, so the n-th reply a client receives answers its n-th request, and a request carries no id. A client may send several requests before it receives, as many as its ring has room for, and the servers work on them at once. A receive that times out or is cancelled leaves its reply to come: the next receive returns it. So a client that sends again after a receive that timed out receives that earlier reply first: replies come in request order.
• call(gpa, request, timeout) never returns another request's reply. A call checks its request first: error.Invalid for an empty one, error.TooLarge over maxPiece(), at once, even while a reply is owed. Then a call on a cancelled connection returns error.Cancelled at once, whatever its timeout, and sends nothing: it could never receive the reply. The connection keeps count of the replies still owed: a call whose receive timed out or was cancelled (or whose reply it couldn't allocate) leaves its reply owed, and the next call first receives and drops that reply (at most one is owed, since every call drops it first; an error.Abandoned in its place counts too), within its own timeout and before it sends. A call that had to drop an owed reply first sends its request only if its timeout has time left: once a deadline or a positive duration has run out, it returns error.Timeout without sending, so no server sees that request; with a duration of zero no step waits, and it sends if there is room. After a call that gave up, a plain receive returns that call's late reply (or error.Abandoned) first: replies come in request order. A call returns error.Invalid while a request sent with send (or commit) still has its reply to be received: that reply is yours, so receive it first; on any connection other than a service's client it returns error.Invalid too. A call sends and receives, so no other thread may send or receive on the connection meanwhile.
• A server never waits for a client. A request takes room in its client's ring for itself and for its reply: its length rounded up to 64 bytes, plus max_message rounded up to 64, from its send until its reply has been received. A client that keeps sending without receiving waits for room its own replies hold.
• A server holds one request at a time: until it replies, its receives return error.Invalid. It may receive on one thread and reply on another. A pool is several servers, each a Conn of its own; each takes the next request of any client. A reply to a client that has left is dropped and succeeds.
• A request is answered once: by its reply, or with error.Abandoned if the server that took it ended before it replied. An abandoned request may or may not have been carried out; the service never sends it again, and the client decides whether to. Not covered: a server that hangs while its process runs (there are no timeouts), and the owner's process, with which the service ends.
• The Service is the service: its close ends it. Then clients' sends and servers' receives get error.Disconnected, and a client receives the replies that came before the end, in order, then error.Disconnected. A member's close never ends the service.