[idd] transport: arbitrate interaction requests

2026-08-22 07:01:30 +00:00 · 2026-08-13 00:20:38 +10:00
parent 5b09d4f90a
commit fbc4640623
10 changed files with 270 additions and 38 deletions
--- a/idd/LGIdd/transport/CInputHub.cpp
+++ b/idd/LGIdd/transport/CInputHub.cpp
@@ -28,6 +28,12 @@ static bool SameSource(const SourceKey& left, const SourceKey& right)
    left.client == right.client && left.generation == right.generation;
 }

+static bool SameClient(const SourceKey& left, const SourceKey& right)
+{
+  return left.backend == right.backend && left.epoch == right.epoch &&
+    left.client == right.client;
+}
+
 CInputHub::CInputHub()
 {
  for (Source& source : m_sources)
@@ -222,6 +228,12 @@ void CInputHub::Unbind(BackendId backend, uint32_t epoch)
          m_owner = {};
          m_ownerDeadline = 0;
        }
+        if (m_interactionOwner.backend == backend &&
+            m_interactionOwner.epoch == epoch)
+        {
+          ClearInteraction();
+        }
+        AdvanceInteractionSerial();
        break;
      }
  }
@@ -245,6 +257,94 @@ void CInputHub::Unbind(BackendId backend, uint32_t epoch)
  }
 }

+InteractionResult CInputHub::CheckInteraction(
+  SourceKey& source, InteractionPermit& permit)
+{
+  permit = {};
+  if (!source.backend || !source.epoch || !source.client)
+    return InteractionResult::STALE;
+
+  CSRWExclusiveLock lock(m_lock);
+  if (m_started && m_sink)
+    CheckState();
+
+  if (m_owner.backend)
+  {
+    if (!SameClient(m_owner, source))
+      return InteractionResult::BUSY;
+    if (source.generation && source.generation != m_owner.generation)
+      return InteractionResult::BUSY;
+    source.generation = m_owner.generation;
+    permit.serial = m_interactionSerial;
+    permit.source = source;
+    return InteractionResult::ACCEPTED;
+  }
+
+  const uint64_t now = GetTickCount64();
+  if (m_interactionOwner.backend && now >= m_interactionDeadline)
+    InvalidateInteraction();
+
+  if (!m_interactionOwner.backend)
+  {
+    permit.serial = m_interactionSerial;
+    permit.source = source;
+    return InteractionResult::ACCEPTED;
+  }
+
+  if (!SameClient(m_interactionOwner, source) ||
+      (source.generation && m_interactionOwner.generation &&
+       source.generation != m_interactionOwner.generation))
+    return InteractionResult::BUSY;
+
+  if (!source.generation)
+    source.generation = m_interactionOwner.generation;
+  permit.serial = m_interactionSerial;
+  permit.source = source;
+  return InteractionResult::ACCEPTED;
+}
+
+void CInputHub::CommitInteraction(
+  const SourceKey& source, const InteractionPermit& permit)
+{
+  if (!source.backend || !source.epoch || !source.client || !permit.serial ||
+      !SameSource(source, permit.source))
+    return;
+
+  CSRWExclusiveLock lock(m_lock);
+  if (m_started && m_sink)
+    CheckState();
+  if (permit.serial != m_interactionSerial)
+    return;
+  if (m_owner.backend)
+    return;
+
+  const uint64_t now = GetTickCount64();
+  if (m_interactionOwner.backend &&
+      (!SameClient(m_interactionOwner, source) ||
+       (source.generation && m_interactionOwner.generation &&
+        source.generation != m_interactionOwner.generation)))
+    return;
+
+  if (!m_interactionOwner.backend)
+    m_interactionOwner = source;
+  else if (!m_interactionOwner.generation)
+    m_interactionOwner.generation = source.generation;
+  m_interactionDeadline = now + INTERACTION_LEASE_MS;
+  AdvanceInteractionSerial();
+}
+
+void CInputHub::RevokeInteraction(BackendId backend, uint32_t epoch)
+{
+  if (!backend || !epoch)
+    return;
+
+  CSRWExclusiveLock lock(m_lock);
+  if (m_interactionOwner.backend == backend &&
+      m_interactionOwner.epoch == epoch)
+    ClearInteraction();
+  AdvanceInteractionSerial();
+}
+
 bool CInputHub::TakeFailure(SourceKey& source)
 {
  CSRWSharedLock lifecycleLock(m_lifecycleLock);
@@ -273,6 +373,7 @@ bool CInputHub::Start(IInputSink& sink)
    m_sink      = &sink;
    m_sinkState = sink.GetState();
    m_started   = true;
+    InvalidateInteraction();
    for (Source& source : m_sources)
      if (source.active)
      {
@@ -298,6 +399,12 @@ bool CInputHub::Start(IInputSink& sink)
          m_owner = {};
          m_ownerDeadline = 0;
        }
+        if (m_interactionOwner.backend == source.backend &&
+            m_interactionOwner.epoch == source.epoch)
+        {
+          ClearInteraction();
+        }
+        AdvanceInteractionSerial();
        source.active         = false;
        source.failed         = true;
        source.failurePending = true;
@@ -329,7 +436,10 @@ void CInputHub::Stop()
  {
    CSRWExclusiveLock lock(m_lock);
    if (!m_started)
+    {
+      InvalidateInteraction();
      return;
+    }
    m_started = false;
    for (Source& source : m_sources)
      if (source.endpoint && source.running)
@@ -341,6 +451,7 @@ void CInputHub::Stop()
    reset = m_owner.backend != 0;
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
  }

  for (unsigned i = count; i > 0; --i)
@@ -389,18 +500,38 @@ bool CInputHub::OwnerValid(const SourceKey& source) const
  return SourceValid(source) && SameSource(m_owner, source);
 }

