export mod events;

//! Shared event rings and terminal records retained until acknowledgement.

use super::abi;

use super::limits;

use super::sync;

/// Shared ring entry count, a power of two.

export constant CAPACITY: u32 = 256;

/// Occupancy limit for ordinary events, leaving 64 entries for lifecycle events.

export constant ORDINARY: u32 = 192;

/// Device interrupt notification.

export constant INTERRUPT: u16 = 1;

/// Timer notification.

export constant TIMEOUT: u16 = 2;

/// Terminal exception report.

export constant FAULT: u16 = 3;

/// Normal domain termination report.

export constant CHILD_EXIT: u16 = 4;

/// Peer wake notification.

export constant WAKEUP: u16 = 5;

/// One asynchronous notification in the shared ABI.

export record Event: Copy {

    /// Notification kind.

    kind: u16,

    /// Zero for every published event.

    reserved: u16,

    /// Interrupt number, status, fault code, or notification token.

    code: u32,

    /// Kind-specific payload, including generation-bearing terminal domain IDs.

    value: u64,

}

/// Shared queue memory. Consumer operations require one serialized reader.

export record Ring: Copy {

    /// Entries published before the producer advances tail.

    data: [Event; CAPACITY],

    /// Consumer progress, accessed through ordered 32-bit operations.

    head: u32,

    /// Producer progress, accessed through ordered 32-bit operations.

    tail: u32,

    /// CAPACITY minus one.

    mask: u32,

}

/// Trusted producer metadata, separate from consumer-writable ring memory.

export record Queue: Copy {

    /// Live receiver generation, or zero for a closed queue.

    generation: u32,

    /// Last validated consumer counter.

    head: u32,

    /// Authoritative producer counter.

    tail: u32,

    /// Terminal domain per occupied slot; generation zero denotes an ordinary event.

    subjects: [abi::Ref; CAPACITY],

}

/// Lifetime state of a terminal-event record.

export union State: Copy {

    /// Available for a new domain's terminal reservation.

    Free,

    /// Capacity owned by an unfinished lifecycle transaction.

    Reserved,

    /// Event retained until its receiver has ring capacity.

    Pending,

    /// Event published and awaiting consumer acknowledgement.

    Queued,

    /// Consumer acknowledged the event, or the receiver exited.

    Acknowledged,

}

/// A domain's terminal notification and its retained identity.

export record Terminal: Copy {

    /// Domain whose identifier must remain reserved.

    subject: abi::Ref,

    /// Receiver selected when termination commits.

    receiver: abi::Ref,

    /// Complete retained event payload.

    event: Event,

    /// Terminal publication and acknowledgement state.

    state: State,

}

/// Fixed event metadata. Callers serialize all kernel-side mutations.

export record Store: Copy {

    /// Producer metadata indexed by receiver domain.

    queues: [Queue; limits::DOMAINS],

    /// Terminal storage indexed by the terminating domain.

    terminals: [Terminal; limits::DOMAINS],

}

/// Terminal capacity that must be committed or cancelled for the same domain.

export union Reservation: Once {

    /// Domain generation owning one terminal record.

    Held(abi::Ref),

}

/// Compute forward distance in the shared 32-bit counter space.

fn distance(end: u32, start: u32) -> u32 {

    return ((end as u64 + 0x100000000 - start as u64) & 0xffffffff) as u32;

}

/// Advance a shared counter with explicit modulo arithmetic.

fn next(value: u32) -> u32 { return ((value as u64 + 1) & 0xffffffff) as u32; }

/// Initialize metadata before any queue or terminal record exists.

export fn initialize(store: &mut Store) {

    for i in 0..limits::DOMAINS {

        set store.queues[i].generation = 0;

        set store.terminals[i].state = State::Free;

    }

}

/// Test the receiver generation without reading beyond the fixed table.

fn live(store: &Store, receiver: abi::Ref) -> bool {

    return receiver.index < limits::DOMAINS and receiver.generation <> 0

        and store.queues[receiver.index].generation == receiver.generation;

}

