danos/system/services/input/input.zig

//! system/services/input — the user-space input service. Shipped in the initial_ramdisk,
//! spawned as a ring-3 process, and bound at `/protocol/input`. It
//! is the fan-out point between **sources** (keyboard, mouse, and joystick/gamepad drivers)
//! and **subscribers** (any program that wants input): a source `publish`es an
//! `InputEvent`, and the service pushes it to every subscriber whose interest mask includes
//! that event's device class (keyboard / mouse / joystick).
//!
//! The delivery discipline is the whole design (see docs/input.md). The kernel's IPC is a
//! synchronous rendezvous: a server holds one pending reply, so it cannot park N
//! subscribers blocked in a "wait for next event" call. Broadcasting therefore has to be
//! *push* — the service delivering to subscribers. But a synchronous push (`ipc_call`)
//! would let one dead or wedged subscriber hang the whole broadcast, since the kernel
//! never wakes a sender parked on a dead peer's endpoint. So delivery uses the
//! asynchronous `ipc.send`: it posts the event to each subscriber's endpoint queue and
//! returns at once, and can never block on a subscriber. That primitive exists for exactly
//! this ([ipc.md](../../../docs/ipc.md), "asynchronous / buffered send").
//!
//! A subscriber registers by handing the service its own endpoint as a capability (M13
//! capability passing — this service is its first real consumer). The service keeps that
//! handle and `ipc.send`s each event to it. That is the envelope's reserved `subscribe`
//! verb, which this protocol adopts rather than defining its own.
//!
//! P4a moved this service onto the shared harness (library/kernel/service.zig). It was the
//! last hand-rolled receive loop in the tree, and the one service that answered neither the
//! universal ping nor a `terminate` signal — so a shutdown had to kill it. The subscriber
//! table, the fan-out, and the prune-on-subscribe below are unchanged; lifting *those* into
//! the harness is a later milestone, and doing it here would have hidden this one.

const envelope = @import("envelope");
const ipc = @import("ipc");
const process = @import("process");
const service = @import("service");
const logging = @import("logging");
const input_protocol = @import("input-protocol");

/// The generated input dispatch. One fan-out point per process, so the handler
/// context is empty and the subscriber table stays in this file's globals.
const Serve = input_protocol.Protocol.Provider(void);

const Invocation = envelope.Invocation;
const Answer = envelope.Answer;

/// One registered subscriber: the endpoint we push events to (a capability it handed us at
/// subscribe time) and the task id that owns it (the subscribe call's badge), so a slot
/// left behind by a subscriber that exited can be reclaimed.
const Subscriber = struct {
    used: bool = false,
    endpoint: ipc.Handle = 0,
    task_id: u32 = 0,
    /// Which device classes this subscriber wants (an OR of input_protocol.device_*). An event
    /// is delivered only if its device's bit is set here.
    device_mask: u32 = 0,
};

var subscribers = [_]Subscriber{.{}} ** 8;

/// Drop any subscriber whose owning process is no longer alive, so its slot (and the
/// endpoint reference it holds) can be reused. Cheap and only run on subscribe — the async
/// `send` to a dead subscriber's orphaned endpoint is harmless (it just fills a queue no
/// one drains), so this is housekeeping, not correctness.
fn pruneDeadSubscribers() void {
    var table: [32]process.ProcessDescriptor = undefined;
    const total = process.processes(&table);
    const count = @min(total, table.len);
    for (&subscribers) |*sub| {
        if (!sub.used) continue;
        var alive = false;
        for (table[0..count]) |descriptor| {
            if (descriptor.id == sub.task_id) {
                alive = true;
                break;
            }
        }
        // The slot owns the endpoint capability it was handed, so reclaiming the
        // slot closes it — otherwise a process that subscribes and dies costs a
        // handle-table slot that never comes back.
        if (!alive) {
            _ = ipc.close(sub.endpoint);
            sub.* = .{};
        }
    }
}

