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-10-02 16:30:59-07:00
log6ba65ca972cf38a57266cf624932ae61d4cbba2d
treeab8273617df91c2261eed28fdb214f464cfa5f2f
parentb174777437ffc537f58e45cd54cc4f9bcf5305d9

Io.Condition: implement full API

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

lib/std/Io.zig+18-7
...@@ -630,7 +630,7 @@ pub const VTable = struct {...@@ -630,7 +630,7 @@ pub const VTable = struct {
630 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,630 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
631631
632 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,632 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex) Cancelable!void,
633 conditionWake: *const fn (?*anyopaque, cond: *Condition) void,633 conditionWake: *const fn (?*anyopaque, cond: *Condition, wake: Condition.Wake) void,
634634
635 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,635 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
636 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,636 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
...@@ -809,9 +809,20 @@ pub const Condition = struct {...@@ -809,9 +809,20 @@ pub const Condition = struct {
809 return io.vtable.conditionWait(io.userdata, cond, mutex);809 return io.vtable.conditionWait(io.userdata, cond, mutex);
810 }810 }
811811
812 pub fn wake(cond: *Condition, io: Io) void {812 pub fn signal(cond: *Condition, io: Io) void {
813 io.vtable.conditionWake(io.userdata, cond);813 io.vtable.conditionWake(io.userdata, cond, .one);
814 }814 }
815
816 pub fn broadcast(cond: *Condition, io: Io) void {
817 io.vtable.conditionWake(io.userdata, cond, .all);
818 }
819
820 pub const Wake = enum {
821 /// wake up only one thread
822 one,
823 /// wake up all thread
824 all,
825 };
815};826};
816827
817pub const TypeErasedQueue = struct {828pub const TypeErasedQueue = struct {
...@@ -863,7 +874,7 @@ pub const TypeErasedQueue = struct {...@@ -863,7 +874,7 @@ pub const TypeErasedQueue = struct {
863 remaining = remaining[copy_len..];874 remaining = remaining[copy_len..];
864 getter.data.remaining = getter.data.remaining[copy_len..];875 getter.data.remaining = getter.data.remaining[copy_len..];
865 if (getter.data.remaining.len == 0) {876 if (getter.data.remaining.len == 0) {
866 getter.data.condition.wake(io);877 getter.data.condition.signal(io);
867 continue;878 continue;
868 }879 }
869 q.getters.prepend(getter);880 q.getters.prepend(getter);
...@@ -946,7 +957,7 @@ pub const TypeErasedQueue = struct {...@@ -946,7 +957,7 @@ pub const TypeErasedQueue = struct {
946 putter.data.remaining = putter.data.remaining[copy_len..];957 putter.data.remaining = putter.data.remaining[copy_len..];
947 remaining = remaining[copy_len..];958 remaining = remaining[copy_len..];
948 if (putter.data.remaining.len == 0) {959 if (putter.data.remaining.len == 0) {
949 putter.data.condition.wake(io);960 putter.data.condition.signal(io);
950 } else {961 } else {
951 assert(remaining.len == 0);962 assert(remaining.len == 0);
952 q.putters.prepend(putter);963 q.putters.prepend(putter);
...@@ -979,7 +990,7 @@ pub const TypeErasedQueue = struct {...@@ -979,7 +990,7 @@ pub const TypeErasedQueue = struct {
979 putter.data.remaining = putter.data.remaining[copy_len..];990 putter.data.remaining = putter.data.remaining[copy_len..];
980 q.put_index += copy_len;991 q.put_index += copy_len;
981 if (putter.data.remaining.len == 0) {992 if (putter.data.remaining.len == 0) {
982 putter.data.condition.wake(io);993 putter.data.condition.signal(io);
983 continue;994 continue;
984 }995 }
985 const second_available = q.buffer[0..q.get_index];996 const second_available = q.buffer[0..q.get_index];
...@@ -988,7 +999,7 @@ pub const TypeErasedQueue = struct {...@@ -988,7 +999,7 @@ pub const TypeErasedQueue = struct {
988 putter.data.remaining = putter.data.remaining[copy_len..];999 putter.data.remaining = putter.data.remaining[copy_len..];
989 q.put_index = copy_len;1000 q.put_index = copy_len;
990 if (putter.data.remaining.len == 0) {1001 if (putter.data.remaining.len == 0) {
991 putter.data.condition.wake(io);1002 putter.data.condition.signal(io);
992 continue;1003 continue;
993 }1004 }
994 q.putters.prepend(putter);1005 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.