zephyr
/
openthread
mirror of https://github.com/zephyrproject-rtos/openthread.git
1
0
Fork 0
Code Issues Releases Wiki Activity

[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.
This commit is contained in:
Abtin Keshavarzian 2022-07-08 13:02:53 -07:00 committed by GitHub
parent 8aca0655e7
commit 2ce3d3bf02
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 438 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
src/core/config/misc.h
View File

@ -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
*

1
src/core/thread/discover_scanner.cpp
View File

@ -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<MeshForwarder>().PauseMessageTransmissions();
mTimer.Start(kDefaultScanDuration);
break;

40
src/core/thread/lowpan.cpp
View File

@ -1179,6 +1179,46 @@ exit:
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

27
src/core/thread/lowpan.hpp
View File

@ -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;

206
src/core/thread/mesh_forwarder.cpp
View File

@ -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<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 @@ 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;
}

53
src/core/thread/mesh_forwarder.hpp
View File

@ -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<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 @@ 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);

17
src/core/thread/mesh_forwarder_ftd.cpp
View File

@ -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;

6
src/core/thread/mesh_forwarder_mtd.cpp
View File

@ -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<uint8_t>(aPriority))

8
tests/toranj/openthread-core-toranj-config.h
View File

@ -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

21
tests/unit/test_lowpan.cpp
View File

@ -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<MessagePool>().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<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();

8
tests/unit/test_lowpan.hpp
View File

@ -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.
*