Skip to content

Commit

Permalink
[mle] update DelayedSender to keep the tx schedule (#10846)
Browse files Browse the repository at this point in the history 
This commit updates how `Mle::DelayedSender` sends MLE messages after
a delay.

Previously, a delayed message was fully prepared and placed in a
queue, waiting for its delay duration to expire. In the new model, a
delayed message is prepared when the requested delay time expires and
the message transmission time is reached. The message is constructed
right before transmission. This ensures that the delayed message
includes the most up-to-date information and simplifies the methods
that prepare and send different types of MLE messages.

`DelayedSender` now keeps track of the message type and the
specific information required to construct the message later. For
example, for a delayed Parent Response to a Parent Request, the
extended address and the `RxChallenge` of the child are saved.

This new model makes it easier to implement the desired behavior when
similar messages are already scheduled to the same destination. For
example:
-   For Data Request, if one is already scheduled, a new request can
    be skipped, as the earlier Data Request will be valid.
-   For Parent Response, Link Accept, and Data Response, a new
    schedule request replaces an earlier one.
-   For Discovery Response, multiple schedules are allowed.
  • Loading branch information
@abtink
abtink authored Nov 5, 2024
1 parent e6a6b9f commit 7169561
Show file tree
Hide file tree
Showing 5 changed files with 372 additions and 217 deletions.
251 changes: 172 additions & 79 deletions src/core/thread/mle.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,7 @@ void Mle::Stop(StopMode aMode)

VerifyOrExit(!IsDisabled());

mDelayedSender.Stop();
Get<KeyManager>().Stop();
SetStateDetached();
Get<ThreadNetif>().UnsubscribeMulticast(mRealmLocalAllThreadNodes);
Expand Down Expand Up @@ -1737,22 +1738,17 @@ Error Mle::SendChildIdRequest(void)
}

Error Mle::SendDataRequest(const Ip6::Address &aDestination)
{
return SendDataRequestAfterDelay(aDestination, /* aDelay */ 0);
}

Error Mle::SendDataRequestAfterDelay(const Ip6::Address &aDestination, uint16_t aDelay)
{
static const uint8_t kTlvs[] = {Tlv::kNetworkData, Tlv::kRoute};

Error error;
Error error = kErrorNone;

mDelayedSender.RemoveDataRequestMessage(aDestination);
VerifyOrExit(IsAttached());

// Based on `mRequestRouteTlv` include both Network Data and Route
// TLVs or only Network Data TLV.

error = SendDataRequest(aDestination, kTlvs, mRequestRouteTlv ? 2 : 1, aDelay);
error = SendDataRequest(aDestination, kTlvs, mRequestRouteTlv ? 2 : 1);

if (IsChild() && !IsRxOnWhenIdle())
{
Expand All @@ -1764,6 +1760,7 @@ Error Mle::SendDataRequestAfterDelay(const Ip6::Address &aDestination, uint16_t
}
}

exit:
return error;
}