/// Register `endpoint` (owned by task `task_id`) to receive the device classes in
/// `device_mask`. Returns false if the subscriber table is full.
fn addSubscriber(endpoint: ipc.Handle, task_id: u32, device_mask: u32) bool {
    for (&subscribers) |*sub| {
        if (!sub.used) {
            sub.* = .{ .used = true, .endpoint = endpoint, .task_id = task_id, .device_mask = device_mask };
            return true;
        }
    }
    return false;
}

/// Push `event` to every subscriber whose interest mask includes its device class.
/// `ipc.send` never blocks, so a slow or dead subscriber cannot stall delivery to others.
///
/// The class is the packet's operation, so the fan-out picks the event by device and the
/// packet is framed once, outside the loop — every subscriber of a class gets identical
/// bytes, which is what "one fan-out point per event domain" means on the wire.
fn broadcast(event: input_protocol.InputEvent) void {
    const class = input_protocol.eventOfDevice(event.device) orelse return; // no class wants it
    var packet: [envelope.post_maximum]u8 = undefined;
    const framed = switch (class) {
        .keyboard => input_protocol.Protocol.encodeEvent(.keyboard, 0, event.asKeyboard() orelse return, &packet),
        .mouse => input_protocol.Protocol.encodeEvent(.mouse, 0, event.asMouse() orelse return, &packet),
        .joystick => input_protocol.Protocol.encodeEvent(.joystick, 0, event.asJoystick() orelse return, &packet),
    } orelse return;

    const bit = input_protocol.deviceBit(event.device);
    for (&subscribers) |*sub| {
        if (sub.used and sub.device_mask & bit != 0) _ = ipc.send(sub.endpoint, framed);
    }
}

/// Set by `onSubscribe` when the subscriber table has taken ownership of the capability the
/// call carried, and read by `onMessage`, which is where the turn's `Arrival` lives. The
/// generated dispatch hands a handler the raw handle rather than the `Arrival` — deliberately,
/// since a handler has no business closing the turn's property — so the *claim* has to travel
/// back out this way. One turn, one handler, one thread: there is nothing here to race.
var capability_claimed = false;

/// The reserved `subscribe` verb: register the caller's endpoint (the call's capability) for
/// the classes in the packet's tail. Refusals simply return, and the turn closes what arrived
/// — the ownership rule the harness states (`ipc.Arrival`), unchanged by the move onto it.
fn onSubscribe(_: void, invocation: Invocation(void), _: Answer(void)) isize {
    const endpoint = invocation.capability orelse return -envelope.EPROTO; // no endpoint passed
    // A zero mask means "everything" (a subscriber that named no class still wants input).
    const requested = input_protocol.decodeSubscribe(invocation.tail).device_mask;
    const mask = if (requested == 0) input_protocol.device_all else requested;
    pruneDeadSubscribers();
    if (!addSubscriber(endpoint, invocation.sender, mask)) return -envelope.ENOSPC; // table full
    capability_claimed = true; // the subscriber table holds it until that task dies
    return 0;
}

fn onPublish(_: void, invocation: Invocation(input_protocol.InputEvent), _: Answer(void)) isize {
    broadcast(invocation.request);
    return 0;
}

const handlers = Serve.Handlers{ .publish = onPublish, .subscribe = onSubscribe };

fn onMessage(message: []const u8, out: []u8, sender: u32, arrived: *ipc.Arrival) usize {
    capability_claimed = false;
    const written = Serve.dispatch({}, handlers, message, sender, arrived.peek(), out);
    if (capability_claimed) _ = arrived.take();
    return written;
}

fn initialise(_: ipc.Handle) bool {
    // The harness has already bound `/protocol/input` by the time this runs, so
    // "ready" still means what it always meant: the name is claimed and the loop
    // is about to serve it.
    _ = logging.write("/system/services/input: ready\n");
    return true;
}

pub fn main() void {
    service.run(input_protocol.message_maximum, .{
        .service = "input",
        .init = initialise,
        .on_message = onMessage,
    });
}