Skip to content

HistoSender/HistoReceiver: fixes memory leak and framed histogram packets to increase reliability #317

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

Open
corrodis wants to merge 2 commits into
base: develop
Choose a base branch
from
+100 −47
Open

HistoSender/HistoReceiver: fixes memory leak and framed histogram packets to increase reliability #317

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f6d3632
HistoSender/HistoReceiver: framed histogram packets and leak fixes
corrodis Jul 17, 2026
f5dfb7a
clangify
corrodis Jul 18, 2026
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
132 changes: 86 additions & 46 deletions otsdaq-mu2e/ArtModules/HistoReceiver.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,58 +5,64 @@

namespace ots
{
namespace
{
constexpr UInt_t kHistoPacketMagic = 0x4f545331; // 'OTS1'
}

void HistoReceiver::addHistogram(TH1* h, TDirectory* subdir, int mode)
{
TH1* h_out = (TH1*)subdir->FindObjectAny(h->GetName()); // find in memory
h->SetDirectory(nullptr);
TH1* h_out = (TH1*)subdir->FindObjectAny(h->GetName());
if(h_out == nullptr)
subdir->WriteTObject(h);
else
{
// __COUT__ << "[HistoReceiver::" << __func__ << "] Updating histogram "<< h->GetName() << std::endl;
if(mode == kAdd)
{
h_out->Add(h);
}
else if(mode == kReplace)
{
h_out->Reset();
h_out->Add(h);
}
h->SetDirectory(subdir);
return;
}
if(mode == kAdd)
h_out->Add(h);
else if(mode == kReplace)
{
h_out->Reset();
h_out->Add(h);
}
delete h;
}
void HistoReceiver::addGraph(TGraph* g, TDirectory* subdir, int mode)
{
TGraph* g_out = (TGraph*)subdir->FindObjectAny(g->GetName()); // find in memory
TGraph* g_out = (TGraph*)subdir->FindObjectAny(g->GetName());
if(g_out == nullptr)
subdir->WriteTObject(g);
{
TGraph* g_copy = (TGraph*)g->Clone(g->GetName());
subdir->Append(g_copy);
subdir->WriteTObject(g_copy, g_copy->GetName(), "Overwrite");
}
else
{
// __COUT__ << "[HistoReceiver::" << __func__ << "] Updating graph "<< h->GetName() << std::endl;
// FIXME: Add graph replacement/addition logic
const int npoints = g->GetN();
g_out->Set(0);
for(int ipoint = 0; ipoint < npoints; ++ipoint)
g_out->AddPoint(g->GetX()[ipoint], g->GetY()[ipoint]);
subdir->WriteTObject(g_out, g_out->GetName(), "Overwrite");
}
delete g;
}

void HistoReceiver::addObject(TObject* readObject, TDirectory* subdir, int mode)
{
if(readObject->InheritsFrom(TH1::Class()))
{
TH1* h = (TH1*)readObject;
addHistogram(h, subdir, mode);
addHistogram((TH1*)readObject, subdir, mode);
return;
}
else if(readObject->InheritsFrom(TGraph::Class()))
{
TGraph* g = (TGraph*)readObject;
addGraph(g, subdir, mode);
}
else
if(readObject->InheritsFrom(TGraph::Class()))
{
__COUT__ << "[HistoReceiver::" << __func__ << "] Unknown object type with name "
<< readObject->GetName() << std::endl;
addGraph((TGraph*)readObject, subdir, mode);
return;
}
__COUT__ << "[HistoReceiver::" << __func__ << "] Unknown object type with name "
<< readObject->GetName() << std::endl;
delete readObject;
}

void HistoReceiver::readPacket(TDirectory* dir, std::string* buf)
Expand All @@ -67,10 +73,11 @@ void HistoReceiver::readPacket(TDirectory* dir, std::string* buf)
message.SetBufferOffset(0); // move pointer

std::string directoryNameStdString;

do
while(message.Length() < message.BufferSize())
{
message.ReadStdString(directoryNameStdString);
if(directoryNameStdString.empty())
break;
// check if the plotting mode was defined
int mode(kAdd);
if(directoryNameStdString.find(":") != std::string::npos)
Expand All @@ -93,30 +100,63 @@ void HistoReceiver::readPacket(TDirectory* dir, std::string* buf)
subdir = subdir->mkdir(dirStr.Data(), "", kTRUE);
}

//now let's add the plottables
TObject* readObject = nullptr;
do
// New framing: magic + object count per directory block.
auto payloadOffset = message.Length();
UInt_t magic = 0;
UInt_t objectCount = 0;
bool useCountFraming = false;
if(static_cast<size_t>(message.BufferSize() - message.Length()) >=
sizeof(UInt_t) * 2)
{
auto lengthBefore = message.Length();
readObject = (TObject*)message.ReadObjectAny(TObject::Class());
if(readObject != nullptr)
message >> magic;
if(magic == kHistoPacketMagic)
{
addObject(readObject, subdir, mode);
message >> objectCount;
useCountFraming = true;
}
else
{
message.SetBufferOffset(lengthBefore);
message.SetBufferOffset(payloadOffset);
}
}

if(useCountFraming)
{
for(UInt_t i = 0; i < objectCount; ++i)
{
TObject* readObject = (TObject*)message.ReadObjectAny(TObject::Class());
if(readObject == nullptr)
{
__COUT__
<< "[HistoReceiver::" << __func__
<< "] Unexpected null object in counted payload for directory "
<< directoryNameStdString << ", index " << i << std::endl;
break;
}
addObject(readObject, subdir, mode);
}
} while(readObject != nullptr);
}
else
{
// Legacy fallback: probe until non-object boundary is reached.
TObject* readObject = nullptr;
do
{
auto lengthBefore = message.Length();
readObject = (TObject*)message.ReadObjectAny(TObject::Class());
if(readObject != nullptr)
{
addObject(readObject, subdir, mode);
}
else
{
message.SetBufferOffset(lengthBefore);
}
} while(readObject != nullptr);
}