+void CInputHub::ClearInteraction()
+{
+  m_interactionOwner    = {};
+  m_interactionDeadline = 0;
+}
+
+void CInputHub::InvalidateInteraction()
+{
+  ClearInteraction();
+  AdvanceInteractionSerial();
+}
+
+void CInputHub::AdvanceInteractionSerial()
+{
+  ++m_interactionSerial;
+  if (!m_interactionSerial)
+    ++m_interactionSerial;
+}
+
 bool CInputHub::CheckState()
 {
  if (!m_started || !m_sink)
    return false;

  const uint64_t state = m_sink->GetState();
-  if (state != m_sinkState || !(state & 1))
+  if (state != m_sinkState)
  {
    if (m_owner.backend)
      m_sink->Reset();
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
    m_sinkState     = state;
  }
  else if (m_owner.backend && GetTickCount64() >= m_ownerDeadline)
@@ -408,6 +539,7 @@ bool CInputHub::CheckState()
    m_sink->Reset();
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
  }
  return (state & 1) != 0;
 }
@@ -449,6 +581,12 @@ void CInputHub::Failed(Source& source)
    m_owner = {};
    m_ownerDeadline = 0;
  }
+  if (m_interactionOwner.backend == source.backend &&
+      m_interactionOwner.epoch == source.epoch)
+  {
+    ClearInteraction();
+  }
+  AdvanceInteractionSerial();
  source.active         = false;
  source.failed         = true;
  source.failurePending = true;
@@ -468,10 +606,12 @@ InputResult CInputHub::Claim(const SourceKey& source)
  {
    const uint64_t state = m_sink->GetState();
    m_sinkState = state;
+    InvalidateInteraction();
    return InputResult::UNAVAILABLE;
  }
  m_owner         = source;
  m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS;
+  InvalidateInteraction();
  return InputResult::ACCEPTED;
 }

@@ -500,6 +640,7 @@ InputResult CInputHub::Release(const SourceKey& source, bool reset)
    accepted = m_sink->Reset();
  m_owner         = {};
  m_ownerDeadline = 0;
+  InvalidateInteraction();
  return accepted ? InputResult::ACCEPTED : InputResult::UNAVAILABLE;
 }

@@ -518,6 +659,7 @@ InputResult CInputHub::SendMouseRelative(const SourceKey& source,
    m_sink->Reset();
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
    return InputResult::UNAVAILABLE;
  }
  m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS;
@@ -539,6 +681,7 @@ InputResult CInputHub::SendMouseAbsolute(const SourceKey& source,
    m_sink->Reset();
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
    return InputResult::UNAVAILABLE;
  }
  m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS;
@@ -560,6 +703,7 @@ InputResult CInputHub::SendKeyboard(const SourceKey& source,
    m_sink->Reset();
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
    return InputResult::UNAVAILABLE;
  }
  m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS;
@@ -579,6 +723,7 @@ InputResult CInputHub::Reset(const SourceKey& source)
  {
    m_owner         = {};
    m_ownerDeadline = 0;
+    InvalidateInteraction();
    return InputResult::UNAVAILABLE;
  }
  m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS;