/// Open a queue over mapped shared memory before publishing its Events handle.

export fn open(store: &mut Store, receiver: abi::Ref, ring: &mut Ring) throws (abi::Error) {

    if receiver.index >= limits::DOMAINS or receiver.generation == 0 { throw abi::Error::InvalidArg; }

    let queue = &mut store.queues[receiver.index];

    if queue.generation <> 0 { throw abi::Error::Busy; }

    set queue.generation = receiver.generation;

    set queue.head = 0;

    set queue.tail = 0;

    set ring.mask = CAPACITY - 1;

    sync::storeRelease32(&mut ring.head, 0);

    sync::storeRelease32(&mut ring.tail, 0);

}

/// Validate consumer progress and acknowledge terminal records before slot reuse.

export fn refresh(store: &mut Store, receiver: abi::Ref, ring: &Ring) throws (abi::Error) {

    if not live(store, receiver) { throw abi::Error::BadHandle; }

    let queue = &mut store.queues[receiver.index];

    let head = sync::loadAcquire32(&ring.head);

    let advance = distance(head, queue.head);

    if advance > distance(queue.tail, queue.head) { throw abi::Error::InvalidArg; }

    for i in 0..advance {

        let subject = queue.subjects[queue.head & (CAPACITY - 1)];

        if subject.generation <> 0 {

            let terminal = &mut store.terminals[subject.index];

            assert terminal.subject == subject

                and terminal.receiver == receiver

                and terminal.state == State::Queued;

            set terminal.state = State::Acknowledged;

        }

        set queue.head = next(queue.head);

    }

}

/// Publish one event after capacity and receiver validation.

fn publish(queue: &mut Queue, ring: &mut Ring, event: Event, subject: abi::Ref) {

    let index = queue.tail & (CAPACITY - 1);

    set ring.data[index] = event;

    set queue.subjects[index] = subject;

    set queue.tail = next(queue.tail);

    sync::storeRelease32(&mut ring.tail, queue.tail);

}

/// Send an ordinary event, returning false when its reserved occupancy limit is reached.

export fn send(store: &mut Store, receiver: abi::Ref, ring: &mut Ring, event: Event) -> bool throws (abi::Error) {

    if event.reserved <> 0 or (event.kind <> INTERRUPT and event.kind <> TIMEOUT and event.kind <> WAKEUP) {

        throw abi::Error::InvalidArg;

    }

    try refresh(store, receiver, ring);

    let queue = &mut store.queues[receiver.index];

    if distance(queue.tail, queue.head) >= ORDINARY { return false; }

    publish(queue, ring, event, abi::Ref { index: 0, generation: 0 });

    return true;

}

/// Read one published event and release its slot. Callers serialize consumers.

export fn pop(ring: &mut Ring) -> ?Event throws (abi::Error) {

    let head = sync::loadAcquire32(&ring.head);

    let tail = sync::loadAcquire32(&ring.tail);

    let count = distance(tail, head);

    if count > CAPACITY or ring.mask <> CAPACITY - 1 { throw abi::Error::InvalidArg; }

    if count == 0 { return nil; }

    let event = ring.data[head & (CAPACITY - 1)];

    sync::storeRelease32(&mut ring.head, next(head));

    return event;

}

/// Reserve terminal capacity before committing domain termination.

export fn reserve(store: &mut Store, subject: abi::Ref) -> Reservation throws (abi::Error) {

    if subject.index >= limits::DOMAINS or subject.generation == 0 { throw abi::Error::InvalidArg; }

    if store.terminals[subject.index].state <> State::Free { throw abi::Error::Busy; }

    set store.terminals[subject.index].subject = subject; set store.terminals[subject.index].state = State::Reserved;

    return Reservation::Held(subject);

}

/// Cancel terminal capacity before domain termination commits.

export fn cancel(store: &mut Store, reservation: Reservation) {

    match reservation {

        case Reservation::Held(subject) => {

            assert store.terminals[subject.index].subject == subject and store.terminals[subject.index].state == State::Reserved;

            set store.terminals[subject.index].state = State::Free;

        },

    }

}

