authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-04-02 18:03:53-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
log7b5886118dabb59967d3e9b17d0502146df2ef92
tree8594534f260e37114700ebe6fb2d1f92e15fafa8
parentfdf15fae9779c1b0fc05b12abf0865245dd95d4f

Io.Condition: implement full API

3 files changed, 78 insertions(+), 24 deletions(-)

lib/std/Io.zig+18-7
...@@ -983,7 +983,7 @@ pub const VTable = struct {...@@ -983,7 +983,7 @@ pub const VTable = struct {
983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,983 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
984984
985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,985 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,
986 conditionWake: *const fn (?*anyopaque, cond: *Condition) void,986 conditionWake: *const fn (?*anyopaque, cond: *Condition, wake: Condition.Wake) void,
987987
988 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,988 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
989 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,989 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
...@@ -1162,9 +1162,20 @@ pub const Condition = struct {...@@ -1162,9 +1162,20 @@ pub const Condition = struct {
1162 return io.vtable.conditionWait(io.userdata, cond, mutex);1162 return io.vtable.conditionWait(io.userdata, cond, mutex);
1163 }1163 }
11641164
1165 pub fn wake(cond: *Condition, io: Io) void {1165 pub fn signal(cond: *Condition, io: Io) void {
1166 io.vtable.conditionWake(io.userdata, cond);1166 io.vtable.conditionWake(io.userdata, cond, .one);
1167 }1167 }
1168
1169 pub fn broadcast(cond: *Condition, io: Io) void {
1170 io.vtable.conditionWake(io.userdata, cond, .all);
1171 }
1172
1173 pub const Wake = enum {
1174 /// wake up only one thread
1175 one,
1176 /// wake up all thread
1177 all,
1178 };
1168};1179};
11691180
1170pub const TypeErasedQueue = struct {1181pub const TypeErasedQueue = struct {
...@@ -1216,7 +1227,7 @@ pub const TypeErasedQueue = struct {...@@ -1216,7 +1227,7 @@ pub const TypeErasedQueue = struct {
1216 remaining = remaining[copy_len..];1227 remaining = remaining[copy_len..];
1217 getter.data.remaining = getter.data.remaining[copy_len..];1228 getter.data.remaining = getter.data.remaining[copy_len..];
1218 if (getter.data.remaining.len == 0) {1229 if (getter.data.remaining.len == 0) {
1219 getter.data.condition.wake(io);1230 getter.data.condition.signal(io);
1220 continue;1231 continue;
1221 }1232 }
1222 q.getters.prepend(getter);1233 q.getters.prepend(getter);
...@@ -1299,7 +1310,7 @@ pub const TypeErasedQueue = struct {...@@ -1299,7 +1310,7 @@ pub const TypeErasedQueue = struct {
1299 putter.data.remaining = putter.data.remaining[copy_len..];1310 putter.data.remaining = putter.data.remaining[copy_len..];
1300 remaining = remaining[copy_len..];1311 remaining = remaining[copy_len..];
1301 if (putter.data.remaining.len == 0) {1312 if (putter.data.remaining.len == 0) {
1302 putter.data.condition.wake(io);1313 putter.data.condition.signal(io);
1303 } else {1314 } else {
1304 assert(remaining.len == 0);1315 assert(remaining.len == 0);
1305 q.putters.prepend(putter);1316 q.putters.prepend(putter);
...@@ -1332,7 +1343,7 @@ pub const TypeErasedQueue = struct {...@@ -1332,7 +1343,7 @@ pub const TypeErasedQueue = struct {
1332 putter.data.remaining = putter.data.remaining[copy_len..];1343 putter.data.remaining = putter.data.remaining[copy_len..];
1333 q.put_index += copy_len;1344 q.put_index += copy_len;
1334 if (putter.data.remaining.len == 0) {1345 if (putter.data.remaining.len == 0) {
1335 putter.data.condition.wake(io);1346 putter.data.condition.signal(io);
1336 continue;1347 continue;
1337 }1348 }
1338 const second_available = q.buffer[0..q.get_index];1349 const second_available = q.buffer[0..q.get_index];
...@@ -1341,7 +1352,7 @@ pub const TypeErasedQueue = struct {...@@ -1341,7 +1352,7 @@ pub const TypeErasedQueue = struct {
1341 putter.data.remaining = putter.data.remaining[copy_len..];1352 putter.data.remaining = putter.data.remaining[copy_len..];
1342 q.put_index = copy_len;1353 q.put_index = copy_len;
1343 if (putter.data.remaining.len == 0) {1354 if (putter.data.remaining.len == 0) {
1344 putter.data.condition.wake(io);1355 putter.data.condition.signal(io);
1345 continue;1356 continue;
1346 }1357 }
1347 q.putters.prepend(putter);1358 q.putters.prepend(putter);
lib/std/Io/EventLoop.zig+55-15
...@@ -555,8 +555,24 @@ const SwitchMessage = struct {...@@ -555,8 +555,24 @@ const SwitchMessage = struct {
555 .condition_wait => |condition_wait| {555 .condition_wait => |condition_wait| {
556 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));556 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
557 assert(prev_fiber.queue_next == null);557 assert(prev_fiber.queue_next == null);
558 const cond_state: *?*Fiber = @ptrCast(&condition_wait.cond.state);558 const cond_impl = prev_fiber.resultPointer(ConditionImpl);
559 assert(@atomicRmw(?*Fiber, cond_state, .Xchg, prev_fiber, .release) == null); // More than one wait on same Condition is illegal.559 cond_impl.* = .{
560 .tail = prev_fiber,
561 .event = .queued,
562 };
563 if (@cmpxchgStrong(
564 ?*Fiber,
565 @as(*?*Fiber, @ptrCast(&condition_wait.cond.state)),
566 null,
567 prev_fiber,
568 .release,
569 .acquire,
570 )) |waiting_fiber| {
571 const waiting_cond_impl = waiting_fiber.?.resultPointer(ConditionImpl);
572 assert(waiting_cond_impl.tail.queue_next == null);
573 waiting_cond_impl.tail.queue_next = prev_fiber;
574 waiting_cond_impl.tail = prev_fiber;
575 }
560 condition_wait.mutex.unlock(el.io());576 condition_wait.mutex.unlock(el.io());
561 },577 },
562 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {578 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
...@@ -1267,10 +1283,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl...@@ -1267,10 +1283,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl
12671283
1268fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {1284fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
1269 const el: *EventLoop = @alignCast(@ptrCast(userdata));1285 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1270 el.yield(null, .{ .mutex_lock = .{1286 el.yield(null, .{ .mutex_lock = .{ .prev_state = prev_state, .mutex = mutex } });
1271 .prev_state = prev_state,
1272 .mutex = mutex,
1273 } });
1274}1287}
1275fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {1288fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
1276 var maybe_waiting_fiber: ?*Fiber = @ptrFromInt(@intFromEnum(prev_state));1289 var maybe_waiting_fiber: ?*Fiber = @ptrFromInt(@intFromEnum(prev_state));
...@@ -1294,21 +1307,48 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut...@@ -1294,21 +1307,48 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut
1294 el.yield(maybe_waiting_fiber.?, .reschedule);1307 el.yield(maybe_waiting_fiber.?, .reschedule);
1295}1308}
12961309
1310const ConditionImpl = struct {
1311 tail: *Fiber,
1312 event: union(enum) {
1313 queued,
1314 wake: Io.Condition.Wake,
1315 },
1316};
1317
1297fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {1318fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
1298 const el: *EventLoop = @alignCast(@ptrCast(userdata));1319 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1299 el.yield(null, .{ .condition_wait = .{1320 el.yield(null, .{ .condition_wait = .{ .cond = cond, .mutex = mutex } });
1300 .cond = cond,1321 const thread = Thread.current();
1301 .mutex = mutex,1322 const fiber = thread.currentFiber();
1302 } });1323 const cond_impl = fiber.resultPointer(ConditionImpl);
1303 try mutex.lock(el.io());1324 try mutex.lock(el.io());
1325 switch (cond_impl.event) {
1326 .queued => {},
1327 .wake => |wake| if (fiber.queue_next) |next_fiber| switch (wake) {
1328 .one => if (@cmpxchgStrong(
1329 ?*Fiber,
1330 @as(*?*Fiber, @ptrCast(&cond.state)),
1331 null,
1332 next_fiber,
1333 .release,
1334 .acquire,
1335 )) |old_fiber| {
1336 const old_cond_impl = old_fiber.?.resultPointer(ConditionImpl);
1337 assert(old_cond_impl.tail.queue_next == null);
1338 old_cond_impl.tail.queue_next = next_fiber;
1339 old_cond_impl.tail = cond_impl.tail;
1340 },
1341 .all => el.schedule(thread, .{ .head = next_fiber, .tail = cond_impl.tail }),
1342 },
1343 }
1344 fiber.queue_next = null;
1304}1345}
13051346
1306fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {1347fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.Wake) void {
1307 const el: *EventLoop = @alignCast(@ptrCast(userdata));1348 const el: *EventLoop = @alignCast(@ptrCast(userdata));
1308 const cond_state: *?*Fiber = @ptrCast(&cond.state);1349 const waiting_fiber = @atomicRmw(?*Fiber, @as(*?*Fiber, @ptrCast(&cond.state)), .Xchg, null, .acquire) orelse return;
1309 if (@atomicRmw(?*Fiber, cond_state, .Xchg, null, .acquire)) |fiber| {1350 waiting_fiber.resultPointer(ConditionImpl).event = .{ .wake = wake };
1310 el.yield(fiber, .reschedule);1351 el.yield(waiting_fiber, .reschedule);
1311 }
1312}1352}
13131353
1314fn errno(signed: i32) std.os.linux.E {1354fn errno(signed: i32) std.os.linux.E {
lib/std/Thread/Pool.zig+5-2
...@@ -666,7 +666,7 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I...@@ -666,7 +666,7 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I
666 }666 }
667}667}
668668
669fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {669fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.Wake) void {
670 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));670 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
671 _ = pool;671 _ = pool;
672 comptime assert(@TypeOf(cond.state) == u64);672 comptime assert(@TypeOf(cond.state) == u64);
...@@ -690,7 +690,10 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {...@@ -690,7 +690,10 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition) void {
690 return;690 return;
691 }691 }
692692
693 const to_wake = 1;693 const to_wake = switch (wake) {
694 .one => 1,
695 .all => wakeable,
696 };
694697
695 // Reserve the amount of waiters to wake by incrementing the signals count.698 // Reserve the amount of waiters to wake by incrementing the signals count.
696 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.699 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.