Add input module: broadcast keyboard events over IPC

Programs can now subscribe to keyboard events (key_down/key_up/key_press)
and drivers can broadcast them, through a new user-space input service.

The delivery model is forced by danos IPC: a synchronous rendezvous holds
one pending reply, so a server cannot park N subscribers blocked in a
"wait for next event" call — delivery must be push. But a synchronous push
has no timeout and the kernel never wakes a sender parked on a dead peer's
endpoint, so one dying subscriber would hang all input. So this lands the
roadmap's planned asynchronous buffered send and builds the service on it:

- ipc_send (syscall 26): non-blocking post to an endpoint's bounded payload
  ring, delivered through reply_wait as a buffered message (notify_message_bit).
  A full ring drops the oldest. It can never hang on a dead/slow peer.
- input-protocol + runtime.input helpers (subscribe/next, connectSource/
  publish) — the first real consumer of M13 capability passing: a subscriber
  hands the service its own endpoint as a capability.
- input service (fan-out via ipc_send, dead-subscriber pruning), a synthetic
  input-source, and input-test; the ps2-bus keyboard driver publishes to it.
  Real IRQ1 scancode decoding (which must live in the bus, the PNP0303 owner)
  is a documented follow-up; the source is synthetic for now.
- build/init wiring, an `input` QEMU case, and docs/input.md.

