From 40878d8a597c9e603c91795e9e260ffcde369bd1 Mon Sep 17 00:00:00 2001 From: Luke Howard Date: Mon, 14 Sep 2026 14:48:54 +1000 Subject: [PATCH] stateMachine: wait until commands need checking, instead of polling every 5 msec The state machine thread woke every 5 msec, 200 times a second, even with nothing to do. A command is sent at once when it can be, by sendAecpCommand/sendAcmpCommand or when a response frees a slot, so the thread only has to time commands out, retry them, and send a queued command once the send interval allows. Advertising, discovery and remote entity timeouts are hundreds of milliseconds or more. The thread now waits until CommandStateMachine::getNextCheckTime(), the earliest inflight command timeout or time a queued command may be sent, but no longer than 250 msec, which is soon enough for advertising, discovery and remote entity timeouts, and for the watchdog. Whenever a command goes inflight, fails to be sent, or has to wait for the send interval, the command state machine tells the manager when it needs checking, and the manager wakes the thread if that is sooner than it waits for. So the thread never delays a command, a queued command is sent as soon as the send interval has elapsed rather than on the next 5 msec tick, and an idle thread wakes 4 times a second. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_012i5Dd3gCyYqp4X7vxhSmxh --- CHANGELOG-avdecc.md | 3 ++ src/stateMachine/commandStateMachine.cpp | 34 +++++++++++++++++ src/stateMachine/commandStateMachine.hpp | 48 ++++++++++++++++++++++-- src/stateMachine/stateMachineManager.cpp | 35 +++++++++++++++-- src/stateMachine/stateMachineManager.hpp | 5 +++ 5 files changed, 118 insertions(+), 7 deletions(-) diff --git a/CHANGELOG-avdecc.md b/CHANGELOG-avdecc.md index 637daf91..6e182e1c 100644 --- a/CHANGELOG-avdecc.md +++ b/CHANGELOG-avdecc.md @@ -10,6 +10,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - Support for all IEEE OUI range for `UniqueIdentifier::getVendorID()` - Missing EntityCapability flags +### Changed +- State machines thread waits until commands next need checking, instead of waking every 5 msec, and checks advertising, discovery and remote entity timeouts at least every 250 msec + ### Fixed - Better debugger support for MacOs and Linux builds - [Crash in MacOSNative ProtocolInterface when an entity goes offline while an AECP command is inflight (semaphore disposed while in use)](https://github.com/L-Acoustics/avdecc/issues/177) diff --git a/src/stateMachine/commandStateMachine.cpp b/src/stateMachine/commandStateMachine.cpp index b280dad4..957ace2a 100644 --- a/src/stateMachine/commandStateMachine.cpp +++ b/src/stateMachine/commandStateMachine.cpp @@ -265,6 +265,40 @@ void CommandStateMachine::checkInflightCommandsTimeoutExpiracy() noexcept } } +std::chrono::time_point CommandStateMachine::getNextCheckTime() noexcept +{ + // Lock + auto const lg = std::scoped_lock{ *_manager }; + + auto nextCheckTime = std::chrono::time_point::max(); + + for (auto const& [entityID, localEntityInfo] : _commandEntities) + { + // Errors to report + if (!localEntityInfo.scheduledAecpErrors.empty() || !localEntityInfo.scheduledAcmpErrors.empty()) + { + return std::chrono::steady_clock::now(); + } + + for (auto const& [targetEntityID, inflight] : localEntityInfo.inflightAecpCommands) + { + nextCheckTime = std::min(nextCheckTime, getNextCheckTimeFor(inflight, localEntityInfo.aecpCommandsQueue, targetEntityID, getMaxInflightAecpMessages(targetEntityID), getAecpSendInterval(targetEntityID))); + } + + for (auto const& [targetMacAddress, inflight] : localEntityInfo.inflightAcmpCommands) + { + nextCheckTime = std::min(nextCheckTime, getNextCheckTimeFor(inflight, localEntityInfo.acmpCommandsQueue, targetMacAddress, getMaxInflightAcmpMessages(targetMacAddress), getAcmpSendInterval(targetMacAddress))); + } + } + + return nextCheckTime; +} + +void CommandStateMachine::scheduleCheck(std::chrono::time_point const time) noexcept +{ + _manager->scheduleStateMachinesCheck(time); +} + void CommandStateMachine::handleAecpResponse(Aecpdu const& aecpdu) noexcept { // Lock diff --git a/src/stateMachine/commandStateMachine.hpp b/src/stateMachine/commandStateMachine.hpp index 04f0cc08..ebd1c1c2 100644 --- a/src/stateMachine/commandStateMachine.hpp +++ b/src/stateMachine/commandStateMachine.hpp @@ -28,6 +28,7 @@ #include "protocolInterfaceDelegate.hpp" +#include #include #include @@ -66,6 +67,8 @@ class CommandStateMachine final void unregisterLocalEntity(entity::LocalEntity& entity) noexcept; void discardAECPCommandsTowardsEntity(la::avdecc::UniqueIdentifier const& entityID) noexcept; void checkInflightCommandsTimeoutExpiracy() noexcept; + /** When checkInflightCommandsTimeoutExpiracy next has something to do: an inflight command timing out, a queued command that may be sent, or an error to report */ + std::chrono::time_point getNextCheckTime() noexcept; void handleAecpResponse(Aecpdu const& aecpdu) noexcept; void handleAcmpResponse(Acmpdu const& acmpdu) noexcept; ProtocolInterface::Error sendAecpCommand(Aecpdu::UniquePointer&& aecpdu, ProtocolInterface::AecpCommandResultHandler const& onResult) noexcept; @@ -178,12 +181,14 @@ class CommandStateMachine final { // Schedule the result handler to be called with the returned error from the delegate info.scheduledAecpErrors.push_back(std::make_pair(error, command.resultHandler)); + scheduleCheck(std::chrono::steady_clock::now()); return it; } else { // Move the command to inflight queue resetAecpCommandTimeoutValue(command); + scheduleCheck(command.timeoutTime); return inflight.inflightCommands.insert(it, std::move(command)); } } @@ -193,8 +198,8 @@ class CommandStateMachine final // Get current time auto const now = std::chrono::steady_clock::now(); - // Check if we don't have too many inflight commands or sending too fast for this destination macAddress - if (inflight.inflightCommands.size() >= getMaxInflightAecpMessages(entityID) || !hasExpired(now, inflight.lastSendTime, getAecpSendInterval(entityID))) + // Check if we don't have too many inflight commands for this destination (a response or a timeout frees one) + if (inflight.inflightCommands.size() >= getMaxInflightAecpMessages(entityID)) { return it; } @@ -206,6 +211,13 @@ class CommandStateMachine final return it; } + // Check if we are sending too fast for this destination, in which case the state machines send it once we may + if (!hasExpired(now, inflight.lastSendTime, getAecpSendInterval(entityID))) + { + scheduleCheck(inflight.lastSendTime + getAecpSendInterval(entityID)); + return it; + } + // Remove command from queue auto command = std::move(queue.front()); queue.pop_front(); @@ -231,12 +243,14 @@ class CommandStateMachine final { // Schedule the result handler to be called with the returned error from the delegate info.scheduledAcmpErrors.push_back(std::make_pair(error, command.resultHandler)); + scheduleCheck(std::chrono::steady_clock::now()); return it; } else { // Move the command to inflight queue resetAcmpCommandTimeoutValue(command); + scheduleCheck(command.timeoutTime); return inflight.inflightCommands.insert(it, std::move(command)); } } @@ -246,8 +260,8 @@ class CommandStateMachine final // Get current time auto const now = std::chrono::steady_clock::now(); - // Check if we don't have too many inflight commands or sending too fast for this destination macAddress - if (inflight.inflightCommands.size() >= getMaxInflightAcmpMessages(targetMacAddress) || !hasExpired(now, inflight.lastSendTime, getAcmpSendInterval(targetMacAddress))) + // Check if we don't have too many inflight commands for this destination (a response or a timeout frees one) + if (inflight.inflightCommands.size() >= getMaxInflightAcmpMessages(targetMacAddress)) { return it; } @@ -259,6 +273,13 @@ class CommandStateMachine final return it; } + // Check if we are sending too fast for this destination, in which case the state machines send it once we may + if (!hasExpired(now, inflight.lastSendTime, getAcmpSendInterval(targetMacAddress))) + { + scheduleCheck(inflight.lastSendTime + getAcmpSendInterval(targetMacAddress)); + return it; + } + // Remove command from queue auto command = std::move(queue.front()); queue.pop_front(); @@ -272,6 +293,24 @@ class CommandStateMachine final return checkQueue(protocolInterface, info, macAddress, inflight, retIt); } + template + std::chrono::time_point getNextCheckTimeFor(InflightInfo const& inflight, CommandsQueue const& queues, Target const& target, size_t const maxInflightCommands, std::chrono::milliseconds const sendInterval) const noexcept + { + auto nextCheckTime = std::chrono::time_point::max(); + for (auto const& command : inflight.inflightCommands) + { + nextCheckTime = std::min(nextCheckTime, command.timeoutTime); + } + + // A queued command waits for a free slot, which a response or a timeout brings, or for the send interval to elapse + if (auto const queueIt = queues.find(target); queueIt != queues.end() && !queueIt->second.queuedCommands.empty() && inflight.inflightCommands.size() < maxInflightCommands) + { + nextCheckTime = std::min(nextCheckTime, inflight.lastSendTime + sendInterval); + } + + return nextCheckTime; + } + bool isAemUnsolicitedResponse(Aecpdu const& aecpdu) const noexcept; bool shouldRearmTimer(Aecpdu const& aecpdu) const noexcept; bool isVuUnsolicitedResponse(Aecpdu const& aecpdu) const noexcept; @@ -283,6 +322,7 @@ class CommandStateMachine final std::chrono::milliseconds getAecpSendInterval(UniqueIdentifier const& entityID) const noexcept; size_t getMaxInflightAcmpMessages(networkInterface::MacAddress const& macAddress) const noexcept; std::chrono::milliseconds getAcmpSendInterval(networkInterface::MacAddress const& macAddress) const noexcept; + void scheduleCheck(std::chrono::time_point const time) noexcept; // Private members Manager* _manager{ nullptr }; diff --git a/src/stateMachine/stateMachineManager.cpp b/src/stateMachine/stateMachineManager.cpp index 7c37ab0b..6ca96bc7 100644 --- a/src/stateMachine/stateMachineManager.cpp +++ b/src/stateMachine/stateMachineManager.cpp @@ -28,6 +28,8 @@ #include "stateMachineManager.hpp" #include "logHelper.hpp" +#include + // Only enable instrumentation in static library and in debug (for unit testing mainly) #if defined(DEBUG) && defined(la_avdecc_static_STATICS) # define SEND_INSTRUMENTATION_NOTIFICATION(eventName) la::avdecc::InstrumentationNotifier::getInstance().triggerEvent(eventName) @@ -231,6 +233,10 @@ void Manager::startStateMachines() noexcept auto& watchDog = *watchDogSharedPointer; watchDog.registerWatch("avdecc::StateMachine", std::chrono::milliseconds{ 1000u }, true); + // Commands need checking when they time out and when a queued one may be sent, which the command state machine + // tells us, but check at least this often: soon enough for advertising, discovery and remote entity timeouts, and for the watchdog + constexpr auto MaximumCheckInterval = std::chrono::milliseconds{ 250u }; + while (!_shouldTerminate) { // Check for local entities announcement @@ -248,8 +254,15 @@ void Manager::startStateMachines() noexcept // Try to detect deadlocks watchDog.alive("avdecc::StateMachine", true); - // Wait a little bit so we don't burn the CPU - std::this_thread::sleep_for(std::chrono::milliseconds(5)); + // Wait until commands next need checking, sooner if one needs it (see scheduleStateMachinesCheck), or MaximumCheckInterval + auto lock = std::unique_lock{ *this }; + auto const nextCheck = std::min(_commandStateMachine.getNextCheckTime(), std::chrono::steady_clock::now() + MaximumCheckInterval); + _nextStateMachinesCheck = nextCheck; + _stateMachinesCondition.wait_until(lock, nextCheck, + [this, nextCheck] + { + return _shouldTerminate || _nextStateMachinesCheck < nextCheck; + }); } watchDog.unregisterWatch("avdecc::StateMachine", true); }); @@ -262,7 +275,11 @@ void Manager::stopStateMachines() noexcept if (_stateMachineThread.joinable()) { // Notify the thread we are shutting down - _shouldTerminate = true; + { + auto const lg = std::scoped_lock{ *this }; + _shouldTerminate = true; + } + _stateMachinesCondition.notify_all(); // Wait for the thread to complete its pending tasks _stateMachineThread.join(); @@ -417,6 +434,18 @@ void Manager::processAcmpdu(Acmpdu const& acmpdu) noexcept } } +void Manager::scheduleStateMachinesCheck(std::chrono::time_point const time) noexcept +{ + auto const lg = std::scoped_lock{ *this }; + + // Only wake the thread if it waits for later: while it checks, it works out when to check next afterwards + if (time < _nextStateMachinesCheck) + { + _nextStateMachinesCheck = time; + _stateMachinesCondition.notify_all(); + } +} + void Manager::lock() noexcept { SEND_INSTRUMENTATION_NOTIFICATION("StateMachineManager::lock::PreLock"); diff --git a/src/stateMachine/stateMachineManager.hpp b/src/stateMachine/stateMachineManager.hpp index 68194d1e..6c14263a 100644 --- a/src/stateMachine/stateMachineManager.hpp +++ b/src/stateMachine/stateMachineManager.hpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include @@ -68,6 +69,8 @@ class Manager final void processAdpdu(Adpdu const& adpdu) noexcept; void processAecpdu(Aecpdu const& aecpdu) noexcept; void processAcmpdu(Acmpdu const& acmpdu) noexcept; + /** Has the state machines thread check no later than 'time', waking it if it waits for longer */ + void scheduleStateMachinesCheck(std::chrono::time_point const time) noexcept; /** BasicLockable concept 'lock' method for the whole StateMachine */ void lock() noexcept; @@ -124,6 +127,8 @@ class Manager final std::uint32_t _lockedCount{ 0u }; // DEBUG status for BasicLockable concept std::thread::id _lockingThreadID{}; // DEBUG status for BasicLockable concept bool _shouldTerminate{ false }; + std::condition_variable_any _stateMachinesCondition{}; /** Wakes the state machines thread, to terminate or to check sooner */ + std::chrono::time_point _nextStateMachinesCheck{}; /** When the state machines thread next checks, while it waits */ ProtocolInterface const* const _protocolInterface{ nullptr }; std::thread _stateMachineThread{}; // Can safely be declared here, will be joined during destruction LocalEntities _localEntities{}; /** Local entities declared by the running program */