From cc24aa8f769fc5c738ed961593e71ec084387199 Mon Sep 17 00:00:00 2001 From: Zhanglong Xia Date: Thu, 27 Sep 2018 11:07:51 +0800 Subject: [PATCH] [message] create new messages with priority (#3015) --- src/core/coap/coap.cpp | 3 +- src/core/coap/coap.hpp | 2 +- src/core/common/message.cpp | 48 ++++++++++++++++---------- src/core/common/message.hpp | 27 +++++++-------- src/core/meshcop/joiner_router.cpp | 3 +- src/core/net/ip6.cpp | 6 ++-- src/core/net/ip6.hpp | 8 ++++- src/core/net/udp6.cpp | 8 ++--- src/core/net/udp6.hpp | 16 +++++++-- src/core/thread/mesh_forwarder.hpp | 10 +++--- src/core/thread/mesh_forwarder_ftd.cpp | 33 +++++++++++++----- src/core/thread/mesh_forwarder_mtd.cpp | 16 +++++++-- src/core/thread/mle.cpp | 3 +- 13 files changed, 119 insertions(+), 64 deletions(-) diff --git a/src/core/coap/coap.cpp b/src/core/coap/coap.cpp index 23fd67066..d597a5410 100644 --- a/src/core/coap/coap.cpp +++ b/src/core/coap/coap.cpp @@ -149,10 +149,9 @@ Message *CoapBase::NewMessage(const Header &aHeader, uint8_t aPriority) // Ensure that header has minimum required length. VerifyOrExit(aHeader.GetLength() >= Header::kMinHeaderLength); - VerifyOrExit((message = mSocket.NewMessage(aHeader.GetLength())) != NULL); + VerifyOrExit((message = mSocket.NewMessage(aHeader.GetLength(), aPriority)) != NULL); message->Prepend(aHeader.GetBytes(), aHeader.GetLength()); message->SetOffset(0); - message->SetPriority(aPriority); exit: return message; diff --git a/src/core/coap/coap.hpp b/src/core/coap/coap.hpp index a47fcfe90..4e35dd1bb 100644 --- a/src/core/coap/coap.hpp +++ b/src/core/coap/coap.hpp @@ -523,7 +523,7 @@ public: * This method creates a new message with a CoAP header. * * @param[in] aHeader A reference to a CoAP header that is used to create the message. - * @param[in] aPrority The message priority level. + * @param[in] aPriority The message priority level. * * @returns A pointer to the message or NULL if failed to allocate message. * diff --git a/src/core/common/message.cpp b/src/core/common/message.cpp index 474796fc1..f8ade9ad8 100644 --- a/src/core/common/message.cpp +++ b/src/core/common/message.cpp @@ -65,28 +65,29 @@ MessagePool::MessagePool(Instance &aInstance) #endif } -Message *MessagePool::New(uint8_t aType, uint16_t aReserved) +Message *MessagePool::New(uint8_t aType, uint16_t aReserved, uint8_t aPriority) { + otError error = OT_ERROR_NONE; Message *message = NULL; - SuccessOrExit(ReclaimBuffers(1)); - - VerifyOrExit((message = static_cast(NewBuffer())) != NULL); + VerifyOrExit((message = static_cast(NewBuffer(aPriority))) != NULL); memset(message, 0, sizeof(*message)); message->SetMessagePool(this); message->SetType(aType); message->SetReserved(aReserved); message->SetLinkSecurityEnabled(true); - message->SetPriority(kDefaultMessagePriority); - if (message->SetLength(0) != OT_ERROR_NONE) + SuccessOrExit(error = message->SetPriority(aPriority)); + SuccessOrExit(error = message->SetLength(0)); + +exit: + if (error != OT_ERROR_NONE) { Free(message); message = NULL; } -exit: return message; } @@ -99,10 +100,12 @@ void MessagePool::Free(Message *aMessage) FreeBuffers(static_cast(aMessage)); } -Buffer *MessagePool::NewBuffer(void) +Buffer *MessagePool::NewBuffer(uint8_t aPriority) { Buffer *buffer = NULL; + SuccessOrExit(ReclaimBuffers(1, aPriority)); + #if OPENTHREAD_CONFIG_PLATFORM_MESSAGE_MANAGEMENT buffer = static_cast(otPlatMessagePoolNew(&GetInstance())); @@ -124,6 +127,7 @@ Buffer *MessagePool::NewBuffer(void) otLogInfoMem(GetInstance(), "No available message buffer"); } +exit: return buffer; } @@ -143,16 +147,18 @@ void MessagePool::FreeBuffers(Buffer *aBuffer) } } -otError MessagePool::ReclaimBuffers(int aNumBuffers) +otError MessagePool::ReclaimBuffers(int aNumBuffers, uint8_t aPriority) { #if OPENTHREAD_MTD || OPENTHREAD_FTD while (aNumBuffers > GetFreeBufferCount()) { MeshForwarder &meshForwarder = GetInstance().GetThreadNetif().GetMeshForwarder(); - SuccessOrExit(meshForwarder.EvictIndirectMessage()); + SuccessOrExit(meshForwarder.EvictMessage(aPriority)); } exit: +#else // OPENTHREAD_MTD || OPENTHREAD_FTD + OT_UNUSED_VARIABLE(aPriority); #endif // OPENTHREAD_MTD || OPENTHREAD_FTD // First comparison is to get around issues with comparing @@ -161,13 +167,19 @@ exit: return (aNumBuffers < 0 || aNumBuffers <= GetFreeBufferCount()) ? OT_ERROR_NONE : OT_ERROR_NO_BUFS; } -#if OPENTHREAD_CONFIG_PLATFORM_MESSAGE_MANAGEMENT uint16_t MessagePool::GetFreeBufferCount(void) const { - return otPlatMessagePoolNumFreeBuffers(&GetInstance()); -} + uint16_t rval; + +#if OPENTHREAD_CONFIG_PLATFORM_MESSAGE_MANAGEMENT + rval = otPlatMessagePoolNumFreeBuffers(&GetInstance()); +#else + rval = mNumFreeBuffers; #endif + return rval; +} + Message *MessagePool::Iterator::Next(void) const { Message *next; @@ -238,7 +250,7 @@ otError Message::ResizeMessage(uint16_t aLength) { if (curBuffer->GetNextBuffer() == NULL) { - curBuffer->SetNextBuffer(GetMessagePool()->NewBuffer()); + curBuffer->SetNextBuffer(GetMessagePool()->NewBuffer(GetPriority())); VerifyOrExit(curBuffer->GetNextBuffer() != NULL, error = OT_ERROR_NO_BUFS); } @@ -303,7 +315,7 @@ otError Message::SetLength(uint16_t aLength) bufs -= (((totalLengthCurrent - kHeadBufferDataSize) - 1) / kBufferDataSize) + 1; } - SuccessOrExit(error = GetMessagePool()->ReclaimBuffers(bufs)); + SuccessOrExit(error = GetMessagePool()->ReclaimBuffers(bufs, GetPriority())); SuccessOrExit(error = ResizeMessage(totalLengthRequest)); mBuffer.mHead.mInfo.mLength = aLength; @@ -425,7 +437,7 @@ otError Message::Prepend(const void *aBuf, uint16_t aLength) while (aLength > GetReserved()) { - VerifyOrExit((newBuffer = GetMessagePool()->NewBuffer()) != NULL, error = OT_ERROR_NO_BUFS); + VerifyOrExit((newBuffer = GetMessagePool()->NewBuffer(GetPriority())) != NULL, error = OT_ERROR_NO_BUFS); newBuffer->SetNextBuffer(GetNextBuffer()); SetNextBuffer(newBuffer); @@ -651,7 +663,8 @@ Message *Message::Clone(uint16_t aLength) const otError error = OT_ERROR_NONE; Message *messageCopy; - VerifyOrExit((messageCopy = GetMessagePool()->New(GetType(), GetReserved())) != NULL, error = OT_ERROR_NO_BUFS); + VerifyOrExit((messageCopy = GetMessagePool()->New(GetType(), GetReserved(), GetPriority())) != NULL, + error = OT_ERROR_NO_BUFS); SuccessOrExit(error = messageCopy->SetLength(aLength)); CopyTo(0, 0, aLength, *messageCopy); @@ -659,7 +672,6 @@ Message *Message::Clone(uint16_t aLength) const messageCopy->SetOffset(GetOffset()); messageCopy->SetInterfaceId(GetInterfaceId()); messageCopy->SetSubType(GetSubType()); - messageCopy->SetPriority(GetPriority()); messageCopy->SetLinkSecurityEnabled(IsLinkSecurityEnabled()); #if OPENTHREAD_CONFIG_ENABLE_TIME_SYNC messageCopy->SetTimeSync(IsTimeSync()); diff --git a/src/core/common/message.hpp b/src/core/common/message.hpp index 35ee3875b..cd124213c 100644 --- a/src/core/common/message.hpp +++ b/src/core/common/message.hpp @@ -359,7 +359,7 @@ public: * If the message is already queued in a priority queue, changing the priority ensures to * update the message in the associated queue. * - * @param[in] aPrority The message priority level. + * @param[in] aPriority The message priority level. * * @retval OT_ERROR_NONE Successfully set the priority for the message. * @retval OT_ERROR_INVALID_ARGS Priority level is not invalid. @@ -1075,7 +1075,6 @@ public: */ void GetInfo(uint16_t &aMessageCount, uint16_t &aBufferCount) const; -private: /** * This method returns the tail of the list (last message in the list) * @@ -1084,6 +1083,7 @@ private: */ Message *GetTail(void) const; +private: /** * This method adds a message to a list. * @@ -1243,11 +1243,12 @@ public: * * @param[in] aType The message type. * @param[in] aReserveHeader The number of header bytes to reserve. + * @param[in] aPriority The priority level of the message. * * @returns A pointer to the message or NULL if no message buffers are available. * */ - Message *New(uint8_t aType, uint16_t aReserveHeader); + Message *New(uint8_t aType, uint16_t aReserveHeader, uint8_t aPriority = kDefaultMessagePriority); /** * This method is used to free a message and return all message buffers to the buffer pool. @@ -1275,17 +1276,13 @@ public: */ Iterator GetAllMessagesTail(void) const { return Iterator(mAllQueue.GetTail()); } - /** - * This method returns the number of free buffers. - * - * @returns The number of free buffers. - * - */ -#if OPENTHREAD_CONFIG_PLATFORM_MESSAGE_MANAGEMENT + /** + * This method returns the number of free buffers. + * + * @returns The number of free buffers. + * + */ uint16_t GetFreeBufferCount(void) const; -#else - uint16_t GetFreeBufferCount(void) const { return mNumFreeBuffers; } -#endif private: enum @@ -1293,9 +1290,9 @@ private: kDefaultMessagePriority = Message::kPriorityLow, }; - Buffer * NewBuffer(void); + Buffer * NewBuffer(uint8_t aPriority); void FreeBuffers(Buffer *aBuffer); - otError ReclaimBuffers(int aNumBuffers); + otError ReclaimBuffers(int aNumBuffers, uint8_t aPriority); PriorityQueue *GetAllMessagesQueue(void) { return &mAllQueue; } #if OPENTHREAD_CONFIG_PLATFORM_MESSAGE_MANAGEMENT == 0 diff --git a/src/core/meshcop/joiner_router.cpp b/src/core/meshcop/joiner_router.cpp index 59a7bee27..86a24a9a4 100644 --- a/src/core/meshcop/joiner_router.cpp +++ b/src/core/meshcop/joiner_router.cpp @@ -244,8 +244,7 @@ void JoinerRouter::HandleRelayTransmit(Coap::Header &aHeader, Message &aMessage, SuccessOrExit(error = Tlv::GetValueOffset(aMessage, Tlv::kJoinerDtlsEncapsulation, offset, length)); - VerifyOrExit((message = mSocket.NewMessage(0)) != NULL, error = OT_ERROR_NO_BUFS); - message->SetPriority(kMeshCoPMessagePriority); + VerifyOrExit((message = mSocket.NewMessage(0, kMeshCoPMessagePriority)) != NULL, error = OT_ERROR_NO_BUFS); message->SetLinkSecurityEnabled(false); while (length) diff --git a/src/core/net/ip6.cpp b/src/core/net/ip6.cpp index ef8c7c54b..55b663df5 100644 --- a/src/core/net/ip6.cpp +++ b/src/core/net/ip6.cpp @@ -67,10 +67,10 @@ Ip6::Ip6(Instance &aInstance) { } -Message *Ip6::NewMessage(uint16_t aReserved) +Message *Ip6::NewMessage(uint16_t aReserved, uint8_t aPriority) { - return GetInstance().GetMessagePool().New(Message::kTypeIp6, - sizeof(Header) + sizeof(HopByHopHeader) + sizeof(OptionMpl) + aReserved); + return GetInstance().GetMessagePool().New( + Message::kTypeIp6, sizeof(Header) + sizeof(HopByHopHeader) + sizeof(OptionMpl) + aReserved, aPriority); } uint16_t Ip6::UpdateChecksum(uint16_t aChecksum, const Address &aAddress) diff --git a/src/core/net/ip6.hpp b/src/core/net/ip6.hpp index a81c87341..f8faa324d 100644 --- a/src/core/net/ip6.hpp +++ b/src/core/net/ip6.hpp @@ -111,11 +111,12 @@ public: * This method allocates a new message buffer from the buffer pool. * * @param[in] aReserved The number of header bytes to reserve following the IPv6 header. + * @param[in] aPriority The priority level of the message. * * @returns A pointer to the message or NULL if insufficient message buffers are available. * */ - Message *NewMessage(uint16_t aReserved); + Message *NewMessage(uint16_t aReserved, uint8_t aPriority = kDefaultIp6MessagePriority); /** * This constructor initializes the object. @@ -379,6 +380,11 @@ public: static const char *IpProtoToString(IpProto aIpProto); private: + enum + { + kDefaultIp6MessagePriority = Message::kPriorityLow, + }; + static void HandleSendQueue(Tasklet &aTasklet); void HandleSendQueue(void); diff --git a/src/core/net/udp6.cpp b/src/core/net/udp6.cpp index 1d8e744cc..c6aec4aa0 100644 --- a/src/core/net/udp6.cpp +++ b/src/core/net/udp6.cpp @@ -55,9 +55,9 @@ Udp &UdpSocket::GetUdp(void) return GetInstance().GetIp6().GetUdp(); } -Message *UdpSocket::NewMessage(uint16_t aReserved) +Message *UdpSocket::NewMessage(uint16_t aReserved, uint8_t aPriority) { - return GetUdp().NewMessage(aReserved); + return GetUdp().NewMessage(aReserved, aPriority); } otError UdpSocket::Open(otUdpReceive aHandler, void *aContext) @@ -241,9 +241,9 @@ uint16_t Udp::GetEphemeralPort(void) return rval; } -Message *Udp::NewMessage(uint16_t aReserved) +Message *Udp::NewMessage(uint16_t aReserved, uint8_t aPriority) { - return GetIp6().NewMessage(sizeof(UdpHeader) + aReserved); + return GetIp6().NewMessage(sizeof(UdpHeader) + aReserved, aPriority); } otError Udp::SendDatagram(Message &aMessage, MessageInfo &aMessageInfo, IpProto aIpProto) diff --git a/src/core/net/udp6.hpp b/src/core/net/udp6.hpp index 4f1ed3000..2539d8b29 100644 --- a/src/core/net/udp6.hpp +++ b/src/core/net/udp6.hpp @@ -110,11 +110,12 @@ public: * This method returns a new UDP message with sufficient header space reserved. * * @param[in] aReserved The number of header bytes to reserve after the UDP header. + * @param[in] aPriority A reference to the priority of the message. * * @returns A pointer to the message or NULL if no buffers are available. * */ - Message *NewMessage(uint16_t aReserved); + Message *NewMessage(uint16_t aReserved, uint8_t aPriority = kDefaultSocketMessagePriority); /** * This method opens the UDP socket. @@ -185,6 +186,11 @@ public: SockAddr &GetPeerName(void) { return *static_cast(&mPeerName); } private: + enum + { + kDefaultSocketMessagePriority = Message::kPriorityLow, + }; + UdpSocket *GetNext(void) { return static_cast(mNext); } void SetNext(UdpSocket *socket) { mNext = static_cast(socket); } @@ -265,11 +271,12 @@ public: * This method returns a new UDP message with sufficient header space reserved. * * @param[in] aReserved The number of header bytes to reserve after the UDP header. + * @param[in] aPriority The priority of the message. * * @returns A pointer to the message or NULL if no buffers are available. * */ - Message *NewMessage(uint16_t aReserved); + Message *NewMessage(uint16_t aReserved, uint8_t aPriority = kDefaultUdpMessagePriority); /** * This method sends an IPv6 datagram. @@ -333,6 +340,11 @@ public: #endif // OPENTHREAD_ENABLE_UDP_PROXY private: + enum + { + kDefaultUdpMessagePriority = Message::kPriorityLow, + }; + enum { kDynamicPortMin = 49152, ///< Service Name and Transport Protocol Port Number Registry diff --git a/src/core/thread/mesh_forwarder.hpp b/src/core/thread/mesh_forwarder.hpp index 7ecb13ed9..98070d1c3 100644 --- a/src/core/thread/mesh_forwarder.hpp +++ b/src/core/thread/mesh_forwarder.hpp @@ -171,13 +171,15 @@ public: void RemoveDataResponseMessages(void); /** - * This method evicts the first indirect message in the indirect send queue. + * This method evicts the message with lowest priority in the send queue. * - * @retval OT_ERROR_NONE Successfully evicted an indirect message. - * @retval OT_ERROR_NOT_FOUND No indirect messages available to evict. + * @param[in] aPriority The highest priority level of the evicted message. + * + * @retval OT_ERROR_NONE Successfully evicted a low priority message. + * @retval OT_ERROR_NOT_FOUND No low priority messages available to evict. * */ - otError EvictIndirectMessage(void); + otError EvictMessage(uint8_t aPriority); /** * This method returns a reference to the send queue. diff --git a/src/core/thread/mesh_forwarder_ftd.cpp b/src/core/thread/mesh_forwarder_ftd.cpp index 8c30e3c47..c871afad1 100644 --- a/src/core/thread/mesh_forwarder_ftd.cpp +++ b/src/core/thread/mesh_forwarder_ftd.cpp @@ -235,20 +235,37 @@ void MeshForwarder::UpdateIndirectMessages(void) } } -otError MeshForwarder::EvictIndirectMessage(void) +otError MeshForwarder::EvictMessage(uint8_t aPriority) { - otError error = OT_ERROR_NOT_FOUND; + otError error = OT_ERROR_NOT_FOUND; + Message *message; - for (Message *message = mSendQueue.GetHead(); message; message = message->GetNext()) + VerifyOrExit((message = mSendQueue.GetTail()) != NULL); + + if (message->GetPriority() > aPriority) { - if (!message->IsChildPending()) - { - continue; - } - RemoveMessage(*message); ExitNow(error = OT_ERROR_NONE); } + else + { + while (1) + { + for (message = mSendQueue.GetHeadForPriority(aPriority); message && (message->GetPriority() == aPriority); + message = message->GetNext()) + { + if (message->IsChildPending()) + { + RemoveMessage(*message); + ExitNow(error = OT_ERROR_NONE); + } + } + + VerifyOrExit(aPriority != Message::kPriorityHigh); + + aPriority--; + } + } exit: return error; diff --git a/src/core/thread/mesh_forwarder_mtd.cpp b/src/core/thread/mesh_forwarder_mtd.cpp index b3c206e1c..5ea2bc1ae 100644 --- a/src/core/thread/mesh_forwarder_mtd.cpp +++ b/src/core/thread/mesh_forwarder_mtd.cpp @@ -54,9 +54,21 @@ exit: return error; } -otError MeshForwarder::EvictIndirectMessage(void) +otError MeshForwarder::EvictMessage(uint8_t aPriority) { - return OT_ERROR_NOT_FOUND; + otError error = OT_ERROR_NOT_FOUND; + Message *message; + + VerifyOrExit((message = mSendQueue.GetTail()) != NULL); + + if (message->GetPriority() > aPriority) + { + RemoveMessage(*message); + ExitNow(error = OT_ERROR_NONE); + } + +exit: + return error; } otError MeshForwarder::GetIndirectTransmission(void) diff --git a/src/core/thread/mle.cpp b/src/core/thread/mle.cpp index 3274e18ba..d141237e8 100644 --- a/src/core/thread/mle.cpp +++ b/src/core/thread/mle.cpp @@ -1081,12 +1081,11 @@ Message *Mle::NewMleMessage(void) { Message *message; - message = mSocket.NewMessage(0); + message = mSocket.NewMessage(0, kMleMessagePriority); VerifyOrExit(message != NULL); message->SetSubType(Message::kSubTypeMleGeneral); message->SetLinkSecurityEnabled(false); - message->SetPriority(kMleMessagePriority); exit: return message;