Skip to content

Commit 64771ff

Browse files
sawenzelclaude
andcommitted
Flush the output files of the o2-sim hit merger concurrently
This runs the per-event merge and flush of the kinematics and of each detector in parallel. - The hit merger merged and filled the kinematics and all detector trees one after another. - In PbPb this serial step took 12.9 s of a 66 s run, after all workers had finished. - Each of these writes to its own TFile, so they now run as tasks of a tbb::task_group. - The final TFile::Write calls run in parallel as well. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 3c27650 commit 64771ff

1 file changed

Lines changed: 31 additions & 41 deletions

File tree

‎run/O2HitMerger.h‎

Lines changed: 31 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,8 @@
8686
#endif
8787

8888
#include <tbb/concurrent_unordered_map.h>
89+
#include <tbb/parallel_for_each.h>
90+
#include <tbb/task_group.h>
8991

9092
namespace o2
9193
{
@@ -787,68 +789,56 @@ class O2HitMerger : public fair::mq::Device
787789
eventheader->putInfo("prims_eta_0.8_pi", eta0Point8CounterPi);
788790
eventheader->putInfo("prims_total", prims);
789791
};
790-
reorderAndMergeMCTracks(flusheventID, mOutTree, nprimaries, subevOrdered, mcheaderhook, eventheader);
791-
792-
if (mOutTree) {
793-
// adjusting and merging track references
794-
remapTrackIdsAndMerge<std::vector<o2::TrackReference>>("TrackRefs", flusheventID, *mOutTree, trackoffsets, nprimaries, subevOrdered, mTrackRefBuffer);
795-
796-
// write MC event headers
797-
{
798-
auto headerbr = o2::base::getOrMakeBranch(*mOutTree, "MCEventHeader.", &eventheader);
799-
headerbr->SetAddress(&eventheader);
800-
headerbr->Fill();
801-
headerbr->ResetAddress();
802-
}
792+
// the kinematics and each detector go to separate files, so we merge and flush them concurrently
793+
tbb::task_group tasks;
794+
tasks.run([&]() {
795+
reorderAndMergeMCTracks(flusheventID, mOutTree, nprimaries, subevOrdered, mcheaderhook, eventheader);
796+
797+
if (mOutTree) {
798+
// adjusting and merging track references
799+
remapTrackIdsAndMerge<std::vector<o2::TrackReference>>("TrackRefs", flusheventID, *mOutTree, trackoffsets, nprimaries, subevOrdered, mTrackRefBuffer);
800+
801+
// write MC event headers
802+
for (auto tree : {mOutTree, mMCHeaderTree}) {
803+
auto headerbr = o2::base::getOrMakeBranch(*tree, "MCEventHeader.", &eventheader);
804+
headerbr->SetAddress(&eventheader);
805+
headerbr->Fill();
806+
headerbr->ResetAddress();
807+
}
803808

804-
{
805-
auto headerbr = o2::base::getOrMakeBranch(*mMCHeaderTree, "MCEventHeader.", &eventheader);
806-
headerbr->SetAddress(&eventheader);
807-
headerbr->Fill();
808-
headerbr->ResetAddress();
809+
// increase the entry count in the trees
810+
mOutTree->SetEntries(mOutTree->GetEntries() + 1);
811+
mMCHeaderTree->SetEntries(mMCHeaderTree->GetEntries() + 1);
809812
}
810-
}
813+
});
811814

812815
// c) do the merge procedure for all hits ... delegate this to detector specific functions
813816
// since they know about types; number of branches; etc.
814817
// this will also fix the trackIDs inside the hits
815818
for (int id = 0; id < mDetectorInstances.size(); ++id) {
816819
auto& det = mDetectorInstances[id];
817-
if (det) {
818-
auto hittree = mDetectorToTTreeMap[id];
819-
if (hittree) {
820+
auto hittree = det ? mDetectorToTTreeMap[id] : nullptr;
821+
if (hittree) {
822+
tasks.run([&, det = det.get(), hittree]() {
820823
det->mergeHitEntriesAndFlush(flusheventID, *hittree, trackoffsets, nprimaries, subevOrdered);
821824
hittree->SetEntries(hittree->GetEntries() + 1);
822-
LOG(info) << "flushing tree to file " << hittree->GetDirectory()->GetFile()->GetName();
823-
}
825+
});
824826
}
825827
}
826-
827-
// increase the entry count in the tree
828-
if (mOutTree) {
829-
mOutTree->SetEntries(mOutTree->GetEntries() + 1);
830-
LOG(info) << "outtree has file " << mOutTree->GetDirectory()->GetFile()->GetName();
831-
}
832-
if (mMCHeaderTree) {
833-
mMCHeaderTree->SetEntries(mMCHeaderTree->GetEntries() + 1);
834-
LOG(info) << "mc header outtree has file " << mMCHeaderTree->GetDirectory()->GetFile()->GetName();
835-
}
828+
tasks.wait();
836829

837830
cleanEvent(flusheventID);
838831
LOG(info) << "Merge/flush for event " << flusheventID << " took " << timer.RealTime();
839832
}
840833
if (mWriteToDisc && mOutFile) {
841834
LOG(info) << "Writing TTrees";
842-
mOutFile->Write("", TObject::kOverwrite);
835+
std::vector<TFile*> files{mOutFile, mMCHeaderOnlyOutFile};
843836
for (int id = 0; id < mDetectorInstances.size(); ++id) {
844-
auto& det = mDetectorInstances[id];
845-
if (det && mDetectorOutFiles[id]) {
846-
mDetectorOutFiles[id]->Write("", TObject::kOverwrite);
837+
if (mDetectorInstances[id] && mDetectorOutFiles[id]) {
838+
files.push_back(mDetectorOutFiles[id]);
847839
}
848840
}
849-
if (mMCHeaderOnlyOutFile) {
850-
mMCHeaderOnlyOutFile->Write("", TObject::kOverwrite);
851-
}
841+
tbb::parallel_for_each(files, [](TFile* file) { file->Write("", TObject::kOverwrite); });
852842
}
853843
return true;
854844
}

0 commit comments

Comments
 (0)