diff --git a/common/include/common/KVMFRRecovery.h b/common/include/common/KVMFRRecovery.h
index e0fa1ed0..bbfac983 100644
--- a/common/include/common/KVMFRRecovery.h
+++ b/common/include/common/KVMFRRecovery.h
@@ -68,7 +68,9 @@ enum
KVMFR_R_ERR_UNSUPPORTED = 1,
KVMFR_R_ERR_HELPER_UNAVAILABLE = 2,
KVMFR_R_ERR_TOPOLOGY_FAILED = 3,
- KVMFR_R_ERR_NO_FALLBACK_DISPLAY = 4
+ KVMFR_R_ERR_NO_FALLBACK_DISPLAY = 4,
+ KVMFR_R_ERR_BUSY = 5,
+ KVMFR_R_ERR_CAPACITY = 6
};
/*
diff --git a/idd/LGIdd/LGIdd.vcxproj b/idd/LGIdd/LGIdd.vcxproj
index ee52da65..bec6dd07 100644
--- a/idd/LGIdd/LGIdd.vcxproj
+++ b/idd/LGIdd/LGIdd.vcxproj
@@ -63,6 +63,7 @@
+
@@ -115,6 +116,7 @@
+
diff --git a/idd/LGIdd/LGIdd.vcxproj.filters b/idd/LGIdd/LGIdd.vcxproj.filters
index bcbb8cbd..8a83bf5c 100644
--- a/idd/LGIdd/LGIdd.vcxproj.filters
+++ b/idd/LGIdd/LGIdd.vcxproj.filters
@@ -163,6 +163,9 @@
Transport
+
+ Transport
+
Transport
@@ -327,6 +330,9 @@
Transport
+
+ Transport
+
Transport
diff --git a/idd/LGIdd/display/CDeviceContext.cpp b/idd/LGIdd/display/CDeviceContext.cpp
index 16348aa0..25fc5f85 100644
--- a/idd/LGIdd/display/CDeviceContext.cpp
+++ b/idd/LGIdd/display/CDeviceContext.cpp
@@ -493,25 +493,6 @@ bool CDeviceContext::InitializeTransport()
CDeviceContext * context =
static_cast(opaque);
- SourceKey source;
- {
- CSRWExclusiveLock routeLock(context->m_recoveryRouteLock);
- if (!route || route != context->m_recoveryRoute ||
- session != context->m_recoverySession ||
- serial != context->m_recoverySerial ||
- active != context->m_recoveryActive)
- {
- DEBUG_WARN("Ignoring stale recovery route");
- return;
- }
- source = context->m_recoverySource;
- context->m_recoveryRoute = 0;
- context->m_recoverySource = {};
- context->m_recoverySession = 0;
- context->m_recoverySerial = 0;
- context->m_recoveryActive = false;
- }
-
ITransport::Recovery state = ITransport::Recovery::FAILED;
uint32_t error = ERROR_SUCCESS;
switch (result)
@@ -537,7 +518,7 @@ bool CDeviceContext::InitializeTransport()
}
context->m_transport->RecoveryStatus(
- source, session, serial, active, state, error);
+ route, session, serial, active, state, error);
},
this);
m_recoveryHandlerSet = true;
@@ -639,37 +620,8 @@ InteractionResult CDeviceContext::OnSetResolution(const SourceKey& source,
return SetResolution(width, height);
}
-void CDeviceContext::OnRecoveryRequest(const SourceKey& source,
- uint64_t session, uint32_t serial, bool active)
+bool CDeviceContext::OnRecoveryAction(const RecoveryAction& action)
{
- if (!source.backend || !source.epoch || !session || !serial ||
- (serial & LGPipeMsg::RECOVERY_ACTIVE))
- return;
-
- CSRWExclusiveLock publishLock(m_recoveryPublishLock);
- uint64_t route;
- {
- CSRWExclusiveLock routeLock(m_recoveryRouteLock);
- route = m_nextRecoveryRoute++;
- if (!m_nextRecoveryRoute)
- ++m_nextRecoveryRoute;
-
- m_recoveryRoute = route;
- m_recoverySource = source;
- m_recoverySession = session;
- m_recoverySerial = serial;
- m_recoveryActive = active;
- }
-
- if (!g_pipe.SetRecovery(this, route, session, serial, active))
- {
- CSRWExclusiveLock routeLock(m_recoveryRouteLock);
- if (m_recoveryRoute != route)
- return;
- m_recoveryRoute = 0;
- m_recoverySource = {};
- m_recoverySession = 0;
- m_recoverySerial = 0;
- m_recoveryActive = false;
- }
+ return g_pipe.SetRecovery(this, action.route, action.session,
+ action.serial, action.active);
}
diff --git a/idd/LGIdd/display/CDeviceContext.h b/idd/LGIdd/display/CDeviceContext.h
index b73d9fb0..2ce8acac 100644
--- a/idd/LGIdd/display/CDeviceContext.h
+++ b/idd/LGIdd/display/CDeviceContext.h
@@ -33,7 +33,7 @@
#include "display/CMonitorManager.h"
#include "transport/CTransportManager.h"
-class CDeviceContext : private ITransportEvents
+class CDeviceContext : private ITransportActions
{
private:
WDFDEVICE m_wdfDevice;
@@ -53,14 +53,6 @@ private:
WDFTIMER m_transportTimer = nullptr;
bool m_recoveryHandlerSet = false;
- CSRWLock m_recoveryPublishLock;
- CSRWLock m_recoveryRouteLock;
- uint64_t m_nextRecoveryRoute = 1;
- uint64_t m_recoveryRoute = 0;
- SourceKey m_recoverySource;
- uint64_t m_recoverySession = 0;
- uint32_t m_recoverySerial = 0;
- bool m_recoveryActive = false;
UINT m_iddCxVersion = 0;
bool m_hasIddCx110DDIs = false;
@@ -78,8 +70,7 @@ private:
const SourceKey& source, int32_t x, int32_t y) override;
InteractionResult OnSetResolution(const SourceKey& source,
uint32_t width, uint32_t height) override;
- void OnRecoveryRequest(const SourceKey& source,
- uint64_t session, uint32_t serial, bool active) override;
+ bool OnRecoveryAction(const RecoveryAction& action) override;
InteractionResult SetResolution(uint32_t width, uint32_t height);
public:
diff --git a/idd/LGIdd/transport/CRecoveryHub.cpp b/idd/LGIdd/transport/CRecoveryHub.cpp
new file mode 100644
index 00000000..45d3e7d6
--- /dev/null
+++ b/idd/LGIdd/transport/CRecoveryHub.cpp
@@ -0,0 +1,416 @@
+/**
+ * Looking Glass
+ * Copyright © 2017-2026 The Looking Glass Authors
+ * https://looking-glass.io
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the Free
+ * Software Foundation; either version 2 of the License, or (at your option)
+ * any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc., 59
+ * Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#include "transport/CRecoveryHub.h"
+
+#include
+
+namespace
+{
+ uint64_t CreateSession(const void * owner)
+ {
+ LARGE_INTEGER counter;
+ QueryPerformanceCounter(&counter);
+
+ uint64_t session = static_cast(counter.QuadPart) ^
+ (GetTickCount64() << 24) ^
+ static_cast(reinterpret_cast(owner)) ^
+ (static_cast(GetCurrentProcessId()) << 32) ^
+ GetCurrentThreadId();
+ if (!session)
+ ++session;
+ return session;
+ }
+}
+
+CRecoveryHub::CRecoveryHub() :
+ m_session(CreateSession(this))
+{
+}
+
+bool CRecoveryHub::SameSource(
+ const SourceKey& left, const SourceKey& right)
+{
+ return left.backend == right.backend && left.epoch == right.epoch &&
+ left.client == right.client && left.generation == right.generation;
+}
+
+bool CRecoveryHub::SameEpoch(
+ const SourceKey& source, BackendId backend, uint32_t epoch)
+{
+ return source.backend == backend && source.epoch == epoch;
+}
+
+bool CRecoveryHub::SameRequest(const Request& request,
+ const SourceKey& source, uint64_t session, uint32_t serial, bool active)
+{
+ return SameSource(request.source, source) && request.session == session &&
+ request.serial == serial && request.active == active;
+}
+
+bool CRecoveryHub::SuccessMatches(RecoveryState state, bool active)
+{
+ return (state == RecoveryState::ACTIVE && active) ||
+ (state == RecoveryState::NORMAL && !active);
+}
+
+bool CRecoveryHub::EpochAttachedLocked(const SourceKey& source) const
+{
+ for (const Epoch& epoch : m_epochs)
+ if (epoch.backend == source.backend && epoch.epoch == source.epoch)
+ return true;
+ return false;
+}
+
+unsigned CRecoveryHub::FindSourceLocked(const SourceKey& source) const
+{
+ for (unsigned i = 0; i < MAX_REQUESTS; ++i)
+ if (m_requests[i].state != SlotState::FREE &&
+ SameSource(m_requests[i].source, source))
+ return i;
+ return MAX_REQUESTS;
+}
+
+unsigned CRecoveryHub::FindFreeLocked() const
+{
+ for (unsigned i = 0; i < MAX_REQUESTS; ++i)
+ if (m_requests[i].state == SlotState::FREE)
+ return i;
+ return MAX_REQUESTS;
+}
+
+bool CRecoveryHub::ActionMatchesLocked(
+ const RecoveryAction& action) const
+{
+ return m_operation.phase != OperationPhase::NONE &&
+ action.route == m_operation.action.route &&
+ action.session == m_operation.action.session &&
+ action.serial == m_operation.action.serial &&
+ action.active == m_operation.action.active;
+}
+
+uint64_t CRecoveryHub::NextNonzero(uint64_t& value)
+{
+ uint64_t result = value++;
+ if (!result)
+ result = value++;
+ if (!value)
+ ++value;
+ return result;
+}
+
+uint32_t CRecoveryHub::NextSerial()
+{
+ uint32_t result = m_nextSerial;
+ m_nextSerial += 2;
+ if (!result)
+ {
+ result = 2;
+ m_nextSerial = 4;
+ }
+ return result;
+}
+
+void CRecoveryHub::ClearRequestLocked(Request& request)
+{
+ request = Request {};
+}
+
+void CRecoveryHub::SetWaitingLocked(Request& request,
+ const SourceKey& source, uint64_t session, uint32_t serial,
+ bool active, uint64_t operation)
+{
+ request = Request {};
+ request.source = source;
+ request.session = session;
+ request.operation = operation;
+ request.sequence = NextNonzero(m_nextSequence);
+ request.serial = serial;
+ request.active = active;
+ request.state = SlotState::WAITING;
+}
+
+void CRecoveryHub::SetReadyLocked(
+ Request& request, RecoveryState state, uint32_t error)
+{
+ request.state = SlotState::READY;
+ request.result = state;
+ request.error = error;
+}
+
+void CRecoveryHub::FinishWaitersLocked(
+ RecoveryState state, uint32_t error)
+{
+ for (Request& request : m_requests)
+ if (request.state == SlotState::WAITING &&
+ request.operation == m_operation.id)
+ SetReadyLocked(request, state, error);
+}
+
+bool CRecoveryHub::Attach(
+ BackendId backend, uint32_t epoch, bool& syncNow)
+{
+ syncNow = false;
+ if (!backend || !epoch)
+ return false;
+
+ CSRWExclusiveLock lock(m_lock);
+ for (const Epoch& item : m_epochs)
+ if (item.backend == backend && item.epoch == epoch)
+ return true;
+
+ for (Epoch& item : m_epochs)
+ if (!item.backend)
+ {
+ item.backend = backend;
+ item.epoch = epoch;
+ item.synced = m_monitorReady;
+ syncNow = m_monitorReady;
+ return true;
+ }
+ return false;
+}
+
+void CRecoveryHub::Remove(BackendId backend, uint32_t epoch)
+{
+ if (!backend || !epoch)
+ return;
+
+ CSRWExclusiveLock lock(m_lock);
+ for (Epoch& item : m_epochs)
+ if (item.backend == backend && item.epoch == epoch)
+ item = Epoch {};
+
+ for (Request& request : m_requests)
+ if (request.state != SlotState::FREE &&
+ SameEpoch(request.source, backend, epoch))
+ ClearRequestLocked(request);
+}
+
+RecoveryAdmission CRecoveryHub::Submit(const SourceKey& source,
+ uint64_t session, uint32_t serial, bool active, uint64_t now,
+ RecoveryAction& action, bool& dispatch)
+{
+ action = RecoveryAction {};
+ dispatch = false;
+
+ RecoveryAdmission result;
+ result.complete = true;
+ result.state = RecoveryState::FAILED;
+ result.error = ERROR_INVALID_PARAMETER;
+
+ CSRWExclusiveLock lock(m_lock);
+ if (!source.backend || !source.epoch || !session || !serial ||
+ !EpochAttachedLocked(source))
+ return result;
+
+ unsigned index = FindSourceLocked(source);
+ if (index != MAX_REQUESTS && SameRequest(
+ m_requests[index], source, session, serial, active))
+ {
+ result.complete = false;
+ result.error = ERROR_SUCCESS;
+ return result;
+ }
+
+ if (index != MAX_REQUESTS)
+ ClearRequestLocked(m_requests[index]);
+ else
+ index = FindFreeLocked();
+
+ if (m_operation.phase == OperationPhase::IN_FLIGHT)
+ {
+ if (active != m_operation.action.active)
+ {
+ result.error = ERROR_BUSY;
+ return result;
+ }
+ if (index == MAX_REQUESTS)
+ {
+ result.error = ERROR_NOT_ENOUGH_QUOTA;
+ return result;
+ }
+
+ SetWaitingLocked(m_requests[index], source, session, serial,
+ active, m_operation.id);
+ result.complete = false;
+ result.error = ERROR_SUCCESS;
+ return result;
+ }
+
+ if (m_knownValid &&
+ ((active && m_known == RecoveryState::ACTIVE) ||
+ (!active && m_known == RecoveryState::NORMAL)))
+ {
+ result.state = m_known;
+ result.error = ERROR_SUCCESS;
+ return result;
+ }
+
+ if (index == MAX_REQUESTS)
+ {
+ result.error = ERROR_NOT_ENOUGH_QUOTA;
+ return result;
+ }
+
+ m_operation = Operation {};
+ m_operation.phase = OperationPhase::IN_FLIGHT;
+ m_operation.id = NextNonzero(m_nextOperation);
+ m_operation.action.route = NextNonzero(m_nextRoute);
+ m_operation.action.session = m_session;
+ m_operation.action.serial = NextSerial();
+ m_operation.action.active = active;
+ m_operation.action.deadline = now + HELPER_TIMEOUT_MS;
+ SetWaitingLocked(m_requests[index], source, session, serial,
+ active, m_operation.id);
+
+ action = m_operation.action;
+ dispatch = true;
+ result.complete = false;
+ result.error = ERROR_SUCCESS;
+ return result;
+}
+
+bool CRecoveryHub::DispatchFailed(
+ const RecoveryAction& action, uint32_t error)
+{
+ CSRWExclusiveLock lock(m_lock);
+ if (!ActionMatchesLocked(action) ||
+ m_operation.phase != OperationPhase::IN_FLIGHT)
+ return false;
+
+ FinishWaitersLocked(RecoveryState::FAILED,
+ error ? error : RPC_S_SERVER_UNAVAILABLE);
+ m_operation = Operation {};
+ return true;
+}
+
+bool CRecoveryHub::Complete(const RecoveryAction& action,
+ RecoveryState state, uint32_t error)
+{
+ CSRWExclusiveLock lock(m_lock);
+ if (!ActionMatchesLocked(action))
+ return false;
+
+ if (!SuccessMatches(state, m_operation.action.active))
+ {
+ state = RecoveryState::FAILED;
+ if (!error)
+ error = ERROR_GEN_FAILURE;
+ m_knownValid = false;
+ }
+ else
+ {
+ error = ERROR_SUCCESS;
+ m_known = state;
+ m_knownValid = true;
+ }
+
+ if (m_operation.phase == OperationPhase::LATCHED)
+ return true;
+
+ FinishWaitersLocked(state, error);
+ m_operation.phase = OperationPhase::LATCHED;
+ m_operation.action.deadline = 0;
+ return true;
+}
+
+bool CRecoveryHub::Tick(uint64_t now)
+{
+ CSRWExclusiveLock lock(m_lock);
+ if (m_operation.phase != OperationPhase::IN_FLIGHT ||
+ now < m_operation.action.deadline)
+ return false;
+
+ FinishWaitersLocked(RecoveryState::FAILED, ERROR_TIMEOUT);
+ m_knownValid = false;
+ m_operation.phase = OperationPhase::LATCHED;
+ m_operation.action.deadline = 0;
+ return true;
+}
+
+bool CRecoveryHub::TakeDelivery(
+ BackendId backend, uint32_t epoch, Delivery& delivery)
+{
+ CSRWExclusiveLock lock(m_lock);
+ unsigned selected = MAX_REQUESTS;
+ uint64_t sequence = 0;
+ for (unsigned i = 0; i < MAX_REQUESTS; ++i)
+ if (m_requests[i].state == SlotState::READY &&
+ SameEpoch(m_requests[i].source, backend, epoch) &&
+ (selected == MAX_REQUESTS || m_requests[i].sequence < sequence))
+ {
+ selected = i;
+ sequence = m_requests[i].sequence;
+ }
+
+ if (selected == MAX_REQUESTS)
+ return false;
+
+ const Request& request = m_requests[selected];
+ delivery.source = request.source;
+ delivery.session = request.session;
+ delivery.serial = request.serial;
+ delivery.active = request.active;
+ delivery.state = request.result;
+ delivery.error = request.error;
+ ClearRequestLocked(m_requests[selected]);
+ return true;
+}
+
+bool CRecoveryHub::HasDelivery(
+ BackendId backend, uint32_t epoch) const
+{
+ CSRWSharedLock lock(m_lock);
+ for (const Request& request : m_requests)
+ if (request.state == SlotState::READY &&
+ SameEpoch(request.source, backend, epoch))
+ return true;
+ return false;
+}
+
+bool CRecoveryHub::MarkMonitorReady()
+{
+ CSRWExclusiveLock lock(m_lock);
+ if (m_monitorReady)
+ return false;
+ m_monitorReady = true;
+ return true;
+}
+
+bool CRecoveryHub::ClaimSync(BackendId backend, uint32_t epoch)
+{
+ CSRWExclusiveLock lock(m_lock);
+ if (!m_monitorReady)
+ return false;
+ for (Epoch& item : m_epochs)
+ if (item.backend == backend && item.epoch == epoch && !item.synced)
+ {
+ item.synced = true;
+ return true;
+ }
+ return false;
+}
+
+bool CRecoveryHub::MonitorReady() const
+{
+ CSRWSharedLock lock(m_lock);
+ return m_monitorReady;
+}
diff --git a/idd/LGIdd/transport/CRecoveryHub.h b/idd/LGIdd/transport/CRecoveryHub.h
new file mode 100644
index 00000000..793a6baa
--- /dev/null
+++ b/idd/LGIdd/transport/CRecoveryHub.h
@@ -0,0 +1,140 @@
+/**
+ * Looking Glass
+ * Copyright © 2017-2026 The Looking Glass Authors
+ * https://looking-glass.io
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License as published by the Free
+ * Software Foundation; either version 2 of the License, or (at your option)
+ * any later version.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
+ * more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc., 59
+ * Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#pragma once
+
+#include "CSRWLock.h"
+#include "transport/ITransport.h"
+
+class CRecoveryHub
+{
+public:
+ struct Delivery
+ {
+ SourceKey source;
+ uint64_t session = 0;
+ uint32_t serial = 0;
+ bool active = false;
+ RecoveryState state = RecoveryState::FAILED;
+ uint32_t error = 0;
+ };
+
+private:
+ static constexpr unsigned MAX_EPOCHS = TRANSPORT_MAX_INSTANCES;
+ // Reserve two independent request correlations per configured instance.
+ static constexpr unsigned MAX_REQUESTS = TRANSPORT_MAX_INSTANCES * 2;
+ static constexpr uint64_t HELPER_TIMEOUT_MS = 20000;
+
+ enum class SlotState
+ {
+ FREE,
+ WAITING,
+ READY,
+ };
+
+ enum class OperationPhase
+ {
+ NONE,
+ IN_FLIGHT,
+ LATCHED,
+ };
+
+ struct Epoch
+ {
+ BackendId backend = 0;
+ uint32_t epoch = 0;
+ bool synced = false;
+ };
+
+ struct Request
+ {
+ SourceKey source;
+ uint64_t session = 0;
+ uint64_t operation = 0;
+ uint64_t sequence = 0;
+ uint32_t serial = 0;
+ uint32_t error = 0;
+ bool active = false;
+ SlotState state = SlotState::FREE;
+ RecoveryState result = RecoveryState::FAILED;
+ };
+
+ struct Operation
+ {
+ OperationPhase phase = OperationPhase::NONE;
+ RecoveryAction action;
+ uint64_t id = 0;
+ };
+
+ mutable CSRWLock m_lock;
+ Epoch m_epochs[MAX_EPOCHS];
+ Request m_requests[MAX_REQUESTS];
+ Operation m_operation;
+ RecoveryState m_known = RecoveryState::FAILED;
+ uint64_t m_session = 0;
+ uint64_t m_nextOperation = 1;
+ uint64_t m_nextRoute = 1;
+ uint64_t m_nextSequence = 1;
+ uint32_t m_nextSerial = 2;
+ bool m_knownValid = false;
+ bool m_monitorReady = false;
+
+ static bool SameSource(const SourceKey& left, const SourceKey& right);
+ static bool SameEpoch(
+ const SourceKey& source, BackendId backend, uint32_t epoch);
+ static bool SameRequest(const Request& request, const SourceKey& source,
+ uint64_t session, uint32_t serial, bool active);
+ static bool SuccessMatches(RecoveryState state, bool active);
+
+ bool EpochAttachedLocked(const SourceKey& source) const;
+ unsigned FindSourceLocked(const SourceKey& source) const;
+ unsigned FindFreeLocked() const;
+ bool ActionMatchesLocked(const RecoveryAction& action) const;
+ uint64_t NextNonzero(uint64_t& value);
+ uint32_t NextSerial();
+ void ClearRequestLocked(Request& request);
+ void SetWaitingLocked(Request& request, const SourceKey& source,
+ uint64_t session, uint32_t serial, bool active, uint64_t operation);
+ void SetReadyLocked(
+ Request& request, RecoveryState state, uint32_t error);
+ void FinishWaitersLocked(RecoveryState state, uint32_t error);
+
+public:
+ CRecoveryHub();
+
+ bool Attach(BackendId backend, uint32_t epoch, bool& syncNow);
+ void Remove(BackendId backend, uint32_t epoch);
+
+ RecoveryAdmission Submit(const SourceKey& source, uint64_t session,
+ uint32_t serial, bool active, uint64_t now,
+ RecoveryAction& action, bool& dispatch);
+ bool DispatchFailed(const RecoveryAction& action, uint32_t error);
+ bool Complete(const RecoveryAction& action,
+ RecoveryState state, uint32_t error);
+ bool Tick(uint64_t now);
+
+ bool TakeDelivery(
+ BackendId backend, uint32_t epoch, Delivery& delivery);
+ bool HasDelivery(BackendId backend, uint32_t epoch) const;
+
+ bool MarkMonitorReady();
+ bool ClaimSync(BackendId backend, uint32_t epoch);
+ bool MonitorReady() const;
+};
diff --git a/idd/LGIdd/transport/CTransportManager.cpp b/idd/LGIdd/transport/CTransportManager.cpp
index 4b865e15..83b47196 100644
--- a/idd/LGIdd/transport/CTransportManager.cpp
+++ b/idd/LGIdd/transport/CTransportManager.cpp
@@ -36,7 +36,8 @@ private:
uint32_t m_epoch;
bool m_interactions;
CInputHub& m_input;
- ITransportEvents& m_events;
+ CRecoveryHub& m_recovery;
+ ITransportActions& m_actions;
SourceKey Stamp(const SourceKey& source) const
{
@@ -49,9 +50,9 @@ private:
public:
CSourceEvents(
BackendId backend, uint32_t epoch, bool interactions, CInputHub& input,
- ITransportEvents& events) :
+ CRecoveryHub& recovery, ITransportActions& actions) :
m_backend(backend), m_epoch(epoch), m_interactions(interactions),
- m_input(input), m_events(events) {}
+ m_input(input), m_recovery(recovery), m_actions(actions) {}
InteractionResult OnSetCursorPos(
const SourceKey& source, int32_t x, int32_t y) override
@@ -65,7 +66,7 @@ public:
if (result != InteractionResult::ACCEPTED)
return result;
const InteractionResult applied =
- m_events.OnSetCursorPos(stamped, x, y);
+ m_actions.OnSetCursorPos(stamped, x, y);
if (applied == InteractionResult::ACCEPTED)
m_input.CommitInteraction(stamped, permit);
return applied;
@@ -83,17 +84,23 @@ public:
if (result != InteractionResult::ACCEPTED)
return result;
const InteractionResult applied =
- m_events.OnSetResolution(stamped, width, height);
+ m_actions.OnSetResolution(stamped, width, height);
if (applied == InteractionResult::ACCEPTED)
m_input.CommitInteraction(stamped, permit);
return applied;
}
- void OnRecoveryRequest(const SourceKey& source,
+ RecoveryAdmission OnRecoveryRequest(const SourceKey& source,
uint64_t session, uint32_t serial, bool active) override
{
- m_events.OnRecoveryRequest(
- Stamp(source), session, serial, active);
+ RecoveryAction action;
+ bool dispatch = false;
+ const RecoveryAdmission admission = m_recovery.Submit(
+ Stamp(source), session, serial, active, GetTickCount64(),
+ action, dispatch);
+ if (dispatch && !m_actions.OnRecoveryAction(action))
+ m_recovery.DispatchFailed(action, RPC_S_SERVER_UNAVAILABLE);
+ return admission;
}
};
@@ -212,12 +219,12 @@ bool CTransportManager::BeginCall(
void CTransportManager::DrainRecovery(Entry& entry,
const std::shared_ptr& transport)
{
- static const unsigned MAX_DRAIN = 8;
+ static const unsigned MAX_DRAIN = 17;
for (unsigned i = 0; i < MAX_DRAIN; ++i)
{
bool sync = false;
- bool update = false;
- RecoveryUpdate recovery;
+ BackendId id = 0;
+ uint32_t epoch = 0;
{
CSRWExclusiveLock entryLock(entry.lock);
if (entry.stopRequested || !transport ||
@@ -225,25 +232,28 @@ void CTransportManager::DrainRecovery(Entry& entry,
(entry.state != State::INITIALIZED &&
entry.state != State::READY))
{
- entry.syncPending = false;
- entry.recoveryPending = false;
+ entry.syncPending = false;
return;
}
- sync = entry.syncPending;
- update = entry.recoveryPending;
- recovery = entry.recovery;
- entry.syncPending = false;
- entry.recoveryPending = false;
+ id = entry.id;
+ epoch = entry.epoch;
+ sync = entry.syncPending;
+ entry.syncPending = false;
}
- if (!sync && !update)
- return;
if (sync)
transport->SyncRecovery();
- if (update)
- transport->RecoveryStatus(recovery.session, recovery.serial,
- recovery.active, recovery.state, recovery.error);
+
+ CRecoveryHub::Delivery delivery;
+ if (!m_recovery.TakeDelivery(id, epoch, delivery))
+ {
+ if (!sync)
+ return;
+ continue;
+ }
+ transport->RecoveryStatus(delivery.source, delivery.session,
+ delivery.serial, delivery.active, delivery.state, delivery.error);
}
}
@@ -256,7 +266,7 @@ void CTransportManager::EndCall(Entry& entry,
DrainRecovery(entry, transport);
CSRWExclusiveLock entryLock(entry.lock);
- if (drain && (entry.syncPending || entry.recoveryPending))
+ if (drain && entry.syncPending)
continue;
entry.call = Call::IDLE;
@@ -266,6 +276,23 @@ void CTransportManager::EndCall(Entry& entry,
}
}
+void CTransportManager::DetachRecovery(Entry& entry)
+{
+ BackendId id = 0;
+ uint32_t epoch = 0;
+ bool attached = false;
+ {
+ CSRWExclusiveLock entryLock(entry.lock);
+ id = entry.id;
+ epoch = entry.epoch;
+ attached = entry.recoveryAttached;
+ entry.recoveryAttached = false;
+ entry.syncPending = false;
+ }
+ if (attached)
+ m_recovery.Remove(id, epoch);
+}
+
bool CTransportManager::Add(TransportInstance config, bool primary,
CreateFn create)
{
@@ -356,6 +383,8 @@ ITransport::OpenResult CTransportManager::OpenEntry(Entry& entry)
bool CTransportManager::InitializeEntry(Entry& entry)
{
std::shared_ptr transport;
+ BackendId id = 0;
+ uint32_t epoch = 0;
uint32_t services = 0;
bool required = false;
{
@@ -365,6 +394,8 @@ bool CTransportManager::InitializeEntry(Entry& entry)
if (entry.state != State::OPEN)
return false;
transport = entry.transport;
+ id = entry.id;
+ epoch = entry.epoch;
services = entry.config.services;
required = entry.required;
}
@@ -388,6 +419,13 @@ bool CTransportManager::InitializeEntry(Entry& entry)
}
}
const bool hasFrameCaps = frameCaps != nullptr;
+ bool syncNow = false;
+ if (!m_recovery.Attach(id, epoch, syncNow))
+ {
+ CSRWExclusiveLock entryLock(entry.lock);
+ entry.state = State::FAILED;
+ return false;
+ }
{
CSRWExclusiveLock entryLock(entry.lock);
// The first successfully initialized instance fixes the advertised
@@ -399,7 +437,17 @@ bool CTransportManager::InitializeEntry(Entry& entry)
// provide its other configured services, but must not receive frames.
entry.frameAbsent = (services & TRANSPORT_SERVICE_FRAME) &&
!hasFrameCaps;
- entry.state = State::INITIALIZED;
+ entry.recoveryAttached = true;
+ entry.syncPending = syncNow;
+ entry.state = State::INITIALIZED;
+ }
+ // Monitor readiness may be published after Attach but before this entry is
+ // visible to SyncRecovery. Claim any synchronization missed in that gap.
+ if (m_recovery.ClaimSync(id, epoch))
+ {
+ CSRWExclusiveLock entryLock(entry.lock);
+ if (entry.id == id && entry.epoch == epoch && entry.recoveryAttached)
+ entry.syncPending = true;
}
return true;
}
@@ -672,6 +720,7 @@ void CTransportManager::RetryEntry(Entry& entry, uint64_t now,
transport = entry.transport;
}
+ DetachRecovery(entry);
RemoveServices(entry);
if (transport)
transport->Stop();
@@ -688,8 +737,6 @@ void CTransportManager::RetryEntry(Entry& entry, uint64_t now,
entry.inputAbsent = false;
entry.frameAbsent = false;
entry.serviceRetryAt = 0;
- entry.recoveryPending = false;
- entry.recovery = RecoveryUpdate {};
++entry.epoch;
if (!entry.epoch)
++entry.epoch;
@@ -729,6 +776,7 @@ void CTransportManager::HandleProcessResult(
name = entry.config.kind;
}
+ DetachRecovery(entry);
RemoveServices(entry);
if (primary && exposed)
{
@@ -947,7 +995,7 @@ bool CTransportManager::Setup(size_t alignment)
}
ITransport::ProcessResult CTransportManager::Process(
- ITransportEvents& events)
+ ITransportActions& actions)
{
if (!BeginPhase(Phase::PROCESS, false))
{
@@ -967,6 +1015,7 @@ ITransport::ProcessResult CTransportManager::Process(
}
const uint64_t now = GetTickCount64();
+ m_recovery.Tick(now);
HandleServiceFailures();
Entry * entries[FRAME_MAX_SINKS] = {};
const unsigned count = Entries(entries);
@@ -1025,7 +1074,7 @@ ITransport::ProcessResult CTransportManager::Process(
}
DrainRecovery(entry, transport);
CSourceEvents sourceEvents(
- id, epoch, interactions, m_input, events);
+ id, epoch, interactions, m_input, m_recovery, actions);
const ProcessResult result = transport->Process(sourceEvents);
DrainRecovery(entry, transport);
HandleProcessResult(entry, result);
@@ -1070,9 +1119,8 @@ void CTransportManager::Stop()
for (unsigned i = 0; i < count; ++i)
{
CSRWExclusiveLock entryLock(entries[i]->lock);
- entries[i]->stopRequested = true;
- entries[i]->syncPending = false;
- entries[i]->recoveryPending = false;
+ entries[i]->stopRequested = true;
+ entries[i]->syncPending = false;
}
for (unsigned i = 0; i < count; ++i)
@@ -1112,7 +1160,10 @@ void CTransportManager::Stop()
}
for (unsigned i = count; i > 0; --i)
+ {
+ DetachRecovery(*entries[i - 1]);
RemoveServices(*entries[i - 1]);
+ }
m_input.Stop();
@@ -1155,6 +1206,8 @@ void CTransportManager::SyncRecovery()
return;
}
+ m_recovery.MarkMonitorReady();
+
Entry * entries[FRAME_MAX_SINKS] = {};
const unsigned count = Entries(entries);
for (unsigned i = 0; i < count; ++i)
@@ -1165,9 +1218,13 @@ void CTransportManager::SyncRecovery()
{
CSRWExclusiveLock entryLock(entry.lock);
if (entry.stopRequested || !entry.transport ||
+ !entry.recoveryAttached ||
(entry.state != State::INITIALIZED && entry.state != State::READY))
continue;
+ if (!m_recovery.ClaimSync(entry.id, entry.epoch))
+ continue;
+
if (entry.call != Call::IDLE)
{
entry.syncPending = true;
@@ -1187,53 +1244,16 @@ void CTransportManager::SyncRecovery()
}
}
-void CTransportManager::RecoveryStatus(const SourceKey& source,
- uint64_t session, uint32_t serial, bool active,
+void CTransportManager::RecoveryStatus(uint64_t route, uint64_t session,
+ uint32_t serial, bool active,
Recovery state, uint32_t error)
{
- {
- CSRWSharedLock managerLock(m_lock);
- if (m_stopping || m_stopped)
- return;
- }
-
- Entry * entries[FRAME_MAX_SINKS] = {};
- const unsigned count = Entries(entries);
- for (unsigned i = 0; i < count; ++i)
- {
- Entry& entry = *entries[i];
- std::shared_ptr transport;
- bool call = false;
- {
- CSRWExclusiveLock entryLock(entry.lock);
- if (entry.id != source.backend || entry.epoch != source.epoch ||
- entry.stopRequested || !entry.transport ||
- (entry.state != State::INITIALIZED && entry.state != State::READY))
- continue;
-
- if (entry.call != Call::IDLE)
- {
- entry.recovery.session = session;
- entry.recovery.serial = serial;
- entry.recovery.active = active;
- entry.recovery.state = state;
- entry.recovery.error = error;
- entry.recoveryPending = true;
- continue;
- }
-
- entry.call = Call::RECOVERY;
- entry.callOwner = GetCurrentThreadId();
- ResetEvent(entry.idleEvent);
- transport = entry.transport;
- call = true;
- }
-
- if (call)
- transport->RecoveryStatus(session, serial, active, state, error);
- EndCall(entry, transport);
- return;
- }
+ RecoveryAction action;
+ action.route = route;
+ action.session = session;
+ action.serial = serial;
+ action.active = active;
+ m_recovery.Complete(action, state, error);
}
bool CTransportManager::CanUseMode(const FrameMode& mode,
diff --git a/idd/LGIdd/transport/CTransportManager.h b/idd/LGIdd/transport/CTransportManager.h
index 7f8c502b..6596364c 100644
--- a/idd/LGIdd/transport/CTransportManager.h
+++ b/idd/LGIdd/transport/CTransportManager.h
@@ -24,6 +24,7 @@
#include "transport/CControlHub.h"
#include "transport/CFrameHub.h"
#include "transport/CInputHub.h"
+#include "transport/CRecoveryHub.h"
#include "transport/ITransport.h"
#include "transport/TransportConfig.h"
@@ -75,15 +76,6 @@ private:
ACCESS,
};
- struct RecoveryUpdate
- {
- uint64_t session = 0;
- uint32_t serial = 0;
- bool active = false;
- Recovery state = Recovery::FAILED;
- uint32_t error = 0;
- };
-
struct Entry
{
Entry();
@@ -115,8 +107,7 @@ private:
bool exposed = false;
bool setupDone = false;
bool syncPending = false;
- bool recoveryPending = false;
- RecoveryUpdate recovery;
+ bool recoveryAttached = false;
std::shared_ptr frameCaps;
DirectFrameBufferMemory directMemory;
bool directMemoryValid = false;
@@ -128,6 +119,7 @@ private:
CControlHub m_control;
CFrameHub m_frames;
CInputHub m_input;
+ CRecoveryHub m_recovery;
Entry * m_primary = nullptr;
bool m_initialized = false;
bool m_setup = false;
@@ -150,6 +142,7 @@ private:
const std::shared_ptr& transport, bool drain = true);
void DrainRecovery(Entry& entry,
const std::shared_ptr& transport);
+ void DetachRecovery(Entry& entry);
OpenResult OpenEntry(Entry& entry);
bool InitializeEntry(Entry& entry);
@@ -175,11 +168,11 @@ public:
OpenResult Open();
bool Initialize();
bool Setup(size_t alignment);
- ProcessResult Process(ITransportEvents& events);
+ ProcessResult Process(ITransportActions& actions);
void Stop();
void SyncRecovery();
- void RecoveryStatus(const SourceKey& source,
- uint64_t session, uint32_t serial, bool active,
+ void RecoveryStatus(uint64_t route, uint64_t session,
+ uint32_t serial, bool active,
Recovery state, uint32_t error);
bool CanUseMode(const FrameMode& mode,
diff --git a/idd/LGIdd/transport/ITransport.h b/idd/LGIdd/transport/ITransport.h
index 930da733..dfccf0db 100644
--- a/idd/LGIdd/transport/ITransport.h
+++ b/idd/LGIdd/transport/ITransport.h
@@ -50,6 +50,29 @@ enum class InteractionResult
FAILED,
};
+enum class RecoveryState
+{
+ NORMAL,
+ ACTIVE,
+ FAILED,
+};
+
+struct RecoveryAdmission
+{
+ bool complete = false;
+ RecoveryState state = RecoveryState::FAILED;
+ uint32_t error = 0;
+};
+
+struct RecoveryAction
+{
+ uint64_t route = 0;
+ uint64_t session = 0;
+ uint64_t deadline = 0;
+ uint32_t serial = 0;
+ bool active = false;
+};
+
class ITransportEvents
{
public:
@@ -59,10 +82,25 @@ public:
const SourceKey& source, int32_t x, int32_t y) = 0;
virtual InteractionResult OnSetResolution(
const SourceKey& source, uint32_t width, uint32_t height) = 0;
- virtual void OnRecoveryRequest(const SourceKey& source,
+ virtual RecoveryAdmission OnRecoveryRequest(const SourceKey& source,
uint64_t session, uint32_t serial, bool active) = 0;
};
+// Actions are implemented above the transport manager. Transport instances
+// receive ITransportEvents only, so recovery work cannot bypass the manager's
+// coordinator.
+class ITransportActions
+{
+public:
+ virtual ~ITransportActions() = default;
+
+ virtual InteractionResult OnSetCursorPos(
+ const SourceKey& source, int32_t x, int32_t y) = 0;
+ virtual InteractionResult OnSetResolution(
+ const SourceKey& source, uint32_t width, uint32_t height) = 0;
+ virtual bool OnRecoveryAction(const RecoveryAction& action) = 0;
+};
+
class ITransport
{
public:
@@ -80,12 +118,7 @@ public:
FAILURE,
};
- enum class Recovery
- {
- NORMAL,
- ACTIVE,
- FAILED,
- };
+ using Recovery = RecoveryState;
virtual ~ITransport() = default;
@@ -97,8 +130,8 @@ public:
virtual ProcessResult Process(ITransportEvents& events) = 0;
virtual void Stop() = 0;
virtual void SyncRecovery() {}
- virtual void RecoveryStatus(
- uint64_t, uint32_t, bool, Recovery, uint32_t) {}
+ virtual void RecoveryStatus(const SourceKey&, uint64_t, uint32_t,
+ bool, Recovery, uint32_t) {}
// Frame capabilities describe the configured instance, not transient
// runtime state. CanUseMode answers are immutable for the returned
diff --git a/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp b/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp
index 161e75d4..0df1f5d5 100644
--- a/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp
+++ b/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp
@@ -98,8 +98,13 @@ ITransport::ProcessResult CLGMPTransport::Process(ITransportEvents& events)
{
const CRecovery::Request recovery = m_recovery.Process();
if (recovery.valid)
- events.OnRecoveryRequest(SourceKey(),
- recovery.session, recovery.serial, recovery.active);
+ {
+ const RecoveryAdmission admission = events.OnRecoveryRequest(
+ SourceKey(), recovery.session, recovery.serial, recovery.active);
+ if (admission.complete)
+ RecoveryStatus(SourceKey(), recovery.session, recovery.serial,
+ recovery.active, admission.state, admission.error);
+ }
// Before the swap chain establishes the frame-buffer alignment, service
// only the protocol-independent recovery channel. This preserves the old
@@ -228,10 +233,11 @@ void CLGMPTransport::SyncRecovery()
m_recovery.Sync();
}
-void CLGMPTransport::RecoveryStatus(
+void CLGMPTransport::RecoveryStatus(const SourceKey& source,
uint64_t session, uint32_t serial, bool active,
Recovery state, uint32_t error)
{
+ UNREFERENCED_PARAMETER(source);
uint32_t wireState = KVMFR_R_STATE_FAILED;
uint32_t wireError = KVMFR_R_ERR_NONE;
switch (state)
@@ -246,9 +252,29 @@ void CLGMPTransport::RecoveryStatus(
case Recovery::FAILED:
wireState = KVMFR_R_STATE_FAILED;
- wireError = error == ERROR_NOT_FOUND ?
- KVMFR_R_ERR_NO_FALLBACK_DISPLAY :
- KVMFR_R_ERR_TOPOLOGY_FAILED;
+ switch (error)
+ {
+ case ERROR_NOT_FOUND:
+ wireError = KVMFR_R_ERR_NO_FALLBACK_DISPLAY;
+ break;
+
+ case ERROR_BUSY:
+ wireError = KVMFR_R_ERR_BUSY;
+ break;
+
+ case ERROR_NOT_ENOUGH_QUOTA:
+ wireError = KVMFR_R_ERR_CAPACITY;
+ break;
+
+ case ERROR_TIMEOUT:
+ case RPC_S_SERVER_UNAVAILABLE:
+ wireError = KVMFR_R_ERR_HELPER_UNAVAILABLE;
+ break;
+
+ default:
+ wireError = KVMFR_R_ERR_TOPOLOGY_FAILED;
+ break;
+ }
break;
}
diff --git a/idd/LGIdd/transport/lgmp/CLGMPTransport.h b/idd/LGIdd/transport/lgmp/CLGMPTransport.h
index 668e7475..265727ae 100644
--- a/idd/LGIdd/transport/lgmp/CLGMPTransport.h
+++ b/idd/LGIdd/transport/lgmp/CLGMPTransport.h
@@ -57,7 +57,7 @@ public:
ProcessResult Process(ITransportEvents& events) override;
void Stop() override;
void SyncRecovery() override;
- void RecoveryStatus(
+ void RecoveryStatus(const SourceKey& source,
uint64_t session, uint32_t serial, bool active,
Recovery state, uint32_t error) override;