|
32 | 32 | #include "CommonUtils/ShmManager.h" |
33 | 33 | #include "CommonUtils/ShmAllocator.h" |
34 | 34 | #include <sys/shm.h> |
| 35 | +#include <atomic> |
35 | 36 | #include <type_traits> |
36 | 37 | #include <unistd.h> |
37 | 38 | #include <cassert> |
@@ -269,11 +270,14 @@ inline std::string demangle(const char* name) |
269 | 270 | return (status == 0) ? res.get() : name; |
270 | 271 | } |
271 | 272 |
|
272 | | -void attachShmMessage(void* hitsptr, fair::mq::Channel& channel, fair::mq::Parts& parts, bool* busy_ptr); |
273 | | -void* decodeShmCore(fair::mq::Parts& dataparts, int index, bool*& busy); |
| 273 | +// a flag in shared memory telling whether the hit merger still reads a hit buffer |
| 274 | +using ShmBusyFlag = std::atomic<bool>; |
| 275 | + |
| 276 | +void attachShmMessage(void* hitsptr, fair::mq::Channel& channel, fair::mq::Parts& parts, ShmBusyFlag* busy_ptr); |
| 277 | +void* decodeShmCore(fair::mq::Parts& dataparts, int index, ShmBusyFlag*& busy); |
274 | 278 |
|
275 | 279 | template <typename T> |
276 | | -T decodeShmMessage(fair::mq::Parts& dataparts, int index, bool*& busy) |
| 280 | +T decodeShmMessage(fair::mq::Parts& dataparts, int index, ShmBusyFlag*& busy) |
277 | 281 | { |
278 | 282 | return reinterpret_cast<T>(decodeShmCore(dataparts, index, busy)); |
279 | 283 | } |
@@ -553,7 +557,7 @@ class DetImpl : public o2::base::Detector |
553 | 557 | mHitCollectorBufferPtr = (char*)&hitcollector; |
554 | 558 |
|
555 | 559 | int probe = 0; |
556 | | - bool* busy = nullptr; |
| 560 | + ShmBusyFlag* busy = nullptr; |
557 | 561 | using HitPtr_t = decltype(static_cast<Det*>(this)->Det::getHits(probe)); |
558 | 562 | std::string name = static_cast<Det*>(this)->getHitBranchNames(probe); |
559 | 563 |
|
@@ -606,7 +610,7 @@ class DetImpl : public o2::base::Detector |
606 | 610 | void fillHitBranch(TTree& tr, fair::mq::Parts& parts, int& index) override |
607 | 611 | { |
608 | 612 | int probe = 0; |
609 | | - bool* busy = nullptr; |
| 613 | + ShmBusyFlag* busy = nullptr; |
610 | 614 | using Hit_t = decltype(static_cast<Det*>(this)->Det::getHits(probe)); |
611 | 615 | std::string name = static_cast<Det*>(this)->getHitBranchNames(probe++); |
612 | 616 | while (name.size() > 0) { |
@@ -697,8 +701,7 @@ class DetImpl : public o2::base::Detector |
697 | 701 | static_cast<Det*>(this)->Det::createHitBuffers(); |
698 | 702 | for (int b = 0; b < NHITBUFFERS; ++b) { |
699 | 703 | auto& instance = o2::utils::ShmManager::Instance(); |
700 | | - mShmBusy[b] = instance.hasSegment() ? (bool*)instance.getmemblock(sizeof(bool)) : new bool; |
701 | | - *mShmBusy[b] = false; |
| 704 | + mShmBusy[b] = instance.hasSegment() ? new (instance.getmemblock(sizeof(ShmBusyFlag))) ShmBusyFlag(false) : new ShmBusyFlag(false); |
702 | 705 | } |
703 | 706 | } |
704 | 707 | mInitialized = true; |
@@ -749,7 +752,7 @@ class DetImpl : public o2::base::Detector |
749 | 752 | static constexpr int NHITBUFFERS = 3; // number of buffers for hits in order to allow async processing |
750 | 753 | // in the hit merger without blocking nor copying the data |
751 | 754 | // (like done in typical data aquisition systems) |
752 | | - bool* mShmBusy[NHITBUFFERS] = {nullptr}; //! pointer to bool in shared mem indicating of IO busy |
| 755 | + ShmBusyFlag* mShmBusy[NHITBUFFERS] = {nullptr}; //! pointer to flag in shared mem indicating of IO busy |
753 | 756 | std::vector<void*> mCachedPtr[NHITBUFFERS]; |
754 | 757 | int mCurrentBuffer = 0; // holding the current buffer information |
755 | 758 | int mInitialized = false; |
|
0 commit comments