server: make concurrent

Signed-off-by: Robert Günzler <r@gnzler.io>

1 files changed, 67 insertions, 30 deletions

diff --git a/src/main.zig b/src/main.zig
index 4a87c86..642cb7a 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -477,46 +477,83 @@ const Cli = struct {
 
         log.info("Serving root={s} addr={s}", .{ root_path, addrStr });
 
-        const serverNode = if (!DEBUG) progress.start(addrStr, 0);
+        const server_node = if (!DEBUG) progress.start(addrStr, 0);
         defer if (!DEBUG) progress.end();
 
         // tcp server
         var server = try addr.listen(io, .{ .reuse_address = true });
         defer server.deinit(io);
 
+        var group: Io.Group = .init;
+        defer group.cancel(io);
+
         while (true) {
             const stream = try server.accept(io);
-            defer stream.close(io);
-
-            var stream_read_buffer: [BUFFER_SIZE]u8 = undefined;
-            var stream_reader = stream.reader(io, &stream_read_buffer);
-            var stream_write_buffer: [BUFFER_SIZE]u8 = undefined;
-            var stream_writer = stream.writer(io, &stream_write_buffer);
-
-            var http_server = std.http.Server.init(&stream_reader.interface, &stream_writer.interface);
-            var req = try http_server.receiveHead();
-
-            log.info("{s} {s}", .{ @tagName(req.head.method), req.head.target });
-
-            const streamNode = if (!DEBUG) serverNode.start(req.head.target, 0);
-            defer if (!DEBUG) streamNode.end();
-
-            switch (req.head.method) {
-                .GET, .HEAD => {},
-                else => return,
-            }
-
-            var fd = File.init(io, root_dir, req.head.target[1..], null) catch |err| {
-                switch (err) {
-                    error.FileNotFound => try req.respond("", .{ .status = .not_found }),
-                    else => try req.respond("", .{ .status = .internal_server_error }),
-                }
-                continue;
-            };
-            defer fd.close(io);
+            errdefer stream.close(io);
 
-            try req.respond(fd.mmap.memory, .{ .status = .ok });
+            try group.concurrent(
+                io,
+                struct {
+                    fn inner(
+                        io_: Io,
+                        stream_: Io.net.Stream,
+                        root_dir_: Dir,
+                        server_node_: ?std.Progress.Node,
+                    ) error{Canceled}!void {
+                        var rbuffer: [BUFFER_SIZE]u8 = undefined;
+                        var r = stream_.reader(io_, &rbuffer);
+                        var wbuffer: [BUFFER_SIZE]u8 = undefined;
+                        var w = stream_.writer(io_, &wbuffer);
+                        var http_server = std.http.Server.init(&r.interface, &w.interface);
+                        while (true) {
+                            var req: std.http.Server.Request = undefined;
+                            (blk: {
+                                req = http_server.receiveHead() catch |err| switch (err) {
+                                    error.HttpConnectionClosing => return, // return from the connection handler
+                                    error.ReadFailed => return,
+                                    else => {
+                                        log.err("ctx=recv", .{});
+                                        break :blk err;
+                                    },
+                                };
+
+                                log.info("{s} {s}", .{ @tagName(req.head.method), req.head.target });
+
+                                const stream_node = if (!DEBUG) server_node_.?.start(req.head.target, 0);
+                                defer if (!DEBUG) stream_node.end();
+
+                                switch (req.head.method) {
+                                    .GET, .HEAD => {},
+                                    else => return,
+                                }
+
+                                var fd = File.init(io_, root_dir_, req.head.target[1..], null) catch |err| {
+                                    return switch (err) {
+                                        error.FileNotFound => break :blk req.respond("", .{ .status = .not_found }),
+                                        else => {
+                                            log.err("ctx=file", .{});
+                                            break :blk err;
+                                        },
+                                    };
+                                };
+                                defer fd.close(io_);
+
+                                req.respond(fd.mmap.memory, .{ .status = .ok }) catch |err| {
+                                    log.err("ctx=send", .{});
+                                    break :blk err;
+                                };
+                            }) catch |err| {
+                                log.err("target={s} {}", .{ req.head.target, err });
+                                return error.Canceled;
+                            };
+                        }
+                    }
+                }.inner,
+                .{ io, stream, root_dir, if (!DEBUG) server_node else null },
+            );
         }
+
+        try group.await(io);
     }
 };