initial commit

src/kern/wait.zig

@@ -0,0 +1,235 @@
+//! Implementation of generic wait mechanisms.
+//! This is mostly based on the work by Arun Kishan on Windows 7.
+//! See more here: https://youtu.be/OAAiOEQhsK0
+const rtl = @import("rtl");
+const std = @import("std");
+const b = @import("base");
+const ke = b.ke;
+
+/// Header for waitable objects.
+/// This must be added to any structured which is considered waitable.
+pub const DispatchHeader = struct {
+    pub const Kind = enum {
+        /// When signaled, `signaled` is kept high and all waiters are woken up.
+        Notification,
+        /// When signaled, `signaled` is decreased until 0, and a single waiter is woken up.
+        Synchronization,
+    };
+
+    /// List of WaitBlocks.
+    waitblocks: rtl.List,
+    /// Kind of object.
+    kind: Kind,
+    /// Lock protecting the object.
+    lock: ke.SpinLock,
+    /// Signaled count.
+    signaled: u32,
+
+    /// Initialize a waitable object.
+    pub fn init(obj: *DispatchHeader, kind: Kind) void {
+        obj.kind = kind;
+        obj.lock = ke.SpinLock.init();
+        obj.signaled = 0;
+        obj.waitblocks.init();
+    }
+
+    fn consume(self: *DispatchHeader) void {
+        switch (self.kind) {
+            .Synchronization => self.signaled -= 1,
+            .Notification => {
+                // Object remains signaled.
+            },
+        }
+    }
+};
+
+pub const WaitBlock = struct {
+    pub const Status = enum(u8) {
+        /// WaitBlock is linked to an object as part of a thread that is waiting.
+        Active,
+        /// The wait associated with this WaitBlock has been satisfied (or timed out).
+        Inactive,
+        /// A signal was delivered to the wait (this could be before it was committed).
+        Signaled,
+    };
+    /// List linkage.
+    link: rtl.List.Entry,
+    /// Object being waited on.
+    object: *DispatchHeader,
+    /// Thread that owns the WaitBlock.
+    thread: *ke.Thread,
+    /// Status of the WaitBlock.
+    status: WaitBlock.Status,
+};
+
+/// Status of a wait operation.
+pub const Status = enum(u8) {
+    /// Wait is currently being processed.
+    InProgress,
+    /// Wait has been committed and is waiting for the object to be signaled.
+    Committed,
+    /// Wait has been satisfied (or timed out).
+    Satisfied,
+};
+
+/// Wait for the provided object to be signaled.
+pub fn wait_one(object: *DispatchHeader, timeout: ?b.Nanoseconds) !usize {
+    return wait_any(object, timeout, null);
+}
+
+/// Wait for any of the provided objects to be signaled.
+/// Returns the index of the object that was signaled, or an error.
+/// If `timeout` is not provided, it will wait indefinitely.
+/// If `waitblocks` is specified, then the wait will use those waitblocks for the operation.
+/// Note that if `timeout` is provided, then one additional waitblock must be allocated.
+pub fn wait_any(objects: []*DispatchHeader, timeout: ?b.Nanoseconds, waitblocks: ?[*]WaitBlock) !usize {
+    const ipl = ke.ipl.raise(.Dispatch);
+    defer ke.ipl.lower(ipl);
+    const curtd = ke.curcpu().current_thread;
+    const obj_count = objects.len;
+    const has_timeout = timeout != null;
+    const blocks = waitblocks orelse blk: {
+        std.debug.assert(obj_count <= curtd.waitblocks.len - @intFromBool(has_timeout));
+        break :blk &curtd.waitblocks;
+    };
+
+    const total_count = obj_count + @intFromBool(has_timeout);
+    const timer_i = obj_count;
+    const timer = &curtd.timer;
+    const timer_block = &curtd.waitblocks[3];
+
+    var satisfier: ?usize = null;
+    curtd.wait_status.store(.InProgress, .acquire);
+
+    for (0..total_count) |i| {
+        const is_timer = has_timeout and i == timer_i;
+        var obj = if (is_timer) &timer.hdr else objects[i];
+        var wb = if (is_timer) timer_block else &blocks[i];
+
+        obj.lock.acquire_no_ipl();
+        defer obj.lock.release_no_ipl();
+
+        if (obj.signaled > 0) {
+            // Object was already signaled. Try consuming it.
+            if (curtd.wait_status.cmpxchgStrong(.InProgress, .Satisfied, .acquire, .monotonic) == null) {
+                obj.consume();
+            }
+            // We have already been satisfied in the meantime, abort.
+            satisfier = i;
+            break;
+        }
+
+        wb.object = obj;
+        wb.thread = curtd;
+        wb.status = .Active;
+        // We are not satisfied yet, so add a waitblock to the object's waitblocks
+        obj.waitblocks.insert_tail(&wb.link);
+    }
+
+    // Wait was already satisfied, back out.
+    if (satisfier != null or (has_timeout and timeout.? == 0)) {
+        std.debug.assert(curtd.wait_status.load(.monotonic) == .Satisfied);
+
+        // Remove any wait block we might've installed
+        for (0..satisfier) |i| {
+            const is_timer = has_timeout and i == timer_i;
+            var obj = if (is_timer) &timer.hdr else objects[i];
+            var wb = if (is_timer) timer_block else &blocks[i];
+
+            obj.lock.acquire_no_ipl();
+            wb.status = .Inactive;
+            wb.link.remove();
+            obj.lock.release_no_ipl();
+        }
+
+        return satisfier orelse error.Timeout;
+    }
+
+    if (has_timeout) {
+        ke.timer.set(timer, timeout.?, null);
+    }
+
+    // Now try committing the wait.
+    // While we're trying to commit the wait, the object locks have been released, and the state could therefore change.
+    // We need to re-check the state of the wait before actually blocking.
+    if (curtd.wait_status.cmpxchgStrong(.InProgress, .Committed, .acquire, .monotonic) == null) {
+        // We're good, now actually block.
+        ke.sched.block();
+    }
+
+    // We're back!
+    // Stop the timer if it was set.
+    if (has_timeout) {
+        ke.timer.cancel(timer);
+    }
+
+    // Find the object that satisfied us.
+    for (0..total_count) |i| {
+        const is_timer = has_timeout and i == timer_i;
+        var obj = if (is_timer) &timer.hdr else objects[i];
+        var wb = if (is_timer) timer_block else &blocks[i];
+
+        obj.lock.acquire_no_ipl();
+        if (wb.status == .Active) {
+            // Waitblock is still active, remove it from the list.
+            wb.link.remove();
+        }
+        if (wb.status == .Signaled or wb.status == .SignaledInProgress) {
+            // This waitblock was signaled, it must be the one that satisfied the wait.
+            std.debug.assert(satisfier == null);
+            satisfier = i;
+        }
+        // Ignore inactive waitblocks
+        obj.lock.release_no_ipl();
+    }
+
+    const final_sat = satisfier orelse return error.Timeout;
+    return if (has_timeout and final_sat == timer_i) error.Timeout else final_sat;
+}
+
+/// Satisfy a wait on an object.
+pub fn satisfy_wait(obj: *DispatchHeader) void {
+    const all = obj.kind == .Notification;
+
+    // Go through (potentially) all wait blocks and try to satisfy them.
+    while (!obj.waitblocks.is_empty() and obj.signaled > 0) {
+        // Get the first waitblock from the list.
+        var wb: *WaitBlock = @fieldParentPtr("link", obj.waitblocks.first());
+        var td = wb.thread;
+
+        // Remove it.
+        wb.link.remove();
+
+        // Three cases may occur here:
+        // 1. The wait was still preparing (status == .InProgress) and we interrupted it.
+        // 2. The wait was already committed (status == .Committed) and we satisfied it.
+        // 3. The wait was already satisfied by another object (status == .Satisfied).
+
+        // 1.
+        if (td.wait_status.cmpxchgStrong(.InProgress, .Satisfied, .acquire, .monotonic) == null) {
+            // We interrupted the wait while it was being prepared.
+            wb.status = .Signaled;
+            obj.consume();
+        }
+
+        // 2.
+        else if (td.wait_status.cmpxchgStrong(.Committed, .Satisfied, .acquire, .monotonic) == null) {
+            // We interrupted the wait while it was committed.
+            // Wake the thread.
+            wb.status = .Signaled;
+            obj.consume();
+            ke.sched.unblock(td);
+        }
+
+        // 3.
+        else if (td.wait_status.load(.acquire) == .Satisfied) {
+            // Someone else satisfied the wait, deactivate the wait block.
+            wb.status = .Inactive;
+            continue;
+        }
+
+        if (!all) {
+            break;
+        }
+    }
+}