Index: /trunk/FACT++/src/EventBuilderWrapper.h =================================================================== --- /trunk/FACT++/src/EventBuilderWrapper.h (revision 15513) +++ /trunk/FACT++/src/EventBuilderWrapper.h (revision 15514) @@ -12,4 +12,5 @@ #include #include +#include #include "DimWriteStatistics.h" @@ -26,4 +27,6 @@ #include "DataWriteFits2.h" +#include "queue.h" + namespace ba = boost::asio; namespace bs = boost::system; @@ -55,15 +58,4 @@ private: boost::thread fThreadMain; - boost::thread fThreadDimQueue1; - boost::thread fThreadDimQueue2; - - bool fThreadDimQueueStop; - - deque> fDimQueue1; - deque> fDimQueue2; - - mutex fMutexDimQueue1; - mutex fMutexDimQueue2; - enum CommandStates_t // g_runStat @@ -117,4 +109,7 @@ DimDescribedService fDimIncomplete; + Queue> fDimQueue1; + Queue> fDimQueue2; + bool fDebugStream; bool fDebugRead; @@ -188,5 +183,4 @@ public: EventBuilderWrapper(MessageImp &imp) : fMsg(imp), - fThreadDimQueueStop(false), fFileFormat(FAD::kNone), fMaxRun(0), fLastOpened(0), fLastClosed(0), fDimWriteStats ("FAD_CONTROL", imp), @@ -272,4 +266,7 @@ fDimFileFormat("FAD_CONTROL/FILE_FORMAT", "S:1", "|format[int]:Current file format"), fDimIncomplete("FAD_CONTROL/INCOMPLETE", "X:1", "|incomplete[bits]:One bit per board"), + // It is important to instantiate them after the DimServices + fDimQueue1(std::bind(&EventBuilderWrapper::factStatSend, this, placeholders::_1)), + fDimQueue2(std::bind(&EventBuilderWrapper::procHeader, this, placeholders::_1)), fDebugStream(false), fDebugRead(false), fDebugLog(false), fNightAsInt(0) { @@ -284,7 +281,4 @@ for (size_t i=0; i<40; i++) ConnectSlot(i, tcp::endpoint()); - - fThreadDimQueue1 = boost::thread(boost::bind(&EventBuilderWrapper::SendStat, this)); - fThreadDimQueue2 = boost::thread(boost::bind(&EventBuilderWrapper::ProcHeader, this)); } @@ -292,11 +286,8 @@ { Abort(); - fThreadDimQueueStop = true; // FIXME: Used timed_join and abort afterwards // What's the maximum time the eb need to abort? fThreadMain.join(); - fThreadDimQueue1.join(); - fThreadDimQueue2.join(); //ffMsg.Info("EventBuilder stopped."); @@ -367,5 +358,5 @@ g_runStat = kModeRun; - g_maxProc = 3; + g_maxProc = 1; fMsg.Message("Starting EventBuilder thread"); @@ -1104,8 +1095,9 @@ case 0: SendRawData(fadhd, event); - return 1; + return 100; + /* case 1: SendFeedbackData(fadhd, event); - return 2; + return 2;*/ } return 100; @@ -1261,29 +1253,13 @@ } - void SendStat() - { - // Performance could be increased signaling the thread to wake up - while (!fThreadDimQueueStop) - { - if (fDimQueue1.size()==0) - { - usleep(1000); - continue; - } - - fMutexDimQueue1.lock(); - const pair stat = fDimQueue1.front(); - fDimQueue1.pop_front(); - fMutexDimQueue1.unlock(); - - fDimStatistics1.setData(&stat.second, sizeof(GUI_STAT)); - fDimStatistics1.Update(stat.first); - } + void factStatSend(const pair &stat) + { + fDimStatistics1.setData(&stat.second, sizeof(GUI_STAT)); + fDimStatistics1.Update(stat.first); } void factStat(const GUI_STAT &stat) { - const lock_guard guard(fMutexDimQueue1); - fDimQueue1.push_back(make_pair(Time(), stat)); + fDimQueue1.post(make_pair(Time(), stat)); } @@ -1404,143 +1380,127 @@ vector fNumConnected; - void ProcHeader() - { - // Performance could be increased signaling the thread to wake up - while (!fThreadDimQueueStop) - { - if (fDimQueue2.size()==0) - { - usleep(100); - continue; - } - - fMutexDimQueue2.lock(); - const auto dat = fDimQueue2.front(); - fDimQueue2.pop_front(); - fMutexDimQueue2.unlock(); - - const Time &t = get<0>(dat); - const bool changed = get<1>(dat); - const FAD::EventHeader &h = get<2>(dat); - - const FAD::EventHeader old = fVecHeader[h.Id()]; - fVecHeader[h.Id()] = h; - - if (old.fVersion != h.fVersion || changed) - { - const array ver = Compare(&fVecHeader[0], &fVecHeader[0].fVersion); - - array data; - for (int i=0; i<42; i++) + void procHeader(const tuple &dat) + { + const Time &t = get<0>(dat); + const bool changed = get<1>(dat); + const FAD::EventHeader &h = get<2>(dat); + + const FAD::EventHeader old = fVecHeader[h.Id()]; + fVecHeader[h.Id()] = h; + + if (old.fVersion != h.fVersion || changed) + { + const array ver = Compare(&fVecHeader[0], &fVecHeader[0].fVersion); + + array data; + for (int i=0; i<42; i++) + { + ostringstream str; + str << (ver[i]>>8) << '.' << (ver[i]&0xff); + data[i] = stof(str.str()); + } + Update(fDimFwVersion, data, t); + } + + if (old.fRunNumber != h.fRunNumber || changed) + { + const array run = Compare(&fVecHeader[0], &fVecHeader[0].fRunNumber); + fDimRunNumber.setData(&run[0], 42*sizeof(uint32_t)); + fDimRunNumber.Update(t); + } + + if (old.fTriggerGeneratorPrescaler != h.fTriggerGeneratorPrescaler || changed) + { + const array pre = Compare(&fVecHeader[0], &fVecHeader[0].fTriggerGeneratorPrescaler); + fDimPrescaler.setData(&pre[0], 42*sizeof(uint16_t)); + fDimPrescaler.Update(t); + } + + if (old.fDNA != h.fDNA || changed) + { + const array dna = Compare(&fVecHeader[0], &fVecHeader[0].fDNA); + Update(fDimDNA, dna, t, 40); + } + + if (old.fStatus != h.fStatus || changed) + { + const array sts = CompareBits(&fVecHeader[0], &fVecHeader[0].fStatus); + Update(fDimStatus, sts, t); + } + + if (memcmp(old.fDac, h.fDac, sizeof(h.fDac)) || changed) + { + array dacs; + + for (int i=0; i dac = Compare(&fVecHeader[0], &fVecHeader[0].fDac[i]); + memcpy(&dacs[i*42], &dac[0], sizeof(uint16_t)*42); + } + + Update(fDimDac, dacs, t); + } + + // ----------- + + static Time oldt(boost::date_time::neg_infin); + Time newt; + + if (newt>oldt+boost::posix_time::seconds(1)) + { + oldt = newt; + + // --- RefClock + + const array clk = Compare(&fVecHeader[0], &fVecHeader[0].fFreqRefClock); + Update(fDimRefClock, clk, t); + + // --- Temperatures + + const array tmp[4] = + { + Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[0]), // 0-39:val, 40:min, 41:max + Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[1]), // 0-39:val, 40:min, 41:max + Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[2]), // 0-39:val, 40:min, 41:max + Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[3]) // 0-39:val, 40:min, 41:max + }; + + vector data; + data.reserve(82); + data.push_back(tmp[0][40]); // min: 0 + data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 1-40 + data.push_back(tmp[0][41]); // max: 41 + data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 42-81 + + for (int j=1; j<=3; j++) + { + const array &ref = tmp[j]; + + // Gloabl min + if (ref[40]data[41]) // 41=max + data[41] = ref[41]; + + for (int i=0; i<40; i++) { - ostringstream str; - str << (ver[i]>>8) << '.' << (ver[i]&0xff); - data[i] = stof(str.str()); + // min per board + if (ref[i]data[i+42]) // data: 42-81 + data[i+42] = ref[i]; // ref: 0-39 } - Update(fDimFwVersion, data, t); - } - - if (old.fRunNumber != h.fRunNumber || changed) - { - const array run = Compare(&fVecHeader[0], &fVecHeader[0].fRunNumber); - fDimRunNumber.setData(&run[0], 42*sizeof(uint32_t)); - fDimRunNumber.Update(t); - } - - if (old.fTriggerGeneratorPrescaler != h.fTriggerGeneratorPrescaler || changed) - { - const array pre = Compare(&fVecHeader[0], &fVecHeader[0].fTriggerGeneratorPrescaler); - fDimPrescaler.setData(&pre[0], 42*sizeof(uint16_t)); - fDimPrescaler.Update(t); - } - - if (old.fDNA != h.fDNA || changed) - { - const array dna = Compare(&fVecHeader[0], &fVecHeader[0].fDNA); - Update(fDimDNA, dna, t, 40); - } - - if (old.fStatus != h.fStatus || changed) - { - const array sts = CompareBits(&fVecHeader[0], &fVecHeader[0].fStatus); - Update(fDimStatus, sts, t); - } - - if (memcmp(old.fDac, h.fDac, sizeof(h.fDac)) || changed) - { - array dacs; - - for (int i=0; i dac = Compare(&fVecHeader[0], &fVecHeader[0].fDac[i]); - memcpy(&dacs[i*42], &dac[0], sizeof(uint16_t)*42); - } - - Update(fDimDac, dacs, t); - } - - // ----------- - - static Time oldt(boost::date_time::neg_infin); - Time newt; - - if (newt>oldt+boost::posix_time::seconds(1)) - { - oldt = newt; - - // --- RefClock - - const array clk = Compare(&fVecHeader[0], &fVecHeader[0].fFreqRefClock); - Update(fDimRefClock, clk, t); - - // --- Temperatures - - const array tmp[4] = - { - Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[0]), // 0-39:val, 40:min, 41:max - Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[1]), // 0-39:val, 40:min, 41:max - Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[2]), // 0-39:val, 40:min, 41:max - Compare(&fVecHeader[0], &fVecHeader[0].fTempDrs[3]) // 0-39:val, 40:min, 41:max - }; - - vector data; - data.reserve(82); - data.push_back(tmp[0][40]); // min: 0 - data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 1-40 - data.push_back(tmp[0][41]); // max: 41 - data.insert(data.end(), tmp[0].data(), tmp[0].data()+40); // val: 42-81 - - for (int j=1; j<=3; j++) - { - const array &ref = tmp[j]; - - // Gloabl min - if (ref[40]data[41]) // 41=max - data[41] = ref[41]; - - for (int i=0; i<40; i++) - { - // min per board - if (ref[i]data[i+42]) // data: 42-81 - data[i+42] = ref[i]; // ref: 0-39 - } - } - - vector deg(82); // 0: global min, 1-40: min - for (int i=0; i<82; i++) // 41: global max, 42-81: max - deg[i] = data[i]/16.; - - fDimTemperature.setData(deg.data(), 82*sizeof(float)); - fDimTemperature.Update(t); - } - + } + + vector deg(82); // 0: global min, 1-40: min + for (int i=0; i<82; i++) // 41: global max, 42-81: max + deg[i] = data[i]/16.; + + fDimTemperature.setData(deg.data(), 82*sizeof(float)); + fDimTemperature.Update(t); } } @@ -1561,6 +1521,8 @@ fNumConnected = con; - const lock_guard guard(fMutexDimQueue2); - fDimQueue2.push_back(make_tuple(Time(), changed, h)); + fDimQueue2.post(make_tuple(Time(), changed, h)); + + //const lock_guard guard(fMutexDimQueue2); + //fDimQueue2.push_back(make_tuple(Time(), changed, h)); } };