From 656b0995f6fa9f01be481d8a72730bc789b90eb3 Mon Sep 17 00:00:00 2001 From: Geoffrey McRae Date: Wed, 12 Aug 2026 17:28:49 +1000 Subject: [PATCH] [client] transport: make video waits cancellable --- client/include/interface/transport.h | 5 ++ client/src/app.c | 17 +++--- client/src/core.c | 67 ++++++++++++---------- client/src/core.h | 6 +- client/src/keybind.c | 3 +- client/src/main.c | 26 +++++---- client/src/main.h | 3 +- client/tests/lgmp_transport_test.c | 86 ++++++++++++++++++++++++++-- client/transports/LGMP/lgmp.c | 82 ++++++++++++++++++++------ 9 files changed, 219 insertions(+), 76 deletions(-) diff --git a/client/include/interface/transport.h b/client/include/interface/transport.h index b54b76da..6c7dae58 100644 --- a/client/include/interface/transport.h +++ b/client/include/interface/transport.h @@ -327,6 +327,9 @@ typedef struct LG_FrameOps void (*getFrameTiming)(LG_Transport * transport, const LG_TransportFrame * frame, LG_TransportFrameTiming * timing); void (*releaseFrame)(LG_Transport * transport, LG_TransportFrame * frame); + /* Optional, thread-safe cancellation of a blocking nextFrame call. This + * does not release a frame already returned to the consumer. */ + void (*cancelFrameWait)(LG_Transport * transport); /* Called by the frame consumer as it exits. A backend may release transient * stream resources; nextFrame must reacquire them when the consumer * restarts. */ @@ -336,6 +339,8 @@ typedef struct LG_FrameOps LG_TransportPointer * pointer); void (*releasePointer)(LG_Transport * transport, LG_TransportPointer * pointer); + /* Optional, thread-safe cancellation of a blocking nextPointer call. */ + void (*cancelPointerWait)(LG_Transport * transport); /* Called by the pointer consumer as it exits. A backend may release * transient stream resources; nextPointer must reacquire them when the * consumer restarts. */ diff --git a/client/src/app.c b/client/src/app.c index 5c076a26..4887b934 100644 --- a/client/src/app.c +++ b/client/src/app.c @@ -1187,14 +1187,14 @@ bool app_guestIsOther(void) void app_stopVideo(bool stop) { - if (g_state.stopVideo == stop) + if (atomic_load_explicit(&g_state.stopVideo, memory_order_acquire) == stop) return; // do not change the state if the host app is not connected if (!atomic_load_explicit(&g_state.lgHostConnected, memory_order_acquire)) return; - g_state.stopVideo = stop; + atomic_store_explicit(&g_state.stopVideo, stop, memory_order_release); app_alert( LG_ALERT_INFO, @@ -1204,10 +1204,7 @@ void app_stopVideo(bool stop) if (stop) { if (g_state.videoOps->type == LG_VIDEO_TYPE_FRAME) - { - core_stopCursorThread(); - core_stopFrameThread(); - } + core_stopVideoThreads(); else g_state.videoOps->swSurface->setActive( g_state.transport.handle, false); @@ -1216,8 +1213,12 @@ void app_stopVideo(bool stop) { if (g_state.videoOps->type == LG_VIDEO_TYPE_FRAME) { - core_startCursorThread(); - core_startFrameThread(); + if (!core_startVideoThreads()) + { + atomic_store_explicit( + &g_state.stopVideo, true, memory_order_release); + app_alert(LG_ALERT_ERROR, "Failed to enable the video stream"); + } } else { diff --git a/client/src/core.c b/client/src/core.c index 9fae5f79..c97f1e0d 100644 --- a/client/src/core.c +++ b/client/src/core.c @@ -621,54 +621,59 @@ bool core_isValidPointerPos(int x, int y) return g_state.ds->isValidPointerPos(x, y); } -bool core_startCursorThread(void) +static void cancelVideoWaits(void) { - if (g_state.cursorThread) - return true; + if (!g_state.videoOps || g_state.videoOps->type != LG_VIDEO_TYPE_FRAME) + return; - g_state.stopVideo = false; + const LG_FrameOps * ops = g_state.videoOps->frame; + if (g_state.frameThread && ops->cancelFrameWait) + ops->cancelFrameWait(g_state.transport.handle); + if (g_state.cursorThread && ops->cancelPointerWait) + ops->cancelPointerWait(g_state.transport.handle); +} + +void core_stopVideoThreads(void) +{ + atomic_store_explicit( + &g_state.stopVideoThreads, true, memory_order_release); + cancelVideoWaits(); + + if (g_state.frameThread) + lgJoinThread(g_state.frameThread, NULL); + if (g_state.cursorThread) + lgJoinThread(g_state.cursorThread, NULL); + + g_state.frameThread = NULL; + g_state.cursorThread = NULL; +} + +bool core_startVideoThreads(void) +{ + if (g_state.frameThread && g_state.cursorThread) + return true; + if (g_state.frameThread || g_state.cursorThread) + core_stopVideoThreads(); + + atomic_store_explicit( + &g_state.stopVideoThreads, false, memory_order_release); if (!lgCreateThread("cursorThread", main_cursorThread, NULL, &g_state.cursorThread)) { DEBUG_ERROR("cursor create thread failed"); return false; } - return true; -} -void core_stopCursorThread(void) -{ - g_state.stopVideo = true; - if (g_state.cursorThread) - lgJoinThread(g_state.cursorThread, NULL); - - g_state.cursorThread = NULL; -} - -bool core_startFrameThread(void) -{ - if (g_state.frameThread) - return true; - - g_state.stopVideo = false; if (!lgCreateThread("frameThread", main_frameThread, NULL, &g_state.frameThread)) { DEBUG_ERROR("frame create thread failed"); + core_stopVideoThreads(); return false; } return true; } -void core_stopFrameThread(void) -{ - g_state.stopVideo = true; - if (g_state.frameThread) - lgJoinThread(g_state.frameThread, NULL); - - g_state.frameThread = NULL; -} - void core_handleGuestMouseUpdate(void) { /* The local cursor is authoritative with absolute input. Sending delayed @@ -932,7 +937,7 @@ void core_handleMouseNormal(double ex, double ey) struct DoublePoint guest; util_localCurToGuest(&guest); - if (!g_state.stopVideo && + if (!atomic_load_explicit(&g_state.stopVideo, memory_order_acquire) && g_state.transportFeatures & LG_TRANSPORT_FEATURE_SET_CURSOR_POS) { const LG_TransportControl control = { diff --git a/client/src/core.h b/client/src/core.h index 82be6879..49e3e653 100644 --- a/client/src/core.h +++ b/client/src/core.h @@ -34,10 +34,8 @@ void core_onWindowSizeChanged(unsigned width, unsigned height); void core_updatePositionInfo(void); void core_alignToGuest(void); bool core_isValidPointerPos(int x, int y); -bool core_startCursorThread(void); -void core_stopCursorThread(void); -bool core_startFrameThread(void); -void core_stopFrameThread(void); +bool core_startVideoThreads(void); +void core_stopVideoThreads(void); void core_handleGuestMouseUpdate(void); void core_handleMouseAbsolute(void); void core_handleMousePosition(double x, double y); diff --git a/client/src/keybind.c b/client/src/keybind.c index 52f6f3cf..66345a69 100644 --- a/client/src/keybind.c +++ b/client/src/keybind.c @@ -37,7 +37,8 @@ static void bind_fullscreen(int sc, void * opaque) static void bind_video(int sc, void * opaque) { - app_stopVideo(!g_state.stopVideo); + app_stopVideo(!atomic_load_explicit( + &g_state.stopVideo, memory_order_acquire)); } static void bind_rotate(int sc, void * opaque) diff --git a/client/src/main.c b/client/src/main.c index 586101ba..d25ca125 100644 --- a/client/src/main.c +++ b/client/src/main.c @@ -1101,8 +1101,7 @@ static int renderThread(void * unused) lgClipboard_dropTransport(); } - core_stopCursorThread(); - core_stopFrameThread(); + core_stopVideoThreads(); if (g_state.videoOps && g_state.videoOps->type == LG_VIDEO_TYPE_FRAME && g_state.videoOps->frame->detachRenderer) @@ -1135,7 +1134,9 @@ int main_cursorThread(void * unused) lgWaitEvent(e_startup, TIMEOUT_INFINITE); // subscribe to the pointer queue - while(app_getState() == APP_STATE_RUNNING && !g_state.stopVideo) + while(app_getState() == APP_STATE_RUNNING && + !atomic_load_explicit( + &g_state.stopVideoThreads, memory_order_acquire)) { LG_TransportPointer pointer; const LG_TransportStatus status = g_state.videoOps->frame->nextPointer( @@ -1144,7 +1145,9 @@ int main_cursorThread(void * unused) { if (status == LG_TRANSPORT_TIMEOUT || status == LG_TRANSPORT_UNAVAILABLE) { - if (!g_state.stopVideo && queueCursorRedraw()) + if (!atomic_load_explicit( + &g_state.stopVideoThreads, memory_order_acquire) && + queueCursorRedraw()) cursorRepaintRequest(); continue; } @@ -1288,7 +1291,8 @@ int main_cursorThread(void * unused) const bool contentChanged = (pointer.flags & (LG_TRANSPORT_POINTER_SHAPE | LG_TRANSPORT_POINTER_COLOR_TRANSFORM)) || whiteLevelChanged; - if (sourceApplied && !g_state.stopVideo && + if (sourceApplied && + !atomic_load_explicit(&g_state.stopVideo, memory_order_acquire) && (wasRendered != isRendered || ((wasRendered || isRendered) && (g_params.mouseRedraw || contentChanged)))) @@ -1324,7 +1328,9 @@ int main_frameThread(void * unused) return 0; } - while(app_getState() == APP_STATE_RUNNING && !g_state.stopVideo) + while(app_getState() == APP_STATE_RUNNING && + !atomic_load_explicit( + &g_state.stopVideoThreads, memory_order_acquire)) { LG_TransportFrame frame; const LG_TransportStatus status = g_state.videoOps->frame->nextFrame( @@ -3485,7 +3491,8 @@ restart: if (g_state.videoOps->type == LG_VIDEO_TYPE_FRAME) { videoSourceBegin(LG_VIDEO_SOURCE_PRIMARY); - if (!core_startCursorThread() || !core_startFrameThread()) + if (!atomic_load_explicit(&g_state.stopVideo, memory_order_acquire) && + !core_startVideoThreads()) return recoveryExit(&recoveryPrompt, -1); } else @@ -3529,10 +3536,7 @@ restart: lgSignalEvent(g_state.frameEvent); if (g_state.videoOps->type == LG_VIDEO_TYPE_FRAME) - { - core_stopFrameThread(); - core_stopCursorThread(); - } + core_stopVideoThreads(); else g_state.videoOps->swSurface->setActive( g_state.transport.handle, false); diff --git a/client/src/main.h b/client/src/main.h index 0a9f08fe..933b41a8 100644 --- a/client/src/main.h +++ b/client/src/main.h @@ -114,7 +114,8 @@ struct AppState atomic_bool lgHostConnected; - bool stopVideo; + atomic_bool stopVideo; + atomic_bool stopVideoThreads; bool ignoreInput; bool escapeActive; uint64_t escapeTime; diff --git a/client/tests/lgmp_transport_test.c b/client/tests/lgmp_transport_test.c index 413187b1..6ee9b91a 100644 --- a/client/tests/lgmp_transport_test.c +++ b/client/tests/lgmp_transport_test.c @@ -28,6 +28,7 @@ #include #include +#include #include #include #include @@ -69,6 +70,77 @@ static bool waitForQueuesEmpty(PLGMPHost host, PLGMPHostQueue frameQueue, return false; } +struct WaitState +{ + LG_Transport * transport; + const LG_FrameOps * ops; + LG_TransportStatus status; + bool frame; +}; + +static void * waitForVideo(void * opaque) +{ + struct WaitState * state = opaque; + if (state->frame) + { + LG_TransportFrame frame; + state->status = state->ops->nextFrame( + state->transport, false, &frame); + } + else + { + LG_TransportPointer pointer; + state->status = state->ops->nextPointer(state->transport, &pointer); + } + return NULL; +} + +static bool checkWaitCancellation(LG_Transport * transport, + const LG_FrameOps * ops, PLGMPHostQueue frameQueue, + PLGMPHostQueue pointerQueue) +{ + struct WaitState frame = { + .transport = transport, + .ops = ops, + .status = LG_TRANSPORT_ERROR, + .frame = true, + }; + struct WaitState pointer = { + .transport = transport, + .ops = ops, + .status = LG_TRANSPORT_ERROR, + }; + pthread_t frameThread; + pthread_t pointerThread; + if (pthread_create(&frameThread, NULL, waitForVideo, &frame) != 0) + return false; + if (pthread_create(&pointerThread, NULL, waitForVideo, &pointer) != 0) + { + ops->cancelFrameWait(transport); + pthread_join(frameThread, NULL); + return false; + } + + bool subscribed = false; + for (unsigned i = 0; i < WAIT_TIMEOUT; ++i) + { + if (lgmpHostQueueHasSubs(frameQueue) && + lgmpHostQueueHasSubs(pointerQueue)) + { + subscribed = true; + break; + } + usleep(1000); + } + ops->cancelFrameWait(transport); + ops->cancelPointerWait(transport); + const int frameJoin = pthread_join(frameThread, NULL); + const int pointerJoin = pthread_join(pointerThread, NULL); + return subscribed && frameJoin == 0 && pointerJoin == 0 && + frame.status == LG_TRANSPORT_TIMEOUT && + pointer.status == LG_TRANSPORT_TIMEOUT; +} + int main(void) { int result = 1; @@ -153,14 +225,16 @@ int main(void) LGT_LGMP.setup(); option_set_string("lgmp", "shmDevice", path); option_set_bool("lgmp", "allowDMA", false); - option_set_int("lgmp", "framePollInterval", 0); - option_set_int("lgmp", "cursorPollInterval", 0); + option_set_int("lgmp", "framePollInterval", 5000000); + option_set_int("lgmp", "cursorPollInterval", 5000000); CHECK(LGT_LGMP.create(&transport)); const LG_VideoOps * videoOps = LGT_LGMP.getVideoOps(transport); CHECK(videoOps); CHECK(videoOps->type == LG_VIDEO_TYPE_FRAME); CHECK(videoOps->frame); frameOps = videoOps->frame; + CHECK(frameOps->cancelFrameWait); + CHECK(frameOps->cancelPointerWait); CHECK(unlink(path) == 0); pathExists = false; @@ -173,8 +247,8 @@ int main(void) LG_TransportFrame frame; LG_TransportPointer pointer; - CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_TIMEOUT); - CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT); + CHECK(checkWaitCancellation( + transport, frameOps, frameQueue, pointerQueue)); CHECK(lgmpHostQueueHasSubs(frameQueue)); CHECK(lgmpHostQueueHasSubs(pointerQueue)); CHECK(lgmpHostQueueNewSubs(frameQueue) == 1); @@ -212,7 +286,9 @@ int main(void) CHECK(lgmpHostQueuePending(frameQueue) == 0); CHECK(lgmpHostQueuePending(pointerQueue) == 0); + frameOps->cancelFrameWait(transport); CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_TIMEOUT); + frameOps->cancelPointerWait(transport); CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT); CHECK(lgmpHostQueueNewSubs(frameQueue) == 1); CHECK(lgmpHostQueueNewSubs(pointerQueue) == 1); @@ -242,7 +318,9 @@ int main(void) frameOps->stopFrame(transport); frameOps->stopPointer(transport); + frameOps->cancelFrameWait(transport); CHECK(frameOps->nextFrame(transport, false, &frame) == LG_TRANSPORT_TIMEOUT); + frameOps->cancelPointerWait(transport); CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT); CHECK(lgmpHostQueueNewSubs(frameQueue) == 1); CHECK(lgmpHostQueueNewSubs(pointerQueue) == 1); diff --git a/client/transports/LGMP/lgmp.c b/client/transports/LGMP/lgmp.c index 85713376..9c301551 100644 --- a/client/transports/LGMP/lgmp.c +++ b/client/transports/LGMP/lgmp.c @@ -26,6 +26,7 @@ #include "common/KVMFRRecovery.h" #include "common/LGMPConfig.h" #include "common/debug.h" +#include "common/event.h" #include "common/ivshmem.h" #include "common/locking.h" #include "common/option.h" @@ -78,6 +79,8 @@ struct LG_Transport LGMPInput * input; LG_Lock frameLock; LG_Lock pointerLock; + LGEvent * frameWake; + LGEvent * pointerWake; unsigned cursorPollInterval; unsigned framePollInterval; @@ -345,6 +348,18 @@ static bool lgmp_create(LG_Transport ** result) this->framePollInterval = framePoll; this->cursorPollInterval = cursorPoll; + this->frameWake = lgCreateEvent(true, 0); + this->pointerWake = lgCreateEvent(true, 0); + if (!this->frameWake || !this->pointerWake) + { + if (this->frameWake) + lgFreeEvent(this->frameWake); + if (this->pointerWake) + lgFreeEvent(this->pointerWake); + free(this); + return false; + } + for (unsigned i = 0; i < LGMP_Q_FRAME_BUFFER_LEN; ++i) this->dma[i].fd = -1; @@ -363,6 +378,8 @@ static bool lgmp_create(LG_Transport ** result) LG_LOCK_FREE(this->frameLock); LG_LOCK_FREE(this->pointerLock); LG_LOCK_FREE(this->recoveryLock); + lgFreeEvent(this->frameWake); + lgFreeEvent(this->pointerWake); free(this); return false; } @@ -399,6 +416,8 @@ static bool lgmp_create(LG_Transport ** result) LG_LOCK_FREE(this->frameLock); LG_LOCK_FREE(this->pointerLock); LG_LOCK_FREE(this->recoveryLock); + lgFreeEvent(this->frameWake); + lgFreeEvent(this->pointerWake); free(this); return false; } @@ -540,6 +559,8 @@ static void lgmp_destroy(LG_Transport ** transport) LG_LOCK_FREE(this->frameLock); LG_LOCK_FREE(this->pointerLock); LG_LOCK_FREE(this->recoveryLock); + lgFreeEvent(this->frameWake); + lgFreeEvent(this->pointerWake); free(this); *transport = NULL; } @@ -1087,8 +1108,25 @@ static LG_TransportStatus lgmp_subscribe(PLGMPClient client, uint32_t id, } } +static void lgmp_waitPoll(LGEvent * event, unsigned interval) +{ + if (!interval) + return; + + const uint64_t timeout = (uint64_t)interval * 1000U; + if (timeout < TIMEOUT_INFINITE) + { + lgWaitEventNS(event, (unsigned)timeout); + return; + } + + const unsigned timeoutMS = interval / 1000U + + (interval % 1000U != 0); + lgWaitEvent(event, timeoutMS); +} + static LG_TransportStatus lgmp_process(PLGMPClientQueue * subscription, - unsigned interval, LGMPMessage * message) + unsigned interval, LGEvent * wake, LGMPMessage * message) { PLGMPClientQueue queue = *subscription; if (!queue) @@ -1101,7 +1139,7 @@ static LG_TransportStatus lgmp_process(PLGMPClientQueue * subscription, return LG_TRANSPORT_OK; case LGMP_ERR_QUEUE_EMPTY: if (interval) - usleep(interval); + lgmp_waitPoll(wake, interval); return LG_TRANSPORT_TIMEOUT; case LGMP_ERR_QUEUE_TIMEOUT: case LGMP_ERR_QUEUE_UNSUBSCRIBED: @@ -1154,7 +1192,7 @@ static LG_TransportStatus lgmp_pollFrameQueue( } const LG_TransportStatus status = - lgmp_process(subscription, 0, &result->message); + lgmp_process(subscription, 0, NULL, &result->message); if (status == LG_TRANSPORT_OK) { result->queue = queue; @@ -1581,10 +1619,15 @@ static LG_TransportStatus lgmp_nextFrame(LG_Transport * this, bool useDMA, LG_UNLOCK(this->frameLock); if ((status == LG_TRANSPORT_TIMEOUT || status == LG_TRANSPORT_UNAVAILABLE) && this->framePollInterval) - usleep(this->framePollInterval); + lgmp_waitPoll(this->frameWake, this->framePollInterval); return status; } +static void lgmp_cancelFrameWait(LG_Transport * this) +{ + lgSignalEvent(this->frameWake); +} + static bool lgmp_frameTimingReady(const KVMFRFrame * frame) { return __atomic_load_n(&frame->timingValid, __ATOMIC_ACQUIRE) && @@ -1693,13 +1736,13 @@ static LG_TransportStatus lgmp_nextPointer(LG_Transport * this, pointerQueue = this->pointerQueue; LG_UNLOCK(this->pointerLock); if (status == LG_TRANSPORT_TIMEOUT) - usleep(1000); + lgmp_waitPoll(this->pointerWake, 1000); if (status == LG_TRANSPORT_OK) { LGMPMessage message; PLGMPClientQueue processedQueue = pointerQueue; status = lgmp_process(&processedQueue, this->cursorPollInterval, - &message); + this->pointerWake, &message); if (!processedQueue) lgmp_clearPointerQueue(this, pointerQueue); if (status == LG_TRANSPORT_OK) @@ -1785,6 +1828,11 @@ static LG_TransportStatus lgmp_nextPointer(LG_Transport * this, return LG_TRANSPORT_OK; } +static void lgmp_cancelPointerWait(LG_Transport * this) +{ + lgSignalEvent(this->pointerWake); +} + static void lgmp_releasePointer(LG_Transport * this, LG_TransportPointer * pointer) { @@ -1895,16 +1943,18 @@ static const LG_InputOps * lgmp_getInputOps(LG_Transport * this, static const LG_FrameOps lgmpFrameOps = { - .supportsDMA = lgmp_supportsDMA, - .attachRenderer = lgmp_attachRenderer, - .detachRenderer = lgmp_detachRenderer, - .nextFrame = lgmp_nextFrame, - .getFrameTiming = lgmp_getFrameTiming, - .releaseFrame = lgmp_releaseFrame, - .stopFrame = lgmp_stopFrame, - .nextPointer = lgmp_nextPointer, - .releasePointer = lgmp_releasePointer, - .stopPointer = lgmp_stopPointer, + .supportsDMA = lgmp_supportsDMA, + .attachRenderer = lgmp_attachRenderer, + .detachRenderer = lgmp_detachRenderer, + .nextFrame = lgmp_nextFrame, + .getFrameTiming = lgmp_getFrameTiming, + .releaseFrame = lgmp_releaseFrame, + .cancelFrameWait = lgmp_cancelFrameWait, + .stopFrame = lgmp_stopFrame, + .nextPointer = lgmp_nextPointer, + .releasePointer = lgmp_releasePointer, + .cancelPointerWait = lgmp_cancelPointerWait, + .stopPointer = lgmp_stopPointer, }; static const LG_VideoOps lgmpVideoOps =