diff --git a/Sources/FinallyTheControllerWorks/Output/UDPHub.swift b/Sources/FinallyTheControllerWorks/Output/UDPHub.swift index 20358ab..b2b6f86 100644 --- a/Sources/FinallyTheControllerWorks/Output/UDPHub.swift +++ b/Sources/FinallyTheControllerWorks/Output/UDPHub.swift @@ -8,7 +8,7 @@ // | f32 lx,ly,rx,ry | u8 lt,rt | u16 battery_mv // | i16 gyro[3] | i16 accel[3] // Rumble in (6 bytes): "S2R1" | u8 strong | u8 weak -// Any inbound datagram registers its sender as a subscriber (30 s TTL). +// An empty hello or valid rumble registers its sender (30 s TTL). import Foundation import Darwin @@ -17,6 +17,7 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable { private static let basePort: UInt16 = 24800 private static let peerTTL: TimeInterval = 30 + private static let maxPeers = 64 var onRumble: ((Int, Double, Double) -> Void)? @@ -93,7 +94,8 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable { private func drainSocket(slot: Int) { guard let s = slots[slot] else { return } var buf = [UInt8](repeating: 0, count: 64) - while true { + // Bound each read dispatch so a noisy local peer cannot starve output. + for _ in 0..<256 { var from = sockaddr_in() var fromLen = socklen_t(MemoryLayout.size) let n = withUnsafeMutablePointer(to: &from) { @@ -102,14 +104,19 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable { } } if n < 0 { break } // EWOULDBLOCK: drained + let isRumble = n == 6 && buf[0...3].elementsEqual([0x53, 0x32, 0x52, 0x31]) + guard n == 0 || isRumble else { continue } + let now = ProcessInfo.processInfo.systemUptime + s.peers = s.peers.filter { now - $0.value <= Self.peerTTL } let peer = SockAddr(addr: from.sin_addr.s_addr, port: from.sin_port) let isNewPeer = s.peers[peer] == nil - s.peers[peer] = CFAbsoluteTimeGetCurrent() + guard !isNewPeer || s.peers.count < Self.maxPeers else { continue } + s.peers[peer] = now if isNewPeer, !s.name.isEmpty { // Late joiners get the name before their first state packet. send(Self.namePacket(s.name), to: peer, via: s.fd) } - if n >= 6, buf[0] == 0x53, buf[1] == 0x32, buf[2] == 0x52, buf[3] == 0x31 { // "S2R1" + if isRumble { onRumble?(slot, Double(buf[4]) / 255.0, Double(buf[5]) / 255.0) } } @@ -154,7 +161,13 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable { func controllerDisconnected(slot: Int) { queue.async { [weak self] in - self?.slots[slot]?.seq = 0 + guard let self, let s = self.slots[slot] else { return } + // Release held input immediately on orderly disconnect. The SDL + // presence watchdog remains necessary for crashes or packet loss. + s.seq &+= 1 + let neutral = Self.statePacket(seq: s.seq, state: ControllerState()) + for peer in s.peers.keys { self.send(neutral, to: peer, via: s.fd) } + s.name = "" } } @@ -163,7 +176,7 @@ final class UDPHub: ControllerOutputSink, @unchecked Sendable { guard let self, let s = self.slots[slot], !s.peers.isEmpty else { return } s.seq &+= 1 let packet = Self.statePacket(seq: s.seq, state: state) - let now = CFAbsoluteTimeGetCurrent() + let now = ProcessInfo.processInfo.systemUptime for (peer, seen) in s.peers { if now - seen > Self.peerTTL { s.peers.removeValue(forKey: peer) diff --git a/tests/udp/UDPTests.swift b/tests/udp/UDPTests.swift new file mode 100644 index 0000000..eead0f4 --- /dev/null +++ b/tests/udp/UDPTests.swift @@ -0,0 +1,112 @@ +import Foundation +#if os(Linux) +import Glibc +let datagram = Int32(SOCK_DGRAM.rawValue) +#else +import Darwin +let datagram = SOCK_DGRAM +#endif + +enum BridgeEngine { static let maxPlayers = 4 } +enum LogLevel { case info, warning, error, debug } +func bridgeLog(_ level: LogLevel, _ category: String, _ message: String) {} +protocol ControllerOutputSink: AnyObject { + var onRumble: ((Int, Double, Double) -> Void)? { get set } + func controllerConnected(slot: Int, model: Switch2.Model) + func controllerDisconnected(slot: Int) + func controllerName(slot: Int, name: String) + func controllerState(slot: Int, state: ControllerState) +} + +@main +enum UDPTests { + static func client(_ port: UInt16) -> Int32 { + let fd = socket(AF_INET, datagram, 0) + precondition(fd >= 0) + var address = sockaddr_in() + address.sin_family = sa_family_t(AF_INET) + address.sin_port = port.bigEndian + address.sin_addr.s_addr = UInt32(0x7f000001).bigEndian + let result = withUnsafePointer(to: &address) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { + connect(fd, $0, socklen_t(MemoryLayout.size)) + } + } + precondition(result == 0) + var timeout = timeval(tv_sec: 1, tv_usec: 0) + precondition(setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, socklen_t(MemoryLayout.size)) == 0) + return fd + } + static func sendBytes(_ fd: Int32, _ data: [UInt8]) { + let n = data.withUnsafeBytes { send(fd, $0.baseAddress, $0.count, 0) } + precondition(n == data.count) + } + static func receive(_ fd: Int32) -> Data { + var bytes = [UInt8](repeating: 0, count: 128) + let n = recv(fd, &bytes, bytes.count, 0) + precondition(n >= 0, "Expected UDP packet was not delivered") + return Data(bytes.prefix(n)) + } + static func sendAndDrain(_ hub: UDPHub, slot: Int, fd: Int32, bytes: [UInt8]) { + // Hold the hub's serial queue so its dispatch source cannot consume + // the packet between poll() and the explicit production drain call. + hub.queue.sync { + sendBytes(fd, bytes) + var ready = pollfd(fd: hub.slots[slot]!.fd, events: Int16(POLLIN), revents: 0) + precondition(poll(&ready, 1, 1000) == 1, "Loopback packet did not arrive") + hub.drainSocket(slot: slot) + } + } + static func main() { + let selected = CommandLine.arguments.last! + let hub = UDPHub() + hub.queue.sync { precondition(hub.slots.count == 4) } + let a = client(24800), b = client(24801) + defer { close(a); close(b) } + sendBytes(a, []); sendBytes(b, []) + for _ in 0..<100 { + if hub.queue.sync(execute: { !hub.slots[0]!.peers.isEmpty && !hub.slots[1]!.peers.isEmpty }) { break } + Thread.sleep(forTimeInterval: 0.005) + } + if selected == "all" || selected == "neutral" { + var held = ControllerState(); held.buttons = [.a]; held.leftTrigger = 255 + hub.controllerState(slot: 0, state: held) + precondition(Switch2.u32(receive(a), 8) == Switch2.Buttons.a.rawValue) + hub.controllerDisconnected(slot: 0) + let packet = receive(a) + precondition(packet.count == 44 && packet.prefix(4) == Data("S2B1".utf8)) + precondition(packet.dropFirst(8).allSatisfy { $0 == 0 }, "Disconnect must emit neutral state") + hub.controllerState(slot: 1, state: held) + precondition(Switch2.u32(receive(b), 8) == Switch2.Buttons.a.rawValue) + print("PASS orderly neutralization and unrelated slot") + } + if selected == "all" || selected == "malformed" { + let c = client(24802); defer { close(c) } + sendAndDrain(hub, slot: 2, fd: c, bytes: [1, 2, 3]) + hub.queue.sync { + precondition(hub.slots[2]!.peers.isEmpty, "Malformed packets must not allocate subscribers") + } + print("PASS malformed subscription rejection") + } + if selected == "all" || selected == "capacity" { + hub.queue.sync { + let s = hub.slots[3]! + for port in 1...64 { + s.peers[UDPHub.SockAddr(addr: 0x0100007f, port: UInt16(port))] = ProcessInfo.processInfo.systemUptime + } + } + let c = client(24803); defer { close(c) } + sendAndDrain(hub, slot: 3, fd: c, bytes: []) + hub.queue.sync { + precondition(hub.slots[3]!.peers.count == 64, "Peer table must stay bounded") + } + // Expired peers must not block a new subscriber even without state traffic. + hub.queue.sync { hub.slots[3]!.peers = hub.slots[3]!.peers.mapValues { _ in -100 } } + sendAndDrain(hub, slot: 3, fd: c, bytes: []) + hub.queue.sync { + precondition(hub.slots[3]!.peers.count == 1) + } + print("PASS peer cap and expiry without input") + } + } +} diff --git a/tests/udp/run.sh b/tests/udp/run.sh new file mode 100755 index 0000000..5466e79 --- /dev/null +++ b/tests/udp/run.sh @@ -0,0 +1,21 @@ +#!/bin/bash +set -euo pipefail +cd "$(dirname "$0")/../.." +work=$(mktemp -d) +trap 'rm -rf "$work"' EXIT +python3 - "$work" <<'PY' +from pathlib import Path +import os,re,sys +out=Path(sys.argv[1]) +s=Path(os.environ.get('UDP_SOURCE','Sources/FinallyTheControllerWorks/Output/UDPHub.swift')).read_text() +s=re.sub(r'\bprivate\s+', '', s) +if sys.platform != 'darwin': + s=s.replace('import Darwin','import Glibc\nimport CoreFoundation').replace('SOCK_DGRAM,','Int32(SOCK_DGRAM.rawValue),') +out.joinpath('UDPHub.swift').write_text(s) +s=Path('Sources/FinallyTheControllerWorks/Bluetooth/ControllerSession.swift').read_text() +a=s.index('struct ControllerState:'); b=s.index('/// Called on the Bluetooth queue.',a) +out.joinpath('State.swift').write_text('import Foundation\n'+s[a:b]) +PY +swiftc -swift-version 5 Sources/FinallyTheControllerWorks/Protocol/Switch2Protocol.swift \ + "$work/State.swift" "$work/UDPHub.swift" tests/udp/UDPTests.swift -o "$work/test" +"$work/test" "${UDP_CASE:-all}"