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 */