karlseguin.tngl.sh
A websocket implementation for zig
final-ish changes to allow integration with httpz
author
Karl Seguin
date
(Aug 3, 2024, 11:26 AM +0800)
commit
56c58d64
56c58d644864dcc8f140e9d70d1545611aafa909
parent
bef44695
bef44695a8da35ba08982dc938b73d2d5d640377
+84
-168
+4
-5
Makefile
+4
-5
Makefile
···
2
2
zig ?= zig
3
3
.PHONY: t
4
4
t:
5
-
TEST_FILTER='${F}' '${zig}' build test -freference-trace --summary all
6
-
TEST_FILTER='${F}' '${zig}' build test_blocking -freference-trace --summary all
5
+
TEST_FILTER='${F}' '${zig}' build test -Dforce_blocking=false -freference-trace --summary all
6
+
TEST_FILTER='${F}' '${zig}' build test -Dforce_blocking=true -freference-trace --summary all
7
7
8
8
.PHONY: tn
9
9
tn:
10
-
TEST_FILTER='${F}' '${zig}' build test -freference-trace --summary all
10
+
TEST_FILTER='${F}' '${zig}' build test -Dforce_blocking=false -freference-trace --summary all
11
11
12
12
.PHONY: tb
13
13
tb:
14
-
TEST_FILTER='${F}' '${zig}' build test_blocking -freference-trace --summary all
14
+
TEST_FILTER='${F}' '${zig}' build test -Dforce_blocking=true -freference-trace --summary all
15
15
16
16
.PHONY: abs
17
17
abs:
18
18
bash support/autobahn/server/run.sh
19
-
20
19
21
20
.PHONY: abc
22
21
abc:
+6
-24
build.zig
+6
-24
build.zig
···
10
10
11
11
{
12
12
const options = b.addOptions();
13
-
options.addOption(bool, "force_blocking", false);
14
-
websocket_module.addOptions("ws_build", options);
15
-
}
16
-
17
-
{
18
-
// run tests in nonblocking mode (only meaningful where epoll/kqueue is supported)
19
-
const tests = b.addTest(.{
20
-
.root_source_file = b.path("src/websocket.zig"),
21
-
.target = target,
22
-
.optimize = optimize,
23
-
.test_runner = b.path("test_runner.zig"),
24
-
});
25
-
tests.linkLibC();
26
-
const options = b.addOptions();
27
-
options.addOption(bool, "force_blocking", false);
28
-
tests.root_module.addOptions("build", options);
29
-
const run_test = b.addRunArtifact(tests);
30
-
run_test.has_side_effects = true;
31
-
32
-
const test_step = b.step("test", "Run tests");
33
-
test_step.dependOn(&run_test.step);
13
+
options.addOption(bool, "websocket_blocking", false);
14
+
websocket_module.addOptions("build", options);
34
15
}
35
16
36
17
{
37
-
// run tests in blocking mode
18
+
// run tests
38
19
const tests = b.addTest(.{
39
20
.root_source_file = b.path("src/websocket.zig"),
40
21
.target = target,
···
42
23
.test_runner = b.path("test_runner.zig"),
43
24
});
44
25
tests.linkLibC();
26
+
const force_blocking = b.option(bool, "force_blocking", "Force blocking mode") orelse false;
45
27
const options = b.addOptions();
46
-
options.addOption(bool, "force_blocking", true);
28
+
options.addOption(bool, "websocket_blocking", force_blocking);
47
29
tests.root_module.addOptions("build", options);
48
30
49
31
const run_test = b.addRunArtifact(tests);
50
32
run_test.has_side_effects = true;
51
33
52
-
const test_step = b.step("test_blocking", "Run tests");
34
+
const test_step = b.step("test", "Run tests");
53
35
test_step.dependOn(&run_test.step);
54
36
}
55
37
}
+1
-1
src/server/handshake.zig
+1
-1
src/server/handshake.zig
···
56
56
},
57
57
10 => if (eql("connection", name)) {
58
58
// find if connection header has upgrade in it, example header:
59
-
// Connection: keep-alive, Upgrade
59
+
// Connection: keep-alive, Upgrade
60
60
if (std.ascii.indexOfIgnoreCase(value, "upgrade") == null) {
61
61
return error.InvalidConnection;
62
62
}
+72
-137
src/server/server.zig
+72
-137
src/server/server.zig
···
427
427
fn NonBlocking(comptime H: type, comptime C: type) type {
428
428
return struct {
429
429
ctx: C,
430
+
431
+
// KQueue or Epoll, depending on the platform
432
+
loop: Loop,
433
+
thread_pool: *ThreadPool,
434
+
430
435
handshake_timeout: u32,
431
436
handshake_pool: *Handshake.Pool,
432
-
base: NonBlockingBase(H, Self.dataAvailable, true),
437
+
base: NonBlockingBase(H, true),
433
438
434
439
const Self = @This();
440
+
const ThreadPool = @import("thread_pool.zig").ThreadPool(Self.dataAvailable);
435
441
436
442
pub fn init(allocator: Allocator, state: *WorkerState, ctx: C) !Self {
437
-
var base = try NonBlockingBase(H, Self.dataAvailable, true).init(allocator, state);
443
+
var base = try NonBlockingBase(H, true).init(allocator, state);
438
444
errdefer base.deinit();
439
445
446
+
const loop = try Loop.init();
447
+
errdefer loop.deinit();
448
+
449
+
const config = &state.config;
450
+
var thread_pool = try ThreadPool.init(allocator, .{
451
+
.count = config.thread_pool.count orelse 4,
452
+
.backlog = config.thread_pool.backlog orelse 500,
453
+
.buffer_size = config.thread_pool.buffer_size orelse 32_768,
454
+
});
455
+
errdefer thread_pool.deinit();
456
+
440
457
return .{
441
458
.ctx = ctx,
459
+
.loop = loop,
442
460
.base = base,
461
+
.thread_pool = thread_pool,
443
462
.handshake_pool = state.handshake_pool,
444
463
.handshake_timeout = state.config.handshake.timeout,
445
464
};
446
465
}
447
466
448
467
pub fn deinit(self: *Self) void {
468
+
self.loop.deinit();
449
469
self.base.deinit();
470
+
self.thread_pool.deinit();
450
471
}
451
472
452
473
fn run(self: *Self, listener: posix.socket_t, signal: posix.fd_t) void {
453
-
if (self.base.setupMonitors(listener, signal) == false) {
474
+
self.loop.monitorAccept(listener) catch |err| {
475
+
log.err("failed to add monitor to listening socket: {}", .{err});
476
+
return;
477
+
};
478
+
479
+
self.loop.monitorSignal(signal) catch |err| {
480
+
log.err("failed to add monitor to signal pipe: {}", .{err});
454
481
return;
455
-
}
482
+
};
456
483
457
-
const thread_pool = self.base.thread_pool;
484
+
const thread_pool = self.thread_pool;
458
485
const conn_manager = &self.base.conn_manager;
459
486
const handshake_timeout = self.handshake_timeout;
460
487
···
469
496
break :blk null;
470
497
};
471
498
472
-
var it = self.base.loop.wait(timeout) catch |err| {
499
+
var it = self.loop.wait(timeout) catch |err| {
473
500
log.err("failed to wait on events: {}", .{err});
474
501
std.time.sleep(std.time.ns_per_s);
475
502
continue;
···
571
598
_ = try posix.fcntl(socket, posix.F.SETFL, flags & ~nonblocking);
572
599
}
573
600
}
574
-
return self.base.newConn(socket, address, now);
601
+
const hc = try self.base.newConn(socket, address, now);
602
+
self.loop.monitorRead(hc, false) catch |err| {
603
+
self.base.cleanupConn(hc);
604
+
return err;
605
+
};
575
606
}
576
607
}
577
608
···
596
627
success = self.base.dataAvailable(hc, thread_buf);
597
628
}
598
629
599
-
self.base.postProcess(hc, success);
630
+
var conn = &hc.conn;
631
+
var closed: bool = undefined;
632
+
if (success == false) {
633
+
conn.close();
634
+
closed = true;
635
+
} else {
636
+
closed = conn.isClosed();
637
+
}
638
+
639
+
if (closed) {
640
+
self.base.cleanupConn(hc);
641
+
} else {
642
+
self.loop.monitorRead(hc, true) catch |err| {
643
+
log.debug("({}) failed to add read event monitor: {}", .{conn.address, err});
644
+
conn.close();
645
+
self.base.cleanupConn(hc);
646
+
};
647
+
}
600
648
}
601
649
602
650
fn dataForHandshake(self: *Self, hc: *HandlerConn(H)) !bool {
···
614
662
};
615
663
}
616
664
617
-
fn NonBlockingBase(comptime H: type, comptime TP: anytype, comptime MANAGE_HS: bool) type {
665
+
fn NonBlockingBase(comptime H: type, comptime MANAGE_HS: bool) type {
618
666
return struct {
619
-
// KQueue or Epoll, depending on the platform
620
-
loop: Loop,
621
667
allocator: Allocator,
622
668
623
-
thread_pool: *ThreadPool,
624
669
buffer_provider: *buffer.Provider,
625
670
626
671
// App can configure the use of a "small" buffer pool. This is the difference
···
632
677
conn_manager: ConnManager(H, MANAGE_HS),
633
678
634
679
const Self = @This();
635
-
const ThreadPool = @import("thread_pool.zig").ThreadPool(TP);
636
680
637
681
fn init(allocator: Allocator, state: *WorkerState) !Self {
638
682
const config = &state.config;
639
683
640
-
const loop = try Loop.init();
641
-
errdefer loop.deinit();
642
-
643
684
var conn_manager = try ConnManager(H, MANAGE_HS).init(allocator);
644
685
errdefer conn_manager.deinit();
645
686
646
-
var thread_pool = try ThreadPool.init(allocator, .{
647
-
.count = config.thread_pool.count orelse 4,
648
-
.backlog = config.thread_pool.backlog orelse 500,
649
-
.buffer_size = config.thread_pool.buffer_size orelse 32_768,
650
-
});
651
-
errdefer thread_pool.deinit();
652
-
653
687
const connection_buffer_size = config.buffers.size orelse DEFAULT_BUFFER_SIZE;
654
688
var small_buffer_pool: ?buffer.Pool = null;
655
689
if (config.buffers.pool) |pool_count| {
···
661
695
};
662
696
663
697
return .{
664
-
.loop = loop,
665
698
.allocator = allocator,
666
-
.thread_pool = thread_pool,
667
699
.max_conn = config.max_conn,
668
700
.conn_manager = conn_manager,
669
701
.small_buffer_pool = small_buffer_pool,
···
674
706
675
707
fn deinit(self: *Self) void {
676
708
self.conn_manager.deinit();
677
-
self.loop.deinit();
678
-
self.thread_pool.deinit();
679
709
if (self.small_buffer_pool) |*sbp| {
680
710
sbp.deinit();
681
711
}
682
712
}
683
713
684
-
fn setupMonitors(self: *Self, listener: ?posix.socket_t, signal: posix.fd_t) bool {
685
-
if (listener) |l| {
686
-
self.loop.monitorAccept(l) catch |err| {
687
-
log.err("failed to add monitor to listening socket: {}", .{err});
688
-
return false;
689
-
};
690
-
}
691
-
692
-
self.loop.monitorSignal(signal) catch |err| {
693
-
log.err("failed to add monitor to signal pipe: {}", .{err});
694
-
return false;
695
-
};
696
-
697
-
return true;
714
+
fn newConn(self: *Self, socket: posix.socket_t, address: net.Address, time: u32) !*HandlerConn(H) {
715
+
return self.conn_manager.create(socket, address, time);
698
716
}
699
717
700
-
fn newConn(self: *Self, socket: posix.socket_t, address: net.Address, time: u32) !void {
701
-
const hc = try self.conn_manager.create(socket, address, time);
702
-
self.loop.monitorRead(hc, false) catch |err| {
703
-
self.conn_manager.cleanup(hc);
704
-
return err;
705
-
};
706
-
}
707
-
708
-
fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) bool {
718
+
pub fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) bool {
709
719
return self._dataAvailable(hc, thread_buf) catch |err| {
710
720
log.err("({}) error processing client message: {}", .{hc.conn.address, err});
711
721
return false;
···
732
742
return ok;
733
743
}
734
744
735
-
fn postProcess(self: *Self, hc: *HandlerConn(H), success: bool) void {
736
-
var conn = &hc.conn;
737
-
var closed: bool = undefined;
738
-
if (success == false) {
739
-
conn.close();
740
-
closed = true;
741
-
} else {
742
-
closed = conn.isClosed();
743
-
}
744
-
745
-
if (closed) {
746
-
self.cleanupConn(hc);
747
-
} else {
748
-
self.loop.monitorRead(hc, true) catch |err| {
749
-
log.debug("({}) failed to add read event monitor: {}", .{conn.address, err});
750
-
conn.close();
751
-
self.cleanupConn(hc);
752
-
};
753
-
}
754
-
}
755
-
756
745
fn cleanupConn(self: *Self, hc: *HandlerConn(H)) void {
757
746
if (hc.reader) |*reader| {
758
747
if (self.small_buffer_pool) |*sbp| {
···
766
755
767
756
fn shutdown(self: *Self) void {
768
757
log.info("received shutdown signal", .{});
769
-
self.thread_pool.stop();
770
758
self.conn_manager.shutdown(self);
771
759
}
772
760
···
824
812
// But monitorRead where rearm == true is also used by our generic ServerLoop when
825
813
// taking over a connection (say from httpz). Hence, we need the EV.ADD flag too.
826
814
fn monitorRead(self: *KQueue, hc: anytype, comptime rearm: bool) !void {
827
-
if (rearm) {
828
-
const event = Kevent{
829
-
.ident = @intCast(hc.socket),
830
-
.filter = posix.system.EVFILT.READ,
831
-
.flags = posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH,
832
-
.fflags = 0,
833
-
.data = 0,
834
-
.udata = @intFromPtr(hc),
835
-
};
836
-
_ = try posix.kevent(self.q, &.{event}, &[_]Kevent{}, null);
837
-
} else {
838
-
try self.change(hc.socket, @intFromPtr(hc), posix.system.EVFILT.READ, posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH);
815
+
if (rearm == false) {
816
+
return self.change(hc.socket, @intFromPtr(hc), posix.system.EVFILT.READ, posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH);
839
817
}
818
+
const event = Kevent{
819
+
.ident = @intCast(hc.socket),
820
+
.filter = posix.system.EVFILT.READ,
821
+
.flags = posix.system.EV.ADD | posix.system.EV.ENABLE | posix.system.EV.DISPATCH,
822
+
.fflags = 0,
823
+
.data = 0,
824
+
.udata = @intFromPtr(hc),
825
+
};
826
+
_ = try posix.kevent(self.q, &.{event}, &[_]Kevent{}, null);
840
827
}
841
828
842
829
fn change(self: *KQueue, fd: posix.fd_t, data: usize, filter: i16, flags: u16) !void {
···
965
952
// with httpz (or any other http library I guess.). Serves a similar purpose
966
953
// as Server, but doesn't accept/listen.
967
954
pub fn Worker(comptime H: type) type {
968
-
969
955
return struct {
970
956
worker: W,
971
957
972
958
const Self = @This();
973
-
const W = if (blockingMode()) Blocking(H) else NonBlockingBase(H, Self.dataAvailable, false);
959
+
const W = if (blockingMode()) Blocking(H) else NonBlockingBase(H, false);
974
960
975
961
pub fn init(allocator: Allocator, state: *WorkerState) !Self {
976
962
return .{
···
993
979
pub fn shutdown(self: *Self) void {
994
980
self.worker.shutdown();
995
981
}
996
-
997
-
fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) void {
998
-
const success = self.worker.dataAvailable(hc, thread_buf);
999
-
self.worker.postProcess(hc, success);
1000
-
}
1001
-
1002
982
};
1003
983
}
1004
984
···
1796
1776
}
1797
1777
}
1798
1778
1799
-
// test "ServerLoop" {
1800
-
// const pipe = try posix.pipe2(.{.NONBLOCK = !blockingMode()});
1801
-
// const server = pipe[0];
1802
-
// errdefer posix.close(server);
1803
-
// const client = pipe[1];
1804
-
// defer posix.close(client);
1805
-
// defer std.debug.print("closing\n", .{});
1806
-
// std.debug.print("test: {d}\n", .{server});
1807
-
1808
-
// var sl = try ServerLoop(TestTakeOverHandler).init(t.allocator, .{});
1809
-
// try sl.run();
1810
-
// errdefer sl.deinit();
1811
-
1812
-
// try sl.takeover(server, try net.Address.parseIp("127.0.0.1", 0), "!!");
1813
-
1814
-
// std.debug.print("test: takeover\n", .{});
1815
-
// var stream = net.Stream{.handle = client};
1816
-
// try stream.writeAll(&proto.frame(.text, "abc"));
1817
-
1818
-
// var buf: [5]u8 = undefined;
1819
-
// _ = try stream.readAtLeast(&buf, 5);
1820
-
// std.debug.print("READ: {s}\n",.{buf});
1821
-
// // try t.expectSlice(u8, &.{129, 3, '1', '2', '3'}, buf[0..5]);
1822
-
1823
-
// sl.stop();
1824
-
// sl.deinit();
1825
-
// }
1826
-
1827
1779
fn testStream(handshake: bool) !net.Stream {
1828
1780
const timeout = std.mem.toBytes(std.posix.timeval{.sec = 0, .usec = 20_000});
1829
1781
const address = try std.net.Address.parseIp("127.0.0.1", 9292);
···
1881
1833
}
1882
1834
}
1883
1835
};
1884
-
1885
-
// const TestTakeOverHandler = struct {
1886
-
// conn: *Conn,
1887
-
// suffix: []const u8,
1888
-
1889
-
// pub fn init(conn: *Conn, suffix: []const u8) !TestTakeOverHandler {
1890
-
// return .{
1891
-
// .conn = conn,
1892
-
// .suffix = suffix,
1893
-
// };
1894
-
// }
1895
-
// pub fn clientMessage(self: *TestTakeOverHandler, allocator: Allocator, data: []const u8,) !void {
1896
-
// if (std.mem.eql(u8, data, "abc")) {
1897
-
// return self.conn.writeText(try std.fmt.allocPrint(allocator, "123{s}", .{self.suffix}));
1898
-
// }
1899
-
// }
1900
-
// };
1901
1836
1902
1837
test "List" {
1903
1838
var list = List(TestNode){};
+1
-1
support/autobahn/server/main.zig
+1
-1
support/autobahn/server/main.zig