/// Commit one terminal event to its selected receiver. A missing receiver needs no delivery.

export fn finish(store: &mut Store, reservation: Reservation, receiver: abi::Ref, kind: u16, code: u32) {

    assert kind == FAULT or kind == CHILD_EXIT;

    let active = live(store, receiver);

    match reservation {

        case Reservation::Held(subject) => {

            let terminal = &mut store.terminals[subject.index];

            assert terminal.subject == subject and terminal.state == State::Reserved;

            set terminal.receiver = receiver;

            set terminal.event = Event { kind, reserved: 0, code, value: abi::id(subject) };

            if active { set terminal.state = State::Pending; }

            else { set terminal.state = State::Acknowledged; }

        },

    }

}

/// Publish retained terminal events in domain-index order as ring capacity permits.

export fn flush(store: &mut Store, receiver: abi::Ref, ring: &mut Ring) -> u32 throws (abi::Error) {

    try refresh(store, receiver, ring);

    let mut count: u32 = 0;

    for i in 0..limits::DOMAINS {

        if distance(store.queues[receiver.index].tail, store.queues[receiver.index].head) == CAPACITY { break; }

        if store.terminals[i].state <> State::Pending or store.terminals[i].receiver <> receiver { continue; }

        set store.terminals[i].state = State::Queued;

        let event = store.terminals[i].event;

        let subject = store.terminals[i].subject;

        publish(&mut store.queues[receiver.index], ring, event, subject);

        set count += 1;

    }

    return count;

}

/// Report whether a retained terminal identity can be released.

export fn acknowledged(store: &Store, subject: abi::Ref) -> bool throws (abi::Error) {

    if subject.index >= limits::DOMAINS { throw abi::Error::BadHandle; }

    if store.terminals[subject.index].state == State::Free or store.terminals[subject.index].subject <> subject { throw abi::Error::BadHandle; }

    return store.terminals[subject.index].state == State::Acknowledged;

}

/// Release terminal metadata only after its queued identifier is no longer observable.

export fn release(store: &mut Store, subject: abi::Ref) throws (abi::Error) {

    if not try acknowledged(store, subject) { throw abi::Error::Busy; }

    set store.terminals[subject.index].state = State::Free;

}

/// Close a stopped domain's queue and end all terminal references held by that receiver.

export fn close(store: &mut Store, receiver: abi::Ref) throws (abi::Error) {

    if not live(store, receiver) { throw abi::Error::BadHandle; }

    set store.queues[receiver.index].generation = 0;

    for i in 0..limits::DOMAINS {

        if (store.terminals[i].state == State::Pending or store.terminals[i].state == State::Queued) and store.terminals[i].receiver == receiver {

            set store.terminals[i].state = State::Acknowledged;

        }

    }

}

export mod events;

//! Event ring saturation, counter validation, and retained terminal identities.

use std::testing;

use kernel::abi;

use kernel::events;

use kernel::sync;

/// Kernel event metadata workspace.

unsafe static STORE: events::Store = undefined;

/// Consumer-visible ring for the receiver.

unsafe static RING: events::Ring = undefined;

/// Return the receiver's current domain reference.

fn receiver() -> abi::Ref { return abi::Ref { index: 0, generation: 1 }; }

/// Reset metadata and open one queue.

unsafe fn initialize() {

    events::initialize(&mut STORE);

    try! events::open(&mut STORE, receiver(), &mut RING);

}

/// Build an ordinary wakeup with a recognizable payload.

fn wake(value: u64) -> events::Event {

    return events::Event { kind: events::WAKEUP, reserved: 0, code: 17, value };

}

/// Fill ordinary capacity while preserving the reserved lifecycle capacity.

