authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-30 19:56:30-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log5508b4c8876074b736ed436f0d20d5cec86f1f85
treee2bddde9b476d377b8035b643571e83927de276d
parentb01244d225eb35eac9e06f0677e6b3fc212b4b26

implement Mutex, Condition, and Queue

3 files changed, 497 insertions(+), 23 deletions(-)

lib/std/Io.zig+311-8
...@@ -6,6 +6,7 @@ const windows = std.os.windows;...@@ -6,6 +6,7 @@ const windows = std.os.windows;
6const posix = std.posix;6const posix = std.posix;
7const math = std.math;7const math = std.math;
8const assert = std.debug.assert;8const assert = std.debug.assert;
9const fs = std.fs;
9const Allocator = std.mem.Allocator;10const Allocator = std.mem.Allocator;
10const Alignment = std.mem.Alignment;11const Alignment = std.mem.Alignment;
1112
...@@ -614,6 +615,12 @@ pub const VTable = struct {...@@ -614,6 +615,12 @@ pub const VTable = struct {
614 /// Thread-safe.615 /// Thread-safe.
615 cancelRequested: *const fn (?*anyopaque) bool,616 cancelRequested: *const fn (?*anyopaque) bool,
616617
618 mutexLock: *const fn (?*anyopaque, mutex: *Mutex) void,
619 mutexUnlock: *const fn (?*anyopaque, mutex: *Mutex) void,
620
621 conditionWait: *const fn (?*anyopaque, cond: *Condition, mutex: *Mutex, timeout_ns: ?u64) Condition.WaitError!void,
622 conditionWake: *const fn (?*anyopaque, cond: *Condition, notify: Condition.Notify) void,
623
617 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,624 createFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.CreateFlags) FileOpenError!fs.File,
618 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,625 openFile: *const fn (?*anyopaque, dir: fs.Dir, sub_path: []const u8, flags: fs.File.OpenFlags) FileOpenError!fs.File,
619 closeFile: *const fn (?*anyopaque, fs.File) void,626 closeFile: *const fn (?*anyopaque, fs.File) void,
...@@ -627,11 +634,11 @@ pub const VTable = struct {...@@ -627,11 +634,11 @@ pub const VTable = struct {
627pub const OpenFlags = fs.File.OpenFlags;634pub const OpenFlags = fs.File.OpenFlags;
628pub const CreateFlags = fs.File.CreateFlags;635pub const CreateFlags = fs.File.CreateFlags;
629636
630pub const FileOpenError = fs.File.OpenError || error{AsyncCancel};637pub const FileOpenError = fs.File.OpenError || error{Canceled};
631pub const FileReadError = fs.File.ReadError || error{AsyncCancel};638pub const FileReadError = fs.File.ReadError || error{Canceled};
632pub const FilePReadError = fs.File.PReadError || error{AsyncCancel};639pub const FilePReadError = fs.File.PReadError || error{Canceled};
633pub const FileWriteError = fs.File.WriteError || error{AsyncCancel};640pub const FileWriteError = fs.File.WriteError || error{Canceled};
634pub const FilePWriteError = fs.File.PWriteError || error{AsyncCancel};641pub const FilePWriteError = fs.File.PWriteError || error{Canceled};
635642
636pub const Timestamp = enum(i96) {643pub const Timestamp = enum(i96) {
637 _,644 _,
...@@ -648,8 +655,8 @@ pub const Deadline = union(enum) {...@@ -648,8 +655,8 @@ pub const Deadline = union(enum) {
648 nanoseconds: i96,655 nanoseconds: i96,
649 timestamp: Timestamp,656 timestamp: Timestamp,
650};657};
651pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{AsyncCancel};658pub const ClockGetTimeError = std.posix.ClockGetTimeError || error{Canceled};
652pub const SleepError = error{ UnsupportedClock, Unexpected, AsyncCancel };659pub const SleepError = error{ UnsupportedClock, Unexpected, Canceled };
653660
654pub const AnyFuture = opaque {};661pub const AnyFuture = opaque {};
655662
...@@ -678,6 +685,302 @@ pub fn Future(Result: type) type {...@@ -678,6 +685,302 @@ pub fn Future(Result: type) type {
678 };685 };
679}686}
680687
688pub const Mutex = struct {
689 state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unlocked),
690
691 pub const unlocked: u32 = 0b00;
692 pub const locked: u32 = 0b01;
693 pub const contended: u32 = 0b11; // must contain the `locked` bit for x86 optimization below
694
695 pub fn tryLock(m: *Mutex) bool {
696 // On x86, use `lock bts` instead of `lock cmpxchg` as:
697 // - they both seem to mark the cache-line as modified regardless: https://stackoverflow.com/a/63350048
698 // - `lock bts` is smaller instruction-wise which makes it better for inlining
699 if (builtin.target.cpu.arch.isX86()) {
700 const locked_bit = @ctz(locked);
701 return m.state.bitSet(locked_bit, .acquire) == 0;
702 }
703
704 // Acquire barrier ensures grabbing the lock happens before the critical section
705 // and that the previous lock holder's critical section happens before we grab the lock.
706 return m.state.cmpxchgWeak(unlocked, locked, .acquire, .monotonic) == null;
707 }
708
709 /// Avoids the vtable for uncontended locks.
710 pub fn lock(m: *Mutex, io: Io) void {
711 if (!m.tryLock()) {
712 @branchHint(.unlikely);
713 io.vtable.mutexLock(io.userdata, m);
714 }
715 }
716
717 pub fn unlock(m: *Mutex, io: Io) void {
718 io.vtable.mutexUnlock(io.userdata, m);
719 }
720};
721
722pub const Condition = struct {
723 state: u64 = 0,
724
725 pub const WaitError = error{
726 Timeout,
727 Canceled,
728 };
729
730 /// How many waiters to wake up.
731 pub const Notify = enum {
732 one,
733 all,
734 };
735
736 pub fn wait(cond: *Condition, io: Io, mutex: *Mutex) void {
737 io.vtable.conditionWait(io.userdata, cond, mutex, null) catch |err| switch (err) {
738 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
739 error.Canceled => return, // handled as spurious wakeup
740 };
741 }
742
743 pub fn timedWait(cond: *Condition, io: Io, mutex: *Mutex, timeout_ns: u64) WaitError!void {
744 return io.vtable.conditionWait(io.userdata, cond, mutex, timeout_ns);
745 }
746
747 pub fn signal(cond: *Condition, io: Io) void {
748 io.vtable.conditionWake(io.userdata, cond, .one);
749 }
750
751 pub fn broadcast(cond: *Condition, io: Io) void {
752 io.vtable.conditionWake(io.userdata, cond, .all);
753 }
754};
755
756pub const TypeErasedQueue = struct {
757 mutex: Mutex,
758
759 /// Ring buffer. This data is logically *after* queued getters.
760 buffer: []u8,
761 put_index: usize,
762 get_index: usize,
763
764 putters: std.DoublyLinkedList(PutNode),
765 getters: std.DoublyLinkedList(GetNode),
766
767 const PutNode = struct {
768 remaining: []const u8,
769 condition: Condition,
770 };
771
772 const GetNode = struct {
773 remaining: []u8,
774 condition: Condition,
775 };
776
777 pub fn init(buffer: []u8) TypeErasedQueue {
778 return .{
779 .mutex = .{},
780 .buffer = buffer,
781 .put_index = 0,
782 .get_index = 0,
783 .putters = .{},
784 .getters = .{},
785 };
786 }
787
788 pub fn put(q: *TypeErasedQueue, io: Io, elements: []const u8, min: usize) usize {
789 assert(elements.len >= min);
790
791 q.mutex.lock(io);
792 defer q.mutex.unlock(io);
793
794 // Getters have first priority on the data, and only when the getters
795 // queue is empty do we start populating the buffer.
796
797 var remaining = elements;
798 while (true) {
799 const getter = q.getters.popFirst() orelse break;
800 const copy_len = @min(getter.data.remaining.len, remaining.len);
801 @memcpy(getter.data.remaining[0..copy_len], remaining[0..copy_len]);
802 remaining = remaining[copy_len..];
803 getter.data.remaining = getter.data.remaining[copy_len..];
804 if (getter.data.remaining.len == 0) {
805 getter.data.condition.signal(io);
806 continue;
807 }
808 q.getters.prepend(getter);
809 assert(remaining.len == 0);
810 return elements.len;
811 }
812
813 while (true) {
814 {
815 const available = q.buffer[q.put_index..];
816 const copy_len = @min(available.len, remaining.len);
817 @memcpy(available[0..copy_len], remaining[0..copy_len]);
818 remaining = remaining[copy_len..];
819 q.put_index += copy_len;
820 if (remaining.len == 0) return elements.len;
821 }
822 {
823 const available = q.buffer[0..q.get_index];
824 const copy_len = @min(available.len, remaining.len);
825 @memcpy(available[0..copy_len], remaining[0..copy_len]);
826 remaining = remaining[copy_len..];
827 q.put_index = copy_len;
828 if (remaining.len == 0) return elements.len;
829 }
830
831 const total_filled = elements.len - remaining.len;
832 if (total_filled >= min) return total_filled;
833
834 var node: std.DoublyLinkedList(PutNode).Node = .{
835 .data = .{ .remaining = remaining, .condition = .{} },
836 };
837 q.putters.append(&node);
838 node.data.condition.wait(io, &q.mutex);
839 remaining = node.data.remaining;
840 }
841 }
842
843 pub fn get(q: *@This(), io: Io, buffer: []u8, min: usize) usize {
844 assert(buffer.len >= min);
845
846 q.mutex.lock(io);
847 defer q.mutex.unlock(io);
848
849 // The ring buffer gets first priority, then data should come from any
850 // queued putters, then finally the ring buffer should be filled with
851 // data from putters so they can be resumed.
852
853 var remaining = buffer;
854 while (true) {
855 if (q.get_index <= q.put_index) {
856 const available = q.buffer[q.get_index..q.put_index];
857 const copy_len = @min(available.len, remaining.len);
858 @memcpy(remaining[0..copy_len], available[0..copy_len]);
859 q.get_index += copy_len;
860 remaining = remaining[copy_len..];
861 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
862 } else {
863 {
864 const available = q.buffer[q.get_index..];
865 const copy_len = @min(available.len, remaining.len);
866 @memcpy(remaining[0..copy_len], available[0..copy_len]);
867 q.get_index += copy_len;
868 remaining = remaining[copy_len..];
869 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
870 }
871 {
872 const available = q.buffer[0..q.put_index];
873 const copy_len = @min(available.len, remaining.len);
874 @memcpy(remaining[0..copy_len], available[0..copy_len]);
875 q.get_index = copy_len;
876 remaining = remaining[copy_len..];
877 if (remaining.len == 0) return fillRingBufferFromPutters(q, io, buffer.len);
878 }
879 }
880 // Copy directly from putters into buffer.
881 while (remaining.len > 0) {
882 const putter = q.putters.popFirst() orelse break;
883 const copy_len = @min(putter.data.remaining.len, remaining.len);
884 @memcpy(remaining[0..copy_len], putter.data.remaining[0..copy_len]);
885 putter.data.remaining = putter.data.remaining[copy_len..];
886 remaining = remaining[copy_len..];
887 if (putter.data.remaining.len == 0) {
888 putter.data.condition.signal(io);
889 } else {
890 assert(remaining.len == 0);
891 q.putters.prepend(putter);
892 return fillRingBufferFromPutters(q, io, buffer.len);
893 }
894 }
895 // Both ring buffer and putters queue is empty.
896 const total_filled = buffer.len - remaining.len;
897 if (total_filled >= min) return total_filled;
898
899 var node: std.DoublyLinkedList(GetNode).Node = .{
900 .data = .{ .remaining = remaining, .condition = .{} },
901 };
902 q.getters.append(&node);
903 node.data.condition.wait(io, &q.mutex);
904 remaining = node.data.remaining;
905 }
906 }
907
908 /// Called when there is nonzero space available in the ring buffer and
909 /// potentially putters waiting. The mutex is already held and the task is
910 /// to copy putter data to the ring buffer and signal any putters whose
911 /// buffers been fully copied.
912 fn fillRingBufferFromPutters(q: *TypeErasedQueue, io: Io, len: usize) usize {
913 while (true) {
914 const putter = q.putters.popFirst() orelse return len;
915 const available = q.buffer[q.put_index..];
916 const copy_len = @min(available.len, putter.data.remaining.len);
917 @memcpy(available[0..copy_len], putter.data.remaining[0..copy_len]);
918 putter.data.remaining = putter.data.remaining[copy_len..];
919 q.put_index += copy_len;
920 if (putter.data.remaining.len == 0) {
921 putter.data.condition.signal(io);
922 continue;
923 }
924 const second_available = q.buffer[0..q.get_index];
925 const second_copy_len = @min(second_available.len, putter.data.remaining.len);
926 @memcpy(second_available[0..second_copy_len], putter.data.remaining[0..second_copy_len]);
927 putter.data.remaining = putter.data.remaining[copy_len..];
928 q.put_index = copy_len;
929 if (putter.data.remaining.len == 0) {
930 putter.data.condition.signal(io);
931 continue;
932 }
933 q.putters.prepend(putter);
934 return len;
935 }
936 }
937};
938
939/// Many producer, many consumer, thread-safe, runtime configurable buffer size.
940/// When buffer is empty, consumers suspend and are resumed by producers.
941/// When buffer is full, producers suspend and are resumed by consumers.
942pub fn Queue(Elem: type) type {
943 return struct {
944 type_erased: TypeErasedQueue,
945
946 pub fn init(buffer: []Elem) @This() {
947 return .{ .type_erased = .init(@ptrCast(buffer)) };
948 }
949
950 /// Appends elements to the end of the queue. The function returns when
951 /// at least `min` elements have been added to the buffer or sent
952 /// directly to a consumer.
953 ///
954 /// Returns how many elements have been added to the queue.
955 ///
956 /// Asserts that `elements.len >= min`.
957 pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) usize {
958 return @divExact(q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
959 }
960
961 /// Receives elements from the beginning of the queue. The function
962 /// returns when at least `min` elements have been populated inside
963 /// `buffer`.
964 ///
965 /// Returns how many elements of `buffer` have been populated.
966 ///
967 /// Asserts that `buffer.len >= min`.
968 pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) usize {
969 return @divExact(q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
970 }
971
972 pub fn putOne(q: *@This(), io: Io, item: Elem) void {
973 assert(q.put(io, &.{item}, 1) == 1);
974 }
975
976 pub fn getOne(q: *@This(), io: Io) Elem {
977 var buf: [1]Elem = undefined;
978 assert(q.get(io, &buf, 1) == 1);
979 return buf[0];
980 }
981 };
982}
983
681/// Calls `function` with `args`, such that the return value of the function is984/// Calls `function` with `args`, such that the return value of the function is
682/// not guaranteed to be available until `await` is called.985/// not guaranteed to be available until `await` is called.
683pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {986pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(function)).@"fn".return_type.?) {
...@@ -685,7 +988,7 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(...@@ -685,7 +988,7 @@ pub fn async(io: Io, function: anytype, args: anytype) Future(@typeInfo(@TypeOf(
685 const Args = @TypeOf(args);988 const Args = @TypeOf(args);
686 const TypeErased = struct {989 const TypeErased = struct {
687 fn start(context: *const anyopaque, result: *anyopaque) void {990 fn start(context: *const anyopaque, result: *anyopaque) void {
688 const args_casted: *const Args = @alignCast(@ptrCast(context));991 const args_casted: *const Args = @ptrCast(@alignCast(context));
689 const result_casted: *Result = @ptrCast(@alignCast(result));992 const result_casted: *Result = @ptrCast(@alignCast(result));
690 result_casted.* = @call(.auto, function, args_casted.*);993 result_casted.* = @call(.auto, function, args_casted.*);
691 }994 }
lib/std/Io/EventLoop.zig+7-7
...@@ -102,7 +102,7 @@ const Fiber = struct {...@@ -102,7 +102,7 @@ const Fiber = struct {
102 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));102 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
103 }103 }
104104
105 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{AsyncCancel}!void {105 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{Canceled}!void {
106 if (@cmpxchgStrong(106 if (@cmpxchgStrong(
107 ?*Thread,107 ?*Thread,
108 &fiber.cancel_thread,108 &fiber.cancel_thread,
...@@ -112,7 +112,7 @@ const Fiber = struct {...@@ -112,7 +112,7 @@ const Fiber = struct {
112 .acquire,112 .acquire,
113 )) |cancel_thread| {113 )) |cancel_thread| {
114 assert(cancel_thread == Thread.canceling);114 assert(cancel_thread == Thread.canceling);
115 return error.AsyncCancel;115 return error.Canceled;
116 }116 }
117 }117 }
118118
...@@ -746,7 +746,7 @@ pub fn createFile(...@@ -746,7 +746,7 @@ pub fn createFile(
746 switch (errno(completion.result)) {746 switch (errno(completion.result)) {
747 .SUCCESS => return .{ .handle = completion.result },747 .SUCCESS => return .{ .handle = completion.result },
748 .INTR => unreachable,748 .INTR => unreachable,
749 .CANCELED => return error.AsyncCancel,749 .CANCELED => return error.Canceled,
750750
751 .FAULT => unreachable,751 .FAULT => unreachable,
752 .INVAL => return error.BadPathName,752 .INVAL => return error.BadPathName,
...@@ -854,7 +854,7 @@ pub fn openFile(...@@ -854,7 +854,7 @@ pub fn openFile(
854 switch (errno(completion.result)) {854 switch (errno(completion.result)) {
855 .SUCCESS => return .{ .handle = completion.result },855 .SUCCESS => return .{ .handle = completion.result },
856 .INTR => unreachable,856 .INTR => unreachable,
857 .CANCELED => return error.AsyncCancel,857 .CANCELED => return error.Canceled,
858858
859 .FAULT => unreachable,859 .FAULT => unreachable,
860 .INVAL => return error.BadPathName,860 .INVAL => return error.BadPathName,
...@@ -950,7 +950,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std...@@ -950,7 +950,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
950 switch (errno(completion.result)) {950 switch (errno(completion.result)) {
951 .SUCCESS => return @as(u32, @bitCast(completion.result)),951 .SUCCESS => return @as(u32, @bitCast(completion.result)),
952 .INTR => unreachable,952 .INTR => unreachable,
953 .CANCELED => return error.AsyncCancel,953 .CANCELED => return error.Canceled,
954954
955 .INVAL => unreachable,955 .INVAL => unreachable,
956 .FAULT => unreachable,956 .FAULT => unreachable,
...@@ -1002,7 +1002,7 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs...@@ -1002,7 +1002,7 @@ pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offs
1002 switch (errno(completion.result)) {1002 switch (errno(completion.result)) {
1003 .SUCCESS => return @as(u32, @bitCast(completion.result)),1003 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1004 .INTR => unreachable,1004 .INTR => unreachable,
1005 .CANCELED => return error.AsyncCancel,1005 .CANCELED => return error.Canceled,
10061006
1007 .INVAL => return error.InvalidArgument,1007 .INVAL => return error.InvalidArgument,
1008 .FAULT => unreachable,1008 .FAULT => unreachable,
...@@ -1080,7 +1080,7 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D...@@ -1080,7 +1080,7 @@ pub fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.D
1080 switch (errno(completion.result)) {1080 switch (errno(completion.result)) {
1081 .SUCCESS, .TIME => return,1081 .SUCCESS, .TIME => return,
1082 .INTR => unreachable,1082 .INTR => unreachable,
1083 .CANCELED => return error.AsyncCancel,1083 .CANCELED => return error.Canceled,
10841084
1085 else => |err| return std.posix.unexpectedErrno(err),1085 else => |err| return std.posix.unexpectedErrno(err),
1086 }1086 }
lib/std/Thread/Pool.zig+179-8
...@@ -332,9 +332,12 @@ pub fn io(pool: *Pool) Io {...@@ -332,9 +332,12 @@ pub fn io(pool: *Pool) Io {
332 .vtable = &.{332 .vtable = &.{
333 .@"async" = @"async",333 .@"async" = @"async",
334 .@"await" = @"await",334 .@"await" = @"await",
335
336 .cancel = cancel,335 .cancel = cancel,
337 .cancelRequested = cancelRequested,336 .cancelRequested = cancelRequested,
337 .mutexLock = mutexLock,
338 .mutexUnlock = mutexUnlock,
339 .conditionWait = conditionWait,
340 .conditionWake = conditionWake,
338341
339 .createFile = createFile,342 .createFile = createFile,
340 .openFile = openFile,343 .openFile = openFile,
...@@ -517,11 +520,179 @@ fn cancelRequested(userdata: ?*anyopaque) bool {...@@ -517,11 +520,179 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
517 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;520 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == AsyncClosure.canceling_tid;
518}521}
519522
520fn checkCancel(pool: *Pool) error{AsyncCancel}!void {523fn checkCancel(pool: *Pool) error{Canceled}!void {
521 if (cancelRequested(pool)) return error.AsyncCancel;524 if (cancelRequested(pool)) return error.Canceled;
525}
526
527fn mutexLock(userdata: ?*anyopaque, m: *Io.Mutex) void {
528 @branchHint(.cold);
529 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
530 _ = pool;
531
532 // Avoid doing an atomic swap below if we already know the state is contended.
533 // An atomic swap unconditionally stores which marks the cache-line as modified unnecessarily.
534 if (m.state.load(.monotonic) == Io.Mutex.contended) {
535 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
536 }
537
538 // Try to acquire the lock while also telling the existing lock holder that there are threads waiting.
539 //
540 // Once we sleep on the Futex, we must acquire the mutex using `contended` rather than `locked`.
541 // If not, threads sleeping on the Futex wouldn't see the state change in unlock and potentially deadlock.
542 // The downside is that the last mutex unlocker will see `contended` and do an unnecessary Futex wake
543 // but this is better than having to wake all waiting threads on mutex unlock.
544 //
545 // Acquire barrier ensures grabbing the lock happens before the critical section
546 // and that the previous lock holder's critical section happens before we grab the lock.
547 while (m.state.swap(Io.Mutex.contended, .acquire) != Io.Mutex.unlocked) {
548 std.Thread.Futex.wait(&m.state, Io.Mutex.contended);
549 }
550}
551
552fn mutexUnlock(userdata: ?*anyopaque, m: *Io.Mutex) void {
553 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
554 _ = pool;
555 // Needs to also wake up a waiting thread if any.
556 //
557 // A waiting thread will acquire with `contended` instead of `locked`
558 // which ensures that it wakes up another thread on the next unlock().
559 //
560 // Release barrier ensures the critical section happens before we let go of the lock
561 // and that our critical section happens before the next lock holder grabs the lock.
562 const state = m.state.swap(Io.Mutex.unlocked, .release);
563 assert(state != Io.Mutex.unlocked);
564
565 if (state == Io.Mutex.contended) {
566 std.Thread.Futex.wake(&m.state, 1);
567 }
568}
569
570fn mutexLockInternal(pool: *std.Thread.Pool, m: *Io.Mutex) void {
571 if (!m.tryLock()) {
572 @branchHint(.unlikely);
573 mutexLock(pool, m);
574 }
575}
576
577fn conditionWait(
578 userdata: ?*anyopaque,
579 cond: *Io.Condition,
580 mutex: *Io.Mutex,
581 timeout: ?u64,
582) Io.Condition.WaitError!void {
583 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
584 comptime assert(@TypeOf(cond.state) == u64);
585 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
586 const cond_state = &ints[0];
587 const cond_epoch = &ints[1];
588 const one_waiter = 1;
589 const waiter_mask = 0xffff;
590 const one_signal = 1 << 16;
591 const signal_mask = 0xffff << 16;
592 // Observe the epoch, then check the state again to see if we should wake up.
593 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:
594 //
595 // - T1: s = LOAD(&state)
596 // - T2: UPDATE(&s, signal)
597 // - T2: UPDATE(&epoch, 1) + FUTEX_WAKE(&epoch)
598 // - T1: e = LOAD(&epoch) (was reordered after the state load)
599 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)
600 //
601 // Acquire barrier to ensure the epoch load happens before the state load.
602 var epoch = cond_epoch.load(.acquire);
603 var state = cond_state.fetchAdd(one_waiter, .monotonic);
604 assert(state & waiter_mask != waiter_mask);
605 state += one_waiter;
606
607 mutexUnlock(pool, mutex);
608 defer mutexLockInternal(pool, mutex);
609
610 var futex_deadline = std.Thread.Futex.Deadline.init(timeout);
611
612 while (true) {
613 futex_deadline.wait(cond_epoch, epoch) catch |err| switch (err) {
614 // On timeout, we must decrement the waiter we added above.
615 error.Timeout => {
616 while (true) {
617 // If there's a signal when we're timing out, consume it and report being woken up instead.
618 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
619 while (state & signal_mask != 0) {
620 const new_state = state - one_waiter - one_signal;
621 state = cond_state.cmpxchgWeak(state, new_state, .acquire, .monotonic) orelse return;
622 }
623
624 // Remove the waiter we added and officially return timed out.
625 const new_state = state - one_waiter;
626 state = cond_state.cmpxchgWeak(state, new_state, .monotonic, .monotonic) orelse return err;
627 }
628 },
629 };
630
631 epoch = cond_epoch.load(.acquire);
632 state = cond_state.load(.monotonic);
633
634 // Try to wake up by consuming a signal and decremented the waiter we added previously.
635 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
636 while (state & signal_mask != 0) {
637 const new_state = state - one_waiter - one_signal;
638 state = cond_state.cmpxchgWeak(state, new_state, .acquire, .monotonic) orelse return;
639 }
640 }
641}
642
643fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, notify: Io.Condition.Notify) void {
644 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
645 _ = pool;
646 comptime assert(@TypeOf(cond.state) == u64);
647 const ints: *[2]std.atomic.Value(u32) = @ptrCast(&cond.state);
648 const cond_state = &ints[0];
649 const cond_epoch = &ints[1];
650 const one_waiter = 1;
651 const waiter_mask = 0xffff;
652 const one_signal = 1 << 16;
653 const signal_mask = 0xffff << 16;
654 var state = cond_state.load(.monotonic);
655 while (true) {
656 const waiters = (state & waiter_mask) / one_waiter;
657 const signals = (state & signal_mask) / one_signal;
658
659 // Reserves which waiters to wake up by incrementing the signals count.
660 // Therefore, the signals count is always less than or equal to the waiters count.
661 // We don't need to Futex.wake if there's nothing to wake up or if other wake() threads have reserved to wake up the current waiters.
662 const wakeable = waiters - signals;
663 if (wakeable == 0) {
664 return;
665 }
666
667 const to_wake = switch (notify) {
668 .one => 1,
669 .all => wakeable,
670 };
671
672 // Reserve the amount of waiters to wake by incrementing the signals count.
673 // Release barrier ensures code before the wake() happens before the signal it posted and consumed by the wait() threads.
674 const new_state = state + (one_signal * to_wake);
675 state = cond_state.cmpxchgWeak(state, new_state, .release, .monotonic) orelse {
676 // Wake up the waiting threads we reserved above by changing the epoch value.
677 // NOTE: a waiting thread could miss a wake up if *exactly* ((1<<32)-1) wake()s happen between it observing the epoch and sleeping on it.
678 // This is very unlikely due to how many precise amount of Futex.wake() calls that would be between the waiting thread's potential preemption.
679 //
680 // Release barrier ensures the signal being added to the state happens before the epoch is changed.
681 // If not, the waiting thread could potentially deadlock from missing both the state and epoch change:
682 //
683 // - T2: UPDATE(&epoch, 1) (reordered before the state change)
684 // - T1: e = LOAD(&epoch)
685 // - T1: s = LOAD(&state)
686 // - T2: UPDATE(&state, signal) + FUTEX_WAKE(&epoch)
687 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed both epoch change and state change)
688 _ = cond_epoch.fetchAdd(1, .release);
689 std.Thread.Futex.wake(cond_epoch, to_wake);
690 return;
691 };
692 }
522}693}
523694
524pub fn createFile(695fn createFile(
525 userdata: ?*anyopaque,696 userdata: ?*anyopaque,
526 dir: std.fs.Dir,697 dir: std.fs.Dir,
527 sub_path: []const u8,698 sub_path: []const u8,
...@@ -532,7 +703,7 @@ pub fn createFile(...@@ -532,7 +703,7 @@ pub fn createFile(
532 return dir.createFile(sub_path, flags);703 return dir.createFile(sub_path, flags);
533}704}
534705
535pub fn openFile(706fn openFile(
536 userdata: ?*anyopaque,707 userdata: ?*anyopaque,
537 dir: std.fs.Dir,708 dir: std.fs.Dir,
538 sub_path: []const u8,709 sub_path: []const u8,
...@@ -543,13 +714,13 @@ pub fn openFile(...@@ -543,13 +714,13 @@ pub fn openFile(
543 return dir.openFile(sub_path, flags);714 return dir.openFile(sub_path, flags);
544}715}
545716
546pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {717fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
547 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));718 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
548 _ = pool;719 _ = pool;
549 return file.close();720 return file.close();
550}721}
551722
552pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {723fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std.posix.off_t) Io.FilePReadError!usize {
553 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));724 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
554 try pool.checkCancel();725 try pool.checkCancel();
555 return switch (offset) {726 return switch (offset) {
...@@ -558,7 +729,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std...@@ -558,7 +729,7 @@ pub fn pread(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8, offset: std
558 };729 };
559}730}
560731
561pub fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {732fn pwrite(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8, offset: std.posix.off_t) Io.FilePWriteError!usize {
562 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));733 const pool: *std.Thread.Pool = @alignCast(@ptrCast(userdata));
563 try pool.checkCancel();734 try pool.checkCancel();
564 return switch (offset) {735 return switch (offset) {