🎓 A1Learner
🔍
0 of 46 lessons completed

Zig Project: Concurrent Server

A concurrent server is a program that handles many requests at the same time with several workers. This lesson simulates requests in memory, so no network socket is ever opened.

Drain a queue with two workers

Each worker claims the next request under the lock, then works without holding the lock. The main thread prints the results in request order after both joins.

const std = @import("std");
const Request = struct { path: []const u8, work: u32 };
const Context = struct {
    requests: []const Request,
    next: usize,
    mutex: std.Thread.Mutex,
    results: []u32,
};
fn worker(ctx: *Context) void {
    while (true) {
        ctx.mutex.lock();
        if (ctx.next >= ctx.requests.len) {
            ctx.mutex.unlock();
            break;
        }
        const idx = ctx.next;
        ctx.next += 1;
        ctx.mutex.unlock();
        ctx.results[idx] = ctx.requests[idx].work * 2;
    }
}
pub fn main() !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();
    var out = std.io.getStdOut().writer();
    const requests = [_]Request{ .{ .path = "/a", .work = 10 }, .{ .path = "/b", .work = 20 }, .{ .path = "/c", .work = 30 }, .{ .path = "/d", .work = 40 } };
    const results = try alloc.alloc(u32, requests.len);
    var ctx = Context{ .requests = &requests, .next = 0, .mutex = .{}, .results = results };
    var t0 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t1 = try std.Thread.spawn(.{}, worker, .{&ctx});
    t0.join();
    t1.join();
    for (results, 0..) |r, i| {
        try out.print("request {d} -> {d}\n", .{ i, r });
    }
    try out.print("handled {d}\n", .{ctx.next});
}

Decide a status per request

The handler decides a status from the request path. Results are stored by request position, so the printed order never depends on which worker handled a request.

const std = @import("std");
const Request = struct { path: []const u8, work: u32 };
const Context = struct {
    requests: []const Request,
    next: usize,
    mutex: std.Thread.Mutex,
    results: []u32,
};
fn worker(ctx: *Context) void {
    while (true) {
        ctx.mutex.lock();
        if (ctx.next >= ctx.requests.len) {
            ctx.mutex.unlock();
            break;
        }
        const idx = ctx.next;
        ctx.next += 1;
        ctx.mutex.unlock();
        if (std.mem.eql(u8, ctx.requests[idx].path, "/missing")) {
            ctx.results[idx] = 404;
        } else {
            ctx.results[idx] = 200;
        }
    }
}
pub fn main() !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();
    var out = std.io.getStdOut().writer();
    const requests = [_]Request{ .{ .path = "/home", .work = 0 }, .{ .path = "/users", .work = 0 }, .{ .path = "/missing", .work = 0 }, .{ .path = "/about", .work = 0 } };
    const results = try alloc.alloc(u32, requests.len);
    var ctx = Context{ .requests = &requests, .next = 0, .mutex = .{}, .results = results };
    var t0 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t1 = try std.Thread.spawn(.{}, worker, .{&ctx});
    t0.join();
    t1.join();
    for (requests, 0..) |req, i| {
        try out.print("{s} -> {d}\n", .{ req.path, results[i] });
    }
}

Three workers, six requests

More workers do not change the result, only how the work is shared. The main thread totals the results after all three workers have joined.

const std = @import("std");
const Request = struct { path: []const u8, work: u32 };
const Context = struct {
    requests: []const Request,
    next: usize,
    mutex: std.Thread.Mutex,
    results: []u32,
};
fn worker(ctx: *Context) void {
    while (true) {
        ctx.mutex.lock();
        if (ctx.next >= ctx.requests.len) {
            ctx.mutex.unlock();
            break;
        }
        const idx = ctx.next;
        ctx.next += 1;
        ctx.mutex.unlock();
        ctx.results[idx] = ctx.requests[idx].work;
    }
}
pub fn main() !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();
    var out = std.io.getStdOut().writer();
    const requests = [_]Request{ .{ .path = "/a", .work = 10 }, .{ .path = "/b", .work = 20 }, .{ .path = "/c", .work = 30 }, .{ .path = "/d", .work = 40 }, .{ .path = "/e", .work = 50 }, .{ .path = "/f", .work = 60 } };
    const results = try alloc.alloc(u32, requests.len);
    var ctx = Context{ .requests = &requests, .next = 0, .mutex = .{}, .results = results };
    var t0 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t1 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t2 = try std.Thread.spawn(.{}, worker, .{&ctx});
    t0.join();
    t1.join();
    t2.join();
    var total: u32 = 0;
    for (results) |r| {
        total += r;
    }
    try out.print("handled {d}\n", .{ctx.next});
    try out.print("total {d}\n", .{total});
}

Example

const std = @import("std");
const Request = struct { path: []const u8, work: u32 };
const Context = struct {
    requests: []const Request,
    next: usize,
    mutex: std.Thread.Mutex,
    results: []u32,
};
fn worker(ctx: *Context) void {
    while (true) {
        ctx.mutex.lock();
        if (ctx.next >= ctx.requests.len) {
            ctx.mutex.unlock();
            break;
        }
        const idx = ctx.next;
        ctx.next += 1;
        ctx.mutex.unlock();
        if (std.mem.eql(u8, ctx.requests[idx].path, "/missing") or std.mem.eql(u8, ctx.requests[idx].path, "/gone")) {
            ctx.results[idx] = 404;
        } else {
            ctx.results[idx] = 200;
        }
    }
}
pub fn main() !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();
    var out = std.io.getStdOut().writer();
    const requests = [_]Request{ .{ .path = "/home", .work = 0 }, .{ .path = "/users", .work = 0 }, .{ .path = "/missing", .work = 0 }, .{ .path = "/settings", .work = 0 }, .{ .path = "/gone", .work = 0 }, .{ .path = "/help", .work = 0 } };
    const results = try alloc.alloc(u32, requests.len);
    var ctx = Context{ .requests = &requests, .next = 0, .mutex = .{}, .results = results };
    var t0 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t1 = try std.Thread.spawn(.{}, worker, .{&ctx});
    var t2 = try std.Thread.spawn(.{}, worker, .{&ctx});
    t0.join();
    t1.join();
    t2.join();
    for (requests, 0..) |req, i| {
        try out.print("{s} -> {d}\n", .{ req.path, results[i] });
    }
    try out.print("handled {d}\n", .{ctx.next});
}

Key points

  • Worker: a worker is a thread that takes requests from a shared queue and handles them.
  • Queue: a queue is a list of work items taken in order.
  • Shared index: a shared index is a counter that says which request is next.
  • Mutex: a mutex is a lock that lets only one thread change shared state at a time.
  • Join: a join is the act of waiting for a thread to finish before continuing.
  • Ordered output: ordered output is results printed in request order after every worker has joined.

Three workers drain six simulated requests through one shared index. Each worker stores its status by request position. The main thread joins all three workers first, then prints every status in request order.

Try it Yourself

Edit the code, then press Run.

Running is simulated

Exercise

When must the main thread print the results of the workers?