From 6631fde35f3a30ecf21b9f1307a5c1973bd09b52 Mon Sep 17 00:00:00 2001 From: Geoffrey McRae Date: Sat, 15 Aug 2026 01:13:13 +1000 Subject: [PATCH] [client] clipboard: reduce remote file transfer latency Mark the last non-empty FILE_DATA payload with BEGIN and END when the producer knows it is terminal. Complete producer bookkeeping in the following fileDataEnd call without consuming another LGMP grant. Keep the standalone terminal record for empty and legacy unknown streams. Drain up to 64 queued clipboard records per worker pass instead of sleeping after every message. Poll at 1 ms while transfers, blocked writes, held input, or pending output are active, and retain the 10 ms interval while idle. This removes a serialized grant round trip from every non-empty file response and reduces scheduling delay while a large transfer is active. Keep the 1 MiB payload size because the bottleneck was stop-and-wait latency, not the chunk capacity. Exercise a full 1 MiB response in a single BEGIN|END grant and retain coverage for empty unknown-size responses. --- client/include/core/clipboard.h | 2 +- client/include/interface/clipboard.h | 5 +- client/src/clipboard.c | 4 +- client/src/clipboard_files.c | 11 ++-- client/tests/clipboard_files_test.c | 33 ++++++----- client/tests/lgmp_clipboard_test.c | 40 ++++++++++--- client/transports/LGMP/clipboard.c | 84 +++++++++++++++++++--------- 7 files changed, 123 insertions(+), 56 deletions(-) diff --git a/client/include/core/clipboard.h b/client/include/core/clipboard.h index 3bba4199..92301ce9 100644 --- a/client/include/core/clipboard.h +++ b/client/include/core/clipboard.h @@ -66,7 +66,7 @@ LG_ClipboardResult lgClipboard_fileDataBegin( const LG_ClipboardFileRequest * request, uint64_t sizeHint); LG_ClipboardResult lgClipboard_fileDataChunk( const LG_ClipboardFileRequest * request, uint64_t responseOffset, - const void * data, size_t size); + const void * data, size_t size, bool end); LG_ClipboardResult lgClipboard_fileDataEnd( const LG_ClipboardFileRequest * request, uint64_t finalSize); bool lgClipboard_fileCancel(uint64_t dataset, uint64_t request, diff --git a/client/include/interface/clipboard.h b/client/include/interface/clipboard.h index d19ec61a..336a5466 100644 --- a/client/include/interface/clipboard.h +++ b/client/include/interface/clipboard.h @@ -247,9 +247,12 @@ typedef struct LG_ClipboardOps const LG_ClipboardFileRequest * request); LG_ClipboardResult (*fileDataBegin)(void * opaque, const LG_ClipboardFileRequest * request, uint64_t sizeHint); + /* end marks a final non-empty chunk. fileDataEnd must still be called to + * complete the producer-side stream; it emits a terminal record only when + * the final chunk was not marked, including for an empty response. */ LG_ClipboardResult (*fileDataChunk)(void * opaque, const LG_ClipboardFileRequest * request, uint64_t responseOffset, - const void * data, size_t size); + const void * data, size_t size, bool end); LG_ClipboardResult (*fileDataEnd)(void * opaque, const LG_ClipboardFileRequest * request, uint64_t finalSize); bool (*fileCancel)(void * opaque, uint64_t dataset, diff --git a/client/src/clipboard.c b/client/src/clipboard.c index 0b1fc7e9..83e023aa 100644 --- a/client/src/clipboard.c +++ b/client/src/clipboard.c @@ -2183,13 +2183,13 @@ LG_ClipboardResult lgClipboard_fileDataBegin( LG_ClipboardResult lgClipboard_fileDataChunk( const LG_ClipboardFileRequest * request, uint64_t responseOffset, - const void * data, size_t size) + const void * data, size_t size, bool end) { LG_ClipboardResult result = LG_CLIPBOARD_RESULT_FAILED; LG_LOCK_SHARED(clipboard.activeLock); if (clipboard.active.ops && clipboard.active.ops->fileDataChunk) result = clipboard.active.ops->fileDataChunk( - clipboard.active.opaque, request, responseOffset, data, size); + clipboard.active.opaque, request, responseOffset, data, size, end); LG_UNLOCK_SHARED(clipboard.activeLock); return result; } diff --git a/client/src/clipboard_files.c b/client/src/clipboard_files.c index 7a08b7dd..2e06da7a 100644 --- a/client/src/clipboard_files.c +++ b/client/src/clipboard_files.c @@ -2764,7 +2764,7 @@ static struct LocalDataset * acquiredLocalDatasetNL(uint64_t wireDataset) } static bool sendJobCall(struct LocalJob * job, int operation, - uint64_t offset, const void * data, size_t size) + uint64_t offset, const void * data, size_t size, bool end) { for (;;) { @@ -2778,7 +2778,7 @@ static bool sendJobCall(struct LocalJob * job, int operation, break; case 1: result = lgClipboard_fileDataChunk( - &job->request, offset, data, size); + &job->request, offset, data, size, end); break; default: result = lgClipboard_fileDataEnd(&job->request, offset); @@ -2796,7 +2796,7 @@ static bool sendJobCall(struct LocalJob * job, int operation, static bool sendBufferJob(struct LocalJob * job, const uint8_t * data, size_t size) { - if (!sendJobCall(job, 0, size, NULL, 0)) + if (!sendJobCall(job, 0, size, NULL, 0, false)) return false; size_t offset = 0; while (offset < size) @@ -2804,11 +2804,12 @@ static bool sendBufferJob(struct LocalJob * job, size_t chunk = size - offset; if (chunk > FILE_STREAM_CHUNK) chunk = FILE_STREAM_CHUNK; - if (!sendJobCall(job, 1, offset, data + offset, chunk)) + const bool end = chunk == size - offset; + if (!sendJobCall(job, 1, offset, data + offset, chunk, end)) return false; offset += chunk; } - return sendJobCall(job, 2, size, NULL, 0); + return sendJobCall(job, 2, size, NULL, 0, true); } static LG_ClipboardFileError appendLocalEntry(uint8_t ** data, size_t * size, diff --git a/client/tests/clipboard_files_test.c b/client/tests/clipboard_files_test.c index 7561751e..a6f47b1e 100644 --- a/client/tests/clipboard_files_test.c +++ b/client/tests/clipboard_files_test.c @@ -53,6 +53,7 @@ static size_t responseSize; static size_t responseCapacity; static bool responseOpen; static bool responseComplete; +static bool responseWireEnded; static unsigned cancelCount; static uint64_t cancelledDataset; static uint64_t cancelledRequest; @@ -102,15 +103,16 @@ LG_ClipboardResult lgClipboard_fileDataBegin( { CHECK(request); (void)sizeHint; - responseSize = 0; - responseOpen = true; - responseComplete = false; + responseSize = 0; + responseOpen = true; + responseComplete = false; + responseWireEnded = false; return LG_CLIPBOARD_RESULT_ACCEPTED; } LG_ClipboardResult lgClipboard_fileDataChunk( const LG_ClipboardFileRequest * request, uint64_t responseOffset, - const void * data, size_t size) + const void * data, size_t size, bool end) { CHECK(request); CHECK(responseOpen); @@ -122,12 +124,14 @@ LG_ClipboardResult lgClipboard_fileDataChunk( { uint8_t * resized = realloc(responseData, wanted); CHECK(resized); - responseData = resized; + responseData = resized; responseCapacity = wanted; } if (size) memcpy(responseData + responseSize, data, size); responseSize = wanted; + if (end) + responseWireEnded = true; return LG_CLIPBOARD_RESULT_ACCEPTED; } @@ -137,7 +141,8 @@ LG_ClipboardResult lgClipboard_fileDataEnd( CHECK(request); CHECK(responseOpen); CHECK(finalSize == responseSize); - responseOpen = false; + CHECK(!finalSize || responseWireEnded); + responseOpen = false; responseComplete = true; return LG_CLIPBOARD_RESULT_ACCEPTED; } @@ -169,13 +174,14 @@ static bool runLocalRequest(uint64_t dataset, uint64_t node, LG_ClipboardFileOperation operation, uint64_t offset, uint32_t length) { static uint64_t requestId = 1; - responseSize = 0; - responseOpen = false; - responseComplete = false; - cancelCount = 0; - cancelledDataset = 0; - cancelledRequest = 0; - cancelledReason = LG_CLIPBOARD_FILE_ERROR_NONE; + responseSize = 0; + responseOpen = false; + responseComplete = false; + responseWireEnded = false; + cancelCount = 0; + cancelledDataset = 0; + cancelledRequest = 0; + cancelledReason = LG_CLIPBOARD_FILE_ERROR_NONE; const LG_ClipboardFileRequest request = { .dataset = dataset, @@ -187,6 +193,7 @@ static bool runLocalRequest(uint64_t dataset, uint64_t node, }; const bool result = lgClipboardFiles_testLocalRequest(&request); CHECK(!result || responseComplete); + CHECK(!result || !responseSize || responseWireEnded); CHECK(result ? cancelCount == 0U : cancelCount == 1U); if (!result) { diff --git a/client/tests/lgmp_clipboard_test.c b/client/tests/lgmp_clipboard_test.c index 1d7110f4..c1a98142 100644 --- a/client/tests/lgmp_clipboard_test.c +++ b/client/tests/lgmp_clipboard_test.c @@ -719,6 +719,7 @@ static bool checkCommit(TestState * state, uint32_t token, } static bool checkFileCommit(TestState * state, uint32_t grantToken, + KVMFRClipboardFileOperation operation, KVMFRClipboardMessage * commit, KVMFRClipboardMessage * slot, const void ** data) { @@ -736,7 +737,7 @@ static bool checkFileCommit(TestState * state, uint32_t grantToken, CHECK(slot->flags == commit->flags); CHECK(slot->length == commit->length); CHECK(slot->sequence == commit->sequence); - CHECK(slot->token == KVMFR_CLIPBOARD_FILE_OP_READ); + CHECK(slot->token == operation); *data = (const uint8_t *)lgmpHostMemPtr( state->grantMemory[grantToken - 1]) + sizeof(*slot); return true; @@ -1280,26 +1281,47 @@ static bool testFileStream(TestState * state) CHECK(postGrant(state, claimRecord.generation, 1)); CHECK(waitMemory(state, state->grantMemory[0])); CHECK(state->ops->fileDataChunk(state->clipboard, - &descriptor, 0, bytes, sizeof(bytes)) == + &descriptor, 0, bytes, sizeof(bytes), true) == LG_CLIPBOARD_RESULT_ACCEPTED); KVMFRClipboardMessage commit; KVMFRClipboardMessage slot; const void * data; - CHECK(checkFileCommit(state, 1, &commit, &slot, &data)); - CHECK(slot.flags == KVMFR_CLIPBOARD_FLAG_BEGIN); + CHECK(checkFileCommit(state, 1, KVMFR_CLIPBOARD_FILE_OP_READ, + &commit, &slot, &data)); + CHECK(slot.flags == (KVMFR_CLIPBOARD_FLAG_BEGIN | + KVMFR_CLIPBOARD_FLAG_END)); CHECK(slot.offset == 0 && slot.sequence == 0 && slot.size == sizeof(bytes)); CHECK(slot.length == sizeof(bytes)); CHECK(memcmp(data, bytes, sizeof(bytes)) == 0); + CHECK(state->ops->fileDataEnd(state->clipboard, + &descriptor, sizeof(bytes)) == LG_CLIPBOARD_RESULT_ACCEPTED); + CHECK(noClientData(state)); + + const unsigned requests = atomic_load(&state->events.fileRequest); + request.transfer = KVMFR_CLIPBOARD_TRANSFER_HELPER | UINT64_C(0x7803); + request.offset = 0; + request.size = KVMFR_CLIPBOARD_FILE_ROOT_NODE; + request.flags = 0; + request.token = KVMFR_CLIPBOARD_FILE_OP_LIST; + CHECK(postRecord(state, &request, + KVMFR_CLIPBOARD_QUEUE_MESSAGE, NULL)); + CHECK(waitAtomic(state, &state->events.fileRequest, requests + 1)); + const LG_ClipboardFileRequest empty = state->events.fileRequestValue; + CHECK(empty.operation == LG_CLIPBOARD_FILE_LIST); + CHECK(state->ops->fileDataBegin(state->clipboard, + &empty, KVMFR_CLIPBOARD_SIZE_UNKNOWN) == + LG_CLIPBOARD_RESULT_ACCEPTED); CHECK(postGrant(state, claimRecord.generation, 1)); CHECK(waitMemory(state, state->grantMemory[0])); CHECK(state->ops->fileDataEnd(state->clipboard, - &descriptor, sizeof(bytes)) == LG_CLIPBOARD_RESULT_ACCEPTED); - CHECK(checkFileCommit(state, 1, &commit, &slot, &data)); - CHECK(slot.flags == KVMFR_CLIPBOARD_FLAG_END); - CHECK(slot.offset == sizeof(bytes) && slot.sequence == 1 && - slot.size == sizeof(bytes)); + &empty, 0) == LG_CLIPBOARD_RESULT_ACCEPTED); + CHECK(checkFileCommit(state, 1, KVMFR_CLIPBOARD_FILE_OP_LIST, + &commit, &slot, &data)); + CHECK(slot.flags == (KVMFR_CLIPBOARD_FLAG_BEGIN | + KVMFR_CLIPBOARD_FLAG_END)); + CHECK(slot.offset == 0 && slot.sequence == 0 && slot.size == 0); CHECK(slot.length == 0); return true; } diff --git a/client/transports/LGMP/clipboard.c b/client/transports/LGMP/clipboard.c index 90fda515..6ee4a514 100644 --- a/client/transports/LGMP/clipboard.c +++ b/client/transports/LGMP/clipboard.c @@ -35,13 +35,14 @@ #include #include -#define CLIPBOARD_PENDING_MAX 128U -#define CLIPBOARD_PENDING_NORMAL_MAX 64U -#define CLIPBOARD_GRANTS KVMFR_CLIPBOARD_SLOT_COUNT -#define CLIPBOARD_POLL_MS 10U -#define CLIPBOARD_RETRY_MS 1U -#define CLIPBOARD_KEEPALIVE_US UINT64_C(200000) -#define CLIPBOARD_RELEASE_TIMEOUT_US UINT64_C(100000) +#define CLIPBOARD_PENDING_MAX 128U +#define CLIPBOARD_PENDING_NORMAL_MAX 64U +#define CLIPBOARD_DRAIN_MAX 64U +#define CLIPBOARD_GRANTS KVMFR_CLIPBOARD_SLOT_COUNT +#define CLIPBOARD_IDLE_POLL_MS 10U +#define CLIPBOARD_ACTIVE_POLL_MS 1U +#define CLIPBOARD_KEEPALIVE_US UINT64_C(200000) +#define CLIPBOARD_RELEASE_TIMEOUT_US UINT64_C(100000) struct PendingRecord { @@ -65,6 +66,7 @@ struct FileTransfer uint32_t sequence; bool began; bool blocked; + bool wireEnded; }; struct FileAcquisition @@ -1418,8 +1420,9 @@ static bool processHeld(LGMPClipboard * clipboard) } } -static bool processMessage(LGMPClipboard * clipboard) +static bool processMessage(LGMPClipboard * clipboard, bool * processed) { + *processed = false; LG_LOCK(clipboard->lock); if (!clipboard->connected || !clipboard->queue || clipboard->held) { @@ -1440,6 +1443,7 @@ static bool processMessage(LGMPClipboard * clipboard) LG_UNLOCK(clipboard->lock); return false; } + *processed = true; const KVMFRClipboardQueueType type = KVMFR_CLIPBOARD_QUEUE_TYPE(message.udata); @@ -1901,13 +1905,28 @@ static int clipboardThread(void * opaque) LGMPClipboard * clipboard = opaque; while (!atomic_load_explicit(&clipboard->stop, memory_order_acquire)) { - const bool held = processHeld(clipboard); - dispatchRetiredFiles(clipboard); - if (!held) - break; - const bool message = processMessage(clipboard); - dispatchRetiredFiles(clipboard); - if (!message) + bool running = true; + unsigned drained = 0; + for (; drained < CLIPBOARD_DRAIN_MAX; ++drained) + { + if (!processHeld(clipboard)) + { + running = false; + break; + } + dispatchRetiredFiles(clipboard); + + bool processed = false; + if (!processMessage(clipboard, &processed)) + { + running = false; + break; + } + dispatchRetiredFiles(clipboard); + if (!processed) + break; + } + if (!running) break; LG_ClipboardRequest readyRequest = LG_CLIPBOARD_REQUEST_INVALID; @@ -1937,7 +1956,10 @@ static int clipboardThread(void * opaque) transfer->blocked = false; fileReady[fileReadyCount++] = transfer->request.request; } - const bool retry = clipboard->pendingCount != 0; + const bool active = drained == CLIPBOARD_DRAIN_MAX || + clipboard->pendingCount || clipboard->held || + clipboard->writeBlocked || clipboard->fileReads || + clipboard->fileWrites; LG_UNLOCK(clipboard->lock); if (readyRequest != LG_CLIPBOARD_REQUEST_INVALID) @@ -1948,7 +1970,7 @@ static int clipboardThread(void * opaque) if (atomic_load_explicit(&clipboard->stop, memory_order_acquire)) break; lgWaitEvent(clipboard->event, - retry ? CLIPBOARD_RETRY_MS : CLIPBOARD_POLL_MS); + active ? CLIPBOARD_ACTIVE_POLL_MS : CLIPBOARD_IDLE_POLL_MS); } dispatchRetiredFiles(clipboard); notifyStatus(clipboard); @@ -2098,7 +2120,7 @@ static void releaseOnDisconnect(LGMPClipboard * clipboard) if (status != LGMP_ERR_QUEUE_BUSY && status != LGMP_ERR_QUEUE_FULL) return; if (clipboard->event) - lgWaitEvent(clipboard->event, CLIPBOARD_RETRY_MS); + lgWaitEvent(clipboard->event, CLIPBOARD_ACTIVE_POLL_MS); } while (microtime() < deadline); @@ -2117,7 +2139,7 @@ static void releaseOnDisconnect(LGMPClipboard * clipboard) if ((int32_t)(processed - serial) >= 0) return; if (clipboard->event) - lgWaitEvent(clipboard->event, CLIPBOARD_RETRY_MS); + lgWaitEvent(clipboard->event, CLIPBOARD_ACTIVE_POLL_MS); } while (microtime() < deadline); DEBUG_WARN("Timed out waiting for LGMP clipboard release"); @@ -2711,6 +2733,7 @@ static LG_ClipboardResult fileDataBegin(void * opaque, transfer->responseOffset = 0; transfer->sequence = 0; transfer->blocked = false; + transfer->wireEnded = false; } LG_UNLOCK(clipboard->lock); return valid ? LG_CLIPBOARD_RESULT_ACCEPTED : @@ -2719,7 +2742,7 @@ static LG_ClipboardResult fileDataBegin(void * opaque, static LG_ClipboardResult fileDataChunk(void * opaque, const LG_ClipboardFileRequest * descriptor, uint64_t responseOffset, - const void * data, size_t size) + const void * data, size_t size, bool end) { LGMPClipboard * clipboard = opaque; if (!descriptor || !data || !size || @@ -2729,11 +2752,13 @@ static LG_ClipboardResult fileDataChunk(void * opaque, LG_LOCK(clipboard->lock); struct FileTransfer * transfer = fileTransferFindNL( clipboard->fileWrites, descriptor->request); - if (!transfer || !transfer->began || + if (!transfer || !transfer->began || transfer->wireEnded || !sameFileRequest(&transfer->request, descriptor) || transfer->responseOffset != responseOffset || (transfer->sizeHint != KVMFR_CLIPBOARD_SIZE_UNKNOWN && responseOffset + size > transfer->sizeHint) || + (end && transfer->sizeHint != KVMFR_CLIPBOARD_SIZE_UNKNOWN && + transfer->sizeHint != responseOffset + size) || (descriptor->operation == LG_CLIPBOARD_FILE_READ && responseOffset + size > descriptor->length)) { @@ -2746,11 +2771,12 @@ static LG_ClipboardResult fileDataChunk(void * opaque, record.transfer = descriptor->request; record.offset = responseOffset; record.sequence = transfer->sequence; - record.size = transfer->sequence ? - KVMFR_CLIPBOARD_SIZE_UNKNOWN : transfer->sizeHint; + record.size = end ? responseOffset + size : + transfer->sequence ? KVMFR_CLIPBOARD_SIZE_UNKNOWN : transfer->sizeHint; record.format = KVMFR_CLIPBOARD_FORMAT_FILES; - record.flags = transfer->sequence ? 0 : - KVMFR_CLIPBOARD_FLAG_BEGIN; + record.flags = (transfer->sequence ? 0 : + KVMFR_CLIPBOARD_FLAG_BEGIN) | + (end ? KVMFR_CLIPBOARD_FLAG_END : 0); record.token = toWireFileOperation(descriptor->operation); record.length = (uint32_t)size; const bool result = enqueueDataNL(clipboard, record, data); @@ -2758,6 +2784,7 @@ static LG_ClipboardResult fileDataChunk(void * opaque, { transfer->responseOffset += size; ++transfer->sequence; + transfer->wireEnded = end; } else transfer->blocked = true; @@ -2786,6 +2813,13 @@ static LG_ClipboardResult fileDataEnd(void * opaque, LG_UNLOCK(clipboard->lock); return LG_CLIPBOARD_RESULT_FAILED; } + if (transfer->wireEnded) + { + free(fileTransferTakeNL( + &clipboard->fileWrites, descriptor->request)); + LG_UNLOCK(clipboard->lock); + return LG_CLIPBOARD_RESULT_ACCEPTED; + } KVMFRClipboardMessage record = { 0 }; record.type = KVMFR_CLIPBOARD_MESSAGE_FILE_DATA; record.clipboardGeneration = descriptor->dataset;