@test unsafe fn saturation() throws (testing::TestError) {

    initialize();

    for i in 0..events::ORDINARY {

        try testing::expect(try! events::send(&mut STORE, receiver(), &mut RING, wake(i as u64)));

    }

    try testing::expect(not try! events::send(&mut STORE, receiver(), &mut RING, wake(999)));

    for i in 0..65 {

        let subject = abi::Ref { index: i + 1, generation: 1 };

        let pending = try! events::reserve(&mut STORE, subject);

        events::finish(&mut STORE, pending, receiver(), events::CHILD_EXIT, i);

    }

    try testing::expect(try! events::flush(&mut STORE, receiver(), &mut RING) == 64);

    let subject = abi::Ref { index: 65, generation: 1 };

    try testing::expect(not try! events::acknowledged(&STORE, subject));

    for i in 0..events::ORDINARY {

        let event = try! events::pop(&mut RING) else panic "ordinary event";

        try testing::expect(event == wake(i as u64));

    }

    try testing::expect(try! events::flush(&mut STORE, receiver(), &mut RING) == 1);

    for i in 0..65 {

        let event = try! events::pop(&mut RING) else panic "terminal event";

        try testing::expect(event.kind == events::CHILD_EXIT and event.code == i);

        try testing::expect(event.value == abi::id(abi::Ref { index: i + 1, generation: 1 }));

    }

    try! events::refresh(&mut STORE, receiver(), &RING);

    try testing::expect(try! events::acknowledged(&STORE, subject));

    try! events::release(&mut STORE, subject);

    let replacement = try! events::reserve(&mut STORE, abi::Ref { index: 65, generation: 2 });

    events::cancel(&mut STORE, replacement);

}

/// Head and tail arithmetic remains valid across the u32 boundary.

@test unsafe fn counterWrap() throws (testing::TestError) {

    initialize();

    set STORE.queues[0].head = 0xfffffffe;

    set STORE.queues[0].tail = 0xfffffffe;

    sync::storeRelease32(&mut RING.head, 0xfffffffe);

    sync::storeRelease32(&mut RING.tail, 0xfffffffe);

    for i in 0..4 { try testing::expect(try! events::send(&mut STORE, receiver(), &mut RING, wake(i as u64))); }

    for i in 0..4 {

        let event = try! events::pop(&mut RING) else panic "wrapped event";

        try testing::expect(event == wake(i as u64));

    }

    try! events::refresh(&mut STORE, receiver(), &RING);

    try testing::expect(STORE.queues[0].head == 2 and STORE.queues[0].tail == 2);

    try testing::expect(try! events::pop(&mut RING) == nil);

}

/// Invalid consumer advances cannot acknowledge or overwrite a pending event.

@test unsafe fn invalidConsumer() throws (testing::TestError) {

    initialize();

    try testing::expect(try! events::send(&mut STORE, receiver(), &mut RING, wake(42)));

    for head in &[2 as u32, 0xffffffff] {

        sync::storeRelease32(&mut RING.head, head);

        let mut failed = false;

        try events::refresh(&mut STORE, receiver(), &RING) catch err {

            try testing::expect(err == abi::Error::InvalidArg); set failed = true;

        };

        try testing::expect(failed and STORE.queues[0].head == 0 and STORE.queues[0].tail == 1);

    }

    sync::storeRelease32(&mut RING.head, 0);

    let event = try! events::pop(&mut RING) else panic "retained event";

    try testing::expect(event == wake(42));

}

/// A receiver's exit ends both queued and unpublished terminal references.

@test unsafe fn receiverExit() throws (testing::TestError) {

    initialize();

    let subject = abi::Ref { index: 1, generation: 7 };

    let pending = try! events::reserve(&mut STORE, subject);

    events::finish(&mut STORE, pending, receiver(), events::FAULT, 23);

    try testing::expect(try! events::flush(&mut STORE, receiver(), &mut RING) == 1);

    let waiting = abi::Ref { index: 2, generation: 3 };

    let second = try! events::reserve(&mut STORE, waiting);

    events::finish(&mut STORE, second, receiver(), events::CHILD_EXIT, 0);

    try! events::close(&mut STORE, receiver());

    try testing::expect(try! events::acknowledged(&STORE, subject));

    try testing::expect(try! events::acknowledged(&STORE, waiting));

    try! events::open(&mut STORE, abi::Ref { index: 0, generation: 2 }, &mut RING);

    try testing::expect(try! events::pop(&mut RING) == nil);

}