Expand All @@ -1773,16 +1770,15 @@ Error Mle::SendDataRequestForLinkMetricsReport(const Ip6::Address
{
static const uint8_t kTlvs[] = {Tlv::kLinkMetricsReport};

return SendDataRequest(aDestination, kTlvs, sizeof(kTlvs), /* aDelay */ 0, &aQueryInfo);
return SendDataRequest(aDestination, kTlvs, sizeof(kTlvs), &aQueryInfo);
}

Error Mle::SendDataRequest(const Ip6::Address &aDestination,
const uint8_t *aTlvs,
uint8_t aTlvsLength,
uint16_t aDelay,
const LinkMetrics::Initiator::QueryInfo *aQueryInfo)
#else
Error Mle::SendDataRequest(const Ip6::Address &aDestination, const uint8_t *aTlvs, uint8_t aTlvsLength, uint16_t aDelay)
Error Mle::SendDataRequest(const Ip6::Address &aDestination, const uint8_t *aTlvs, uint8_t aTlvsLength)
#endif
{
Error error = kErrorNone;
Expand All @@ -1798,22 +1794,14 @@ Error Mle::SendDataRequest(const Ip6::Address &aDestination, const uint8_t *aTlv
}
#endif

if (aDelay)
{
SuccessOrExit(error = message->SendAfterDelay(aDestination, aDelay));
Log(kMessageDelay, kTypeDataRequest, aDestination);
}
else
{
SuccessOrExit(error = message->AppendActiveAndPendingTimestampTlvs());
SuccessOrExit(error = message->AppendActiveAndPendingTimestampTlvs());

SuccessOrExit(error = message->SendTo(aDestination));
Log(kMessageSend, kTypeDataRequest, aDestination);
SuccessOrExit(error = message->SendTo(aDestination));
Log(kMessageSend, kTypeDataRequest, aDestination);

if (!IsRxOnWhenIdle())
{
Get<DataPollSender>().SendFastPolls(DataPollSender::kDefaultFastPolls);
}
if (!IsRxOnWhenIdle())
{
Get<DataPollSender>().SendFastPolls(DataPollSender::kDefaultFastPolls);
}

exit:
Expand Down Expand Up @@ -2719,8 +2707,8 @@ void Mle::HandleAdvertisement(RxInfo &aRxInfo)

if (mRetrieveNewNetworkData || IsNetworkDataNewer(leaderData))
{
delay = Random::NonCrypto::GetUint16InRange(0, kMleMaxResponseDelay);
IgnoreError(SendDataRequestAfterDelay(aRxInfo.mMessageInfo.GetPeerAddr(), delay));
delay = 1 + Random::NonCrypto::GetUint16InRange(0, kMleMaxResponseDelay);
mDelayedSender.ScheduleDataRequest(aRxInfo.mMessageInfo.GetPeerAddr(), delay);
}

aRxInfo.mClass = RxInfo::kPeerMessage;
Expand Down Expand Up @@ -2901,7 +2889,7 @@ Error Mle::HandleLeaderData(RxInfo &aRxInfo)

if (aRxInfo.mMessageInfo.GetSockAddr().IsMulticast())
{
delay = Random::NonCrypto::GetUint16InRange(0, kMleMaxResponseDelay);
delay = 1 + Random::NonCrypto::GetUint16InRange(0, kMleMaxResponseDelay);
}
else
{
Expand All @@ -2911,7 +2899,7 @@ Error Mle::HandleLeaderData(RxInfo &aRxInfo)
delay = 10;
}

IgnoreError(SendDataRequestAfterDelay(aRxInfo.mMessageInfo.GetPeerAddr(), delay));
mDelayedSender.ScheduleDataRequest(aRxInfo.mMessageInfo.GetPeerAddr(), delay);
}
else if (error == kErrorNone)
{
Expand Down Expand Up @@ -4328,100 +4316,210 @@ Mle::DelayedSender::DelayedSender(Instance &aInstance)
{
}

Error Mle::DelayedSender::SendMessage(TxMessage &aMessage, const Ip6::Address &aDestination, uint16_t aDelay)
void Mle::DelayedSender::Stop(void)
{
mTimer.Stop();
mSchedules.DequeueAndFreeAll();
}

void Mle::DelayedSender::ScheduleDataRequest(const Ip6::Address &aDestination, uint16_t aDelay)
{
VerifyOrExit(!HasMatchingSchedule(kTypeDataRequest, aDestination));
AddSchedule(kTypeDataRequest, aDestination, aDelay, nullptr, 0);

exit:
return;
}

#if OPENTHREAD_FTD

void Mle::DelayedSender::ScheduleParentResponse(const ParentResponseInfo &aInfo, uint16_t aDelay)
{
Ip6::Address destination;

destination.SetToLinkLocalAddress(aInfo.mChildExtAddress);

RemoveMatchingSchedules(kTypeParentResponse, destination);
AddSchedule(kTypeParentResponse, destination, aDelay, &aInfo, sizeof(aInfo));
}

void Mle::DelayedSender::ScheduleMulticastDataResponse(uint16_t aDelay)
{
Ip6::Address destination;

destination.SetToLinkLocalAllNodesMulticast();

Get<MeshForwarder>().RemoveDataResponseMessages();
RemoveMatchingSchedules(kTypeDataResponse, destination);
AddSchedule(kTypeDataResponse, destination, aDelay, nullptr, 0);
}

void Mle::DelayedSender::ScheduleLinkAccept(const LinkAcceptInfo &aInfo, uint16_t aDelay)
{
Ip6::Address destination;

destination.SetToLinkLocalAddress(aInfo.mExtAddress);

RemoveMatchingSchedules(kTypeLinkAccept, destination);
AddSchedule(kTypeLinkAccept, destination, aDelay, &aInfo, sizeof(aInfo));
}

void Mle::DelayedSender::ScheduleDiscoveryResponse(const Ip6::Address &aDestination,
const DiscoveryResponseInfo &aInfo,
uint16_t aDelay)
{
AddSchedule(kTypeDiscoveryResponse, aDestination, aDelay, &aInfo, sizeof(aInfo));
}

#endif // OPENTHREAD_FTD

void Mle::DelayedSender::AddSchedule(MessageType aMessageType,
const Ip6::Address &aDestination,
uint16_t aDelay,
const void *aInfo,
uint16_t aInfoSize)
{
Error error = kErrorNone;
Metadata metadata;
Schedule *schedule = Get<MessagePool>().Allocate(Message::kTypeOther);
Header header;

metadata.mSendTime = TimerMilli::GetNow() + aDelay;
metadata.mDestination = aDestination;
VerifyOrExit(schedule != nullptr);

SuccessOrExit(error = metadata.AppendTo(aMessage));
mQueue.Enqueue(aMessage);
header.mSendTime = TimerMilli::GetNow() + aDelay;
header.mDestination = aDestination;
header.mMessageType = aMessageType;
SuccessOrExit(schedule->Append(header));

mTimer.FireAtIfEarlier(metadata.mSendTime);
if (aInfo != nullptr)
{
SuccessOrExit(schedule->AppendBytes(aInfo, aInfoSize));
}

mTimer.FireAtIfEarlier(header.mSendTime);

mSchedules.Enqueue(*schedule);
schedule = nullptr;

Log(kMessageDelay, aMessageType, aDestination);

exit:
return error;
FreeMessage(schedule);
}

void Mle::DelayedSender::HandleTimer(void)
{
NextFireTime nextSendTime;
MessageQueue schedulesToExecute;

for (Message &message : mQueue)
for (Schedule &schedule : mSchedules)
{
Metadata metadata;
Header header;

metadata.ReadFrom(message);
header.ReadFrom(schedule);

if (nextSendTime.GetNow() < metadata.mSendTime)
if (nextSendTime.GetNow() < header.mSendTime)
{
nextSendTime.UpdateIfEarlier(metadata.mSendTime);
nextSendTime.UpdateIfEarlier(header.mSendTime);
}
else
{
mQueue.Dequeue(message);
Send(static_cast<TxMessage &>(message), metadata);
mSchedules.Dequeue(schedule);
schedulesToExecute.Enqueue(schedule);
}
}

mTimer.FireAt(nextSendTime);

for (Schedule &schedule : schedulesToExecute)
{
Execute(schedule);
}

schedulesToExecute.DequeueAndFreeAll();
}

void Mle::DelayedSender::Send(TxMessage &aMessage, const Metadata &aMetadata)
void Mle::DelayedSender::Execute(const Schedule &aSchedule)
{
Error error = kErrorNone;
Header header;

aMetadata.RemoveFrom(aMessage);
header.ReadFrom(aSchedule);

if (aMessage.IsMleCommand(kCommandDataRequest))
switch (header.mMessageType)
{
SuccessOrExit(error = aMessage.AppendActiveAndPendingTimestampTlvs());
}
case kTypeDataRequest:
IgnoreError(Get<Mle>().SendDataRequest(header.mDestination));
break;

#if OPENTHREAD_FTD
case kTypeParentResponse:
{
ParentResponseInfo info;

SuccessOrExit(error = aMessage.SendTo(aMetadata.mDestination));
IgnoreError(aSchedule.Read(sizeof(Header), info));
Get<MleRouter>().SendParentResponse(info);
break;
}

Log(kMessageSend, kTypeGenericDelayed, aMetadata.mDestination);
case kTypeDataResponse:
Get<MleRouter>().SendMulticastDataResponse();
break;

if (!Get<Mle>().IsRxOnWhenIdle())
case kTypeLinkAccept:
{
// Start fast poll mode, assuming enqueued msg is MLE Data Request.
// Note: Finer-grade check may be required when deciding whether or
// not to enter fast poll mode for other type of delayed message.
LinkAcceptInfo info;

Get<DataPollSender>().SendFastPolls(DataPollSender::kDefaultFastPolls);
IgnoreError(aSchedule.Read(sizeof(Header), info));
IgnoreError(Get<MleRouter>().SendLinkAccept(info));
break;
}

exit:
if (error != kErrorNone)
case kTypeDiscoveryResponse:
{
aMessage.Free();
DiscoveryResponseInfo info;

IgnoreError(aSchedule.Read(sizeof(Header), info));
IgnoreError(Get<MleRouter>().SendDiscoveryResponse(header.mDestination, info));
break;
}
}
#endif // OPENTHREAD_FTD

void Mle::DelayedSender::RemoveDataResponseMessage(void)
{
RemoveMessage(kCommandDataResponse, kTypeDataResponse, nullptr);
default:
break;
}
}

void Mle::DelayedSender::RemoveDataRequestMessage(const Ip6::Address &aDestination)
bool Mle::DelayedSender::Match(const Schedule &aSchedule, MessageType aMessageType, const Ip6::Address &aDestination)
{
RemoveMessage(kCommandDataRequest, kTypeDataRequest, &aDestination);
Header header;

header.ReadFrom(aSchedule);

return (header.mMessageType == aMessageType) && (header.mDestination == aDestination);
}

void Mle::DelayedSender::RemoveMessage(Command aCommand, MessageType aMessageType, const Ip6::Address *aDestination)
bool Mle::DelayedSender::HasMatchingSchedule(MessageType aMessageType, const Ip6::Address &aDestination) const
{
for (Message &message : mQueue)
bool hasMatching = false;

for (const Schedule &schedule : mSchedules)
{
Metadata metadata;
if (Match(schedule, aMessageType, aDestination))
{
hasMatching = true;
break;
}
}

metadata.ReadFrom(message);
return hasMatching;
}

if (message.IsMleCommand(aCommand) && ((aDestination == nullptr) || (metadata.mDestination == *aDestination)))
void Mle::DelayedSender::RemoveMatchingSchedules(MessageType aMessageType, const Ip6::Address &aDestination)
{
for (Schedule &schedule : mSchedules)
{
if (Match(schedule, aMessageType, aDestination))
{
mQueue.DequeueAndFree(message);
Log(kMessageRemoveDelayed, aMessageType, metadata.mDestination);
mSchedules.DequeueAndFree(schedule);
Log(kMessageRemoveDelayed, aMessageType, aDestination);
}
}
}
Expand Down Expand Up @@ -4846,11 +4944,6 @@ Error Mle::TxMessage::SendTo(const Ip6::Address &aDestination)
return error;
}

Error Mle::TxMessage::SendAfterDelay(const Ip6::Address &aDestination, uint16_t aDelay)
{
return Get<Mle>().mDelayedSender.SendMessage(*this, aDestination, aDelay);
}

#if OPENTHREAD_FTD

Error Mle::TxMessage::AppendConnectivityTlv(void)
Expand Down
Loading

0 comments on commit 7169561

Please sign in to comment.