From 13277e803dafdb003010c52c06b5b4169905f7cb Mon Sep 17 00:00:00 2001 From: Abtin Keshavarzian Date: Mon, 26 Jan 2026 10:35:39 -0800 Subject: [PATCH] [mesh-forwarder] evict lower priority msg on direct tx queue limit (#12320) This commit updates `ApplyDirectTxQueueLimit()` to improve how the direct tx queue handles frame limits. Previously, when the direct tx queue reached its configured frame count threshold, the code attempted to remove aged messages. If this was insufficient, the new incoming message was dropped. With this change, after attempting to remove aged messages, the code now attempts to evict an existing lower-priority message (and all its frames) from the direct tx queue to make room for a new higher priority message. If no existing message can be evicted (i.e., all messages are of higher or equal priority), the new message is dropped. --- src/core/common/message.cpp | 5 +++- src/core/thread/mesh_forwarder.cpp | 33 ++++++++++++-------------- src/core/thread/mesh_forwarder.hpp | 25 +++++++++---------- src/core/thread/mesh_forwarder_ftd.cpp | 20 ++++++++++++++-- src/core/thread/mesh_forwarder_mtd.cpp | 6 ++++- 5 files changed, 53 insertions(+), 36 deletions(-) diff --git a/src/core/common/message.cpp b/src/core/common/message.cpp index eb8b7334a..2c18c4f7a 100644 --- a/src/core/common/message.cpp +++ b/src/core/common/message.cpp @@ -157,7 +157,10 @@ void MessagePool::FreeBuffers(Buffer *aBuffer) } } -Error MessagePool::ReclaimBuffers(Message::Priority aPriority) { return Get().EvictMessage(aPriority); } +Error MessagePool::ReclaimBuffers(Message::Priority aPriority) +{ + return Get().EvictMessage(aPriority, MeshForwarder::kEvictReasonNoMessageBuffer); +} uint16_t MessagePool::GetFreeBufferCount(void) const { diff --git a/src/core/thread/mesh_forwarder.cpp b/src/core/thread/mesh_forwarder.cpp index 5e53b2350..20cb111d8 100644 --- a/src/core/thread/mesh_forwarder.cpp +++ b/src/core/thread/mesh_forwarder.cpp @@ -390,14 +390,16 @@ bool MeshForwarder::IsDirectTxQueueOverMaxFrameThreshold(void) const void MeshForwarder::ApplyDirectTxQueueLimit(Message &aMessage) { + Error error; + bool originalEvictFlag; + VerifyOrExit(aMessage.IsDirectTransmission()); VerifyOrExit(IsDirectTxQueueOverMaxFrameThreshold()); -#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE - { - bool originalEvictFlag = aMessage.GetDoNotEvict(); - Error error; + originalEvictFlag = aMessage.GetDoNotEvict(); + do + { // We mark the "do not evict" flag on the new `aMessage` so // that it will not be removed from `RemoveAgedMessages()`. // This protects against the unlikely case where the newly @@ -408,15 +410,15 @@ void MeshForwarder::ApplyDirectTxQueueLimit(Message &aMessage) // freed twice. aMessage.SetDoNotEvict(true); - error = RemoveAgedMessages(); + error = EvictMessage(aMessage.GetPriority(), kEvictReasonDirectTxQueueAtLimit); aMessage.SetDoNotEvict(originalEvictFlag); if (error == kErrorNone) { VerifyOrExit(IsDirectTxQueueOverMaxFrameThreshold()); } - } -#endif + + } while (error == kErrorNone); LogMessage(kMessageFullQueueDrop, aMessage); FinalizeMessageDirectTx(aMessage, kErrorDrop); @@ -1369,8 +1371,10 @@ const char *MeshForwarder::MessageActionToString(MessageAction aAction, Error aE _(kMessagePrepareIndirect, "Prepping indir tx") \ _(kMessageDrop, "Dropping") \ _(kMessageReassemblyDrop, "Dropping (reassembly queue)") \ - _(kMessageEvict, "Evicting") \ - QueueMgmntMessageActionMapList(_) DropQueueFullMessageActionMapList(_) + _(kMessageEvict, "Evicting (no msg buff)") \ + _(kMessageFullQueueEvict, "Evicting (dir queue full)") \ + _(kMessageFullQueueDrop, "Dropping (dir queue full)") \ + QueueMgmntMessageActionMapList(_) #if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE #define QueueMgmntMessageActionMapList(_) \ @@ -1378,12 +1382,6 @@ const char *MeshForwarder::MessageActionToString(MessageAction aAction, Error aE _(kMessageQueueMgmtDrop, "Dropping (queue mgmt)") #else #define QueueMgmntMessageActionMapList(_) -#endif - -#if (OPENTHREAD_CONFIG_MAX_FRAMES_IN_DIRECT_TX_QUEUE > 0) -#define DropQueueFullMessageActionMapList(_) _(kMessageFullQueueDrop, "Dropping (dir queue full)") -#else -#define DropQueueFullMessageActionMapList(_) #endif DefineEnumStringArray(MessageActionMapList); @@ -1539,12 +1537,11 @@ void MeshForwarder::LogMessage(MessageAction aAction, case kMessageDrop: case kMessageReassemblyDrop: + case kMessageFullQueueDrop: case kMessageEvict: + case kMessageFullQueueEvict: #if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE case kMessageQueueMgmtDrop: -#endif -#if (OPENTHREAD_CONFIG_MAX_FRAMES_IN_DIRECT_TX_QUEUE > 0) - case kMessageFullQueueDrop: #endif // default kLogLevelInfo for dropped message break; diff --git a/src/core/thread/mesh_forwarder.hpp b/src/core/thread/mesh_forwarder.hpp index d4ed6514d..70db267ac 100644 --- a/src/core/thread/mesh_forwarder.hpp +++ b/src/core/thread/mesh_forwarder.hpp @@ -84,6 +84,7 @@ class MeshForwarder : public InstanceLocator, private NonCopyable friend class Ip6::Ip6; friend class Mle::DiscoverScanner; friend class TimeTicker; + friend class ot::MessagePool; public: /** @@ -179,16 +180,6 @@ public: */ void RemoveDataResponseMessages(void); - /** - * Evicts the message with lowest priority in the send queue. - * - * @param[in] aPriority The highest priority level of the evicted message. - * - * @retval kErrorNone Successfully evicted a low priority message. - * @retval kErrorNotFound No low priority messages available to evict. - */ - Error EvictMessage(Message::Priority aPriority); - /** * Retrieves information about the send queue and the reassembly queue. * @@ -299,6 +290,12 @@ private: static constexpr uint32_t kTimeInQueueDropMsg = OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_DROP_MSG_INTERVAL; #endif + enum EvictReason : uint8_t // Used in EvictMessage() + { + kEvictReasonNoMessageBuffer, + kEvictReasonDirectTxQueueAtLimit, + }; + enum MessageAction : uint8_t { kMessageReceive, // Indicates that the message was received. @@ -306,13 +303,12 @@ private: kMessagePrepareIndirect, // Indicates that the message is being prepared for indirect tx. kMessageDrop, // Indicates that the outbound message is dropped (e.g., dst unknown). kMessageReassemblyDrop, // Indicates that the message is being dropped from reassembly list. - kMessageEvict, // Indicates that the message was evicted. + kMessageEvict, // Indicates that the message was evicted due to no available message buffers. + kMessageFullQueueEvict, // Indicates that a lower priority message eviction due to direct tx queue at limit. + kMessageFullQueueDrop, // Indicates message drop due to direct tx queue at limit. #if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE kMessageMarkEcn, // Indicates that ECN is marked on an outbound message by delay-aware queue management. kMessageQueueMgmtDrop, // Indicates that an outbound message is dropped by delay-aware queue management. -#endif -#if (OPENTHREAD_CONFIG_MAX_FRAMES_IN_DIRECT_TX_QUEUE > 0) - kMessageFullQueueDrop, // Indicates message drop due to reaching max allowed frames in direct tx queue. #endif }; @@ -475,6 +471,7 @@ private: void HandleTimeTick(void); void ScheduleTransmissionTask(void); + Error EvictMessage(Message::Priority aPriority, EvictReason aEvictReason); Error GetFramePriority(RxInfo &aRxInfo, Message::Priority &aPriority); #if OPENTHREAD_FTD diff --git a/src/core/thread/mesh_forwarder_ftd.cpp b/src/core/thread/mesh_forwarder_ftd.cpp index 77073e84c..874e3c5b6 100644 --- a/src/core/thread/mesh_forwarder_ftd.cpp +++ b/src/core/thread/mesh_forwarder_ftd.cpp @@ -196,7 +196,7 @@ void MeshForwarder::HandleResolved(const Ip6::Address &aEid, Error aError) } } -Error MeshForwarder::EvictMessage(Message::Priority aPriority) +Error MeshForwarder::EvictMessage(Message::Priority aPriority, EvictReason aEvictReason) { Error error = kErrorNotFound; Message *evict = nullptr; @@ -222,12 +222,19 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) continue; } + if ((aEvictReason == kEvictReasonDirectTxQueueAtLimit) && !message->IsDirectTransmission()) + { + continue; + } + evict = message; error = kErrorNone; ExitNow(); } } + VerifyOrExit(aEvictReason == kEvictReasonNoMessageBuffer); + for (uint8_t priority = aPriority; priority < Message::kNumPriorities; priority++) { // search for an equal or higher priority indirect message to evict @@ -254,7 +261,16 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) exit: if ((error == kErrorNone) && (evict != nullptr)) { - FinalizeAndRemoveMessage(*evict, kErrorNoBufs, kMessageEvict); + switch (aEvictReason) + { + case kEvictReasonDirectTxQueueAtLimit: + FinalizeAndRemoveMessage(*evict, kErrorDrop, kMessageFullQueueEvict); + break; + + case kEvictReasonNoMessageBuffer: + FinalizeAndRemoveMessage(*evict, kErrorNoBufs, kMessageEvict); + break; + } } return error; diff --git a/src/core/thread/mesh_forwarder_mtd.cpp b/src/core/thread/mesh_forwarder_mtd.cpp index 9dec26b43..df47e56a7 100644 --- a/src/core/thread/mesh_forwarder_mtd.cpp +++ b/src/core/thread/mesh_forwarder_mtd.cpp @@ -54,8 +54,10 @@ void MeshForwarder::SendMessage(OwnedPtr aMessagePtr) #endif } -Error MeshForwarder::EvictMessage(Message::Priority aPriority) +Error MeshForwarder::EvictMessage(Message::Priority aPriority, EvictReason aEvictReason) { + OT_UNUSED_VARIABLE(aEvictReason); + Error error = kErrorNotFound; Message *message; @@ -66,6 +68,8 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) VerifyOrExit((message = mSendQueue.GetTail()) != nullptr); + VerifyOrExit(!message->GetDoNotEvict()); + if (message->GetPriority() < static_cast(aPriority)) { FinalizeAndRemoveMessage(*message, kErrorNoBufs, kMessageEvict);