library: the harness keeps the subscribers, and an id belongs to whoever opened it
Three services had each written the same thing and got it three different ways: input polled the process list to notice a dead subscriber, and only when someone else subscribed; the power service never noticed at all; the device manager noticed drivers but not subscribers. The harness owns the table now, driven by the events a protocol declares — it registers on the reserved verb, frames each event once, posts to everyone interested without waiting on any of them, and reclaims a slot when the kernel says its owner died. Interest masks moved to the envelope, so a subscriber that wants only mice asks the same way everywhere. Two consequences the plan had not foreseen. The device manager now hears a supervised child's death twice, once as its supervisor and once as a subscriber, so restart backoff counted every crash twice and gave up after half as many; it retires the id before counting. And the kernel's published exit table had eight slots for what is now six subscriptions in a plain boot, so it holds sixteen. The other half is a hole the design named early and left standing: a backend handed out a small integer and then honoured it from anyone. A process that guessed a file's node id read another client's file; a display layer had no owner at all, so any client could reconfigure or destroy any layer; a USB device token was never checked against the client that opened it. Each is now bound to the task that opened it, and a wrong owner gets exactly what an unknown id gets — the refusal must not become the oracle the identical answers elsewhere were designed to remove. Closing a file changed with it: it used to succeed unconditionally, which would have told a caller which ids existed. Suite 111/111, with a new case in which one process holds a file and a layer, hands both ids to a second process, and finds them untouched after that process has tried everything with them.
This commit is contained in:
@@ -87,11 +87,14 @@ pub fn build(b: *std.Build) void {
|
||||
});
|
||||
// The harness binds the service's contract name at startup, which is a
|
||||
// conversation with the registry — hence channel (and time, for the patience
|
||||
// a provider that beat init to the mount needs).
|
||||
// a provider that beat init to the mount needs). It also owns the subscriber
|
||||
// table and the fan-out, which are expressed in the envelope's vocabulary
|
||||
// (the reserved subscribe verb, the push floor) — hence envelope.
|
||||
_ = b.addModule("service", .{
|
||||
.root_source_file = b.path("service.zig"),
|
||||
.imports = &.{
|
||||
.{ .name = "channel", .module = channel },
|
||||
.{ .name = "envelope", .module = protocol.module("envelope") },
|
||||
.{ .name = "ipc", .module = ipc },
|
||||
.{ .name = "process", .module = process },
|
||||
},
|
||||
|
||||
@@ -12,6 +12,12 @@
|
||||
//! therefore safe, and keeping is explicit; the opposite arrangement quietly
|
||||
//! spends a handle-table slot per request.
|
||||
//!
|
||||
//! The harness also owns the **subscriber side** of a protocol that declares
|
||||
//! `.events` — see `Subscribers`. The table, the reserved subscribe/unsubscribe
|
||||
//! verbs, the fan-out, and the dead-subscriber sweep live here rather than in
|
||||
//! each provider, so every event stream in the system has identical semantics
|
||||
//! (docs/os-development/protocol-namespace.md, "Wiring").
|
||||
//!
|
||||
//! The liveness probe: a **zero-length request is the universal ping**, answered
|
||||
//! with a zero-length reply by the harness itself. No protocol's requests start
|
||||
//! at length zero, so the encoding cannot collide, and there is nothing for a
|
||||
@@ -19,9 +25,23 @@
|
||||
//! is the diagnosis (see docs/ipc.md).
|
||||
|
||||
const channel = @import("channel");
|
||||
const envelope = @import("envelope");
|
||||
const ipc = @import("ipc");
|
||||
const process = @import("process");
|
||||
|
||||
/// The harness's handle on a provider's subscriber table, type-erased because
|
||||
/// `run` is not generic over the protocol while `Subscribers` is. A service names
|
||||
/// its table once, as `Callbacks.subscribers`, and the loop does the rest: it
|
||||
/// subscribes to published process exits at startup and drops a dead task's
|
||||
/// subscriptions before the service's own notification callback ever sees the
|
||||
/// badge.
|
||||
pub const SubscriberHooks = struct {
|
||||
/// Ask the kernel for published exit events on this service's endpoint.
|
||||
watch: *const fn (endpoint: ipc.Handle) void,
|
||||
/// Drop everything task `dead` had subscribed.
|
||||
forget: *const fn (dead: u32) void,
|
||||
};
|
||||
|
||||
pub const Callbacks = struct {
|
||||
/// Called once with the service's endpoint before the loop starts — the
|
||||
/// place to subscribe to exit events, bind IRQs, or announce readiness.
|
||||
@@ -60,8 +80,216 @@ pub const Callbacks = struct {
|
||||
/// `init` runs, so the service is reachable the moment it serves. A refusal
|
||||
/// (not granted, or a live provider already holds the name) aborts startup.
|
||||
service: ?[]const u8 = null,
|
||||
/// This provider's subscriber table — `Subscribers(Protocol, Context).hooks`
|
||||
/// — for a protocol that declares `.events`. Naming it here is what buys the
|
||||
/// exit-notification sweep: the loop subscribes to published deaths at
|
||||
/// startup and releases a dead subscriber's slot (and the endpoint capability
|
||||
/// in it) when one lands.
|
||||
subscribers: ?SubscriberHooks = null,
|
||||
};
|
||||
|
||||
/// How many subscribers one provider fans out to. Bounded like every table in
|
||||
/// this system; a subscribe past the end is refused with `-ENOSPC` rather than
|
||||
/// silently forgetting an earlier one.
|
||||
pub const subscriber_capacity = 8;
|
||||
|
||||
/// The interest mask that means "every event of this protocol" — what a
|
||||
/// subscriber which named no class gets, and what a provider passes when the
|
||||
/// event it is publishing belongs to no class.
|
||||
pub const every_event: u32 = 0;
|
||||
|
||||
/// The subscriber side of a protocol, for a provider whose contract declares
|
||||
/// `.events` (docs/os-development/protocol-namespace.md: *the harness owns the
|
||||
/// machinery — the subscriber table, the dead-subscriber sweep, and the fan-out
|
||||
/// loop*). Three services hand-rolled this, with three different ideas of when a
|
||||
/// dead subscriber goes away — a poll of the process list on subscribe, a drop on
|
||||
/// a failed send, and nothing at all. This is the one idiom.
|
||||
///
|
||||
/// ```zig
|
||||
/// const Subscriptions = service.Subscribers(power_protocol.Protocol, void);
|
||||
/// ...
|
||||
/// fn onMessage(message: []const u8, reply: []u8, sender: u32, arrived: *ipc.Arrival) usize {
|
||||
/// return Subscriptions.dispatch({}, handlers, message, sender, arrived, reply);
|
||||
/// }
|
||||
/// pub fn main() void {
|
||||
/// service.run(power_protocol.message_maximum, .{
|
||||
/// .service = "power",
|
||||
/// .on_message = onMessage,
|
||||
/// .subscribers = Subscriptions.hooks,
|
||||
/// });
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// What the provider still writes is its own events — `publish(.power_button, 0,
|
||||
/// .{})`. Everything else happens here: registering the caller's endpoint on the
|
||||
/// reserved `subscribe` verb, taking that capability out of the turn, dropping it
|
||||
/// on `unsubscribe` or on the subscriber's death, and framing one packet for the
|
||||
/// whole fan-out.
|
||||
///
|
||||
/// The table is per instantiation (a container-level `var` inside the generic
|
||||
/// type), so a process providing two contracts gets two tables and neither can
|
||||
/// see the other's subscribers.
|
||||
pub fn Subscribers(comptime Protocol: type, comptime Context: type) type {
|
||||
return struct {
|
||||
/// The generated dispatch this provider answers with.
|
||||
pub const Provider = Protocol.Provider(Context);
|
||||
pub const Handlers = Provider.Handlers;
|
||||
|
||||
/// One registered subscriber: the endpoint events are pushed to (the
|
||||
/// capability it handed over at subscribe time, which this slot owns),
|
||||
/// the task that handed it over — the kernel-stamped badge, the only
|
||||
/// source identity there is — and which classes of event it asked for.
|
||||
const Slot = struct {
|
||||
used: bool = false,
|
||||
endpoint: ipc.Handle = 0,
|
||||
task: u32 = 0,
|
||||
interest: u32 = every_event,
|
||||
};
|
||||
|
||||
var slots: [subscriber_capacity]Slot = .{Slot{}} ** subscriber_capacity;
|
||||
|
||||
/// Set when a slot has taken the capability the turn carried, and read
|
||||
/// back in `dispatch`, 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 claimed = false;
|
||||
|
||||
/// What `Callbacks.subscribers` is given.
|
||||
pub const hooks: SubscriberHooks = .{ .watch = watchExits, .forget = forget };
|
||||
|
||||
fn watchExits(endpoint: ipc.Handle) void {
|
||||
// Published exits, not a poll of the process list: a service must
|
||||
// never depend on clients cleaning up after themselves, and it must
|
||||
// not have to walk the whole table on every subscribe to find out
|
||||
// either (docs/process-lifecycle.md, "Who learns of a death").
|
||||
_ = process.subscribeExits(endpoint);
|
||||
}
|
||||
|
||||
/// Release everything task `dead` had subscribed. The slot owns the
|
||||
/// endpoint capability, so reclaiming the slot closes it — otherwise a
|
||||
/// process that subscribes and dies costs a handle-table slot that never
|
||||
/// comes back.
|
||||
pub fn forget(dead: u32) void {
|
||||
for (&slots) |*slot| {
|
||||
if (slot.used and slot.task == dead) {
|
||||
_ = ipc.close(slot.endpoint);
|
||||
slot.* = .{};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `task` is a subscriber — the gate for an operation a provider
|
||||
/// honours from its subscribers and nobody else. The power service's
|
||||
/// shutdown is the one: the badge is kernel-stamped, so nothing in a
|
||||
/// packet can claim to be the subscriber that already ran the stop
|
||||
/// sequence.
|
||||
pub fn has(task: u32) bool {
|
||||
for (&slots) |*slot| {
|
||||
if (slot.used and slot.task == task) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// Answer one received packet, with the reserved `subscribe` and
|
||||
/// `unsubscribe` verbs already wired — a provider that leaves those two
|
||||
/// handlers null (every provider should) gets the harness's. The turn's
|
||||
/// capability is peeked, never taken, unless a slot actually kept it.
|
||||
pub fn dispatch(
|
||||
context: Context,
|
||||
handlers: Handlers,
|
||||
packet: []const u8,
|
||||
sender: u32,
|
||||
arrived: *ipc.Arrival,
|
||||
reply: []u8,
|
||||
) usize {
|
||||
var wired = handlers;
|
||||
if (wired.subscribe == null) wired.subscribe = onSubscribe;
|
||||
if (wired.unsubscribe == null) wired.unsubscribe = onUnsubscribe;
|
||||
claimed = false;
|
||||
const written = Provider.dispatch(context, wired, packet, sender, arrived.peek(), reply);
|
||||
if (claimed) _ = arrived.take();
|
||||
return written;
|
||||
}
|
||||
|
||||
/// Push one event to every subscriber.
|
||||
pub fn publish(
|
||||
comptime event: Protocol.Event,
|
||||
target: u64,
|
||||
payload: Protocol.PayloadOf(event),
|
||||
) void {
|
||||
publishClass(event, target, payload, every_event);
|
||||
}
|
||||
|
||||
/// Push one event to the subscribers whose interest mask includes
|
||||
/// `class` (a subscriber that named no class takes everything). The
|
||||
/// packet is framed **once**, outside the loop, so every subscriber of a
|
||||
/// class receives identical bytes; and delivery is `ipc.send`, which
|
||||
/// never blocks, so one slow or dead subscriber can never stall the rest
|
||||
/// — the whole reason broadcast is a provider pattern and not a kernel
|
||||
/// primitive.
|
||||
pub fn publishClass(
|
||||
comptime event: Protocol.Event,
|
||||
target: u64,
|
||||
payload: Protocol.PayloadOf(event),
|
||||
class: u32,
|
||||
) void {
|
||||
var packet: [envelope.post_maximum]u8 = undefined;
|
||||
const framed = Protocol.encodeEvent(event, target, payload, &packet) orelse return;
|
||||
for (&slots) |*slot| {
|
||||
if (!slot.used) continue;
|
||||
if (!wants(slot.*, class)) continue;
|
||||
// The sweep is what normally reclaims a dead subscriber, promptly
|
||||
// and with its capability closed. This is the backstop for a
|
||||
// notification that never arrived: an endpoint's notify ring is
|
||||
// bounded, so a burst of deaths can drop one, and a send to an
|
||||
// endpoint whose owner is gone fails rather than blocking.
|
||||
if (!ipc.send(slot.endpoint, framed)) {
|
||||
_ = ipc.close(slot.endpoint);
|
||||
slot.* = .{};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn wants(slot: Slot, class: u32) bool {
|
||||
if (class == every_event) return true; // the event belongs to no class
|
||||
if (slot.interest == every_event) return true; // the subscriber named none
|
||||
return slot.interest & class != 0;
|
||||
}
|
||||
|
||||
/// The reserved `subscribe` verb: register the caller's endpoint (the
|
||||
/// call's capability) for the classes its tail names. A refusal simply
|
||||
/// returns and the turn closes what arrived — the harness's ownership
|
||||
/// rule (`ipc.Arrival`), which is why a subscribe storm against a full
|
||||
/// table cannot spend the handle table.
|
||||
fn onSubscribe(_: Context, invocation: envelope.Invocation(void), _: envelope.Answer(void)) isize {
|
||||
const endpoint = invocation.capability orelse return -envelope.EPROTO; // no endpoint passed
|
||||
const interest = envelope.decodeSubscribe(invocation.tail).interest;
|
||||
for (&slots) |*slot| {
|
||||
if (slot.used) continue;
|
||||
// Appended, not replaced: one task may hold several subscriptions
|
||||
// on different endpoints (a client taking keyboard and mouse as
|
||||
// two streams), and each is its own conversation.
|
||||
slot.* = .{ .used = true, .endpoint = endpoint, .task = invocation.sender, .interest = interest };
|
||||
claimed = true; // the table holds it until that task dies
|
||||
return 0;
|
||||
}
|
||||
return -envelope.ENOSPC; // table full
|
||||
}
|
||||
|
||||
/// The reserved `unsubscribe` verb: every subscription the calling task
|
||||
/// holds here goes, which is exactly what its death would do. It names no
|
||||
/// endpoint because the badge already names the only subscriber a caller
|
||||
/// can speak for — its own.
|
||||
fn onUnsubscribe(_: Context, invocation: envelope.Invocation(void), _: envelope.Answer(void)) isize {
|
||||
if (!has(invocation.sender)) return -envelope.ENOENT;
|
||||
forget(invocation.sender);
|
||||
return 0;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/// Run the service: create the endpoint, bind it under the service's contract
|
||||
/// name (if it has one), bind signals to it, call `init`, then serve until
|
||||
/// `terminate` arrives — at which point the loop returns and main's return is
|
||||
@@ -74,6 +302,9 @@ pub fn run(comptime maximum_message: usize, callbacks: Callbacks) void {
|
||||
if (!channel.bindPatiently(name, endpoint)) return;
|
||||
}
|
||||
_ = process.bindSignals(endpoint);
|
||||
// Before `init`, so a subscriber that arrives the instant the name is bound
|
||||
// is already covered by the sweep that will release it.
|
||||
if (callbacks.subscribers) |subscribers| subscribers.watch(endpoint);
|
||||
if (callbacks.init) |initialise| {
|
||||
if (!initialise(endpoint)) return;
|
||||
}
|
||||
@@ -105,6 +336,13 @@ pub fn run(comptime maximum_message: usize, callbacks: Callbacks) void {
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// A death sweeps the subscriber table first, then still reaches the
|
||||
// service: a provider often has its own per-client state to release
|
||||
// (open file handles, device tokens, layers) and the same badge is
|
||||
// the notice for both.
|
||||
if (got.isChildExit()) {
|
||||
if (callbacks.subscribers) |subscribers| subscribers.forget(got.childProcessId());
|
||||
}
|
||||
if (callbacks.on_notification) |onNotification| onNotification(got.badge);
|
||||
continue;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user