Mirrored_Repos/LookingGlass
1
0
Fork 0
mirror of https://github.com/gnif/LookingGlass.git synced 2026-11-12 14:57:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

[client/idd] input: complete network latency path

Browse Source 
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.
This commit is contained in:
Geoffrey McRae
2026-10-08 21:34:24 +11:00
parent 1af2f13440
commit b1dda92f09
9 changed files with 695 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files

515
client/transports/Network/connection.c
View File

@@ -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,
&parameters, 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(
&registration, &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(&parameters, 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,
&parameters, 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,
&parameters, 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 =

2
client/transports/Network/connection.h
View File

@@ -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,

128
client/transports/Network/input.c
View File

@@ -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, &notify);
if (input->sendReadyPending)
networkInputResumeSendsNL(input, &notify);
}
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, &notify);
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, &notify);
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, &notify);
if (input->sendReadyPending)
networkInputResumeSendsNL(input, &notify);
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, &notify);
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, &notify);
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, &notify);
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, &notify);
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, &notify);
LG_UNLOCK(input->stateLock);
if (notify)
@@ -1661,6 +1707,8 @@ static bool networkInputHandleStatus(NetworkInput * input,
input->reliableBarrierSequence = 0;
networkInputResumeSendsNL(input, &notify);
}
if (input->sendReadyPending)
networkInputResumeSendsNL(input, &notify);
LG_UNLOCK(input->stateLock);
if (notify)
@@ -1703,6 +1751,8 @@ static bool networkInputHandleLEDs(NetworkInput * input,
NETWORK_INPUT_LED_MASK;
networkInputAcknowledgeLEDsNL(input, changed);
networkInputSyncKeyboardLEDsNL(input, &notify);
if (input->sendReadyPending)
networkInputResumeSendsNL(input, &notify);
notify |= oldValid != input->keyboardLEDsValid ||
oldState != input->keyboardLEDs;
if (notify)

8
idd/LGCommon/CPipeEndpoint.cpp
View File

@@ -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)
{

7
idd/LGCommon/CPipeEndpoint.h
View File

@@ -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<bool> m_running { false };
std::atomic<bool> m_connected { false };

26
idd/LGIdd/ipc/CInputPipeServer.cpp
View File

@@ -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<int32_t>(x);
first.payload.mouseRelative.deltaY = static_cast<int32_t>(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<int32_t>(x);
tail.payload.mouseRelative.deltaY = static_cast<int32_t>(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)

1
idd/LGIdd/ipc/CInputPipeServer.h
View File

@@ -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

325
idd/LGIdd/transport/network/CNetworkServer.cpp
View File

@@ -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<uint8_t> inputReceiveData;
std::vector<uint8_t> inputReceivePayload;
std::shared_ptr<CNetworkSession> 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> 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<uint32_t>(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<uint32_t>(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, &parameters, 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<uint32_t>(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<uint32_t>(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<CNetworkSession>& 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<CNetworkSession> 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<uint8_t>(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<CNetworkSession> 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<CNetworkSession> 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> send(
static_cast<Send *>(event->SEND_COMPLETE.ClientContext));
InputSend& send = *static_cast<InputSend *>(
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(
&registrationConfig, &registration);
if (QUIC_FAILED(status))

27
idd/LGInput/CHIDDevice.cpp
View File

@@ -248,6 +248,8 @@ static bool CompactQueuedMotion(_Inout_ HIDDeviceContext * context)
previous->x = static_cast<int16_t>(x);
previous->y = static_cast<int16_t>(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<int16_t>(x);
previous->y = static_cast<int16_t>(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);
}
}