3 files changed, 97 insertions, 1 deletions
diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig
index 995b9eb43c..109b467c41 100644
--- a/lib/std/Io/Threaded.zig
+++ b/lib/std/Io/Threaded.zig
@@ -6114,3 +6114,7 @@ fn initializeWsa(t: *Threaded) error{NetworkDown}!void {
     }
     return error.NetworkDown;
 }
+
+test {
+    _ = @import("Threaded/test.zig");
+}
diff --git a/lib/std/Io/Threaded/test.zig b/lib/std/Io/Threaded/test.zig
new file mode 100644
index 0000000000..6b8b858075
--- /dev/null
+++ b/lib/std/Io/Threaded/test.zig
@@ -0,0 +1,46 @@
+const std = @import("std");
+const Io = std.Io;
+const testing = std.testing;
+const assert = std.debug.assert;
+
+test "concurrent vs main prevents deadlock via oversubscription" {
+    var threaded: Io.Threaded = .init(std.testing.allocator);
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    threaded.cpu_count = 1;
+
+    var queue: Io.Queue(u8) = .init(&.{});
+
+    var putter = try io.concurrent(put, .{ io, &queue });
+    defer putter.cancel(io);
+
+    try testing.expectEqual(42, queue.getOneUncancelable(io));
+}
+
+fn put(io: Io, queue: *Io.Queue(u8)) void {
+    queue.putOneUncancelable(io, 42);
+}
+
+fn get(io: Io, queue: *Io.Queue(u8)) void {
+    assert(queue.getOneUncancelable(io) == 42);
+}
+
+test "concurrent vs concurrent prevents deadlock via oversubscription" {
+    var threaded: Io.Threaded = .init(std.testing.allocator);
+    defer threaded.deinit();
+    const io = threaded.io();
+
+    threaded.cpu_count = 1;
+
+    var queue: Io.Queue(u8) = .init(&.{});
+
+    var putter = try io.concurrent(put, .{ io, &queue });
+    defer putter.cancel(io);
+
+    var getter = try io.concurrent(get, .{ io, &queue });
+    defer getter.cancel(io);
+
+    getter.await(io);
+    putter.await(io);
+}
diff --git a/lib/std/Io/test.zig b/lib/std/Io/test.zig
index 22f211d6e4..f47da2c9e1 100644
--- a/lib/std/Io/test.zig
+++ b/lib/std/Io/test.zig
@@ -1,4 +1,8 @@
+const builtin = @import("builtin");
+const native_endian = builtin.cpu.arch.endian();
+
 const std = @import("std");
+const Io = std.Io;
 const DefaultPrng = std.Random.DefaultPrng;
 const expect = std.testing.expect;
 const expectEqual = std.testing.expectEqual;
@@ -6,7 +10,6 @@ const expectError = std.testing.expectError;
 const mem = std.mem;
 const fs = std.fs;
 const File = std.fs.File;
-const native_endian = @import("builtin").target.cpu.arch.endian();
 
 const tmpDir = std.testing.tmpDir;
 
@@ -123,3 +126,46 @@ test "updateTimes" {
     try expect(stat_new.atime.nanoseconds < stat_old.atime.nanoseconds);
     try expect(stat_new.mtime.nanoseconds < stat_old.mtime.nanoseconds);
 }
+
+test "Group" {
+    const io = std.testing.io;
+
+    var group: Io.Group = .init;
+    var results: [2]usize = undefined;
+
+    group.async(io, count, .{ 1, 10, &results[0] });
+    group.async(io, count, .{ 20, 30, &results[1] });
+
+    group.wait(io);
+
+    try std.testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
+}
+
+fn count(a: usize, b: usize, result: *usize) void {
+    var sum: usize = 0;
+    for (a..b) |i| {
+        sum += i;
+    }
+    result.* = sum;
+}
+
+test "Group cancellation" {
+    const io = std.testing.io;
+
+    var group: Io.Group = .init;
+    var results: [2]usize = undefined;
+
+    group.async(io, sleep, .{ io, &results[0] });
+    group.async(io, sleep, .{ io, &results[1] });
+
+    group.cancel(io);
+
+    try std.testing.expectEqualSlices(usize, &.{ 1, 1 }, &results);
+}
+
+fn sleep(io: Io, result: *usize) void {
+    // TODO when cancellation race bug is fixed, make this timeout much longer so that
+    // it causes the unit test to be failed if not cancelled.
+    io.sleep(.fromMilliseconds(1), .awake) catch {};
+    result.* = 1;
+}