[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

@@ -157,7 +157,10 @@ void MessagePool::FreeBuffers(Buffer *aBuffer)
     }
 }
 
-Error MessagePool::ReclaimBuffers(Message::Priority aPriority) { return Get<MeshForwarder>().EvictMessage(aPriority); }
+Error MessagePool::ReclaimBuffers(Message::Priority aPriority)
+{
+    return Get<MeshForwarder>().EvictMessage(aPriority, MeshForwarder::kEvictReasonNoMessageBuffer);
+}
 
 uint16_t MessagePool::GetFreeBufferCount(void) const
 {

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;

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

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;

src/core/thread/mesh_forwarder_mtd.cpp

@@ -54,8 +54,10 @@ void MeshForwarder::SendMessage(OwnedPtr<Message> 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<uint8_t>(aPriority))
     {
         FinalizeAndRemoveMessage(*message, kErrorNoBufs, kMessageEvict);