espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-10-22 23:48:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

[message] create new messages with priority (#3015)

Browse Source
This commit is contained in:
Zhanglong Xia
2018-09-26 20:07:51 -07:00
committed by Jonathan Hui
parent ea1b83a4e7
commit cc24aa8f76
13 changed files with 119 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/coap/coap.cpp
+1 -2
View File
@@ -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;
src/core/coap/coap.hpp
+1 -1
View File
@@ -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.
*
src/core/common/message.cpp
+30 -18
View File
@@ -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<Message *>(NewBuffer())) != NULL);
VerifyOrExit((message = static_cast<Message *>(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<Buffer *>(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<Buffer *>(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());
src/core/common/message.hpp
+12 -15
View File
@@ -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
src/core/meshcop/joiner_router.cpp
+1 -2
View File
@@ -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)
src/core/net/ip6.cpp
+3 -3
View File
@@ -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)
src/core/net/ip6.hpp
+7 -1
View File
@@ -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);
src/core/net/udp6.cpp
+4 -4
View File
@@ -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)
src/core/net/udp6.hpp
+14 -2
View File
@@ -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<SockAddr *>(&mPeerName); }
private:
enum
{
kDefaultSocketMessagePriority = Message::kPriorityLow,
};
UdpSocket *GetNext(void) { return static_cast<UdpSocket *>(mNext); }
void SetNext(UdpSocket *socket) { mNext = static_cast<otUdpSocket *>(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
src/core/thread/mesh_forwarder.hpp
+6 -4
View File
@@ -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.
src/core/thread/mesh_forwarder_ftd.cpp
+25 -8
View File
@@ -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;
src/core/thread/mesh_forwarder_mtd.cpp
+14 -2
View File
@@ -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)
src/core/thread/mle.cpp
+1 -2
View File
@@ -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;