diff --git a/src/core/config/misc.h b/src/core/config/misc.h index 27a11e806..8909c4559 100644 --- a/src/core/config/misc.h +++ b/src/core/config/misc.h @@ -421,6 +421,68 @@ #define OPENTHREAD_CONFIG_NUM_FRAGMENT_PRIORITY_ENTRIES 8 #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 * diff --git a/src/core/thread/discover_scanner.cpp b/src/core/thread/discover_scanner.cpp index ff0b6f899..bf6d60f44 100644 --- a/src/core/thread/discover_scanner.cpp +++ b/src/core/thread/discover_scanner.cpp @@ -213,6 +213,7 @@ void DiscoverScanner::HandleDiscoveryRequestFrameTxDone(Message &aMessage) // the next scan channel. Also pause message tx on `MeshForwarder` // while listening to receive Discovery Responses. aMessage.SetDirectTransmission(); + aMessage.SetTimestampToNow(); Get().PauseMessageTransmissions(); mTimer.Start(kDefaultScanDuration); break; diff --git a/src/core/thread/lowpan.cpp b/src/core/thread/lowpan.cpp index b683c46f4..3aca0f8d5 100644 --- a/src/core/thread/lowpan.cpp +++ b/src/core/thread/lowpan.cpp @@ -1179,6 +1179,46 @@ exit: return (error == kErrorNone) ? static_cast(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((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(Ip6::kEcnMarked << kEcnOffset); + + aMessage.Write(aOffset, byte); +} + //--------------------------------------------------------------------------------------------------------------------- // MeshHeader diff --git a/src/core/thread/lowpan.hpp b/src/core/thread/lowpan.hpp index ae42d1e80..c0edcca38 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 @@ public: */ 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 @@ private: 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 f6e07c84d..be6c7d7e5 100644 --- a/src/core/thread/mesh_forwarder.cpp +++ b/src/core/thread/mesh_forwarder.cpp @@ -260,6 +260,184 @@ void MeshForwarder::HandleTxDelayTimer(void) } #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().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().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().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().ScheduleTransmissionTask(); @@ -294,12 +472,24 @@ Message *MeshForwarder::PrepareNextDirectTransmission(void) 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 @@ const char *MeshForwarder::MessageActionToString(MessageAction aAction, Error aE "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 @@ const char *MeshForwarder::MessageActionToString(MessageAction aAction, Error aE 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 @@ void MeshForwarder::LogMessage(MessageAction aAction, 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 88e4fa5de..317a16640 100644 --- a/src/core/thread/mesh_forwarder.hpp +++ b/src/core/thread/mesh_forwarder.hpp @@ -335,14 +335,23 @@ private: 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 @@ private: 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(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 @@ private: 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 @@ private: 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 d7c43aef0..ab9cdbcde 100644 --- a/src/core/thread/mesh_forwarder_ftd.cpp +++ b/src/core/thread/mesh_forwarder_ftd.cpp @@ -53,6 +53,7 @@ Error MeshForwarder::SendMessage(Message &aMessage) aMessage.SetOffset(0); aMessage.SetDatagramTag(0); + aMessage.SetTimestampToNow(); mSendQueue.Enqueue(aMessage); switch (aMessage.GetType()) @@ -199,6 +200,11 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) 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 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) } exit: - - if (error == kErrorNone) + if ((error == kErrorNone) && (evict != nullptr)) { RemoveMessage(*evict); } @@ -881,11 +886,18 @@ void MeshForwarder::UpdateFragmentPriority(Lowpan::FragmentHeader &aFragmentHead 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 @@ MeshForwarder::FragmentPriorityList::Entry *MeshForwarder::FragmentPriorityList: { 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 76db595f9..d74fba23f 100644 --- a/src/core/thread/mesh_forwarder_mtd.cpp +++ b/src/core/thread/mesh_forwarder_mtd.cpp @@ -42,6 +42,7 @@ Error MeshForwarder::SendMessage(Message &aMessage) aMessage.SetDirectTransmission(); aMessage.SetOffset(0); aMessage.SetDatagramTag(0); + aMessage.SetTimestampToNow(); mSendQueue.Enqueue(aMessage); mScheduleTransmissionTask.Post(); @@ -54,6 +55,11 @@ Error MeshForwarder::EvictMessage(Message::Priority aPriority) 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(aPriority)) diff --git a/tests/toranj/openthread-core-toranj-config.h b/tests/toranj/openthread-core-toranj-config.h index 7a180cc3e..508f60553 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 209fcdc1a..355807498 100644 --- a/tests/unit/test_lowpan.cpp +++ b/tests/unit/test_lowpan.cpp @@ -170,6 +170,8 @@ static void Test(TestIphcVector &aVector, bool aCompress, bool aDecompress) if (aCompress) { Lowpan::BufferWriter buffer(result, 127); + Message * compressedMsg; + Ip6::Ecn ecn; VerifyOrQuit((message = sInstance->Get().Allocate(Message::kTypeIp6)) != nullptr); @@ -192,6 +194,25 @@ static void Test(TestIphcVector &aVector, bool aCompress, bool aDecompress) 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().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 43d1c967f..b06fb07cb 100644 --- a/tests/unit/test_lowpan.hpp +++ b/tests/unit/test_lowpan.hpp @@ -99,6 +99,14 @@ public: */ 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. *