| ... | @@ -1206,7 +1206,7 @@ pub fn Select(comptime U: type) type { | ... | @@ -1206,7 +1206,7 @@ pub fn Select(comptime U: type) type { |
| 1206 | /// already been called and completed, or it has successfully been | 1206 | /// already been called and completed, or it has successfully been |
| 1207 | /// assigned a unit of concurrency. | 1207 | /// assigned a unit of concurrency. |
| 1208 | /// | 1208 | /// |
| 1209 | /// After this is called, `wait` or `cancel` must be called before the | 1209 | /// After this is called, `await` or `cancel` must be called before the |
| 1210 | /// select is deinitialized. | 1210 | /// select is deinitialized. |
| 1211 | /// | 1211 | /// |
| 1212 | /// Threadsafe. | 1212 | /// Threadsafe. |
| ... | @@ -1225,10 +1225,13 @@ pub fn Select(comptime U: type) type { | ... | @@ -1225,10 +1225,13 @@ pub fn Select(comptime U: type) type { |
| 1225 | args: @TypeOf(args), | 1225 | args: @TypeOf(args), |
| 1226 | fn start(type_erased_context: *const anyopaque) Cancelable!void { | 1226 | fn start(type_erased_context: *const anyopaque) Cancelable!void { |
| 1227 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); | 1227 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); |
| 1228 | const elem = @unionInit(U, @tagName(field), @call(.auto, function, context.args)); | 1228 | const raw_result = @call(.auto, function, context.args); |
| | 1229 | const elem = @unionInit(U, @tagName(field), raw_result); |
| 1229 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { | 1230 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { |
| 1230 | error.Closed => unreachable, | 1231 | error.Closed => unreachable, |
| 1231 | }; | 1232 | }; |
| | 1233 | if (@typeInfo(@TypeOf(raw_result)) == .error_union) |
| | 1234 | raw_result catch |err| if (err == error.Canceled) return error.Canceled; |
| 1232 | } | 1235 | } |
| 1233 | }; | 1236 | }; |
| 1234 | const context: Context = .{ .select = s, .args = args }; | 1237 | const context: Context = .{ .select = s, .args = args }; |
| ... | @@ -1236,6 +1239,46 @@ pub fn Select(comptime U: type) type { | ... | @@ -1236,6 +1239,46 @@ pub fn Select(comptime U: type) type { |
| 1236 | s.io.vtable.groupAsync(s.io.userdata, &s.group, @ptrCast(&context), .of(Context), Context.start); | 1239 | s.io.vtable.groupAsync(s.io.userdata, &s.group, @ptrCast(&context), .of(Context), Context.start); |
| 1237 | } | 1240 | } |
| 1238 | | 1241 | |
| | 1242 | /// Calls `function` with `args` concurrently. The resource spawned is |
| | 1243 | /// owned by the select. |
| | 1244 | /// |
| | 1245 | /// `function` must have return type matching the `field` field of `Union`. |
| | 1246 | /// |
| | 1247 | /// After this function returns successfully, it is guaranteed that |
| | 1248 | /// `function` has been assigned a unit of concurrency, and `await` or |
| | 1249 | /// `cancel` must be called before the select is deinitialized. |
| | 1250 | /// |
| | 1251 | /// |
| | 1252 | /// Threadsafe. |
| | 1253 | /// |
| | 1254 | /// Related: |
| | 1255 | /// * `Io.concurrent` |
| | 1256 | /// * `Group.concurrent` |
| | 1257 | pub fn concurrent( |
| | 1258 | s: *S, |
| | 1259 | comptime field: Field, |
| | 1260 | function: anytype, |
| | 1261 | args: std.meta.ArgsTuple(@TypeOf(function)), |
| | 1262 | ) ConcurrentError!void { |
| | 1263 | const Context = struct { |
| | 1264 | select: *S, |
| | 1265 | args: @TypeOf(args), |
| | 1266 | fn start(type_erased_context: *const anyopaque) Cancelable!void { |
| | 1267 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); |
| | 1268 | const raw_result = @call(.auto, function, context.args); |
| | 1269 | const elem = @unionInit(U, @tagName(field), raw_result); |
| | 1270 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { |
| | 1271 | error.Closed => unreachable, |
| | 1272 | }; |
| | 1273 | if (@typeInfo(@TypeOf(raw_result)) == .error_union) |
| | 1274 | raw_result catch |err| if (err == error.Canceled) return error.Canceled; |
| | 1275 | } |
| | 1276 | }; |
| | 1277 | const context: Context = .{ .select = s, .args = args }; |
| | 1278 | try s.io.vtable.groupConcurrent(s.io.userdata, &s.group, @ptrCast(&context), .of(Context), Context.start); |
| | 1279 | _ = @atomicRmw(usize, &s.outstanding, .Add, 1, .monotonic); |
| | 1280 | } |
| | 1281 | |
| 1239 | /// Blocks until another task of the select finishes. | 1282 | /// Blocks until another task of the select finishes. |
| 1240 | /// | 1283 | /// |
| 1241 | /// Asserts there is at least one more `outstanding` task. | 1284 | /// Asserts there is at least one more `outstanding` task. |
| ... | @@ -1249,12 +1292,12 @@ pub fn Select(comptime U: type) type { | ... | @@ -1249,12 +1292,12 @@ pub fn Select(comptime U: type) type { |
| 1249 | }; | 1292 | }; |
| 1250 | } | 1293 | } |
| 1251 | | 1294 | |
| 1252 | /// Equivalent to `wait` but requests cancelation on all remaining | 1295 | /// Equivalent to `await` but requests cancelation on all remaining |
| 1253 | /// tasks owned by the select. | 1296 | /// tasks owned by the select. |
| 1254 | /// | 1297 | /// |
| 1255 | /// For a description of cancelation and cancelation points, see `Future.cancel`. | 1298 | /// For a description of cancelation and cancelation points, see `Future.cancel`. |
| 1256 | /// | 1299 | /// |
| 1257 | /// It is illegal to call `wait` after this. | 1300 | /// It is illegal to call `await` after this. |
| 1258 | /// | 1301 | /// |
| 1259 | /// Idempotent. Not threadsafe. | 1302 | /// Idempotent. Not threadsafe. |
| 1260 | pub fn cancel(s: *S) void { | 1303 | pub fn cancel(s: *S) void { |
Andrew Kelley <andrew@ziglang.org> 2026-02-21 01:29:06+01:00
lib