ke: add generic IPI mechanism

src/kern/amd64/impl.zig

@@ -113,14 +113,6 @@ pub inline fn restore_interrupts(state: bool) void {
     }
 }
 
-pub inline fn send_resched_ipi(target: u32) void {
-    pl.impl.send_resched_ipi(target);
-}
-
-pub inline fn send_tlb_ipi(target: u32) void {
-    pl.impl.send_tlb_ipi(target);
-}
-
 pub inline fn flush_full_tlb() void {
     const prev_cr3 = amd64.read_cr(3);
     // Reload CR3 to flush the entire TLB.

src/kern/amd64/int.zig

@@ -85,8 +85,9 @@ export fn isr_handler_main(frame: *const amd64.IrqFrame) callconv(.{ .x86_64_sys
         ki.timer.clock();
     }
 
-    if (frame.intno == 241) {
-        ki.shootdown.ipi_handler();
+    // Platform IPI number.
+    if (frame.intno == 240) {
+        ki.ipi.ipi_handler();
     }
 
     if (frame.intno >= 32) {

src/kern/ipi.zig

@@ -0,0 +1,140 @@
+//! Generic cross-core call mechanisms.
+
+const r = @import("root");
+const std = @import("std");
+
+const ke = r.ke;
+const ki = r.ki;
+const pl = r.pl;
+
+const State = enum(u8) {
+    /// Not currently processing IPIs.
+    Idle = 0,
+    /// Has IPIs pending.
+    Pending = 1,
+    /// Is currently processing IPIs.
+    Running = 2,
+};
+
+const PerCpu = struct {
+    /// Function to be called on other CPUs.
+    func: *const fn (u32, ?*anyopaque) void,
+    /// Argument passed to the function.
+    arg: ?*anyopaque,
+    /// Pending work from other CPUs.
+    pending: ke.AtomicCpuMask,
+    /// Pending CPUs.
+    counter: std.atomic.Value(u16),
+    /// IPI state.
+    state: std.atomic.Value(State),
+};
+
+const percpu = ke.CpuLocal(PerCpu, undefined);
+
+/// Call a function on another CPU.
+/// The function will be executed synchronously, i.e this will block until
+/// the function finishes running. The callback is called with the sender cpu
+/// and provided argument as arguments.
+pub fn unicast(cpu: u32, func: *const fn (u32, ?*anyopaque) void, arg: ?*anyopaque) void {
+    const ipl = ke.ipl.raise(.Dispatch);
+    const curcpu = percpu.local();
+    const remote = percpu.remote(cpu);
+
+    curcpu.func = func;
+    curcpu.arg = arg;
+    curcpu.counter.store(1, .monotonic);
+
+    remote.pending.set(ke.cpu.current(), .monotonic);
+
+    const state = remote.state.swap(.Pending, .release);
+
+    if (state == .Idle) {
+        std.log.info("Sent ipi", .{});
+        pl.send_ipi(cpu);
+    }
+
+    while (curcpu.counter.load(.acquire) != 0) {
+        std.atomic.spinLoopHint();
+    }
+
+    ke.ipl.lower(ipl);
+}
+
+pub fn broadcast(func: *const fn (u32, ?*anyopaque) void, arg: ?*anyopaque) void {
+    const ipl = ke.ipl.raise(.High);
+    const curcpu = percpu.local();
+
+    curcpu.func = func;
+    curcpu.arg = arg;
+    curcpu.counter = .init(@truncate(ke.ncpus - 1));
+
+    for (0..ke.ncpus) |i| {
+        if (i == ke.cpu.current()) continue;
+        const remote = percpu.remote(@intCast(i));
+
+        remote.pending.set(ke.cpu.current(), .monotonic);
+
+        const state = remote.state.swap(.Pending, .release);
+
+        if (state == .Idle) {
+            pl.send_ipi(@intCast(i));
+        }
+    }
+
+    while (curcpu.counter.load(.acquire) != 0) {
+        std.atomic.spinLoopHint();
+    }
+
+    ke.ipl.lower(ipl);
+}
+
+pub fn ipi_handler() void {
+    const curcpu = percpu.local();
+
+    while (true) {
+        // Mark the current CPU as processing IPIs, if the CAS fails here,
+        // then it's probably some kind of IPI that was sent to force us to check
+        // software interrupts (like a rescheduling IPI).
+        const val = curcpu.state.cmpxchgStrong(
+            .Pending,
+            .Running,
+            .acquire,
+            .monotonic,
+        );
+
+        if (val != null) {
+            // CAS failed.
+            return;
+        }
+
+        const words = curcpu.pending.storage.len;
+
+        for (0..words) |i| {
+            var word = curcpu.pending.storage[i].swap(0, .monotonic);
+
+            // Go through every CPU in our pending mask.
+            while (word != 0) {
+                const bit = @ctz(word);
+                const cpu = @bitSizeOf(usize) * i + bit;
+
+                // Call its function and decrement the counter.
+                const remote = percpu.remote(@intCast(cpu));
+                remote.func(@intCast(cpu), remote.arg);
+
+                // Ensure all previous memory operations complete.
+                _ = remote.counter.fetchSub(1, .release);
+
+                word &= (word - 1);
+            }
+        }
+
+        // Try to make the CPU state go from running to idle, if this fails
+        // then we still have work to do.
+        _ = curcpu.state.cmpxchgStrong(
+            .Running,
+            .Idle,
+            .monotonic,
+            .monotonic,
+        ) orelse return;
+    }
+}

src/kern/private.zig

@@ -28,17 +28,6 @@ const ImplSchema = struct {
     /// Read a per-cpu pointer on the current CPU.
     pub fn percpu_ptr(variable: anytype) @TypeOf(variable) {}
 
-    /// Send a rescheduling inter-processor interrupt (IPI)
-    /// on the target CPU.
-    pub fn send_resched_ipi(target: u32) void {
-        _ = target;
-    }
-
-    /// Send a TLB shootdown IPI on the target CPU.
-    pub fn send_tlb_ipi(target: u32) void {
-        _ = target;
-    }
-
     /// Flush the entire TLB on the current CPU.
     pub fn flush_full_tlb() void {}
 
@@ -91,3 +80,4 @@ pub const queue = @import("queue.zig");
 pub const shootdown = @import("shootdown.zig");
 pub const tunable = @import("tunable.zig");
 pub const smr = @import("smr.zig");
+pub const ipi = @import("ipi.zig");

src/kern/root.zig

@@ -103,5 +103,10 @@ pub const smr = struct {
     pub const Sequence = p.smr.Sequence;
 };
 
+pub const ipi = struct {
+    pub const unicast = p.ipi.unicast;
+    pub const broadcast = p.ipi.broadcast;
+};
+
 /// Number of CPUs on the system
 pub var ncpus: usize = 0;

src/kern/sched.zig

@@ -95,6 +95,7 @@ const config = @import("config");
 const r = @import("root");
 const ke = r.ke;
 const ki = ke.private;
+const pl = r.pl;
 
 const runqueues_n = 64;
 const interactivity_threshold = 30;
@@ -624,7 +625,7 @@ pub fn update_priority_locked(td: *ke.Thread, new_prio: u8) void {
         ki.ipl.set_softint_pending(c, .Dispatch);
 
         if (c != ke.cpu.current()) {
-            ki.impl.send_resched_ipi(c);
+            pl.send_ipi(c);
         }
     }
 
@@ -1101,7 +1102,7 @@ fn enqueue_on_cpu(c: u32, td: *ke.Thread) void {
 
             if (c != ke.cpu.current()) {
                 // Send an IPI.
-                ki.impl.send_resched_ipi(c);
+                pl.send_ipi(c);
             }
 
             // queues_lock dropped

src/kern/shootdown.zig

@@ -7,6 +7,7 @@ const r = @import("root");
 const ke = r.ke;
 const ki = ke.private;
 const mm = r.mm;
+const pl = r.pl;
 
 /// Structure representing a single shootdown request.
 pub const ShootdownState = struct {
@@ -36,6 +37,9 @@ const PerCpu = struct {
     valid_states: [config.ncpus]std.atomic.Value(u64),
     /// CPUs that have sent this CPU a shootdown.
     senders: ke.AtomicCpuMask,
+
+    sync_addr: r.VAddr,
+    sync_count: usize,
 };
 
 const slot_free: u16 = std.math.maxInt(u16);
@@ -44,11 +48,6 @@ const percpu = ke.CpuLocal(PerCpu, undefined);
 
 pub var shootdowns: rtl.HandoffList = undefined;
 
-var sync_counter: std.atomic.Value(usize) = .init(0);
-var sync_addr: r.VAddr = 0;
-var sync_count: usize = 0;
-var shootdown_lock: ke.SpinLock = .init();
-
 fn pcpu_init() linksection(r.init) void {
     const local = percpu.local();
 
@@ -89,30 +88,17 @@ fn flush_range(va: r.VAddr, npages: usize) void {
 
 /// Synchronous shootdown path.
 fn do_synchronous_shootdown(state: ShootdownState) void {
-    shootdown_lock.acquire_no_ipl();
+    const cpu = percpu.local();
 
-    sync_addr = state.base;
-    sync_count = state.npages;
-    sync_counter.store(ke.ncpus - 1, .release);
+    cpu.sync_addr = state.base;
+    cpu.sync_count = state.npages;
 
-    const cur = ke.cpu.current();
-
-    for (0..ke.ncpus) |i| {
-        if (i == cur) continue;
-        ki.impl.send_tlb_ipi(@truncate(i));
-    }
-
-    while (sync_counter.load(.monotonic) != 0) {
-        std.atomic.spinLoopHint();
-    }
-
-    shootdown_lock.release_no_ipl();
+    ke.ipi.broadcast(flush_tlb, null);
 }
 
-/// Called by the TLB interrupt handler.
-pub fn ipi_handler() void {
-    flush_range(sync_addr, sync_count);
-    _ = sync_counter.fetchSub(1, .release);
+fn flush_tlb(sender: u32, _: ?*anyopaque) void {
+    const cpu = percpu.remote(sender);
+    flush_range(cpu.sync_addr, cpu.sync_count);
 }
 
 /// Called in a quiescent state to process the pending shootdowns.

src/platform/pc/impl.zig

@@ -9,7 +9,7 @@ pub const smp = @import("smp.zig");
 
 const ki = r.ke.private;
 
-pub const name = "PC";
+pub const name: []const u8 = "PC";
 
 const com1 = 0x3F8;
 
@@ -113,10 +113,6 @@ pub fn arm_timer(deadline: r.Nanoseconds) void {
     apic.arm_timer(deadline);
 }
 
-pub fn send_resched_ipi(target_cpu: u32) void {
+pub fn send_ipi(target_cpu: u32) void {
     apic.send_ipi(smp.cpu_id_to_apic_id[target_cpu], 240, 0);
 }
-
-pub fn send_tlb_ipi(target_cpu: u32) void {
-    apic.send_ipi(smp.cpu_id_to_apic_id[target_cpu], 241, 0);
-}

src/platform/root.zig

@@ -3,6 +3,7 @@
 
 const config = @import("config");
 const r = @import("root");
+const rtl = @import("rtl");
 
 pub const impl = switch (config.arch) {
     .amd64 => @import("pc/impl.zig"),
@@ -11,6 +12,28 @@ pub const impl = switch (config.arch) {
 
 pub const acpi = @import("acpi/root.zig");
 
+const ImplSchema = struct {
+    pub const name: []const u8 = undefined;
+    pub fn early_init() void {}
+    pub fn late_init() void {}
+    pub fn debug_write(c: u8) void {
+        _ = c;
+    }
+    pub fn debug_read() u8 {
+        unreachable;
+    }
+    pub fn send_ipi(cpu: u32) void {
+        _ = cpu;
+    }
+    pub fn arm_timer(ns: r.Nanoseconds) void {
+        _ = ns;
+    }
+};
+
+comptime {
+    rtl.assert_interface(impl, ImplSchema);
+}
+
 /// Early platform-specific initialization, called before anything else.
 pub inline fn early_init() void {
     return impl.early_init();
@@ -37,5 +60,10 @@ pub inline fn arm_timer(ns: r.Nanoseconds) void {
     return impl.arm_timer(ns);
 }
 
+/// Send an inter-processor interrupt (IPI) to a target CPU.
+pub inline fn send_ipi(cpu: u32) void {
+    return impl.send_ipi(cpu);
+}
+
 /// Pretty name for the platform.
 pub const name = impl.name;