From b1dda92f095857322e36a02d4b51bf51b3770c31 Mon Sep 17 00:00:00 2001 From: Geoffrey McRae Date: Thu, 8 Oct 2026 21:34:24 +1100 Subject: [PATCH] [client/idd] input: complete network latency path Isolate input sends from shared transport accounting and reserve reliable capacity. Prioritize QUIC execution and bound stale motion through the Windows pipe and HID queues. --- client/transports/Network/connection.c | 515 +++++++++++++----- client/transports/Network/connection.h | 2 + client/transports/Network/input.c | 128 +++-- idd/LGCommon/CPipeEndpoint.cpp | 8 +- idd/LGCommon/CPipeEndpoint.h | 7 +- idd/LGIdd/ipc/CInputPipeServer.cpp | 26 +- idd/LGIdd/ipc/CInputPipeServer.h | 1 + .../transport/network/CNetworkServer.cpp | 325 ++++++----- idd/LGInput/CHIDDevice.cpp | 27 +- 9 files changed, 695 insertions(+), 344 deletions(-) diff --git a/client/transports/Network/connection.c b/client/transports/Network/connection.c index 86c68292..c7ca32ed 100644 --- a/client/transports/Network/connection.c +++ b/client/transports/Network/connection.c @@ -63,6 +63,7 @@ (2U * (LG_NET_ENVELOPE_WIRE_SIZE + NETWORK_MAX_INPUT_PAYLOAD)) #define NETWORK_MAX_NEGOTIATED_PAYLOAD LG_NET_MAX_PAYLOAD_LENGTH #define NETWORK_MAX_DATAGRAM_PAYLOAD 1136U +#define NETWORK_DATAGRAM_SEND_ENABLED UINT32_C(0x00010000) #define NETWORK_HELLO_REQUEST_ID UINT64_C(1) #define NETWORK_SEQUENCE_CONNECTION_EPOCH UINT64_C(1) #define NETWORK_AUTH_PBKDF2_ROUNDS_MIN 10000U @@ -91,7 +92,10 @@ #define NETWORK_SEND_RESUME_NUMERATOR 3U #define NETWORK_SEND_RESUME_DENOMINATOR 4U #define NETWORK_INPUT_SEND_POOL_COUNT 64U +#define NETWORK_INPUT_SEND_RELIABLE_RESERVE 16U #define NETWORK_INPUT_SEND_POOL_DATA_SIZE 256U +#define NETWORK_INPUT_SEND_READY_RELIABLE UINT8_C(0x01) +#define NETWORK_INPUT_SEND_READY_DATAGRAM UINT8_C(0x02) _Static_assert(NETWORK_AUTH_SHA256_SIZE == LG_NET_PASSWORD_SHA256_LENGTH, "Network authentication SHA-256 sizes differ"); @@ -106,6 +110,9 @@ _Static_assert(LG_NET_INPUT_CLAIM_WIRE_SIZE <= NETWORK_MAX_INPUT_PAYLOAD && _Static_assert(NETWORK_INPUT_SEND_POOL_DATA_SIZE >= LG_NET_ENVELOPE_WIRE_SIZE + NETWORK_MAX_INPUT_PAYLOAD, "Network input send pool entries are too small"); +_Static_assert(NETWORK_INPUT_SEND_RELIABLE_RESERVE < + NETWORK_INPUT_SEND_POOL_COUNT, + "Network input reliable reserve consumes the send pool"); struct NetworkSend { @@ -125,6 +132,7 @@ struct NetworkSend struct NetworkInputSend { struct NetworkSend send; + bool reliableReserved; uint8_t data[NETWORK_INPUT_SEND_POOL_DATA_SIZE]; }; @@ -193,6 +201,7 @@ struct NetworkConnection LGEvent * bulkReceiveIdle; LGThread * bulkReceiveThread; LG_Lock lock; + LG_Lock datagramSendLock; LG_Lock inputStreamLock; LG_Lock inputSendPoolLock; LG_Lock stateCallbackLock; @@ -264,8 +273,8 @@ struct NetworkConnection [LG_NET_SERVICE_CONTROL + 1U]; uint32_t maxPayloadLength; uint32_t maxDatagramPayload; - uint16_t maxDatagramSendLength; - bool datagramSendEnabled; + _Atomic uint32_t datagramSendState; + _Atomic uint16_t inputDatagramBlockedLength; uint32_t serviceMaxPayload[LG_NET_SERVICE_CONTROL + 1U]; uint32_t serviceMaxDatagram[LG_NET_SERVICE_CONTROL + 1U]; @@ -275,6 +284,8 @@ struct NetworkConnection struct NetworkInputSend inputSendPool[NETWORK_INPUT_SEND_POOL_COUNT]; struct NetworkSend * inputSendFree; + struct NetworkSend * inputReliableSendFree; + uint8_t inputSendBlocked; bool inputStreamBound; struct NetworkBulkStream * outboundBulk @@ -311,6 +322,8 @@ static QUIC_STATUS QUIC_API networkBulkStreamCallback(HQUIC handle, static QUIC_STATUS QUIC_API networkRejectedStreamCallback(HQUIC handle, void * opaque, QUIC_STREAM_EVENT * event); static bool networkConnectionOpenInputStream(NetworkConnection * this); +static void networkConnectionNotifySendReady(NetworkConnection * this, + uint16_t reliable, uint16_t datagram, uint16_t bulk); static char * networkDuplicate(const char * value) { @@ -342,34 +355,112 @@ static uint8_t * networkSendData(struct NetworkSend * send) return (uint8_t *)(send + 1); } -static struct NetworkSend * networkConnectionAllocSend( - NetworkConnection * this, LGNetService service, size_t dataSize) +static void networkConnectionInitSend(struct NetworkSend * send, + NetworkConnection * this, bool pooled) { - struct NetworkSend * send = NULL; - if (service == LG_NET_SERVICE_INPUT && - dataSize <= NETWORK_INPUT_SEND_POOL_DATA_SIZE) - { - LG_LOCK(this->inputSendPoolLock); - send = this->inputSendFree; - if (send) - this->inputSendFree = send->next; - LG_UNLOCK(this->inputSendPoolLock); - } - - if (!send) - { - send = malloc(sizeof(*send) + dataSize); - if (!send) - return NULL; - send->pooled = false; - } - send->connection = this; send->next = NULL; send->componentEpoch = 0; send->applicationSequence = 0; send->accounted = false; send->lossReported = false; + send->pooled = pooled; +} + +static uint8_t networkConnectionInputReadyBit( + NetworkPacketDelivery delivery) +{ + switch (delivery) + { + case NETWORK_PACKET_DELIVERY_RELIABLE: + return NETWORK_INPUT_SEND_READY_RELIABLE; + + case NETWORK_PACKET_DELIVERY_DATAGRAM: + return NETWORK_INPUT_SEND_READY_DATAGRAM; + + case NETWORK_PACKET_DELIVERY_BULK: + return 0; + } + return 0; +} + +static bool networkConnectionNextDatagramSequence( + NetworkConnection * this, uint64_t * sequence) +{ + if (this->datagramSendSequence == UINT64_MAX) + return false; + *sequence = ++this->datagramSendSequence; + return true; +} + +static bool networkConnectionDatagramAvailable( + NetworkConnection * this, uint64_t packetLength) +{ + const uint32_t state = atomic_load_explicit( + &this->datagramSendState, memory_order_acquire); + return (state & NETWORK_DATAGRAM_SEND_ENABLED) && + packetLength <= (uint16_t)state; +} + +static bool networkConnectionPrepareInputDatagram( + NetworkConnection * this, uint16_t packetLength) +{ + if (networkConnectionDatagramAvailable(this, packetLength)) + { + atomic_store_explicit( + &this->inputDatagramBlockedLength, 0, memory_order_release); + return true; + } + + atomic_store_explicit(&this->inputDatagramBlockedLength, + packetLength, memory_order_release); + if (!networkConnectionDatagramAvailable(this, packetLength)) + return false; + + atomic_store_explicit( + &this->inputDatagramBlockedLength, 0, memory_order_release); + return true; +} + +static struct NetworkSend * networkConnectionAllocSend( + NetworkConnection * this, size_t dataSize) +{ + struct NetworkSend * send = malloc(sizeof(*send) + dataSize); + if (!send) + return NULL; + networkConnectionInitSend(send, this, false); + return send; +} + +static struct NetworkSend * networkConnectionAllocInputSend( + NetworkConnection * this, size_t dataSize, + NetworkPacketDelivery delivery) +{ + if (dataSize > NETWORK_INPUT_SEND_POOL_DATA_SIZE) + return NULL; + + LG_LOCK(this->inputSendPoolLock); + struct NetworkSend * send = NULL; + if (delivery == NETWORK_PACKET_DELIVERY_RELIABLE && + this->inputReliableSendFree) + { + send = this->inputReliableSendFree; + this->inputReliableSendFree = send->next; + } + else if (this->inputSendFree) + { + send = this->inputSendFree; + this->inputSendFree = send->next; + } + if (!send) + this->inputSendBlocked |= networkConnectionInputReadyBit(delivery); + LG_UNLOCK(this->inputSendPoolLock); + if (!send) + return NULL; + + networkConnectionInitSend(send, this, true); + send->service = LG_NET_SERVICE_INPUT; + send->delivery = delivery; return send; } @@ -383,11 +474,32 @@ static void networkConnectionFreeSend(struct NetworkSend * send) return; } - NetworkConnection * this = send->connection; + NetworkConnection * this = send->connection; + struct NetworkInputSend * inputSend = (struct NetworkInputSend *)send; + const uint16_t ready = + UINT16_C(1) << LG_NET_SERVICE_INPUT; + uint8_t deliveries; LG_LOCK(this->inputSendPoolLock); - send->next = this->inputSendFree; - this->inputSendFree = send; + if (inputSend->reliableReserved) + { + send->next = this->inputReliableSendFree; + this->inputReliableSendFree = send; + deliveries = this->inputSendBlocked & + NETWORK_INPUT_SEND_READY_RELIABLE; + } + else + { + send->next = this->inputSendFree; + this->inputSendFree = send; + deliveries = this->inputSendBlocked; + } + this->inputSendBlocked &= (uint8_t)~deliveries; LG_UNLOCK(this->inputSendPoolLock); + networkConnectionNotifySendReady(this, + (uint16_t)(deliveries & NETWORK_INPUT_SEND_READY_RELIABLE ? + ready : 0), + (uint16_t)(deliveries & NETWORK_INPUT_SEND_READY_DATAGRAM ? + ready : 0), 0); } static bool networkGuestText(char * destination, size_t capacity, @@ -840,8 +952,7 @@ static bool networkConnectionSendCanResumeNL( (!this->inputStream || !this->inputStreamBound) : !this->controlStream)) || (delivery == NETWORK_PACKET_DELIVERY_DATAGRAM && - (!this->datagramSendEnabled || - blockedBytes > this->maxDatagramSendLength)) || + !networkConnectionDatagramAvailable(this, blockedBytes)) || (delivery == NETWORK_PACKET_DELIVERY_BULK && (!this->outboundBulk[service] || !this->outboundBulk[service]->bound))) @@ -958,10 +1069,10 @@ static LG_TransportStatus networkConnectionSendInputPacket( { const size_t packetSize = LG_NET_ENVELOPE_WIRE_SIZE + payloadLength; - struct NetworkSend * send = networkConnectionAllocSend( - this, LG_NET_SERVICE_INPUT, packetSize); + struct NetworkSend * send = networkConnectionAllocInputSend( + this, packetSize, NETWORK_PACKET_DELIVERY_RELIABLE); if (!send) - return LG_TRANSPORT_ERROR; + return LG_TRANSPORT_UNAVAILABLE; uint8_t * data = networkSendData(send); LGNetPacketParameters parameters; @@ -972,68 +1083,49 @@ static LG_TransportStatus networkConnectionSendInputPacket( parameters.requestID = requestID; LG_LOCK(this->inputStreamLock); - LG_LOCK(this->lock); - HQUIC stream = this->inputStream; + HQUIC stream = this->inputStream; + const bool sessionValid = atomic_load_explicit( + &this->sessionValid, memory_order_acquire); + if (!stream || !sessionValid) + { + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_DISCONNECTED; + } const uint16_t inputMask = UINT16_C(1) << LG_NET_SERVICE_INPUT; const bool serviceReady = (this->receivedServices & inputMask) && (this->reliableServices & inputMask) && this->serviceVersions[LG_NET_SERVICE_INPUT] == serviceVersion && this->inputStreamBound; - if (!stream || - !atomic_load_explicit(&this->sessionValid, memory_order_acquire)) - { - LG_UNLOCK(this->lock); - LG_UNLOCK(this->inputStreamLock); - networkConnectionFreeSend(send); - return LG_TRANSPORT_DISCONNECTED; - } if (!serviceReady || payloadLength > this->serviceMaxPayload[LG_NET_SERVICE_INPUT]) { - LG_UNLOCK(this->lock); LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return LG_TRANSPORT_UNAVAILABLE; } if (this->inputSendSequence == UINT64_MAX) { - LG_UNLOCK(this->lock); LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return LG_TRANSPORT_ERROR; } - const LG_TransportStatus reserveStatus = - networkConnectionReserveSendNL(this, LG_NET_SERVICE_INPUT, - NETWORK_PACKET_DELIVERY_RELIABLE, packetSize); - if (reserveStatus != LG_TRANSPORT_OK) - { - LG_UNLOCK(this->lock); - LG_UNLOCK(this->inputStreamLock); - networkConnectionFreeSend(send); - return reserveStatus; - } - parameters.sessionEpoch = this->sessionEpoch; parameters.sequence = ++this->inputSendSequence; send->service = LG_NET_SERVICE_INPUT; send->length = (uint32_t)packetSize; send->delivery = NETWORK_PACKET_DELIVERY_RELIABLE; - send->accounted = true; - LG_UNLOCK(this->lock); + send->accounted = false; size_t wireSize; if (!lgNetPacketConstruct(data, packetSize, &wireSize, ¶meters, payload, payloadLength) || wireSize != packetSize) { - LG_LOCK(this->lock); --this->inputSendSequence; - const uint16_t ready = networkConnectionReleaseSendNL(this, send); - LG_UNLOCK(this->lock); LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); - networkConnectionNotifySendReady(this, ready, 0, 0); return LG_TRANSPORT_ERROR; } @@ -1041,20 +1133,15 @@ static LG_TransportStatus networkConnectionSendInputPacket( send->buffer.Length = (uint32_t)packetSize; const QUIC_STATUS status = this->api->StreamSend(stream, &send->buffer, 1, QUIC_SEND_FLAG_PRIORITY_WORK, send); - uint16_t ready = 0; if (QUIC_FAILED(status)) { - LG_LOCK(this->lock); --this->inputSendSequence; - ready = networkConnectionReleaseSendNL(this, send); - LG_UNLOCK(this->lock); } LG_UNLOCK(this->inputStreamLock); if (QUIC_FAILED(status)) { networkConnectionFreeSend(send); - networkConnectionNotifySendReady(this, ready, 0, 0); return LG_TRANSPORT_ERROR; } return LG_TRANSPORT_OK; @@ -1096,7 +1183,7 @@ static LG_TransportStatus networkConnectionSendPacket( parameters.requestID = requestID; struct NetworkSend * send = networkConnectionAllocSend( - this, service, packetSize); + this, packetSize); if (!send) return LG_TRANSPORT_ERROR; uint8_t * data = networkSendData(send); @@ -2924,16 +3011,29 @@ static QUIC_STATUS QUIC_API networkConnectionCallback(HQUIC handle, case QUIC_CONNECTION_EVENT_DATAGRAM_STATE_CHANGED: { - uint16_t ready; - LG_LOCK(this->lock); - this->datagramSendEnabled = + const bool sendEnabled = event->DATAGRAM_STATE_CHANGED.SendEnabled != FALSE; - this->maxDatagramSendLength = + const uint16_t maxLength = event->DATAGRAM_STATE_CHANGED.MaxSendLength; - ready = - networkConnectionReadyServicesNL( + const uint32_t sendState = + (sendEnabled ? NETWORK_DATAGRAM_SEND_ENABLED : 0) | maxLength; + uint16_t ready; + LG_LOCK(this->lock); + LG_LOCK(this->datagramSendLock); + atomic_store_explicit( + &this->datagramSendState, sendState, memory_order_release); + LG_UNLOCK(this->datagramSendLock); + ready = networkConnectionReadyServicesNL( this, NETWORK_PACKET_DELIVERY_DATAGRAM); LG_UNLOCK(this->lock); + + uint16_t blockedLength = atomic_load_explicit( + &this->inputDatagramBlockedLength, memory_order_acquire); + if (sendEnabled && blockedLength && maxLength >= blockedLength && + atomic_compare_exchange_strong_explicit( + &this->inputDatagramBlockedLength, &blockedLength, 0, + memory_order_acq_rel, memory_order_acquire)) + ready |= UINT16_C(1) << LG_NET_SERVICE_INPUT; networkConnectionNotifySendReady(this, 0, ready, 0); return QUIC_STATUS_SUCCESS; } @@ -2957,10 +3057,12 @@ static QUIC_STATUS QUIC_API networkConnectionCallback(HQUIC handle, (send->service == LG_NET_SERVICE_INPUT && state == QUIC_DATAGRAM_SEND_SENT)) { - LG_LOCK(this->lock); if (send->accounted) + { + LG_LOCK(this->lock); ready = networkConnectionReleaseSendNL(this, send); - LG_UNLOCK(this->lock); + LG_UNLOCK(this->lock); + } } if (reportLoss && this->datagramLost) this->datagramLost(this->opaque, send->service, @@ -3018,10 +3120,14 @@ static QUIC_STATUS QUIC_API networkStreamCallback(HQUIC handle, case QUIC_STREAM_EVENT_SEND_COMPLETE: { - struct NetworkSend * send = event->SEND_COMPLETE.ClientContext; - LG_LOCK(this->lock); - const uint16_t ready = networkConnectionReleaseSendNL(this, send); - LG_UNLOCK(this->lock); + struct NetworkSend * send = event->SEND_COMPLETE.ClientContext; + uint16_t ready = 0; + if (send->accounted) + { + LG_LOCK(this->lock); + ready = networkConnectionReleaseSendNL(this, send); + LG_UNLOCK(this->lock); + } networkConnectionFreeSend(send); networkConnectionNotifySendReady(this, ready, 0, 0); return QUIC_STATUS_SUCCESS; @@ -3118,8 +3224,8 @@ static QUIC_STATUS QUIC_API networkInputStreamCallback(HQUIC handle, LGNetStreamBinding binding; lgNetStreamBindingInit(&binding, LG_NET_SERVICE_INPUT, LG_NET_ROLE_CLIENT, event->START_COMPLETE.ID, sessionEpoch, 0); - struct NetworkSend * send = networkConnectionAllocSend(this, - LG_NET_SERVICE_INPUT, LG_NET_STREAM_BINDING_WIRE_SIZE); + struct NetworkSend * send = networkConnectionAllocSend( + this, LG_NET_STREAM_BINDING_WIRE_SIZE); uint8_t * data = send ? networkSendData(send) : NULL; if (!send || !lgNetStreamBindingEncode(data, LG_NET_STREAM_BINDING_WIRE_SIZE, &binding)) @@ -3188,10 +3294,14 @@ static QUIC_STATUS QUIC_API networkInputStreamCallback(HQUIC handle, case QUIC_STREAM_EVENT_SEND_COMPLETE: { - struct NetworkSend * send = event->SEND_COMPLETE.ClientContext; - LG_LOCK(this->lock); - const uint16_t ready = networkConnectionReleaseSendNL(this, send); - LG_UNLOCK(this->lock); + struct NetworkSend * send = event->SEND_COMPLETE.ClientContext; + uint16_t ready = 0; + if (send->accounted) + { + LG_LOCK(this->lock); + ready = networkConnectionReleaseSendNL(this, send); + LG_UNLOCK(this->lock); + } networkConnectionFreeSend(send); networkConnectionNotifySendReady(this, ready, 0, 0); return QUIC_STATUS_SUCCESS; @@ -3278,8 +3388,8 @@ static QUIC_STATUS QUIC_API networkBulkStreamCallback(HQUIC handle, event->START_COMPLETE.ID, sessionEpoch, LG_NET_STREAM_BINDING_UNIDIRECTIONAL | LG_NET_STREAM_BINDING_BULK); - struct NetworkSend * send = networkConnectionAllocSend(this, - stream->service, LG_NET_STREAM_BINDING_WIRE_SIZE); + struct NetworkSend * send = networkConnectionAllocSend( + this, LG_NET_STREAM_BINDING_WIRE_SIZE); uint8_t * data = send ? networkSendData(send) : NULL; if (!send || !lgNetStreamBindingEncode(data, LG_NET_STREAM_BINDING_WIRE_SIZE, &binding)) @@ -3411,18 +3521,32 @@ bool networkConnectionCreate(NetworkConnection ** result, } LG_LOCK_INIT(this->lock); + LG_LOCK_INIT(this->datagramSendLock); LG_LOCK_INIT(this->inputStreamLock); LG_LOCK_INIT(this->inputSendPoolLock); LG_LOCK_INIT(this->stateCallbackLock); atomic_init(&this->stopping, false); atomic_init(&this->sessionValid, false); + atomic_init(&this->datagramSendState, 0); + atomic_init(&this->inputDatagramBlockedLength, 0); for (size_t i = 0; i < NETWORK_INPUT_SEND_POOL_COUNT; ++i) { - struct NetworkSend * send = &this->inputSendPool[i].send; - send->connection = this; - send->pooled = true; - send->next = this->inputSendFree; - this->inputSendFree = send; + struct NetworkInputSend * inputSend = &this->inputSendPool[i]; + struct NetworkSend * send = &inputSend->send; + inputSend->reliableReserved = + i < NETWORK_INPUT_SEND_RELIABLE_RESERVE; + send->connection = this; + send->pooled = true; + if (inputSend->reliableReserved) + { + send->next = this->inputReliableSendFree; + this->inputReliableSendFree = send; + } + else + { + send->next = this->inputSendFree; + this->inputSendFree = send; + } } uint16_t offeredBulkStreams = 0; @@ -3509,7 +3633,7 @@ bool networkConnectionCreate(NetworkConnection ** result, const QUIC_REGISTRATION_CONFIG registration = { .AppName = "looking-glass-client", - .ExecutionProfile = QUIC_EXECUTION_PROFILE_LOW_LATENCY, + .ExecutionProfile = QUIC_EXECUTION_PROFILE_TYPE_REAL_TIME, }; status = this->api->RegistrationOpen( ®istration, &this->registration); @@ -3640,6 +3764,7 @@ void networkConnectionDestroy(NetworkConnection ** connection) LG_LOCK_FREE(this->stateCallbackLock); LG_LOCK_FREE(this->inputSendPoolLock); LG_LOCK_FREE(this->inputStreamLock); + LG_LOCK_FREE(this->datagramSendLock); LG_LOCK_FREE(this->lock); free(this->inputReceivePayload); free(this->inputReceiveData); @@ -3670,6 +3795,10 @@ LG_TransportStatus networkConnectionConnect(NetworkConnection * this, lgResetEvent(this->connectEvent); lgResetEvent(this->closeEvent); + LG_LOCK(this->inputSendPoolLock); + this->inputSendBlocked = 0; + LG_UNLOCK(this->inputSendPoolLock); + LG_LOCK(this->lock); this->connectStatus = LG_TRANSPORT_UNAVAILABLE; this->connecting = true; @@ -3696,8 +3825,10 @@ LG_TransportStatus networkConnectionConnect(NetworkConnection * this, this->peerBulkStreams = 0; this->maxPayloadLength = NETWORK_MAX_NEGOTIATED_PAYLOAD; this->maxDatagramPayload = 0; - this->maxDatagramSendLength = 0; - this->datagramSendEnabled = false; + atomic_store_explicit( + &this->datagramSendState, 0, memory_order_release); + atomic_store_explicit( + &this->inputDatagramBlockedLength, 0, memory_order_release); this->inputStreamBound = false; networkConnectionResetSendAccountingNL(this); networkConnectionClearAuthProofNL(this); @@ -3893,6 +4024,23 @@ bool networkConnectionServiceVersion(NetworkConnection * this, return available; } +bool networkConnectionAwaitInputSend(NetworkConnection * this, + NetworkPacketDelivery delivery) +{ + const uint8_t readyBit = networkConnectionInputReadyBit(delivery); + if (!this || !readyBit) + return false; + + LG_LOCK(this->inputSendPoolLock); + const bool available = this->inputSendFree != NULL || + (delivery == NETWORK_PACKET_DELIVERY_RELIABLE && + this->inputReliableSendFree != NULL); + if (!available) + this->inputSendBlocked |= readyBit; + LG_UNLOCK(this->inputSendPoolLock); + return available; +} + LG_TransportStatus networkConnectionSend(NetworkConnection * this, LGNetService service, uint16_t messageType, uint16_t serviceVersion, LGNetEnvelopeFlags flags, uint64_t componentEpoch, uint64_t requestID, @@ -3906,6 +4054,107 @@ LG_TransportStatus networkConnectionSend(NetworkConnection * this, payload, payloadLength, false); } +static LG_TransportStatus networkConnectionSendInputDatagram( + NetworkConnection * this, uint16_t messageType, + uint16_t serviceVersion, LGNetEnvelopeFlags flags, + uint64_t componentEpoch, uint64_t requestID, + const void * payload, uint32_t payloadLength, + uint64_t applicationSequence, bool discardIfBlocked) +{ + const size_t packetSize = LG_NET_ENVELOPE_WIRE_SIZE + + payloadLength; + const uint16_t packetLength = (uint16_t)packetSize; + struct NetworkSend * send = networkConnectionAllocInputSend( + this, packetSize, NETWORK_PACKET_DELIVERY_DATAGRAM); + if (!send) + return LG_TRANSPORT_UNAVAILABLE; + uint8_t * data = networkSendData(send); + + LGNetPacketParameters parameters; + lgNetPacketParametersInit(¶meters, LG_NET_SERVICE_INPUT, + messageType, serviceVersion, LG_NET_MESSAGE_VERSION_CURRENT); + parameters.flags = flags | LG_NET_ENVELOPE_DATAGRAM | + LG_NET_ENVELOPE_FINAL; + parameters.componentEpoch = componentEpoch; + parameters.requestID = requestID; + + LG_LOCK(this->inputStreamLock); + const uint16_t serviceMask = UINT16_C(1) << LG_NET_SERVICE_INPUT; + HQUIC connection = this->connection; + if (!connection || + !atomic_load_explicit(&this->sessionValid, memory_order_acquire)) + { + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_DISCONNECTED; + } + if (!(this->receivedServices & serviceMask) || + !(this->datagramServices & serviceMask) || + this->serviceVersions[LG_NET_SERVICE_INPUT] != serviceVersion || + payloadLength > this->serviceMaxDatagram[LG_NET_SERVICE_INPUT]) + { + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_UNAVAILABLE; + } + + LG_LOCK(this->datagramSendLock); + if (!networkConnectionPrepareInputDatagram(this, packetLength)) + { + LG_UNLOCK(this->datagramSendLock); + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_UNAVAILABLE; + } + + uint64_t sequence; + if (!networkConnectionNextDatagramSequence(this, &sequence)) + { + LG_UNLOCK(this->datagramSendLock); + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_ERROR; + } + + parameters.sessionEpoch = this->sessionEpoch; + parameters.sequence = sequence; + send->length = (uint32_t)packetSize; + send->componentEpoch = componentEpoch; + send->applicationSequence = applicationSequence; + send->accounted = false; + + size_t wireSize; + if (!lgNetPacketConstruct(data, packetSize, &wireSize, + ¶meters, payload, payloadLength) || wireSize != packetSize) + { + --this->datagramSendSequence; + LG_UNLOCK(this->datagramSendLock); + LG_UNLOCK(this->inputStreamLock); + networkConnectionFreeSend(send); + return LG_TRANSPORT_ERROR; + } + + send->buffer.Buffer = data; + send->buffer.Length = (uint32_t)packetSize; + QUIC_SEND_FLAGS sendFlags = QUIC_SEND_FLAG_PRIORITY_WORK | + QUIC_SEND_FLAG_DGRAM_PRIORITY; + if (discardIfBlocked) + sendFlags |= QUIC_SEND_FLAG_CANCEL_ON_BLOCKED; + const QUIC_STATUS status = this->api->DatagramSend(connection, + &send->buffer, 1, sendFlags, send); + if (QUIC_FAILED(status)) + --this->datagramSendSequence; + LG_UNLOCK(this->datagramSendLock); + LG_UNLOCK(this->inputStreamLock); + + if (QUIC_FAILED(status)) + { + networkConnectionFreeSend(send); + return LG_TRANSPORT_ERROR; + } + return LG_TRANSPORT_OK; +} + LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, LGNetService service, uint16_t messageType, uint16_t serviceVersion, LGNetEnvelopeFlags flags, uint64_t componentEpoch, uint64_t requestID, @@ -3928,6 +4177,11 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, if (payloadLength > LG_NET_MAX_PAYLOAD_LENGTH) return LG_TRANSPORT_ERROR; + if (service == LG_NET_SERVICE_INPUT) + return networkConnectionSendInputDatagram(this, messageType, + serviceVersion, flags, componentEpoch, requestID, + payload, payloadLength, applicationSequence, discardIfBlocked); + const size_t packetSize = LG_NET_ENVELOPE_WIRE_SIZE + payloadLength; LGNetPacketParameters parameters; @@ -3939,14 +4193,11 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, parameters.requestID = requestID; struct NetworkSend * send = networkConnectionAllocSend( - this, service, packetSize); + this, packetSize); if (!send) return LG_TRANSPORT_ERROR; - uint8_t * data = networkSendData(send); - const bool inputDatagram = service == LG_NET_SERVICE_INPUT; + uint8_t * data = networkSendData(send); - if (inputDatagram) - LG_LOCK(this->inputStreamLock); LG_LOCK(this->lock); const uint16_t serviceMask = UINT16_C(1) << service; HQUIC connection = this->connection; @@ -3954,8 +4205,6 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, !atomic_load_explicit(&this->sessionValid, memory_order_acquire)) { LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return LG_TRANSPORT_DISCONNECTED; } @@ -3965,21 +4214,10 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, payloadLength > this->serviceMaxDatagram[service]) { LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return LG_TRANSPORT_UNAVAILABLE; } - if (this->datagramSendSequence == UINT64_MAX) - { - LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); - networkConnectionFreeSend(send); - return LG_TRANSPORT_ERROR; - } - if (!this->datagramSendEnabled || - packetSize > this->maxDatagramSendLength) + if (!networkConnectionDatagramAvailable(this, packetSize)) { const struct NetworkSendLimits limits = networkConnectionSendLimits( @@ -3993,8 +4231,6 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, this->datagramSend.serviceBlockedBytes[service] : packetSize; } LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return LG_TRANSPORT_UNAVAILABLE; } @@ -4005,35 +4241,39 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, if (reserveStatus != LG_TRANSPORT_OK) { LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); return reserveStatus; } - parameters.sessionEpoch = this->sessionEpoch; - parameters.sequence = ++this->datagramSendSequence; - send->service = service; send->length = (uint32_t)packetSize; send->componentEpoch = componentEpoch; send->applicationSequence = applicationSequence; send->delivery = NETWORK_PACKET_DELIVERY_DATAGRAM; send->accounted = true; - if (inputDatagram) + + LG_LOCK(this->datagramSendLock); + uint64_t sequence; + if (!networkConnectionNextDatagramSequence(this, &sequence)) + { + const uint16_t ready = networkConnectionReleaseSendNL(this, send); + LG_UNLOCK(this->datagramSendLock); LG_UNLOCK(this->lock); + networkConnectionFreeSend(send); + networkConnectionNotifySendReady(this, 0, ready, 0); + return LG_TRANSPORT_ERROR; + } + + parameters.sessionEpoch = this->sessionEpoch; + parameters.sequence = sequence; size_t wireSize; if (!lgNetPacketConstruct(data, packetSize, &wireSize, ¶meters, payload, payloadLength) || wireSize != packetSize) { - if (inputDatagram) - LG_LOCK(this->lock); - if (!inputDatagram) - --this->datagramSendSequence; + --this->datagramSendSequence; const uint16_t ready = networkConnectionReleaseSendNL(this, send); + LG_UNLOCK(this->datagramSendLock); LG_UNLOCK(this->lock); - if (inputDatagram) - LG_UNLOCK(this->inputStreamLock); networkConnectionFreeSend(send); networkConnectionNotifySendReady(this, 0, ready, 0); return LG_TRANSPORT_ERROR; @@ -4042,8 +4282,6 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, send->buffer.Buffer = data; send->buffer.Length = (uint32_t)packetSize; QUIC_SEND_FLAGS sendFlags = QUIC_SEND_FLAG_PRIORITY_WORK; - if (service == LG_NET_SERVICE_INPUT) - sendFlags |= QUIC_SEND_FLAG_DGRAM_PRIORITY; if (discardIfBlocked) sendFlags |= QUIC_SEND_FLAG_CANCEL_ON_BLOCKED; const QUIC_STATUS status = this->api->DatagramSend(connection, @@ -4051,22 +4289,11 @@ LG_TransportStatus networkConnectionSendDatagram(NetworkConnection * this, uint16_t ready = 0; if (QUIC_FAILED(status)) { - if (inputDatagram) - LG_LOCK(this->lock); - if (!inputDatagram) - --this->datagramSendSequence; + --this->datagramSendSequence; ready = networkConnectionReleaseSendNL(this, send); - if (inputDatagram) - LG_UNLOCK(this->lock); - } - if (inputDatagram) - { - LG_UNLOCK(this->inputStreamLock); - } - else - { - LG_UNLOCK(this->lock); } + LG_UNLOCK(this->datagramSendLock); + LG_UNLOCK(this->lock); if (QUIC_FAILED(status)) { @@ -4166,7 +4393,7 @@ retry: } struct NetworkSend * send = networkConnectionAllocSend( - this, service, packetSize); + this, packetSize); if (!send) { struct NetworkSend rollback = diff --git a/client/transports/Network/connection.h b/client/transports/Network/connection.h index e041f68f..f42ac9ab 100644 --- a/client/transports/Network/connection.h +++ b/client/transports/Network/connection.h @@ -86,6 +86,8 @@ bool networkConnectionServiceAvailable(NetworkConnection * connection, LGNetService service, LGNetCapabilityFlags required); bool networkConnectionServiceVersion(NetworkConnection * connection, LGNetService service, uint16_t * version); +bool networkConnectionAwaitInputSend(NetworkConnection * connection, + NetworkPacketDelivery delivery); LG_TransportStatus networkConnectionSend(NetworkConnection * connection, LGNetService service, uint16_t messageType, uint16_t serviceVersion, diff --git a/client/transports/Network/input.c b/client/transports/Network/input.c index 38085ce3..fa14f0a5 100644 --- a/client/transports/Network/input.c +++ b/client/transports/Network/input.c @@ -458,22 +458,21 @@ static bool networkInputQueueReliableNL(NetworkInput * input, static LG_TransportStatus networkInputDrainReliableNL(NetworkInput * input) { - while (input->reliableCount) - { - const NetworkInputPkt * packet = - &input->reliableQueue[input->reliableHead]; - const LG_TransportStatus status = networkInputSendReliableNL(input, - packet->messageType, packet->flags, packet->componentEpoch, - packet->requestID, packet->payload, packet->payloadLength); - if (status != LG_TRANSPORT_OK) - return status; + if (!input->reliableCount) + return LG_TRANSPORT_OK; - input->reliableHead = - (input->reliableHead + 1U) % NETWORK_INPUT_QUEUE_CAPACITY; - --input->reliableCount; - } + const NetworkInputPkt * packet = + &input->reliableQueue[input->reliableHead]; + const LG_TransportStatus status = networkInputSendReliableNL(input, + packet->messageType, packet->flags, packet->componentEpoch, + packet->requestID, packet->payload, packet->payloadLength); + if (status != LG_TRANSPORT_OK) + return status; - input->reliableHead = 0; + input->reliableHead = + (input->reliableHead + 1U) % NETWORK_INPUT_QUEUE_CAPACITY; + if (!--input->reliableCount) + input->reliableHead = 0; return LG_TRANSPORT_OK; } @@ -792,6 +791,8 @@ static bool networkInputTimer(void * opaque) } networkInputSyncKeyboardLEDsNL(input, ¬ify); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); } LG_UNLOCK(input->stateLock); @@ -905,6 +906,8 @@ static bool networkInputUpdateKey(void * opaque, int key, bool down) } if (!down && networkInputTransportFatal(status)) notify = networkInputAbandonClaimNL(input, true); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); if (notify) networkInputNotifyStatus(input); @@ -923,7 +926,11 @@ static bool networkInputUpdateKey(void * opaque, int key, bool down) NETWORK_INPUT_LED_SYNC_DELAY_MS); } } + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); + if (notify) + networkInputNotifyStatus(input); return true; } @@ -959,6 +966,8 @@ static bool networkInputKeyboardLEDs(void * opaque, bool numLock, if (input->keyboardLEDsSyncAfter < syncAfter) input->keyboardLEDsSyncAfter = syncAfter; networkInputSyncKeyboardLEDsNL(input, ¬ify); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); const bool available = input->available; LG_UNLOCK(input->stateLock); if (notify) @@ -1213,32 +1222,51 @@ static LG_TransportStatus networkInputSendAbsoluteNL(NetworkInput * input, static void networkInputResumeSendsNL(NetworkInput * input, bool * notify) { - input->sendReadyPending = false; - if (!input->connected || !input->serviceAvailable || input->destroying) - return; - - const LG_TransportStatus reliableStatus = - networkInputDrainReliableNL(input); - if (networkInputTransportFatal(reliableStatus)) + do { - *notify |= networkInputAbandonClaimNL(input, true); - return; - } - if (reliableStatus != LG_TRANSPORT_OK) - return; + input->sendReadyPending = false; + if (!input->connected || !input->serviceAvailable || input->destroying) + return; - if (!input->claimPending && !input->claimed) - { - networkInputSendClaimNL(input); - return; - } - if (!input->claimed || - (!input->pendingRelative && !input->pendingAbsolute)) - return; + const LG_TransportStatus reliableStatus = + networkInputDrainReliableNL(input); + if (networkInputTransportFatal(reliableStatus)) + { + *notify |= networkInputAbandonClaimNL(input, true); + return; + } + if (reliableStatus != LG_TRANSPORT_OK) + return; + if (input->reliableCount) + { + if (input->sendReadyPending) + continue; + if (networkConnectionAwaitInputSend(input->connection, + NETWORK_PACKET_DELIVERY_RELIABLE)) + { + input->sendReadyPending = true; + continue; + } + return; + } - const LG_TransportStatus status = networkInputFlushMotionNL(input); - if (networkInputTransportFatal(status)) - *notify |= networkInputAbandonClaimNL(input, true); + if (!input->claimPending && !input->claimed) + { + networkInputSendClaimNL(input); + continue; + } + if (!input->claimed || + (!input->pendingRelative && !input->pendingAbsolute)) + continue; + + const LG_TransportStatus status = networkInputFlushMotionNL(input); + if (networkInputTransportFatal(status)) + { + *notify |= networkInputAbandonClaimNL(input, true); + return; + } + } + while (input->sendReadyPending); } void networkInputSendReady(NetworkInput * input, @@ -1307,9 +1335,15 @@ static bool networkInputMouseMotion(void * opaque, int32_t x, int32_t y) if (eligible) status = networkInputSendRelativeNL(input, x, y, 0, 0, input->mouseButtons, 0, true); - const bool result = eligible && + bool result = eligible && !networkInputTransportFatal(status); + bool notify = false; + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); + result &= input->available; LG_UNLOCK(input->stateLock); + if (notify) + networkInputNotifyStatus(input); return result; } @@ -1330,9 +1364,15 @@ static bool networkInputMousePosition(void * opaque, uint32_t x, if (eligible) status = networkInputSendAbsoluteNL(input, x, y, width, height, 0, 0, input->mouseButtons, 0, true); - const bool result = eligible && + bool result = eligible && !networkInputTransportFatal(status); + bool notify = false; + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); + result &= input->available; LG_UNLOCK(input->stateLock); + if (notify) + networkInputNotifyStatus(input); return result; } @@ -1411,6 +1451,8 @@ static bool networkInputUpdateButton(void * opaque, input->mouseButtons = buttons; else if (networkInputTransportFatal(status) && !down && mask) notify = networkInputAbandonClaimNL(input, true); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); if (notify) networkInputNotifyStatus(input); @@ -1433,6 +1475,8 @@ static void networkInputReset(void * opaque) bool notify = false; LG_LOCK(input->stateLock); notify = networkInputSendResetNL(input); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); if (notify) networkInputNotifyStatus(input); @@ -1574,6 +1618,8 @@ void networkInputConnectionState(NetworkInput * input, bool connected) notify |= networkInputSetAvailableNL(input, false); networkInputSendClaimNL(input); } + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); if (notify) @@ -1661,6 +1707,8 @@ static bool networkInputHandleStatus(NetworkInput * input, input->reliableBarrierSequence = 0; networkInputResumeSendsNL(input, ¬ify); } + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); LG_UNLOCK(input->stateLock); if (notify) @@ -1703,6 +1751,8 @@ static bool networkInputHandleLEDs(NetworkInput * input, NETWORK_INPUT_LED_MASK; networkInputAcknowledgeLEDsNL(input, changed); networkInputSyncKeyboardLEDsNL(input, ¬ify); + if (input->sendReadyPending) + networkInputResumeSendsNL(input, ¬ify); notify |= oldValid != input->keyboardLEDsValid || oldState != input->keyboardLEDs; if (notify) diff --git a/idd/LGCommon/CPipeEndpoint.cpp b/idd/LGCommon/CPipeEndpoint.cpp index 4c799da8..be70a5fd 100644 --- a/idd/LGCommon/CPipeEndpoint.cpp +++ b/idd/LGCommon/CPipeEndpoint.cpp @@ -38,7 +38,7 @@ const DWORD CPipeEndpoint::CLIENT_RETRY_MAX_MS = 2000; const DWORD CPipeEndpoint::SERVER_RETRY_MS = 1000; const DWORD CPipeEndpoint::AUTHENTICATION_TIMEOUT_MS = 2000; const DWORD CPipeEndpoint::AUTHORIZATION_POLL_MS = 1000; -const DWORD CPipeEndpoint::WRITE_TIMEOUT_MS = 250; +const DWORD CPipeEndpoint::DEFAULT_WRITE_TIMEOUT_MS = 250; bool CPipeEndpoint::IsDisconnectedError(DWORD error) { @@ -159,7 +159,7 @@ CPipeEndpoint::PipeIoResult CPipeEndpoint::WriteMessage( m_writeEvent, &overlapped, &bytesWritten, - WRITE_TIMEOUT_MS); + m_writeTimeoutMs); else if (IsDisconnectedError(error)) result = PipeIoResult::Disconnected; else if (error == ERROR_OPERATION_ABORTED && @@ -199,7 +199,8 @@ bool CPipeEndpoint::Start( const SECURITY_ATTRIBUTES * serverSecurity, DWORD serverOpenMode, DWORD serverPipeMode, - ThreadScheduling threadScheduling) + ThreadScheduling threadScheduling, + DWORD writeTimeoutMs) { Stop(); @@ -215,6 +216,7 @@ bool CPipeEndpoint::Start( SECURITY_ATTRIBUTES {}; m_serverOpenMode = serverOpenMode; m_serverPipeMode = serverPipeMode; + m_writeTimeoutMs = writeTimeoutMs; m_stopEvent = CreateEventW(nullptr, TRUE, FALSE, nullptr); if (!m_stopEvent) { diff --git a/idd/LGCommon/CPipeEndpoint.h b/idd/LGCommon/CPipeEndpoint.h index 57c799d9..66c6b55d 100644 --- a/idd/LGCommon/CPipeEndpoint.h +++ b/idd/LGCommon/CPipeEndpoint.h @@ -74,6 +74,8 @@ public: class CPipeEndpoint { public: + static const DWORD DEFAULT_WRITE_TIMEOUT_MS; + enum class Mode { Server, @@ -99,7 +101,8 @@ public: _In_opt_ const SECURITY_ATTRIBUTES * serverSecurity = nullptr, _In_ DWORD serverOpenMode = 0, _In_ DWORD serverPipeMode = 0, - _In_ ThreadScheduling threadScheduling = ThreadScheduling::Normal); + _In_ ThreadScheduling threadScheduling = ThreadScheduling::Normal, + _In_ DWORD writeTimeoutMs = DEFAULT_WRITE_TIMEOUT_MS); void Stop(); void DisconnectClient(); @@ -132,7 +135,6 @@ private: static const DWORD SERVER_RETRY_MS; static const DWORD AUTHENTICATION_TIMEOUT_MS; static const DWORD AUTHORIZATION_POLL_MS; - static const DWORD WRITE_TIMEOUT_MS; static bool IsDisconnectedError(_In_ DWORD error); PipeIoResult WaitForOverlapped( @@ -173,6 +175,7 @@ private: bool m_hasServerSecurity = false; DWORD m_serverOpenMode = 0; DWORD m_serverPipeMode = 0; + DWORD m_writeTimeoutMs = DEFAULT_WRITE_TIMEOUT_MS; std::atomic m_running { false }; std::atomic m_connected { false }; diff --git a/idd/LGIdd/ipc/CInputPipeServer.cpp b/idd/LGIdd/ipc/CInputPipeServer.cpp index 0a0f6d51..ffdd18e3 100644 --- a/idd/LGIdd/ipc/CInputPipeServer.cpp +++ b/idd/LGIdd/ipc/CInputPipeServer.cpp @@ -76,7 +76,12 @@ bool CInputPipeServer::Init() if (!m_endpoint.Start( LG_INPUT_PIPE_NAME, CPipeEndpoint::Mode::Server, - sizeof(LGInputPipeMessage))) + sizeof(LGInputPipeMessage), + nullptr, + 0, + 0, + CPipeEndpoint::ThreadScheduling::Normal, + PIPE_WRITE_TIMEOUT_MS)) { DeInit(); return false; @@ -193,6 +198,8 @@ bool CInputPipeServer::CompactQueuedMotionLocked() first.payload.mouseRelative.deltaX = static_cast(x); first.payload.mouseRelative.deltaY = static_cast(y); + if (second.enqueueTicks) + first.enqueueTicks = second.enqueueTicks; RemoveQueueItemLocked(i + 1); ++m_statRelativeCoalesced; return true; @@ -207,6 +214,8 @@ bool CInputPipeServer::CompactQueuedMotionLocked() first.payload.mouseAbsolute.x = second.payload.mouseAbsolute.x; first.payload.mouseAbsolute.y = second.payload.mouseAbsolute.y; + if (second.enqueueTicks) + first.enqueueTicks = second.enqueueTicks; RemoveQueueItemLocked(i + 1); ++m_statAbsoluteCoalesced; return true; @@ -253,6 +262,9 @@ bool CInputPipeServer::QueueLocked( { tail.payload.mouseRelative.deltaX = static_cast(x); tail.payload.mouseRelative.deltaY = static_cast(y); + LARGE_INTEGER enqueueTime = {}; + if (QueryPerformanceCounter(&enqueueTime)) + tail.enqueueTicks = enqueueTime.QuadPart; ++m_statRelativeCoalesced; return true; } @@ -266,6 +278,9 @@ bool CInputPipeServer::QueueLocked( { tail.payload.mouseAbsolute.x = payload.mouseAbsolute.x; tail.payload.mouseAbsolute.y = payload.mouseAbsolute.y; + LARGE_INTEGER enqueueTime = {}; + if (QueryPerformanceCounter(&enqueueTime)) + tail.enqueueTicks = enqueueTime.QuadPart; ++m_statAbsoluteCoalesced; return true; } @@ -503,7 +518,14 @@ bool CInputPipeServer::Pop(QueueItem& item) if (!m_queueCount) return false; - item = m_queue[m_queueHead]; + // A blocked pipe write can leave a run of stale motion queued behind it. + // Collapse that run before the next write without combining fresh motion. + if (MotionBudgetExceededLocked()) + while (CompactQueuedMotionLocked()) + { + } + + item = m_queue[m_queueHead]; m_queueHead = (m_queueHead + 1) % QUEUE_LENGTH; --m_queueCount; if (m_queueCount) diff --git a/idd/LGIdd/ipc/CInputPipeServer.h b/idd/LGIdd/ipc/CInputPipeServer.h index eb2d44c0..c0475374 100644 --- a/idd/LGIdd/ipc/CInputPipeServer.h +++ b/idd/LGIdd/ipc/CInputPipeServer.h @@ -35,6 +35,7 @@ private: static constexpr size_t QUEUE_LENGTH = 128; static constexpr size_t MOTION_EMERGENCY_DEPTH = 8; static constexpr DWORD MOTION_AGE_BUDGET_MS = 1; + static constexpr DWORD PIPE_WRITE_TIMEOUT_MS = 4; static constexpr DWORD STATISTICS_INTERVAL_MS = 5000; struct QueueItem diff --git a/idd/LGIdd/transport/network/CNetworkServer.cpp b/idd/LGIdd/transport/network/CNetworkServer.cpp index 8f57edae..a951f067 100644 --- a/idd/LGIdd/transport/network/CNetworkServer.cpp +++ b/idd/LGIdd/transport/network/CNetworkServer.cpp @@ -385,7 +385,6 @@ struct CNetworkServer::Impl uint32_t length = 0; Delivery delivery = Delivery::RELIABLE; bool accounted = true; - bool inputAck = false; }; struct SendLimits @@ -398,6 +397,8 @@ struct CNetworkServer::Impl { uint64_t claimEpoch = 0; uint64_t sequence = 0; + int64_t previousX = 0; + int64_t previousY = 0; int64_t cumulativeX = 0; int64_t cumulativeY = 0; uint16_t absoluteX = 0; @@ -407,6 +408,18 @@ struct CNetworkServer::Impl bool useAbsolute = false; }; + struct InputSend + { + static constexpr size_t CAPACITY = + LG_NET_ENVELOPE_WIRE_SIZE + NETWORK_INPUT_MAX_PAYLOAD; + + QUIC_BUFFER buffer = {}; + uint8_t data[CAPACITY] = {}; + uint32_t length = 0; + bool active = false; + bool inputAck = false; + }; + struct SendAccounting { uint32_t count; @@ -448,6 +461,13 @@ struct CNetworkServer::Impl FAILED, }; + enum class InputPrepareResult + { + READY, + STALE, + INVALID, + }; + enum class HandshakeState { WAITING_HELLO, @@ -485,7 +505,6 @@ struct CNetworkServer::Impl Impl& owner; CSRWLock lock; CSRWLock inputLock; - CSRWLock inputDispatchLock; CSRWLock inputSendLock; HQUIC handle = nullptr; HQUIC controlStream = nullptr; @@ -539,6 +558,7 @@ struct CNetworkServer::Impl uint32_t inputDatagramLimit = 0; uint32_t inputStreamSendCount = 0; uint64_t inputStreamSendBytes = 0; + uint32_t inputSendNext = 0; uint16_t inputStreamServiceVersion = 0; bool inputStreamBound = false; bool inputClosing = false; @@ -550,6 +570,7 @@ struct CNetworkServer::Impl std::vector inputReceiveData; std::vector inputReceivePayload; std::shared_ptr inputSession; + InputSend inputSends[NETWORK_INPUT_SEND_COUNT_MAX] = {}; LGNetSequenceTracker inputDatagramReceive = {}; uint64_t inputClaimantID = 0; @@ -1619,19 +1640,44 @@ struct CNetworkServer::Impl video.SendReady(VideoPeer(connection), videoDelivery); } - static void ReleaseInputSendLocked(Connection& connection, - const Send& send) + static InputSend * AcquireInputSendLocked(Connection& connection) { - if (!connection.inputStreamSendCount || + for (uint32_t offset = 0; offset < NETWORK_INPUT_SEND_COUNT_MAX; + ++offset) + { + const uint32_t index = + (connection.inputSendNext + offset) % NETWORK_INPUT_SEND_COUNT_MAX; + InputSend& send = connection.inputSends[index]; + if (send.active) + continue; + + send.active = true; + connection.inputSendNext = + (index + 1U) % NETWORK_INPUT_SEND_COUNT_MAX; + return &send; + } + + return nullptr; + } + + static void ReleaseInputSendLocked(Connection& connection, + InputSend& send) + { + if (!send.active || !connection.inputStreamSendCount || connection.inputStreamSendBytes < send.length) { DEBUG_ERROR("Network input send accounting underflow for client %u", connection.client); - return; } - - --connection.inputStreamSendCount; - connection.inputStreamSendBytes -= send.length; + else + { + --connection.inputStreamSendCount; + connection.inputStreamSendBytes -= send.length; + } + send.buffer = {}; + send.length = 0; + send.active = false; + send.inputAck = false; } SendResult TrySendInputPacket(Connection& connection, @@ -1649,26 +1695,13 @@ struct CNetworkServer::Impl LG_NET_ENVELOPE_EARLY_DATA)) return SendResult::FAILED; - const size_t packetSize = + const size_t packetSize = LG_NET_ENVELOPE_WIRE_SIZE + payloadLength; - std::unique_ptr send(new (std::nothrow) Send()); - if (!send) + if (packetSize > InputSend::CAPACITY) return SendResult::FAILED; - try - { - send->data.resize(packetSize); - } - catch (...) - { - return SendResult::FAILED; - } - send->service = LG_NET_SERVICE_INPUT; - send->length = static_cast(packetSize); - send->accounted = false; - send->inputAck = inputAck; - - QUIC_STATUS status = QUIC_STATUS_SUCCESS; + InputSend * send = nullptr; + QUIC_STATUS status = QUIC_STATUS_SUCCESS; { LGNetPacketParameters parameters = {}; HQUIC stream = nullptr; @@ -1687,6 +1720,12 @@ struct CNetworkServer::Impl NETWORK_INPUT_SEND_BYTES_MAX - packetSize) return SendResult::BLOCKED; + send = AcquireInputSendLocked(connection); + if (!send) + return SendResult::BLOCKED; + + send->length = static_cast(packetSize); + send->inputAck = inputAck; ++connection.inputStreamSendCount; connection.inputStreamSendBytes += packetSize; @@ -1702,7 +1741,7 @@ struct CNetworkServer::Impl } size_t wireSize = 0; - if (!lgNetPacketConstruct(send->data.data(), send->data.size(), + if (!lgNetPacketConstruct(send->data, sizeof(send->data), &wireSize, ¶meters, payload, payloadLength) || wireSize != packetSize) { @@ -1713,11 +1752,10 @@ struct CNetworkServer::Impl return SendResult::FAILED; } - send->buffers[0].Buffer = send->data.data(); - send->buffers[0].Length = static_cast(wireSize); - status = api->StreamSend(stream, - send->buffers, send->bufferCount, QUIC_SEND_FLAG_PRIORITY_WORK, - send.get()); + send->buffer.Buffer = send->data; + send->buffer.Length = static_cast(wireSize); + status = api->StreamSend(stream, + &send->buffer, 1, QUIC_SEND_FLAG_PRIORITY_WORK, send); if (QUIC_FAILED(status)) { CSRWExclusiveLock guard(connection.inputLock); @@ -1729,7 +1767,6 @@ struct CNetworkServer::Impl if (QUIC_FAILED(status)) return SendResult::FAILED; - send.release(); return SendResult::SENT; } @@ -3732,7 +3769,7 @@ retry: uint64_t claimantID, LGNetInputClaimFlags requiredFlags) { CSRWExclusiveLock guard(connection.inputLock); - uint64_t& lastSequence = datagram ? + uint64_t& lastSequence = datagram ? connection.inputDatagramSequence : connection.inputReliableSequence; if (!connection.inputClaimantID || @@ -3776,33 +3813,73 @@ retry: return LG_INPUT_MOUSE_DELTA_MIN; } - bool PrepareInputPointer(Connection& connection, uint64_t claimEpoch, - uint64_t sequence, bool datagram, bool relative, + InputPrepareResult PrepareInputPointer(Connection& connection, + const LGNetPacketView& packet, uint64_t sequence, bool datagram, + bool relative, int32_t wheel, LGNetPointerButtons changedButtons, int64_t cumulativeX, int64_t cumulativeY, - LGNetPointerButtons& buttons, InputPointerMotion& motion) + LGNetPointerButtons& buttons, InputPointerMotion& motion, + std::shared_ptr& session) { + const bool acknowledge = + (packet.envelope.flags & LG_NET_ENVELOPE_ACK_REQUIRED) != 0; + if (!sequence) + return InputPrepareResult::INVALID; + if (datagram) + { + if (wheel || changedButtons || acknowledge || + packet.envelope.requestID) + return InputPrepareResult::INVALID; + } + else if (acknowledge ? packet.envelope.requestID != sequence : + packet.envelope.requestID != 0) + return InputPrepareResult::INVALID; + CSRWExclusiveLock guard(connection.inputLock); - uint64_t& lastSequence = datagram ? + uint64_t& lastSequence = datagram ? connection.inputDatagramSequence : connection.inputReliableSequence; + session = connection.inputSession; + if (!session) + return InputPrepareResult::INVALID; if (!connection.inputClaimantID || - connection.inputClaimEpoch != claimEpoch || + connection.inputClaimEpoch != packet.envelope.componentEpoch || !(connection.inputFlags & LG_NET_INPUT_CLAIM_POINTER) || sequence <= lastSequence) - return false; + return InputPrepareResult::STALE; + + const bool motionApplies = + sequence > connection.inputMotionSequence; + const bool baseline = connection.inputMotionBaseline; + const bool modeChange = motionApplies && baseline && + connection.inputMotionAbsolute == relative; + const bool buttonChange = buttons != connection.inputPointerButtons; + if (datagram) + { + if (motionApplies && (!baseline || modeChange)) + return InputPrepareResult::INVALID; + } + else + { + const bool transition = wheel || changedButtons || buttonChange || + (motionApplies && (!baseline || modeChange)); + if (transition && !acknowledge) + return InputPrepareResult::INVALID; + } lastSequence = sequence; if (!datagram) connection.inputMotionBaseline = true; - motion.claimEpoch = claimEpoch; + motion.claimEpoch = packet.envelope.componentEpoch; motion.sequence = sequence; + motion.previousX = connection.inputCumulativeX; + motion.previousY = connection.inputCumulativeY; motion.previousButtons = connection.inputPointerButtons; if (datagram) buttons = connection.inputPointerButtons; else connection.inputPointerButtons = buttons; - motion.apply = sequence > connection.inputMotionSequence; + motion.apply = motionApplies; if (motion.apply) { connection.inputMotionSequence = sequence; @@ -3827,7 +3904,7 @@ retry: motion.absoluteX = connection.inputAbsoluteX; motion.absoluteY = connection.inputAbsoluteY; } - return true; + return InputPrepareResult::READY; } InputResult SendInputRelativeMotion(Connection& connection, @@ -3835,26 +3912,22 @@ retry: const InputPointerMotion& motion, int32_t wheel, LGNetPointerButtons buttons) { - int64_t cumulativeX = 0; - int64_t cumulativeY = 0; - { - CSRWSharedLock guard(connection.inputLock); - cumulativeX = connection.inputCumulativeX; - cumulativeY = connection.inputCumulativeY; - } + int64_t cumulativeX = motion.previousX; + int64_t cumulativeY = motion.previousY; InputResult result = InputResult::STALE; for (uint32_t chunk = 0; chunk < NETWORK_INPUT_RECOVERY_LIMIT; ++chunk) { - int64_t nextX = cumulativeX; - int64_t nextY = cumulativeY; - const int32_t deltaX = InputCumulativeStep( + int64_t nextX = cumulativeX; + int64_t nextY = cumulativeY; + const int32_t deltaX = InputCumulativeStep( motion.cumulativeX, cumulativeX, nextX); - const int32_t deltaY = InputCumulativeStep( + const int32_t deltaY = InputCumulativeStep( motion.cumulativeY, cumulativeY, nextY); - const bool final = - (nextX == motion.cumulativeX && nextY == motion.cumulativeY) || - chunk + 1U == NETWORK_INPUT_RECOVERY_LIMIT; + const bool reached = + nextX == motion.cumulativeX && nextY == motion.cumulativeY; + const bool final = + reached || chunk + 1U == NETWORK_INPUT_RECOVERY_LIMIT; result = session->SendMouseRelative(deltaX, deltaY, final ? wheel : 0, final ? buttons : motion.previousButtons); @@ -3864,7 +3937,14 @@ retry: cumulativeX = nextX; cumulativeY = nextY; if (final) + { + if (!reached) + { + cumulativeX = motion.cumulativeX; + cumulativeY = motion.cumulativeY; + } break; + } } { @@ -3980,14 +4060,20 @@ retry: packet.envelope.requestID) return false; - if (!ValidateInputOwner(connection, packet.envelope.componentEpoch, - keyboard.sequence, false, 0, LG_NET_INPUT_CLAIM_KEYBOARD)) - return true; - - uint8_t keys[NETWORK_KEYBOARD_KEY_COUNT] = {}; + uint8_t keys[NETWORK_KEYBOARD_KEY_COUNT] = {}; + std::shared_ptr session = {}; + uint8_t modifiers = 0; { CSRWExclusiveLock guard(connection.inputLock); - connection.keyboardModifiers = + if (!connection.inputClaimantID || + connection.inputClaimEpoch != + packet.envelope.componentEpoch || + !(connection.inputFlags & LG_NET_INPUT_CLAIM_KEYBOARD) || + keyboard.sequence <= connection.inputReliableSequence) + return true; + + connection.inputReliableSequence = keyboard.sequence; + connection.keyboardModifiers = static_cast(keyboard.modifiers); if (keyboard.usage < 224 || keyboard.usage > 231) connection.keyboardKeys[keyboard.usage] = @@ -4004,12 +4090,6 @@ retry: } if (count > NETWORK_KEYBOARD_KEY_COUNT) memset(keys, NETWORK_KEYBOARD_ROLLOVER, sizeof(keys)); - } - - std::shared_ptr session = {}; - uint8_t modifiers = 0; - { - CSRWSharedLock guard(connection.inputLock); session = connection.inputSession; modifiers = connection.keyboardModifiers; } @@ -4035,84 +4115,27 @@ retry: (extent - 1U)); } - static bool InputPointerEnvelopeValid(Connection& connection, - const LGNetPacketView& packet, uint64_t sequence, bool datagram, - bool relative, int32_t wheel, LGNetPointerButtons buttons, - LGNetPointerButtons changedButtons) - { - const bool acknowledge = - (packet.envelope.flags & LG_NET_ENVELOPE_ACK_REQUIRED) != 0; - if (!sequence) - return false; - if (datagram) - { - if (wheel || changedButtons || acknowledge || - packet.envelope.requestID) - return false; - } - else if (acknowledge ? packet.envelope.requestID != sequence : - packet.envelope.requestID != 0) - return false; - - bool current; - bool motionApplies; - bool baseline; - bool modeChange; - bool buttonChange; - { - CSRWSharedLock guard(connection.inputLock); - const uint64_t lastSequence = datagram ? - connection.inputDatagramSequence : - connection.inputReliableSequence; - current = connection.inputClaimantID && - connection.inputClaimEpoch == packet.envelope.componentEpoch && - (connection.inputFlags & LG_NET_INPUT_CLAIM_POINTER) && - sequence > lastSequence; - motionApplies = sequence > connection.inputMotionSequence; - baseline = connection.inputMotionBaseline; - modeChange = motionApplies && baseline && - connection.inputMotionAbsolute == relative; - buttonChange = buttons != connection.inputPointerButtons; - } - if (!current) - return true; - if (datagram) - return !motionApplies || (baseline && !modeChange); - - const bool transition = wheel || changedButtons || buttonChange || - (motionApplies && (!baseline || modeChange)); - if (transition && !acknowledge) - return false; - return true; - } - bool HandleInputPointer(Connection& connection, const LGNetPacketView& packet, bool datagram) { InputResult result = InputResult::STALE; std::shared_ptr session = {}; - { - CSRWSharedLock guard(connection.inputLock); - session = connection.inputSession; - } - if (!session) - return false; if (packet.envelope.messageType == LG_NET_INPUT_MESSAGE_MOUSE_RELATIVE) { LGNetInputRelative relative = {}; if (lgNetInputRelativeDecode(&relative, packet.payload, - packet.payloadLength) != LG_NET_PARSE_OK || relative.wheelX || - !InputPointerEnvelopeValid(connection, packet, relative.sequence, - datagram, true, relative.wheelY, relative.buttons, - relative.changedButtons)) + packet.payloadLength) != LG_NET_PARSE_OK || relative.wheelX) return false; - InputPointerMotion motion; - if (!PrepareInputPointer(connection, - packet.envelope.componentEpoch, relative.sequence, datagram, - true, relative.cumulativeX, relative.cumulativeY, - relative.buttons, motion)) + InputPointerMotion motion = {}; + const InputPrepareResult prepared = PrepareInputPointer(connection, + packet, relative.sequence, datagram, true, relative.wheelY, + relative.changedButtons, relative.cumulativeX, + relative.cumulativeY, relative.buttons, motion, session); + if (prepared == InputPrepareResult::INVALID) + return false; + if (prepared == InputPrepareResult::STALE) return true; if (datagram && !motion.apply) return true; @@ -4131,18 +4154,18 @@ retry: { LGNetInputAbsolute absolute = {}; if (lgNetInputAbsoluteDecode(&absolute, packet.payload, - packet.payloadLength) != LG_NET_PARSE_OK || absolute.wheelX || - !InputPointerEnvelopeValid(connection, packet, absolute.sequence, - datagram, false, absolute.wheelY, absolute.buttons, - absolute.changedButtons)) + packet.payloadLength) != LG_NET_PARSE_OK || absolute.wheelX) return false; - InputPointerMotion motion; + InputPointerMotion motion = {}; motion.absoluteX = NormalizeAbsolute(absolute.x, absolute.width); motion.absoluteY = NormalizeAbsolute(absolute.y, absolute.height); - if (!PrepareInputPointer(connection, - packet.envelope.componentEpoch, absolute.sequence, datagram, - false, absolute.cumulativeX, absolute.cumulativeY, - absolute.buttons, motion)) + const InputPrepareResult prepared = PrepareInputPointer(connection, + packet, absolute.sequence, datagram, false, absolute.wheelY, + absolute.changedButtons, absolute.cumulativeX, + absolute.cumulativeY, absolute.buttons, motion, session); + if (prepared == InputPrepareResult::INVALID) + return false; + if (prepared == InputPrepareResult::STALE) return true; if (datagram && !motion.apply) return true; @@ -4183,7 +4206,6 @@ retry: return false; } - CSRWExclusiveLock dispatchGuard(connection.inputDispatchLock); switch (packet.envelope.messageType) { case LG_NET_INPUT_MESSAGE_CLAIM: @@ -5898,13 +5920,15 @@ retry: case QUIC_STREAM_EVENT_SEND_COMPLETE: { - std::unique_ptr send( - static_cast(event->SEND_COMPLETE.ClientContext)); + InputSend& send = *static_cast( + event->SEND_COMPLETE.ClientContext); + bool inputAck = false; { CSRWExclusiveLock guard(connection.inputLock); - ReleaseInputSendLocked(connection, *send); + inputAck = send.inputAck; + ReleaseInputSendLocked(connection, send); connection.inputLEDWaitForSendReady = false; - if (send->inputAck) + if (inputAck) connection.inputAckInFlight = false; } if (owner.TrySendPendingInputAck(connection) == SendResult::FAILED || @@ -6253,7 +6277,8 @@ retry: QUIC_REGISTRATION_CONFIG registrationConfig = {}; registrationConfig.AppName = "looking-glass-idd"; - registrationConfig.ExecutionProfile = QUIC_EXECUTION_PROFILE_LOW_LATENCY; + registrationConfig.ExecutionProfile = + QUIC_EXECUTION_PROFILE_TYPE_REAL_TIME; QUIC_STATUS status = api->RegistrationOpen( ®istrationConfig, ®istration); if (QUIC_FAILED(status)) diff --git a/idd/LGInput/CHIDDevice.cpp b/idd/LGInput/CHIDDevice.cpp index bca1ff5b..43fa7ea0 100644 --- a/idd/LGInput/CHIDDevice.cpp +++ b/idd/LGInput/CHIDDevice.cpp @@ -248,6 +248,8 @@ static bool CompactQueuedMotion(_Inout_ HIDDeviceContext * context) previous->x = static_cast(x); previous->y = static_cast(y); + if (second.enqueueTicks) + first.enqueueTicks = second.enqueueTicks; RemoveQueuedReport(context, i + 1); ++context->statistics.relativeCoalesced; return true; @@ -265,6 +267,8 @@ static bool CompactQueuedMotion(_Inout_ HIDDeviceContext * context) continue; CopyMemory(first.data, second.data, second.size); + if (second.enqueueTicks) + first.enqueueTicks = second.enqueueTicks; first.size = second.size; RemoveQueuedReport(context, i + 1); ++context->statistics.staleAbsoluteCompacted; @@ -311,6 +315,9 @@ static NTSTATUS QueueReport( { previous->x = static_cast(x); previous->y = static_cast(y); + LARGE_INTEGER enqueueTime = {}; + if (QueryPerformanceCounter(&enqueueTime)) + tail.enqueueTicks = enqueueTime.QuadPart; ++context->statistics.relativeCoalesced; return STATUS_SUCCESS; } @@ -326,6 +333,9 @@ static NTSTATUS QueueReport( !previous->wheel && !current->wheel) { CopyMemory(tail.data, data, size); + LARGE_INTEGER enqueueTime = {}; + if (QueryPerformanceCounter(&enqueueTime)) + tail.enqueueTicks = enqueueTime.QuadPart; tail.size = size; ++context->statistics.absoluteCoalesced; return STATUS_SUCCESS; @@ -361,9 +371,9 @@ static NTSTATUS ReadReport( _In_ WDFREQUEST request, _Out_ bool * complete) { - HIDQueuedReport report = {}; + HIDQueuedReport report = {}; bool haveReport = false; - *complete = true; + *complete = true; NTSTATUS status; { @@ -372,13 +382,22 @@ static NTSTATUS ReadReport( status = STATUS_DEVICE_NOT_READY; else if (context->reportCount) { + // If Windows stopped issuing reads temporarily, deliver one current + // aggregate instead of replaying every stale motion report accumulated + // while it was stalled. Transitions remain ordered and are never + // compacted. + if (MotionBudgetExceeded(context)) + while (CompactQueuedMotion(context)) + { + } + PopReport(context, &report); haveReport = true; - status = STATUS_SUCCESS; + status = STATUS_SUCCESS; } else { - status = WdfRequestForwardToIoQueue(request, context->reportQueue); + status = WdfRequestForwardToIoQueue(request, context->reportQueue); *complete = !NT_SUCCESS(status); } }