authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-27 13:38:48-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-28 01:46:43+01:00
log8c70fd0a57936cca236a000a17364cbbf7a09673
treedd753fb0a0e0c1e0e39c863985ece4f4e6581afb
parente779ca72230b5975f48c84ae19c51f0c74f9ecbe

std.Io.Select: introduce cancelDiscard

and make the return value of `cancel` return queue items. I don't think it's possible to make `cancel` not deadlock with an empty queue buffer without introducing a new Group primitive. This is the best I could come up with based on existing primitives. Let's see if applications find these APIs palatable.

2 files changed, 87 insertions(+), 12 deletions(-)

lib/std/Io.zig+62-11
...@@ -1184,8 +1184,6 @@ pub fn Select(comptime U: type) type {...@@ -1184,8 +1184,6 @@ pub fn Select(comptime U: type) type {
1184 return struct {1184 return struct {
1185 io: Io,1185 io: Io,
1186 group: Group,1186 group: Group,
1187 /// The queue is never closed because there may be live resources
1188 /// inserted into it which would otherwise leak.
1189 queue: Queue(U),1187 queue: Queue(U),
11901188
1191 const S = @This();1189 const S = @This();
...@@ -1235,7 +1233,7 @@ pub fn Select(comptime U: type) type {...@@ -1235,7 +1233,7 @@ pub fn Select(comptime U: type) type {
1235 const raw_result = @call(.auto, function, context.args);1233 const raw_result = @call(.auto, function, context.args);
1236 const elem = @unionInit(U, @tagName(field), raw_result);1234 const elem = @unionInit(U, @tagName(field), raw_result);
1237 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {1235 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {
1238 error.Closed => unreachable,1236 error.Closed => {},
1239 };1237 };
1240 if (@typeInfo(@TypeOf(raw_result)) == .error_union)1238 if (@typeInfo(@TypeOf(raw_result)) == .error_union)
1241 _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled;1239 _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled;
...@@ -1274,7 +1272,7 @@ pub fn Select(comptime U: type) type {...@@ -1274,7 +1272,7 @@ pub fn Select(comptime U: type) type {
1274 const raw_result = @call(.auto, function, context.args);1272 const raw_result = @call(.auto, function, context.args);
1275 const elem = @unionInit(U, @tagName(field), raw_result);1273 const elem = @unionInit(U, @tagName(field), raw_result);
1276 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {1274 context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) {
1277 error.Closed => unreachable,1275 error.Closed => {},
1278 };1276 };
1279 if (@typeInfo(@TypeOf(raw_result)) == .error_union)1277 if (@typeInfo(@TypeOf(raw_result)) == .error_union)
1280 _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled;1278 _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled;
...@@ -1286,6 +1284,8 @@ pub fn Select(comptime U: type) type {...@@ -1286,6 +1284,8 @@ pub fn Select(comptime U: type) type {
12861284
1287 /// Blocks until another task of the select finishes.1285 /// Blocks until another task of the select finishes.
1288 ///1286 ///
1287 /// It is legal to call `async` and `concurrent` after this.
1288 ///
1289 /// Threadsafe.1289 /// Threadsafe.
1290 pub fn await(s: *S) Cancelable!U {1290 pub fn await(s: *S) Cancelable!U {
1291 return s.queue.getOne(s.io) catch |err| switch (err) {1291 return s.queue.getOne(s.io) catch |err| switch (err) {
...@@ -1299,6 +1299,8 @@ pub fn Select(comptime U: type) type {...@@ -1299,6 +1299,8 @@ pub fn Select(comptime U: type) type {
1299 ///1299 ///
1300 /// Asserts that `buffer.len >= min`.1300 /// Asserts that `buffer.len >= min`.
1301 ///1301 ///
1302 /// It is legal to call `async` and `concurrent` after this.
1303 ///
1302 /// Threadsafe.1304 /// Threadsafe.
1303 pub fn awaitMany(s: *S, buffer: []U, min: usize) Cancelable!usize {1305 pub fn awaitMany(s: *S, buffer: []U, min: usize) Cancelable!usize {
1304 return s.queue.get(s.io, buffer, min) catch |err| switch (err) {1306 return s.queue.get(s.io, buffer, min) catch |err| switch (err) {
...@@ -1307,16 +1309,53 @@ pub fn Select(comptime U: type) type {...@@ -1307,16 +1309,53 @@ pub fn Select(comptime U: type) type {
1307 };1309 };
1308 }1310 }
13091311
1310 /// Equivalent to `await` but requests cancelation on all remaining1312 /// Requests cancelation on all remaining tasks owned by the select,
1311 /// tasks owned by the select.1313 /// then blocks until they all finish. If the select was initialized
1314 /// with insufficient buffer space for all remaining tasks to finish, a
1315 /// deadlock occurs.
1312 ///1316 ///
1313 /// For a description of cancelation and cancelation points, see `Future.cancel`.1317 /// If any of the select tasks allocate resources, those tasks may have
1318 /// completed, meaning that this function must be called in a loop
1319 /// until `null` is returned in order to deallocate those resources. If
1320 /// there is no possibility of resource leaks, `cancelDiscard` is
1321 /// preferable.
1314 ///1322 ///
1315 /// It is illegal to call `await` after this.1323 /// It is illegal to call `await` or `awaitMany` after this.
1316 ///1324 ///
1317 /// Idempotent. Threadsafe.1325 /// It is safe to call this multiple times, even after `null` is
1318 pub fn cancel(s: *S) void {1326 /// returned.
1319 s.group.cancel(s.io);1327 ///
1328 /// Threadsafe.
1329 pub fn cancel(s: *S) ?U {
1330 const io = s.io;
1331 if (s.group.token.load(.acquire)) |token| {
1332 io.vtable.groupCancel(io.userdata, &s.group, token);
1333 assert(s.group.token.raw == null);
1334 s.queue.close(io);
1335 }
1336 return s.queue.getOneUncancelable(io) catch |err| switch (err) {
1337 error.Closed => return null,
1338 };
1339 }
1340
1341 /// Requests cancelation on all remaining tasks owned by the select,
1342 /// then blocks until they all finish.
1343 ///
1344 /// All return values from outstanding tasks are discarded. This
1345 /// function is therefore inappropriate to call when a task can return
1346 /// an allocated resource. For that use case, see `cancel`.
1347 ///
1348 /// It is illegal to call `await` or `awaitMany` after this.
1349 ///
1350 /// It is safe to call this multiple times.
1351 ///
1352 /// Threadsafe.
1353 pub fn cancelDiscard(s: *S) void {
1354 const io = s.io;
1355 const token = s.group.token.load(.acquire) orelse return;
1356 s.queue.close(io);
1357 io.vtable.groupCancel(io.userdata, &s.group, token);
1358 assert(s.group.token.raw == null);
1320 }1359 }
1321 };1360 };
1322}1361}
...@@ -1693,6 +1732,12 @@ pub const TypeErasedQueue = struct {...@@ -1693,6 +1732,12 @@ pub const TypeErasedQueue = struct {
1693 };1732 };
1694 }1733 }
16951734
1735 /// After this is called, the queue enters a "closed" state. A closed
1736 /// queue always returns `error.Closed` for put attempts even when
1737 /// there is space in the buffer. However, existing elements of the
1738 /// queue are retrieved before `error.Closed` is returned.
1739 ///
1740 /// Threadsafe.
1696 pub fn close(q: *TypeErasedQueue, io: Io) void {1741 pub fn close(q: *TypeErasedQueue, io: Io) void {
1697 q.mutex.lockUncancelable(io);1742 q.mutex.lockUncancelable(io);
1698 defer q.mutex.unlock(io);1743 defer q.mutex.unlock(io);
...@@ -1967,6 +2012,12 @@ pub fn Queue(Elem: type) type {...@@ -1967,6 +2012,12 @@ pub fn Queue(Elem: type) type {
1967 return .{ .type_erased = .init(@ptrCast(buffer)) };2012 return .{ .type_erased = .init(@ptrCast(buffer)) };
1968 }2013 }
19692014
2015 /// After this is called, the queue enters a "closed" state. A closed
2016 /// queue always returns `error.Closed` for put attempts even when
2017 /// there is space in the buffer. However, existing elements of the
2018 /// queue are retrieved before `error.Closed` is returned.
2019 ///
2020 /// Threadsafe.
1970 pub fn close(q: *@This(), io: Io) void {2021 pub fn close(q: *@This(), io: Io) void {
1971 q.type_erased.close(io);2022 q.type_erased.close(io);
1972 }2023 }
lib/std/Io/test.zig+25-1
...@@ -835,7 +835,7 @@ test "Select" {...@@ -835,7 +835,7 @@ test "Select" {
835 };835 };
836 var buffer: [4]U = undefined;836 var buffer: [4]U = undefined;
837 var select: Io.Select(U) = .init(io, &buffer);837 var select: Io.Select(U) = .init(io, &buffer);
838 defer select.cancel();838 defer _ = select.cancel();
839839
840 select.async(.foo, S.foo, .{});840 select.async(.foo, S.foo, .{});
841 select.concurrent(.bar, S.bar, .{io}) catch |err| switch (err) {841 select.concurrent(.bar, S.bar, .{io}) catch |err| switch (err) {
...@@ -864,3 +864,27 @@ test "Select" {...@@ -864,3 +864,27 @@ test "Select" {
864864
865 try testing.expectEqual(42, result);865 try testing.expectEqual(42, result);
866}866}
867
868test "Select with empty buffer, no deadlock" {
869 const S = struct {
870 fn sleeper(io: Io, duration: Io.Duration) Io.Cancelable!void {
871 try io.sleep(duration, .awake);
872 }
873 };
874
875 const io = testing.io;
876
877 const U = union(enum) {
878 sleeper: Io.Cancelable!void,
879 };
880 var select: Io.Select(U) = .init(io, &.{});
881 defer select.cancelDiscard();
882
883 select.concurrent(.sleeper, S.sleeper, .{ io, .fromNanoseconds(1) }) catch |err| switch (err) {
884 error.ConcurrencyUnavailable => return error.SkipZigTest,
885 };
886 select.concurrent(.sleeper, S.sleeper, .{ io, .fromSeconds(600) }) catch |err| switch (err) {
887 error.ConcurrencyUnavailable => return error.SkipZigTest,
888 };
889 assert((try select.await()) == .sleeper);
890}