dir->cd();
// __COUT__ << "[HistoReceiver::readPacket] message.BufferSize() - message.Length() = " << (message.BufferSize() - message.Length()) << std::endl;
// if (message.BufferSize() - message.Length()){
// char nn[10];
// message.ReadArray(&nn[0]);
// nn[9] = '\0';
// __COUT__ << "[HistoReceiver::readPacket] nn[10] = " << std::string(nn) << std::endl;
// }
} while(directoryNameStdString != "");
}
buf = nullptr;
}

Expand Down
2 changes: 1 addition & 1 deletion otsdaq-mu2e/ArtModules/HistoReceiver.hh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ class HistoReceiver
};
void addHistogram(TH1* h, TDirectory* subdir, int mode);
void addGraph(TGraph* g, TDirectory* subdir, int mode);
void addObject(TObject* h, TDirectory* subdir, int mode);
void addObject(TObject* readObject, TDirectory* subdir, int mode);
void readPacket(TDirectory* dir, std::string* buf);
int parseMode(std::string mode);
};
Expand Down
13 changes: 13 additions & 0 deletions otsdaq-mu2e/ArtModules/HistoSender.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,11 @@

namespace ots
{
namespace
{
constexpr UInt_t kHistoPacketMagic = 0x4f545331; // 'OTS1'
}

HistoSender::HistoSender(std::string serverHost, int serverPort)
: sender_(serverHost, serverPort)
{
Expand All @@ -22,6 +27,8 @@ void HistoSender::sendHistogram(std::string directoryName, TH1* hist)
TBufferFile buffer_(TBufferFile::kWrite);
buffer_.SetWriteMode();
buffer_.WriteStdString(directoryName);
buffer_ << kHistoPacketMagic;
buffer_ << static_cast<UInt_t>(1);
buffer_.WriteObject(hist);
sender_.sendPacket(buffer_.Buffer(), buffer_.Length());
buffer_.Reset();
Expand All @@ -34,6 +41,8 @@ void HistoSender::sendHistograms(std::string directoryName, std::vector<TH1*>& h
TBufferFile buffer_(TBufferFile::kWrite);
buffer_.SetWriteMode();
buffer_.WriteStdString(directoryName);
buffer_ << kHistoPacketMagic;
buffer_ << static_cast<UInt_t>(hists.size());
for(size_t i = 0; i < hists.size(); ++i)
{
buffer_.WriteObject(hists[i]);
Expand All @@ -50,6 +59,8 @@ void HistoSender::sendHistograms(std::map<std::string, std::vector<TH1*>>& hists
{
std::string dirName = hist_iter.first;
buffer_.WriteStdString(dirName);
buffer_ << kHistoPacketMagic;
buffer_ << static_cast<UInt_t>(hist_iter.second.size());
for(size_t i = 0; i < hist_iter.second.size(); ++i)
{
buffer_.WriteObject(hist_iter.second[i]);
Expand All @@ -67,6 +78,8 @@ void HistoSender::sendGraphs(std::map<std::string, std::vector<TGraph*>>& graphs
{
std::string dirName = iter.first;
buffer_.WriteStdString(dirName);
buffer_ << kHistoPacketMagic;
buffer_ << static_cast<UInt_t>(iter.second.size());
for(size_t i = 0; i < iter.second.size(); ++i)
{
buffer_.WriteObject(iter.second[i]);
Expand Down
Loading