1const builtin = @import("builtin");
2const native_endian = builtin.cpu.arch.endian();
3
4const std = @import("std");
5const Io = std.Io;
6const DefaultPrng = std.Random.DefaultPrng;
7const mem = std.mem;
8const fs = std.fs;
9const File = std.Io.File;
10const assert = std.debug.assert;
11
12const testing = std.testing;
13const expect = std.testing.expect;
14const expectEqual = std.testing.expectEqual;
15const expectError = std.testing.expectError;
16const expectEqualStrings = std.testing.expectEqualStrings;
17const tmpDir = std.testing.tmpDir;
18
19test "write a file, read it, then delete it" {
20 const io = testing.io;
21
22 var tmp = tmpDir(.{});
23 defer tmp.cleanup();
24
25 var data: [1024]u8 = undefined;
26 var prng = DefaultPrng.init(testing.random_seed);
27 const random = prng.random();
28 random.bytes(data[0..]);
29 const tmp_file_name = "temp_test_file.txt";
30 {
31 var file = try tmp.dir.createFile(io, tmp_file_name, .{});
32 defer file.close(io);
33
34 var file_writer = file.writer(io, &.{});
35 const st = &file_writer.interface;
36 try st.print("begin", .{});
37 try st.writeAll(&data);
38 try st.print("end", .{});
39 try st.flush();
40 }
41
42 {
43 // Make sure the exclusive flag is honored.
44 try expectError(File.OpenError.PathAlreadyExists, tmp.dir.createFile(io, tmp_file_name, .{ .exclusive = true }));
45 }
46
47 {
48 var file = try tmp.dir.openFile(io, tmp_file_name, .{});
49 defer file.close(io);
50
51 const file_size = try file.length(io);
52 const expected_file_size: u64 = "begin".len + data.len + "end".len;
53 try expectEqual(expected_file_size, file_size);
54
55 var file_buffer: [1024]u8 = undefined;
56 var file_reader = file.reader(io, &file_buffer);
57 const contents = try file_reader.interface.allocRemaining(testing.allocator, .limited(2 * 1024));
58 defer testing.allocator.free(contents);
59
60 try expect(mem.eql(u8, contents[0.."begin".len], "begin"));
61 try expect(mem.eql(u8, contents["begin".len .. contents.len - "end".len], &data));
62 try expect(mem.eql(u8, contents[contents.len - "end".len ..], "end"));
63 }
64 try tmp.dir.deleteFile(io, tmp_file_name);
65}
66
67test "File.Writer.seekTo" {
68 var tmp = tmpDir(.{});
69 defer tmp.cleanup();
70
71 const io = testing.io;
72
73 var data: [8192]u8 = undefined;
74 @memset(&data, 0x55);
75
76 const tmp_file_name = "temp_test_file.txt";
77 var file = try tmp.dir.createFile(io, tmp_file_name, .{ .read = true });
78 defer file.close(io);
79
80 var fw = file.writerStreaming(io, &.{});
81
82 try fw.interface.writeAll(&data);
83 try expect(fw.logicalPos() == try file.length(io));
84 try fw.seekTo(1234);
85 try expect(fw.logicalPos() == 1234);
86}
87
88test "file discard" {
89 var tmp = tmpDir(.{});
90 defer tmp.cleanup();
91
92 const io = testing.io;
93
94 const tmp_file_name = "temp_test_file.txt";
95 var file = try tmp.dir.createFile(io, tmp_file_name, .{ .read = true });
96 defer file.close(io);
97
98 var fw = file.writerStreaming(io, &.{});
99
100 try fw.interface.writeAll("test");
101
102 var fr = file.reader(io, &.{});
103 const r = &fr.interface;
104
105 try std.testing.expectEqual(error.EndOfStream, r.discardAll(1024));
106 try fr.seekTo(0);
107 try std.testing.expectEqual(4, fr.interface.discardRemaining());
108}
109
110test "File.setLength" {
111 const io = testing.io;
112
113 var tmp = tmpDir(.{});
114 defer tmp.cleanup();
115
116 const tmp_file_name = "temp_test_file.txt";
117 var file = try tmp.dir.createFile(io, tmp_file_name, .{ .read = true });
118 defer file.close(io);
119
120 var fw = file.writerStreaming(io, &.{});
121
122 // Verify that the file size changes and the file offset is not moved
123 try expect((try file.length(io)) == 0);
124 try expect(fw.logicalPos() == 0);
125 try file.setLength(io, 8192);
126 try expect((try file.length(io)) == 8192);
127 try expect(fw.logicalPos() == 0);
128 try fw.seekTo(100);
129 try file.setLength(io, 4096);
130 try expect((try file.length(io)) == 4096);
131 try expect(fw.logicalPos() == 100);
132 try file.setLength(io, 0);
133 try expect((try file.length(io)) == 0);
134 try expect(fw.logicalPos() == 100);
135}
136
137test "legacy setLength" {
138 // https://github.com/ziglang/zig/issues/20747 (open fd does not have write permission)
139 if (builtin.os.tag == .wasi and builtin.link_libc) return error.SkipZigTest;
140
141 const io = testing.io;
142
143 var tmp = tmpDir(.{});
144 defer tmp.cleanup();
145
146 const file_name = "afile.txt";
147 try tmp.dir.writeFile(io, .{ .sub_path = file_name, .data = "ninebytes" });
148 const f = try tmp.dir.openFile(io, file_name, .{ .mode = .read_write });
149 defer f.close(io);
150
151 const initial_size = try f.length(io);
152 var buffer: [32]u8 = undefined;
153 var reader = f.reader(io, &.{});
154
155 {
156 try f.setLength(io, initial_size);
157 try expectEqual(initial_size, try f.length(io));
158 try reader.seekTo(0);
159 try expectEqual(initial_size, try reader.interface.readSliceShort(&buffer));
160 try expectEqualStrings("ninebytes", buffer[0..@intCast(initial_size)]);
161 }
162
163 {
164 const larger = initial_size + 4;
165 try f.setLength(io, larger);
166 try expectEqual(larger, try f.length(io));
167 try reader.seekTo(0);
168 try expectEqual(larger, try reader.interface.readSliceShort(&buffer));
169 try expectEqualStrings("ninebytes\x00\x00\x00\x00", buffer[0..@intCast(larger)]);
170 }
171
172 {
173 const smaller = initial_size - 5;
174 try f.setLength(io, smaller);
175 try expectEqual(smaller, try f.length(io));
176 try reader.seekTo(0);
177 try expectEqual(smaller, try reader.interface.readSliceShort(&buffer));
178 try expectEqualStrings("nine", buffer[0..@intCast(smaller)]);
179 }
180
181 try f.setLength(io, 0);
182 try expectEqual(0, try f.length(io));
183 try reader.seekTo(0);
184 try expectEqual(0, try reader.interface.readSliceShort(&buffer));
185}
186
187test "setTimestamps" {
188 const io = testing.io;
189
190 var tmp = tmpDir(.{});
191 defer tmp.cleanup();
192
193 const tmp_file_name = "just_a_temporary_file.txt";
194 var file = try tmp.dir.createFile(io, tmp_file_name, .{ .read = true });
195 defer file.close(io);
196
197 const stat_old = try file.stat(io);
198
199 // Set atime and mtime to 5s before
200 try file.setTimestamps(io, .{
201 .access_timestamp = if (stat_old.atime) |atime| .{ .new = atime.subDuration(.fromSeconds(5)) } else .unchanged,
202 .modify_timestamp = .{ .new = stat_old.mtime.subDuration(.fromSeconds(5)) },
203 });
204 const stat_new = try file.stat(io);
205 // NetBSD with noatime will just not update the timestamp, and noatime is default in at least NetBSD 11+.
206 if (builtin.os.tag != .netbsd) {
207 if (stat_old.atime) |old_atime| try expect(stat_new.atime.?.nanoseconds < old_atime.nanoseconds);
208 }
209 try expect(stat_new.mtime.nanoseconds < stat_old.mtime.nanoseconds);
210}
211
212test "Group" {
213 const io = testing.io;
214
215 var group: Io.Group = .init;
216 var results: [2]usize = undefined;
217
218 group.async(io, count, .{ 1, 10, &results[0] });
219 group.async(io, count, .{ 20, 30, &results[1] });
220
221 try group.await(io);
222
223 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
224}
225
226fn count(a: usize, b: usize, result: *usize) void {
227 var sum: usize = 0;
228 for (a..b) |i| {
229 sum += i;
230 }
231 result.* = sum;
232}
233
234test "Group.cancel" {
235 const global = struct {
236 fn sleep(io: Io, result: *usize) Io.Cancelable!void {
237 defer result.* = 1;
238 io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {
239 error.Canceled => |e| return e,
240 };
241 }
242
243 fn sleepRecancel(io: Io, result: *usize) void {
244 io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {
245 error.Canceled => io.recancel(),
246 };
247 result.* = 1;
248 }
249
250 fn sleepUncancelable(io: Io, result: *usize) void {
251 const old_prot = io.swapCancelProtection(.blocked);
252 defer _ = io.swapCancelProtection(old_prot);
253 // Short sleep interval, because this one won't be canceled (that's the point!).
254 io.sleep(.fromMilliseconds(50), .awake) catch {};
255 result.* = 1;
256 }
257 };
258
259 const io = testing.io;
260
261 var group: Io.Group = .init;
262 var results: [5]usize = @splat(0);
263
264 group.concurrent(io, global.sleep, .{ io, &results[0] }) catch |err| switch (err) {
265 error.ConcurrencyUnavailable => return error.SkipZigTest,
266 };
267 try group.concurrent(io, global.sleep, .{ io, &results[1] });
268 try group.concurrent(io, global.sleepRecancel, .{ io, &results[2] });
269 try group.concurrent(io, global.sleepUncancelable, .{ io, &results[3] });
270 // Because this one doesn't block until canceled, it is safe to run asynchronously.
271 group.async(io, global.sleepUncancelable, .{ io, &results[4] });
272
273 group.cancel(io);
274
275 try testing.expectEqualSlices(usize, &.{ 1, 1, 1, 1, 1 }, &results);
276}
277
278test "Group.concurrent" {
279 const io = testing.io;
280
281 var group: Io.Group = .init;
282 defer group.cancel(io);
283 var results: [2]usize = undefined;
284
285 group.concurrent(io, count, .{ 1, 10, &results[0] }) catch |err| switch (err) {
286 error.ConcurrencyUnavailable => {
287 try expect(builtin.single_threaded);
288 return;
289 },
290 };
291
292 group.concurrent(io, count, .{ 20, 30, &results[1] }) catch |err| switch (err) {
293 error.ConcurrencyUnavailable => {
294 try expect(builtin.single_threaded);
295 return;
296 },
297 };
298
299 try group.await(io);
300
301 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
302}
303
304test "Group materializes error.Cancel" {
305 const S = struct {
306 fn task() Io.Cancelable!void {
307 return error.Canceled;
308 }
309 };
310
311 const io = testing.io;
312
313 var group: Io.Group = .init;
314
315 group.async(io, S.task, .{});
316 group.concurrent(io, S.task, .{}) catch |err| switch (err) {
317 error.ConcurrencyUnavailable => {
318 try expect(builtin.single_threaded);
319 return;
320 },
321 };
322
323 try group.await(io);
324}
325
326test "Group task receives cancelation unknowingly" {
327 const S = struct {
328 io: Io,
329 err: ?Io.Cancelable!void,
330
331 fn task(s: *@This()) void {
332 foo(s);
333 }
334
335 fn foo(s: *@This()) void {
336 s.err = s.io.sleep(.fromSeconds(300), .awake);
337 }
338 };
339
340 const io = testing.io;
341
342 var group: Io.Group = .init;
343 var result: S = .{ .io = io, .err = null };
344 group.concurrent(io, S.task, .{&result}) catch |err| switch (err) {
345 error.ConcurrencyUnavailable => {
346 try expect(builtin.single_threaded);
347 return;
348 },
349 };
350 group.cancel(io);
351
352 try expectError(error.Canceled, result.err.?);
353}
354
355fn testQueue(comptime len: usize) !void {
356 const io = testing.io;
357 var buf: [len]usize = undefined;
358 var queue: Io.Queue(usize) = .init(&buf);
359 var begin: usize = 0;
360 for (1..len + 1) |n| {
361 const end = begin + n;
362 for (begin..end) |i| try queue.putOne(io, i);
363 for (begin..end) |i| try expect(try queue.getOne(io) == i);
364 begin = end;
365 }
366}
367
368test "Queue" {
369 try testQueue(1);
370 try testQueue(2);
371 try testQueue(3);
372 try testQueue(4);
373 try testQueue(5);
374}
375
376test "Queue.close single-threaded" {
377 const io = std.testing.io;
378
379 var buf: [10]u8 = undefined;
380 var queue: Io.Queue(u8) = .init(&buf);
381
382 try queue.putAll(io, &.{ 0, 1, 2, 3, 4, 5, 6 });
383 try expectEqual(3, try queue.put(io, &.{ 7, 8, 9, 10 }, 0)); // there is capacity for 3 more items
384
385 var get_buf: [4]u8 = undefined;
386
387 // Receive some elements before closing
388 try expectEqual(4, try queue.get(io, &get_buf, 0));
389 try expectEqual(0, get_buf[0]);
390 try expectEqual(1, get_buf[1]);
391 try expectEqual(2, get_buf[2]);
392 try expectEqual(3, get_buf[3]);
393 try expectEqual(4, try queue.getOne(io));
394
395 // ...and add a couple more now there's space
396 try queue.putAll(io, &.{ 20, 21 });
397
398 queue.close(io);
399
400 // Receive more elements *after* closing
401 try expectEqual(4, try queue.get(io, &get_buf, 0));
402 try expectEqual(5, get_buf[0]);
403 try expectEqual(6, get_buf[1]);
404 try expectEqual(7, get_buf[2]);
405 try expectEqual(8, get_buf[3]);
406 try expectEqual(9, try queue.getOne(io));
407
408 // Cannot put anything while closed, even if the buffer has space
409 try expectError(error.Closed, queue.putOne(io, 100));
410 try expectError(error.Closed, queue.putAll(io, &.{ 101, 102 }));
411 try expectError(error.Closed, queue.putUncancelable(io, &.{ 103, 104 }, 0));
412
413 // Even if we ask for 3 items, the queue is closed, so we only get the last 2
414 try expectEqual(2, try queue.get(io, &get_buf, 4));
415 try expectEqual(20, get_buf[0]);
416 try expectEqual(21, get_buf[1]);
417
418 // The queue is now empty, so `get` should return `error.Closed` too
419 try expectError(error.Closed, queue.getOne(io));
420 try expectError(error.Closed, queue.get(io, &get_buf, 0));
421 try expectError(error.Closed, queue.putUncancelable(io, &get_buf, 2));
422}
423
424test "Event" {
425 const global = struct {
426 fn waitAndRead(io: Io, event: *Io.Event, ptr: *const u32) Io.Cancelable!u32 {
427 try event.wait(io);
428 return ptr.*;
429 }
430 };
431
432 const io = std.testing.io;
433
434 var event: Io.Event = .unset;
435 var buffer: u32 = undefined;
436
437 {
438 var future = io.concurrent(global.waitAndRead, .{ io, &event, &buffer }) catch |err| switch (err) {
439 error.ConcurrencyUnavailable => return error.SkipZigTest,
440 };
441
442 buffer = 123;
443 event.set(io);
444
445 const result = try future.await(io);
446
447 try std.testing.expectEqual(123, result);
448 }
449
450 event.reset();
451
452 {
453 var future = io.concurrent(global.waitAndRead, .{ io, &event, &buffer }) catch |err| switch (err) {
454 error.ConcurrencyUnavailable => return error.SkipZigTest,
455 };
456 try std.testing.expectError(error.Canceled, future.cancel(io));
457 }
458}
459
460test "recancel" {
461 const global = struct {
462 fn worker(io: Io) Io.Cancelable!void {
463 var dummy_event: Io.Event = .unset;
464
465 if (dummy_event.wait(io)) {
466 return;
467 } else |err| switch (err) {
468 error.Canceled => io.recancel(),
469 }
470
471 // Now we expect to see `error.Canceled` again.
472 return dummy_event.wait(io);
473 }
474 };
475
476 const io = std.testing.io;
477 var future = io.concurrent(global.worker, .{io}) catch |err| switch (err) {
478 error.ConcurrencyUnavailable => return error.SkipZigTest,
479 };
480 if (future.cancel(io)) {
481 return error.UnexpectedSuccess; // both `wait` calls should have returned `error.Canceled`
482 } else |err| switch (err) {
483 error.Canceled => {},
484 }
485}
486
487test "swapCancelProtection" {
488 const global = struct {
489 fn waitTwice(
490 io: Io,
491 event: *Io.Event,
492 ) error{ Canceled, CanceledWhileProtected }!void {
493 // Wait for `event` while protected from cancelation.
494 {
495 const old_prot = io.swapCancelProtection(.blocked);
496 defer _ = io.swapCancelProtection(old_prot);
497 event.wait(io) catch |err| switch (err) {
498 error.Canceled => return error.CanceledWhileProtected,
499 };
500 }
501 // Reset the event (it will never be set again), and this time wait for it without protection.
502 event.reset();
503 _ = try event.wait(io);
504 }
505 fn sleepThenSet(io: Io, event: *Io.Event) !void {
506 // Give `waitTwice` a chance to get canceled.
507 try io.sleep(.fromMilliseconds(200), .awake);
508 event.set(io);
509 }
510 };
511
512 const io = std.testing.io;
513
514 var event: Io.Event = .unset;
515
516 var wait_future = io.concurrent(global.waitTwice, .{ io, &event }) catch |err| switch (err) {
517 error.ConcurrencyUnavailable => return error.SkipZigTest,
518 };
519 defer wait_future.cancel(io) catch {};
520
521 var set_future = try io.concurrent(global.sleepThenSet, .{ io, &event });
522 defer set_future.cancel(io) catch {};
523
524 if (wait_future.cancel(io)) {
525 return error.UnexpectedSuccess; // there was no `set` call to unblock the second `wait`
526 } else |err| switch (err) {
527 error.Canceled => {},
528 error.CanceledWhileProtected => |e| return e,
529 }
530
531 // Because it reached the `set`, it should be too late for `sleepThenSet` to see `error.Canceled`.
532 try set_future.cancel(io);
533}
534
535test "cancel futex wait" {
536 const global = struct {
537 fn blockUntilCanceled(io: Io) void {
538 while (true) io.futexWait(u32, &0, 0) catch |err| switch (err) {
539 error.Canceled => return,
540 };
541 }
542 };
543
544 const io = std.testing.io;
545
546 var future = io.concurrent(global.blockUntilCanceled, .{io}) catch |err| switch (err) {
547 error.ConcurrencyUnavailable => return error.SkipZigTest,
548 };
549 defer future.cancel(io);
550
551 // Give the task some time to start so that we cancel while it is blocked.
552 try io.sleep(.fromMilliseconds(20), .awake);
553}
554
555test "cancel sleep" {
556 const global = struct {
557 fn blockUntilCanceled(io: Io) void {
558 while (true) io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {
559 error.Canceled => return,
560 };
561 }
562 };
563
564 const io = std.testing.io;
565
566 var future = io.concurrent(global.blockUntilCanceled, .{io}) catch |err| switch (err) {
567 error.ConcurrencyUnavailable => return error.SkipZigTest,
568 };
569 defer future.cancel(io);
570
571 // Give the task some time to start so that we cancel while it is blocked.
572 try io.sleep(.fromMilliseconds(20), .awake);
573}
574
575test "tasks spawned in group after Group.cancel are canceled" {
576 const global = struct {
577 fn waitThenSpawn(io: Io, group: *Io.Group) void {
578 _ = io.swapCancelProtection(.blocked);
579 group.concurrent(io, blockUntilCanceled, .{io}) catch {};
580 io.sleep(.fromMilliseconds(10), .awake) catch unreachable;
581 group.concurrent(io, blockUntilCanceled, .{io}) catch {};
582 group.async(io, blockUntilCanceled, .{io});
583 }
584 fn blockUntilCanceled(io: Io) Io.Cancelable!void {
585 while (true) io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {
586 error.Canceled => |e| return e,
587 };
588 }
589 };
590
591 const io = std.testing.io;
592
593 var group: Io.Group = .init;
594 defer group.cancel(io);
595
596 group.concurrent(io, global.blockUntilCanceled, .{io}) catch |err| switch (err) {
597 error.ConcurrencyUnavailable => return error.SkipZigTest,
598 };
599 try io.sleep(.fromMilliseconds(10), .awake); // let that first sleep start up
600 try group.concurrent(io, global.waitThenSpawn, .{ io, &group });
601}
602
603test "random" {
604 const io = testing.io;
605
606 var a: u64 = undefined;
607 var b: u64 = undefined;
608 var c: u64 = undefined;
609
610 io.random(@ptrCast(&a));
611 io.random(@ptrCast(&b));
612 io.random(@ptrCast(&c));
613
614 try expect(a ^ b ^ c != 0);
615}
616
617test "randomSecure" {
618 const io = testing.io;
619
620 var buf_a: [50]u8 = undefined;
621 var buf_b: [50]u8 = undefined;
622 try io.randomSecure(&buf_a);
623 try io.randomSecure(&buf_b);
624 // If this test fails the chance is significantly higher that there is a bug than
625 // that two sets of 50 bytes were equal.
626 try expect(!mem.eql(u8, &buf_a, &buf_b));
627}
628
629test "memory mapping" {
630 if (builtin.cpu.arch == .hexagon) return error.SkipZigTest; // mmap returned EINVAL
631 if (builtin.cpu.arch.isSPARC()) return error.SkipZigTest; // mmap returned EINVAL
632 if (builtin.os.tag == .wasi and builtin.link_libc) {
633 // https://github.com/ziglang/zig/issues/20747 (open fd does not have write permission)
634 return error.SkipZigTest;
635 }
636
637 const io = testing.io;
638
639 var tmp = tmpDir(.{});
640 defer tmp.cleanup();
641
642 try tmp.dir.writeFile(io, .{
643 .sub_path = "blah.txt",
644 .data = "this is my data123",
645 });
646
647 {
648 var file = try tmp.dir.openFile(io, "blah.txt", .{ .mode = .read_write });
649 defer file.close(io);
650
651 var mm = try file.createMemoryMap(io, .{ .len = "this is my data123".len });
652 defer mm.destroy(io);
653
654 try expectEqualStrings("this is my data123", mm.memory);
655 mm.memory[4] = '9';
656 mm.memory[7] = '9';
657
658 try mm.write(io);
659 }
660
661 var buffer: [100]u8 = undefined;
662 const updated_contents = try tmp.dir.readFile(io, "blah.txt", &buffer);
663 try expectEqualStrings("this9is9my data123", updated_contents);
664
665 {
666 var file = try tmp.dir.openFile(io, "blah.txt", .{ .mode = .read_write });
667 defer file.close(io);
668
669 var mm = try file.createMemoryMap(io, .{
670 .len = "this9is9my".len,
671 });
672 defer mm.destroy(io);
673
674 try expectEqualStrings("this9is9my", mm.memory);
675
676 // Cross a page boundary to require an actual remap.
677 const new_len = std.heap.pageSize() * 2;
678 mm.setLength(io, new_len) catch |err| switch (err) {
679 error.OperationUnsupported => {
680 mm.destroy(io);
681 mm = try file.createMemoryMap(io, .{ .len = new_len });
682 },
683 else => |e| return e,
684 };
685 try mm.read(io);
686
687 try expectEqualStrings("this9is9my data123\x00\x00", mm.memory[0.."this9is9my data123\x00\x00".len]);
688 }
689}
690
691test "read from a file using Batch.awaitAsync API" {
692 const io = testing.io;
693
694 var tmp = tmpDir(.{});
695 defer tmp.cleanup();
696
697 try tmp.dir.writeFile(io, .{
698 .sub_path = "eyes.txt",
699 .data = "Heaven's been cheating the Hell out of me",
700 });
701 try tmp.dir.writeFile(io, .{
702 .sub_path = "saviour.txt",
703 .data = "Burn your thoughts, erase your will / to gods of suffering and tears",
704 });
705
706 var eyes_file = try tmp.dir.openFile(io, "eyes.txt", .{});
707 defer eyes_file.close(io);
708
709 var saviour_file = try tmp.dir.openFile(io, "saviour.txt", .{});
710 defer saviour_file.close(io);
711
712 var eyes_buf: [100]u8 = undefined;
713 var saviour_buf: [100]u8 = undefined;
714 var storage: [2]Io.Operation.Storage = undefined;
715 var batch: Io.Batch = .init(&storage);
716
717 // Tests add API because this provides coverage for both add and addAt.
718 try testing.expectEqual(0, batch.add(.{ .file_read_streaming = .{
719 .file = eyes_file,
720 .data = &.{&eyes_buf},
721 } }));
722 try testing.expectEqual(1, batch.add(.{ .file_read_streaming = .{
723 .file = saviour_file,
724 .data = &.{&saviour_buf},
725 } }));
726
727 // This API is supposed to *always* work even if the target has no
728 // concurrency primitives available.
729 try batch.awaitAsync(io);
730
731 while (batch.next()) |completion| {
732 switch (completion.index) {
733 0 => {
734 const n = try completion.result.file_read_streaming;
735 try expectEqualStrings(
736 "Heaven's been cheating the Hell out of me"[0..n],
737 eyes_buf[0..n],
738 );
739 },
740 1 => {
741 const n = try completion.result.file_read_streaming;
742 try expectEqualStrings(
743 "Burn your thoughts, erase your will / to gods of suffering and tears"[0..n],
744 saviour_buf[0..n],
745 );
746 },
747 else => return error.TestFailure,
748 }
749 }
750}
751
752test "Event smoke test" {
753 const io = testing.io;
754
755 var event: Io.Event = .unset;
756 try testing.expectEqual(false, event.isSet());
757
758 // make sure the event gets set
759 event.set(io);
760 try testing.expectEqual(true, event.isSet());
761
762 // make sure the event gets unset again
763 event.reset();
764 try testing.expectEqual(false, event.isSet());
765
766 // waits should timeout as there's no other thread to set the event
767 try testing.expectError(error.Timeout, event.waitTimeout(io, .{ .duration = .{
768 .raw = .zero,
769 .clock = .awake,
770 } }));
771 try testing.expectError(error.Timeout, event.waitTimeout(io, .{ .duration = .{
772 .raw = .fromMilliseconds(1),
773 .clock = .awake,
774 } }));
775
776 // set the event again and make sure waits complete
777 event.set(io);
778 try event.wait(io);
779 try event.waitTimeout(io, .{ .duration = .{ .raw = .fromMilliseconds(1), .clock = .awake } });
780 try testing.expectEqual(true, event.isSet());
781}
782
783test "Event signaling" {
784 if (builtin.single_threaded) {
785 // This test requires spawning threads.
786 return error.SkipZigTest;
787 }
788
789 const io = testing.io;
790
791 const Context = struct {
792 in: Io.Event = .unset,
793 out: Io.Event = .unset,
794 value: usize = 0,
795
796 fn input(self: *@This()) !void {
797 // wait for the value to become 1
798 try self.in.wait(io);
799 self.in.reset();
800 try testing.expectEqual(self.value, 1);
801
802 // bump the value and wake up output()
803 self.value = 2;
804 self.out.set(io);
805
806 // wait for output to receive 2, bump the value and wake us up with 3
807 try self.in.wait(io);
808 self.in.reset();
809 try testing.expectEqual(self.value, 3);
810
811 // bump the value and wake up output() for it to see 4
812 self.value = 4;
813 self.out.set(io);
814 }
815
816 fn output(self: *@This()) !void {
817 // start with 0 and bump the value for input to see 1
818 try testing.expectEqual(self.value, 0);
819 self.value = 1;
820 self.in.set(io);
821
822 // wait for input to receive 1, bump the value to 2 and wake us up
823 try self.out.wait(io);
824 self.out.reset();
825 try testing.expectEqual(self.value, 2);
826
827 // bump the value to 3 for input to see (rhymes)
828 self.value = 3;
829 self.in.set(io);
830
831 // wait for input to bump the value to 4 and receive no more (rhymes)
832 try self.out.wait(io);
833 self.out.reset();
834 try testing.expectEqual(self.value, 4);
835 }
836 };
837
838 var ctx = Context{};
839
840 const thread = try std.Thread.spawn(.{}, Context.output, .{&ctx});
841 defer thread.join();
842
843 try ctx.input();
844}
845
846test "Event broadcast" {
847 if (builtin.single_threaded) {
848 // This test requires spawning threads.
849 return error.SkipZigTest;
850 }
851
852 const io = testing.io;
853
854 const num_threads = 10;
855 const Barrier = struct {
856 event: Io.Event = .unset,
857 counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(num_threads),
858
859 fn wait(self: *@This()) !void {
860 if (self.counter.fetchSub(1, .acq_rel) == 1) {
861 self.event.set(io);
862 }
863 try self.event.wait(io);
864 }
865 };
866
867 const Context = struct {
868 start_barrier: Barrier = .{},
869 finish_barrier: Barrier = .{},
870
871 fn run(self: *@This()) !void {
872 try self.start_barrier.wait();
873 try self.finish_barrier.wait();
874 }
875 };
876
877 var ctx = Context{};
878 var threads: [num_threads - 1]std.Thread = undefined;
879
880 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Context.run, .{&ctx});
881 defer for (threads) |t| t.join();
882
883 try ctx.run();
884}
885
886test "Select" {
887 const S = struct {
888 fn foo() bool {
889 return true;
890 }
891
892 fn bar(io: Io) Io.Cancelable!void {
893 try io.sleep(.fromSeconds(300), .awake);
894 }
895
896 fn baz() error{Ignored}!u8 {
897 return 42;
898 }
899 };
900
901 const io = testing.io;
902
903 const U = union(enum) {
904 foo: bool,
905 bar: Io.Cancelable!void,
906 baz: error{Ignored}!u8,
907 };
908 var buffer: [4]U = undefined;
909 var select: Io.Select(U) = .init(io, &buffer);
910 defer _ = select.cancel();
911
912 select.async(.foo, S.foo, .{});
913 select.concurrent(.bar, S.bar, .{io}) catch |err| switch (err) {
914 error.ConcurrencyUnavailable => return error.SkipZigTest,
915 };
916
917 switch (try select.await()) {
918 .foo => {},
919 .bar => return error.TestFailed, // should be sleeping
920 .baz => return error.TestFailed, // not called yet
921 }
922 select.async(.foo, S.foo, .{});
923 select.async(.foo, S.foo, .{});
924
925 var finished_buffer: [3]U = undefined;
926 const finished = finished_buffer[0..try select.awaitMany(&finished_buffer, 2)];
927 try testing.expectEqualSlices(U, &.{ .{ .foo = true }, .{ .foo = true } }, finished);
928
929 select.async(.baz, S.baz, .{});
930
931 const result = switch (try select.await()) {
932 .baz => |n| try n,
933 .foo => return error.TestFailed, // not called
934 .bar => return error.TestFailed, // should be sleeping
935 };
936
937 try testing.expectEqual(42, result);
938}
939
940test "Select with empty buffer, no deadlock" {
941 const S = struct {
942 fn sleeper(io: Io, duration: Io.Duration) Io.Cancelable!void {
943 try io.sleep(duration, .awake);
944 }
945 };
946
947 const io = testing.io;
948
949 const U = union(enum) {
950 sleeper: Io.Cancelable!void,
951 };
952 var select: Io.Select(U) = .init(io, &.{});
953 defer select.cancelDiscard();
954
955 select.concurrent(.sleeper, S.sleeper, .{ io, .fromNanoseconds(1) }) catch |err| switch (err) {
956 error.ConcurrencyUnavailable => return error.SkipZigTest,
957 };
958 select.concurrent(.sleeper, S.sleeper, .{ io, .fromSeconds(600) }) catch |err| switch (err) {
959 error.ConcurrencyUnavailable => return error.SkipZigTest,
960 };
961 assert((try select.await()) == .sleeper);
962}
963
964test "Select.cancel with no tasks, no deadlock" {
965 const io = testing.io;
966
967 const U = union(enum) {
968 nothing: void,
969 also_nothing: void,
970 };
971 var select: Io.Select(U) = .init(io, &.{});
972 try expectEqual(null, select.cancel());
973}
974
975test "Condition.waitTimeout" {
976 const io = testing.io;
977
978 const Context = struct {
979 ready: Io.Event = .unset,
980 mutex: Io.Mutex = .init,
981 cond: Io.Condition = .init,
982 value: u32 = 0,
983
984 fn worker(ctx: *@This()) !void {
985 defer ctx.ready.set(io);
986
987 try ctx.mutex.lock(io);
988 defer ctx.mutex.unlock(io);
989
990 try expectError(error.Timeout, ctx.cond.waitTimeout(io, &ctx.mutex, .{ .duration = .{
991 .raw = .fromMilliseconds(1),
992 .clock = .awake,
993 } }));
994 try expectEqual(0, ctx.value);
995
996 ctx.ready.set(io);
997
998 while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex);
999 try expectEqual(1, ctx.value);
1000 }
1001 };
1002
1003 var ctx: Context = .{};
1004
1005 var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) {
1006 error.ConcurrencyUnavailable => return error.SkipZigTest,
1007 };
1008 defer future.cancel(io) catch {};
1009
1010 try ctx.ready.wait(io);
1011
1012 try ctx.mutex.lock(io);
1013 ctx.value = 1;
1014 ctx.mutex.unlock(io);
1015 ctx.cond.signal(io);
1016
1017 try future.await(io);
1018}
1019
1020test "Condition.waitUncancelable" {
1021 const io = testing.io;
1022
1023 const Context = struct {
1024 ready: Io.Event = .unset,
1025 mutex: Io.Mutex = .init,
1026 cond: Io.Condition = .init,
1027 value: u32 = 0,
1028
1029 fn worker(ctx: *@This()) !void {
1030 defer ctx.ready.set(io);
1031
1032 try ctx.mutex.lock(io);
1033 defer ctx.mutex.unlock(io);
1034
1035 try expectEqual(0, ctx.value);
1036
1037 ctx.ready.set(io);
1038
1039 ctx.cond.waitUncancelable(io, &ctx.mutex);
1040
1041 while (ctx.value == 0) try ctx.cond.wait(io, &ctx.mutex);
1042 try expectEqual(1, ctx.value);
1043 }
1044 };
1045
1046 var ctx: Context = .{};
1047
1048 var future = io.concurrent(Context.worker, .{&ctx}) catch |err| switch (err) {
1049 error.ConcurrencyUnavailable => return error.SkipZigTest,
1050 };
1051 defer future.cancel(io) catch {};
1052
1053 try ctx.ready.wait(io);
1054
1055 try ctx.mutex.lock(io);
1056 ctx.value = 1;
1057 ctx.mutex.unlock(io);
1058 ctx.cond.signal(io);
1059
1060 try future.await(io);
1061}