[mesh-forwarder] add delay-aware queue management (#7568)
This commit implements delay-aware queue management. When enabled the
device will monitor time-in-queue of messages in the direct tx queue
and if it is lager than specified thresholds it updates ECN flag
(if message indicates it is ECN-capable) and/or drop the message. This
mechanism is applied to IPv6 messages on the first device that sends
the message into Thread mesh and also on intermediate routers that are
forwarding the message (e.g., as a "mesh lowpan fragment" frame). On an
intermediate router when forwarding the fragments of a message, if any
fragment is dropped by the queue management policy, all subsequent
fragments will also be dropped.
In particular, this commit contains the following:
- Adds `DecompressEcn()` and `MarkCompressedEcn()` in `Lowpan` class
to decompress or update the ECN field in a compressed IPHC header
(unit test `test_lowpan` is also updated to test the new methods).
- Adds `UpdateEcnOrDrop()` which implements the main queue management
logic. This method is used when preparing next direct tx message. It
decides whether to keep the message as is, update ECN on it or drop
it.
- Updates `EvictMessage()` to first apply the queue management rule
to see if any message can be dropped before using the eviction
logic based on message priority.
- Updates and reuses the `FragmentPriorityList` to track whether
queue management dropped any of the fragments of same message so
to also drop any subsequent ones.
- Updates `LogMessage()` to log when a message is dropped by
queue-management or when ECN is marked on a message.
diff --git a/src/core/config/misc.h b/src/core/config/misc.h
index 27a11e8..8909c45 100644
--- a/src/core/config/misc.h
+++ b/src/core/config/misc.h
@@ -422,6 +422,68 @@
#endif
/**
+ * @def OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ *
+ * Define to 1 to enable delay-aware queue management for the send queue.
+ *
+ * When enabled device will monitor time-in-queue of messages in the direct tx queue and if the wait time is lager than
+ * specified thresholds it may update ECN flag (if message indicates it is ECN-capable) or drop the message.
+ *
+ */
+#ifndef OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE \
+ (OPENTHREAD_CONFIG_THREAD_VERSION >= OT_THREAD_VERSION_1_3)
+#endif
+
+/**
+ * @OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_MARK_ECN_INTERVAL
+ *
+ * Specifies the time-in-queue threshold interval in milliseconds to mark ECN on a message if it is ECN-capable or
+ * drop the message if not ECN-capable.
+ *
+ */
+#ifndef OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_MARK_ECN_INTERVAL
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_MARK_ECN_INTERVAL 500
+#endif
+
+/**
+ * @OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_DROP_MSG_INTERVAL
+ *
+ * Specifies the time-in-queue threshold interval in milliseconds to drop a message.
+ *
+ */
+#ifndef OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_DROP_MSG_INTERVAL
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_DROP_MSG_INTERVAL 1000
+#endif
+
+/**
+ * OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_RETAIN_TIME
+ *
+ * Specifies the max retain time in seconds of a mesh header fragmentation tag entry in the list.
+ *
+ * The entry in list is used to track whether an earlier fragment of same message was dropped by the router and if so
+ * the next fragments are also dropped. The entry is removed once last fragment is processed or after the retain time
+ * specified by this config parameter expires.
+ *
+ */
+#ifndef OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_RETAIN_TIME
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_RETAIN_TIME (4 * 60) // 4 minutes
+#endif
+
+/**
+ * OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_ENTRY_LIST_SIZE
+ *
+ * Specifies the number of mesh header fragmentation tag entries in the list for delay-aware queue management.
+ *
+ * The list is used to track whether an earlier fragment of same message was dropped by the router and if so the next
+ * fragments are also dropped.
+ *
+ */
+#ifndef OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_ENTRY_LIST_SIZE
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_ENTRY_LIST_SIZE 16
+#endif
+
+/**
* @def OPENTHREAD_CONFIG_PLATFORM_RADIO_PROPRIETARY_SUPPORT
*
* Define to 1 to support proprietary radio configurations defined by platform.
diff --git a/src/core/thread/discover_scanner.cpp b/src/core/thread/discover_scanner.cpp
index ff0b6f8..bf6d60f 100644
--- a/src/core/thread/discover_scanner.cpp
+++ b/src/core/thread/discover_scanner.cpp
@@ -213,6 +213,7 @@
// the next scan channel. Also pause message tx on `MeshForwarder`
// while listening to receive Discovery Responses.
aMessage.SetDirectTransmission();
+ aMessage.SetTimestampToNow();
Get<MeshForwarder>().PauseMessageTransmissions();
mTimer.Start(kDefaultScanDuration);
break;
diff --git a/src/core/thread/lowpan.cpp b/src/core/thread/lowpan.cpp
index b683c46..3aca0f8 100644
--- a/src/core/thread/lowpan.cpp
+++ b/src/core/thread/lowpan.cpp
@@ -1179,6 +1179,46 @@
return (error == kErrorNone) ? static_cast<int>(compressedLength) : -1;
}
+Ip6::Ecn Lowpan::DecompressEcn(const Message &aMessage, uint16_t aOffset) const
+{
+ Ip6::Ecn ecn = Ip6::kEcnNotCapable;
+ uint16_t hcCtl;
+ uint8_t byte;
+
+ SuccessOrExit(aMessage.Read(aOffset, hcCtl));
+ hcCtl = HostSwap16(hcCtl);
+
+ VerifyOrExit((hcCtl & kHcDispatchMask) == kHcDispatch);
+ aOffset += sizeof(uint16_t);
+
+ if ((hcCtl & kHcTrafficFlowMask) == kHcTrafficFlow)
+ {
+ // ECN is elided and is zero (`kEcnNotCapable`).
+ ExitNow();
+ }
+
+ // When ECN is not elided, it is always included as the
+ // first two bits of the next byte.
+ SuccessOrExit(aMessage.Read(aOffset, byte));
+ ecn = static_cast<Ip6::Ecn>((byte & kEcnMask) >> kEcnOffset);
+
+exit:
+ return ecn;
+}
+
+void Lowpan::MarkCompressedEcn(Message &aMessage, uint16_t aOffset)
+{
+ uint8_t byte;
+
+ aOffset += sizeof(uint16_t);
+ IgnoreError(aMessage.Read(aOffset, byte));
+
+ byte &= ~kEcnMask;
+ byte |= static_cast<uint8_t>(Ip6::kEcnMarked << kEcnOffset);
+
+ aMessage.Write(aOffset, byte);
+}
+
//---------------------------------------------------------------------------------------------------------------------
// MeshHeader
diff --git a/src/core/thread/lowpan.hpp b/src/core/thread/lowpan.hpp
index ae42d1e..c0edcca 100644
--- a/src/core/thread/lowpan.hpp
+++ b/src/core/thread/lowpan.hpp
@@ -43,6 +43,7 @@
#include "mac/mac_types.hpp"
#include "net/ip6.hpp"
#include "net/ip6_address.hpp"
+#include "net/ip6_types.hpp"
namespace ot {
@@ -308,6 +309,29 @@
*/
int DecompressUdpHeader(Ip6::Udp::Header &aUdpHeader, const uint8_t *aBuf, uint16_t aBufLength);
+ /**
+ * This method decompresses the IPv6 ECN field in a LOWPAN_IPHC header.
+ *
+ * @param[in] aMessage The message to read the IPHC header from.
+ * @param[in] aOffset The offset in @p aMessage to start of IPHC header.
+ *
+ * @returns The decompressed ECN field. If the IPHC header is not valid `kEcnNotCapable` is returned.
+ *
+ */
+ Ip6::Ecn DecompressEcn(const Message &aMessage, uint16_t aOffset) const;
+
+ /**
+ * This method updates the compressed ECN field in a LOWPAN_IPHC header to `kEcnMarked`.
+ *
+ * This method MUST be used when the ECN field is not elided in the IPHC header. Note that the ECN is not elided
+ * when it is not zero (`kEcnNotCapable`).
+ *
+ * @param[in,out] aMessage The message containing the IPHC header and to update.
+ * @param[in] aOffset The offset in @p aMessage to start of IPHC header.
+ *
+ */
+ void MarkCompressedEcn(Message &aMessage, uint16_t aOffset);
+
private:
static constexpr uint16_t kHcDispatch = 3 << 13;
static constexpr uint16_t kHcDispatchMask = 7 << 13;
@@ -336,6 +360,9 @@
static constexpr uint16_t kHcDstAddrMode3 = 3 << 0;
static constexpr uint16_t kHcDstAddrModeMask = 3 << 0;
+ static constexpr uint8_t kEcnOffset = 6;
+ static constexpr uint8_t kEcnMask = 3 << kEcnOffset;
+
static constexpr uint8_t kExtHdrDispatch = 0xe0;
static constexpr uint8_t kExtHdrDispatchMask = 0xf0;
diff --git a/src/core/thread/mesh_forwarder.cpp b/src/core/thread/mesh_forwarder.cpp
index f6e07c8..be6c7d7 100644
--- a/src/core/thread/mesh_forwarder.cpp
+++ b/src/core/thread/mesh_forwarder.cpp
@@ -260,6 +260,184 @@
}
#endif
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+
+Error MeshForwarder::UpdateEcnOrDrop(Message &aMessage, bool aPreparingToSend)
+{
+ // This method performs delay-aware active queue management for
+ // direct message transmission. It parses the IPv6 header from
+ // `aMessage` to determine if message is ECN-capable. This is
+ // then used along with the message's time-in-queue to decide
+ // whether to keep the message as is, change the ECN field to
+ // mark congestion, or drop the message. If the message is to be
+ // dropped, this method clears the direct tx flag on `aMessage`
+ // and removes it from the send queue (if no pending indirect tx)
+ // and returns `kErrorDrop`. This method returns `kErrorNone`
+ // when the message is kept as is or ECN field is updated.
+
+ Error error = kErrorNone;
+ uint32_t timeInQueue = TimerMilli::GetNow() - aMessage.GetTimestamp();
+ bool shouldMarkEcn = (timeInQueue >= kTimeInQueueMarkEcn);
+ bool isEcnCapable = false;
+
+ VerifyOrExit(aMessage.IsDirectTransmission() && (aMessage.GetOffset() == 0));
+
+ if (aMessage.GetType() == Message::kTypeIp6)
+ {
+ Ip6::Header ip6Header;
+
+ IgnoreError(aMessage.Read(0, ip6Header));
+
+ VerifyOrExit(!Get<ThreadNetif>().HasUnicastAddress(ip6Header.GetSource()));
+
+ isEcnCapable = (ip6Header.GetEcn() != Ip6::kEcnNotCapable);
+
+ if ((shouldMarkEcn && !isEcnCapable) || (timeInQueue >= kTimeInQueueDropMsg))
+ {
+ ExitNow(error = kErrorDrop);
+ }
+
+ if (shouldMarkEcn)
+ {
+ switch (ip6Header.GetEcn())
+ {
+ case Ip6::kEcnCapable0:
+ case Ip6::kEcnCapable1:
+ ip6Header.SetEcn(Ip6::kEcnMarked);
+ aMessage.Write(0, ip6Header);
+ LogMessage(kMessageMarkEcn, aMessage);
+ break;
+
+ case Ip6::kEcnMarked:
+ case Ip6::kEcnNotCapable:
+ break;
+ }
+ }
+ }
+#if OPENTHREAD_FTD
+ else if (aMessage.GetType() == Message::kType6lowpan)
+ {
+ uint16_t headerLength = 0;
+ uint16_t offset;
+ bool hasFragmentHeader = false;
+ Lowpan::FragmentHeader fragmentHeader;
+ Lowpan::MeshHeader meshHeader;
+
+ IgnoreError(meshHeader.ParseFrom(aMessage, headerLength));
+
+ offset = headerLength;
+
+ if (fragmentHeader.ParseFrom(aMessage, offset, headerLength) == kErrorNone)
+ {
+ hasFragmentHeader = true;
+ offset += headerLength;
+ }
+
+ if (!hasFragmentHeader || (fragmentHeader.GetDatagramOffset() == 0))
+ {
+ Ip6::Ecn ecn = Get<Lowpan::Lowpan>().DecompressEcn(aMessage, offset);
+
+ isEcnCapable = (ecn != Ip6::kEcnNotCapable);
+
+ if ((shouldMarkEcn && !isEcnCapable) || (timeInQueue >= kTimeInQueueDropMsg))
+ {
+ FragmentPriorityList::Entry *entry;
+
+ entry = mFragmentPriorityList.FindEntry(meshHeader.GetSource(), fragmentHeader.GetDatagramTag());
+
+ if (entry != nullptr)
+ {
+ entry->MarkToDrop();
+ entry->ResetLifetime();
+ }
+
+ ExitNow(error = kErrorDrop);
+ }
+
+ if (shouldMarkEcn)
+ {
+ switch (ecn)
+ {
+ case Ip6::kEcnCapable0:
+ case Ip6::kEcnCapable1:
+ Get<Lowpan::Lowpan>().MarkCompressedEcn(aMessage, offset);
+ LogMessage(kMessageMarkEcn, aMessage);
+ break;
+
+ case Ip6::kEcnMarked:
+ case Ip6::kEcnNotCapable:
+ break;
+ }
+ }
+ }
+ else if (hasFragmentHeader)
+ {
+ FragmentPriorityList::Entry *entry;
+
+ entry = mFragmentPriorityList.FindEntry(meshHeader.GetSource(), fragmentHeader.GetDatagramTag());
+ VerifyOrExit(entry != nullptr);
+
+ if (entry->ShouldDrop())
+ {
+ error = kErrorDrop;
+ }
+
+ // We can clear the entry if it is the last fragment and
+ // only if the message is being prepared to be sent out.
+ if (aPreparingToSend && (fragmentHeader.GetDatagramOffset() + aMessage.GetLength() - offset >=
+ fragmentHeader.GetDatagramSize()))
+ {
+ entry->Clear();
+ }
+ }
+ }
+#else
+ OT_UNUSED_VARIABLE(aPreparingToSend);
+#endif // OPENTHREAD_FTD
+
+exit:
+ if (error == kErrorDrop)
+ {
+ LogMessage(kMessageQueueMgmtDrop, aMessage);
+ aMessage.ClearDirectTransmission();
+ RemoveMessageIfNoPendingTx(aMessage);
+ }
+
+ return error;
+}
+
+Error MeshForwarder::RemoveAgedMessages(void)
+{
+ // This method goes through all messages in the send queue and
+ // removes all aged messages determined based on the delay-aware
+ // active queue management rules. It may also mark ECN on some
+ // messages. It returns `kErrorNone` if at least one message was
+ // removed, or `kErrorNotFound` if none was removed.
+
+ Error error = kErrorNotFound;
+ Message *nextMessage;
+
+ for (Message *message = mSendQueue.GetHead(); message != nullptr; message = nextMessage)
+ {
+ nextMessage = message->GetNext();
+
+ // Exclude the current message being sent `mSendMessage`.
+ if ((message == mSendMessage) || !message->IsDirectTransmission())
+ {
+ continue;
+ }
+
+ if (UpdateEcnOrDrop(*message, /* aPreparingToSend */ false) == kErrorDrop)
+ {
+ error = kErrorNone;
+ }
+ }
+
+ return error;
+}
+
+#endif // OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+
void MeshForwarder::ScheduleTransmissionTask(Tasklet &aTasklet)
{
aTasklet.Get<MeshForwarder>().ScheduleTransmissionTask();
@@ -294,12 +472,24 @@
for (curMessage = mSendQueue.GetHead(); curMessage; curMessage = nextMessage)
{
+ // We set the `nextMessage` here but it can be updated again
+ // after the `switch(message.GetType())` since it may be
+ // evicted during message processing (e.g., from the call to
+ // `UpdateIp6Route()` due to Address Solicit).
+
+ nextMessage = curMessage->GetNext();
+
if (!curMessage->IsDirectTransmission() || curMessage->IsResolvingAddress())
{
- nextMessage = curMessage->GetNext();
continue;
}
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ if (UpdateEcnOrDrop(*curMessage) == kErrorDrop)
+ {
+ continue;
+ }
+#endif
curMessage->SetDoNotEvict(true);
switch (curMessage->GetType())
@@ -1634,6 +1824,10 @@
"Dropping", // (3) kMessageDrop
"Dropping (reassembly queue)", // (4) kMessageReassemblyDrop
"Evicting", // (5) kMessageEvict
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ "Marked ECN", // (6) kMessageMarkEcn
+ "Dropping (queue mgmt)", // (7) kMessageQueueMgmtDrop
+#endif
};
const char *string = kMessageActionStrings[aAction];
@@ -1644,6 +1838,10 @@
static_assert(kMessageDrop == 3, "kMessageDrop value is incorrect");
static_assert(kMessageReassemblyDrop == 4, "kMessageReassemblyDrop value is incorrect");
static_assert(kMessageEvict == 5, "kMessageEvict value is incorrect");
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ static_assert(kMessageMarkEcn == 6, "kMessageMarkEcn is incorrect");
+ static_assert(kMessageQueueMgmtDrop == 7, "kMessageQueueMgmtDrop is incorrect");
+#endif
if ((aAction == kMessageTransmit) && (aError != kErrorNone))
{
@@ -1741,12 +1939,18 @@
case kMessageReceive:
case kMessageTransmit:
case kMessagePrepareIndirect:
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ case kMessageMarkEcn:
+#endif
logLevel = (aError == kErrorNone) ? kLogLevelInfo : kLogLevelNote;
break;
case kMessageDrop:
case kMessageReassemblyDrop:
case kMessageEvict:
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ case kMessageQueueMgmtDrop:
+#endif
logLevel = kLogLevelNote;
break;
}
diff --git a/src/core/thread/mesh_forwarder.hpp b/src/core/thread/mesh_forwarder.hpp
index 88e4fa5..317a166 100644
--- a/src/core/thread/mesh_forwarder.hpp
+++ b/src/core/thread/mesh_forwarder.hpp
@@ -335,14 +335,23 @@
static constexpr uint32_t kTxDelayInterval = OPENTHREAD_CONFIG_MAC_COLLISION_AVOIDANCE_DELAY_INTERVAL; // In msec
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ static constexpr uint32_t kTimeInQueueMarkEcn = OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_MARK_ECN_INTERVAL;
+ static constexpr uint32_t kTimeInQueueDropMsg = OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_DROP_MSG_INTERVAL;
+#endif
+
enum MessageAction : uint8_t
{
kMessageReceive, // Indicates that the message was received.
kMessageTransmit, // Indicates that the message was sent.
kMessagePrepareIndirect, // Indicates that the message is being prepared for indirect tx.
- kMessageDrop, // Indicates that the outbound message is being dropped (e.g., dst unknown).
+ 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.
+#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
};
enum AnycastType : uint8_t
@@ -361,21 +370,39 @@
friend class FragmentPriorityList;
public:
- Message::Priority GetPriority(void) const { return mPriority; }
+ // Lifetime of an entry in seconds.
+ static constexpr uint8_t kLifetime =
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ OT_MAX(kReassemblyTimeout, OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_RETAIN_TIME);
+#else
+ kReassemblyTimeout;
+#endif
+
+ Message::Priority GetPriority(void) const { return static_cast<Message::Priority>(mPriority); }
bool IsExpired(void) const { return (mLifetime == 0); }
void DecrementLifetime(void) { mLifetime--; }
- void ResetLifetime(void) { mLifetime = kReassemblyTimeout; }
+ void ResetLifetime(void) { mLifetime = kLifetime; }
bool Matches(uint16_t aSrcRloc16, uint16_t aTag) const
{
return (mSrcRloc16 == aSrcRloc16) && (mDatagramTag == aTag);
}
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ bool ShouldDrop(void) const { return mShouldDrop; }
+ void MarkToDrop(void) { mShouldDrop = true; }
+#endif
+
private:
- uint16_t mSrcRloc16;
- uint16_t mDatagramTag;
- Message::Priority mPriority;
- uint8_t mLifetime;
+ uint16_t mSrcRloc16;
+ uint16_t mDatagramTag;
+ uint8_t mLifetime;
+ uint8_t mPriority : 2;
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ bool mShouldDrop : 1;
+#endif
+
+ static_assert(Message::kNumPriorities <= 4, "mPriority as a 2-bit does not fit all `Priority` values");
};
Entry *AllocateEntry(uint16_t aSrcRloc16, uint16_t aTag, Message::Priority aPriority);
@@ -383,7 +410,13 @@
bool UpdateOnTimeTick(void);
private:
- static constexpr uint16_t kNumEntries = OPENTHREAD_CONFIG_NUM_FRAGMENT_PRIORITY_ENTRIES;
+ static constexpr uint16_t kNumEntries =
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ OT_MAX(OPENTHREAD_CONFIG_NUM_FRAGMENT_PRIORITY_ENTRIES,
+ OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_FRAG_TAG_ENTRY_LIST_SIZE);
+#else
+ OPENTHREAD_CONFIG_NUM_FRAGMENT_PRIORITY_ENTRIES;
+#endif
Entry mEntries[kNumEntries];
};
@@ -433,6 +466,10 @@
bool aAddFragHeader = false);
void PrepareEmptyFrame(Mac::TxFrame &aFrame, const Mac::Address &aMacDest, bool aAckRequest);
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ Error UpdateEcnOrDrop(Message &aMessage, bool aPreparingToSend = true);
+ Error RemoveAgedMessages(void);
+#endif
void SendMesh(Message &aMessage, Mac::TxFrame &aFrame);
void SendDestinationUnreachable(uint16_t aMeshSource, const Message &aMessage);
Error UpdateIp6Route(Message &aMessage);
diff --git a/src/core/thread/mesh_forwarder_ftd.cpp b/src/core/thread/mesh_forwarder_ftd.cpp
index d7c43ae..ab9cdbc 100644
--- a/src/core/thread/mesh_forwarder_ftd.cpp
+++ b/src/core/thread/mesh_forwarder_ftd.cpp
@@ -53,6 +53,7 @@
aMessage.SetOffset(0);
aMessage.SetDatagramTag(0);
+ aMessage.SetTimestampToNow();
mSendQueue.Enqueue(aMessage);
switch (aMessage.GetType())
@@ -199,6 +200,11 @@
Error error = kErrorNotFound;
Message *evict = nullptr;
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ error = RemoveAgedMessages();
+ VerifyOrExit(error == kErrorNotFound);
+#endif
+
// Search for a lower priority message to evict
for (uint8_t priority = 0; priority < aPriority; priority++)
{
@@ -245,8 +251,7 @@
}
exit:
-
- if (error == kErrorNone)
+ if ((error == kErrorNone) && (evict != nullptr))
{
RemoveMessage(*evict);
}
@@ -881,11 +886,18 @@
ExitNow();
}
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ OT_UNUSED_VARIABLE(aFragmentLength);
+#else
+ // We can clear the entry in `mFragmentPriorityList` if it is the
+ // last fragment. But if "delay aware active queue management" is
+ // used we need to keep entry until the message is sent.
if (aFragmentHeader.GetDatagramOffset() + aFragmentLength >= aFragmentHeader.GetDatagramSize())
{
entry->Clear();
}
else
+#endif
{
entry->ResetLifetime();
}
@@ -922,6 +934,7 @@
{
if (entry.IsExpired())
{
+ entry.Clear();
entry.mSrcRloc16 = aSrcRloc16;
entry.mDatagramTag = aTag;
entry.mPriority = aPriority;
diff --git a/src/core/thread/mesh_forwarder_mtd.cpp b/src/core/thread/mesh_forwarder_mtd.cpp
index 76db595..d74fba2 100644
--- a/src/core/thread/mesh_forwarder_mtd.cpp
+++ b/src/core/thread/mesh_forwarder_mtd.cpp
@@ -42,6 +42,7 @@
aMessage.SetDirectTransmission();
aMessage.SetOffset(0);
aMessage.SetDatagramTag(0);
+ aMessage.SetTimestampToNow();
mSendQueue.Enqueue(aMessage);
mScheduleTransmissionTask.Post();
@@ -54,6 +55,11 @@
Error error = kErrorNotFound;
Message *message;
+#if OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ error = RemoveAgedMessages();
+ VerifyOrExit(error == kErrorNotFound);
+#endif
+
VerifyOrExit((message = mSendQueue.GetTail()) != nullptr);
if (message->GetPriority() < static_cast<uint8_t>(aPriority))
diff --git a/tests/toranj/openthread-core-toranj-config.h b/tests/toranj/openthread-core-toranj-config.h
index 7a180cc..508f605 100644
--- a/tests/toranj/openthread-core-toranj-config.h
+++ b/tests/toranj/openthread-core-toranj-config.h
@@ -530,6 +530,14 @@
*/
#define OPENTHREAD_CONFIG_DNSSD_SERVER_ENABLE 1
+/**
+ * @def OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE
+ *
+ * Define to 1 to enable delay-aware queue management for the send queue.
+ *
+ */
+#define OPENTHREAD_CONFIG_DELAY_AWARE_QUEUE_MANAGEMENT_ENABLE 1
+
#if OPENTHREAD_RADIO
/**
* @def OPENTHREAD_CONFIG_MAC_SOFTWARE_ACK_TIMEOUT_ENABLE
diff --git a/tests/unit/test_lowpan.cpp b/tests/unit/test_lowpan.cpp
index 209fcdc..3558074 100644
--- a/tests/unit/test_lowpan.cpp
+++ b/tests/unit/test_lowpan.cpp
@@ -170,6 +170,8 @@
if (aCompress)
{
Lowpan::BufferWriter buffer(result, 127);
+ Message * compressedMsg;
+ Ip6::Ecn ecn;
VerifyOrQuit((message = sInstance->Get<MessagePool>().Allocate(Message::kTypeIp6)) != nullptr);
@@ -192,6 +194,25 @@
VerifyOrQuit(compressBytes == aVector.mIphcHeader.mLength, "Lowpan::Compress failed");
VerifyOrQuit(message->GetOffset() == aVector.mPayloadOffset, "Lowpan::Compress failed");
VerifyOrQuit(memcmp(iphc, result, iphcLength) == 0, "Lowpan::Compress failed");
+
+ // Validate `DecompressEcn()` and `MarkCompressedEcn()`
+
+ VerifyOrQuit((compressedMsg = sInstance->Get<MessagePool>().Allocate(Message::kTypeIp6)) != nullptr);
+ SuccessOrQuit(compressedMsg->AppendBytes(result, compressBytes));
+
+ ecn = sLowpan->DecompressEcn(*compressedMsg, /* aOffset */ 0);
+ VerifyOrQuit(ecn == aVector.GetIpHeader().GetEcn());
+ printf("Decompressed ECN is %d\n", ecn);
+
+ if (ecn != Ip6::kEcnNotCapable)
+ {
+ sLowpan->MarkCompressedEcn(*compressedMsg, /*a aOffset */ 0);
+ ecn = sLowpan->DecompressEcn(*compressedMsg, /* aOffset */ 0);
+ VerifyOrQuit(ecn == Ip6::kEcnMarked);
+ printf("ECN is updated to %d\n", ecn);
+ }
+
+ compressedMsg->Free();
}
message->Free();
diff --git a/tests/unit/test_lowpan.hpp b/tests/unit/test_lowpan.hpp
index 43d1c96..b06fb07 100644
--- a/tests/unit/test_lowpan.hpp
+++ b/tests/unit/test_lowpan.hpp
@@ -100,6 +100,14 @@
void SetMacDestination(uint16_t aAddress) { mMacDestination.SetShort(aAddress); }
/**
+ * This method gets the IPv6 header
+ *
+ * @returns the IPv6 header.
+ *
+ */
+ const Ip6::Header &GetIpHeader(void) const { return mIpHeader; }
+
+ /**
* This method initializes IPv6 Header.
*
* @param aVersionClassFlow Value of the Version, Traffic class and Flow control fields.
