stateMachine: wait until commands need checking, instead of polling every 5 msec

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)

src/stateMachine/commandStateMachine.cpp

@@ -265,6 +265,40 @@ void CommandStateMachine::checkInflightCommandsTimeoutExpiracy() noexcept
 	}
 }
 
+std::chrono::time_point<std::chrono::steady_clock> CommandStateMachine::getNextCheckTime() noexcept
+{
+	// Lock
+	auto const lg = std::scoped_lock{ *_manager };
+
+	auto nextCheckTime = std::chrono::time_point<std::chrono::steady_clock>::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<std::chrono::steady_clock> const time) noexcept
+{
+	_manager->scheduleStateMachinesCheck(time);
+}
+
 void CommandStateMachine::handleAecpResponse(Aecpdu const& aecpdu) noexcept
 {
 	// Lock

src/stateMachine/commandStateMachine.hpp

@@ -28,6 +28,7 @@
 
 #include "protocolInterfaceDelegate.hpp"
 
+#include <algorithm>
 #include <chrono>
 #include <unordered_map>
 
@@ -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<std::chrono::steady_clock> 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<typename InflightInfo, typename CommandsQueue, typename Target>
+	std::chrono::time_point<std::chrono::steady_clock> 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<std::chrono::steady_clock>::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<std::chrono::steady_clock> const time) noexcept;
 
 	// Private members
 	Manager* _manager{ nullptr };

src/stateMachine/stateMachineManager.cpp

@@ -28,6 +28,8 @@
 #include "stateMachineManager.hpp"
 #include "logHelper.hpp"
 
+#include <algorithm>
+
 // 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)
@@ -223,7 +225,7 @@
 
 		// Create the state machine thread
 		_stateMachineThread = std::thread(
 			[this]
 			{
 				utils::setCurrentThreadName("avdecc::StateMachine");
 
@@ -231,6 +233,10 @@
 				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 @@
 					// 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 @@
 	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::scheduleStateMachinesCheck(std::chrono::time_point<std::chrono::steady_clock> 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");

src/stateMachine/stateMachineManager.hpp

@@ -34,6 +34,7 @@
 #include <chrono>
 #include <unordered_map>
 #include <mutex>
+#include <condition_variable>
 #include <thread>
 #include <cstdint>
 
@@ -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<std::chrono::steady_clock> 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<std::chrono::steady_clock> _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 */