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.

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

/// 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.
fn broadcast(event: input_protocol.InputEvent) void {
    const bytes = std.mem.asBytes(&event);
    const bit = input_protocol.deviceBit(event.device);
    for (&subscribers) |*sub| {
        if (sub.used and sub.device_mask & bit != 0) _ = ipc.send(sub.endpoint, bytes);
    }
}

/// Handle one request. `got` carries the sender badge (a task id); `arrived` carries the
/// capability the request came with, under the same ownership rule the service harness
/// states (`ipc.Arrival`): **it belongs to the turn, and only a handler that means to keep
/// it says `take`.** Everything else here — a short message, a `publish`, a subscribe that
/// finds the table full — simply returns, and the loop closes what arrived. Writes a
/// `Reply` into `out` and returns its length.
fn handle(message: []const u8, got: ipc.Received, out: []u8, arrived: *ipc.Arrival) usize {
    const reply = struct {
        fn write(buffer: []u8, status: i32) usize {
            const header = input_protocol.Reply{ .status = status };
            @memcpy(buffer[0..input_protocol.reply_size], std.mem.asBytes(&header));
            return input_protocol.reply_size;
        }
    };

    if (message.len < input_protocol.request_size) return reply.write(out, -1);
    const request = std.mem.bytesToValue(input_protocol.Request, message[0..input_protocol.request_size]);

    switch (@as(input_protocol.Operation, @enumFromInt(request.operation))) {
        .subscribe => {
            const endpoint = arrived.peek() orelse return reply.write(out, -1); // no endpoint passed
            // A zero mask means "everything" (a subscriber that named no class still wants input).
            const mask = if (request.device_mask == 0) input_protocol.device_all else request.device_mask;
            pruneDeadSubscribers();
            if (!addSubscriber(endpoint, @intCast(got.badge), mask)) return reply.write(out, -1); // table full
            _ = arrived.take(); // claimed: the subscriber table holds it until that task dies
            return reply.write(out, 0);
        },
        .publish => {
            broadcast(request.event);
            return reply.write(out, 0);
        },
    }
}

pub fn main() void {
    const endpoint = ipc.createIpcEndpoint() orelse {
        _ = logging.write("/system/services/input: no endpoint\n");
        return;
    };
    if (!channel.bindPatiently("input", endpoint)) {
        _ = logging.write("/system/services/input: could not bind /protocol/input\n");
        return;
    }
    _ = logging.write("/system/services/input: ready\n");

    var reply_buffer: [input_protocol.reply_size]u8 = undefined;
    var reply_len: usize = 0;
    var receive: [input_protocol.request_size]u8 = undefined;
    while (true) {
        const got = ipc.replyWait(endpoint, reply_buffer[0..reply_len], &receive, null);
        // Whatever capability came with this turn is the turn's, and the turn closes it
        // unless `handle` claims it (`ipc.Arrival`). The kernel installs a sent capability
        // whatever the message's length or kind, so this covers the notification
        // `continue` and every refusal inside `handle` — otherwise about thirty-two
        // capability-carrying calls, which need no authorization at all, exhaust this
        // service's handle table and no further subscribe can ever land.
        var arrived: ipc.Arrival = .{ .handle = got.cap };
        defer arrived.release();

        // Only synchronous client requests (subscribe/publish) arrive here; nothing sends
        // this service asynchronous messages, so a notification wake would be spurious.
        if (got.isNotification()) {
            reply_len = 0;
            continue;
        }
        reply_len = handle(receive[0..got.len], got, &reply_buffer, &arrived);
    }
}