--- a/idd/LGIdd/transport/CInputHub.h
+++ b/idd/LGIdd/transport/CInputHub.h
@@ -31,9 +31,17 @@ class IInputSink;

 class CInputHub final : public IInputTransport
 {
+public:
+  struct InteractionPermit
+  {
+    uint64_t serial = 0;
+    SourceKey source;
+  };
+
 private:
  static constexpr unsigned MAX_SOURCES = 8;
  static constexpr uint64_t OWNER_LEASE_MS = 500;
+  static constexpr uint64_t INTERACTION_LEASE_MS = 500;

  struct Source final : public IInputTarget
  {
@@ -72,12 +80,18 @@ private:
  uint64_t         m_sinkState = 0;
  SourceKey        m_owner;
  uint64_t         m_ownerDeadline = 0;
+  SourceKey        m_interactionOwner;
+  uint64_t         m_interactionDeadline = 0;
+  uint64_t         m_interactionSerial = 1;
  bool             m_started = false;

  bool SourceValid(const SourceKey& source) const;
  bool BindingPresent(const SourceKey& source) const;
  bool BindingValid(const SourceKey& source) const;
  bool OwnerValid(const SourceKey& source) const;
+  void ClearInteraction();
+  void InvalidateInteraction();
+  void AdvanceInteractionSerial();
  bool CheckState();
  InputTargetState GetState(const SourceKey& source);
  void Failed(Source& source);
@@ -101,6 +115,11 @@ public:

  bool Bind(BackendId backend, uint32_t epoch, IInputSource& source);
  void Unbind(BackendId backend, uint32_t epoch);
+  InteractionResult CheckInteraction(
+    SourceKey& source, InteractionPermit& permit);
+  void CommitInteraction(
+    const SourceKey& source, const InteractionPermit& permit);
+  void RevokeInteraction(BackendId backend, uint32_t epoch);
  bool TakeFailure(SourceKey& source);

  bool Start(IInputSink& sink) override;
--- a/idd/LGIdd/transport/CTransportManager.cpp
+++ b/idd/LGIdd/transport/CTransportManager.cpp
@@ -34,6 +34,8 @@ class CSourceEvents final : public ITransportEvents
 private:
  BackendId         m_backend;
  uint32_t          m_epoch;
+  bool              m_interactions;
+  CInputHub&         m_input;
  ITransportEvents& m_events;

  SourceKey Stamp(const SourceKey& source) const
@@ -46,19 +48,45 @@ private:

 public:
  CSourceEvents(
-    BackendId backend, uint32_t epoch, ITransportEvents& events) :
-    m_backend(backend), m_epoch(epoch), m_events(events) {}
+    BackendId backend, uint32_t epoch, bool interactions, CInputHub& input,
+    ITransportEvents& events) :
+    m_backend(backend), m_epoch(epoch), m_interactions(interactions),
+    m_input(input), m_events(events) {}

-  void OnSetCursorPos(
+  InteractionResult OnSetCursorPos(
    const SourceKey& source, int32_t x, int32_t y) override
  {
-    m_events.OnSetCursorPos(Stamp(source), x, y);
+    if (!m_interactions)
+      return InteractionResult::UNAVAILABLE;
+    SourceKey stamped = Stamp(source);
+    CInputHub::InteractionPermit permit;
+    const InteractionResult result =
+      m_input.CheckInteraction(stamped, permit);
+    if (result != InteractionResult::ACCEPTED)
+      return result;
+    const InteractionResult applied =
+      m_events.OnSetCursorPos(stamped, x, y);
+    if (applied == InteractionResult::ACCEPTED)
+      m_input.CommitInteraction(stamped, permit);
+    return applied;
  }

-  void OnSetResolution(const SourceKey& source,
+  InteractionResult OnSetResolution(const SourceKey& source,
    uint32_t width, uint32_t height) override
  {
-    m_events.OnSetResolution(Stamp(source), width, height);
+    if (!m_interactions)
+      return InteractionResult::UNAVAILABLE;
+    SourceKey stamped = Stamp(source);
+    CInputHub::InteractionPermit permit;
+    const InteractionResult result =
+      m_input.CheckInteraction(stamped, permit);
+    if (result != InteractionResult::ACCEPTED)
+      return result;
+    const InteractionResult applied =
+      m_events.OnSetResolution(stamped, width, height);
+    if (applied == InteractionResult::ACCEPTED)
+      m_input.CommitInteraction(stamped, permit);
+    return applied;
  }

  void OnRecoveryRequest(const SourceKey& source,
@@ -518,6 +546,7 @@ void CTransportManager::HandleServiceFailures()
      }

      m_control.Remove(token.backend, token.epoch);
+      m_input.RevokeInteraction(token.backend, token.epoch);
      if (restart)
      {
        RemoveServices(entry);
@@ -622,6 +651,7 @@ void CTransportManager::RemoveServices(Entry& entry)
    entry.inputAdded     = false;
  }

+  m_input.RevokeInteraction(id, epoch);
  if (inputAdded)
    m_input.Unbind(id, epoch);
  if (controlAdded)
@@ -988,8 +1018,14 @@ ITransport::ProcessResult CTransportManager::Process(
    if (setup)
      SetupEntry(entry, alignment);

+    bool interactions = false;
+    {
+      CSRWSharedLock entryLock(entry.lock);
+      interactions = entry.controlAdded;
+    }
    DrainRecovery(entry, transport);
-    CSourceEvents sourceEvents(id, epoch, events);
+    CSourceEvents sourceEvents(
+      id, epoch, interactions, m_input, events);
    const ProcessResult result = transport->Process(sourceEvents);
    DrainRecovery(entry, transport);
    HandleProcessResult(entry, result);
--- a/idd/LGIdd/transport/ITransport.h
+++ b/idd/LGIdd/transport/ITransport.h
@@ -40,14 +40,24 @@ struct SourceKey
  uint32_t  generation = 0;
 };

+enum class InteractionResult
+{
+  ACCEPTED,
+  BUSY,
+  UNAVAILABLE,
+  STALE,
+  REJECTED,
+  FAILED,
+};
+
 class ITransportEvents
 {
 public:
  virtual ~ITransportEvents() = default;

-  virtual void OnSetCursorPos(
+  virtual InteractionResult OnSetCursorPos(
    const SourceKey& source, int32_t x, int32_t y) = 0;
-  virtual void OnSetResolution(
+  virtual InteractionResult OnSetResolution(
    const SourceKey& source, uint32_t width, uint32_t height) = 0;
  virtual void OnRecoveryRequest(const SourceKey& source,
    uint64_t session, uint32_t serial, bool active) = 0;
--- a/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp
+++ b/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp
@@ -138,26 +138,43 @@ ITransport::ProcessResult CLGMPTransport::Process(ITransportEvents& events)
  while ((status = m_control.ReadDataWithSource(
      data, &size, &sourceClientID)) == LGMP_OK)
  {
+    if (size < sizeof(KVMFRMessage))
+    {
+      DEBUG_WARN("Ignoring invalid KVMFR message");
+      m_control.AckData();
+      continue;
+    }
+
    KVMFRMessage * msg = reinterpret_cast<KVMFRMessage *>(data);
    switch (msg->type)
    {
      case KVMFR_MESSAGE_SETCURSORPOS:
      {
-        SourceKey source;
-        source.client = sourceClientID;
-        KVMFRSetCursorPos * position =
-          reinterpret_cast<KVMFRSetCursorPos *>(msg);
-        events.OnSetCursorPos(source, position->x, position->y);
+        if (size == sizeof(KVMFRSetCursorPos))
+        {
+          SourceKey source;
+          source.client = sourceClientID;
+          const KVMFRSetCursorPos * position =
+            reinterpret_cast<const KVMFRSetCursorPos *>(msg);
+          events.OnSetCursorPos(source, position->x, position->y);
+        }
+        else
+          DEBUG_WARN("Ignoring invalid KVMFR cursor position");
        break;
      }

      case KVMFR_MESSAGE_WINDOWSIZE:
      {
-        SourceKey source;
-        source.client = sourceClientID;
-        KVMFRWindowSize * window =
-          reinterpret_cast<KVMFRWindowSize *>(msg);
-        events.OnSetResolution(source, window->w, window->h);
+        if (size == sizeof(KVMFRWindowSize))
+        {
+          SourceKey source;
+          source.client = sourceClientID;
+          const KVMFRWindowSize * window =
+            reinterpret_cast<const KVMFRWindowSize *>(msg);
+          events.OnSetResolution(source, window->w, window->h);
+        }
+        else
+          DEBUG_WARN("Ignoring invalid KVMFR window size");
        break;
      }