/// Terminal IDs stay reserved through counter wrap and until producer acknowledgement.

@test unsafe fn terminalIdentity() throws (testing::TestError) {

    initialize();

    set STORE.queues[0].head = 0xffffffff;

    set STORE.queues[0].tail = 0xffffffff;

    sync::storeRelease32(&mut RING.head, 0xffffffff);

    sync::storeRelease32(&mut RING.tail, 0xffffffff);

    let subject = abi::Ref { index: 1, generation: 4 };

    let replacement = abi::Ref { index: 1, generation: 5 };

    let pending = try! events::reserve(&mut STORE, subject);

    events::finish(&mut STORE, pending, receiver(), events::FAULT, 13);

    try testing::expect(try! events::flush(&mut STORE, receiver(), &mut RING) == 1);

    let event = try! events::pop(&mut RING) else panic "terminal wrap";

    try testing::expect(event.kind == events::FAULT and event.code == 13 and event.value == abi::id(subject));

    let mut busy: u32 = 0;

    try events::release(&mut STORE, subject) catch err {

        try testing::expect(err == abi::Error::Busy); set busy += 1;

    };

    try reserveAndCancel(replacement) catch err {

        try testing::expect(err == abi::Error::Busy); set busy += 1;

    };

    sync::storeRelease32(&mut RING.head, 1);

    let mut invalid = false;

    try events::refresh(&mut STORE, receiver(), &RING) catch err {

        try testing::expect(err == abi::Error::InvalidArg); set invalid = true;

    };

    try testing::expect(invalid and busy == 2 and not try! events::acknowledged(&STORE, subject));

    sync::storeRelease32(&mut RING.head, 0);

    try! events::refresh(&mut STORE, receiver(), &RING);

    try! events::release(&mut STORE, subject);

    let next = try! events::reserve(&mut STORE, replacement);

    events::finish(&mut STORE, next, abi::Ref { index: 0, generation: 0 }, events::CHILD_EXIT, 0);

    try testing::expect(try! events::acknowledged(&STORE, replacement));

    let mut stale = false;

    try events::acknowledged(&STORE, subject) catch err {

        try testing::expect(err == abi::Error::BadHandle); set stale = true;

    };

    try testing::expect(stale);

}

/// Consume a successful reservation probe without committing termination.

unsafe fn reserveAndCancel(subject: abi::Ref) throws (abi::Error) {

    let pending = try events::reserve(&mut STORE, subject);

    events::cancel(&mut STORE, pending);

}

/// Queue and terminal pools reject invalid generations and restore cancelled capacity.

@test unsafe fn capacityAndValidation() throws (testing::TestError) {

    initialize();

    try testing::expect(@sizeOf(events::Event) == 16 and @sizeOf(events::Ring) == 4112);

    for i in 0..512 { try! reserveAndCancel(abi::Ref { index: i, generation: 1 }); }

    let mut invalid: u32 = 0;

    for subject in &[abi::Ref { index: 512, generation: 1 }, abi::Ref { index: 0, generation: 0 }] {

        try reserveAndCancel(subject) catch err {

            try testing::expect(err == abi::Error::InvalidArg); set invalid += 1;

        };

    }

    let mut event = wake(0);

    set event.kind = events::FAULT;

    try events::send(&mut STORE, receiver(), &mut RING, event) catch err {

        try testing::expect(err == abi::Error::InvalidArg); set invalid += 1;

    };

    set event.kind = events::WAKEUP; set event.reserved = 1;

    try events::send(&mut STORE, receiver(), &mut RING, event) catch err {

        try testing::expect(err == abi::Error::InvalidArg); set invalid += 1;

    };

    try testing::expect(invalid == 4 and STORE.queues[0].tail == 0);

}