[client] transport: make video waits cancellable

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. */

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_stopVideoThreads();
-      core_stopCursorThread();
-      core_stopFrameThread();
-    }
     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();
+      if (!core_startVideoThreads())
-      core_startFrameThread();
+      {
+        atomic_store_explicit(
+            &g_state.stopVideo, true, memory_order_release);
+        app_alert(LG_ALERT_ERROR, "Failed to enable the video stream");
+      }
     }
     else
     {

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)
+  if (!g_state.videoOps || g_state.videoOps->type != LG_VIDEO_TYPE_FRAME)
-    return true;
+    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 = {

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);
+bool core_startVideoThreads(void);
-void core_stopCursorThread(void);
+void core_stopVideoThreads(void);
-bool core_startFrameThread(void);
-void core_stopFrameThread(void);
 void core_handleGuestMouseUpdate(void);
 void core_handleMouseAbsolute(void);
 void core_handleMousePosition(double x, double y);

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)

client/src/main.c

@@ -1101,8 +1101,7 @@ static int renderThread(void * unused)
     lgClipboard_dropTransport();
   }
 
-  core_stopCursorThread();
+  core_stopVideoThreads();
-  core_stopFrameThread();
 
   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_stopVideoThreads();
-      core_stopFrameThread();
-      core_stopCursorThread();
-    }
     else
       g_state.videoOps->swSurface->setActive(
           g_state.transport.handle, false);

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;

client/tests/lgmp_transport_test.c

@@ -28,6 +28,7 @@
 #include <lgmp/host.h>
 
 #include <fcntl.h>
+#include <pthread.h>
 #include <stdatomic.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -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", "framePollInterval", 5000000);
-  option_set_int("lgmp", "cursorPollInterval", 0);
+  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(checkWaitCancellation(
-  CHECK(frameOps->nextPointer(transport, &pointer) == LG_TRANSPORT_TIMEOUT);
+        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);

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,
+  .supportsDMA       = lgmp_supportsDMA,
-  .attachRenderer = lgmp_attachRenderer,
+  .attachRenderer    = lgmp_attachRenderer,
-  .detachRenderer = lgmp_detachRenderer,
+  .detachRenderer    = lgmp_detachRenderer,
-  .nextFrame      = lgmp_nextFrame,
+  .nextFrame         = lgmp_nextFrame,
-  .getFrameTiming = lgmp_getFrameTiming,
+  .getFrameTiming    = lgmp_getFrameTiming,
-  .releaseFrame   = lgmp_releaseFrame,
+  .releaseFrame      = lgmp_releaseFrame,
-  .stopFrame      = lgmp_stopFrame,
+  .cancelFrameWait   = lgmp_cancelFrameWait,
-  .nextPointer    = lgmp_nextPointer,
+  .stopFrame         = lgmp_stopFrame,
-  .releasePointer = lgmp_releasePointer,
+  .nextPointer       = lgmp_nextPointer,
-  .stopPointer    = lgmp_stopPointer,
+  .releasePointer    = lgmp_releasePointer,
+  .cancelPointerWait = lgmp_cancelPointerWait,
+  .stopPointer       = lgmp_stopPointer,
 };
 
 static const LG_VideoOps lgmpVideoOps =