diff --git a/src/lib/__tests__/socket-child.spec.ts b/src/lib/__tests__/socket-child.spec.ts index 1e370bd4..e79631a9 100644 --- a/src/lib/__tests__/socket-child.spec.ts +++ b/src/lib/__tests__/socket-child.spec.ts @@ -898,4 +898,323 @@ describe('SocketChild', () => { } } }) + + function genPayloadMessage(pktId: number): Buffer { + // Payload carries the packet id, so delivery order can be checked from the payload itself + const payload = Buffer.alloc(2) + payload.writeUInt16BE(pktId, 0) + return Buffer.concat([genAckRequestMessage(pktId, 2), payload]) + } + + function genNewSessionMessage(pktId: number): Buffer { + const buffer = Buffer.from([ + 0x10, + 0x0c, // Length & Type + 0x53, + 0x1b, // Session Id + 0x00, + 0x00, // Not acking + 0x00, + 0x00, // Not asking for retransmit + 0x00, + 0x00, // 'Client pkt id' Not needed + 0x00, + 0x00, // Packet Id + ]) + buffer.writeUInt16BE(pktId, 10) // Packet Id + + return buffer + } + + function createOutOfOrderHarness() { + const delivered: number[] = [] + const child = createSocketChild(async (buf, packetId) => { + // The payload must belong to the packet id it is delivered with + expect(buf.readUInt16BE(0)).toEqual(packetId) + delivered.push(packetId) + return Promise.resolve() + }) + fakeConnect(child) + const socket = getSocket(child) + + const acked: number[] = [] + socket.sendImpl = (msg: Buffer): void => { + const opcode = msg.readUInt8(0) >> 3 + expect(opcode).toEqual(PacketFlag.AckReply) + acked.push(msg.readUInt16BE(4)) + } + + return { child, socket, delivered, acked } + } + + test('Inbound commands - out of order packets are buffered until the gap is filled', async () => { + const { child, socket, delivered, acked } = createOutOfOrderHarness() + try { + // Packet 1 is lost, 2-4 arrive ahead of sequence + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(3)) + await socket.emitMessage(clock, genPayloadMessage(4)) + await clock.tickAsync(20) + expect(delivered).toEqual([]) + expect(acked).toEqual([]) + + // The missing packet arrives: everything is delivered in order, and acked straight away + await socket.emitMessage(clock, genPayloadMessage(1)) + expect(delivered).toEqual([1, 2, 3, 4]) + expect(acked).toEqual([4]) + expect((child as any)._lastReceivedPacketId).toEqual(4) + + // Sequence continues normally afterwards + await socket.emitMessage(clock, genPayloadMessage(5)) + await clock.tickAsync(20) + expect(delivered).toEqual([1, 2, 3, 4, 5]) + expect(acked).toEqual([4, 5]) + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - out of order packets around wrap', async () => { + const { child, socket, delivered, acked } = createOutOfOrderHarness() + try { + ;(child as any)._lastReceivedPacketId = 32766 // 32767 is max + + await socket.emitMessage(clock, genPayloadMessage(0)) + await socket.emitMessage(clock, genPayloadMessage(1)) + await clock.tickAsync(20) + expect(delivered).toEqual([]) + + await socket.emitMessage(clock, genPayloadMessage(32767)) + expect(delivered).toEqual([32767, 0, 1]) + expect(acked).toEqual([1]) + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - retransmits of buffered packets are not delivered twice', async () => { + const { child, socket, delivered, acked } = createOutOfOrderHarness() + try { + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(3)) + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(3)) + expect(delivered).toEqual([]) + + await socket.emitMessage(clock, genPayloadMessage(1)) + expect(delivered).toEqual([1, 2, 3]) + expect(acked).toEqual([3]) + + // Late retransmits of already delivered packets are re-acked, not re-delivered + await socket.emitMessage(clock, genPayloadMessage(1)) + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(3)) + await clock.tickAsync(20) + expect(delivered).toEqual([1, 2, 3]) + expect(acked).toEqual([3, 3]) + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - packets too far ahead are not buffered', async () => { + const { child, socket, delivered } = createOutOfOrderHarness() + try { + // Way beyond any sane reorder window + await socket.emitMessage(clock, genPayloadMessage(5000)) + expect((child as any)._outOfOrderPackets.size).toEqual(0) + + await socket.emitMessage(clock, genPayloadMessage(1)) + expect(delivered).toEqual([1]) + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - packet ids beyond 15 bits are ignored', async () => { + const { child, socket, delivered } = createOutOfOrderHarness() + try { + // Fill the whole reorder window, along with the 16 bit alias of every id in it + for (let pktId = 2; pktId <= 256; pktId++) { + await socket.emitMessage(clock, genPayloadMessage(pktId)) + await socket.emitMessage(clock, genPayloadMessage(pktId + 32768)) + } + expect((child as any)._outOfOrderPackets.size).toEqual(255) + expect([...(child as any)._outOfOrderPackets.keys()].every((pktId: number) => pktId < 32768)).toBeTrue() + + // The alias of the next packet is not accepted in its place + await socket.emitMessage(clock, genPayloadMessage(1 + 32768)) + expect(delivered).toEqual([]) + expect((child as any)._lastReceivedPacketId).toEqual(0) + + // The real packet drains the whole window, leaving nothing behind + await socket.emitMessage(clock, genPayloadMessage(1)) + expect(delivered).toHaveLength(256) + expect((child as any)._outOfOrderPackets.size).toEqual(0) + expect((child as any)._outOfOrderStalledSince).toBeUndefined() + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - out of order buffer is cleared by a new session', async () => { + const { child, socket, delivered, acked } = createOutOfOrderHarness() + try { + await socket.emitMessage(clock, genPayloadMessage(3)) + await socket.emitMessage(clock, genPayloadMessage(4)) + expect(delivered).toEqual([]) + expect((child as any)._outOfOrderPackets.size).toEqual(2) + + // The device starts a new session, the buffered packets belong to the old one + await socket.emitMessage(clock, genNewSessionMessage(0)) + expect((child as any)._outOfOrderPackets.size).toEqual(0) + expect(acked).toEqual([0]) + + await socket.emitMessage(clock, genPayloadMessage(1)) + await socket.emitMessage(clock, genPayloadMessage(2)) + expect(delivered).toEqual([1, 2]) + + // Buffer again, then a reconnect must also clear it + await socket.emitMessage(clock, genPayloadMessage(5)) + expect((child as any)._outOfOrderPackets.size).toEqual(1) + await child.connect(ADDRESS, DEFAULT_PORT) + expect((child as any)._outOfOrderPackets.size).toEqual(0) + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - a gap that never fills triggers a reconnect, even with other traffic', async () => { + let connected = true + const child = createSocketChild(undefined, undefined, async () => { + connected = false + }) + try { + fakeConnect(child) + const socket = getSocket(child) + socket.sendImpl = (): void => { + // Ignore acks and the reconnect hello + } + + // Packet 1 is lost + await socket.emitMessage(clock, genPayloadMessage(2)) + expect((child as any)._outOfOrderPackets.size).toEqual(1) + + // Keep hearing ack replies from the atem, which would otherwise keep the connection alive + for (let i = 0; i < 9; i++) { + await clock.tickAsync(500) + await socket.emitMessage(clock, genAckCommandMessage(0)) + // More packets pile up behind the gap + await socket.emitMessage(clock, genPayloadMessage(3 + i)) + } + expect(connected).toBeTrue() + + // The stall exceeds the connection timeout + await clock.tickAsync(1000) + await socket.emitMessage(clock, genAckCommandMessage(0)) + expect(connected).toBeFalse() + expect((child as any)._outOfOrderPackets.size).toEqual(0) + expect((child as any)._outOfOrderStalledSince).toBeUndefined() + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - progress through the buffer restarts the stall clock', async () => { + let connected = true + const delivered: number[] = [] + const child = createSocketChild( + async (_buf, packetId) => { + delivered.push(packetId) + }, + undefined, + async () => { + connected = false + } + ) + try { + fakeConnect(child) + const socket = getSocket(child) + socket.sendImpl = (): void => { + // Ignore acks + } + + // 1 and 3 lost + await socket.emitMessage(clock, genPayloadMessage(2)) + await socket.emitMessage(clock, genPayloadMessage(4)) + + // 1 is recovered after 4s, which drains 2 but leaves 4 waiting behind 3 + await clock.tickAsync(4000) + await socket.emitMessage(clock, genPayloadMessage(1)) + expect(delivered).toEqual([1, 2]) + expect((child as any)._outOfOrderPackets.size).toEqual(1) + + // 4s more is beyond 5s since the first gap, but not since the progress + await clock.tickAsync(4000) + expect(connected).toBeTrue() + + // 3 is recovered, everything drains and the stall is over + await socket.emitMessage(clock, genPayloadMessage(3)) + expect(delivered).toEqual([1, 2, 3, 4]) + expect((child as any)._outOfOrderStalledSince).toBeUndefined() + + // Normal traffic keeps it alive well past the old stall deadline + for (let i = 5; i < 10; i++) { + await clock.tickAsync(1000) + await socket.emitMessage(clock, genPayloadMessage(i)) + } + expect(connected).toBeTrue() + } finally { + await child.disconnect() + } + }) + + test('Inbound commands - backlog drains under per-frame retransmit bursts', async () => { + // Models a capture against a real ATEM: after one lost packet the ATEM retransmits its unacked window + // once per frame in bursts where the packets are interleaved with a stride of 3, while also sending a + // new unsolicited packet (Time) every frame. Without a reorder buffer only one packet per frame is + // accepted, which equals the arrival rate, so the backlog never drains. + const { child, socket, delivered, acked } = createOutOfOrderHarness() + try { + const lastAcked = (): number => (acked.length ? acked[acked.length - 1] : 0) + + let atemNextId = 1 + // Initial stream at one packet per frame, with packet 1 lost on the way + for (let frame = 0; frame < 22; frame++) { + const id = atemNextId++ + if (id !== 1) await socket.emitMessage(clock, genPayloadMessage(id)) + await clock.tickAsync(20) + } + + const backlogs: number[] = [] + for (let frame = 0; frame < 30; frame++) { + const oldestUnacked = lastAcked() + 1 + const newestSent = atemNextId - 1 + // Retransmit burst: the oldest unacked packet, then part of the window at stride 3 with a rotating + // phase that never includes oldest + 1. So exactly one packet per burst is next in sequence. + const burst: number[] = [] + if (oldestUnacked <= newestSent) burst.push(oldestUnacked) + for (let id = oldestUnacked + 2 + (frame % 3); id <= newestSent; id += 3) burst.push(id) + for (const id of burst) await socket.emitMessage(clock, genPayloadMessage(id)) + + // Plus one new packet per frame + await socket.emitMessage(clock, genPayloadMessage(atemNextId++)) + await clock.tickAsync(20) + + backlogs.push(atemNextId - 1 - lastAcked()) + } + + // The backlog must have drained, rather than staying constant + expect(backlogs[backlogs.length - 1]).toEqual(0) + expect(backlogs.slice(-10)).toEqual(Array(10).fill(0)) + + // Every packet delivered exactly once, in order + const expected: number[] = [] + for (let id = 1; id < atemNextId; id++) expected.push(id) + expect(delivered).toEqual(expected) + } finally { + await child.disconnect() + } + }) }) diff --git a/src/lib/atemSocketChild.ts b/src/lib/atemSocketChild.ts index 781863df..874f4479 100644 --- a/src/lib/atemSocketChild.ts +++ b/src/lib/atemSocketChild.ts @@ -11,6 +11,12 @@ const RETRANSMIT_INTERVAL = 10 // ms const MAX_PACKET_RETRIES = 10 const MAX_PACKET_ID = 1 << 15 // Atem expects 15 not 16 bits before wrapping const MAX_PACKET_PER_ACK = 16 +/** + * How far ahead of the expected packet id an inbound packet may be and still get buffered, which also bounds the + * buffer size. The atem keeps resending unacked packets, so this only needs to cover its unacked window (observed + * ~20-30 packets deep when one packet was lost at 50 packets/s). Packets are at most 2047 bytes, so this is <512KB. + */ +const MAX_OUT_OF_ORDER_PACKETS = 256 export const COMMAND_CONNECT_HELLO = Buffer.from([ 0x10, 0x14, 0x53, 0xab, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3a, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, @@ -63,6 +69,10 @@ export class AtemSocketChild { private _lastReceivedAt: number = performance.now() private _lastReceivedPacketId = 0 + /** Payloads of AckRequest packets that arrived ahead of sequence, keyed by packet id */ + private _outOfOrderPackets = new Map() + /** When the inbound stream last made progress while there were packets waiting in `_outOfOrderPackets` */ + private _outOfOrderStalledSince: number | undefined private _inFlight: InFlightPacket[] = [] private _ackTimer: NodeJS.Timeout | undefined private _receivedWithoutAck = 0 @@ -94,11 +104,20 @@ export class AtemSocketChild { private startTimers(): void { if (!this._reconnectTimer) { this._reconnectTimer = setInterval(() => { - if (this._lastReceivedAt + CONNECTION_TIMEOUT > performance.now()) { - // We heard from the atem recently + const now = performance.now() + const heardRecently = this._lastReceivedAt + CONNECTION_TIMEOUT > now + const inboundStalled = + this._outOfOrderStalledSince !== undefined && + this._outOfOrderStalledSince + CONNECTION_TIMEOUT <= now + if (heardRecently && !inboundStalled) { + // We heard from the atem recently, and the inbound stream is not stuck behind a lost packet return } + if (inboundStalled) { + this.log(`Inbound stalled waiting for packet ${(this._lastReceivedPacketId + 1) % MAX_PACKET_ID}`) + } + this.restartConnection().catch((e) => { this.log(`Reconnect failed: ${e}`) }) @@ -118,6 +137,7 @@ export class AtemSocketChild { return this._closeSocket().then(async () => { this._connectionState = ConnectionState.Disconnected + this._clearOutOfOrderPackets() return this.onDisconnect() }) } @@ -149,6 +169,7 @@ export class AtemSocketChild { this._nextSendPacketId = 1 this._sessionId = 0 this._inFlight = [] + this._clearOutOfOrderPackets() this.log('reconnect') this.startTimers() @@ -256,6 +277,7 @@ export class AtemSocketChild { if (flags & PacketFlag.NewSessionId) { this._connectionState = ConnectionState.Established this._lastReceivedPacketId = remotePacketId + this._clearOutOfOrderPackets() this._sendAck(remotePacketId) return } @@ -271,17 +293,50 @@ export class AtemSocketChild { ps.push(this._retransmitFrom(fromPacketId)) } - // Got a packet that needs an ack - if (flags & PacketFlag.AckRequest) { + // Got a packet that needs an ack. Ids are only 15 bits, anything larger is malformed and could never be drained + if (flags & PacketFlag.AckRequest && remotePacketId < MAX_PACKET_ID) { + const distance = (remotePacketId - this._lastReceivedPacketId + MAX_PACKET_ID) % MAX_PACKET_ID // Check if it next in the sequence - if (remotePacketId === (this._lastReceivedPacketId + 1) % MAX_PACKET_ID) { + if (distance === 1) { this._lastReceivedPacketId = remotePacketId - this._sendOrQueueAck() // It might have commands if (length > 12) { ps.push(this.onCommandsReceived(packet.subarray(12), remotePacketId)) } + + // Deliver any buffered packets which are now in sequence + let drained = 0 + let nextPacketId = (this._lastReceivedPacketId + 1) % MAX_PACKET_ID + let nextPayload = this._outOfOrderPackets.get(nextPacketId) + while (nextPayload) { + this._outOfOrderPackets.delete(nextPacketId) + this._lastReceivedPacketId = nextPacketId + drained++ + + if (nextPayload.length > 0) { + ps.push(this.onCommandsReceived(nextPayload, nextPacketId)) + } + + nextPacketId = (nextPacketId + 1) % MAX_PACKET_ID + nextPayload = this._outOfOrderPackets.get(nextPacketId) + } + + // Progress was made, so restart the stall clock for any packets still waiting behind a new gap + this._outOfOrderStalledSince = this._outOfOrderPackets.size > 0 ? performance.now() : undefined + + if (drained > 0) { + // Ack immediately, so the atem stops retransmitting what we now have + this._sendAckNow() + } else { + this._sendOrQueueAck() + } + } else if (distance > 1 && distance <= MAX_OUT_OF_ORDER_PACKETS) { + // It is ahead of sequence, so hold onto it until the gap is filled + if (!this._outOfOrderPackets.has(remotePacketId)) { + if (this._outOfOrderPackets.size === 0) this._outOfOrderStalledSince = performance.now() + this._outOfOrderPackets.set(remotePacketId, packet.subarray(12)) + } } else if (this._isPacketCoveredByAck(this._lastReceivedPacketId, remotePacketId)) { // We got a retransmit of something we have already acked, so reack it this._sendOrQueueAck() @@ -321,17 +376,15 @@ export class AtemSocketChild { this._socket.send(packet, 0, packet.length, this._port, this._address) } + private _clearOutOfOrderPackets(): void { + this._outOfOrderPackets.clear() + this._outOfOrderStalledSince = undefined + } + private _sendOrQueueAck(): void { this._receivedWithoutAck++ if (this._receivedWithoutAck >= MAX_PACKET_PER_ACK) { - this._receivedWithoutAck = 0 - - if (this._ackTimer) { - clearTimeout(this._ackTimer) - delete this._ackTimer - } - - this._sendAck(this._lastReceivedPacketId) + this._sendAckNow() } else if (!this._ackTimer) { this._ackTimer = setTimeout(() => { delete this._ackTimer @@ -341,6 +394,17 @@ export class AtemSocketChild { } } + private _sendAckNow(): void { + this._receivedWithoutAck = 0 + + if (this._ackTimer) { + clearTimeout(this._ackTimer) + delete this._ackTimer + } + + this._sendAck(this._lastReceivedPacketId) + } + private _sendAck(packetId: number): void { const opcode = PacketFlag.AckReply << 11 const length = 12