karlseguin.tngl.sh
A websocket implementation for zig
make writes thread-safe
author
Karl Seguin
date
(Jul 14, 2024, 5:10 PM +0800)
commit
b60f46ac
b60f46acefd96d9194d9de739f6c8909b564d02e
parent
692648b0
692648b06867e4ebd1bbe141f31e2330e8e487bf
+58
-18
+11
-1
Dockerfile
+11
-1
Dockerfile
···
1
-
Dockerfile
1
+
from ubuntu:latest
2
+
run apt-get update && apt-get install -y wget xz-utils
3
+
4
+
run wget "https://ziglang.org/builds/zig-linux-aarch64-0.14.0-dev.244+0d79aa017.tar.xz"
5
+
run tar -xJvf zig-linux-aarch64-0.14.0-dev.244+0d79aa017.tar.xz && \
6
+
mv /zig-linux-aarch64-0.14.0-dev.244+0d79aa017/ /zig && \
7
+
chmod a+x /zig && \
8
+
rm -fr /zig-*
9
+
10
+
workdir /opt
11
+
entrypoint ["/zig/zig", "build", "test"]
+46
-16
src/server/server.zig
+46
-16
src/server/server.zig
···
713
713
714
714
// Called in a thread-pool thread/
715
715
// !! Access to self has to be synchronized !!
716
-
fn dataAvailable(self: *Self, hc: *HandlerConn(H), thread_buf: []u8) void {
716
+
fn dataAvailable(self: *Self, hc: *HandlerConn(H), _: []u8) void {
717
717
// Both doHandshake and processIncoming return !bool. This is to deal with
718
718
// Zig's lack of error payloads. The error is used for uncaught errors -
719
719
// usually rare things. The bool, when false, is used to indicate that
···
727
727
break :blk false;
728
728
};
729
729
} else {
730
-
ok = processIncoming(hc, thread_buf) catch |err| blk: {
730
+
ok = processIncoming(hc) catch |err| blk: {
731
731
log.warn("({}) uncaugh error handling data: {}", .{hc.conn.address, err});
732
732
break :blk false;
733
733
};
···
827
827
return true;
828
828
}
829
829
830
-
fn processIncoming(hc: *HandlerConn(H), thread_buf: []u8) !bool {
831
-
_ = thread_buf;
832
-
830
+
fn processIncoming(hc: *HandlerConn(H)) !bool {
833
831
var conn = &hc.conn;
834
832
var reader = &hc.reader.?;
835
833
reader.fill(conn.stream) catch |err| {
···
1089
1087
_closed: bool,
1090
1088
stream: net.Stream,
1091
1089
address: net.Address,
1090
+
lock: Thread.Mutex = .{},
1092
1091
1093
1092
pub fn isClosed(self: *Conn) bool {
1094
-
return @atomicLoad(bool, &self._closed, .monotonic);
1093
+
self.lock.lock();
1094
+
defer self.lock.unlock();
1095
+
return self._closed;
1095
1096
}
1096
1097
1097
1098
pub fn close(self: *Conn) void {
1098
-
if (self.isClosed() == false) {
1099
+
self.lock.lock();
1100
+
defer self.lock.unlock();
1101
+
if (self._closed == false) {
1099
1102
posix.close(self.stream.handle);
1100
-
@atomicStore(bool, &self._closed, true, .monotonic);
1103
+
self._closed = true;
1101
1104
}
1102
1105
}
1103
1106
···
1137
1140
1138
1141
// maximum possible prefix length. op_code + length_type + 8byte length
1139
1142
var buf: [10]u8 = undefined;
1143
+
var header: []const u8 = undefined;
1140
1144
buf[0] = @intFromEnum(op_code);
1141
1145
1142
1146
if (l <= 125) {
1143
1147
buf[1] = @intCast(l);
1144
-
try stream.writeAll(buf[0..2]);
1148
+
header = buf[0..2];
1145
1149
} else if (l < 65536) {
1146
1150
buf[1] = 126;
1147
1151
buf[2] = @intCast((l >> 8) & 0xFF);
1148
1152
buf[3] = @intCast(l & 0xFF);
1149
-
try stream.writeAll(buf[0..4]);
1153
+
header = buf[0..4];
1150
1154
} else {
1151
1155
buf[1] = 127;
1152
1156
buf[2] = @intCast((l >> 56) & 0xFF);
···
1157
1161
buf[7] = @intCast((l >> 16) & 0xFF);
1158
1162
buf[8] = @intCast((l >> 8) & 0xFF);
1159
1163
buf[9] = @intCast(l & 0xFF);
1160
-
try stream.writeAll(buf[0..]);
1164
+
header = buf[0..];
1161
1165
}
1162
-
if (l > 0) {
1163
-
try stream.writeAll(data);
1166
+
1167
+
if (l == 0) {
1168
+
// no body, just write the header
1169
+
self.lock.lock();
1170
+
defer self.lock.unlock();
1171
+
return stream.writeAll(header);
1172
+
}
1173
+
1174
+
var vec = [2]std.posix.iovec_const{
1175
+
.{ .len = header.len, .base = header.ptr },
1176
+
.{ .len = data.len, .base = data.ptr },
1177
+
};
1178
+
1179
+
var i: usize = 0;
1180
+
const socket = stream.handle;
1181
+
1182
+
self.lock.lock();
1183
+
defer self.lock.unlock();
1184
+
1185
+
while (true) {
1186
+
var n = try std.posix.writev(socket, vec[i..]);
1187
+
while (n >= vec[i].len) {
1188
+
n -= vec[i].len;
1189
+
i += 1;
1190
+
if (i >= vec.len) return;
1191
+
}
1192
+
vec[i].base += n;
1193
+
vec[i].len -= n;
1164
1194
}
1165
1195
}
1166
1196
1167
-
// TODO: SO MUCH
1168
-
// (WouldBlock handling, thread safety...)
1169
-
pub fn writeFramed(self: Conn, data: []const u8) !void {
1197
+
pub fn writeFramed(self: *Conn, data: []const u8) !void {
1198
+
self.lock.lock();
1199
+
defer self.lock.unlock();
1170
1200
try self.stream.writeAll(data);
1171
1201
}
1172
1202
};
+1
-1
support/autobahn/server/run.sh
+1
-1
support/autobahn/server/run.sh