Learn Zig Series):Exercise 1: Multi-producer shared ring buffer
const std = @import("std");
const c = @cImport({
@cInclude("sys/mman.h");
@cInclude("fcntl.h");
@cInclude("unistd.h");
});
const SLOT_SIZE = 64;
const NUM_SLOTS = 512;
const SharedRing = extern struct {
head: u64 align(8), // atomic: next write position
tail: u64 align(8), // atomic: next read position
total_written: u64 align(8),
slots: [NUM_SLOTS][SLOT_SIZE]u8,
};
fn producerWork(ring: *SharedRing, producer_id: usize, count: usize) void {
var written: usize = 0;
while (written < count) {
// atomically claim a slot by incrementing head
const slot_idx_raw = @atomicRmw(u64, &ring.head, .Add, 1, .seq_cst);
const slot_idx = slot_idx_raw % NUM_SLOTS;
// spin until the slot is free (consumer has read past it)
// simple approach: wait until tail has advanced past previous use
while (true) {
const tail = @atomicLoad(u64, &ring.tail, .acquire);
// slot is available if head hasn't lapped tail by more than NUM_SLOTS
if (slot_idx_raw - tail < NUM_SLOTS) break;
std.atomic.spinLoopHint();
}
// write message into the claimed slot
var msg_buf: [SLOT_SIZE]u8 = [_]u8{0} ** SLOT_SIZE;
_ = std.fmt.bufPrint(&msg_buf, "p{d}-msg-{d}", .{ producer_id, written }) catch {};
@memcpy(&ring.slots[slot_idx], &msg_buf);
_ = @atomicRmw(u64, &ring.total_written, .Add, 1, .seq_cst);
written += 1;
}
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
const shm_name: [*:0]const u8 = "/zig_mpring";
_ = c.shm_unlink(shm_name);
const fd = c.shm_open(shm_name, c.O_CREAT | c.O_RDWR, 0o666);
if (fd < 0) return;
_ = c.ftruncate(fd, @intCast(@sizeOf(SharedRing)));
const ptr = c.mmap(null, @sizeOf(SharedRing), c.PROT_READ | c.PROT_WRITE, c.MAP_SHARED, fd, 0);
if (ptr == c.MAP_FAILED) return;
const ring: *SharedRing = @ptrCast(@alignCast(ptr));
@atomicStore(u64, &ring.head, 0, .seq_cst);
@atomicStore(u64, &ring.tail, 0, .seq_cst);
@atomicStore(u64, &ring.total_written, 0, .seq_cst);
@memset(std.mem.asBytes(&ring.slots), 0);
const msgs_per_producer: usize = 100;
const num_producers: usize = 3;
var pids: [3]std.posix.pid_t = undefined;
for (0..num_producers) |p| {
const pid = try std.posix.fork();
if (pid == 0) {
producerWork(ring, p, msgs_per_producer);
std.process.exit(0);
}
pids[p] = pid;
}
// parent = consumer: read all 300 messages
var received: usize = 0;
const total_expected = num_producers * msgs_per_producer;
while (received < total_expected) {
const tw = @atomicLoad(u64, &ring.total_written, .acquire);
while (received < tw) {
const slot_idx = @atomicLoad(u64, &ring.tail, .acquire) % NUM_SLOTS;
const slot = &ring.slots[slot_idx];
const end = std.mem.indexOfScalar(u8, slot, 0) orelse SLOT_SIZE;
if (received < 5 or received >= total_expected - 3) {
try stdout.print("[consumer] #{d}: {s}\n", .{ received, slot[0..end] });
}
_ = @atomicRmw(u64, &ring.tail, .Add, 1, .seq_cst);
received += 1;
}
if (received < total_expected) std.atomic.spinLoopHint();
}
for (pids[0..num_producers]) |pid| _ = std.posix.waitpid(pid, 0);
try stdout.print("Received {d}/{d} messages\n", .{ received, total_expected });
_ = c.munmap(ptr, @sizeOf(SharedRing));
_ = c.close(fd);
_ = c.shm_unlink(shm_name);
}
Each producer atomically claims a write slot by incrementing head with @atomicRmw. No mutex needed -- the atomic increment guarantees each producer gets a unique slot index. The consumer advances tail as it reads. With 512 slots and only 300 total messages, the ring never wraps.
Exercise 2: Process-safe shared hash map
const std = @import("std");
const c = @cImport({
@cInclude("sys/mman.h");
@cInclude("fcntl.h");
@cInclude("unistd.h");
});
const KEY_SIZE = 32;
const VAL_SIZE = 64;
const NUM_BUCKETS = 2048;
const Bucket = extern struct {
lock: u32 align(4), // 0 = unlocked, 1 = locked
occupied: u32, // 0 = empty, 1 = in use
key: [KEY_SIZE]u8,
value: [VAL_SIZE]u8,
};
const SharedMap = extern struct {
count: u64 align(8),
buckets: [NUM_BUCKETS]Bucket,
};
fn hashKey(key: []const u8) usize {
var h: u64 = 5381;
for (key) |ch| {
h = ((h << 5) +% h) +% ch;
}
return @intCast(h % NUM_BUCKETS);
}
fn spinLock(lock: *u32) void {
while (true) {
const prev = @atomicRmw(u32, lock, .Xchg, 1, .acquire);
if (prev == 0) return;
std.atomic.spinLoopHint();
}
}
fn spinUnlock(lock: *u32) void {
@atomicStore(u32, lock, 0, .release);
}
fn mapPut(map: *SharedMap, key: []const u8, value: []const u8) bool {
var idx = hashKey(key);
var probes: usize = 0;
while (probes < NUM_BUCKETS) : (probes += 1) {
const bucket = &map.buckets[idx];
spinLock(&bucket.lock);
if (bucket.occupied == 0) {
// empty slot, insert here
var k: [KEY_SIZE]u8 = [_]u8{0} ** KEY_SIZE;
@memcpy(k[0..@min(key.len, KEY_SIZE)], key[0..@min(key.len, KEY_SIZE)]);
bucket.key = k;
var v: [VAL_SIZE]u8 = [_]u8{0} ** VAL_SIZE;
@memcpy(v[0..@min(value.len, VAL_SIZE)], value[0..@min(value.len, VAL_SIZE)]);
bucket.value = v;
bucket.occupied = 1;
_ = @atomicRmw(u64, &map.count, .Add, 1, .seq_cst);
spinUnlock(&bucket.lock);
return true;
}
// check if key matches (update in place)
const end = std.mem.indexOfScalar(u8, &bucket.key, 0) orelse KEY_SIZE;
if (std.mem.eql(u8, bucket.key[0..end], key)) {
var v: [VAL_SIZE]u8 = [_]u8{0} ** VAL_SIZE;
@memcpy(v[0..@min(value.len, VAL_SIZE)], value[0..@min(value.len, VAL_SIZE)]);
bucket.value = v;
spinUnlock(&bucket.lock);
return true;
}
spinUnlock(&bucket.lock);
idx = (idx + 1) % NUM_BUCKETS;
}
return false;
}
fn mapGet(map: *SharedMap, key: []const u8, out: *[VAL_SIZE]u8) bool {
var idx = hashKey(key);
var probes: usize = 0;
while (probes < NUM_BUCKETS) : (probes += 1) {
const bucket = &map.buckets[idx];
spinLock(&bucket.lock);
if (bucket.occupied == 0) {
spinUnlock(&bucket.lock);
return false;
}
const end = std.mem.indexOfScalar(u8, &bucket.key, 0) orelse KEY_SIZE;
if (std.mem.eql(u8, bucket.key[0..end], key)) {
out.* = bucket.value;
spinUnlock(&bucket.lock);
return true;
}
spinUnlock(&bucket.lock);
idx = (idx + 1) % NUM_BUCKETS;
}
return false;
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
const shm_name: [*:0]const u8 = "/zig_hashmap";
_ = c.shm_unlink(shm_name);
const fd = c.shm_open(shm_name, c.O_CREAT | c.O_RDWR, 0o666);
if (fd < 0) return;
_ = c.ftruncate(fd, @intCast(@sizeOf(SharedMap)));
const ptr = c.mmap(null, @sizeOf(SharedMap), c.PROT_READ | c.PROT_WRITE, c.MAP_SHARED, fd, 0);
if (ptr == c.MAP_FAILED) return;
const map: *SharedMap = @ptrCast(@alignCast(ptr));
@memset(std.mem.asBytes(map), 0);
const pid = try std.posix.fork();
if (pid == 0) {
// child: insert 1000 key-value pairs
var key_buf: [32]u8 = undefined;
var val_buf: [64]u8 = undefined;
for (0..1000) |i| {
const klen = std.fmt.bufPrint(&key_buf, "key-{d}", .{i}) catch continue;
const vlen = std.fmt.bufPrint(&val_buf, "value-{d}", .{i}) catch continue;
_ = mapPut(map, klen, vlen);
}
std.process.exit(0);
}
// parent: concurrently try to read (will find some, miss some in-flight)
// after child finishes, verify all 1000
_ = std.posix.waitpid(pid, 0);
var found: usize = 0;
var key_buf: [32]u8 = undefined;
var val_out: [VAL_SIZE]u8 = undefined;
for (0..1000) |i| {
const klen = std.fmt.bufPrint(&key_buf, "key-{d}", .{i}) catch continue;
if (mapGet(map, klen, &val_out)) found += 1;
}
const count = @atomicLoad(u64, &map.count, .seq_cst);
try stdout.print("Inserted: {d}, Found: {d}\n", .{ count, found });
_ = c.munmap(ptr, @sizeOf(SharedMap));
_ = c.close(fd);
_ = c.shm_unlink(shm_name);
}
Per-bucket spinlocks let the writer and reader operate on different hash buckets simultaneously. The @atomicRmw exchange on the lock field is the lightest possible synchronization -- no kernel calls unless there's actual contention on the same bucket.
Exercise 3: Shared blackboard with publish/subscribe channels
const std = @import("std");
const c = @cImport({
@cInclude("sys/mman.h");
@cInclude("fcntl.h");
@cInclude("unistd.h");
});
const CHANNEL_NAME_SIZE = 32;
const CHANNEL_DATA_SIZE = 256;
const MAX_CHANNELS = 8;
const Channel = extern struct {
name: [CHANNEL_NAME_SIZE]u8,
active: u32 align(4),
sequence: u64 align(8), // bumped on each publish
data: [CHANNEL_DATA_SIZE]u8,
};
const Blackboard = extern struct {
num_channels: u32,
_pad: [4]u8,
channels: [MAX_CHANNELS]Channel,
};
fn findChannel(board: *Blackboard, name: []const u8) ?*Channel {
for (&board.channels) |*ch| {
if (@atomicLoad(u32, &ch.active, .acquire) == 0) continue;
const end = std.mem.indexOfScalar(u8, &ch.name, 0) orelse CHANNEL_NAME_SIZE;
if (std.mem.eql(u8, ch.name[0..end], name)) return ch;
}
return null;
}
fn publish(ch: *Channel, data: []const u8) void {
var buf: [CHANNEL_DATA_SIZE]u8 = [_]u8{0} ** CHANNEL_DATA_SIZE;
@memcpy(buf[0..@min(data.len, CHANNEL_DATA_SIZE)], data[0..@min(data.len, CHANNEL_DATA_SIZE)]);
ch.data = buf;
// release ensures data write is visible before sequence bump
_ = @atomicRmw(u64, &ch.sequence, .Add, 1, .release);
}
fn subscribe(ch: *Channel, last_seq: *u64, out: *[CHANNEL_DATA_SIZE]u8) bool {
// acquire ensures we see the data written before the sequence bump
const seq = @atomicLoad(u64, &ch.sequence, .acquire);
if (seq > last_seq.*) {
out.* = ch.data;
last_seq.* = seq;
return true;
}
return false;
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
const shm_name: [*:0]const u8 = "/zig_blackboard";
_ = c.shm_unlink(shm_name);
const fd = c.shm_open(shm_name, c.O_CREAT | c.O_RDWR, 0o666);
if (fd < 0) return;
_ = c.ftruncate(fd, @intCast(@sizeOf(Blackboard)));
const ptr = c.mmap(null, @sizeOf(Blackboard), c.PROT_READ | c.PROT_WRITE, c.MAP_SHARED, fd, 0);
if (ptr == c.MAP_FAILED) return;
const board: *Blackboard = @ptrCast(@alignCast(ptr));
@memset(std.mem.asBytes(board), 0);
// set up two channels
var temp_name: [CHANNEL_NAME_SIZE]u8 = [_]u8{0} ** CHANNEL_NAME_SIZE;
@memcpy(temp_name[0..11], "temperature");
board.channels[0].name = temp_name;
@atomicStore(u32, &board.channels[0].active, 1, .release);
var pres_name: [CHANNEL_NAME_SIZE]u8 = [_]u8{0} ** CHANNEL_NAME_SIZE;
@memcpy(pres_name[0..8], "pressure");
board.channels[1].name = pres_name;
@atomicStore(u32, &board.channels[1].active, 1, .release);
board.num_channels = 2;
// subscriber 1: watches temperature
const pid1 = try std.posix.fork();
if (pid1 == 0) {
const ch = findChannel(board, "temperature").?;
var last_seq: u64 = 0;
var buf: [CHANNEL_DATA_SIZE]u8 = undefined;
var reads: usize = 0;
while (reads < 10) {
if (subscribe(ch, &last_seq, &buf)) {
const end = std.mem.indexOfScalar(u8, &buf, 0) orelse CHANNEL_DATA_SIZE;
const stderr = std.io.getStdErr().writer();
stderr.print("[temp-sub] seq={d}: {s}\n", .{ last_seq, buf[0..end] }) catch {};
reads += 1;
}
std.time.sleep(10 * std.time.ns_per_ms);
}
std.process.exit(0);
}
// subscriber 2: watches pressure
const pid2 = try std.posix.fork();
if (pid2 == 0) {
const ch = findChannel(board, "pressure").?;
var last_seq: u64 = 0;
var buf: [CHANNEL_DATA_SIZE]u8 = undefined;
var reads: usize = 0;
while (reads < 10) {
if (subscribe(ch, &last_seq, &buf)) {
const end = std.mem.indexOfScalar(u8, &buf, 0) orelse CHANNEL_DATA_SIZE;
const stderr = std.io.getStdErr().writer();
stderr.print("[pres-sub] seq={d}: {s}\n", .{ last_seq, buf[0..end] }) catch {};
reads += 1;
}
std.time.sleep(10 * std.time.ns_per_ms);
}
std.process.exit(0);
}
// parent = publisher: write to both channels
const temp_ch = findChannel(board, "temperature").?;
const pres_ch = findChannel(board, "pressure").?;
var msg_buf: [CHANNEL_DATA_SIZE]u8 = undefined;
for (0..10) |i| {
const tlen = std.fmt.bufPrint(&msg_buf, "temp={d}.{d}C", .{ 20 + i, i * 3 }) catch continue;
_ = tlen;
publish(temp_ch, &msg_buf);
const plen = std.fmt.bufPrint(&msg_buf, "pres={d}.{d}hPa", .{ 1013 + i, i * 7 }) catch continue;
_ = plen;
publish(pres_ch, &msg_buf);
std.time.sleep(20 * std.time.ns_per_ms);
}
_ = std.posix.waitpid(pid1, 0);
_ = std.posix.waitpid(pid2, 0);
try stdout.print("Blackboard test complete\n", .{});
_ = c.munmap(ptr, @sizeOf(Blackboard));
_ = c.close(fd);
_ = c.shm_unlink(shm_name);
}
The .release ordering on publish and .acquire on subscribe form a release-acquire pair. This is sufficient because we only need the subscriber to see the data that was written before the sequence number was bumped -- we don't need a total global ordering between unrelated channels. This is cheaper than .seq_cst on architectures like ARM where sequential consistency requires full memory barriers.
Last episode we covered shared memory and semaphores -- the fastest IPC mechanism for processes on the same machine. We could map physical pages into multiple address spaces and coordinate access through atomics and semaphores. But there was one thing we didn't really address: what happens when a process gets rudely interrupted from the outside?
That's what signals are. Every Unix program has been dealing with them since the day it first ran -- you've been sending SIGINT every time you press Ctrl+C. But actually handling signals correctly is one of the trickiest parts of systems programming. Signal handlers run asynchronously, at arbitrary points in your program's execution. They can interrupt system calls, corrupt data structures, and deadlock your program if you're not careful. Most programs get signal handling subtley wrong, and most of the time it doesn't matter because programs usually just die on signals anyway.
But when you're building long-running services -- the kind we've been building in this OS programming arc -- getting signal handling right is the difference between a server that shuts down gracefully (finishes in-flight requests, flushes buffers, closes connections) and one that just dies mid-write and corrupts your data.
Here we go!
A signal is a software interrupt delivered to a process by the kernel. When a signal arrives, the kernel suspends whatever the process was doing, runs the signal handler (if one is registered), and then resumes the original code. Think of it like a hardware interrupt but at the process level.
There are about 30 standard signals on Linux. The important ones for application developers:
kill <pid> sends by default.You cannot catch SIGKILL or SIGSTOP. Everything else is fair game ;-)
const std = @import("std");
const linux = std.os.linux;
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
// print a few signal numbers to show they're just integers
try stdout.print("SIGINT = {d}\n", .{linux.SIG.INT});
try stdout.print("SIGTERM = {d}\n", .{linux.SIG.TERM});
try stdout.print("SIGCHLD = {d}\n", .{linux.SIG.CHLD});
try stdout.print("SIGKILL = {d}\n", .{linux.SIG.KILL});
try stdout.print("SIGUSR1 = {d}\n", .{linux.SIG.USR1});
try stdout.print("SIGPIPE = {d}\n", .{linux.SIG.PIPE});
// you can send signals to yourself
try stdout.print("\nSending SIGUSR1 to ourselves...\n", .{});
// default action for SIGUSR1 is terminate, so this would kill us
// let's ignore it first
var sa: linux.Sigaction = .{
.handler = .{ .handler = linux.SIG.IGN },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.USR1, &sa, null);
// now send it -- we'll survive because we set IGN
_ = linux.kill(linux.getpid(), linux.SIG.USR1);
try stdout.print("Survived SIGUSR1 (we're ignoring it)\n", .{});
}
The linux.SIG.IGN constant tells the kernel to silently discard the signal. The opposite is linux.SIG.DFL which restores the default action. Notice we're using linux.sigaction here -- Zig's standard library exposes the Linux sigaction syscall directly through std.os.linux, which is exactly what we need. No C interop required for this part.
The old signal() function from C is broken in several ways (it resets the handler after each delivery, it doesn't let you control which signals are blocked during handling, and its behavior varies between Unix variants). The proper interface is sigaction, and Zig gives us direct access to it:
const std = @import("std");
const linux = std.os.linux;
var signal_count: u32 = 0;
fn handler(sig: c_int) callconv(.c) void {
// this is our signal handler -- it runs asynchronously!
_ = sig;
signal_count += 1;
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
var sa: linux.Sigaction = .{
.handler = .{ .handler = handler },
.mask = linux.empty_sigset,
.flags = linux.SA.RESTART,
};
// register our handler for SIGUSR1
_ = linux.sigaction(linux.SIG.USR1, &sa, null);
try stdout.print("Handler registered. Sending 5 signals to ourselves...\n", .{});
const my_pid = linux.getpid();
var i: usize = 0;
while (i < 5) : (i += 1) {
_ = linux.kill(my_pid, linux.SIG.USR1);
}
try stdout.print("Received {d} signals\n", .{signal_count});
}
A few important things happening here:
The handler function has callconv(.c) -- signal handlers must use the C calling convention because the kernel expects it.
The SA.RESTART flag tells the kernel to automatically restart interrupted system calls. Without this, a signal arriving during a read() or write() would cause it to fail with EINTR, and you'd have to retry manually. With SA.RESTART, the kernel handles the retry for you. Almost always what you want.
The mask field controls which signals are blocked while this handler runs. empty_sigset means no additional signals are blocked. We could fill this with other signal numbers to prevent nested signal delivery.
The global signal_count variable is modified from the handler. This is technically a data race -- the handler and main code both access it without synchronization. For a simple counter it mostly works, but this is NOT safe in general. We'll fix this with atomics shortly.
Here's where signal handling gets truly tricky. When a signal arrives, your handler runs in the context of the interrupted code. If your main code was in the middle of malloc(), and your signal handler calls malloc(), you've got a deadlock -- malloc holds an internal lock, and the handler tries to acquire the same lock on the same thread.
The POSIX standard defines a list of async-signal-safe functions that are safe to call from signal handlers. The list is surprisingly short. What you CAN do in a signal handler:
volatile flag (or use an atomic)write() (the raw syscall, NOT stdio printf)_exit()sigaction, kill)What you CANNOT do:
malloc() or free() (Zig's allocators)printf() or any stdio buffered I/O (includes std.debug.print)std.io.getStdOut().writer().print() -- this uses buffered I/OThis is why most real signal handlers are tiny -- they just set a flag:
const std = @import("std");
const linux = std.os.linux;
var shutdown_requested: bool = false;
fn handleShutdown(sig: c_int) callconv(.c) void {
_ = sig;
// THIS is the correct pattern: set a flag, nothing else
@atomicStore(bool, &shutdown_requested, true, .release);
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
var sa: linux.Sigaction = .{
.handler = .{ .handler = handleShutdown },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.INT, &sa, null);
_ = linux.sigaction(linux.SIG.TERM, &sa, null);
try stdout.print("Running... press Ctrl+C to stop\n", .{});
var iteration: u64 = 0;
while (!@atomicLoad(bool, &shutdown_requested, .acquire)) {
// do work
iteration += 1;
if (iteration % 10_000_000 == 0) {
try stdout.print("Working... iteration {d}\n", .{iteration});
}
}
// clean shutdown
try stdout.print("\nShutdown requested after {d} iterations. Cleaning up...\n", .{iteration});
// flush buffers, close files, save state, etc.
try stdout.print("Goodbye!\n", .{});
}
The @atomicStore in the handler and @atomicLoad in the main loop form an acquire-release pair -- the same pattern we used for shared memory in episode 66. The signal handler is effectively a "different thread" from the main code's perspective, so atomics are necessary for correctness.
The flag-checking approach works when your main loop is busy doing computation. But what if your main loop is blocked on I/O? If you're sitting in a poll() or read() call waiting for data, a flag set by a signal handler won't be checked until the I/O completes.
The self-pipe trick (invented by Daniel J. Bernstein in the 1990s) elegantly solves this. You create a pipe, and the signal handler writes a byte to it. Your main loop includes the pipe's read end in its poll() set. When a signal arrives, the write wakes up poll(), and you can check which signal was delivered:
const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;
var signal_pipe_fd: posix.fd_t = -1;
fn signalWriter(sig: c_int) callconv(.c) void {
// write the signal number as a single byte to the pipe
// write() is async-signal-safe
const byte: [1]u8 = .{@intCast(@as(u32, @bitCast(sig)))};
_ = posix.write(signal_pipe_fd, &byte) catch {};
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
// create the self-pipe
const pipe_fds = try posix.pipe();
signal_pipe_fd = pipe_fds[1]; // handler writes to this end
// set pipe to non-blocking so the handler never blocks
var flags = linux.fcntl(pipe_fds[1], linux.F.GETFL, @as(linux.fd_t, 0));
_ = linux.fcntl(pipe_fds[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));
// register signal handler
var sa: linux.Sigaction = .{
.handler = .{ .handler = signalWriter },
.mask = linux.empty_sigset,
.flags = linux.SA.RESTART,
};
_ = linux.sigaction(linux.SIG.USR1, &sa, null);
_ = linux.sigaction(linux.SIG.INT, &sa, null);
_ = linux.sigaction(linux.SIG.TERM, &sa, null);
try stdout.print("PID: {d} -- send SIGUSR1 or Ctrl+C\n", .{linux.getpid()});
// main event loop: poll on the pipe + any other fds
var running = true;
while (running) {
var pollfds = [_]linux.pollfd{
.{
.fd = pipe_fds[0],
.events = linux.POLL.IN,
.revents = 0,
},
// you'd add other fds here (sockets, timers, etc.)
};
const ready = linux.poll(&pollfds, pollfds.len, 2000); // 2 second timeout
if (@as(isize, @bitCast(@as(usize, ready))) <= 0) {
try stdout.print("(tick -- no signals)\n", .{});
continue;
}
if (pollfds[0].revents & linux.POLL.IN != 0) {
// read the signal byte(s) from the pipe
var buf: [16]u8 = undefined;
const n = posix.read(pipe_fds[0], &buf) catch 0;
var j: usize = 0;
while (j < n) : (j += 1) {
const sig_num = buf[j];
switch (sig_num) {
@intCast(linux.SIG.USR1) => {
try stdout.print("Got SIGUSR1 -- doing custom action\n", .{});
},
@intCast(linux.SIG.INT), @intCast(linux.SIG.TERM) => {
try stdout.print("Got shutdown signal ({d}) -- exiting\n", .{sig_num});
running = false;
},
else => {
try stdout.print("Got signal {d}\n", .{sig_num});
},
}
}
}
}
posix.close(pipe_fds[0]);
posix.close(pipe_fds[1]);
}
The self-pipe trick is used in basically every production event loop -- nginx, Node.js's libuv, Python's asyncio, and countless others. It converts the asynchronous, dangerous world of signal handlers into the synchronous, safe world of file descriptor I/O.
On modern Linux there's also signalfd() which does the same thing more directly (you get a file descriptor that becomes readable when a signal arrives), but the self-pipe trick works on all Unix systems and is good to understand regardless.
Sometimes you need a critical section where signals must not be delivered. You don't want a SIGTERM arriving while you're halfway through writing a transaction log. Signal masks let you temporarily block signal delivery:
const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;
var got_signal: bool = false;
fn handler(sig: c_int) callconv(.c) void {
_ = sig;
@atomicStore(bool, &got_signal, true, .release);
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
// register handler for SIGUSR1
var sa: linux.Sigaction = .{
.handler = .{ .handler = handler },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.USR1, &sa, null);
// block SIGUSR1 during critical section
var block_set = linux.empty_sigset;
linux.sigaddset(&block_set, linux.SIG.USR1);
var old_set: linux.sigset_t = undefined;
try stdout.print("Blocking SIGUSR1...\n", .{});
_ = linux.sigprocmask(linux.SIG.BLOCK, &block_set, &old_set);
// send signal while blocked -- it will be QUEUED, not delivered
_ = linux.kill(linux.getpid(), linux.SIG.USR1);
try stdout.print("Signal sent while blocked. got_signal = {}\n", .{
@atomicLoad(bool, &got_signal, .acquire),
});
// simulate critical work
try stdout.print("Doing critical work...\n", .{});
std.time.sleep(100 * std.time.ns_per_ms);
try stdout.print("Critical work done. Unblocking...\n", .{});
// unblock -- the queued signal will be delivered NOW
_ = linux.sigprocmask(linux.SIG.SETMASK, &old_set, null);
try stdout.print("After unblock: got_signal = {}\n", .{
@atomicLoad(bool, &got_signal, .acquire),
});
}
When you block a signal, the kernel doesn't discard it -- it queues it. As soon as you unblock, the queued signal is delivered. This is how you protect critical sections: block signals before, do the work, unblock after. The pending signals arrive immediately after unblocking.
Important: standard (non-real-time) signals are NOT queued multiply. If you send SIGUSR1 five times while it's blocked, you get ONE delivery when you unblock. Real-time signals (SIGRTMIN through SIGRTMAX) ARE individually queued, but most programs use standard signals.
We've been using waitpid() to wait for child processes since episode 64. But waitpid is blocking -- your parent process sits there doing nothing until the child exits. For a server that manages multiple workers, you want to be notified asynchronously when a child exits so you can clean up (and maybe restart it).
That's what SIGCHLD is for. The kernel sends SIGCHLD to the parent whenever a child process terminates or is stopped. Combined with the self-pipe trick, you can handle child exits in your main event loop:
const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;
var sigchld_pipe: posix.fd_t = -1;
fn sigchldHandler(sig: c_int) callconv(.c) void {
_ = sig;
const byte = [_]u8{@intCast(linux.SIG.CHLD)};
_ = posix.write(sigchld_pipe, &byte) catch {};
}
fn reapChildren(stdout: anytype) !void {
// reap ALL finished children (there might be several)
while (true) {
const result = linux.waitpid(-1, null, linux.W.NOHANG);
const pid = @as(i32, @bitCast(@as(u32, @truncate(result))));
if (pid <= 0) break;
try stdout.print("[parent] child {d} exited\n", .{pid});
}
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
// self-pipe for SIGCHLD
const pipe_fds = try posix.pipe();
sigchld_pipe = pipe_fds[1];
var flags = linux.fcntl(pipe_fds[1], linux.F.GETFL, @as(linux.fd_t, 0));
_ = linux.fcntl(pipe_fds[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));
var sa: linux.Sigaction = .{
.handler = .{ .handler = sigchldHandler },
.mask = linux.empty_sigset,
.flags = linux.SA.RESTART | linux.SA.NOCLDSTOP,
};
_ = linux.sigaction(linux.SIG.CHLD, &sa, null);
// spawn 3 children that exit at different times
for (0..3) |i| {
const pid = try posix.fork();
if (pid == 0) {
const sleep_ms: u64 = (i + 1) * 200;
std.time.sleep(sleep_ms * std.time.ns_per_ms);
const stderr = std.io.getStdErr().writer();
stderr.print("[child {d}] exiting after {d}ms\n", .{ linux.getpid(), sleep_ms }) catch {};
std.process.exit(0);
}
try stdout.print("[parent] spawned child {d}\n", .{pid});
}
// event loop: wait for all children to exit
var children_remaining: u32 = 3;
while (children_remaining > 0) {
var pollfds = [_]linux.pollfd{.{
.fd = pipe_fds[0],
.events = linux.POLL.IN,
.revents = 0,
}};
const ready = linux.poll(&pollfds, 1, 1000);
if (@as(isize, @bitCast(@as(usize, ready))) > 0) {
if (pollfds[0].revents & linux.POLL.IN != 0) {
// drain the pipe
var buf: [16]u8 = undefined;
_ = posix.read(pipe_fds[0], &buf) catch {};
// reap children -- might be more than one
const old_remaining = children_remaining;
try reapChildren(stdout);
// crude count: check how many we reaped
// in production you'd track PIDs properly
_ = old_remaining;
children_remaining -|= 1;
}
}
}
try stdout.print("All children done.\n", .{});
posix.close(pipe_fds[0]);
posix.close(pipe_fds[1]);
}
The SA.NOCLDSTOP flag tells the kernel to only send SIGCHLD when children terminate, not when they're stopped (e.g. by SIGSTOP). Without this flag you'd get spurious notifications.
Notice the reapChildren function calls waitpid in a loop with WNOHANG. This is critical because multiple children might exit before the parent gets to handle the SIGCHLD. Remember -- standard signals don't queue. If three children exit between two iterations of our event loop, we get ONE SIGCHLD. The loop with WNOHANG ensures we reap all of them.
Let's put everything together into a pattern that every long-running server should implement. The idea is simple: when SIGTERM or SIGINT arrives, stop accepting new work, finish what's in progress, flush buffers, and exit cleanly.
const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;
const ServerState = struct {
signal_pipe: [2]posix.fd_t,
shutdown_requested: bool,
active_workers: u32,
total_requests: u64,
};
var server: ServerState = undefined;
fn shutdownHandler(sig: c_int) callconv(.c) void {
const byte = [_]u8{@intCast(@as(u32, @bitCast(sig)))};
_ = posix.write(server.signal_pipe[1], &byte) catch {};
}
fn workerTask(id: usize) void {
const stderr = std.io.getStdErr().writer();
// simulate handling a request
stderr.print("[worker {d}] processing request...\n", .{id}) catch {};
std.time.sleep(300 * std.time.ns_per_ms);
stderr.print("[worker {d}] request complete\n", .{id}) catch {};
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
// set up signal pipe
server.signal_pipe = try posix.pipe();
server.shutdown_requested = false;
server.active_workers = 0;
server.total_requests = 0;
var flags = linux.fcntl(server.signal_pipe[1], linux.F.GETFL, @as(linux.fd_t, 0));
_ = linux.fcntl(server.signal_pipe[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));
// register handlers
var sa: linux.Sigaction = .{
.handler = .{ .handler = shutdownHandler },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.INT, &sa, null);
_ = linux.sigaction(linux.SIG.TERM, &sa, null);
// also ignore SIGPIPE (common in network servers)
var sa_ign: linux.Sigaction = .{
.handler = .{ .handler = linux.SIG.IGN },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.PIPE, &sa_ign, null);
try stdout.print("Server started (PID {d}). Send SIGTERM or press Ctrl+C to stop.\n", .{
linux.getpid(),
});
// main server loop
var request_id: usize = 0;
while (!server.shutdown_requested) {
var pollfds = [_]linux.pollfd{.{
.fd = server.signal_pipe[0],
.events = linux.POLL.IN,
.revents = 0,
}};
// simulate accepting work with a timeout
const ready = linux.poll(&pollfds, 1, 500);
// check for signals
if (@as(isize, @bitCast(@as(usize, ready))) > 0) {
if (pollfds[0].revents & linux.POLL.IN != 0) {
var buf: [16]u8 = undefined;
_ = posix.read(server.signal_pipe[0], &buf) catch {};
try stdout.print("\nShutdown signal received.\n", .{});
server.shutdown_requested = true;
break;
}
}
// simulate accepting a new request every tick
request_id += 1;
server.total_requests += 1;
const pid = try posix.fork();
if (pid == 0) {
workerTask(request_id);
std.process.exit(0);
}
server.active_workers += 1;
// reap finished workers (non-blocking)
while (true) {
const result = linux.waitpid(-1, null, linux.W.NOHANG);
const rpid = @as(i32, @bitCast(@as(u32, @truncate(result))));
if (rpid <= 0) break;
if (server.active_workers > 0) server.active_workers -= 1;
}
if (request_id >= 6 and !server.shutdown_requested) {
// for demo purposes, auto-shutdown after 6 requests
try stdout.print("(demo: processed 6 requests, shutting down)\n", .{});
server.shutdown_requested = true;
}
}
// graceful shutdown: wait for in-flight workers
try stdout.print("Waiting for {d} active workers to finish...\n", .{server.active_workers});
while (server.active_workers > 0) {
const result = linux.waitpid(-1, null, 0);
const rpid = @as(i32, @bitCast(@as(u32, @truncate(result))));
if (rpid > 0) {
server.active_workers -= 1;
try stdout.print(" worker exited, {d} remaining\n", .{server.active_workers});
}
}
try stdout.print("Server shut down cleanly. Processed {d} total requests.\n", .{
server.total_requests,
});
posix.close(server.signal_pipe[0]);
posix.close(server.signal_pipe[1]);
}
This is the standard pattern used by production servers (nginx, PostgreSQL, Apache, Redis):
waitpid loopHaving said that, production servers usually add a shutdown timeout -- if workers don't finish within, say, 30 seconds, send them SIGTERM. If they STILL don't finish after another 10 seconds, send SIGKILL. You don't want a hung worker to prevent your server from ever restarting.
One common pattern in init systems (systemd, Docker) is to send SIGTERM first, wait a grace period, then send SIGKILL. Your server should handle the first SIGTERM gracefully and the second one (if the user is impatient and hits Ctrl+C again) forcefully:
const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;
var shutdown_count: u32 = 0;
var sig_pipe: posix.fd_t = -1;
fn shutdownEscalation(sig: c_int) callconv(.c) void {
// atomically increment the shutdown counter
const count = @atomicRmw(u32, &shutdown_count, .Add, 1, .seq_cst);
if (count >= 1) {
// second signal -- force exit immediately
// _exit is async-signal-safe, unlike exit()
std.posix._exit(1);
}
// first signal -- write to pipe for graceful handling
const byte = [_]u8{@intCast(@as(u32, @bitCast(sig)))};
_ = posix.write(sig_pipe, &byte) catch {};
}
pub fn main() !void {
const stdout = std.io.getStdOut().writer();
const pipe_fds = try posix.pipe();
sig_pipe = pipe_fds[1];
var flags = linux.fcntl(pipe_fds[1], linux.F.GETFL, @as(linux.fd_t, 0));
_ = linux.fcntl(pipe_fds[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));
var sa: linux.Sigaction = .{
.handler = .{ .handler = shutdownEscalation },
.mask = linux.empty_sigset,
.flags = 0,
};
_ = linux.sigaction(linux.SIG.INT, &sa, null);
_ = linux.sigaction(linux.SIG.TERM, &sa, null);
try stdout.print("PID {d}: press Ctrl+C once for graceful, twice for force quit\n", .{
linux.getpid(),
});
// simulate a long-running task
var i: u32 = 0;
while (i < 100) : (i += 1) {
var pollfds = [_]linux.pollfd{.{
.fd = pipe_fds[0],
.events = linux.POLL.IN,
.revents = 0,
}};
const ready = linux.poll(&pollfds, 1, 500);
if (@as(isize, @bitCast(@as(usize, ready))) > 0) {
if (pollfds[0].revents & linux.POLL.IN != 0) {
var buf: [4]u8 = undefined;
_ = posix.read(pipe_fds[0], &buf) catch {};
try stdout.print("\nFirst signal -- shutting down gracefully...\n", .{});
try stdout.print("(press Ctrl+C again to force quit)\n", .{});
// simulate cleanup that takes a while
try stdout.print("Flushing buffers...\n", .{});
std.time.sleep(500 * std.time.ns_per_ms);
try stdout.print("Closing connections...\n", .{});
std.time.sleep(500 * std.time.ns_per_ms);
try stdout.print("Saving state...\n", .{});
std.time.sleep(500 * std.time.ns_per_ms);
break;
}
}
try stdout.print("Working... ({d})\n", .{i});
}
try stdout.print("Clean shutdown complete.\n", .{});
posix.close(pipe_fds[0]);
posix.close(pipe_fds[1]);
}
The key insight is the atomic counter. First SIGINT/SIGTERM triggers graceful shutdown through the pipe. If the user sends a second signal while we're still cleaning up, the handler sees count >= 1 and calls _exit(1) directly -- immediate termination, no messing about. This is exacly what you want: patience is rewarded with a clean shutdown, impatience still works (just less cleanly) ;-)
Note we use std.posix._exit (which calls the _exit syscall directly) rather than std.process.exit (which runs atexit handlers and flushes stdio buffers). In a signal handler, _exit is safe; exit is not.
Build a "watchdog timer" process. The parent forks a child that does work (simulated with sleep). The parent sets a SIGALRM alarm for 2 seconds using the alarm() syscall (via @cImport). If the child finishes before the alarm fires, cancel the alarm and print success. If the alarm fires first (child is taking too long), the SIGALRM handler should kill the child with SIGTERM, wait for it, and print a timeout message. Test with both a "fast" child (sleeps 500ms) and a "slow" child (sleeps 5 seconds).
Implement a signal-driven log rotator. Write a long-running process that writes numbered log lines to a file (/tmp/zig_log_test.log). When it receives SIGUSR1 (via self-pipe), it should: close the current log file, rename it to zig_log_test.log.1 (moving the old .1 to .2 if it exists), open a fresh log file, and continue writing. Use kill from another terminal to send SIGUSR1 and verify the rotation works. The signal handler must NOT do the rotation itself -- it must only write to the self-pipe, and the main loop does the actual file operations.
Build a process pool with signal-based lifecycle management. The parent process starts 4 worker children. Each worker does simulated work in a loop (sleep 200ms per "task"). The parent handles three signals: SIGUSR1 means "add a worker" (fork one more, up to a max of 8), SIGUSR2 means "remove a worker" (send SIGTERM to the most recently spawned child), and SIGTERM means "shut down all workers gracefully" (send SIGTERM to each child, wait for all, then exit). Track all child PIDs in an array. Handle SIGCHLD to detect when workers exit unexpectedly and print a warning.
sigaction syscall (via std.os.linux.sigaction) is the proper way to register handlers -- the old signal() function is broken and non-portablesigprocmask) block signal delivery during critical sections -- blocked signals are queued (one per signal number) and delivered when unblockedwaitpid with WNOHANG in a loop because multiple children can exit between SIGCHLD deliveriesSignals are one of the oldest Unix APIs and also one of the most misunderstood. The rules are strict (async-signal-safety) but the patterns are well established (self-pipe, atomic flags, graceful shutdown). Now that we've covered fork, pipes, shared memory, and signals, we have most of the foundational IPC and process management primitives. The next steps in this OS programming arc will take us into Unix domain sockets for local networking and the machinery that turns a regular process into a background daemon.
Bedankt en tot de volgende keer!
@scipio