zzstoatzz.io
This repository has no description
1//! Zio: the hybrid std.Io backend.
2//!
3//! wraps an inner Io.Threaded and delegates the entire vtable to it, then
4//! overrides the ops that benefit from the fiber scheduler:
5//!
6//! concurrent/await/cancel -> fibers on the sched loop thread
7//! sleep -> sched timer heap (when called from a fiber)
8//! netConnectIp/Read/Write/Close -> park on poller readiness
9//! futexWait/futexWake -> fiber-aware futex table; this makes
10//! Io.Mutex, Io.Condition and Io.Queue work
11//! correctly for fiber callers with zero
12//! changes to their code
13//! netLookup -> spilled to an inner-Threaded thread so DNS
14//! never blocks the loop
15//!
16//! every op detects its caller domain: on-a-fiber -> scheduler path;
17//! anywhere else (main thread, pool threads) -> delegate to Io.Threaded.
18//! this is what makes the backend a drop-in `const Backend = zio.Zio;`.
19const std = @import("std");
20const builtin = @import("builtin");
21const c = std.c;
22const Io = std.Io;
23const sched_mod = @import("fiber/sched.zig");
24const engine = @import("engine.zig");
25const Sched = sched_mod.Sched;
26const Fiber = sched_mod.Fiber;
27
28const max_result_size = 512;
29const max_context_size = 1024;
30
31pub const Zio = struct {
32 allocator: std.mem.Allocator,
33 threaded: Io.Threaded,
34 vtable: Io.VTable,
35 sched: Sched,
36 loop_thread: ?std.Thread,
37
38 /// identifies the loop thread; ops compare against Sched.tls_active.
39 pub fn init(zio: *Zio, allocator: std.mem.Allocator, options: Io.Threaded.InitOptions) !void {
40 zio.allocator = allocator;
41 zio.threaded = .init(allocator, options);
42 zio.sched = try Sched.init(allocator);
43 try zio.sched.attachWake();
44 zio.loop_thread = null;
45
46 const inner = zio.threaded.io();
47 zio.vtable = inner.vtable.*;
48 zio.vtable.concurrent = vtConcurrent;
49 zio.vtable.await = vtAwait;
50 zio.vtable.cancel = vtCancel;
51 zio.vtable.checkCancel = vtCheckCancel;
52 zio.vtable.sleep = vtSleep;
53 zio.vtable.netConnectIp = vtNetConnectIp;
54 zio.vtable.netAccept = vtNetAccept;
55 zio.vtable.netRead = vtNetRead;
56 zio.vtable.netWrite = vtNetWrite;
57 zio.vtable.futexWait = vtFutexWait;
58 zio.vtable.futexWaitUncancelable = vtFutexWaitUncancelable;
59 zio.vtable.futexWake = vtFutexWake;
60 zio.vtable.netLookup = vtNetLookup;
61 }
62
63 pub fn start(zio: *Zio) !void {
64 zio.loop_thread = try std.Thread.spawn(.{}, loopMain, .{zio});
65 }
66
67 pub fn deinit(zio: *Zio) void {
68 if (zio.loop_thread) |t| {
69 zio.sched.requestStop();
70 t.join();
71 zio.loop_thread = null;
72 }
73 zio.sched.deinit();
74 zio.threaded.deinit();
75 }
76
77 pub fn io(zio: *Zio) Io {
78 return .{ .userdata = &zio.threaded, .vtable = &zio.vtable };
79 }
80
81 fn loopMain(zio: *Zio) void {
82 zio.sched.runForever();
83 }
84
85 fn fromUserdata(userdata: ?*anyopaque) *Zio {
86 const t: *Io.Threaded = @ptrCast(@alignCast(userdata.?));
87 return @fieldParentPtr("threaded", t);
88 }
89
90 /// non-null when the calling code is a fiber owned by this Zio's sched.
91 fn currentFiber(zio: *Zio) ?*Fiber {
92 if (Sched.tls_active != &zio.sched) return null;
93 return zio.sched.current;
94 }
95};
96
97// --- task: unit of concurrency behind *AnyFuture ---------------------------
98
99const Task = struct {
100 zio: *Zio,
101 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
102 context: [max_context_size]u8 align(16),
103 result: [max_result_size]u8 align(16),
104 /// 0 = running, 1 = done. futex word for foreign-thread awaiters.
105 state: std.atomic.Value(u32),
106 cancel_requested: std.atomic.Value(bool),
107 /// fiber awaiting this task, parked on the sched (sched thread only).
108 awaiter: ?*Fiber,
109 /// the fiber executing this task (sched thread only).
110 fiber: ?*Fiber,
111 /// two owners: the completing fiber and the awaiting caller.
112 refs: std.atomic.Value(u32),
113
114 fn unref(task: *Task) void {
115 if (task.refs.fetchSub(1, .acq_rel) == 1) {
116 task.zio.allocator.destroy(task);
117 }
118 }
119};
120
121fn taskFiberEntry(s: *Sched, arg: ?*anyopaque) void {
122 const task: *Task = @ptrCast(@alignCast(arg.?));
123 task.fiber = s.current;
124 s.current.?.task = task;
125 task.start(&task.context, &task.result);
126 task.fiber = null;
127 task.state.store(1, .release);
128 // wake a fiber awaiter (same thread) and any foreign-thread awaiter.
129 if (task.awaiter) |af| {
130 task.awaiter = null;
131 if (af.state == .parked) s.pushReady(af);
132 }
133 zioFutexWake(task.zio, &task.state.raw, 1);
134 task.unref();
135}
136
137fn vtConcurrent(
138 userdata: ?*anyopaque,
139 result_len: usize,
140 result_alignment: std.mem.Alignment,
141 context: []const u8,
142 context_alignment: std.mem.Alignment,
143 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
144) Io.ConcurrentError!*Io.AnyFuture {
145 const zio = Zio.fromUserdata(userdata);
146 std.debug.assert(result_len <= max_result_size);
147 std.debug.assert(context.len <= max_context_size);
148 std.debug.assert(result_alignment.compare(.lte, .@"16"));
149 std.debug.assert(context_alignment.compare(.lte, .@"16"));
150
151 const task = zio.allocator.create(Task) catch return error.ConcurrencyUnavailable;
152 task.* = .{
153 .zio = zio,
154 .start = start,
155 .context = undefined,
156 .result = undefined,
157 .state = .init(0),
158 .cancel_requested = .init(false),
159 .awaiter = null,
160 .fiber = null,
161 .refs = .init(2),
162 };
163 @memcpy(task.context[0..context.len], context);
164
165 if (zio.currentFiber() != null) {
166 // spawning from a fiber: same thread, spawn directly.
167 zio.sched.spawn(taskFiberEntry, task) catch {
168 zio.allocator.destroy(task);
169 return error.ConcurrencyUnavailable;
170 };
171 } else {
172 zio.sched.injectSpawn(taskFiberEntry, task) catch {
173 zio.allocator.destroy(task);
174 return error.ConcurrencyUnavailable;
175 };
176 }
177 return @ptrCast(task);
178}
179
180fn awaitImpl(zio: *Zio, task: *Task, result: []u8) void {
181 if (zio.currentFiber()) |f| {
182 while (task.state.load(.acquire) == 0) {
183 task.awaiter = f;
184 zio.sched.parkCurrent();
185 }
186 } else {
187 // foreign thread: wait on the task state word through the inner
188 // Threaded futex (real futex/ulock). completion wakes via zioFutexWake.
189 const inner = zio.threaded.io();
190 while (task.state.load(.acquire) == 0) {
191 inner.vtable.futexWaitUncancelable(inner.userdata, &task.state.raw, 0);
192 }
193 }
194 @memcpy(result, task.result[0..result.len]);
195 task.unref();
196}
197
198fn vtAwait(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8, result_alignment: std.mem.Alignment) void {
199 _ = result_alignment;
200 const zio = Zio.fromUserdata(userdata);
201 const task: *Task = @ptrCast(@alignCast(any_future));
202 awaitImpl(zio, task, result);
203}
204
205fn vtCancel(userdata: ?*anyopaque, any_future: *Io.AnyFuture, result: []u8, result_alignment: std.mem.Alignment) void {
206 _ = result_alignment;
207 const zio = Zio.fromUserdata(userdata);
208 const task: *Task = @ptrCast(@alignCast(any_future));
209 task.cancel_requested.store(true, .release);
210 // wake the task's fiber if it is parked so it can observe cancellation.
211 zio.sched.injectInterrupt(task) catch {};
212 awaitImpl(zio, task, result);
213}
214
215fn vtCheckCancel(userdata: ?*anyopaque) Io.Cancelable!void {
216 const zio = Zio.fromUserdata(userdata);
217 if (zio.currentFiber()) |f| {
218 if (fiberCanceled(f)) return error.Canceled;
219 return;
220 }
221 const inner = zio.threaded.io();
222 return inner.vtable.checkCancel(inner.userdata);
223}
224
225fn fiberCanceled(f: *Fiber) bool {
226 const task_ptr = f.task orelse return false;
227 const task: *Task = @ptrCast(@alignCast(task_ptr));
228 return task.cancel_requested.load(.acquire);
229}
230
231// --- sleep ------------------------------------------------------------------
232
233fn timeoutToDeadlineMs(zio: *Zio, timeout: Io.Timeout) ?u64 {
234 // all supported clocks are mapped onto the sched's monotonic ms clock.
235 const inner = zio.threaded.io();
236 const now_ns = Io.Timestamp.now(inner, .awake).nanoseconds;
237 return switch (timeout) {
238 .none => null,
239 .duration => |d| sched_mod.nowMs() + @as(u64, @intCast(@divTrunc(d.raw.nanoseconds, std.time.ns_per_ms))),
240 .deadline => |d| blk: {
241 const delta_ns = d.raw.nanoseconds - now_ns;
242 if (delta_ns <= 0) break :blk sched_mod.nowMs();
243 break :blk sched_mod.nowMs() + @as(u64, @intCast(@divTrunc(delta_ns, std.time.ns_per_ms)));
244 },
245 };
246}
247
248fn vtSleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.Cancelable!void {
249 const zio = Zio.fromUserdata(userdata);
250 if (zio.currentFiber()) |f| {
251 if (fiberCanceled(f)) return error.Canceled;
252 const deadline = timeoutToDeadlineMs(zio, timeout) orelse {
253 // sleep forever: park until canceled.
254 while (true) {
255 zio.sched.parkCurrent();
256 if (fiberCanceled(f)) return error.Canceled;
257 }
258 };
259 zio.sched.sleepUntil(deadline);
260 if (fiberCanceled(f)) return error.Canceled;
261 return;
262 }
263 const inner = zio.threaded.io();
264 return inner.vtable.sleep(inner.userdata, timeout);
265}
266
267// --- networking --------------------------------------------------------------
268
269fn vtNetConnectIp(
270 userdata: ?*anyopaque,
271 address: *const Io.net.IpAddress,
272 options: Io.net.IpAddress.ConnectOptions,
273) Io.net.IpAddress.ConnectError!Io.net.Socket {
274 const zio = Zio.fromUserdata(userdata);
275 const f = zio.currentFiber() orelse {
276 const inner = zio.threaded.io();
277 return inner.vtable.netConnectIp(inner.userdata, address, options);
278 };
279 if (options.mode != .stream) {
280 const inner = zio.threaded.io();
281 return inner.vtable.netConnectIp(inner.userdata, address, options);
282 }
283
284 const ip4 = switch (address.*) {
285 .ip4 => |a| a,
286 else => {
287 const inner = zio.threaded.io();
288 return inner.vtable.netConnectIp(inner.userdata, address, options);
289 },
290 };
291 const fd = engine.tcpConnectIp4Start(ip4.bytes, ip4.port) catch
292 return error.ConnectionRefused;
293 errdefer _ = c.close(fd);
294 while (true) {
295 if (fiberCanceled(f)) return error.Canceled;
296 zio.sched.waitWritable(fd);
297 if (fiberCanceled(f)) return error.Canceled;
298 engine.connectResult(fd) catch return error.ConnectionRefused;
299 var peer: c.sockaddr.in = undefined;
300 var len: c.socklen_t = @sizeOf(c.sockaddr.in);
301 if (c.getpeername(fd, @ptrCast(&peer), &len) == 0) break;
302 if (c.errno(@as(isize, -1)) != .NOTCONN) return error.ConnectionRefused;
303 }
304 var bound: c.sockaddr.in = undefined;
305 var blen: c.socklen_t = @sizeOf(c.sockaddr.in);
306 _ = c.getsockname(fd, @ptrCast(&bound), &blen);
307 return .{ .handle = fd, .address = .{ .ip4 = .{
308 .bytes = @bitCast(bound.addr),
309 .port = std.mem.bigToNative(u16, bound.port),
310 } } };
311}
312
313fn vtNetAccept(
314 userdata: ?*anyopaque,
315 server: Io.net.Socket.Handle,
316 options: Io.net.Server.AcceptOptions,
317) Io.net.Server.AcceptError!Io.net.Socket {
318 const zio = Zio.fromUserdata(userdata);
319 const f = zio.currentFiber() orelse {
320 const inner = zio.threaded.io();
321 return inner.vtable.netAccept(inner.userdata, server, options);
322 };
323 // the listener may have been created by the (blocking) Threaded path;
324 // flip it non-blocking so the loop never blocks in accept. idempotent.
325 engine.setNonblock(server) catch return error.SocketNotListening;
326 while (true) {
327 if (fiberCanceled(f)) return error.Canceled;
328 const maybe = engine.acceptNonblock(server) catch return error.SocketNotListening;
329 if (maybe) |fd| {
330 var peer: c.sockaddr.in = undefined;
331 var plen: c.socklen_t = @sizeOf(c.sockaddr.in);
332 _ = c.getpeername(fd, @ptrCast(&peer), &plen);
333 return .{ .handle = fd, .address = .{ .ip4 = .{
334 .bytes = @bitCast(peer.addr),
335 .port = std.mem.bigToNative(u16, peer.port),
336 } } };
337 }
338 zio.sched.waitReadable(server);
339 }
340}
341
342fn vtNetRead(userdata: ?*anyopaque, src: Io.net.Socket.Handle, data: [][]u8) Io.net.Stream.Reader.Error!usize {
343 const zio = Zio.fromUserdata(userdata);
344 const f = zio.currentFiber() orelse {
345 const inner = zio.threaded.io();
346 return inner.vtable.netRead(inner.userdata, src, data);
347 };
348 const buf = for (data) |d| {
349 if (d.len > 0) break d;
350 } else return 0;
351 while (true) {
352 if (fiberCanceled(f)) return error.Canceled;
353 const n = c.read(src, buf.ptr, buf.len);
354 if (n >= 0) return @intCast(n);
355 if (c.errno(n) != engine.wouldBlock) return error.Unexpected;
356 zio.sched.waitReadable(src);
357 }
358}
359
360fn vtNetWrite(
361 userdata: ?*anyopaque,
362 dest: Io.net.Socket.Handle,
363 header: []const u8,
364 data: []const []const u8,
365 splat: usize,
366) Io.net.Stream.Writer.Error!usize {
367 const zio = Zio.fromUserdata(userdata);
368 const f = zio.currentFiber() orelse {
369 const inner = zio.threaded.io();
370 return inner.vtable.netWrite(inner.userdata, dest, header, data, splat);
371 };
372 // short writes are allowed by the contract: send the first non-empty
373 // chunk, parking on EAGAIN until at least one byte lands.
374 const chunk: []const u8 = if (header.len > 0) header else blk: {
375 if (data.len == 0) return 0;
376 for (data[0 .. data.len - 1]) |d| {
377 if (d.len > 0) break :blk d;
378 }
379 const last = data[data.len - 1];
380 if (last.len == 0 or splat == 0) return 0;
381 break :blk last;
382 };
383 while (true) {
384 if (fiberCanceled(f)) return error.Canceled;
385 const n = c.write(dest, chunk.ptr, chunk.len);
386 if (n >= 0) return @intCast(n);
387 if (c.errno(n) != engine.wouldBlock) return error.Unexpected;
388 zio.sched.waitWritable(dest);
389 }
390}
391
392// --- fiber-aware futex --------------------------------------------------------
393// Io.Mutex, Io.Condition and Io.Queue all park through these two ops. fibers
394// park on the sched futex table; foreign threads keep the inner Threaded
395// futex. wakes always hit both domains.
396
397fn vtFutexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Io.Timeout) Io.Cancelable!void {
398 const zio = Zio.fromUserdata(userdata);
399 if (zio.currentFiber()) |f| {
400 if (fiberCanceled(f)) return error.Canceled;
401 const deadline = timeoutToDeadlineMs(zio, timeout);
402 zio.sched.futexWaitFiber(ptr, expected, deadline);
403 if (fiberCanceled(f)) return error.Canceled;
404 return;
405 }
406 const inner = zio.threaded.io();
407 return inner.vtable.futexWait(inner.userdata, ptr, expected, timeout);
408}
409
410fn vtFutexWaitUncancelable(userdata: ?*anyopaque, ptr: *const u32, expected: u32) void {
411 const zio = Zio.fromUserdata(userdata);
412 if (zio.currentFiber() != null) {
413 zio.sched.futexWaitFiber(ptr, expected, null);
414 return;
415 }
416 const inner = zio.threaded.io();
417 inner.vtable.futexWaitUncancelable(inner.userdata, ptr, expected);
418}
419
420fn zioFutexWake(zio: *Zio, ptr: *const u32, max_waiters: u32) void {
421 // fiber domain (thread-safe: goes through the sched's external queue
422 // when called off-loop) ...
423 zio.sched.futexWakeFibers(ptr, max_waiters);
424 // ... and thread domain.
425 const inner = zio.threaded.io();
426 inner.vtable.futexWake(inner.userdata, ptr, max_waiters);
427}
428
429fn vtFutexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
430 zioFutexWake(Zio.fromUserdata(userdata), ptr, max_waiters);
431}
432
433// --- dns spill -----------------------------------------------------------------
434
435fn vtNetLookup(
436 userdata: ?*anyopaque,
437 host_name: Io.net.HostName,
438 queue: *Io.Queue(Io.net.HostName.LookupResult),
439 options: Io.net.HostName.LookupOptions,
440) Io.net.HostName.LookupError!void {
441 const zio = Zio.fromUserdata(userdata);
442 const inner = zio.threaded.io();
443 if (zio.currentFiber() == null) {
444 return inner.vtable.netLookup(inner.userdata, host_name, queue, options);
445 }
446 // fiber caller: getaddrinfo must not run on the loop thread. run the
447 // lookup as an inner-Threaded task (real OS thread); the fiber parks in
448 // queue.get(), whose futexWait is fiber-aware (above), so the loop keeps
449 // running while DNS resolves.
450 const Spill = struct {
451 fn lookup(inner_io: Io, hn: Io.net.HostName, q: *Io.Queue(Io.net.HostName.LookupResult), opts: Io.net.HostName.LookupOptions) void {
452 inner_io.vtable.netLookup(inner_io.userdata, hn, q, opts) catch {
453 // error detail is lost across the spill; consumers observe a
454 // closed queue with no results (same shape as zero answers).
455 q.close(inner_io);
456 };
457 }
458 };
459 var future = inner.concurrent(Spill.lookup, .{ inner, host_name, queue, options }) catch
460 return inner.vtable.netLookup(inner.userdata, host_name, queue, options);
461 // detach: the queue is the completion channel; the task outlives this call.
462 _ = &future;
463 return;
464}
465
466test "Zio: concurrent + await through the vtable (fiber tasks)" {
467 var zio: Zio = undefined;
468 try zio.init(std.heap.c_allocator, .{});
469 defer zio.deinit();
470 try zio.start();
471 const io = zio.io();
472
473 const Worker = struct {
474 fn double(x: u64) u64 {
475 return x * 2;
476 }
477 };
478 var fut = try io.concurrent(Worker.double, .{21});
479 const answer = fut.await(io);
480 try std.testing.expectEqual(@as(u64, 42), answer);
481}
482
483test "Zio: io.sleep on a fiber task goes through the sched timers" {
484 var zio: Zio = undefined;
485 try zio.init(std.heap.c_allocator, .{});
486 defer zio.deinit();
487 try zio.start();
488 const io = zio.io();
489
490 const Worker = struct {
491 fn nap(io_inner: Io) u64 {
492 io_inner.sleep(Io.Duration.fromMilliseconds(20), .awake) catch return 0;
493 return 7;
494 }
495 };
496 const t0 = sched_mod.nowMs();
497 var fut = try io.concurrent(Worker.nap, .{io});
498 const r = fut.await(io);
499 const elapsed = sched_mod.nowMs() - t0;
500 try std.testing.expectEqual(@as(u64, 7), r);
501 try std.testing.expect(elapsed >= 15);
502}
503
504test "Zio: blocking-style net through Io.net on fibers (tcp echo)" {
505 var zio: Zio = undefined;
506 try zio.init(std.heap.c_allocator, .{});
507 defer zio.deinit();
508 try zio.start();
509 const io = zio.io();
510
511 // listener stays on the inner Threaded (accept isn't overridden yet) —
512 // exactly the hybrid contract: unoverridden ops fall back to threads.
513 const bind_addr = try Io.net.IpAddress.parse("127.0.0.1", 0);
514 var server = try Io.net.IpAddress.listen(&bind_addr, io, .{});
515 defer server.deinit(io);
516 const addr = server.socket.address;
517
518 const Server = struct {
519 fn serve(io_inner: Io, srv: *Io.net.Server) u64 {
520 const stream = srv.accept(io_inner) catch return 0;
521 defer stream.close(io_inner);
522 var rbuf: [128]u8 = undefined;
523 var wbuf: [128]u8 = undefined;
524 var reader = stream.reader(io_inner, &rbuf);
525 var writer = stream.writer(io_inner, &wbuf);
526 var total: u64 = 0;
527 // fixed-size protocol: read exactly one 11-byte ping per round.
528 // (readSliceShort would park waiting to fill its buffer — short
529 // reads only happen at stream end, not at message boundaries.)
530 var chunk: [11]u8 = undefined;
531 while (true) {
532 reader.interface.readSliceAll(&chunk) catch return total;
533 writer.interface.writeAll(&chunk) catch return total;
534 writer.interface.flush() catch return total;
535 total += chunk.len;
536 }
537 }
538 };
539
540 const Client = struct {
541 fn ping(io_inner: Io, a: *const Io.net.IpAddress) u64 {
542 const stream = Io.net.IpAddress.connect(a, io_inner, .{ .mode = .stream }) catch return 0;
543 defer stream.close(io_inner);
544 var rbuf: [128]u8 = undefined;
545 var wbuf: [128]u8 = undefined;
546 var reader = stream.reader(io_inner, &rbuf);
547 var writer = stream.writer(io_inner, &wbuf);
548 var echoed: u64 = 0;
549 var round: usize = 0;
550 while (round < 50) : (round += 1) {
551 writer.interface.writeAll("fiber echo!") catch return echoed;
552 writer.interface.flush() catch return echoed;
553 var got: [11]u8 = undefined;
554 reader.interface.readSliceAll(&got) catch return echoed;
555 echoed += got.len;
556 }
557 return echoed;
558 }
559 };
560
561 var server_fut = try io.concurrent(Server.serve, .{ io, &server });
562 var client_fut = try io.concurrent(Client.ping, .{ io, &addr });
563 const echoed = client_fut.await(io);
564 const served = server_fut.await(io);
565 try std.testing.expectEqual(@as(u64, 550), echoed);
566 try std.testing.expectEqual(@as(u64, 550), served);
567}
568
569test "Zio: Io.Mutex contended across fiber and pool-thread domains" {
570 var zio: Zio = undefined;
571 try zio.init(std.heap.c_allocator, .{});
572 defer zio.deinit();
573 try zio.start();
574 const io = zio.io();
575
576 const Shared = struct {
577 var mutex: Io.Mutex = .init;
578 var counter: u64 = 0;
579
580 fn fiberSide(io_inner: Io) u64 {
581 var i: u32 = 0;
582 while (i < 1000) : (i += 1) {
583 mutex.lock(io_inner) catch return 0;
584 counter += 1;
585 mutex.unlock(io_inner);
586 }
587 return 1;
588 }
589 fn threadSide(inner_io: Io) u64 {
590 var i: u32 = 0;
591 while (i < 1000) : (i += 1) {
592 mutex.lock(inner_io) catch return 0;
593 counter += 1;
594 mutex.unlock(inner_io);
595 }
596 return 1;
597 }
598 };
599 Shared.counter = 0;
600
601 // fiber domain through Zio; thread domain through the inner Threaded —
602 // the exact zlay shape (subscriber fibers vs pool_io worker threads).
603 const inner = zio.threaded.io();
604 var f1 = try io.concurrent(Shared.fiberSide, .{io});
605 var f2 = try inner.concurrent(Shared.threadSide, .{inner});
606 _ = f1.await(io);
607 _ = f2.await(inner);
608 try std.testing.expectEqual(@as(u64, 2000), Shared.counter);
609}
610
611test "Zio: cancel reaches a fiber parked on io promptly" {
612 var zio: Zio = undefined;
613 try zio.init(std.heap.c_allocator, .{});
614 defer zio.deinit();
615 try zio.start();
616 const io = zio.io();
617
618 // a connected socket pair where no data ever arrives: the reader fiber
619 // parks in netRead indefinitely; cancel must wake it with error.Canceled.
620 const listener = try engine.tcpListenIp4(.{ 127, 0, 0, 1 }, 0, 4);
621 defer _ = c.close(listener.fd);
622
623 const Reader = struct {
624 fn readForever(io_inner: Io, port: u16) u64 {
625 const addr = Io.net.IpAddress{ .ip4 = .{ .bytes = .{ 127, 0, 0, 1 }, .port = port } };
626 const stream = Io.net.IpAddress.connect(&addr, io_inner, .{ .mode = .stream }) catch return 99;
627 defer stream.close(io_inner);
628 var rbuf: [64]u8 = undefined;
629 var reader = stream.reader(io_inner, &rbuf);
630 var byte: [1]u8 = undefined;
631 reader.interface.readSliceAll(&byte) catch return 1; // canceled -> ReadFailed path
632 return 0;
633 }
634 };
635
636 var fut = try io.concurrent(Reader.readForever, .{ io, listener.port });
637 // let it connect and park (accept side never sends anything)
638 const accepted = try engine.acceptNonblock(listener.fd) orelse blk: {
639 var spins: u32 = 0;
640 while (spins < 500) : (spins += 1) {
641 const ts: c.timespec = .{ .sec = 0, .nsec = 5 * std.time.ns_per_ms };
642 _ = c.nanosleep(&ts, null);
643 if (try engine.acceptNonblock(listener.fd)) |fd2| break :blk fd2;
644 }
645 return error.NeverConnected;
646 };
647 defer _ = c.close(accepted);
648 {
649 const ts: c.timespec = .{ .sec = 0, .nsec = 50 * std.time.ns_per_ms };
650 _ = c.nanosleep(&ts, null); // give the fiber time to park in netRead
651 }
652
653 const t0 = sched_mod.nowMs();
654 const r = fut.cancel(io); // must return promptly, not hang
655 const elapsed = sched_mod.nowMs() - t0;
656 try std.testing.expectEqual(@as(u64, 1), r);
657 try std.testing.expect(elapsed < 500);
658}
659
660test {
661 _ = Zio;
662}