From 5792b6d53ca2606e06d16b5c9252713c2136e610 Mon Sep 17 00:00:00 2001 From: Johnny D Date: Thu, 17 Sep 2026 12:07:46 -0400 Subject: [PATCH] Drain bounded observation inboxes without shifting their backlog Replace EventMailbox and native pull-reader pending arrays with a private, lazily grown circular FIFO. Drop only the obsolete ordered prefix after resynchronization and release consumed envelope references immediately. The existing capacities, single scheduled drain, locks, generation/retirement checks and overflow/snapshot contracts are unchanged. Retain all existing tests; add seven tests covering 10,000-event wraps across non-power-of-two and maximum capacities, lazy bounded allocation, token release, partial reads, overflow recovery and repeated full-rate hub backlogs. Add an optional actual-mailbox release microbenchmark without a CI timing threshold. Linux validation: 89 package tests with warnings as errors and 6 real Swift/C/SDL/Cemu host CTests pass. The identical queue-only benchmark reduces 4096-envelope median drain time from about 42 ms to 0.94 ms on this runner; this is not controller latency or a platform guarantee. Native Apple-SDK and complete emulator qualification must run on the published head. --- Sources/Switch2Kit/Public/Observation.swift | 78 +++++++-- .../PendingControllerEventsTests.swift | 149 ++++++++++++++++++ tests/observation-buffer/Benchmark.swift | 30 ++++ tests/observation-buffer/README.md | 23 +++ tests/observation-buffer/benchmark.sh | 11 ++ 5 files changed, 282 insertions(+), 9 deletions(-) create mode 100644 Tests/Switch2KitTests/PendingControllerEventsTests.swift create mode 100644 tests/observation-buffer/Benchmark.swift create mode 100644 tests/observation-buffer/README.md create mode 100755 tests/observation-buffer/benchmark.sh diff --git a/Sources/Switch2Kit/Public/Observation.swift b/Sources/Switch2Kit/Public/Observation.swift index 866fd58..f450a2a 100644 --- a/Sources/Switch2Kit/Public/Observation.swift +++ b/Sources/Switch2Kit/Public/Observation.swift @@ -40,15 +40,74 @@ package protocol EventSink: Sendable { func cancel() } +// FIFO specialized for the hub's monotonically numbered envelopes. Storage grows +// lazily up to the observation's existing bound. Popping releases the envelope +// immediately without shifting the rest of the backlog under a producer lock. +struct PendingControllerEvents: Sendable { + private let capacity: Int + private var storage: [EventEnvelope?] = [] + private var head = 0 + private(set) var count = 0 + var isEmpty: Bool { count == 0 } + var allocatedCount: Int { storage.count } + + init(capacity: Int) { + precondition((1...4096).contains(capacity)) + self.capacity = capacity + } + mutating func append(_ event: EventEnvelope) { + precondition(count < capacity) + if count == storage.count { + var grown = [EventEnvelope?](repeating: nil, + count: min(capacity, max(1, storage.count * 2))) + for index in 0.. EventEnvelope? { + guard count != 0 else { return nil } + let event = storage[head] + storage[head] = nil + head = (head + 1) % storage.count + count -= 1 + return event + } + mutating func takeFirst(_ maximum: Int) -> [EventEnvelope] { + var result: [EventEnvelope] = [] + result.reserveCapacity(min(maximum, count)) + for _ in 0.. private let capacity: Int private let queue: DispatchQueue private let interval: TimeInterval @@ -58,6 +117,7 @@ package final class EventMailbox: EventSink { current: @escaping @Sendable () -> EventEnvelope, handler: @escaping @Sendable (Switch2ControllerEvent) -> Void) { self.capacity = min(4096, max(1, capacity)); self.queue = queue + self.state = Mutex(State(pending: PendingControllerEvents(capacity: self.capacity))) self.interval = interval; self.current = current; self.handler = handler } package func enqueue(_ event: EventEnvelope) { @@ -79,7 +139,7 @@ package final class EventMailbox: EventSink { guard !value.cancelled else { return (false, nil) } if value.overflow { value.overflow = false; return (true, nil) } guard !value.pending.isEmpty else { return (false, nil) } - return (false, value.pending.removeFirst()) + return (false, value.pending.popFirst()) } var envelope: EventEnvelope if next.0 { envelope = current() } @@ -93,7 +153,7 @@ package final class EventMailbox: EventSink { let deliver = state.withLock { value in guard !value.cancelled, envelope.sequence > value.delivered else { return false } value.delivered = envelope.sequence - value.pending.removeAll { $0.sequence <= value.delivered } + value.pending.discard(through: value.delivered) return true } if deliver { handler(envelope.event) } @@ -209,18 +269,19 @@ package final class ControllerEventHub: Sendable { // The only producer is ControllerEventHub; one host thread drains each reader. package final class ControllerEventReader: EventSink { private struct State: Sendable { - var pending: [EventEnvelope] = [] + var pending: PendingControllerEvents var overflow = true var cancelled = false var delivered: UInt64 = 0 } - private let state = Mutex(State()) + private let state: Mutex private let capacity: Int private let current: @Sendable () -> EventEnvelope private let remove: @Sendable () -> Void package init(capacity: Int, current: @escaping @Sendable () -> EventEnvelope, remove: @escaping @Sendable () -> Void) { self.capacity = capacity; self.current = current; self.remove = remove + self.state = Mutex(State(pending: PendingControllerEvents(capacity: capacity))) } package func enqueue(_ event: EventEnvelope) { state.withLock { value in @@ -237,8 +298,7 @@ package final class ControllerEventReader: EventSink { let picked: (events: [EventEnvelope], resync: Bool) = state.withLock { value in guard !value.cancelled, maximum > 0 else { return ([], false) } if value.overflow { value.overflow = false; return ([], true) } - let events = Array(value.pending.prefix(maximum)) - value.pending.removeFirst(events.count) + let events = value.pending.takeFirst(maximum) return (events, events.contains { $0.lifetime?.isActive == false || $0.snapshotLifetimes.contains { !$0.isActive } }) } let now = current() @@ -251,7 +311,7 @@ package final class ControllerEventReader: EventSink { guard !value.cancelled else { return ([], snapshot, false, false) } if resync { value.delivered = max(value.delivered, now.sequence) - value.pending.removeAll { $0.sequence <= value.delivered } + value.pending.discard(through: value.delivered) // A concurrent producer may have overflowed after current() was read. // Keep that overflow bit, forcing another authoritative snapshot next read. return ([], snapshot, true, value.overflow || !value.pending.isEmpty) diff --git a/Tests/Switch2KitTests/PendingControllerEventsTests.swift b/Tests/Switch2KitTests/PendingControllerEventsTests.swift new file mode 100644 index 0000000..f2be30e --- /dev/null +++ b/Tests/Switch2KitTests/PendingControllerEventsTests.swift @@ -0,0 +1,149 @@ +import Foundation +import Synchronization +import XCTest +@testable import Switch2Kit + +final class PendingControllerEventsTests: XCTestCase { + private func envelope(_ sequence: UInt64, lifetime: SessionLifetime? = nil) -> EventEnvelope { + .init(sequence: sequence, event: .status(.init()), lifetime: lifetime) + } + + func testLazyGrowthAndRepeatedWrapStayWithinConfiguredCapacity() { + for capacity in [1, 2, 3, 7, 256, 4096] { + var inbox = PendingControllerEvents(capacity: capacity) + var expected: [UInt64] = [] + XCTAssertEqual(inbox.allocatedCount, 0, "Idle observers must not preallocate their maximum backlog") + for sequence in UInt64(1)...10_000 { + if expected.count == capacity { + XCTAssertEqual(inbox.popFirst()?.sequence, expected.removeFirst()) + } + inbox.append(envelope(sequence)); expected.append(sequence) + if sequence % 3 == 0 { + XCTAssertEqual(inbox.popFirst()?.sequence, expected.removeFirst()) + } + XCTAssertEqual(inbox.count, expected.count) + XCTAssertLessThanOrEqual(inbox.allocatedCount, capacity) + } + XCTAssertEqual(inbox.takeFirst(capacity).map(\.sequence), expected) + XCTAssertTrue(inbox.isEmpty) + XCTAssertNil(inbox.popFirst()) + } + } + + func testGrowthPreservesWrappedEntriesAndPartialReads() { + var inbox = PendingControllerEvents(capacity: 7) + for n in UInt64(1)...4 { inbox.append(envelope(n)) } + XCTAssertEqual(inbox.takeFirst(3).map(\.sequence), [1, 2, 3]) + for n in UInt64(5)...10 { inbox.append(envelope(n)) } + XCTAssertEqual(inbox.allocatedCount, 7) + XCTAssertTrue(inbox.takeFirst(0).isEmpty) + XCTAssertEqual(inbox.takeFirst(2).map(\.sequence), [4, 5]) + XCTAssertEqual(inbox.takeFirst(100).map(\.sequence), [6, 7, 8, 9, 10]) + XCTAssertTrue(inbox.isEmpty) + } + + func testResynchronizationDiscardsOnlyObsoletePrefixAfterWrap() { + var inbox = PendingControllerEvents(capacity: 4) + for n in UInt64(1)...4 { inbox.append(envelope(n)) } + _ = inbox.takeFirst(3) + for n in UInt64(5)...7 { inbox.append(envelope(n)) } + inbox.discard(through: 3) + XCTAssertEqual(inbox.count, 4) + inbox.discard(through: 5) + XCTAssertEqual(inbox.takeFirst(4).map(\.sequence), [6, 7]) + inbox.append(envelope(8)); inbox.discard(through: .max) + XCTAssertTrue(inbox.isEmpty) + } + + func testPoppedAndDiscardedEnvelopesReleaseTheirLifetimeImmediately() { + var inbox = PendingControllerEvents(capacity: 8) + weak var popped: SessionLifetime? + weak var discarded: SessionLifetime? + do { + let first = SessionLifetime(), second = SessionLifetime() + popped = first; discarded = second + inbox.append(envelope(1, lifetime: first)) + inbox.append(envelope(2, lifetime: second)) + } + XCTAssertNotNil(popped); XCTAssertNotNil(discarded) + _ = inbox.popFirst() + XCTAssertNil(popped, "Consumed storage must not pin a retired attempt") + XCTAssertNotNil(discarded) + inbox.discard(through: 2) + XCTAssertNil(discarded) + } + + func testClearReleasesSnapshotTokensAndOptionallyRetainsOnlyEmptySlots() { + for keep in [true, false] { + var inbox = PendingControllerEvents(capacity: 4) + weak var token: SessionLifetime? + do { + let lifetime = SessionLifetime(); token = lifetime + var item = envelope(1) + item.snapshotLifetimes = [lifetime] + inbox.append(item) + } + XCTAssertNotNil(token) + inbox.removeAll(keepingCapacity: keep) + XCTAssertNil(token) + XCTAssertTrue(inbox.isEmpty) + XCTAssertEqual(inbox.allocatedCount, keep ? 1 : 0) + inbox.append(envelope(2)) + XCTAssertEqual(inbox.popFirst()?.sequence, 2) + } + } + + func testFullRateObservationDrainsRepeatedMaximumBacklogsInOrder() throws { + let hub = ControllerEventHub(), lifetime = SessionLifetime() + let id = Switch2ControllerID(rawValue: UUID()) + let queue = DispatchQueue(label: "test.observation.circular") + let reports = Mutex<[UInt64]>([]) + let done = DispatchSemaphore(value: 0) + let observation = try hub.observe(queue: queue, capacity: 4096) { event in + if case .input(let controller) = event { + reports.withLock { $0.append(controller.state.sequence) } + if controller.state.sequence % 4095 == 0 { done.signal() } + } + } + defer { observation.cancel() } + for round in 0..<3 { + queue.suspend() + for n in 1...4095 { + let sequence = UInt64(round * 4095 + n) + let value = Switch2Controller(id: id, model: .joyCon2Left, + state: .init(buttons: sequence % 2 == 0 ? [] : .slL, sequence: sequence), + connectedAt: Date(timeIntervalSince1970: 0), bodyColor: nil, buttonColor: nil, + serialNumber: nil, sessionGeneration: lifetime.id, lastActivityAt: 0) + hub.publish(.init(controllers: [value]), event: .input(value), lifetime: lifetime) + } + queue.resume() + XCTAssertEqual(done.wait(timeout: .now() + 10), .success) + } + XCTAssertEqual(reports.withLock { $0 }, Array(UInt64(1)...12_285)) + } + + func testPullReaderPreservesPartialReadFlagsAndOverflowRecovery() throws { + let hub = ControllerEventHub(), reader = try hub.makeReader(capacity: 7) + defer { reader.cancel() } + XCTAssertTrue(reader.read(maximum: 1).resync) + for _ in 0..<1000 { + for _ in 0..<7 { hub.publish(.init(), event: .status(.init())) } + for count in [3, 2, 2] { + let batch = reader.read(maximum: count) + XCTAssertFalse(batch.resync) + XCTAssertEqual(batch.events.count, count) + XCTAssertEqual(batch.more, reader.pendingCount > 0) + } + XCTAssertEqual(reader.pendingCount, 0) + } + for _ in 0..<100 { hub.publish(.init(), event: .status(.init())) } + let recovered = reader.read(maximum: 7) + XCTAssertTrue(recovered.resync); XCTAssertTrue(recovered.events.isEmpty) + XCTAssertFalse(recovered.more) + hub.publish(.init(), event: .status(.init())) + XCTAssertEqual(reader.read(maximum: 1).events.count, 1) + reader.cancel() + hub.publish(.init(), event: .status(.init())) + XCTAssertTrue(reader.read(maximum: 7).events.isEmpty) + } +} diff --git a/tests/observation-buffer/Benchmark.swift b/tests/observation-buffer/Benchmark.swift new file mode 100644 index 0000000..b3ca96f --- /dev/null +++ b/tests/observation-buffer/Benchmark.swift @@ -0,0 +1,30 @@ +import Foundation +import Synchronization + +@main enum Benchmark { + static func main() { + for capacity in [256, 4096] { + var elapsed: [Double] = [] + for _ in 0..<9 { + let queue = DispatchQueue(label: "buffer.benchmark") + queue.suspend() + let done = DispatchSemaphore(value: 0) + let calls = Mutex(0) + let box = EventMailbox(capacity: capacity, queue: queue, current: { + EventEnvelope(sequence: UInt64(capacity + 1), event: .snapshot(.init()), lifetime: nil) + }, handler: { _ in + let count = calls.withLock { $0 += 1; return $0 } + if count == capacity { done.signal() } + }) + for n in 1...capacity { box.enqueue(.init(sequence: UInt64(n), event: .status(.init()), lifetime: nil)) } + let start = DispatchTime.now().uptimeNanoseconds + queue.resume() + precondition(done.wait(timeout: .now() + 20) == .success) + elapsed.append(Double(DispatchTime.now().uptimeNanoseconds - start) / 1e6) + box.cancel(); queue.sync {} + precondition(calls.withLock { $0 } == capacity) + } + print("capacity=\(capacity) median_drain_ms=\(elapsed.sorted()[4]) samples_ms=\(elapsed)") + } + } +} diff --git a/tests/observation-buffer/README.md b/tests/observation-buffer/README.md new file mode 100644 index 0000000..cc7f2c2 --- /dev/null +++ b/tests/observation-buffer/README.md @@ -0,0 +1,23 @@ +# Observation inbox microbenchmark + +From any working directory: + +```sh +bash /path/to/Switch2Kit/tests/observation-buffer/benchmark.sh +``` + +This optional benchmark compiles the real `EventMailbox` in release mode. It queues +exactly 256 or 4096 status envelopes while its delivery queue is suspended, then +measures resume-to-last-handler time for nine repetitions. The output contains all +samples and their median. It checks completion and delivered count, not a timing +threshold. It is not a Bluetooth, SDL, emulator-frame, or physical input latency test. + +For a before/after comparison, run the identical `Benchmark.swift` and compile +command against each revision's `Observation.swift` and its own source dependencies +in separate checkouts. Do not compare a debug build against a release build. CPU +load, scheduler behavior and platform affect the measured times. + +Correctness and storage-bound tests are in +`Tests/Switch2KitTests/PendingControllerEventsTests.swift`; the existing observation, +reader, generation and native consumer regressions remain mandatory. This optional +microbenchmark is deliberately not discovered as a timing-sensitive CI test. diff --git a/tests/observation-buffer/benchmark.sh b/tests/observation-buffer/benchmark.sh new file mode 100755 index 0000000..0f4ad97 --- /dev/null +++ b/tests/observation-buffer/benchmark.sh @@ -0,0 +1,11 @@ +#!/bin/bash +# Optional queue-only microbenchmark; not a hardware latency or CI timing assertion. +set -euo pipefail +cd "$(dirname "$0")/../.." +source tests/support/kit-sources.sh +work=$(mktemp -d) +trap 'rm -rf "$work"' EXIT +swiftc "${kit_flags[@]}" -swift-version 6 -O -warnings-as-errors "${kit_sources[@]}" \ + Sources/Switch2Kit/Public/Observation.swift tests/observation-buffer/Benchmark.swift \ + -o "$work/benchmark" +"$work/benchmark"