karlseguin.tngl.sh
A websocket implementation for zig
Nonblocking worker is working, but the code is a mess.
Need to improve authbahn test reliability, re-enable a bunch of deleted unit
tests, and then start refactoring.
author
Karl Seguin
date
(Jul 14, 2024, 2:14 PM +0800)
commit
692648b0
692648b06867e4ebd1bbe141f31e2330e8e487bf
parent
2d83ac34
2d83ac3462cf6cc9119dc73cf49635d47dbdbcff
+532
-279
+1
Dockerfile
+1
Dockerfile
···
1
+
Dockerfile
+2
-2
autobahn_client.zig
support/autobahn/client/main.zig
+2
-2
autobahn_client.zig
support/autobahn/client/main.zig
···
1
1
const std = @import("std");
2
-
const websocket = @import("./src/websocket.zig");
2
+
const websocket = @import("websocket");
3
3
4
4
pub fn main() !void {
5
5
var gpa = std.heap.GeneralPurposeAllocator(.{}){};
···
57
57
};
58
58
59
59
// wait 5 seconds for autobanh server to be up
60
-
std.time.sleep(5000000000);
60
+
std.time.sleep(std.time.ns_per_s * 5);
61
61
62
62
var buffer_provider = try websocket.bufferProvider(allocator, .{.count = 10, .size = 32768, .max = 20_000_000});
63
63
defer buffer_provider.deinit();
+3
-3
autobahn_server.zig
support/autobahn/server/main.zig
+3
-3
autobahn_server.zig
support/autobahn/server/main.zig
···
1
1
const std = @import("std");
2
-
const websocket = @import("./src/websocket.zig");
2
+
const websocket = @import("websocket");
3
3
4
4
const Conn = websocket.Conn;
5
5
const Message = websocket.Message;
···
15
15
var gpa = std.heap.GeneralPurposeAllocator(.{}){};
16
16
defer _ = gpa.detectLeaks();
17
17
18
-
const allocator = gpa.allocator();
18
+
const allocator = if (@import("builtin").mode == .Debug) gpa.allocator() else std.heap.c_allocator;
19
19
20
20
var server = try websocket.Server(Handler).init(allocator, .{
21
21
.port = 9223,
···
62
62
context: *Context,
63
63
64
64
pub fn init(_: Handshake, conn: *Conn, context: *Context) !Handler {
65
-
return Handler{
65
+
return .{
66
66
.conn = conn,
67
67
.context = context,
68
68
};
+7
-1
build.zig
+7
-1
build.zig
···
4
4
const target = b.standardTargetOptions(.{});
5
5
const optimize = b.standardOptimizeOption(.{});
6
6
7
-
_ = b.addModule("websocket", .{
7
+
const websocket_module = b.addModule("websocket", .{
8
8
.root_source_file = b.path("src/websocket.zig"),
9
9
});
10
+
11
+
{
12
+
const options = b.addOptions();
13
+
options.addOption(bool, "force_blocking", false);
14
+
websocket_module.addOptions("build", options);
15
+
}
10
16
11
17
const lib_test = b.addTest(.{
12
18
.root_source_file = b.path("src/websocket.zig"),
+1
-1
readme.md
+1
-1
readme.md
···
394
394
# Autobahn
395
395
Every mandatory [Autobahn Testsuite](https://github.com/crossbario/autobahn-testsuite) case is passing. (Three fragmented UTF-8 are flagged as non-strict and as compression is not implemented, these are all flagged as "Unimplemented").
396
396
397
-
You can see `autobahn_server.zig` and `autobahn_client.zig` in the root of the project for the handler that's used for the autobahn tests.
397
+
You can see `support/autobahn/server/main.zig` and `support/autobahn/client/main.zig` for the handler that's used for the autobahn tests.
398
398
399
399
## http.zig
400
400
I'm also working on an HTTP 1.1 server for zig: [https://github.com/karlseguin/http.zig](https://github.com/karlseguin/http.zig).
+450
-267
src/server/server.zig
+450
-267
src/server/server.zig
···
1
1
const std = @import("std");
2
+
const builtin = @import("builtin");
2
3
const proto = @import("../proto.zig");
3
4
const buffer = @import("../buffer.zig");
4
5
···
32
33
};
33
34
34
35
pub fn blockingMode() bool {
35
-
return true;
36
-
// if (force_blocking) {
37
-
// return true;
38
-
// }
39
-
// return switch (builtin.os.tag) {
40
-
// .linux, .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => false,
41
-
// else => true,
42
-
// };
36
+
if (force_blocking) {
37
+
return true;
38
+
}
39
+
return switch (builtin.os.tag) {
40
+
.linux, .macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => false,
41
+
else => true,
42
+
};
43
43
}
44
44
45
45
pub const Config = struct {
···
47
47
address: []const u8 = "127.0.0.1",
48
48
unix_path: ?[]const u8 = null,
49
49
50
-
max_message_size: usize = 65536,
51
-
connection_buffer_size: usize = 4096,
50
+
max_conn: usize = 16_384,
51
+
max_message_size: usize = 65_536,
52
+
connection_buffer_size: usize = 4_096,
52
53
53
54
handshake: Config.Handshake = .{},
54
55
shutdown: Shutdown = .{},
···
89
90
// these aren't used by the server themselves, but both Blocking and NonBlocking
90
91
// workers need them, so we put them here, and reference them in the workers,
91
92
// to avoid the duplicate code.
92
-
bp: buffer.Provider,
93
-
hp: Handshake.Pool,
93
+
handshake_pool: Handshake.Pool,
94
+
buffer_provider: buffer.Provider,
94
95
95
96
const Self = @This();
96
97
97
98
pub fn init(allocator: Allocator, config: Config) !Self {
98
-
var bp = try buffer.Provider.init(allocator, .{
99
+
var handshake_pool = try Handshake.Pool.init(allocator, config.handshake.pool_count orelse 32, config.handshake.max_size orelse 1024, config.handshake.max_headers orelse 10);
100
+
errdefer handshake_pool.deinit();
101
+
102
+
var buffer_provider = try buffer.Provider.init(allocator, .{
99
103
.max = config.max_message_size,
100
104
.count = config.large_buffers.count orelse 8,
101
-
.size = config.large_buffers.size orelse @min(config.max_message_size * 2, config.max_message_size),
105
+
.size = config.large_buffers.size orelse @min(config.connection_buffer_size * 2, config.max_message_size),
102
106
});
103
-
errdefer bp.deinit();
104
-
105
-
var hp = try Handshake.Pool.init(allocator, config.handshake.pool_count orelse 32, config.handshake.max_size orelse 1024, config.handshake.max_headers orelse 10);
106
-
errdefer hp.deinit();
107
+
errdefer buffer_provider.deinit();
107
108
108
109
return .{
109
-
.bp = bp,
110
-
.hp = hp,
110
+
.handshake_pool = handshake_pool,
111
+
.buffer_provider = buffer_provider,
111
112
.cond = .{},
112
113
.config = config,
113
114
.allocator = allocator,
···
115
116
}
116
117
117
118
pub fn deinit(self: *Self) void {
118
-
self.bp.deinit();
119
-
self.hp.deinit();
119
+
self.handshake_pool.deinit();
120
+
self.buffer_provider.deinit();
120
121
}
121
122
122
123
pub fn listenInNewThread(self: *Self, context: anytype) !Thread {
···
129
130
var no_delay = true;
130
131
const address = blk: {
131
132
if (config.unix_path) |unix_path| {
132
-
if (comptime @import("builtin").os.tag == .windows) {
133
+
if (comptime builtin.os.tag == .windows) {
133
134
return error.UnixPathNotSupported;
134
135
}
135
136
no_delay = false;
···
194
195
thrd.join();
195
196
} else {
196
197
defer posix.close(socket);
198
+
const C = @TypeOf(context);
197
199
const signals = try posix.pipe2(.{.NONBLOCK = true});
198
200
199
-
var w = try NonBlocking(H).init(self);
201
+
var w = try NonBlocking(H, C).init(self, signals[1], context);
200
202
defer w.deinit();
201
203
202
-
const thrd = try Thread.spawn(.{}, NonBlocking(H).listen, .{&w, socket, signals[0], context});
204
+
const thrd = try Thread.spawn(.{}, NonBlocking(H, C).listen, .{&w, socket, signals[0]});
203
205
log.info("starting nonblocking worker to listen on {}", .{address});
204
206
205
207
// is this really the best way?
···
221
223
fn Blocking(comptime H: type) type {
222
224
return struct {
223
225
running: bool,
224
-
hp: *Handshake.Pool,
225
-
bp: *buffer.Provider,
226
226
allocator: Allocator,
227
227
config: *const Config,
228
228
handshake_timeout: Timeout,
229
229
handshake_max_size: u16,
230
230
handshake_max_headers: u16,
231
+
handshake_pool: *Handshake.Pool,
232
+
buffer_provider: *buffer.Provider,
231
233
232
234
// protects access to the conn_list and conn_pool
233
235
conn_lock: Thread.Mutex,
···
257
259
errdefer conn_pool.deinit();
258
260
259
261
return .{
260
-
.bp = &server.bp,
261
-
.hp = &server.hp,
262
262
.running = true,
263
263
.config = config,
264
264
.conn_list = .{},
265
265
.conn_lock = .{},
266
266
.conn_pool = conn_pool,
267
267
.allocator = allocator,
268
+
.handshake_pool = &server.handshake_pool,
269
+
.buffer_provider = &server.buffer_provider,
268
270
.handshake_timeout = Timeout.init(config.handshake.timeout orelse MAX_TIMEOUT),
269
271
.handshake_max_size = config.handshake.max_size orelse DEFAULT_MAX_HANDSHAKE_SIZE,
270
272
.handshake_max_headers = config.handshake.max_headers orelse 0,
···
322
324
.socket = socket,
323
325
.handler = null,
324
326
.handshake = null,
325
-
.reader = undefined,
327
+
.reader = null,
326
328
.conn = .{
327
329
._closed = false,
328
-
._rw_mode = .none,
329
-
._io_mode = .blocking,
330
-
._socket_flags = 0, // not needed in the blocking worker
331
330
.address = address,
332
331
.stream = .{.handle = socket},
333
332
},
···
362
361
363
362
// we delay initializing the reader until AFTER the handshake to avoid
364
363
// unecessarily acllocate config.buffer_size
365
-
hc.reader = Reader.init(self.config.connection_buffer_size, self.bp) catch |err| {
364
+
hc.reader = Reader.init(self.config.connection_buffer_size, self.buffer_provider) catch |err| {
366
365
log.err("({}) error creating reader: {}", .{address, err});
367
366
return;
368
367
};
369
368
370
-
var reader = &hc.reader;
369
+
var reader = &hc.reader.?;
371
370
defer reader.deinit();
372
371
373
372
log.debug("({}) connection successfully upgraded", .{address});
···
422
421
var arena = std.heap.ArenaAllocator.init(self.allocator);
423
422
defer arena.deinit();
424
423
425
-
const state = try self.hp.acquire();
426
-
defer self.hp.release(state);
424
+
const state = try self.handshake_pool.acquire();
425
+
defer self.handshake_pool.release(state);
427
426
428
427
const handshake = self.readHandshake(conn, state) catch |err| {
429
428
log.debug("({}) error reading handshake: {}", .{conn.address, err});
430
-
respondToHandshakeError(conn, err);
429
+
switch (err) {
430
+
error.BrokenPipe, error.Closed, error.ConnectionResetByPeer => {},
431
+
else => respondToHandshakeError(conn, err),
432
+
}
431
433
return null;
432
434
};
433
435
···
460
462
while (state.len < buf.len) {
461
463
const n = try posix.read(socket, buf[state.len..]);
462
464
if (n == 0) {
463
-
return error.Close;
465
+
return error.Closed;
464
466
}
465
467
466
468
state.len += n;
···
479
481
};
480
482
}
481
483
482
-
fn NonBlocking(comptime H: type) type {
483
-
const ThreadPool = @import("thread_pool.zig").ThreadPool(NonBlocking(H).dataAvailable);
484
-
484
+
fn NonBlocking(comptime H: type, comptime C: type) type {
485
485
return struct {
486
+
ctx: C,
486
487
// KQueue or Epoll, depending on the platform
487
488
loop: Loop,
488
489
allocator: Allocator,
489
490
config: *const Config,
490
491
491
-
bp: *buffer.Provider,
492
-
hp: *Handshake.Pool,
493
-
tp: ThreadPool,
492
+
// The server creates a pipe. It's used for two things.
493
+
// 1-
494
+
// This worker monitors (via the loop) one end of the pipe.
495
+
// When the server is shutdown, it closes the write end of the pipe,
496
+
// which the worker detects and will then also shutdown.
497
+
// 2-
498
+
// When a thread pool thread is done processing a request, it needs to
499
+
// either rearm the monitor or cleanup the connection. It does so by
500
+
// writing the intFromPtr to the write end of the pipe, which the worker
501
+
// detects and then takes back control fo the connection.
502
+
503
+
// where in buf we've read up to (incase of a partial read)
504
+
signal_pos: usize,
505
+
506
+
// Buffer for reading from the signal fd
507
+
signal_buf: [64]usize,
508
+
509
+
// The write side of the signal. Ideally, this wouldn't be here and the worker
510
+
// would only know about the read side of the signal. This is needed by the
511
+
// server when shutting down (to close), and the thread pool to hand the connection
512
+
// back to the worker. But...the theradpool is really just a method of this
513
+
// worker (NonBlocking(H, C).dataAvailable), so having it here is the most
514
+
// logical place since the thread pool already reaches into the worker for
515
+
// various things (like the handshake_pool, buffer_provider and config).
516
+
signal_write: posix.fd_t,
517
+
// synchronize writes to signal_write
518
+
signal_lock: Thread.Mutex,
519
+
520
+
thread_pool: *ThreadPool,
521
+
handshake_pool: *Handshake.Pool,
522
+
buffer_provider: *buffer.Provider,
494
523
495
524
max_conn: usize,
496
525
conn_list: List(HandlerConn(H)),
497
526
conn_pool: std.heap.MemoryPool(HandlerConn(H)),
498
527
499
528
const Self = @This();
529
+
const ThreadPool = @import("thread_pool.zig").ThreadPool(Self.dataAvailable);
500
530
501
531
const Loop = switch (@import("builtin").os.tag) {
502
-
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KQueue(H),
503
-
.linux => EPoll(H),
532
+
.macos, .ios, .tvos, .watchos, .freebsd, .netbsd, .dragonfly, .openbsd => KQueue,
533
+
.linux => EPoll,
504
534
else => unreachable,
505
535
};
506
536
507
-
pub fn init(server: *Server(H)) !Self {
537
+
pub fn init(server: *Server(H), signal_write: posix.fd_t, ctx: C) !Self {
508
538
const config = &server.config;
509
539
const allocator = server.allocator;
510
540
511
541
const loop = try Loop.init();
512
542
errdefer loop.deinit();
513
543
514
-
const conn_pool = std.heap.MemoryPool(HandlerConn(H)).init(allocator);
544
+
var conn_pool = std.heap.MemoryPool(HandlerConn(H)).init(allocator);
515
545
errdefer conn_pool.deinit();
516
546
517
-
var tp = try ThreadPool.init(allocator, .{
518
-
.count = config.thread_pool.count,
547
+
var thread_pool = try ThreadPool.init(allocator, .{
548
+
.count = config.thread_pool.count orelse 4,
519
549
.backlog = config.thread_pool.backlog orelse 500,
520
550
.buffer_size = config.thread_pool.buffer_size orelse 32_768,
521
551
});
522
-
errdefer tp.deinit();
552
+
errdefer thread_pool.deinit();
523
553
524
554
return .{
555
+
.ctx = ctx,
525
556
.loop = loop,
526
557
.config = config,
527
-
.bp = &server.pb,
528
-
.hp = &server.hb,
529
-
.tp = tp,
558
+
.signal_pos = 0,
559
+
.signal_lock = .{},
560
+
.signal_buf = undefined,
561
+
.signal_write = signal_write,
562
+
.thread_pool = thread_pool,
563
+
.handshake_pool = &server.handshake_pool,
564
+
.buffer_provider = &server.buffer_provider,
530
565
.conn_list = .{},
531
566
.conn_pool = conn_pool,
532
567
.allocator = allocator,
533
-
.max_conn = config.workers.max_conn orelse 8192,
568
+
.max_conn = config.max_conn,
534
569
};
535
570
}
536
571
···
538
573
closeAll(H, self.conn_list, &self.conn_pool, self.config.shutdown);
539
574
self.conn_pool.deinit();
540
575
self.loop.deinit();
541
-
self.bp.deinit();
542
576
}
543
577
544
-
pub fn listen(self: *Self, listener: posix.socket_t, signal: posix.fd_t, context: anytype) void {
545
-
_ = context;
578
+
pub fn listen(self: *Self, listener: posix.socket_t, signal_read: posix.fd_t) void {
546
579
self.loop.monitorAccept(listener) catch |err| {
547
580
log.err("failed to add monitor to listening socket: {}", .{err});
548
581
return;
549
582
};
550
583
551
-
self.loop.monitorSignal(signal) catch |err| {
584
+
self.loop.monitorSignal(signal_read) catch |err| {
552
585
log.err("failed to add monitor to signal pipe: {}", .{err});
553
586
return;
554
587
};
···
570
603
}
571
604
572
605
if (data == 1) {
573
-
log.info("received shutdown signal", .{});
574
-
// for now, activity on the signal can only mean shutdown
575
-
return;
606
+
if (self.processSignal(signal_read)) {
607
+
log.info("received shutdown signal", .{});
608
+
// signal was closed, we're being told to shutdown
609
+
return;
610
+
}
611
+
continue;
576
612
}
577
613
578
614
const hc: *HandlerConn(H) = @ptrFromInt(data);
579
-
self.tp.spawn(self, hc);
615
+
self.thread_pool.spawn(.{self, hc});
580
616
}
581
617
}
582
618
}
···
599
635
600
636
log.debug("({}) connected", .{address});
601
637
602
-
// set non blocking
603
-
const flags = (try posix.fcntl(socket, posix.F.GETFL, 0)) | @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
604
-
_ = try posix.fcntl(socket, posix.F.SETFL, flags);
638
+
{
639
+
// socket is _probably_ in NONBLOCKING mode (it inherits
640
+
// the flag from the listening socket).
641
+
const flags = try posix.fcntl(socket, posix.F.GETFL, 0);
642
+
const nonblocking = @as(u32, @bitCast(posix.O{ .NONBLOCK = true }));
643
+
if (flags & nonblocking == nonblocking) {
644
+
// Yup, it's in nonblocking mode. Disable that flag to
645
+
// put it in blocking mode.
646
+
_ = try posix.fcntl(socket, posix.F.SETFL, flags & ~nonblocking);
647
+
}
648
+
}
605
649
606
650
const hc = try self.conn_pool.create();
607
651
errdefer self.conn_pool.destroy(hc);
···
610
654
.socket = socket,
611
655
.handler = null,
612
656
.handshake = null,
613
-
.reader = undefined,
657
+
.reader = null,
614
658
.conn = .{
615
659
._closed = false,
616
-
._rw_mode = .none,
617
-
._io_mode = .nonblocking,
618
660
.address = address,
619
661
.stream = .{.handle = socket},
620
-
._socket_flags = flags,
621
662
},
622
663
};
623
664
try self.loop.monitorRead(hc, false);
···
626
667
}
627
668
}
628
669
670
+
fn processSignal(self: *Self, signal: posix.fd_t) bool {
671
+
const s_t = @sizeOf(usize);
672
+
673
+
const buflen = @typeInfo(@TypeOf(self.signal_buf)).Array.len * @sizeOf(usize);
674
+
const buf: *[buflen]u8 = @ptrCast(&self.signal_buf);
675
+
const start = self.signal_pos;
676
+
677
+
const n = posix.read(signal, buf[start..]) catch |err| switch (err) {
678
+
error.WouldBlock => return false,
679
+
else => 0,
680
+
};
681
+
682
+
if (n == 0) {
683
+
// assume closed, indicating a shutdown
684
+
return true;
685
+
}
686
+
687
+
const pos = start + n;
688
+
const connections = pos / s_t;
689
+
690
+
for (0..connections) |i| {
691
+
const data_start = (i * s_t);
692
+
const data_end = data_start + s_t;
693
+
const hc: *HandlerConn(H) = @ptrFromInt(@as(*usize, @alignCast(@ptrCast(buf[data_start..data_end]))).*);
694
+
if (hc.conn.isClosed()) {
695
+
self.cleanupConn(hc);
696
+
} else {
697
+
self.loop.monitorRead(hc, true) catch |err| {
698
+
log.debug("({}) failed to add read event monitor: {}", .{hc.conn.address, err});
699
+
hc.conn.close();
700
+
self.cleanupConn(hc);
701
+
};
702
+
}
703
+
}
704
+
705
+
const partial_len = @mod(pos, s_t);
706
+
const partial_start = pos - partial_len;
707
+
for (0..partial_len) |i| {
708
+
buf[i] = buf[partial_start + i];
709
+
}
710
+
self.signal_pos = partial_len;
711
+
return false;
712
+
}
629
713
630
714
// Called in a thread-pool thread/
631
715
// !! Access to self has to be synchronized !!
632
-
// Will not be called concurrently for the same hc.
633
-
// But, as always, the app can have a reference to hc.conn, so hc.conn
634
-
// has to be synchornized (which its methods are).
635
-
fn dataAvailable(self: *Self, hc: *HandlerConn, thread_buf: []u8) void {
636
-
if (hc.handshake) |*state| {
637
-
self.doHandshake(hc, state) catch {}; // TODO log
716
+
fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) void {
717
+
// Both doHandshake and processIncoming return !bool. This is to deal with
718
+
// Zig's lack of error payloads. The error is used for uncaught errors -
719
+
// usually rare things. The bool, when false, is used to indicate that
720
+
// the connection should be closed, but not to log a generic uncaught
721
+
// error message (likely because something more meaningful was already logged)
722
+
var ok = true;
723
+
if (hc.handler == null) {
724
+
// if we don't have a handler yet, then we haven't done our handshake. It's that simple.
725
+
ok = self.doHandshake(hc) catch |err| blk: {
726
+
log.warn("({}) uncaugh error handling handshake: {}", .{hc.conn.address, err});
727
+
break :blk false;
728
+
};
638
729
} else {
639
-
self.processIncoming(hc, thread_buf) catch {}; // TODO log
730
+
ok = processIncoming(hc, thread_buf) catch |err| blk: {
731
+
log.warn("({}) uncaugh error handling data: {}", .{hc.conn.address, err});
732
+
break :blk false;
733
+
};
640
734
}
641
-
}
642
735
643
-
fn doHandshake(self: *Self, hc: *HandlerConn, state: *Handshake.State) !void {
644
-
_ = self;
645
-
_ = hc;
646
-
_ = state;
647
-
}
736
+
if (ok == false) {
737
+
hc.conn.close();
738
+
}
648
739
740
+
const buf = std.mem.asBytes(&@intFromPtr(hc));
741
+
var to_write: usize = @sizeOf(usize);
742
+
const pipe = self.signal_write;
649
743
650
-
fn doDataAvailable(self: *Self, hc: *HandlerConn, thread_buf: []u8) !void {
651
-
_ = self;
652
-
_ = hc;
653
-
_ = thread_buf;
744
+
self.signal_lock.lock();
745
+
defer self.signal_lock.unlock();
746
+
while (to_write > 0) {
747
+
to_write -= std.posix.write(pipe, buf) catch @panic("TODO");
748
+
}
654
749
}
655
750
656
-
fn closeConn(self: *Self, hc: *HandlerConn) void {
657
-
posix.close(hc.conn.stream.handle);
658
-
self.conn_list.remove(hc.node);
659
-
self.coon_pool.destroy(hc);
660
-
}
661
-
};
662
-
}
751
+
// This is being usexecuted in a thread-pool thread. Access to self has to
752
+
// be synchronized.
753
+
// Normally access to hc.conn also has to be synchronized, but for doHandshake,
754
+
// that's only true after H.init is called (at which point, the application
755
+
// has access to &hc.conn and could do things concurrently to it).
756
+
fn doHandshake(self: *Self, hc: *HandlerConn(H)) !bool {
757
+
var state = hc.handshake orelse blk: {
758
+
const s = try self.handshake_pool.acquire();
759
+
hc.handshake = s;
760
+
break :blk s;
761
+
};
663
762
664
-
fn KQueue(comptime H: type) type {
665
-
return struct {
666
-
q: i32,
667
-
change_count: usize,
668
-
change_buffer: [16]Kevent,
669
-
event_list: [64]Kevent,
763
+
var buf = state.buf;
764
+
var conn = &hc.conn;
765
+
const len = state.len;
670
766
671
-
const Kevent = posix.Kevent;
767
+
if (len == buf.len) {
768
+
log.warn("({}) handshake request exceeded maximum configured size ({d})", .{conn.address, buf.len});
769
+
return false;
770
+
}
672
771
673
-
fn init() !KQueue {
674
-
return .{
675
-
.q = try posix.kqueue(),
676
-
.change_count = 0,
677
-
.change_buffer = undefined,
678
-
.event_list = undefined,
772
+
// Normally, we need to manage the reality of having nonblocking reads
773
+
// and blocking writes. But here this isn't a concern. (a) the app hasn't
774
+
// been giving &conn yet, so can't be doing writes and (b) doHandshake
775
+
// is only called by a single thread per hc.
776
+
const n = posix.read(hc.socket, buf[len..]) catch |err| {
777
+
switch (err) {
778
+
error.BrokenPipe, error.ConnectionResetByPeer => log.debug("({}) handshake connection closed: {}", .{conn.address, err}),
779
+
else => log.warn("({}) handshake error reading from socket: {}", .{conn.address, err}),
780
+
}
781
+
return false;
679
782
};
680
-
}
681
783
682
-
fn deinit(self: KQueue) void {
683
-
posix.close(self.q);
684
-
}
784
+
if (n == 0) {
785
+
return error.Closed;
786
+
}
787
+
state.len += n;
685
788
686
-
fn monitorAccept(self: *KQueue, fd: c_int) !void {
687
-
try self.change(fd, 0, posix.system.EVFILT_READ, posix.system.EV_ADD);
688
-
}
789
+
const handshake = Handshake.parse(state) catch |err| {
790
+
log.debug("({}) error parsing handshake: {}", .{conn.address, err});
791
+
respondToHandshakeError(conn, err);
792
+
return false;
793
+
} orelse {
794
+
// we need more data
795
+
return true;
796
+
};
689
797
690
-
fn monitorSignal(self: *KQueue, fd: c_int) !void {
691
-
try self.change(fd, 1, posix.system.EVFILT_READ, posix.system.EV_ADD);
692
-
}
798
+
self.handshake_pool.release(state);
799
+
hc.handshake = null;
693
800
694
-
fn monitorRead(self: *KQueue, hc: *HandlerConn(H), comptime rearm: bool) !void {
695
-
_ = rearm; // used by epoll
696
-
try self.change(hc.socket, @intFromPtr(hc), posix.system.EVFILT_READ, posix.system.EV_ADD | posix.system.EV_ENABLE | posix.system.EV_DISPATCH);
697
-
}
801
+
// After this, the app has access to &hc.conn, so any access to the
802
+
// conn has to be synchronized (which the conn does internally).
698
803
699
-
fn remove(self: *KQueue, hc: *HandlerConn(H)) !void {
700
-
try self.change(hc.socket, 0, posix.system.EVFILT_READ, posix.system.EV_DELETE);
701
-
}
804
+
hc.handler = H.init(handshake, conn, self.ctx) catch |err| {
805
+
if (comptime std.meta.hasFn(H, "handshakeErrorResponse")) {
806
+
preHandOffWrite(H.handshakeErrorResponse(err));
807
+
} else {
808
+
respondToHandshakeError(conn, err);
809
+
}
810
+
log.debug("({}) " ++ @typeName(H) ++ ".init rejected request {}", .{conn.address, err});
811
+
return false;
812
+
};
702
813
703
-
fn change(self: *KQueue, fd: posix.fd_t, data: usize, filter: i16, flags: u16) !void {
704
-
var change_count = self.change_count;
705
-
var change_buffer = &self.change_buffer;
814
+
try conn.writeFramed(&handshake.reply());
706
815
707
-
if (change_count == change_buffer.len) {
708
-
// calling this with an empty event_list will return immediate
709
-
_ = try posix.kevent(self.q, change_buffer, &[_]Kevent{}, null);
710
-
change_count = 0;
816
+
if (comptime std.meta.hasFn(H, "afterInit")) {
817
+
hc.handler.?.afterInit() catch |err| {
818
+
log.debug("({}) " ++ @typeName(H) ++ ".afterInit error: {}", .{conn.address, err});
819
+
return false;
820
+
};
711
821
}
712
-
change_buffer[change_count] = .{
713
-
.ident = @intCast(fd),
714
-
.filter = filter,
715
-
.flags = flags,
716
-
.fflags = 0,
717
-
.data = 0,
718
-
.udata = data,
822
+
823
+
hc.reader = Reader.init(self.config.connection_buffer_size, self.buffer_provider) catch |err| {
824
+
log.err("({}) error creating reader: {}", .{conn.address, err});
825
+
return false;
719
826
};
720
-
self.change_count = change_count + 1;
827
+
return true;
721
828
}
722
829
723
-
fn wait(self: *KQueue) !Iterator {
724
-
const event_list = &self.event_list;
725
-
const event_count = try posix.kevent(self.q, self.change_buffer[0..self.change_count], event_list, null);
726
-
self.change_count = 0;
830
+
fn processIncoming(hc: *HandlerConn(H), thread_buf: []u8) !bool {
831
+
_ = thread_buf;
727
832
728
-
return .{
729
-
.index = 0,
730
-
.events = event_list[0..event_count],
833
+
var conn = &hc.conn;
834
+
var reader = &hc.reader.?;
835
+
reader.fill(conn.stream) catch |err| {
836
+
switch (err) {
837
+
error.BrokenPipe, error.Closed, error.ConnectionResetByPeer => log.debug("({}) connection closed: {}", .{conn.address, err}),
838
+
else => log.warn("({}) error reading from connection: {}", .{conn.address, err}),
839
+
}
840
+
return false;
731
841
};
842
+
843
+
while (true) {
844
+
const has_more, const message = reader.read() catch |err| {
845
+
switch (err) {
846
+
error.LargeControl => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {},
847
+
error.ReservedFlags => conn.writeFramed(CLOSE_PROTOCOL_ERROR) catch {},
848
+
else => {},
849
+
}
850
+
log.debug("({}) invalid websocket packet: {}", .{conn.address, err});
851
+
return false;
852
+
} orelse {
853
+
// we need more data
854
+
return true;
855
+
};
856
+
857
+
handleMessage(H, hc, message) catch |err| {
858
+
log.warn("({}) handle message error: {}", .{conn.address, err});
859
+
return false;
860
+
};
861
+
862
+
if (conn.isClosed()) {
863
+
return false;
864
+
}
865
+
866
+
if (has_more == false) {
867
+
// we don't have more data ready to be processed in our buffer
868
+
return true;
869
+
}
870
+
}
732
871
}
733
872
734
-
const Iterator = struct {
735
-
index: usize,
736
-
events: []Kevent,
873
+
fn cleanupConn(self: *Self, hc: *HandlerConn(H)) void {
874
+
if (hc.handshake) |hs| {
875
+
self.handshake_pool.release(hs);
876
+
}
877
+
878
+
if (hc.reader) |*r| {
879
+
r.deinit();
880
+
}
737
881
738
-
fn next(self: *Iterator) ?usize {
739
-
const index = self.index;
740
-
const events = self.events;
741
-
if (index == events.len) {
742
-
return null;
882
+
if (comptime std.meta.hasFn(H, "close")) {
883
+
if (hc.handler) |*h| {
884
+
h.close();
743
885
}
744
-
self.index = index + 1;
745
-
return self.events[index].udata;
746
886
}
747
-
};
887
+
self.conn_list.remove(hc);
888
+
self.conn_pool.destroy(hc);
889
+
}
748
890
};
749
891
}
750
892
751
-
fn EPoll(comptime H: type) type {
752
-
return struct {
753
-
q: i32,
754
-
event_list: [64]EpollEvent,
893
+
const KQueue = struct {
894
+
q: i32,
895
+
change_count: usize,
896
+
change_buffer: [16]Kevent,
897
+
event_list: [64]Kevent,
898
+
899
+
const Self = @This();
900
+
const Kevent = posix.Kevent;
755
901
756
-
const linux = std.os.linux;
757
-
const EpollEvent = linux.epoll_event;
902
+
fn init() !Self {
903
+
return .{
904
+
.q = try posix.kqueue(),
905
+
.change_count = 0,
906
+
.change_buffer = undefined,
907
+
.event_list = undefined,
908
+
};
909
+
}
758
910
759
-
fn init() !EPoll {
760
-
return .{
761
-
.event_list = undefined,
762
-
.q = try posix.epoll_create1(0),
763
-
};
764
-
}
911
+
fn deinit(self: Self) void {
912
+
posix.close(self.q);
913
+
}
765
914
766
-
fn deinit(self: EPoll) void {
767
-
posix.close(self.q);
768
-
}
915
+
fn monitorAccept(self: *Self, fd: c_int) !void {
916
+
try self.change(fd, 0, posix.system.EVFILT_READ, posix.system.EV_ADD);
917
+
}
769
918
770
-
fn monitorAccept(self: *EPoll, fd: c_int) !void {
771
-
var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 0 } };
772
-
return std.posix.epoll_ctl(self.q, linux.EPOLL.CTL_ADD, fd, &event);
773
-
}
919
+
fn monitorSignal(self: *Self, fd: c_int) !void {
920
+
try self.change(fd, 1, posix.system.EVFILT_READ, posix.system.EV_ADD);
921
+
}
774
922
775
-
fn monitorSignal(self: *EPoll, fd: c_int) !void {
776
-
var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 1 } };
777
-
return std.posix.epoll_ctl(self.q, linux.EPOLL.CTL_ADD, fd, &event);
778
-
}
923
+
fn monitorRead(self: *Self, hc: anytype, comptime rearm: bool) !void {
924
+
_ = rearm; // used by epoll
925
+
try self.change(hc.socket, @intFromPtr(hc), posix.system.EVFILT_READ, posix.system.EV_ADD | posix.system.EV_ENABLE | posix.system.EV_DISPATCH);
926
+
}
779
927
780
-
fn monitorRead(self: *EPoll, hc: *HandlerConn(H), comptime rearm: bool) !void {
781
-
const op = if (rearm) linux.EPOLL.CTL_MOD else linux.EPOLL.CTL_ADD;
782
-
var event = linux.epoll_event{ .events = linux.EPOLL.IN | linux.EPOLL.ONESHOT, .data = .{ .ptr = @intFromPtr(hc) } };
783
-
return posix.epoll_ctl(self.q, op, hc.socket, &event);
784
-
}
928
+
fn change(self: *Self, fd: posix.fd_t, data: usize, filter: i16, flags: u16) !void {
929
+
var change_count = self.change_count;
930
+
var change_buffer = &self.change_buffer;
785
931
786
-
fn remove(self: *EPoll, hc: *HandlerConn(H)) !void {
787
-
return posix.epoll_ctl(self.q, linux.EPOLL.CTL_DEL, hc.socket, null);
932
+
if (change_count == change_buffer.len) {
933
+
// calling this with an empty event_list will return immediate
934
+
_ = try posix.kevent(self.q, change_buffer, &[_]Kevent{}, null);
935
+
change_count = 0;
788
936
}
937
+
change_buffer[change_count] = .{
938
+
.ident = @intCast(fd),
939
+
.filter = filter,
940
+
.flags = flags,
941
+
.fflags = 0,
942
+
.data = 0,
943
+
.udata = data,
944
+
};
945
+
self.change_count = change_count + 1;
946
+
}
789
947
790
-
fn wait(self: *EPoll) !Iterator {
791
-
const event_list = &self.event_list;
792
-
const event_count = blk: while (true) {
793
-
const rc = linux.syscall6(
794
-
.epoll_pwait2,
795
-
@as(usize, @bitCast(@as(isize, self.q))),
796
-
@intFromPtr(event_list.ptr),
797
-
event_list.len,
798
-
0,
799
-
0,
800
-
@sizeOf(linux.sigset_t),
801
-
);
948
+
fn wait(self: *Self) !Iterator {
949
+
const event_list = &self.event_list;
950
+
const event_count = try posix.kevent(self.q, self.change_buffer[0..self.change_count], event_list, null);
951
+
self.change_count = 0;
952
+
953
+
return .{
954
+
.index = 0,
955
+
.events = event_list[0..event_count],
956
+
};
957
+
}
802
958
803
-
// taken from std.os.epoll_waits
804
-
switch (posix.errno(rc)) {
805
-
.SUCCESS => break :blk @as(usize, @intCast(rc)),
806
-
.INTR => continue,
807
-
.BADF => unreachable,
808
-
.FAULT => unreachable,
809
-
.INVAL => unreachable,
810
-
else => unreachable,
811
-
}
812
-
};
959
+
const Iterator = struct {
960
+
index: usize,
961
+
events: []Kevent,
813
962
814
-
return .{
815
-
.index = 0,
816
-
.events = event_list[0..event_count],
817
-
};
963
+
fn next(self: *Iterator) ?usize {
964
+
const index = self.index;
965
+
const events = self.events;
966
+
if (index == events.len) {
967
+
return null;
968
+
}
969
+
self.index = index + 1;
970
+
return self.events[index].udata;
818
971
}
972
+
};
973
+
};
819
974
820
-
const Iterator = struct {
821
-
index: usize,
822
-
events: []EpollEvent,
975
+
const EPoll = struct {
976
+
q: i32,
977
+
event_list: [64]EpollEvent,
978
+
979
+
const linux = std.os.linux;
980
+
981
+
const Self = @This();
982
+
const EpollEvent = linux.epoll_event;
983
+
984
+
fn init() !Self {
985
+
return .{
986
+
.event_list = undefined,
987
+
.q = try posix.epoll_create1(0),
988
+
};
989
+
}
990
+
991
+
fn deinit(self: Self) void {
992
+
posix.close(self.q);
993
+
}
994
+
995
+
fn monitorAccept(self: *Self, fd: c_int) !void {
996
+
var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 0 } };
997
+
return std.posix.epoll_ctl(self.q, linux.EPOLL.CTL_ADD, fd, &event);
998
+
}
999
+
1000
+
fn monitorSignal(self: *Self, fd: c_int) !void {
1001
+
var event = linux.epoll_event{ .events = linux.EPOLL.IN, .data = .{ .ptr = 1 } };
1002
+
return std.posix.epoll_ctl(self.q, linux.EPOLL.CTL_ADD, fd, &event);
1003
+
}
1004
+
1005
+
fn monitorRead(self: *Self, hc: anytype, comptime rearm: bool) !void {
1006
+
const op = if (rearm) linux.EPOLL.CTL_MOD else linux.EPOLL.CTL_ADD;
1007
+
var event = linux.epoll_event{ .events = linux.EPOLL.IN | linux.EPOLL.ONESHOT, .data = .{ .ptr = @intFromPtr(hc) } };
1008
+
return posix.epoll_ctl(self.q, op, hc.socket, &event);
1009
+
}
1010
+
1011
+
fn wait(self: *Self) !Iterator {
1012
+
const event_list = &self.event_list;
1013
+
const event_count = blk: while (true) {
1014
+
const rc = linux.syscall6(
1015
+
.epoll_pwait2,
1016
+
@as(usize, @bitCast(@as(isize, self.q))),
1017
+
@intFromPtr(event_list.ptr),
1018
+
event_list.len,
1019
+
0,
1020
+
0,
1021
+
@sizeOf(linux.sigset_t),
1022
+
);
823
1023
824
-
fn next(self: *Iterator) ?usize {
825
-
const index = self.index;
826
-
const events = self.events;
827
-
if (index == events.len) {
828
-
return null;
829
-
}
830
-
self.index = index + 1;
831
-
return self.events[index].data.ptr;
1024
+
// taken from std.os.epoll_waits
1025
+
switch (posix.errno(rc)) {
1026
+
.SUCCESS => break :blk @as(usize, @intCast(rc)),
1027
+
.INTR => continue,
1028
+
.BADF => unreachable,
1029
+
.FAULT => unreachable,
1030
+
.INVAL => unreachable,
1031
+
else => unreachable,
832
1032
}
833
1033
};
1034
+
1035
+
return .{
1036
+
.index = 0,
1037
+
.events = event_list[0..event_count],
1038
+
};
1039
+
}
1040
+
1041
+
const Iterator = struct {
1042
+
index: usize,
1043
+
events: []EpollEvent,
1044
+
1045
+
fn next(self: *Iterator) ?usize {
1046
+
const index = self.index;
1047
+
const events = self.events;
1048
+
if (index == events.len) {
1049
+
return null;
1050
+
}
1051
+
self.index = index + 1;
1052
+
return self.events[index].data.ptr;
1053
+
}
834
1054
};
835
-
}
1055
+
};
836
1056
837
1057
// We want to extract as much common logic as possible from the Blocking and
838
1058
// NonBlockign workers. The code from this point on is meant to be used with
···
857
1077
return struct {
858
1078
conn: Conn,
859
1079
handler: ?H,
860
-
reader: Reader,
1080
+
reader: ?Reader,
861
1081
socket: posix.socket_t, // denormalization from conn.stream.handle
862
-
handshake: ?Handshake.State,
1082
+
handshake: ?*Handshake.State,
863
1083
next: ?*HandlerConn(H) = null,
864
1084
prev: ?*HandlerConn(H) = null,
865
1085
};
···
867
1087
868
1088
pub const Conn = struct {
869
1089
_closed: bool,
870
-
_io_mode: IOMode,
871
-
_rw_mode: RWMode,
872
-
_socket_flags: usize,
873
1090
stream: net.Stream,
874
1091
address: net.Address,
875
1092
876
-
const IOMode = enum {
877
-
blocking,
878
-
nonblocking,
879
-
};
880
-
881
-
const RWMode = enum {
882
-
none,
883
-
read,
884
-
write,
885
-
};
886
-
887
-
pub fn blocking(self: *Conn) !void {
888
-
// in blockingMode, io_mode is ALWAYS blocking, so we can skip this
889
-
if (comptime blockingMode() == false) {
890
-
if (@atomicRmw(IOMode, &self._io_mode, .Xchg, .blocking, .monotonic) == .nonblock) {
891
-
// we don't care if the above check + this call isn't atomic.
892
-
// at worse, we're doing an unecessary syscall
893
-
_ = try posix.fcntl(self.stream.handle, posix.F.SETFL, self.socket_flags & ~@as(u32, @bitCast(posix.O{ .NONBLOCK = true })));
894
-
}
895
-
}
896
-
}
897
-
898
1093
pub fn isClosed(self: *Conn) bool {
899
1094
return @atomicLoad(bool, &self._closed, .monotonic);
900
1095
}
···
978
1173
979
1174
fn handleMessage(comptime H: type, hc: *HandlerConn(H), message: Message) !void {
980
1175
const message_type = message.type;
981
-
defer hc.reader.done(message_type);
1176
+
defer hc.reader.?.done(message_type);
982
1177
983
1178
log.debug("({}) received {s} message", .{hc.conn.address, @tagName(message_type)});
984
1179
···
1098
1293
1099
1294
var pos: usize = 0;
1100
1295
while (pos < response.len) {
1101
-
const n = posix.write(socket, response[pos..]) catch |err| switch (err) {
1102
-
error.WouldBlock => {
1103
-
if (conn._io_mode == .blocking) {
1104
-
// We were in blocking mode, which means we reached our timeout
1105
-
// We're done trying to send the response.
1106
-
return;
1107
-
}
1108
-
try conn.blocking();
1109
-
continue;
1110
-
},
1111
-
else => return,
1112
-
};
1113
-
1296
+
const n = posix.write(socket, response[pos..]) catch return;
1114
1297
if (n == 0) {
1115
1298
// closed
1116
1299
return;
+1
-1
src/server/thread_pool.zig
+1
-1
src/server/thread_pool.zig
+21
support/autobahn/client/build.zig
+21
support/autobahn/client/build.zig
···
1
+
const std = @import("std");
2
+
3
+
pub fn build(b: *std.Build) void {
4
+
const target = b.standardTargetOptions(.{});
5
+
const optimize = b.standardOptimizeOption(.{});
6
+
7
+
const exe = b.addExecutable(.{
8
+
.name = "autobahn_test_client",
9
+
.root_source_file = b.path("main.zig"),
10
+
.target = target,
11
+
.optimize = optimize,
12
+
});
13
+
exe.root_module.addImport("websocket", b.dependency("websocket", .{}).module("websocket"));
14
+
15
+
b.installArtifact(exe);
16
+
17
+
const run_cmd = b.addRunArtifact(exe);
18
+
run_cmd.step.dependOn(b.getInstallStep());
19
+
const run_step = b.step("run", "Run the app");
20
+
run_step.dependOn(&run_cmd.step);
21
+
}
+10
support/autobahn/client/build.zig.zon
+10
support/autobahn/client/build.zig.zon
+1
-1
support/autobahn/client/run.sh
+1
-1
support/autobahn/client/run.sh
+21
support/autobahn/server/build.zig
+21
support/autobahn/server/build.zig
···
1
+
const std = @import("std");
2
+
3
+
pub fn build(b: *std.Build) void {
4
+
const target = b.standardTargetOptions(.{});
5
+
const optimize = b.standardOptimizeOption(.{});
6
+
7
+
const exe = b.addExecutable(.{
8
+
.name = "autobahn_test_server",
9
+
.root_source_file = b.path("main.zig"),
10
+
.target = target,
11
+
.optimize = optimize,
12
+
});
13
+
exe.root_module.addImport("websocket", b.dependency("websocket", .{}).module("websocket"));
14
+
15
+
b.installArtifact(exe);
16
+
17
+
const run_cmd = b.addRunArtifact(exe);
18
+
run_cmd.step.dependOn(b.getInstallStep());
19
+
const run_step = b.step("run", "Run the app");
20
+
run_step.dependOn(&run_cmd.step);
21
+
}
+10
support/autobahn/server/build.zig.zon
+10
support/autobahn/server/build.zig.zon
+4
-3
support/autobahn/server/run.sh
+4
-3
support/autobahn/server/run.sh
···
3
3
set -o nounset
4
4
5
5
root=$(dirname $(realpath $BASH_SOURCE))
6
+
echo "starting server..."
7
+
cd support/autobahn/server/ && zig build -Doptimize=ReleaseFast run &
8
+
pid=$!
6
9
7
-
zig run autobahn_server.zig &
8
-
pid=$!
9
-
sleep 2 # give chance for socket to listen
10
+
sleep 3 # give chance for socket to listen
10
11
trap "kill ${pid} || true;" EXIT
11
12
12
13
docker run --rm \