From 5b9d4923a786cc55abbe11fa140c278cc28a6e1e Mon Sep 17 00:00:00 2001 From: Geoffrey McRae Date: Thu, 13 Aug 2026 01:01:00 +1000 Subject: [PATCH] [idd] transport: coordinate recovery requests --- common/include/common/KVMFRRecovery.h | 4 +- idd/LGIdd/LGIdd.vcxproj | 2 + idd/LGIdd/LGIdd.vcxproj.filters | 6 + idd/LGIdd/display/CDeviceContext.cpp | 56 +-- idd/LGIdd/display/CDeviceContext.h | 13 +- idd/LGIdd/transport/CRecoveryHub.cpp | 416 ++++++++++++++++++++ idd/LGIdd/transport/CRecoveryHub.h | 140 +++++++ idd/LGIdd/transport/CTransportManager.cpp | 174 ++++---- idd/LGIdd/transport/CTransportManager.h | 21 +- idd/LGIdd/transport/ITransport.h | 51 ++- idd/LGIdd/transport/lgmp/CLGMPTransport.cpp | 38 +- idd/LGIdd/transport/lgmp/CLGMPTransport.h | 2 +- 12 files changed, 752 insertions(+), 171 deletions(-) create mode 100644 idd/LGIdd/transport/CRecoveryHub.cpp create mode 100644 idd/LGIdd/transport/CRecoveryHub.h 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;