Skip to content

[MessageControl] Make sure no messages are missed via the UDP output #362

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pwielders merged 7 commits into from
Sep 20, 2025
Merged

[MessageControl] Make sure no messages are missed via the UDP output #362

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6ed437d
Use a queue of ProxyType<string> instead of a buffer for UDP output t…
VeithMetro Aug 5, 2025
57119a0
Remove ProxyType entirely, as the lifetime of the string is managed b…
VeithMetro Aug 6, 2025
bd1e613
Make sure to use std::move
VeithMetro Aug 6, 2025
df1c6e2
Indeed no need for this std::move..
VeithMetro Aug 6, 2025
219f0de
Make the string const
VeithMetro Aug 6, 2025
8a0c03a
Optimize with std::move to avoid all possible copies
VeithMetro Aug 21, 2025
0877d7f
Optimize Text::Convert to create less copies
VeithMetro Aug 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 21 additions & 17 deletions MessageControl/MessageOutput.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,12 @@ namespace Publishers {
{
ASSERT(metadata.Type() != Core::Messaging::Metadata::type::INVALID);

string output = metadata.ToString(_abbreviated).c_str() +
Core::Format("%s\n", text.c_str());
string output = metadata.ToString(_abbreviated);

output.reserve(output.size() + text.size() + 1);

output.append(text);
output.push_back('\n');

return (output);
}
Expand Down Expand Up @@ -131,9 +135,8 @@ namespace Publishers {
//UDPOutput
UDPOutput::Channel::Channel(const Core::NodeId& nodeId)
: Core::SocketDatagram(false, nodeId.Origin(), nodeId, Messaging::MessageUnit::Instance().DataSize(), 0)
, _loaded(0)
, _queue()
{
::memset(_sendBuffer, 0, sizeof(_sendBuffer));
}
UDPOutput::Channel::~Channel()
{
Expand All @@ -142,13 +145,21 @@ namespace Publishers {

uint16_t UDPOutput::Channel::SendData(uint8_t* dataFrame, const uint16_t maxSendSize)
{
uint16_t actualByteCount = 0;

_adminLock.Lock();

uint16_t actualByteCount = (_loaded > maxSendSize ? maxSendSize : _loaded);
memcpy(dataFrame, _sendBuffer, actualByteCount);
_loaded = 0;
if (_queue.empty() == true) {
Comment thread
VeithMetro marked this conversation as resolved.
Comment thread
VeithMetro marked this conversation as resolved.
Comment thread
VeithMetro marked this conversation as resolved.
Comment thread
VeithMetro marked this conversation as resolved.
_adminLock.Unlock();
}
else {
string msg = std::move(_queue.front());
_queue.pop();
_adminLock.Unlock();
Comment thread
VeithMetro marked this conversation as resolved.

_adminLock.Unlock();
actualByteCount = std::min<uint16_t>(msg.size(), maxSendSize);
memcpy(dataFrame, msg.c_str(), actualByteCount);
Comment thread
VeithMetro marked this conversation as resolved.
Comment thread
VeithMetro marked this conversation as resolved.
}

return (actualByteCount);
}
Expand All @@ -162,18 +173,11 @@ namespace Publishers {
{
}

void UDPOutput::Channel::Output(const string& text)
void UDPOutput::Channel::Output(string&& text)
{
_adminLock.Lock();

ASSERT((_loaded + text.length() + 1) < sizeof(_sendBuffer));

if ((_loaded + text.length() + 1) < sizeof(_sendBuffer)) {
Core::FrameType<0> frame(_sendBuffer + _loaded, sizeof(_sendBuffer) - _loaded, sizeof(_sendBuffer) - _loaded);
Core::FrameType<0>::Writer frameWriter(frame, 0);
frameWriter.NullTerminatedText(text);
_loaded += frameWriter.Offset();
}
_queue.emplace(std::move(text));
Comment thread
VeithMetro marked this conversation as resolved.
Comment thread
VeithMetro marked this conversation as resolved.

_adminLock.Unlock();

Expand Down
7 changes: 2 additions & 5 deletions MessageControl/MessageOutput.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -345,8 +345,6 @@ namespace Publishers {

class UDPOutput : public IPublish {
private:
static constexpr uint16_t UDPBufferSize = 4 * 1024;

class Channel : public Core::SocketDatagram {
public:
Channel() = delete;
Expand All @@ -356,16 +354,15 @@ namespace Publishers {
explicit Channel(const Core::NodeId& nodeId);
~Channel() override;

void Output(const string& text);
void Output(string&& text);

private:
uint16_t SendData(uint8_t* dataFrame, const uint16_t maxSendSize) override;
// Unused
uint16_t ReceiveData(uint8_t*, const uint16_t) override;
void StateChange() override;

uint8_t _sendBuffer[UDPBufferSize];
uint16_t _loaded;
std::queue<string> _queue;
Core::CriticalSection _adminLock;
};

Expand Down
Loading