Skip to content

Commit 72dab53

Browse files
sawenzelclaude
andcommitted
Release per-event buffers in the o2-sim hit merger
This fixes a memory leak in the hit merger and makes its merge flag thread-safe. - cleanEvent was empty, so the decoded SubEventInfo objects of every event were never freed. - The MC tracks and track references of events dropped by --noemptyevents were never freed either. - cleanEvent now owns the release of all three buffers; the merge functions no longer delete. - Buffer entries are emptied rather than erased, since erasing from a tbb::concurrent_unordered_map is not safe while the receiving thread inserts. - handleSimData keeps the event id and event count as values, since a merge thread may free the event info. - mergingInProgress is shared by two threads and is now std::atomic<bool>. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent c372745 commit 72dab53

1 file changed

Lines changed: 23 additions & 17 deletions

File tree

‎run/O2HitMerger.h‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767
#include <list>
6868
#include <csignal>
6969
#include <mutex>
70+
#include <atomic>
7071
#include <filesystem>
7172
#include <functional>
7273

@@ -399,6 +400,9 @@ class O2HitMerger : public fair::mq::Device
399400
int index = 0;
400401
auto infoptr = o2::base::decodeTMessage<o2::data::SubEventInfo*>(data, index++);
401402
o2::data::SubEventInfo& info = *infoptr;
403+
// once a merge thread runs, the buffered info of a complete event may be freed at any time
404+
const auto eventID = info.eventID;
405+
const auto maxEvents = info.maxEvents;
402406
auto accum = insertAdd<uint32_t, uint32_t>(mPartsCheckSum, info.eventID, (uint32_t)info.part);
403407

404408
LOG(info) << "SIMDATA channel got " << data.Size() << " parts for event " << info.eventID << " part " << info.part << " out of " << info.nparts;
@@ -423,19 +427,19 @@ class O2HitMerger : public fair::mq::Device
423427
mMergerIOThread.join();
424428
}
425429
// start hit merging and flushing in a separate thread in order not to block
426-
mMergerIOThread = std::thread([info, this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; });
430+
mMergerIOThread = std::thread([this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; });
427431
}
428432

429-
mEventChecksum += info.eventID;
433+
mEventChecksum += eventID;
430434
// we also need to check if we have all events
431-
if (isDataComplete<uint32_t>(mEventChecksum, info.maxEvents)) {
435+
if (isDataComplete<uint32_t>(mEventChecksum, maxEvents)) {
432436
LOG(info) << "ALL EVENTS HERE; CHECKSUM " << mEventChecksum;
433437

434438
// flush remaining data and close file
435439
if (mMergerIOThread.joinable()) {
436440
mMergerIOThread.join();
437441
}
438-
mMergerIOThread = std::thread([info, this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; });
442+
mMergerIOThread = std::thread([this]() { mergingInProgress = true; mergeAndFlushData(); mergingInProgress = false; });
439443
if (mMergerIOThread.joinable()) {
440444
mMergerIOThread.join();
441445
}
@@ -444,17 +448,29 @@ class O2HitMerger : public fair::mq::Device
444448
}
445449

446450
if (mPipeToDriver != -1) {
447-
if (write(mPipeToDriver, &info.eventID, sizeof(info.eventID)) == -1) {
451+
if (write(mPipeToDriver, &eventID, sizeof(eventID)) == -1) {
448452
LOG(error) << "FAILED WRITING TO PIPE";
449453
};
450454
}
451455
}
452456
return expectmore;
453457
}
454458

459+
// releases the buffered data of an event once it is flushed or discarded
455460
void cleanEvent(int eventID)
456461
{
457-
// cleanup intermediate per-Event buffers
462+
auto release = [eventID](auto& buffer) {
463+
auto iter = buffer.find(eventID);
464+
if (iter != buffer.end()) {
465+
for (auto ptr : iter->second) {
466+
delete ptr;
467+
}
468+
iter->second = {};
469+
}
470+
};
471+
release(mMCTrackBuffer);
472+
release(mTrackRefBuffer);
473+
release(mSubEventInfoBuffer);
458474
}
459475

460476
template <typename T>
@@ -558,11 +574,6 @@ class O2HitMerger : public fair::mq::Device
558574
channel.Send(reply);
559575
LOG(info) << "Forward publish MC tracks on channel";
560576
}
561-
562-
// cleanup buffered data
563-
for (auto ptr : vectorOfSubEventMCTracks) {
564-
delete ptr; // avoid this by using unique ptr
565-
}
566577
}
567578

568579
template <typename T, typename M>
@@ -609,11 +620,6 @@ class O2HitMerger : public fair::mq::Device
609620
targetbr->SetAddress(&dataaddr);
610621
targetbr->Fill();
611622
targetbr->ResetAddress();
612-
613-
// cleanup mem
614-
for (auto ptr : vectorOfT) {
615-
delete ptr; // avoid this by using unique ptr
616-
}
617623
}
618624

619625
void updateTrackIdWithOffset(MCTrack& track, Int_t nprim, Int_t idelta0, Int_t idelta1)
@@ -855,7 +861,7 @@ class O2HitMerger : public fair::mq::Device
855861

856862
// intermediate structures to collect data per event
857863
std::thread mMergerIOThread; //! a thread used to do hit merging and IO flushing asynchronously
858-
bool mergingInProgress = false;
864+
std::atomic<bool> mergingInProgress{false};
859865

860866
Hashtable<int, std::vector<std::vector<o2::MCTrack>*>> mMCTrackBuffer; //! vector of sub-event track vectors; one per event
861867
Hashtable<int, std::vector<std::vector<o2::TrackReference>*>> mTrackRefBuffer; //!

0 commit comments

Comments
 (0)