diff --git a/idd/LGIdd/LGIdd.vcxproj b/idd/LGIdd/LGIdd.vcxproj index b83e345b..59174665 100644 --- a/idd/LGIdd/LGIdd.vcxproj +++ b/idd/LGIdd/LGIdd.vcxproj @@ -62,6 +62,7 @@ + @@ -111,13 +112,16 @@ + + + diff --git a/idd/LGIdd/LGIdd.vcxproj.filters b/idd/LGIdd/LGIdd.vcxproj.filters index 01466c65..723e3880 100644 --- a/idd/LGIdd/LGIdd.vcxproj.filters +++ b/idd/LGIdd/LGIdd.vcxproj.filters @@ -160,6 +160,9 @@ Transport + + Transport + Transport @@ -169,6 +172,9 @@ Transport + + Transport + Transport @@ -181,6 +187,9 @@ Transport + + Transport + Transport @@ -309,6 +318,9 @@ Transport + + Transport + Transport diff --git a/idd/LGIdd/display/CDeviceContext.cpp b/idd/LGIdd/display/CDeviceContext.cpp index c329a64e..a2d37271 100644 --- a/idd/LGIdd/display/CDeviceContext.cpp +++ b/idd/LGIdd/display/CDeviceContext.cpp @@ -85,15 +85,21 @@ void CDeviceContext::QueryIddCxCapabilities() #ifdef HAS_IDDCX_110 const bool hasIddCx110DDIs = - !!IDD_IS_FUNCTION_AVAILABLE(IddCxSwapChainReleaseAndAcquireBuffer2) && - !!IDD_IS_FUNCTION_AVAILABLE(IddCxMonitorQueryHardwareCursor3) && - !!IDD_IS_FUNCTION_AVAILABLE(IddCxMonitorUpdateModes2) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxAdapterQueryTargetInfo) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxAdapterCommitModes2) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxParseMonitorDescription2) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorQueryTargetModes2) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorSetDefaultHdrMetaData) && - IDD_IS_FIELD_AVAILABLE(IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorSetGammaRamp); + IDD_IS_FUNCTION_AVAILABLE(IddCxSwapChainReleaseAndAcquireBuffer2) && + IDD_IS_FUNCTION_AVAILABLE(IddCxMonitorQueryHardwareCursor3) && + IDD_IS_FUNCTION_AVAILABLE(IddCxMonitorUpdateModes2) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxAdapterQueryTargetInfo) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxAdapterCommitModes2) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxParseMonitorDescription2) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorQueryTargetModes2) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorSetDefaultHdrMetaData) && + IDD_IS_FIELD_AVAILABLE( + IDD_CX_CLIENT_CONFIG, EvtIddCxMonitorSetGammaRamp); #else const bool hasIddCx110DDIs = false; #endif @@ -474,12 +480,31 @@ bool CDeviceContext::InitializeTransport() return true; g_pipe.SetRecoveryHandler( - [](void * opaque, uint64_t session, uint32_t serial, bool active, - LGPipeMsg::Type result) + [](void * opaque, uint64_t route, uint64_t session, + uint32_t serial, bool active, LGPipeMsg::Type result) { 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) @@ -505,7 +530,7 @@ bool CDeviceContext::InitializeTransport() } context->m_transport->RecoveryStatus( - session, serial, active, state, error); + source, session, serial, active, state, error); }, this); m_recoveryHandlerSet = true; @@ -563,8 +588,7 @@ bool CDeviceContext::SetupTransport(size_t alignSize) return false; } - IInputTransport * input = m_transport->Input(); - if (input && !input->Start(g_inputPipeServer)) + if (!m_transport->Input().Start(g_inputPipeServer)) { DEBUG_ERROR("Failed to start input transport"); return false; @@ -610,6 +634,34 @@ void CDeviceContext::OnSetResolution(const SourceKey& source, void CDeviceContext::OnRecoveryRequest(const SourceKey& source, uint64_t session, uint32_t serial, bool active) { - UNREFERENCED_PARAMETER(source); - g_pipe.SetRecovery(this, session, serial, active); + 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; + } } diff --git a/idd/LGIdd/display/CDeviceContext.h b/idd/LGIdd/display/CDeviceContext.h index d2c0a416..569b6008 100644 --- a/idd/LGIdd/display/CDeviceContext.h +++ b/idd/LGIdd/display/CDeviceContext.h @@ -53,6 +53,14 @@ 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; diff --git a/idd/LGIdd/ipc/CPipeServer.cpp b/idd/LGIdd/ipc/CPipeServer.cpp index d01b0ff2..0fabc961 100644 --- a/idd/LGIdd/ipc/CPipeServer.cpp +++ b/idd/LGIdd/ipc/CPipeServer.cpp @@ -136,7 +136,7 @@ void CPipeServer::HandleRecovery(const LGPipeMsg & msg) queueLock.Unlock(); if (m_recoveryHandler) m_recoveryHandler(m_recoveryOpaque, - msg.recovery.session, serial, active, msg.type); + m_recoveryRoute, msg.recovery.session, serial, active, msg.type); } void CPipeServer::SetDeviceContext(CDeviceContext * context) @@ -150,6 +150,7 @@ void CPipeServer::SetRecoveryHandler( { CSRWExclusiveLock queueLock(m_queueLock); CSRWExclusiveLock recoveryLock(m_recoveryLock); + m_recoveryRoute = 0; m_recoveryValid = false; m_recoveryRequest = {}; m_recoveryHandler = handler; @@ -163,6 +164,7 @@ void CPipeServer::ClearRecoveryHandler(void * opaque) if (m_recoveryOpaque != opaque) return; + m_recoveryRoute = 0; m_recoveryValid = false; m_recoveryRequest = {}; m_recoveryHandler = nullptr; @@ -218,14 +220,14 @@ void CPipeServer::ResolutionRejected(uint32_t width, uint32_t height, WriteMsg(msg); } -void CPipeServer::SetRecovery( - void * owner, uint64_t session, uint32_t serial, bool active) +bool CPipeServer::SetRecovery(void * owner, uint64_t route, + uint64_t session, uint32_t serial, bool active) { - if (!session || !serial || + if (!route || !session || !serial || (serial & LGPipeMsg::RECOVERY_ACTIVE)) { DEBUG_ERROR("Invalid recovery request correlation"); - return; + return false; } LGPipeMsg msg = {}; @@ -236,11 +238,13 @@ void CPipeServer::SetRecovery( (active ? LGPipeMsg::RECOVERY_ACTIVE : 0U); CSRWExclusiveLock queueLock(m_queueLock); - CSRWSharedLock recoveryLock(m_recoveryLock); + CSRWExclusiveLock recoveryLock(m_recoveryLock); if (!m_recoveryHandler || m_recoveryOpaque != owner) - return; + return false; m_recoveryValid = true; + m_recoveryRoute = route; m_recoveryRequest = msg; m_endpoint.Send(&msg, sizeof(msg)); + return true; } diff --git a/idd/LGIdd/ipc/CPipeServer.h b/idd/LGIdd/ipc/CPipeServer.h index 16d7e7a7..07d5d8ab 100644 --- a/idd/LGIdd/ipc/CPipeServer.h +++ b/idd/LGIdd/ipc/CPipeServer.h @@ -35,7 +35,7 @@ class CPipeServer : private IPipeEndpointHandler { public: using RecoveryHandler = void (*)(void * opaque, - uint64_t session, uint32_t serial, bool active, + uint64_t route, uint64_t session, uint32_t serial, bool active, LGPipeMsg::Type result); private: @@ -51,6 +51,7 @@ class CPipeServer : private IPipeEndpointHandler CSRWLock m_recoveryLock; RecoveryHandler m_recoveryHandler = nullptr; void * m_recoveryOpaque = nullptr; + uint64_t m_recoveryRoute = 0; void WriteMsg(const LGPipeMsg & msg); void QueueMsgLocked(const LGPipeMsg & msg); @@ -77,8 +78,8 @@ class CPipeServer : private IPipeEndpointHandler void SetGPUStatus(bool software); void ResolutionRejected(uint32_t width, uint32_t height, uint32_t requiredSizeMiB); - void SetRecovery( - void * owner, uint64_t session, uint32_t serial, bool active); + bool SetRecovery(void * owner, uint64_t route, + uint64_t session, uint32_t serial, bool active); }; extern CPipeServer g_pipe; diff --git a/idd/LGIdd/transport/CControlHub.cpp b/idd/LGIdd/transport/CControlHub.cpp index 179a1751..09813c9e 100644 --- a/idd/LGIdd/transport/CControlHub.cpp +++ b/idd/LGIdd/transport/CControlHub.cpp @@ -20,120 +20,529 @@ #include "transport/CControlHub.h" -#include -#include -#include +#include "CDebug.h" -std::vector> CControlHub::Snapshot() const +#include +#include + +CControlHub::CControlHub() { - CSRWSharedLock lock(m_sinkLock); - return m_sinks; + m_stopEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr); + if (!m_stopEvent) + return; + + m_valid = true; + for (unsigned i = 0; i < MAX_SINKS; ++i) + { + Sink& sink = m_sinks[i]; + sink.owner = this; + sink.index = i; + sink.wake = CreateEvent(nullptr, FALSE, FALSE, nullptr); + sink.idle = CreateEvent(nullptr, TRUE, TRUE, nullptr); + if (!sink.wake || !sink.idle) + { + m_valid = false; + break; + } + + sink.thread = CreateThread(nullptr, 0, WorkerProc, &sink, 0, nullptr); + if (!sink.thread) + { + m_valid = false; + break; + } + } + + if (!m_valid) + { + SetEvent(m_stopEvent); + for (Sink& sink : m_sinks) + if (sink.wake) + SetEvent(sink.wake); + } +} + +CControlHub::~CControlHub() +{ + for (Sink& sink : m_sinks) + { + BackendId backend; + uint32_t epoch; + { + CSRWSharedLock lock(sink.lock); + backend = sink.backend; + epoch = sink.epoch; + } + if (backend && epoch) + Remove(backend, epoch); + } + + if (m_stopEvent) + SetEvent(m_stopEvent); + for (Sink& sink : m_sinks) + if (sink.wake) + SetEvent(sink.wake); + + for (Sink& sink : m_sinks) + { + if (sink.thread) + WaitForSingleObject(sink.thread, INFINITE); + if (sink.thread) + CloseHandle(sink.thread); + if (sink.idle) + CloseHandle(sink.idle); + if (sink.wake) + CloseHandle(sink.wake); + } + if (m_stopEvent) + CloseHandle(m_stopEvent); +} + +DWORD WINAPI CControlHub::WorkerProc(void * opaque) +{ + Sink * sink = static_cast(opaque); + sink->owner->Worker(*sink); + return 0; +} + +uint64_t CControlHub::NextRevision(uint64_t revision) +{ + ++revision; + return revision ? revision : 1; +} + +bool CControlHub::TokenMatches( + const Sink& sink, const ControlToken& token) +{ + return sink.backend == token.backend && sink.epoch == token.epoch; } bool CControlHub::Add( - BackendId backend, uint32_t epoch, IControlTransport& control) + BackendId backend, uint32_t epoch, IControlSink& control) { - std::shared_ptr sink(new (std::nothrow) Sink); - if (!sink) + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + if (!backend || !epoch || !m_valid) return false; - sink->backend = backend; - sink->epoch = epoch; - sink->control = &control; + Sink * selected = nullptr; { - CSRWExclusiveLock lock(m_sinkLock); - for (const auto& current : m_sinks) - if (current->backend == backend && current->epoch == epoch) + CSRWExclusiveLock lock(m_listLock); + for (Sink& sink : m_sinks) + { + CSRWSharedLock sinkLock(sink.lock); + if ((sink.active || sink.reserved) && + sink.backend == backend && sink.epoch == epoch) return false; - m_sinks.push_back(sink); - } + } - Replay(sink); + for (Sink& sink : m_sinks) + { + CSRWExclusiveLock sinkLock(sink.lock); + if (!sink.active && !sink.failed && !sink.reserved) + { + sink.reserved = true; + sink.backend = backend; + sink.epoch = epoch; + selected = &sink; + break; + } + } + } + if (!selected) + return false; + + const ControlToken token = { backend, epoch }; + { + CSRWExclusiveLock lock(selected->lock); + selected->target = &control; + selected->deliveredPosition = 0; + selected->deliveredShape = 0; + selected->deliveredTransform = 0; + memset(selected->retryAt, 0, sizeof(selected->retryAt)); + selected->nextWork = 0; + selected->bindingSerial = NextRevision(selected->bindingSerial); + selected->replaySerial = NextRevision(selected->replaySerial); + selected->calling = false; + selected->active = false; + selected->failed = false; + selected->failurePending = false; + } + control.SetControlEvents(this, token); + bool attached = false; + { + CSRWExclusiveLock lock(selected->lock); + if (selected->target == &control && selected->backend == backend && + selected->epoch == epoch && selected->reserved && !selected->failed) + { + selected->replaySerial = NextRevision(selected->replaySerial); + selected->active = true; + selected->reserved = false; + attached = true; + } + } + if (!attached) + { + control.SetControlEvents(nullptr, {}); + SetEvent(selected->wake); + WaitForSingleObject(selected->idle, INFINITE); + CSRWExclusiveLock listLock(m_listLock); + CSRWExclusiveLock sinkLock(selected->lock); + if (selected->target == &control && selected->backend == backend && + selected->epoch == epoch) + { + selected->target = nullptr; + selected->backend = 0; + selected->epoch = 0; + selected->active = false; + selected->reserved = false; + selected->failed = false; + selected->failurePending = false; + selected->calling = false; + memset(selected->retryAt, 0, sizeof(selected->retryAt)); + } + return false; + } + SetEvent(selected->wake); return true; } void CControlHub::Remove(BackendId backend, uint32_t epoch) { - std::shared_ptr removed; + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + Sink * selected = nullptr; + IControlSink * target = nullptr; { - CSRWExclusiveLock lock(m_sinkLock); - for (auto current = m_sinks.begin(); current != m_sinks.end(); ++current) - if ((*current)->backend == backend && (*current)->epoch == epoch) + CSRWExclusiveLock listLock(m_listLock); + for (Sink& sink : m_sinks) + { + CSRWExclusiveLock lock(sink.lock); + if ((sink.active || sink.failed) && + sink.backend == backend && sink.epoch == epoch) { - removed = *current; - m_sinks.erase(current); + sink.active = false; + sink.failed = false; + sink.failurePending = false; + sink.reserved = true; + target = sink.target; + selected = &sink; break; } + } } + if (!selected) + return; + + SetEvent(selected->wake); + WaitForSingleObject(selected->idle, INFINITE); + if (target) + target->SetControlEvents(nullptr, {}); - if (removed) { - CSRWExclusiveLock lock(removed->lock); - removed->control = nullptr; + CSRWExclusiveLock listLock(m_listLock); + CSRWExclusiveLock sinkLock(selected->lock); + if (selected->reserved && selected->backend == backend && + selected->epoch == epoch) + { + selected->target = nullptr; + selected->backend = 0; + selected->epoch = 0; + selected->reserved = false; + selected->failed = false; + selected->failurePending = false; + selected->calling = false; + memset(selected->retryAt, 0, sizeof(selected->retryAt)); + } } } -void CControlHub::Replay(const std::shared_ptr& sink) +bool CControlHub::TakeFailure(ControlToken& token) { - IDARG_OUT_QUERY_HWCURSOR cursor = {}; - std::vector cursorData; - UINT sdrWhiteLevel; - bool cursorValid; - std::shared_ptr transform; + CSRWSharedLock lifecycleLock(m_lifecycleLock); + CSRWSharedLock listLock(m_listLock); + for (Sink& sink : m_sinks) + { + CSRWExclusiveLock lock(sink.lock); + if (!sink.failurePending) + continue; + token.backend = sink.backend; + token.epoch = sink.epoch; + sink.failurePending = false; + return true; + } + return false; +} + +void CControlHub::OnControlReplay(const ControlToken& token) +{ + CSRWSharedLock listLock(m_listLock); + for (Sink& sink : m_sinks) + { + CSRWExclusiveLock lock(sink.lock); + if ((!sink.active && !sink.reserved) || + !TokenMatches(sink, token)) + continue; + + sink.deliveredPosition = 0; + sink.deliveredShape = 0; + sink.deliveredTransform = 0; + sink.replaySerial = NextRevision(sink.replaySerial); + memset(sink.retryAt, 0, sizeof(sink.retryAt)); + SetEvent(sink.wake); + return; + } +} + +bool CControlHub::BeginWork(Sink& sink, Work& work, DWORD& wait) +{ + const uint64_t now = GetTickCount64(); + State state; { CSRWSharedLock lock(m_stateLock); - cursor = m_cursor; - cursorData = m_cursorData; - sdrWhiteLevel = m_sdrWhiteLevel; - cursorValid = m_cursorValid; - transform = m_colorTransform; + state.transform = m_state.transform; + state.cursor = m_state.cursor; + state.cursorData = m_state.cursorData; + state.sdrWhiteLevel = m_state.sdrWhiteLevel; + state.positionRevision = m_state.positionRevision; + state.shapeRevision = m_state.shapeRevision; + state.transformRevision = m_state.transformRevision; } - CSRWSharedLock lock(sink->lock); - if (!sink->control) - return; - sink->control->SetColorTransform(std::move(transform)); - if (cursorValid) - sink->control->SendCursor( - cursor, cursorData.empty() ? nullptr : cursorData.data(), - sdrWhiteLevel); + CSRWExclusiveLock lock(sink.lock); + if (!sink.active || !sink.target) + { + wait = INFINITE; + return false; + } + + const bool pending[] = + { + state.positionRevision != sink.deliveredPosition, + state.shapeRevision != sink.deliveredShape, + state.transformRevision != sink.deliveredTransform, + }; + + unsigned selected = static_cast(WorkType::COUNT); + uint64_t earliest = 0; + for (unsigned offset = 0; + offset < static_cast(WorkType::COUNT); ++offset) + { + const unsigned index = + (sink.nextWork + offset) % static_cast(WorkType::COUNT); + if (!pending[index]) + continue; + if (!sink.retryAt[index] || sink.retryAt[index] <= now) + { + selected = index; + break; + } + if (!earliest || sink.retryAt[index] < earliest) + earliest = sink.retryAt[index]; + } + + if (selected == static_cast(WorkType::COUNT)) + { + wait = earliest ? static_cast(std::min( + earliest - now, MAXDWORD - 1)) : INFINITE; + return false; + } + + sink.nextWork = + (selected + 1) % static_cast(WorkType::COUNT); + sink.calling = true; + ResetEvent(sink.idle); + + work.type = static_cast(selected); + work.target = sink.target; + work.sdrWhiteLevel = state.sdrWhiteLevel; + work.positionRevision = state.positionRevision; + work.shapeRevision = state.shapeRevision; + work.transformRevision = state.transformRevision; + work.bindingSerial = sink.bindingSerial; + work.replaySerial = sink.replaySerial; + + if (work.type == WorkType::TRANSFORM) + work.transform = std::move(state.transform); + else + { + work.cursor.IsCursorVisible = state.cursor.IsCursorVisible; + work.cursor.X = state.cursor.X; + work.cursor.Y = state.cursor.Y; + if (work.type == WorkType::SHAPE) + { + work.cursor.IsCursorShapeUpdated = true; + work.cursor.CursorShapeInfo = state.cursor.CursorShapeInfo; + work.cursorData = std::move(state.cursorData); + } + else + work.cursor.CursorShapeInfo.CursorType = + IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED; + } + wait = INFINITE; + return true; +} + +bool CControlHub::CompleteWork( + Sink& sink, const Work& work, ControlResult result) +{ + CSRWExclusiveLock lock(sink.lock); + if (sink.target != work.target || + sink.bindingSerial != work.bindingSerial) + return false; + if (!sink.active) + { + sink.calling = false; + SetEvent(sink.idle); + return false; + } + + if (result == ControlResult::FAILED) + { + sink.active = false; + sink.failed = true; + sink.failurePending = true; + sink.calling = false; + SetEvent(sink.idle); + return true; + } + + if (sink.replaySerial != work.replaySerial) + { + sink.calling = false; + SetEvent(sink.idle); + return true; + } + + const unsigned index = static_cast(work.type); + if (result == ControlResult::RETRY) + sink.retryAt[index] = GetTickCount64() + RETRY_MS; + else + { + sink.retryAt[index] = 0; + switch (work.type) + { + case WorkType::POSITION: + sink.deliveredPosition = work.positionRevision; + break; + + case WorkType::SHAPE: + sink.deliveredPosition = work.positionRevision; + sink.deliveredShape = work.shapeRevision; + break; + + case WorkType::TRANSFORM: + sink.deliveredTransform = work.transformRevision; + break; + + case WorkType::COUNT: + break; + } + } + + sink.calling = false; + SetEvent(sink.idle); + + return true; +} + +void CControlHub::Worker(Sink& sink) +{ + DWORD wait = INFINITE; + HANDLE events[] = { m_stopEvent, sink.wake }; + for (;;) + { + const DWORD status = WaitForMultipleObjects(2, events, FALSE, wait); + if (status == WAIT_OBJECT_0 || status == WAIT_FAILED) + return; + + Work work; + if (!BeginWork(sink, work, wait)) + continue; + + ControlResult result; + if (work.type == WorkType::TRANSFORM) + result = work.target->SetColorTransform(std::move(work.transform)); + else + result = work.target->SendCursor(work.cursor, + !work.cursorData || work.cursorData->empty() ? nullptr : + work.cursorData->data(), + work.cursorData ? work.cursorData->size() : 0, + work.sdrWhiteLevel); + + const bool completed = CompleteWork(sink, work, result); + if (result == ControlResult::FAILED && completed) + DEBUG_WARN("Control update delivery failed"); + if (completed) + { + wait = result == ControlResult::RETRY ? RETRY_MS : 0; + if (result != ControlResult::RETRY) + SetEvent(sink.wake); + } + else + wait = INFINITE; + } +} + +void CControlHub::WakeAll(WorkType type) +{ + const unsigned index = static_cast(type); + CSRWSharedLock listLock(m_listLock); + for (Sink& sink : m_sinks) + { + CSRWExclusiveLock lock(sink.lock); + if (!sink.active) + continue; + sink.retryAt[index] = 0; + SetEvent(sink.wake); + } } void CControlHub::SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, const BYTE * data, UINT sdrWhiteLevel) { + const bool shape = info.CursorShapeInfo.CursorType != + IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED; + std::shared_ptr> cursorData; + if (shape) + { + if (info.CursorShapeInfo.Height && + info.CursorShapeInfo.Pitch > + SIZE_MAX / info.CursorShapeInfo.Height) + { + DEBUG_ERROR("Pointer shape size overflow"); + return; + } + const size_t size = static_cast( + info.CursorShapeInfo.Height) * info.CursorShapeInfo.Pitch; + if (size && !data) + { + DEBUG_ERROR("Missing pointer shape payload"); + return; + } + cursorData = std::make_shared>(size); + if (size) + memcpy(cursorData->data(), data, size); + } + { CSRWExclusiveLock lock(m_stateLock); - m_cursor.IsCursorVisible = info.IsCursorVisible; - m_cursor.X = info.X; - m_cursor.Y = info.Y; - m_cursorValid = true; - m_sdrWhiteLevel = sdrWhiteLevel; + m_state.cursor.IsCursorVisible = info.IsCursorVisible; + m_state.cursor.X = info.X; + m_state.cursor.Y = info.Y; + m_state.sdrWhiteLevel = sdrWhiteLevel; + m_state.positionRevision = NextRevision(m_state.positionRevision); - if (info.CursorShapeInfo.CursorType != - IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED) + if (shape) { - const size_t size = - (size_t)info.CursorShapeInfo.Height * info.CursorShapeInfo.Pitch; - m_cursor.IsCursorShapeUpdated = info.IsCursorShapeUpdated; - m_cursor.CursorShapeInfo = info.CursorShapeInfo; - m_cursorData.resize(size); - if (size) - memcpy(m_cursorData.data(), data, size); - m_shapeValid = true; + m_state.cursor.IsCursorShapeUpdated = info.IsCursorShapeUpdated; + m_state.cursor.CursorShapeInfo = info.CursorShapeInfo; + m_state.cursorData = std::move(cursorData); + m_state.shapeRevision = NextRevision(m_state.shapeRevision); } - else if (!m_shapeValid) - m_cursor.CursorShapeInfo.CursorType = - IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED; } - const auto sinks = Snapshot(); - for (const auto& sink : sinks) - { - CSRWSharedLock lock(sink->lock); - if (sink->control) - sink->control->SendCursor(info, data, sdrWhiteLevel); - } + WakeAll(WorkType::POSITION); + if (shape) + WakeAll(WorkType::SHAPE); } void CControlHub::SetColorTransform( @@ -141,21 +550,15 @@ void CControlHub::SetColorTransform( { { CSRWExclusiveLock lock(m_stateLock); - m_colorTransform = transform; - } - - const auto sinks = Snapshot(); - for (const auto& sink : sinks) - { - CSRWSharedLock lock(sink->lock); - if (sink->control) - sink->control->SetColorTransform(transform); + m_state.transform = std::move(transform); + m_state.transformRevision = NextRevision(m_state.transformRevision); } + WakeAll(WorkType::TRANSFORM); } std::shared_ptr CControlHub::GetColorTransform() const { CSRWSharedLock lock(m_stateLock); - return m_colorTransform; + return m_state.transform; } diff --git a/idd/LGIdd/transport/CControlHub.h b/idd/LGIdd/transport/CControlHub.h index 9086e2aa..010ff195 100644 --- a/idd/LGIdd/transport/CControlHub.h +++ b/idd/LGIdd/transport/CControlHub.h @@ -21,40 +21,109 @@ #pragma once #include "CSRWLock.h" +#include "transport/IControlSink.h" #include "transport/IControlTransport.h" #include "transport/ITransport.h" -#include +#include + +#include #include -class CControlHub final : public IControlTransport +class CControlHub final : public IControlTransport, public IControlEvents { private: - struct Sink + static constexpr unsigned MAX_SINKS = 8; + static constexpr DWORD RETRY_MS = 5; + + enum class WorkType { - BackendId backend; - uint32_t epoch; - CSRWLock lock; - IControlTransport * control; + POSITION, + SHAPE, + TRANSFORM, + COUNT, }; - mutable CSRWLock m_sinkLock; - std::vector> m_sinks; + struct State + { + std::shared_ptr transform; + IDARG_OUT_QUERY_HWCURSOR cursor = {}; + std::shared_ptr> cursorData; + UINT sdrWhiteLevel = 0; + uint64_t positionRevision = 0; + uint64_t shapeRevision = 0; + uint64_t transformRevision = 1; + }; + struct Work + { + WorkType type = WorkType::POSITION; + IControlSink * target = nullptr; + IDARG_OUT_QUERY_HWCURSOR cursor = {}; + std::shared_ptr> cursorData; + std::shared_ptr transform; + UINT sdrWhiteLevel = 0; + uint64_t positionRevision = 0; + uint64_t shapeRevision = 0; + uint64_t transformRevision = 0; + uint64_t bindingSerial = 0; + uint64_t replaySerial = 0; + }; + + struct Sink + { + CControlHub * owner = nullptr; + unsigned index = 0; + BackendId backend = 0; + uint32_t epoch = 0; + CSRWLock lock; + IControlSink * target = nullptr; + HANDLE wake = nullptr; + HANDLE idle = nullptr; + HANDLE thread = nullptr; + uint64_t deliveredPosition = 0; + uint64_t deliveredShape = 0; + uint64_t deliveredTransform = 0; + uint64_t retryAt[static_cast(WorkType::COUNT)] = {}; + unsigned nextWork = 0; + uint64_t bindingSerial = 0; + uint64_t replaySerial = 0; + bool active = false; + bool failed = false; + bool failurePending = false; + bool reserved = false; + bool calling = false; + }; + + mutable CSRWLock m_listLock; mutable CSRWLock m_stateLock; - std::shared_ptr m_colorTransform; - IDARG_OUT_QUERY_HWCURSOR m_cursor = {}; - std::vector m_cursorData; - UINT m_sdrWhiteLevel = 0; - bool m_cursorValid = false; - bool m_shapeValid = false; + CSRWLock m_lifecycleLock; + State m_state; + Sink m_sinks[MAX_SINKS]; + HANDLE m_stopEvent = nullptr; + bool m_valid = false; - std::vector> Snapshot() const; - void Replay(const std::shared_ptr& sink); + static DWORD WINAPI WorkerProc(void * opaque); + static uint64_t NextRevision(uint64_t revision); + static bool TokenMatches(const Sink& sink, const ControlToken& token); + + bool BeginWork(Sink& sink, Work& work, DWORD& wait); + bool CompleteWork(Sink& sink, const Work& work, ControlResult result); + void Worker(Sink& sink); + void WakeAll(WorkType type); public: - bool Add(BackendId backend, uint32_t epoch, IControlTransport& control); + CControlHub(); + ~CControlHub() override; + + CControlHub(const CControlHub&) = delete; + CControlHub& operator=(const CControlHub&) = delete; + + bool Add(BackendId backend, uint32_t epoch, IControlSink& control); void Remove(BackendId backend, uint32_t epoch); + bool TakeFailure(ControlToken& token); + + void OnControlReplay(const ControlToken& token) override; void SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, const BYTE * data, UINT sdrWhiteLevel) override; diff --git a/idd/LGIdd/transport/CInputHub.cpp b/idd/LGIdd/transport/CInputHub.cpp new file mode 100644 index 00000000..51b9d26b --- /dev/null +++ b/idd/LGIdd/transport/CInputHub.cpp @@ -0,0 +1,586 @@ +/** + * 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/CInputHub.h" + +#include "input/IInputSink.h" + +static bool SameSource(const SourceKey& left, const SourceKey& right) +{ + return left.backend == right.backend && left.epoch == right.epoch && + left.client == right.client && left.generation == right.generation; +} + +CInputHub::CInputHub() +{ + for (Source& source : m_sources) + source.owner = this; +} + +CInputHub::~CInputHub() +{ + Stop(); + for (Source& source : m_sources) + { + BackendId backend; + uint32_t epoch; + { + CSRWSharedLock lock(m_lock); + backend = source.backend; + epoch = source.epoch; + } + if (backend && epoch) + Unbind(backend, epoch); + } +} + +SourceKey CInputHub::Source::Key(const InputSourceId& sourceId) const +{ + SourceKey source; + source.backend = backend; + source.epoch = epoch; + source.client = sourceId.client; + source.generation = sourceId.generation; + return source; +} + +InputTargetState CInputHub::Source::GetState( + const InputSourceId& source) +{ + return owner->GetState(Key(source)); +} + +void CInputHub::Source::Failed() +{ + owner->Failed(*this); +} + +InputResult CInputHub::Source::Claim(const InputSourceId& source) +{ + return owner->Claim(Key(source)); +} + +InputResult CInputHub::Source::Touch(const InputSourceId& source) +{ + return owner->Touch(Key(source)); +} + +InputResult CInputHub::Source::Release( + const InputSourceId& source, bool reset) +{ + return owner->Release(Key(source), reset); +} + +InputResult CInputHub::Source::SendMouseRelative( + const InputSourceId& source, + int32_t deltaX, int32_t deltaY, int32_t wheel, uint32_t buttons) +{ + return owner->SendMouseRelative( + Key(source), deltaX, deltaY, wheel, buttons); +} + +InputResult CInputHub::Source::SendMouseAbsolute( + const InputSourceId& source, + uint16_t x, uint16_t y, int32_t wheel, uint32_t buttons) +{ + return owner->SendMouseAbsolute(Key(source), x, y, wheel, buttons); +} + +InputResult CInputHub::Source::SendKeyboard(const InputSourceId& source, + uint8_t modifiers, const uint8_t * keys) +{ + return owner->SendKeyboard(Key(source), modifiers, keys); +} + +InputResult CInputHub::Source::Reset(const InputSourceId& source) +{ + return owner->Reset(Key(source)); +} + +bool CInputHub::Bind( + BackendId backend, uint32_t epoch, IInputSource& input) +{ + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + if (!backend || !epoch) + return false; + + Source * selected = nullptr; + bool start = false; + { + CSRWExclusiveLock lock(m_lock); + for (Source& source : m_sources) + if ((source.active || source.failed || source.reserved) && + source.backend == backend && source.epoch == epoch) + return false; + for (Source& source : m_sources) + if (!source.active && !source.failed && !source.reserved) + { + source.backend = backend; + source.epoch = epoch; + source.endpoint = &input; + source.active = true; + source.reserved = true; + selected = &source; + start = m_started; + break; + } + } + if (!selected) + return false; + + if (start && !input.Start(*selected)) + { + input.Stop(); + CSRWExclusiveLock lock(m_lock); + if (selected->reserved && selected->endpoint == &input && + selected->backend == backend && selected->epoch == epoch) + { + selected->backend = 0; + selected->epoch = 0; + selected->endpoint = nullptr; + selected->running = false; + selected->active = false; + selected->failed = false; + selected->failurePending = false; + selected->reserved = false; + } + return false; + } + { + CSRWExclusiveLock lock(m_lock); + if (!selected->reserved || selected->endpoint != &input || + selected->backend != backend || selected->epoch != epoch || + selected->failed) + { + lock.Unlock(); + if (start) + input.Stop(); + { + CSRWExclusiveLock clearLock(m_lock); + if (selected->endpoint == &input && + selected->backend == backend && selected->epoch == epoch) + { + selected->backend = 0; + selected->epoch = 0; + selected->endpoint = nullptr; + selected->running = false; + selected->active = false; + selected->failed = false; + selected->failurePending = false; + selected->reserved = false; + } + } + return false; + } + selected->running = start; + selected->reserved = false; + } + return true; +} + +void CInputHub::Unbind(BackendId backend, uint32_t epoch) +{ + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + Source * selected = nullptr; + IInputSource * input = nullptr; + bool stop = false; + { + CSRWExclusiveLock lock(m_lock); + for (Source& source : m_sources) + if ((source.active || source.failed || source.reserved) && + source.backend == backend && source.epoch == epoch) + { + source.active = false; + source.failed = false; + source.failurePending = false; + source.reserved = true; + input = source.endpoint; + stop = source.running; + source.running = false; + selected = &source; + if (m_owner.backend == backend && m_owner.epoch == epoch) + { + if (m_sink) + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + } + break; + } + } + if (!selected) + return; + + if (input && stop) + input->Stop(); + + CSRWExclusiveLock lock(m_lock); + if (selected->reserved && selected->backend == backend && + selected->epoch == epoch) + { + selected->backend = 0; + selected->epoch = 0; + selected->endpoint = nullptr; + selected->running = false; + selected->reserved = false; + selected->failed = false; + selected->failurePending = false; + } +} + +bool CInputHub::TakeFailure(SourceKey& source) +{ + CSRWSharedLock lifecycleLock(m_lifecycleLock); + CSRWExclusiveLock lock(m_lock); + for (Source& slot : m_sources) + { + if (!slot.failurePending) + continue; + source.backend = slot.backend; + source.epoch = slot.epoch; + slot.failurePending = false; + return true; + } + return false; +} + +bool CInputHub::Start(IInputSink& sink) +{ + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + Source * sources[MAX_SOURCES] = {}; + unsigned count = 0; + { + CSRWExclusiveLock lock(m_lock); + if (m_started) + return m_sink == &sink; + m_sink = &sink; + m_sinkState = sink.GetState(); + m_started = true; + for (Source& source : m_sources) + if (source.active) + { + source.reserved = true; + sources[count++] = &source; + } + } + + for (unsigned i = 0; i < count; ++i) + { + Source& source = *sources[i]; + const bool started = source.endpoint->Start(source); + bool stop = !started; + { + CSRWExclusiveLock lock(m_lock); + if (!started || source.failed || !source.active || !source.reserved) + { + if (m_owner.backend == source.backend && + m_owner.epoch == source.epoch) + { + if (m_sink) + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + } + source.active = false; + source.failed = true; + source.failurePending = true; + stop = true; + } + else + source.reserved = false; + source.running = started && !stop; + } + + if (!stop) + continue; + source.endpoint->Stop(); + { + CSRWExclusiveLock lock(m_lock); + source.running = false; + } + } + return true; +} + +void CInputHub::Stop() +{ + CSRWExclusiveLock lifecycleLock(m_lifecycleLock); + Source * sources[MAX_SOURCES] = {}; + unsigned count = 0; + IInputSink * sink = nullptr; + bool reset = false; + { + CSRWExclusiveLock lock(m_lock); + if (!m_started) + return; + m_started = false; + for (Source& source : m_sources) + if (source.endpoint && source.running) + { + source.running = false; + sources[count++] = &source; + } + sink = m_sink; + reset = m_owner.backend != 0; + m_owner = {}; + m_ownerDeadline = 0; + } + + for (unsigned i = count; i > 0; --i) + sources[i - 1]->endpoint->Stop(); + if (reset && sink) + sink->Reset(); + + CSRWExclusiveLock lock(m_lock); + m_sink = nullptr; + m_sinkState = 0; +} + +bool CInputHub::SourceValid(const SourceKey& source) const +{ + if (!source.client || !source.generation) + return false; + return BindingValid(source); +} + +bool CInputHub::BindingValid(const SourceKey& source) const +{ + if (!BindingPresent(source)) + return false; + for (const Source& slot : m_sources) + if (slot.active && !slot.reserved && slot.endpoint && + slot.backend == source.backend && + slot.epoch == source.epoch) + return true; + return false; +} + +bool CInputHub::BindingPresent(const SourceKey& source) const +{ + if (!source.backend || !source.epoch) + return false; + for (const Source& slot : m_sources) + if (slot.active && slot.endpoint && + slot.backend == source.backend && + slot.epoch == source.epoch) + return true; + return false; +} + +bool CInputHub::OwnerValid(const SourceKey& source) const +{ + return SourceValid(source) && SameSource(m_owner, source); +} + +bool CInputHub::CheckState() +{ + if (!m_started || !m_sink) + return false; + + const uint64_t state = m_sink->GetState(); + if (state != m_sinkState || !(state & 1)) + { + if (m_owner.backend) + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + m_sinkState = state; + } + else if (m_owner.backend && GetTickCount64() >= m_ownerDeadline) + { + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + } + return (state & 1) != 0; +} + +InputTargetState CInputHub::GetState(const SourceKey& source) +{ + CSRWExclusiveLock lock(m_lock); + InputTargetState result; + if (!BindingPresent(source)) + return result; + const bool available = CheckState(); + result.state = m_sinkState; + if (!available) + return result; + if (!BindingValid(source)) + return result; + if (!m_owner.backend) + result.available = true; + else if (source.client && source.generation && + SameSource(m_owner, source)) + { + result.available = true; + result.owned = true; + result.owner.client = m_owner.client; + result.owner.generation = m_owner.generation; + } + return result; +} + +void CInputHub::Failed(Source& source) +{ + CSRWExclusiveLock lock(m_lock); + if (!source.active) + return; + if (m_owner.backend == source.backend && m_owner.epoch == source.epoch) + { + if (m_sink) + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + } + source.active = false; + source.failed = true; + source.failurePending = true; +} + +InputResult CInputHub::Claim(const SourceKey& source) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (m_owner.backend) + return SameSource(m_owner, source) ? + InputResult::ACCEPTED : InputResult::BUSY; + if (!m_sink->Reset() || m_sink->GetState() != m_sinkState) + { + const uint64_t state = m_sink->GetState(); + m_sinkState = state; + return InputResult::UNAVAILABLE; + } + m_owner = source; + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} + +InputResult CInputHub::Touch(const SourceKey& source) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} + +InputResult CInputHub::Release(const SourceKey& source, bool reset) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + bool accepted = true; + if (reset && m_sink) + accepted = m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + return accepted ? InputResult::ACCEPTED : InputResult::UNAVAILABLE; +} + +InputResult CInputHub::SendMouseRelative(const SourceKey& source, + int32_t deltaX, int32_t deltaY, int32_t wheel, uint32_t buttons) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + if (!m_sink->SendMouseRelative(deltaX, deltaY, wheel, buttons)) + { + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + return InputResult::UNAVAILABLE; + } + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} + +InputResult CInputHub::SendMouseAbsolute(const SourceKey& source, + uint16_t x, uint16_t y, int32_t wheel, uint32_t buttons) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + if (!m_sink->SendMouseAbsolute(x, y, wheel, buttons)) + { + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + return InputResult::UNAVAILABLE; + } + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} + +InputResult CInputHub::SendKeyboard(const SourceKey& source, + uint8_t modifiers, const uint8_t * keys) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + if (!m_sink->SendKeyboard(modifiers, keys)) + { + m_sink->Reset(); + m_owner = {}; + m_ownerDeadline = 0; + return InputResult::UNAVAILABLE; + } + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} + +InputResult CInputHub::Reset(const SourceKey& source) +{ + CSRWExclusiveLock lock(m_lock); + if (!SourceValid(source)) + return InputResult::STALE; + if (!CheckState()) + return InputResult::UNAVAILABLE; + if (!OwnerValid(source)) + return m_owner.backend ? InputResult::BUSY : InputResult::STALE; + if (!m_sink->Reset()) + { + m_owner = {}; + m_ownerDeadline = 0; + return InputResult::UNAVAILABLE; + } + m_ownerDeadline = GetTickCount64() + OWNER_LEASE_MS; + return InputResult::ACCEPTED; +} diff --git a/idd/LGIdd/transport/CInputHub.h b/idd/LGIdd/transport/CInputHub.h new file mode 100644 index 00000000..08d83d42 --- /dev/null +++ b/idd/LGIdd/transport/CInputHub.h @@ -0,0 +1,108 @@ +/** + * 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/IInputSource.h" +#include "transport/IInputTransport.h" +#include "transport/ITransport.h" + +#include + +class IInputSink; + +class CInputHub final : public IInputTransport +{ +private: + static constexpr unsigned MAX_SOURCES = 8; + static constexpr uint64_t OWNER_LEASE_MS = 500; + + struct Source final : public IInputTarget + { + CInputHub * owner = nullptr; + BackendId backend = 0; + uint32_t epoch = 0; + IInputSource * endpoint = nullptr; + bool running = false; + bool active = false; + bool reserved = false; + bool failed = false; + bool failurePending = false; + + SourceKey Key(const InputSourceId& source) const; + InputTargetState GetState(const InputSourceId& source) override; + void Failed() override; + InputResult Claim(const InputSourceId& source) override; + InputResult Touch(const InputSourceId& source) override; + InputResult Release( + const InputSourceId& source, bool reset) override; + InputResult SendMouseRelative(const InputSourceId& source, + int32_t deltaX, int32_t deltaY, int32_t wheel, + uint32_t buttons) override; + InputResult SendMouseAbsolute(const InputSourceId& source, + uint16_t x, uint16_t y, int32_t wheel, + uint32_t buttons) override; + InputResult SendKeyboard(const InputSourceId& source, + uint8_t modifiers, const uint8_t * keys) override; + InputResult Reset(const InputSourceId& source) override; + }; + + mutable CSRWLock m_lock; + CSRWLock m_lifecycleLock; + Source m_sources[MAX_SOURCES]; + IInputSink * m_sink = nullptr; + uint64_t m_sinkState = 0; + SourceKey m_owner; + uint64_t m_ownerDeadline = 0; + 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; + bool CheckState(); + InputTargetState GetState(const SourceKey& source); + void Failed(Source& source); + InputResult Claim(const SourceKey& source); + InputResult Touch(const SourceKey& source); + InputResult Release(const SourceKey& source, bool reset); + InputResult SendMouseRelative(const SourceKey& source, + int32_t deltaX, int32_t deltaY, int32_t wheel, uint32_t buttons); + InputResult SendMouseAbsolute(const SourceKey& source, + uint16_t x, uint16_t y, int32_t wheel, uint32_t buttons); + InputResult SendKeyboard(const SourceKey& source, + uint8_t modifiers, const uint8_t * keys); + InputResult Reset(const SourceKey& source); + +public: + CInputHub(); + ~CInputHub() override; + + CInputHub(const CInputHub&) = delete; + CInputHub& operator=(const CInputHub&) = delete; + + bool Bind(BackendId backend, uint32_t epoch, IInputSource& source); + void Unbind(BackendId backend, uint32_t epoch); + bool TakeFailure(SourceKey& source); + + bool Start(IInputSink& sink) override; + void Stop() override; +}; diff --git a/idd/LGIdd/transport/CTransportManager.cpp b/idd/LGIdd/transport/CTransportManager.cpp index 25e0aca0..4ce34cd5 100644 --- a/idd/LGIdd/transport/CTransportManager.cpp +++ b/idd/LGIdd/transport/CTransportManager.cpp @@ -27,6 +27,7 @@ #include static const uint64_t RETRY_DELAY_MS = 500; +static const uint64_t SERVICE_RETRY_DELAY_MS = 250; class CSourceEvents final : public ITransportEvents { @@ -353,7 +354,14 @@ bool CTransportManager::AddServices(Entry& entry) uint32_t epoch = 0; bool primary = false; bool controlAdded = false; + bool controlFailed = false; + bool controlAbsent = false; + bool inputAdded = false; + bool inputFailed = false; + bool inputAbsent = false; bool frameAdded = false; + bool frameAbsent = false; + uint64_t retryAt = 0; { CSRWSharedLock entryLock(entry.lock); transport = entry.transport; @@ -361,38 +369,149 @@ bool CTransportManager::AddServices(Entry& entry) epoch = entry.epoch; primary = entry.primary; controlAdded = entry.controlAdded; + controlFailed = entry.controlFailed; + controlAbsent = entry.controlAbsent; + inputAdded = entry.inputAdded; + inputFailed = entry.inputFailed; + inputAbsent = entry.inputAbsent; frameAdded = entry.frameAdded; + frameAbsent = entry.frameAbsent; + retryAt = entry.serviceRetryAt; } if (!transport) - return false; + return !primary; - if (!controlAdded) + const uint64_t now = GetTickCount64(); + const bool attach = now >= retryAt; + bool controlRetry = false; + bool frameRetry = false; + bool inputRetry = false; + if (attach && !controlAdded && !controlFailed && !controlAbsent) { - if (!m_control.Add(id, epoch, transport->Control())) - return false; - controlAdded = true; - CSRWExclusiveLock entryLock(entry.lock); - entry.controlAdded = true; + IControlSink * control = transport->Control(); + if (control && m_control.Add(id, epoch, *control)) + { + controlAdded = true; + CSRWExclusiveLock entryLock(entry.lock); + entry.controlAdded = true; + } + else if (control) + controlRetry = true; + else + { + controlAbsent = true; + CSRWExclusiveLock entryLock(entry.lock); + entry.controlAbsent = true; + } } - if (!frameAdded && - m_frames.Bind(id, epoch, primary, transport->FrameSink())) + if (attach && !frameAdded && !frameAbsent) { - CSRWExclusiveLock entryLock(entry.lock); - entry.frameAdded = true; - return true; + IFrameSink * frame = transport->FrameSink(); + if (frame && m_frames.Bind(id, epoch, primary, *frame)) + { + CSRWExclusiveLock entryLock(entry.lock); + entry.frameAdded = true; + frameAdded = true; + } + else if (frame || primary) + frameRetry = true; + else + { + frameAbsent = true; + CSRWExclusiveLock entryLock(entry.lock); + entry.frameAbsent = true; + } } - if (frameAdded) - return true; + if (attach && !inputAdded && !inputFailed && !inputAbsent) + { + IInputSource * input = transport->Input(); + if (input && m_input.Bind(id, epoch, *input)) + { + CSRWExclusiveLock entryLock(entry.lock); + entry.inputAdded = true; + inputAdded = true; + } + else if (input) + inputRetry = true; + else + { + inputAbsent = true; + CSRWExclusiveLock entryLock(entry.lock); + entry.inputAbsent = true; + } + } - m_control.Remove(id, epoch); + if (attach && (controlRetry || inputRetry || frameRetry)) { CSRWExclusiveLock entryLock(entry.lock); - entry.controlAdded = false; + entry.serviceRetryAt = now + SERVICE_RETRY_DELAY_MS; + } + + return !primary || frameAdded; +} + +void CTransportManager::HandleServiceFailures() +{ + ControlToken token; + while (m_control.TakeFailure(token)) + { + Entry * entries[FRAME_MAX_SINKS] = {}; + const unsigned count = Entries(entries); + for (unsigned i = 0; i < count; ++i) + { + Entry& entry = *entries[i]; + bool restart = false; + { + CSRWExclusiveLock entryLock(entry.lock); + if (entry.id != token.backend || entry.epoch != token.epoch || + !entry.controlAdded) + continue; + entry.controlAdded = false; + entry.controlFailed = true; + restart = !entry.exposed; + } + + m_control.Remove(token.backend, token.epoch); + if (restart) + { + RemoveServices(entry); + ScheduleRetry(entry); + } + break; + } + } + + SourceKey source; + while (m_input.TakeFailure(source)) + { + Entry * entries[FRAME_MAX_SINKS] = {}; + const unsigned count = Entries(entries); + for (unsigned i = 0; i < count; ++i) + { + Entry& entry = *entries[i]; + bool restart = false; + { + CSRWExclusiveLock entryLock(entry.lock); + if (entry.id != source.backend || entry.epoch != source.epoch || + !entry.inputAdded) + continue; + entry.inputAdded = false; + entry.inputFailed = true; + restart = !entry.exposed; + } + + m_input.Unbind(source.backend, source.epoch); + if (restart) + { + RemoveServices(entry); + ScheduleRetry(entry); + } + break; + } } - return false; } bool CTransportManager::SetupEntry(Entry& entry, size_t alignment) @@ -400,9 +519,7 @@ bool CTransportManager::SetupEntry(Entry& entry, size_t alignment) std::shared_ptr transport; { CSRWSharedLock entryLock(entry.lock); - if (entry.state == State::READY) - return true; - if (entry.state != State::INITIALIZED) + if (entry.state != State::INITIALIZED && entry.state != State::READY) return false; transport = entry.transport; } @@ -449,20 +566,25 @@ void CTransportManager::RemoveServices(Entry& entry) uint32_t epoch = 0; bool frameAdded = false; bool controlAdded = false; + bool inputAdded = false; { CSRWExclusiveLock entryLock(entry.lock); id = entry.id; epoch = entry.epoch; frameAdded = entry.frameAdded; controlAdded = entry.controlAdded; + inputAdded = entry.inputAdded; entry.frameAdded = false; entry.controlAdded = false; + entry.inputAdded = false; } - if (frameAdded) - m_frames.Unbind(id, epoch); + if (inputAdded) + m_input.Unbind(id, epoch); if (controlAdded) m_control.Remove(id, epoch); + if (frameAdded) + m_frames.Unbind(id, epoch); } void CTransportManager::RetryEntry(Entry& entry, uint64_t now, @@ -489,6 +611,14 @@ void CTransportManager::RetryEntry(Entry& entry, uint64_t now, entry.directMemory = DirectFrameBufferMemory {}; entry.directMemoryValid = false; entry.setupDone = false; + entry.controlFailed = false; + entry.controlAbsent = false; + entry.inputFailed = false; + entry.inputAbsent = false; + entry.frameAbsent = false; + entry.serviceRetryAt = 0; + entry.recoveryPending = false; + entry.recovery = RecoveryUpdate {}; ++entry.epoch; if (!entry.epoch) ++entry.epoch; @@ -503,8 +633,12 @@ void CTransportManager::RetryEntry(Entry& entry, uint64_t now, } if (setup && !SetupEntry(entry, alignment)) { - RemoveServices(entry); - ScheduleRetry(entry); + CSRWSharedLock entryLock(entry.lock); + if (entry.state == State::FAILED) + { + entryLock.Unlock(); + ScheduleRetry(entry); + } } } @@ -566,7 +700,7 @@ ITransport::OpenResult CTransportManager::Open() const unsigned count = Entries(entries); OpenResult aggregate = Primary() ? OpenResult::SUCCESS : OpenResult::FAILURE; - for (unsigned i = 0; i < count && aggregate != OpenResult::FAILURE; ++i) + for (unsigned i = 0; i < count; ++i) { Entry& entry = *entries[i]; if (!BeginCall(entry, Call::LIFECYCLE, true)) @@ -593,8 +727,11 @@ ITransport::OpenResult CTransportManager::Open() } EndCall(entry, transport); - if (entry.required && result != OpenResult::SUCCESS) - aggregate = result; + if (entry.required && result == OpenResult::FAILURE) + aggregate = OpenResult::FAILURE; + else if (entry.required && result == OpenResult::RETRY && + aggregate == OpenResult::SUCCESS) + aggregate = OpenResult::RETRY; } EndPhase(); @@ -609,7 +746,7 @@ bool CTransportManager::Initialize() Entry * entries[FRAME_MAX_SINKS] = {}; const unsigned count = Entries(entries); bool success = true; - for (unsigned i = 0; i < count && success; ++i) + for (unsigned i = 0; i < count; ++i) { Entry& entry = *entries[i]; if (!BeginCall(entry, Call::LIFECYCLE, true)) @@ -676,12 +813,12 @@ bool CTransportManager::Setup(size_t alignment) Entry * entries[FRAME_MAX_SINKS] = {}; const unsigned count = Entries(entries); bool success = initialized; - for (unsigned i = 0; i < count && success; ++i) + for (unsigned i = 0; i < count; ++i) { Entry& entry = *entries[i]; if (!BeginCall(entry, Call::LIFECYCLE, true)) { - if (entry.required) + if (entry.primary) success = false; continue; } @@ -694,12 +831,16 @@ bool CTransportManager::Setup(size_t alignment) transport = entry.transport; } - if (state == State::INITIALIZED && !SetupEntry(entry, alignment)) + if ((state == State::INITIALIZED || state == State::READY) && + !SetupEntry(entry, alignment)) { - RemoveServices(entry); - if (entry.required) + { + CSRWSharedLock entryLock(entry.lock); + state = entry.state; + } + if (entry.primary) success = false; - else + else if (state == State::FAILED) ScheduleRetry(entry); } { @@ -709,14 +850,17 @@ bool CTransportManager::Setup(size_t alignment) EndCall(entry, transport); } - for (unsigned i = 0; i < count && success; ++i) + Entry * primary = Primary(); + if (primary) { - CSRWSharedLock entryLock(entries[i]->lock); - if (entries[i]->required && entries[i]->state != State::READY) + CSRWSharedLock entryLock(primary->lock); + if (primary->state != State::READY || !primary->frameAdded) success = false; } + else + success = false; - if (success) + if (initialized) { CSRWExclusiveLock managerLock(m_lock); m_setup = true; @@ -746,6 +890,7 @@ ITransport::ProcessResult CTransportManager::Process( } const uint64_t now = GetTickCount64(); + HandleServiceFailures(); Entry * entries[FRAME_MAX_SINKS] = {}; const unsigned count = Entries(entries); for (unsigned i = 0; i < count; ++i) @@ -793,6 +938,9 @@ ITransport::ProcessResult CTransportManager::Process( entry.setupDone = true; } + if (setup) + SetupEntry(entry, alignment); + DrainRecovery(entry, transport); CSourceEvents sourceEvents(id, epoch, events); const ProcessResult result = transport->Process(sourceEvents); @@ -880,6 +1028,11 @@ void CTransportManager::Stop() return; } + for (unsigned i = count; i > 0; --i) + RemoveServices(*entries[i - 1]); + + m_input.Stop(); + for (unsigned i = count; i > 0; --i) { Entry& entry = *entries[i - 1]; @@ -891,7 +1044,6 @@ void CTransportManager::Stop() state = entry.state; } - RemoveServices(entry); if (transport && state != State::STOPPED) transport->Stop(); { @@ -952,8 +1104,9 @@ void CTransportManager::SyncRecovery() } } -void CTransportManager::RecoveryStatus(uint64_t session, uint32_t serial, - bool active, Recovery state, uint32_t error) +void CTransportManager::RecoveryStatus(const SourceKey& source, + uint64_t session, uint32_t serial, bool active, + Recovery state, uint32_t error) { { CSRWSharedLock managerLock(m_lock); @@ -970,7 +1123,8 @@ void CTransportManager::RecoveryStatus(uint64_t session, uint32_t serial, bool call = false; { CSRWExclusiveLock entryLock(entry.lock); - if (entry.stopRequested || !entry.transport || + if (entry.id != source.backend || entry.epoch != source.epoch || + entry.stopRequested || !entry.transport || (entry.state != State::INITIALIZED && entry.state != State::READY)) continue; @@ -995,6 +1149,7 @@ void CTransportManager::RecoveryStatus(uint64_t session, uint32_t serial, if (call) transport->RecoveryStatus(session, serial, active, state, error); EndCall(entry, transport); + return; } } @@ -1095,25 +1250,7 @@ IControlTransport& CTransportManager::Control() return m_control; } -IInputTransport * CTransportManager::Input() +IInputTransport& CTransportManager::Input() { - Entry * primary = Primary(); - if (!primary || !BeginPhase(Phase::ACCESS, true)) - return nullptr; - if (!BeginCall(*primary, Call::ACCESS, true)) - { - EndPhase(); - return nullptr; - } - - std::shared_ptr transport; - { - CSRWSharedLock entryLock(primary->lock); - transport = primary->transport; - } - Expose(*primary); - IInputTransport * input = transport ? transport->Input() : nullptr; - EndCall(*primary, transport); - EndPhase(); - return input; + return m_input; } diff --git a/idd/LGIdd/transport/CTransportManager.h b/idd/LGIdd/transport/CTransportManager.h index e3dcc406..ae3424da 100644 --- a/idd/LGIdd/transport/CTransportManager.h +++ b/idd/LGIdd/transport/CTransportManager.h @@ -20,8 +20,10 @@ #pragma once +#include "CSRWLock.h" #include "transport/CControlHub.h" #include "transport/CFrameHub.h" +#include "transport/CInputHub.h" #include "transport/ITransport.h" #include @@ -96,8 +98,15 @@ private: State state = State::CLOSED; uint32_t epoch = 1; uint64_t retryAt = 0; - bool controlAdded = false; - bool frameAdded = false; + bool controlAdded = false; + bool controlFailed = false; + bool controlAbsent = false; + bool inputAdded = false; + bool inputFailed = false; + bool inputAbsent = false; + bool frameAdded = false; + bool frameAbsent = false; + uint64_t serviceRetryAt = 0; bool exposed = false; bool setupDone = false; bool syncPending = false; @@ -114,6 +123,7 @@ private: unsigned m_entryCount = 0; CControlHub m_control; CFrameHub m_frames; + CInputHub m_input; Entry * m_primary = nullptr; bool m_initialized = false; bool m_setup = false; @@ -141,6 +151,7 @@ private: bool InitializeEntry(Entry& entry); bool SetupEntry(Entry& entry, size_t alignment); bool AddServices(Entry& entry); + void HandleServiceFailures(); void RetryEntry(Entry& entry, uint64_t now, bool initialized, bool setup, size_t alignment); void HandleProcessResult(Entry& entry, ProcessResult result); @@ -164,7 +175,8 @@ public: ProcessResult Process(ITransportEvents& events); void Stop(); void SyncRecovery(); - void RecoveryStatus(uint64_t session, uint32_t serial, bool active, + void RecoveryStatus(const SourceKey& source, + uint64_t session, uint32_t serial, bool active, Recovery state, uint32_t error); FrameMemoryLimits GetMemoryLimits() const; @@ -172,5 +184,5 @@ public: IFrameTransport& Frames(); IControlTransport& Control(); - IInputTransport * Input(); + IInputTransport& Input(); }; diff --git a/idd/LGIdd/transport/IControlSink.h b/idd/LGIdd/transport/IControlSink.h new file mode 100644 index 00000000..1c5b4fbf --- /dev/null +++ b/idd/LGIdd/transport/IControlSink.h @@ -0,0 +1,69 @@ +/** + * 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 "postprocess/D12FrameFormat.h" + +#include +#include +#include + +#include +#include +#include + +struct ControlToken +{ + uint32_t backend = 0; + uint32_t epoch = 0; +}; + +enum class ControlResult +{ + APPLIED, + RETRY, + FAILED, +}; + +class IControlEvents +{ +public: + virtual ~IControlEvents() = default; + + virtual void OnControlReplay(const ControlToken& token) = 0; +}; + +class IControlSink +{ +public: + virtual ~IControlSink() = default; + + // Passing nullptr is a callback-quiescence barrier. Once this returns, no + // callback using the previous events pointer may still be running. + virtual void SetControlEvents( + IControlEvents * events, const ControlToken& token) = 0; + + // These calls perform at most one delivery attempt. + virtual ControlResult SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, + const BYTE * data, size_t size, UINT sdrWhiteLevel) = 0; + virtual ControlResult SetColorTransform( + std::shared_ptr transform) = 0; +}; diff --git a/idd/LGIdd/transport/IInputSource.h b/idd/LGIdd/transport/IInputSource.h new file mode 100644 index 00000000..4ec6fbe3 --- /dev/null +++ b/idd/LGIdd/transport/IInputSource.h @@ -0,0 +1,76 @@ +/** + * 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 + +struct InputSourceId +{ + uint32_t client = 0; + uint32_t generation = 0; +}; + +struct InputTargetState +{ + uint64_t state = 0; + InputSourceId owner; + bool available = false; + bool owned = false; +}; + +enum class InputResult +{ + ACCEPTED, + BUSY, + UNAVAILABLE, + STALE, +}; + +class IInputTarget +{ +public: + virtual ~IInputTarget() = default; + + virtual InputTargetState GetState(const InputSourceId& source) = 0; + virtual void Failed() = 0; + virtual InputResult Claim(const InputSourceId& source) = 0; + virtual InputResult Touch(const InputSourceId& source) = 0; + virtual InputResult Release( + const InputSourceId& source, bool reset) = 0; + virtual InputResult SendMouseRelative(const InputSourceId& source, + int32_t deltaX, int32_t deltaY, int32_t wheel, uint32_t buttons) = 0; + virtual InputResult SendMouseAbsolute(const InputSourceId& source, + uint16_t x, uint16_t y, int32_t wheel, uint32_t buttons) = 0; + virtual InputResult SendKeyboard(const InputSourceId& source, + uint8_t modifiers, const uint8_t * keys) = 0; + virtual InputResult Reset(const InputSourceId& source) = 0; +}; + +class IInputSource +{ +public: + virtual ~IInputSource() = default; + + // Start may issue target calls before it returns. Stop is a callback- + // quiescence barrier and leaves no work that can access the target. + virtual bool Start(IInputTarget& target) = 0; + virtual void Stop() = 0; +}; diff --git a/idd/LGIdd/transport/ITransport.h b/idd/LGIdd/transport/ITransport.h index 26d6167d..b6607449 100644 --- a/idd/LGIdd/transport/ITransport.h +++ b/idd/LGIdd/transport/ITransport.h @@ -26,9 +26,9 @@ #include #include -class IControlTransport; +class IControlSink; class IFrameSink; -class IInputTransport; +class IInputSource; using BackendId = uint32_t; @@ -93,7 +93,8 @@ public: virtual FrameMemoryLimits GetMemoryLimits() const = 0; virtual DirectFrameBufferMemory GetDirectMemory() const = 0; - virtual IFrameSink& FrameSink() = 0; - virtual IControlTransport& Control() = 0; - virtual IInputTransport * Input() { return nullptr; } + // Component pointers are fixed after Setup and remain valid until Stop. + virtual IFrameSink * FrameSink() { return nullptr; } + virtual IControlSink * Control() { return nullptr; } + virtual IInputSource * Input() { return nullptr; } }; diff --git a/idd/LGIdd/transport/lgmp/CLGMPControl.cpp b/idd/LGIdd/transport/lgmp/CLGMPControl.cpp index bba7473c..e3384376 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPControl.cpp +++ b/idd/LGIdd/transport/lgmp/CLGMPControl.cpp @@ -98,6 +98,7 @@ bool CLGMPControl::Initialize() void CLGMPControl::DeInit() { + SetControlEvents(nullptr, {}); for (int i = 0; i < LGMP_Q_POINTER_LEN; ++i) lgmpHostMemFree(&m_pointerMemory[i]); for (int i = 0; i < POINTER_SHAPE_BUFFERS; ++i) @@ -106,7 +107,6 @@ void CLGMPControl::DeInit() lgmpHostMemFree(&m_pointerTransformMemory[i]); m_pointerQueue = nullptr; - m_pointerShape = nullptr; m_pointerMemoryIndex = 0; m_pointerShapeIndex = 0; m_pointerTransformIndex = 0; @@ -129,43 +129,101 @@ bool CLGMPControl::HasNewSubscribers() return lgmpHostQueueNewSubs(m_pointerQueue) != 0; } -void CLGMPControl::SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, - const BYTE * data, UINT sdrWhiteLevel) +void CLGMPControl::SetControlEvents( + IControlEvents * events, const ControlToken& token) { - PLGMPMemory mem; - if (info.CursorShapeInfo.CursorType == IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED) + CSRWExclusiveLock lock(m_eventLock); + m_events = events; + m_token = events ? token : ControlToken {}; +} + +PLGMPMemory CLGMPControl::FindAvailable( + PLGMPMemory * memory, int count, int& index) const +{ + for (int offset = 0; offset < count; ++offset) { - mem = m_pointerMemory[m_pointerMemoryIndex]; - if (++m_pointerMemoryIndex == LGMP_Q_POINTER_LEN) - m_pointerMemoryIndex = 0; + const int candidate = (index + offset) % count; + if (memory[candidate] && + !lgmpHostQueuePayloadPending(m_pointerQueue, memory[candidate])) + { + index = candidate; + return memory[candidate]; + } + } + return nullptr; +} + +ControlResult CLGMPControl::SendCursor( + const IDARG_OUT_QUERY_HWCURSOR& info, + const BYTE * data, size_t size, UINT sdrWhiteLevel) +{ + if (!m_pointerQueue) + return ControlResult::FAILED; + + const bool hasShape = info.CursorShapeInfo.CursorType != + IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED; + if (hasShape) + { + if (info.CursorShapeInfo.CursorType != IDDCX_CURSOR_SHAPE_TYPE_ALPHA && + info.CursorShapeInfo.CursorType != + IDDCX_CURSOR_SHAPE_TYPE_MASKED_COLOR) + { + DEBUG_ERROR("Unsupported pointer shape type: %u", + static_cast(info.CursorShapeInfo.CursorType)); + return ControlResult::FAILED; + } + + if (info.CursorShapeInfo.Height && + info.CursorShapeInfo.Pitch > SIZE_MAX / info.CursorShapeInfo.Height) + { + DEBUG_ERROR("Pointer shape size overflow"); + return ControlResult::FAILED; + } + const size_t required = static_cast( + info.CursorShapeInfo.Height) * info.CursorShapeInfo.Pitch; + if (required != size || (required && !data) || + required > MAX_POINTER_SIZE - sizeof(KVMFRCursor)) + { + DEBUG_ERROR("Invalid pointer shape payload: %zu bytes", size); + return ControlResult::FAILED; + } + } + + PLGMPMemory mem; + int * index; + int count; + if (!hasShape) + { + index = &m_pointerMemoryIndex; + count = LGMP_Q_POINTER_LEN; + mem = FindAvailable(m_pointerMemory, count, *index); } else { - mem = m_pointerShapeMemory[m_pointerShapeIndex]; - if (++m_pointerShapeIndex == POINTER_SHAPE_BUFFERS) - m_pointerShapeIndex = 0; + index = &m_pointerShapeIndex; + count = POINTER_SHAPE_BUFFERS; + mem = FindAvailable(m_pointerShapeMemory, count, *index); } + if (!mem) + return ControlResult::RETRY; KVMFRCursor * cursor = (KVMFRCursor *)lgmpHostMemPtr(mem); cursor->sdrWhiteLevel = sdrWhiteLevel ? sdrWhiteLevel : KVMFR_SDR_WHITE_LEVEL_DEFAULT; - m_cursorVisible = info.IsCursorVisible; uint32_t flags = CURSOR_FLAG_VISIBLE_VALID; if (info.IsCursorVisible) { - m_cursorX = info.X; - m_cursorY = info.Y; cursor->x = (int16_t)info.X; cursor->y = (int16_t)info.Y; flags |= CURSOR_FLAG_POSITION | CURSOR_FLAG_VISIBLE; } - if (info.CursorShapeInfo.CursorType != IDDCX_CURSOR_SHAPE_TYPE_UNINITIALIZED) + if (hasShape) { - memcpy(cursor + 1, data, - (size_t)info.CursorShapeInfo.Height * info.CursorShapeInfo.Pitch); + if (size) + memcpy(cursor + 1, data, size); cursor->hx = (int8_t )info.CursorShapeInfo.XHot; cursor->hy = (int8_t )info.CursorShapeInfo.YHot; @@ -185,56 +243,43 @@ void CLGMPControl::SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, } flags |= CURSOR_FLAG_SHAPE; - m_pointerShape = mem; } - LGMP_STATUS status; - while ((status = lgmpHostQueuePost( - m_pointerQueue, flags, mem)) != LGMP_OK) + const LGMP_STATUS status = + lgmpHostQueuePost(m_pointerQueue, flags, mem); + if (status == LGMP_OK) { - if (status == LGMP_ERR_QUEUE_FULL) - { - Sleep(1); - continue; - } - - DEBUG_ERROR("lgmpHostQueuePost Failed (Pointer): %s", - lgmpStatusString(status)); - break; + *index = (*index + 1) % count; + return ControlResult::APPLIED; } + if (status == LGMP_ERR_QUEUE_FULL) + return ControlResult::RETRY; + + DEBUG_ERROR("lgmpHostQueuePost Failed (Pointer): %s", + lgmpStatusString(status)); + return ControlResult::FAILED; } -void CLGMPControl::SetColorTransform( +ControlResult CLGMPControl::SetColorTransform( std::shared_ptr transform) { - { - CSRWExclusiveLock lock(m_colorTransformLock); - m_colorTransform = std::move(transform); - } - SendColorTransform(); + return SendColorTransform(transform); } -std::shared_ptr -CLGMPControl::GetColorTransform() const -{ - CSRWSharedLock lock(m_colorTransformLock); - std::shared_ptr transform = m_colorTransform; - return transform; -} - -void CLGMPControl::SendColorTransform() +ControlResult CLGMPControl::SendColorTransform( + const std::shared_ptr& transform) { if (!m_pointerQueue || !m_pointerTransformMemory[0]) - return; + return ControlResult::FAILED; - PLGMPMemory mem = m_pointerTransformMemory[m_pointerTransformIndex]; - if (++m_pointerTransformIndex == COLOR_TRANSFORM_BUFFERS) - m_pointerTransformIndex = 0; + PLGMPMemory mem = FindAvailable(m_pointerTransformMemory, + COLOR_TRANSFORM_BUFFERS, m_pointerTransformIndex); + if (!mem) + return ControlResult::RETRY; KVMFRCursor * cursor = (KVMFRCursor *)lgmpHostMemPtr(mem); KVMFRColorTransform * output = (KVMFRColorTransform *)(cursor + 1); - const auto transform = GetColorTransform(); output->flags = 0; if (transform) @@ -248,54 +293,25 @@ void CLGMPControl::SendColorTransform() memcpy(output->lut, transform->lut, sizeof(output->lut)); } - LGMP_STATUS status; - while ((status = lgmpHostQueuePost(m_pointerQueue, - CURSOR_FLAG_COLOR_TRANSFORM, mem)) != LGMP_OK) + const LGMP_STATUS status = lgmpHostQueuePost(m_pointerQueue, + CURSOR_FLAG_COLOR_TRANSFORM, mem); + if (status == LGMP_OK) { - if (status == LGMP_ERR_QUEUE_FULL) - { - Sleep(1); - continue; - } - - DEBUG_ERROR("lgmpHostQueuePost Failed (Pointer Transform): %s", - lgmpStatusString(status)); - break; + m_pointerTransformIndex = + (m_pointerTransformIndex + 1) % COLOR_TRANSFORM_BUFFERS; + return ControlResult::APPLIED; } + if (status == LGMP_ERR_QUEUE_FULL) + return ControlResult::RETRY; + + DEBUG_ERROR("lgmpHostQueuePost Failed (Pointer Transform): %s", + lgmpStatusString(status)); + return ControlResult::FAILED; } -void CLGMPControl::ResendCursor() +void CLGMPControl::RequestReplay() { - PLGMPMemory mem = m_pointerShape; - if (!mem) - return; - - KVMFRCursor* cursor = (KVMFRCursor*)lgmpHostMemPtr(mem); - cursor->x = (int16_t)m_cursorX; - cursor->y = (int16_t)m_cursorY; - - const uint32_t flags = - CURSOR_FLAG_POSITION | CURSOR_FLAG_SHAPE | CURSOR_FLAG_VISIBLE_VALID | - (m_cursorVisible ? CURSOR_FLAG_VISIBLE : 0); - - LGMP_STATUS status; - while ((status = lgmpHostQueuePost( - m_pointerQueue, flags, mem)) != LGMP_OK) - { - if (status == LGMP_ERR_QUEUE_FULL) - { - Sleep(1); - continue; - } - - DEBUG_ERROR("lgmpHostQueuePost Failed (Pointer): %s", - lgmpStatusString(status)); - break; - } -} - -void CLGMPControl::ResendState() -{ - ResendCursor(); - SendColorTransform(); + CSRWSharedLock lock(m_eventLock); + if (m_events) + m_events->OnControlReplay(m_token); } diff --git a/idd/LGIdd/transport/lgmp/CLGMPControl.h b/idd/LGIdd/transport/lgmp/CLGMPControl.h index 3540b629..aec71511 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPControl.h +++ b/idd/LGIdd/transport/lgmp/CLGMPControl.h @@ -23,7 +23,7 @@ #include "CSRWLock.h" #include "transport/lgmp/CLGMPHost.h" -#include "transport/IControlTransport.h" +#include "transport/IControlSink.h" #include "common/KVMFR.h" @@ -35,7 +35,7 @@ class CLGMPTransport; -class CLGMPControl final : public IControlTransport +class CLGMPControl final : public IControlSink { private: friend class CLGMPTransport; @@ -49,26 +49,25 @@ private: PLGMPMemory m_pointerMemory[LGMP_Q_POINTER_LEN] = {}; PLGMPMemory m_pointerShapeMemory[POINTER_SHAPE_BUFFERS] = {}; PLGMPMemory m_pointerTransformMemory[COLOR_TRANSFORM_BUFFERS] = {}; - PLGMPMemory m_pointerShape = nullptr; int m_pointerMemoryIndex = 0; int m_pointerShapeIndex = 0; int m_pointerTransformIndex = 0; - bool m_cursorVisible = false; - int m_cursorX = 0; - int m_cursorY = 0; - mutable CSRWLock m_colorTransformLock; - std::shared_ptr m_colorTransform; + mutable CSRWLock m_eventLock; + IControlEvents * m_events = nullptr; + ControlToken m_token = {}; - void SendColorTransform(); - void ResendCursor(); + PLGMPMemory FindAvailable( + PLGMPMemory * memory, int count, int& index) const; + ControlResult SendColorTransform( + const std::shared_ptr& transform); bool Initialize(); void DeInit(); LGMP_STATUS ReadDataWithSource(void * data, size_t * size, uint32_t * sourceClientID); LGMP_STATUS AckData(); bool HasNewSubscribers(); - void ResendState(); + void RequestReplay(); public: explicit CLGMPControl(CLGMPHost& host) : @@ -78,10 +77,10 @@ public: CLGMPControl(const CLGMPControl&) = delete; CLGMPControl& operator=(const CLGMPControl&) = delete; - void SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, const BYTE * data, - UINT sdrWhiteLevel) override; - void SetColorTransform( + void SetControlEvents( + IControlEvents * events, const ControlToken& token) override; + ControlResult SendCursor(const IDARG_OUT_QUERY_HWCURSOR& info, + const BYTE * data, size_t size, UINT sdrWhiteLevel) override; + ControlResult SetColorTransform( std::shared_ptr transform) override; - std::shared_ptr - GetColorTransform() const override; }; diff --git a/idd/LGIdd/transport/lgmp/CLGMPInputTransport.cpp b/idd/LGIdd/transport/lgmp/CLGMPInputTransport.cpp index 280731f7..83b79daa 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPInputTransport.cpp +++ b/idd/LGIdd/transport/lgmp/CLGMPInputTransport.cpp @@ -20,7 +20,6 @@ #include "transport/lgmp/CLGMPInputTransport.h" -#include "input/IInputSink.h" #include "transport/lgmp/CLGMPHost.h" #include "CDebug.h" #include "CSRWLock.h" @@ -101,7 +100,10 @@ bool CLGMPInputTransport::Initialize() memset(lgmpHostMemPtr(memory), 0, sizeof(KVMFRInputStatus)); } - m_statusDirty = true; + { + CSRWExclusiveLock lock(m_statusLock); + m_statusDirty = true; + } return true; } @@ -113,26 +115,44 @@ void CLGMPInputTransport::DeInit() m_queue = nullptr; } -void CLGMPInputTransport::UpdateSinkState(uint64_t state) +InputSourceId CLGMPInputTransport::Owner() const { - if (state == m_sinkState) + InputSourceId source; + source.client = m_ownerClientID; + source.generation = m_ownerGeneration; + return source; +} + +void CLGMPInputTransport::UpdateTargetState( + const InputTargetState& state) +{ + CSRWExclusiveLock lock(m_statusLock); + if (state.state == m_targetState.state && + state.available == m_targetState.available && + state.owned == m_targetState.owned && + state.owner.client == m_targetState.owner.client && + state.owner.generation == m_targetState.owner.generation) return; - m_sinkState = state; - if (++m_endpointGeneration == 0) - ++m_endpointGeneration; + if (state.state != m_targetState.state) + { + if (++m_endpointGeneration == 0) + ++m_endpointGeneration; + } + m_targetState = state; m_statusDirty = true; } -void CLGMPInputTransport::PublishStatus() +bool CLGMPInputTransport::PublishStatus() { + CSRWExclusiveLock lock(m_statusLock); if (!m_queue) - return; + return true; if (lgmpHostQueueNewSubs(m_queue)) m_statusDirty = true; if (!m_statusDirty || !lgmpHostQueueHasSubs(m_queue)) - return; + return true; PLGMPMemory memory = nullptr; for (PLGMPMemory candidate : m_statusMemory) @@ -142,9 +162,9 @@ void CLGMPInputTransport::PublishStatus() break; } if (!memory) - return; + return true; - const bool available = (m_sinkState & 1) != 0; + const bool available = m_targetState.available; KVMFRInputStatus status = {}; status.version = KVMFR_INPUT_VERSION; status.capabilities = available ? @@ -152,11 +172,11 @@ void CLGMPInputTransport::PublishStatus() KVMFR_INPUT_CAP_MOUSE_ABSOLUTE | KVMFR_INPUT_CAP_KEYBOARD : 0; status.flags = available ? KVMFR_INPUT_STATUS_AVAILABLE : 0; - if (m_ownerClientID) + if (m_targetState.owned) { status.flags |= KVMFR_INPUT_STATUS_HAS_OWNER; - status.ownerClientID = m_ownerClientID; - status.ownerGeneration = m_ownerGeneration; + status.ownerClientID = m_targetState.owner.client; + status.ownerGeneration = m_targetState.owner.generation; } status.generation = m_endpointGeneration; status.lease = static_cast(OWNER_LEASE_MS); @@ -173,11 +193,26 @@ void CLGMPInputTransport::PublishStatus() m_statusDirty = false; } else if (result != LGMP_ERR_QUEUE_FULL) + { DEBUG_WARN("lgmpHostQueuePost Failed (Input Status): %s", lgmpStatusString(result)); + return false; + } + return true; } -bool CLGMPInputTransport::Start(IInputSink& sink) +void CLGMPInputTransport::FlushStatus() +{ + if (m_statusFailed.load(std::memory_order_acquire) || PublishStatus()) + return; + + m_statusFailed.store(true, std::memory_order_release); + CSRWSharedLock lock(m_lifecycleLock); + if (m_stopEvent) + SetEvent(m_stopEvent); +} + +bool CLGMPInputTransport::Start(IInputTarget& target) { CSRWExclusiveLock lock(m_lifecycleLock); if (m_thread) @@ -198,7 +233,7 @@ bool CLGMPInputTransport::Start(IInputSink& sink) m_thread = nullptr; m_pollTimer = nullptr; m_stopEvent = nullptr; - m_sink = nullptr; + m_target = nullptr; } if (!m_queue) @@ -224,19 +259,25 @@ bool CLGMPInputTransport::Start(IInputSink& sink) return false; } - m_sink = &sink; - UpdateSinkState(sink.GetState()); - if (!m_endpointGeneration) + m_target = ⌖ { - m_endpointGeneration = 1; - m_statusDirty = true; + CSRWExclusiveLock statusLock(m_statusLock); + m_targetState = target.GetState({}); + if (++m_endpointGeneration == 0) + ++m_endpointGeneration; + m_statusDirty = true; } + m_statusFailed.store(false, std::memory_order_release); m_thread = CreateThread(nullptr, 0, ThreadProc, this, 0, nullptr); if (!m_thread) { DEBUG_ERROR_HR(GetLastError(), "Failed to create LGMP input worker"); - m_sink = nullptr; - m_sinkState = 0; + m_target = nullptr; + { + CSRWExclusiveLock statusLock(m_statusLock); + m_targetState = {}; + m_statusDirty = true; + } CloseHandle(m_pollTimer); CloseHandle(m_stopEvent); m_pollTimer = nullptr; @@ -271,12 +312,16 @@ void CLGMPInputTransport::Stop() CloseHandle(m_stopEvent); m_stopEvent = nullptr; } - m_sink = nullptr; + m_target = nullptr; m_ownerClientID = 0; m_ownerGeneration = 0; m_ownerSequence = 0; m_ownerDeadline = 0; - m_sinkState = 0; + { + CSRWExclusiveLock statusLock(m_statusLock); + m_targetState = {}; + m_statusDirty = true; + } } bool CLGMPInputTransport::IsOwner( @@ -294,16 +339,19 @@ bool CLGMPInputTransport::Claim( ++m_statistics.sequenceErrors; return false; } - if (!m_sink) + if (!m_target) { ++m_statistics.deliveryFailures; return false; } - const uint64_t sinkState = m_sink->GetState(); - if (!(sinkState & 1) || sinkState != m_sinkState || - !m_sink->Reset() || m_sink->GetState() != sinkState) + InputSourceId source; + source.client = sourceClientID; + source.generation = message.generation; + const InputResult result = m_target->Claim(source); + if (result != InputResult::ACCEPTED) { + UpdateTargetState(m_target->GetState(source)); ++m_statistics.deliveryFailures; return false; } @@ -312,7 +360,7 @@ bool CLGMPInputTransport::Claim( m_ownerGeneration = message.generation; m_ownerSequence = message.sequence; RenewLease(); - m_statusDirty = true; + UpdateTargetState(m_target->GetState(source)); ++m_statistics.claims; DEBUG_INFO("Input owner %u generation %u acquired", m_ownerClientID, m_ownerGeneration); @@ -332,14 +380,16 @@ void CLGMPInputTransport::ReleaseOwner( const uint32_t clientID = m_ownerClientID; const uint32_t generation = m_ownerGeneration; - if (reset && m_sink) - m_sink->Reset(); + if (m_target) + { + m_target->Release(Owner(), reset); + UpdateTargetState(m_target->GetState({})); + } m_ownerClientID = 0; m_ownerGeneration = 0; m_ownerSequence = 0; m_ownerDeadline = 0; - m_statusDirty = true; ++m_statistics.releases; DEBUG_INFO("Input owner %u generation %u released (%s)", clientID, generation, reason); @@ -347,23 +397,18 @@ void CLGMPInputTransport::ReleaseOwner( void CLGMPInputTransport::CheckOwner() { - if (!m_sink) + if (!m_target) return; - const uint64_t state = m_sink->GetState(); - if (state != m_sinkState) - { - UpdateSinkState(state); - ReleaseOwner(true, "input endpoint changed"); - return; - } + const InputTargetState state = m_target->GetState(Owner()); + UpdateTargetState(state); if (!m_ownerClientID) return; - if (!(state & 1)) + if (!state.available || !state.owned) { - ReleaseOwner(true, "input unavailable"); + ReleaseOwner(false, "input ownership changed"); return; } @@ -418,17 +463,9 @@ bool CLGMPInputTransport::ProcessMessage( uint32_t sourceClientID, const KVMFRInputMessage& message) { const bool owner = IsOwner(sourceClientID, message.generation); - const uint64_t sinkState = m_sink ? - m_sink->GetState() : m_sinkState; - if (sinkState != m_sinkState) - { - UpdateSinkState(sinkState); - if (m_ownerClientID) - { - ReleaseOwner(true, "input endpoint changed"); - return false; - } - } + InputSourceId source; + source.client = sourceClientID; + source.generation = message.generation; if (!message.generation || !message.sequence || message.reserved || !ValidatePayload(message)) @@ -439,6 +476,10 @@ bool CLGMPInputTransport::ProcessMessage( return false; } + if (m_target) + UpdateTargetState(m_target->GetState( + m_ownerClientID ? Owner() : source)); + if (message.type == KVMFR_INPUT_MESSAGE_CLAIM) { if (m_ownerClientID) @@ -474,7 +515,7 @@ bool CLGMPInputTransport::ProcessMessage( return false; } - bool accepted = false; + InputResult result = InputResult::STALE; bool inputReport = false; switch (message.type) { @@ -483,56 +524,52 @@ bool CLGMPInputTransport::ProcessMessage( return true; case KVMFR_INPUT_MESSAGE_KEEPALIVE: - accepted = true; + result = m_target ? m_target->Touch(source) : + InputResult::UNAVAILABLE; break; case KVMFR_INPUT_MESSAGE_RESET: - accepted = m_sink && m_sink->Reset(); + result = m_target ? m_target->Reset(source) : + InputResult::UNAVAILABLE; break; case KVMFR_INPUT_MESSAGE_MOUSE_RELATIVE: inputReport = true; - accepted = m_sink && m_sink->SendMouseRelative( + result = m_target ? m_target->SendMouseRelative(source, message.payload.mouseRelative.deltaX, message.payload.mouseRelative.deltaY, message.payload.mouseRelative.wheel, - message.payload.mouseRelative.buttons); + message.payload.mouseRelative.buttons) : InputResult::UNAVAILABLE; break; case KVMFR_INPUT_MESSAGE_MOUSE_ABSOLUTE: inputReport = true; - accepted = m_sink && m_sink->SendMouseAbsolute( + result = m_target ? m_target->SendMouseAbsolute(source, message.payload.mouseAbsolute.x, message.payload.mouseAbsolute.y, message.payload.mouseAbsolute.wheel, - message.payload.mouseAbsolute.buttons); + message.payload.mouseAbsolute.buttons) : InputResult::UNAVAILABLE; break; case KVMFR_INPUT_MESSAGE_KEYBOARD: inputReport = true; - accepted = m_sink && m_sink->SendKeyboard( + result = m_target ? m_target->SendKeyboard(source, message.payload.keyboard.modifiers, - message.payload.keyboard.keys); + message.payload.keyboard.keys) : InputResult::UNAVAILABLE; break; default: break; } - if (!accepted) + if (result != InputResult::ACCEPTED) { ++m_statistics.deliveryFailures; ReleaseOwner(true, "input delivery failed"); return false; } - const uint64_t deliveredState = m_sink->GetState(); - if (deliveredState != m_sinkState) - { - UpdateSinkState(deliveredState); - ReleaseOwner(true, "input endpoint changed"); - return false; - } + UpdateTargetState(m_target->GetState(source)); m_ownerSequence = message.sequence; RenewLease(); @@ -541,9 +578,9 @@ bool CLGMPInputTransport::ProcessMessage( return true; } -bool CLGMPInputTransport::DrainMessages() +bool CLGMPInputTransport::DrainMessages(bool& received) { - bool received = false; + received = false; unsigned count = 0; for (; count < 256; ++count) { @@ -558,7 +595,7 @@ bool CLGMPInputTransport::DrainMessages() { DEBUG_ERROR("lgmpHostReadData Failed (Input): %s", lgmpStatusString(status)); - break; + return false; } received = true; @@ -577,13 +614,19 @@ bool CLGMPInputTransport::DrainMessages() ProcessMessage(sourceClientID, message); } - lgmpHostAckData(m_queue); + const LGMP_STATUS ackStatus = lgmpHostAckData(m_queue); + if (ackStatus != LGMP_OK) + { + DEBUG_ERROR("lgmpHostAckData Failed (Input): %s", + lgmpStatusString(ackStatus)); + return false; + } } if (count > m_statistics.maxDrain) m_statistics.maxDrain = count; if (count == 256) ++m_statistics.drainLimit; - return received; + return true; } void CLGMPInputTransport::LogStatistics(ULONGLONG now) @@ -643,11 +686,22 @@ void CLGMPInputTransport::Thread() m_statistics = {}; m_statistics.lastLog = GetTickCount64(); const HANDLE waitHandles[] = { m_stopEvent, m_pollTimer }; + bool failed = false; for (;;) { CheckOwner(); - const bool received = DrainMessages(); - PublishStatus(); + bool received = false; + if (!DrainMessages(received)) + { + failed = true; + break; + } + if (!PublishStatus()) + { + m_statusFailed.store(true, std::memory_order_release); + failed = true; + break; + } const ULONGLONG now = GetTickCount64(); if (received) activeUntil = now + ACTIVE_POLL_MS; @@ -659,23 +713,30 @@ void CLGMPInputTransport::Thread() DEBUG_ERROR_HR(GetLastError(), "Failed to arm LGMP input timer"); if (WaitForSingleObject(m_stopEvent, 1) != WAIT_TIMEOUT) break; - continue; + failed = true; + break; } const DWORD wait = WaitForMultipleObjects( _countof(waitHandles), waitHandles, FALSE, INFINITE); if (wait == WAIT_FIRST_OBJECT_VALUE) + { + failed = m_statusFailed.load(std::memory_order_acquire); break; + } if (wait != WAIT_FIRST_OBJECT_VALUE + 1) { DEBUG_ERROR_HR(GetLastError(), "LGMP input worker wait failed"); + failed = true; break; } } ReleaseOwner(true, "transport stopped"); - UpdateSinkState(0); + UpdateTargetState({}); PublishStatus(); + if (failed && m_target) + m_target->Failed(); if (avTaskHandle) AvRevertMmThreadCharacteristics(avTaskHandle); } diff --git a/idd/LGIdd/transport/lgmp/CLGMPInputTransport.h b/idd/LGIdd/transport/lgmp/CLGMPInputTransport.h index 74b3c233..aeb96099 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPInputTransport.h +++ b/idd/LGIdd/transport/lgmp/CLGMPInputTransport.h @@ -21,11 +21,12 @@ #pragma once #include "CSRWLock.h" -#include "transport/IInputTransport.h" +#include "transport/IInputSource.h" #include "common/LGMPConfig.h" #include +#include #include extern "C" { @@ -33,10 +34,10 @@ extern "C" { } class CLGMPHost; -class IInputSink; +class IInputTarget; struct KVMFRInputMessage; -class CLGMPInputTransport final : public IInputTransport +class CLGMPInputTransport final : public IInputSource { private: static constexpr ULONGLONG OWNER_LEASE_MS = 500; @@ -63,9 +64,10 @@ private: PLGMPHostQueue m_queue = nullptr; PLGMPMemory m_statusMemory[LGMP_Q_INPUT_LEN] = {}; - IInputSink * m_sink = nullptr; + IInputTarget * m_target = nullptr; CSRWLock m_lifecycleLock; + CSRWLock m_statusLock; HANDLE m_stopEvent = nullptr; HANDLE m_pollTimer = nullptr; HANDLE m_thread = nullptr; @@ -74,18 +76,21 @@ private: uint32_t m_ownerGeneration = 0; uint32_t m_ownerSequence = 0; ULONGLONG m_ownerDeadline = 0; - uint64_t m_sinkState = 0; + InputTargetState m_targetState; uint32_t m_endpointGeneration = 0; uint32_t m_statusSerial = 0; bool m_statusDirty = false; + std::atomic m_statusFailed = false; Statistics m_statistics = {}; bool Initialize(); void DeInit(); - void UpdateSinkState(uint64_t state); - void PublishStatus(); + InputSourceId Owner() const; + void UpdateTargetState(const InputTargetState& state); + bool PublishStatus(); + void FlushStatus(); void LogStatistics(ULONGLONG now); - bool DrainMessages(); + bool DrainMessages(bool& received); bool ProcessMessage(uint32_t sourceClientID, const KVMFRInputMessage& message); bool ValidatePayload(const KVMFRInputMessage& message) const; @@ -109,6 +114,6 @@ public: CLGMPInputTransport(const CLGMPInputTransport&) = delete; CLGMPInputTransport& operator=(const CLGMPInputTransport&) = delete; - bool Start(IInputSink& sink) override; + bool Start(IInputTarget& target) override; void Stop() override; }; diff --git a/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp b/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp index 96823d3d..cd7f92b6 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp +++ b/idd/LGIdd/transport/lgmp/CLGMPTransport.cpp @@ -121,6 +121,8 @@ ITransport::ProcessResult CLGMPTransport::Process(ITransportEvents& events) return ProcessResult::FAILURE; } + m_input.FlushStatus(); + const uint64_t now = CFrameScheduler::Nanotime(); // Take the frame subscriber snapshot before processing scheduling messages, @@ -192,7 +194,7 @@ ITransport::ProcessResult CLGMPTransport::Process(ITransportEvents& events) m_frames.FinalizeSubscribers(subscribers, now); if (m_control.HasNewSubscribers()) - m_control.ResendState(); + m_control.RequestReplay(); return ProcessResult::OK; } diff --git a/idd/LGIdd/transport/lgmp/CLGMPTransport.h b/idd/LGIdd/transport/lgmp/CLGMPTransport.h index a51b98b9..c7730159 100644 --- a/idd/LGIdd/transport/lgmp/CLGMPTransport.h +++ b/idd/LGIdd/transport/lgmp/CLGMPTransport.h @@ -63,7 +63,7 @@ public: FrameMemoryLimits GetMemoryLimits() const override; DirectFrameBufferMemory GetDirectMemory() const override; - IFrameSink& FrameSink() override { return m_frames; } - IControlTransport& Control() override { return m_control; } - IInputTransport * Input() override { return &m_input; } + IFrameSink * FrameSink() override { return &m_frames; } + IControlSink * Control() override { return &m_control; } + IInputSource * Input() override { return &m_input; } };