Full QEMU suite 48/48, including the new input case and every IPC/endpoint
regression (ipc, ipc-call, ipc-cap, vfs, hpet, bus, irqfree).
This commit is contained in:
Daniel Samson
2026-07-11 15:03:24 +01:00
parent 2a583d55a8
commit 65244e3103
19 changed files with 765 additions and 5 deletions
+79 -1
View File
@@ -57,6 +57,29 @@ pub const EPERM: i64 = 9; // not permitted (process_kill by anyone but the super
/// shared kernel↔user ABI (system/abi.zig), because ring 3 has to test the same bit.
pub const notify_badge_bit: u64 = abi.notify_badge_bit;
/// Set (with `notify_badge_bit`) when a `replyWait` wake carries a buffered payload
/// posted by `send` (`ipc_send`), rather than a bare IRQ/exit notification. Shared with
/// ring 3 through the ABI so the receiver can tell "a message arrived" from "the hardware
/// spoke".
pub const notify_message_bit: u64 = abi.notify_message_bit;
/// Largest payload a single `send` (`ipc_send`) may post. Kept small — the payload rides
/// inline in every `Endpoint`, and the async path is for events (a `KeyEvent` is 16
/// bytes), not bulk transfer, which is what `call` and future shared pages are for.
pub const POST_MAXIMUM: usize = 64;
/// Depth of an endpoint's async payload ring. Absorbs a burst while a receiver is briefly
/// busy; a full ring drops the *oldest* message (see `send`).
const post_capacity: usize = 16;
/// One buffered message: a length-prefixed payload plus the sender's task id (delivered
/// in the low bits of the receiver's badge).
const PostSlot = struct {
length: u16 = 0,
sender_id: u64 = 0,
bytes: [POST_MAXIMUM]u8 = undefined,
};
/// End of the user (low) canonical half — user buffers must lie below it.
const user_half_end: u64 = 0x0000_8000_0000_0000;
@@ -74,6 +97,12 @@ pub const Endpoint = struct {
notify_buffer: [8]u64 = undefined,
notify_head: u8 = 0,
notify_tail: u8 = 0,
// Pending buffered messages (payloads posted by `send`), a small FIFO ring. Unlike
// notifications — which are a level and coalesce — these are discrete messages, so a
// full ring drops the oldest rather than merging.
post_buffer: [post_capacity]PostSlot = undefined,
post_head: u16 = 0,
post_tail: u16 = 0,
};
pub fn createIpcEndpoint() ?*Endpoint {
@@ -265,12 +294,23 @@ pub fn replyWait(endpoint: *Endpoint, reply_ptr: u64, reply_len: u64, receive_pt
scheduler.readyLocked(client); // its `call` now returns
}
// (2) Receive the next request (or notification), blocking until one is ready.
// (2) Receive the next request (or notification / buffered message), blocking until
// one is ready. Bare notifications (IRQ/exit) come first — they're latency-sensitive
// and carry no payload — then buffered messages, then synchronous client requests.
while (true) {
if (popNotify(endpoint)) |badge| {
out_badge.* = badge | notify_badge_bit;
return 0; // notification: no payload, no reply owed, no cap
}
if (popPost(endpoint)) |slot| {
const n = @min(@as(usize, slot.length), receive_cap);
// Copy from the kernel-resident ring slot (source aspace 0) into the receiver.
if (!copyAcross(0, @intFromPtr(&slot.bytes), me.aspace, receive_ptr, n)) {
continue; // bad receive buffer: drop this message, keep serving
}
out_badge.* = slot.sender_id | notify_badge_bit | notify_message_bit;
return @intCast(n); // async message: payload delivered, no reply owed, no cap
}
if (dequeueSender(endpoint)) |caller| {
const n = @min(caller.ipc_send_len, receive_cap);
if (!copyAcross(caller.aspace, caller.ipc_send_ptr, me.aspace, receive_ptr, n)) {
@@ -306,6 +346,44 @@ fn popNotify(endpoint: *Endpoint) ?u64 {
return badge;
}
/// Take the oldest buffered message from the post ring, or null if empty. Returns a
/// pointer into the endpoint's own storage — valid until the next `send`/`popPost` under
/// the same lock region, which is all the copy-out in `replyWait` needs.
fn popPost(endpoint: *Endpoint) ?*const PostSlot {
if (endpoint.post_head == endpoint.post_tail) return null;
const slot = &endpoint.post_buffer[endpoint.post_head % post_capacity];
endpoint.post_head +%= 1;
return slot;
}
/// Client-free side of async IPC (`ipc_send`): copy `[source_va, len)` from address space
/// `source_as` into `endpoint`'s post ring and wake a waiting receiver — **without
/// blocking the sender** and with no reply owed. `sender_id` rides along, delivered in the
/// low bits of the receiver's badge. Returns 0, or a negative errno (`-E2BIG` if the
/// payload exceeds `POST_MAXIMUM`, `-EFAULT` if the source buffer is unmapped / out of the
/// user half). A full ring drops the *oldest* message (advancing `post_head`), because a
/// buffered message is discrete, not a level: keeping the newest keeps input responsive.
/// Precondition: the big kernel lock is held.
pub fn sendLocked(endpoint: *Endpoint, source_as: u64, source_va: u64, len: u64, sender_id: u64) i64 {
if (len > POST_MAXIMUM) return -E2BIG;
// Drop the oldest if the ring is full, so this newest message always lands.
if (endpoint.post_tail -% endpoint.post_head >= post_capacity) endpoint.post_head +%= 1;
const slot = &endpoint.post_buffer[endpoint.post_tail % post_capacity];
if (!copyFromUser(source_as, source_va, slot.bytes[0..@intCast(len)])) return -EFAULT;
slot.length = @intCast(len);
slot.sender_id = sender_id;
endpoint.post_tail +%= 1;
scheduler.wakeLocked(&endpoint.receive_wait_queue);
return 0;
}
/// `sendLocked` wrapped in its own critical section, for the `ipc_send` syscall path.
pub fn send(endpoint: *Endpoint, source_as: u64, source_va: u64, len: u64, sender_id: u64) i64 {
const flags = sync.enter();
defer sync.leave(flags);
return sendLocked(endpoint, source_as, source_va, len, sender_id);
}
/// Post an asynchronous notification carrying `badge` to `endpoint` and wake a waiting
/// receiver. Precondition: the big kernel lock is held.
///
+13
View File
@@ -183,6 +183,7 @@ fn system_call(state: *architecture.CpuState) void {
.ipc_lookup => systemIpcLookup(state),
.ipc_call => systemIpcCall(state),
.ipc_reply_wait => systemIpcReplyWait(state),
.ipc_send => systemIpcSend(state),
.device_enumerate => systemDeviceEnumerate(state),
.device_claim => systemDeviceClaim(state),
.mmio_map => systemMmioMap(state),
@@ -263,6 +264,18 @@ fn systemIpcReplyWait(state: *architecture.CpuState) void {
architecture.setSystemCallResult3(state, received_cap);
}
/// ipc_send(handle, message_ptr, message_len) -> 0/-errno: post a payload to an
/// endpoint's async queue and wake a receiver, without blocking the caller. The async
/// counterpart of ipc_call — for broadcasts (the input service) where a rendezvous would
/// let one dead subscriber hang the sender. Delivered through ipc_reply_wait as a
/// buffered message (badge carries notify_message_bit and the caller's task id).
fn systemIpcSend(state: *architecture.CpuState) void {
const me = scheduler.current();
const endpoint = ipc.resolveHandle(me, architecture.systemCallArg(state, 0)) orelse return failErr(state, ipc.EBADF);
const r = ipc.send(endpoint, me.aspace, architecture.systemCallArg(state, 1), architecture.systemCallArg(state, 2), me.id);
architecture.setSystemCallResult(state, @bitCast(r));
}
/// device_enumerate(buffer, maximum) -> total: snapshot the device table into the caller's
/// buffer (up to `maximum` entries), returning the total device count.
fn systemDeviceEnumerate(state: *architecture.CpuState) void {
+46
View File
@@ -136,6 +136,8 @@ pub fn run(case: []const u8, boot_information: *const BootInformation) void {
initialRamdiskTest(boot_information);
} else if (eql(case, "vfs")) {
vfsTest(boot_information);
} else if (eql(case, "input")) {
inputTest(boot_information);
} else if (eql(case, "hpet")) {
hpetTest(boot_information);
} else if (eql(case, "iopass")) {
@@ -1585,6 +1587,50 @@ fn vfsTest(boot_information: *const BootInformation) void {
result();
}
/// The full input path: spawn the input service, a synthetic keyboard source, and a
/// subscriber from the initial_ramdisk. The source publishes key events; the service
/// broadcasts them (with the asynchronous ipc_send); the subscriber receives them and —
/// only once it has — heartbeats "input-test: ok". Seeing that marker proves an event
/// travelled source -> service -> subscriber over IPC, exercising the async buffered-send
/// primitive and capability-passing subscription. The source and service stay silent
/// after startup so the subscriber's line is the one left in the shared evidence buffer.
fn inputTest(boot_information: *const BootInformation) void {
log("DANOS-TEST-BEGIN: input\n", .{});
if (boot_information.initial_ramdisk_len == 0) {
check("bootloader handed over an initial_ramdisk", false);
result();
return;
}
const image = @as([*]const u8, @ptrFromInt(boot_handoff.physicalToVirtual(boot_information.initial_ramdisk_base)))[0..boot_information.initial_ramdisk_len];
const rd = initial_ramdisk.Reader.init(image) orelse {
check("initial_ramdisk image is valid", false);
result();
return;
};
process.write_count = 0;
process.write_from_user = false;
_ = spawnNamed(rd, "input"); // the fan-out service
_ = spawnNamed(rd, "input-source"); // a synthetic keyboard publishing events
_ = spawnNamed(rd, "input-test"); // the subscriber whose "ok" line is the marker
// Wait for the subscriber's success heartbeat (it beats once per received event).
const prefix = "input-test: ok";
scheduler.setPriority(1);
const deadline = architecture.millis() + 12000;
while (architecture.millis() < deadline) {
if (process.write_len >= prefix.len and eql(process.write_buffer[0..prefix.len], prefix) and process.write_count >= 2) break;
scheduler.yield();
}
scheduler.setPriority(4);
const ok = process.write_len >= prefix.len and eql(process.write_buffer[0..prefix.len], prefix);
check("a subscriber received a broadcast key event over IPC (source -> service -> subscriber)", ok);
check("events kept flowing (service + async send stay up)", process.write_count >= 2);
check("client syscalls came from user mode (CPL 3)", process.write_from_user);
result();
}
/// Process arguments, end to end: spawn args-echo bare (its argv[0] is the
/// initial-ramdisk name). Instance 1 sees argc == 1 and respawns itself through
/// `system_spawn` with the extra arguments "alpha beta-42" — the syscall argument