#ifndef FACT_EventBuilderWrapper #define FACT_EventBuilderWrapper #include #if BOOST_VERSION < 104400 #if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ > 4)) #undef BOOST_HAS_RVALUE_REFS #endif #endif #include #include #include #include #include "DimWriteStatistics.h" #include "DataCalib.h" #include "DataWriteRaw.h" #ifdef HAVE_FITS #include "DataWriteFits.h" #else #define DataWriteFits DataWriteFits2 #endif #include "DataWriteFits2.h" #include "queue.h" namespace ba = boost::asio; namespace bs = boost::system; namespace fs = boost::filesystem; using ba::ip::tcp; using namespace std; // ======================================================================== #include "EventBuilder.h" extern "C" { extern void StartEvtBuild(); extern int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt); } // ======================================================================== class EventBuilderWrapper { public: // FIXME static EventBuilderWrapper *This; MessageImp &fMsg; private: boost::thread fThreadMain; enum CommandStates_t // g_runStat { kAbort = -2, // quit as soon as possible ('abort') kExit = -1, // stop reading, quit when buffered events done ('exit') kInitialize = 0, // 'initialize' (e.g. dim not yet started) kHybernate = 1, // do nothing for long time ('hybernate') [wakeup within ~1sec] kSleep = 2, // do nothing ('sleep') [wakeup within ~10msec] kModeFlush = 10, // read data from camera, but skip them ('flush') kModeTest = 20, // read data and process them, but do not write to disk ('test') kModeFlag = 30, // read data, process and write all to disk ('flag') kModeRun = 40, // read data, process and write selected to disk ('run') }; enum { kCurrent = 0, kTotal = 1, kEventId = 2, kTriggerId = 3, }; FAD::FileFormat_t fFileFormat; uint32_t fMaxRun; uint32_t fLastOpened; uint32_t fLastClosed; array fNumEvts; DimWriteStatistics fDimWriteStats; DimDescribedService fDimRuns; DimDescribedService fDimEvents; DimDescribedService fDimRawData; DimDescribedService fDimEventData; DimDescribedService fDimFeedbackData; DimDescribedService fDimFwVersion; DimDescribedService fDimRunNumber; DimDescribedService fDimStatus; DimDescribedService fDimDNA; DimDescribedService fDimTemperature; DimDescribedService fDimPrescaler; DimDescribedService fDimRefClock; DimDescribedService fDimRoi; DimDescribedService fDimDac; DimDescribedService fDimDrsRuns; DimDescribedService fDimDrsCalibration; DimDescribedService fDimStatistics1; //DimDescribedService fDimStatistics2; DimDescribedService fDimFileFormat; DimDescribedService fDimIncomplete; Queue> fDimQueue1; Queue> fDimQueue2; Queue>> fDimQueue3; bool fDebugStream; bool fDebugRead; bool fDebugLog; string fPath; uint64_t fNightAsInt; uint32_t fRunNumber; protected: bool InitRunNumber(const string &path="") { if (!path.empty()) { if (!DimWriteStatistics::DoesPathExist(path, fMsg)) { fMsg.Error("Data path "+path+" does not exist!"); return false; } fPath = path; fDimWriteStats.SetCurrentFolder(fPath); fMsg.Info("Data path set to "+path+"."); } // Get current night const uint64_t night = Time().NightAsInt(); if (night==fNightAsInt) return true; // Check for run numbers fRunNumber = 1000; while (--fRunNumber>0) { const string name = DataProcessorImp::FormFileName(fPath, night, fRunNumber, ""); if (access((name+"bin").c_str(), F_OK) == 0) break; if (access((name+"fits").c_str(), F_OK) == 0) break; if (access((name+"drs.fits").c_str(), F_OK) == 0) break; } // This is now the first file which does not exist fRunNumber++; fLastOpened = 0; // Check if we have exceeded the maximum if (fRunNumber==1000) { fMsg.Error("You have a file with run-number 1000 in "+fPath+" ["+to_string(night)+"]"); return false; } ostringstream str; if (fNightAsInt==0) str << "First night..."; else str << "Night has changd from " << fNightAsInt << "... new"; str << " run-number is " << night << "-" << setfill('0') << setw(3) << fRunNumber << " [" << (fPath.empty()?".":fPath) << "]"; fMsg.Message(str); fNightAsInt = night; return true; } public: EventBuilderWrapper(MessageImp &imp) : fMsg(imp), fFileFormat(FAD::kNone), fMaxRun(0), fLastOpened(0), fLastClosed(0), fDimWriteStats ("FAD_CONTROL", imp), fDimRuns ("FAD_CONTROL/RUNS", "I:5;C", "Run files statistics" "|stats[int]:num of open files, min/max run no, last opened or closed run" "|file[string]:filename of last opened file"), fDimEvents ("FAD_CONTROL/EVENTS", "I:4", "Event counts" "|evtsCount[int]:Num evts cur. run, total (all run), evt ID, trig. Num"), fDimRawData ("FAD_CONTROL/RAW_DATA", "S:1;S:1;I:1;I:1;S:1;I:1;C:4;I:1;I:2;I:40;S:1440;S:160;F", "|roi[uint16]:number of samples per pixel" "|roi_tm[uint16]:number of samples per time-marker channel" "|num_fad[uint32]:event number from FADs" "|num_ftm[uint32]:trigger number from FTM" "|type[uint16]:trigger type from FTM" "|num_boards[uint32]:number of active boards" "|error[uint8]:event builder error counters" "|dummy[]:" "|time[uint32]:PC time as unix time stamp" "|time_board[uint32]:Time stamp of FAD boards" "|start_pix[int16]:start sample of pixels" "|start_tm[int16]:start sample of time marker channels" "|adc[int16]:adc data"), fDimEventData ("FAD_CONTROL/EVENT_DATA", "F:1440;F:1440;F:1440;F:1440", "|avg:|rms:|max:|pos"), fDimFeedbackData("FAD_CONTROL/FEEDBACK_DATA", "F:1440", ""), fDimFwVersion ("FAD_CONTROL/FIRMWARE_VERSION", "F:42", "Firmware version number of fad boards" "|firmware[float]:Version number of firmware, for each board. 40=min, 41=max"), fDimRunNumber ("FAD_CONTROL/RUN_NUMBER", "I:42", "Run numbers coming from FAD boards" "|runNumbers[int]:current run number of each FAD board. 40=min, 41=max"), fDimStatus ("FAD_CONTROL/STATUS", "S:42", "Status of FAD boards" "|status[bitpattern]:Status of each FAD board. Maybe buggy"), fDimDNA ("FAD_CONTROL/DNA", "X:40", "DNA of FAD boards" "|DNA[hex]:Hex identifier of each FAD board"), fDimTemperature ("FAD_CONTROL/TEMPERATURE", "F:82", "FADs temperatures" "|temp[deg. C]:0 global min, 1-40 min, 41 global max, 42-81 max"), fDimPrescaler ("FAD_CONTROL/PRESCALER", "S:42", "Trigger generator prescaler of fad boards" "|prescaler[int]:Trigger generator prescaler value, for each board"), fDimRefClock ("FAD_CONTROL/REFERENCE_CLOCK", "I:42", "Reference clock of FAD boards" "|refClocks[t]:ref clocks of FAD boards. 40=min, 41=max"), fDimRoi ("FAD_CONTROL/REGION_OF_INTEREST", "S:2", "roi:|roi_rm:"), fDimDac ("FAD_CONTROL/DAC", "S:336", "DAC settings of each FAD board" "|DAC[int]:DAC counts, sequentially DAC 0 board 0, 0/1, 0/2... (plus min max)"), fDimDrsRuns ("FAD_CONTROL/DRS_RUNS", "I:1;I:3", "|roi:Region of interest of secondary baseline" "|run:Run numbers of DRS runs (0=none)"), fDimDrsCalibration("FAD_CONTROL/DRS_CALIBRATION", "I:1;I:3;F:1474560;F:1474560;F:1474560;F:1474560;F:1474560;F:1474560;F:163840;F:163840", "|roi:Region of interest of secondary baseline" "|run:Run numbers of DRS runs (0=none)"), fDimStatistics1 ("FAD_CONTROL/STATISTICS1", "I:3;I:5;X:4;I:3;I:3;I:40;I:1;I:2;C:40;I:40;X:40", "Event Builder status for GUI display" "|threadInfo[int]:Number of read, proc and writes" "|bufferInfo[int]:Events in buffer, incomp., comp., tot., max past cycle, total" "|memInfo[int]:total buf. mem, used mem, max used, max past cycle" "|EvtCnt[int]:Number of events skipped, written, with errors" "|badRoi[int]:Num boards with wrong ROI in event, run or board" "|badRoiBoard[int]:Num boards with wrong ROI" "|deltaT[ms]:Time in ms for rates" "|rateNew[int]:Number of new start events received" "|numConn[int]:Number of connections per board" "|rateBytes[int]:Bytes read this cycle" "|totBytes[int]:Bytes read (counter)"), /*fDimStatistics2 ("FAD_CONTROL/STATISTICS2", "I:1;I:280;X:40;I:40;I:4;I:4;I:2;I:2;I:3;C:40", "Event Builder status, events oriented" "|reset[int]:If increased, reset all counters" "|numRead[int]:How often sucessful read from N sockets per loop" "|gotByte[int]:number of bytes read per board" "|gotErr[int]:number of com. errors per board" "|evtStat[int]:number of evts read, completed, with errors, incomplete" "|procStat[int]:num. of evts proc., w probs, acc. or rej. by SW trigger" "|feedStat[int]:number of evts used or rejected by feedback system" "|wrtStat[int]:number of evts written to disk, with errors" "|runStat[int]:number of run opened, closed, with open or close errors" "|numConn[int]:number of sockets successfully opened per board"),*/ 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)), fDimQueue3(std::bind(&EventBuilderWrapper::updateEvents, this, placeholders::_1)), fDebugStream(false), fDebugRead(false), fDebugLog(false), fNightAsInt(0) { if (This) throw logic_error("EventBuilderWrapper cannot be instantiated twice."); This = this; memset(fNumEvts.data(), 0, sizeof(fNumEvts)); fDimEvents.Update(fNumEvts); for (size_t i=0; i<40; i++) ConnectSlot(i, tcp::endpoint()); } virtual ~EventBuilderWrapper() { Abort(); // FIXME: Used timed_join and abort afterwards // What's the maximum time the eb need to abort? fThreadMain.join(); //ffMsg.Info("EventBuilder stopped."); for (vector::iterator it=fFiles.begin(); it!=fFiles.end(); it++) delete *it; } set fIsRunStarted; map fExpectedRuns; uint32_t StartNewRun(int64_t maxtime, int64_t maxevt, const pair &ref) { if (maxtime<=0 || maxtime>24*60*60) maxtime = 24*60*60; if (maxevt<=0 || maxevt>INT32_MAX) maxevt = INT32_MAX; const FAD::RunDescription descr = { uint32_t(maxtime), uint32_t(maxevt), ref.first, ref.second, }; if (!InitRunNumber()) return 0; // FIMXE: Maybe reset an event counter so that the mcp can count events? //fMsg.Info(" ==> TODO: Set a limit on the size of fExpectedRuns!"); fExpectedRuns[fRunNumber] = descr; fIsRunStarted.insert(fRunNumber); return fRunNumber++; } bool IsThreadRunning() { return !fThreadMain.timed_join(boost::posix_time::microseconds(0)); } void SetMaxMemory(unsigned int mb) const { /* if (mb*1000000 &addr) { if (IsThreadRunning()) { fMsg.Warn("Start - EventBuilder still running"); return; } fLastMessage.clear(); for (size_t i=0; i<40; i++) ConnectSlot(i, addr[i]); g_runStat = kModeRun; g_maxProc = 1; fMsg.Message("Starting EventBuilder thread"); fThreadMain = boost::thread(StartEvtBuild); } void ConnectSlot(unsigned int i, const tcp::endpoint &addr) { if (i>39) return; if (addr==tcp::endpoint()) { DisconnectSlot(i); return; } struct sockaddr_in sockaddr; //IP for each socket sockaddr.sin_family = AF_INET; sockaddr.sin_addr.s_addr = htonl(addr.address().to_v4().to_ulong()); sockaddr.sin_port = htons(addr.port()); memcpy(&g_port[i].sockAddr, &sockaddr, sizeof(struct sockaddr_in)); // In this order g_port[i].sockDef = 1; fDimIncomplete.setQuality(0); fDimIncomplete.Update(uint64_t(0)); } void DisconnectSlot(unsigned int i) { if (i>39) return; g_port[i].sockDef = 0; // In this order struct sockaddr_in addr; //IP for each socket addr.sin_family = AF_INET; addr.sin_addr.s_addr = 0; addr.sin_port = 0; memcpy(&g_port[i].sockAddr, &addr, sizeof(struct sockaddr_in)); fDimIncomplete.setQuality(0); fDimIncomplete.Update(uint64_t(0)); } void IgnoreSlot(unsigned int i) { if (i>39) return; if (g_port[i].sockAddr.sin_port==0) return; g_port[i].sockDef = -1; } void Abort() { fMsg.Message("Signal abort to EventBuilder thread..."); g_runStat = kAbort; } void ResetThread(bool soft) { /* if (g_reset > 0) * suspend reading * reset = g_reset; * g_reset=0 * reset% 10 == 0 leave event Buffers as they are == 1 let all buffers drain (write (incomplete) events) > 1 flush all buffers (do not write buffered events) * (reset/10)%10 > 0 close all sockets and destroy them (also free the allocated read-buffers) recreate before resuming operation [ this is more than just close/open that can be triggered by e.g. close/open the base-socket ] * (reset/100)%10 > 0 close all open run-files * (reset/1000) sleep so many seconds before resuming operation (does not (yet) take into account time left when waiting for buffers getting empty ...) * resume_reading */ fMsg.Message("Signal reset to EventBuilder thread..."); g_reset = soft ? 101 : 102; } void Exit() { fMsg.Message("Signal exit to EventBuilder thread..."); g_runStat = kExit; } /* void Wait() { fThread.join(); ffMsg.Message("EventBuilder stopped."); }*/ void Hybernate() const { g_runStat = kHybernate; } void Sleep() const { g_runStat = kSleep; } void FlushMode() const { g_runStat = kModeFlush; } void TestMode() const { g_runStat = kModeTest; } void FlagMode() const { g_runStat = kModeFlag; } void RunMode() const { g_runStat = kModeRun; } // FIXME: To be removed //void SetMode(int mode) const { g_runStat = mode; } bool IsConnected(int i) const { return gi_NumConnect[i]==7; } bool IsConnecting(int i) const { return !IsConnected(i) && !IsDisconnected(i); } bool IsDisconnected(int i) const { return gi_NumConnect[i]<=0 && g_port[i].sockDef==0; } int GetNumConnected(int i) const { return gi_NumConnect[i]; } int GetNumFilesOpen() const { return fFiles.size(); } /* bool IsConnected(int i) const { return gi_NumConnect[i]>0; } bool IsConnecting(int i) const { return !IsConnected(i) && !IsDisconnected(i); } bool IsDisconnected(int i) const { return gi_NumConnect[i]<=0 && g_port[i].sockDef==0; } int GetNumConnected(int i) const { return gi_NumConnect[i]; } */ void SetIgnore(int i, bool b) const { if (g_port[i].sockDef!=0) g_port[i].sockDef=b?-1:1; } bool IsIgnored(int i) const { return g_port[i].sockDef==-1; } void SetOutputFormat(FAD::FileFormat_t f) { const bool changed = f!=fFileFormat; fFileFormat = f; fDimFileFormat.Update(uint16_t(f)); string msg = "File format set to: "; switch (f) { case FAD::kNone: msg += "kNone."; break; case FAD::kDebug: msg += "kDebug."; break; case FAD::kFits: msg += "kFits."; break; case FAD::kCfitsio: msg += "kCfitsio"; break; case FAD::kRaw: msg += "kRaw"; break; case FAD::kCalib: DataCalib::Restart(); DataCalib::Update(fDimDrsCalibration, fDimDrsRuns); fMsg.Message("Resetted DRS calibration."); return; } if (changed) fMsg.Message(msg); } virtual int ResetSecondaryDrsBaseline() { if (DataCalib::ResetTrgOff(fDimDrsCalibration, fDimDrsRuns)) { fFileFormat = FAD::kCalib; fDimFileFormat.Update(uint16_t(fFileFormat)); fMsg.Message("Resetted DRS calibration for secondary baseline."); } else fMsg.Warn("Could not reset DRS calibration of secondary baseline."); return 0; } void SetDebugLog(bool b) { fDebugLog = b; } void SetDebugStream(bool b) { fDebugStream = b; if (b) return; for (int i=0; i<40; i++) { if (!fDumpStream[i].is_open()) continue; fDumpStream[i].close(); ostringstream name; name << "socket_dump-" << setfill('0') << setw(2) << i << ".bin"; fMsg.Message("Closed file '"+name.str()+"'"); } } void SetDebugRead(bool b) { fDebugRead = b; if (b || !fDumpRead.is_open()) return; fDumpRead.close(); fMsg.Message("Closed file 'socket_events.txt'"); } // size_t GetUsedMemory() const { return gi_usedMem; } void LoadDrsCalibration(const char *fname) { if (!DataCalib::ReadFits(fname, fMsg)) return; fMsg.Info("Successfully loaded DRS calibration from "+string(fname)); DataCalib::Update(fDimDrsCalibration, fDimDrsRuns); } virtual int CloseOpenFiles() { CloseRunFile(0, 0, 0); return 0; } /* struct OpenFileToDim { int code; char fileName[FILENAME_MAX]; }; SignalRunOpened(runid, filename); // Send num open files // Send runid, (more info about the run?), filename via dim SignalEvtWritten(runid); // Send num events written of newest file SignalRunClose(runid); // Send new num open files // Send empty file-name if no file is open */ // -------------- Mapped event builder callbacks ------------------ void UpdateRuns(const string &fname="") { uint32_t values[5] = { static_cast(fFiles.size()), 0xffffffff, 0, fLastOpened, fLastClosed }; for (vector::const_iterator it=fFiles.begin(); it!=fFiles.end(); it++) { const DataProcessorImp *file = *it; if (file->GetRunId()GetRunId(); if (file->GetRunId()>values[2]) values[2] = file->GetRunId(); } fMaxRun = values[2]; vector data(sizeof(values)+fname.size()+1); memcpy(data.data(), values, sizeof(values)); strcpy(data.data()+sizeof(values), fname.c_str()); fDimRuns.Update(data); } vector fFiles; void updateEvents(const pair> &stat) { fDimEvents.setData(stat.second.data(), sizeof(uint32_t)*4); fDimEvents.Update(stat.first); } FileHandle_t runOpen(uint32_t runid, RUN_HEAD *h, size_t) { //fMsg.Info(" ==> TODO: Update run configuration in database!"); map::iterator it = fExpectedRuns.begin(); while (it!=fExpectedRuns.end()) { if (it->firstfirst << "."; fMsg.Info(str); fExpectedRuns.erase(it++); continue; } if (it->first==runid) break; it++; } FAD::RunDescription desc; if (it==fExpectedRuns.end()) { ostringstream str; str << "runOpen - Run " << runid << " wasn't expected (maybe manual triggers)"; fMsg.Warn(str); } else { desc = it->second; fExpectedRuns.erase(it); } // Check if file already exists... DataProcessorImp *file = 0; switch (fFileFormat) { case FAD::kNone: file = new DataDump(fPath, fNightAsInt, runid, fMsg); break; case FAD::kDebug: file = new DataDebug(fPath, fNightAsInt, runid, fMsg); break; case FAD::kCfitsio: file = new DataWriteFits(fPath, fNightAsInt, runid, fMsg); break; case FAD::kFits: file = new DataWriteFits2(fPath, fNightAsInt, runid, fMsg); break; case FAD::kRaw: file = new DataWriteRaw(fPath, fNightAsInt, runid, fMsg); break; case FAD::kCalib: file = new DataCalib(fPath, fNightAsInt, runid, fDimDrsCalibration, fDimDrsRuns, fMsg); break; } try { if (!file->Open(h, desc)) return 0; } catch (const exception &e) { fMsg.Error("Exception trying to open file: "+string(e.what())); return 0; } fFiles.push_back(file); ostringstream str; str << "Opened: " << file->GetFileName() << " (" << file->GetRunId() << ")"; fMsg.Info(str); fDimWriteStats.FileOpened(file->GetFileName()); fLastOpened = runid; UpdateRuns(file->GetFileName()); fNumEvts[kEventId] = 0; fNumEvts[kTriggerId] = 0; fNumEvts[kCurrent] = 0; fDimQueue3.post(make_pair(Time(), fNumEvts)); // fDimCurrentEvent.Update(uint32_t(0)); return reinterpret_cast(file); } int runWrite(FileHandle_t handler, EVENT *e, size_t /*sz*/) { DataProcessorImp *file = reinterpret_cast(handler); if (!file->WriteEvt(e)) return -1; if (file->GetRunId()==fMaxRun) { fNumEvts[kCurrent]++; fNumEvts[kEventId] = e->EventNum; fNumEvts[kTriggerId] = e->TriggerNum; } fNumEvts[kTotal]++; static Time oldt(boost::date_time::neg_infin); Time newt; if (newt>oldt+boost::posix_time::seconds(1)) { fDimQueue3.post(make_pair(Time(), fNumEvts)); //fDimEvents.Update(fNumEvts); oldt = newt; } // ===> SignalEvtWritten(runid); // Send num events written of newest file /* close run runId (all all runs if runId=0) */ /* return: 0=close scheduled / >0 already closed / <0 does not exist */ //CloseRunFile(file->GetRunId(), time(NULL)+2) ; return 0; } virtual void CloseRun(uint32_t /*runid*/) { } int runClose(FileHandle_t handler, RUN_TAIL *tail, size_t) { //fMsg.Info(" ==> TODO: Update run configuration in database!"); DataProcessorImp *file = reinterpret_cast(handler); const vector::iterator it = find(fFiles.begin(), fFiles.end(), file); if (it==fFiles.end()) { ostringstream str; str << "File handler (" << handler << ") requested to close by event builder doesn't exist."; fMsg.Fatal(str); return -1; } /* fFiles.erase(it); fLastClosed = file->GetRunId(); CloseRun(fLastClosed); UpdateRuns(); fDimEvents.Update(fNumEvts); */ const bool rc = file->Close(tail); if (!rc) { // Error message } // Note that this is the signal for the fadctrl to change from // WritingData back to Connected. If this is done too early, // a new run might be started before this is closed. This is // faster, but leads to problems with the DRS calibration // if the system is fast enough to start the new run before // this one has really been closed. fFiles.erase(it); fLastClosed = file->GetRunId(); CloseRun(fLastClosed); UpdateRuns(); fDimQueue3.post(make_pair(Time(),fNumEvts)); //fDimEvents.Update(fNumEvts); ostringstream str; str << "Closed: " << file->GetFileName() << " (" << file->GetRunId() << ")"; fMsg.Info(str); delete file; // ==> SignalRunClose(runid); // Send new num open files // Send empty file-name if no file is open return rc ? 0 : -1; } ofstream fDumpStream[40]; void debugStream(int isock, void *buf, int len) { if (!fDebugStream) return; const int slot = isock/7; if (slot<0 || slot>39) return; if (!fDumpStream[slot].is_open()) { ostringstream name; name << "socket_dump-" << setfill('0') << setw(2) << slot << ".bin"; fDumpStream[slot].open(name.str().c_str(), ios::app); if (!fDumpStream[slot]) { ostringstream str; str << "Open file '" << name << "': " << strerror(errno) << " (errno=" << errno << ")"; fMsg.Error(str); return; } fMsg.Message("Opened file '"+name.str()+"' for writing."); } fDumpStream[slot].write(reinterpret_cast(buf), len); } ofstream fDumpRead; // Stream to possibly dump docket events void debugRead(int isock, int ibyte, uint32_t event, uint32_t ftmevt, uint32_t runno, int state, uint32_t tsec, uint32_t tusec) { // isock = socketID (0-279) // ibyte = #bytes gelesen // event = eventId (oder 0 wenn noch nicht bekannt) // state : 1=finished reading data // 0=reading data // -1=start reading data (header) // -2=start reading data, // eventId not known yet (too little data) // tsec, tusec = time when reading seconds, microseconds // if (!fDebugRead || ibyte==0) return; if (!fDumpRead.is_open()) { fDumpRead.open("socket_events.txt", ios::app); if (!fDumpRead) { ostringstream str; str << "Open file 'socket_events.txt': " << strerror(errno) << " (errno=" << errno << ")"; fMsg.Error(str); return; } fMsg.Message("Opened file 'socket_events.txt' for writing."); fDumpRead << "# START: " << Time().GetAsStr() << endl; fDumpRead << "# state time_sec time_usec socket slot runno event_id trigger_id bytes_received" << endl; } fDumpRead << setw(2) << state << " " << setw(8) << tsec << " " << setw(9) << tusec << " " << setw(3) << isock << " " << setw(2) << isock/7 << " " << runno << " " << event << " " << ftmevt << " " << ibyte << endl; } array fVecRoi; int eventCheck(uint32_t runNr, PEVNT_HEADER *fadhd, EVENT *event, int /*iboard*/) { /* fadhd[i] ist ein array mit den 40 fad-headers (falls ein board nicht gelesen wurde, ist start_package_flag =0 ) event ist die Struktur, die auch die write routine erhaelt; darin sind im header die 'soll-werte' fuer z.B. eventID als auch die ADC-Werte (falls Du die brauchst) Wenn die routine einen negativen Wert liefert, wird das event geloescht (nicht an die write-routine weitergeleitet [mind. im Prinzip] */ const array roi = {{ event->Roi, event->RoiTM }}; if (roi!=fVecRoi) { Update(fDimRoi, roi); fVecRoi = roi; } const FAD::EventHeader *beg = reinterpret_cast(fadhd); const FAD::EventHeader *end = reinterpret_cast(fadhd)+40; // FIMXE: Compare with target configuration for (const FAD::EventHeader *ptr=beg; ptr!=end; ptr++) { // FIXME: Compare with expectations!!! if (ptr->fStartDelimiter==0) { if (ptr==beg) beg++; continue; } if (beg->fStatus != ptr->fStatus) { fMsg.Error("Inconsistency in FAD status detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } if (beg->fRunNumber != ptr->fRunNumber) { fMsg.Error("Inconsistent run number detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } /* if (beg->fVersion != ptr->fVersion) { Error("Inconsist firmware version detected.... closing run."); CloseRunFile(runNr, 0, 0); break; } */ if (beg->fEventCounter != ptr->fEventCounter) { fMsg.Error("Inconsistent FAD event number detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } if (beg->fTriggerCounter != ptr->fTriggerCounter) { fMsg.Error("Inconsistent FTM trigger number detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } if (beg->fAdcClockPhaseShift != ptr->fAdcClockPhaseShift) { fMsg.Error("Inconsistent phase shift detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } if (memcmp(beg->fDac, ptr->fDac, sizeof(beg->fDac))) { fMsg.Error("Inconsistent DAC values detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } if (beg->fTriggerType != ptr->fTriggerType) { fMsg.Error("Inconsistent trigger type detected.... closing run."); CloseRunFile(runNr, 0, 0); return -1; } } // check REFCLK_frequency // check consistency with command configuration // how to log errors? // need gotNewRun/closedRun to know it is finished return 0; } void SendRawData(PEVNT_HEADER *fadhd, EVENT *event) { // Currently we send any event no matter what its trigger id is... // To be changed. static Time oldt(boost::date_time::neg_infin); Time newt; static int skip = 0; // FIXME: Only send events if the have newer run-numbers if (newt0 ? 100 : 1000)) return; oldt = newt; // Workaround to find a valid header..... const FAD::EventHeader *beg = reinterpret_cast(fadhd); const FAD::EventHeader *end = reinterpret_cast(fadhd)+40; // FIMXE: Compare with target configuration const FAD::EventHeader *ptr=beg; for (; ptr!=end; ptr++) { if (ptr->fStartDelimiter!=0) break; } if (ptr==end) return; vector data(sizeof(EVENT)+event->Roi*sizeof(float)*(1440+160)); memcpy(data.data(), event, sizeof(EVENT)); float *vec = reinterpret_cast(data.data()+sizeof(EVENT)); DataCalib::Apply(vec, event->Adc_Data, event->StartPix, event->Roi); DrsCalibrate::RemoveSpikes(vec, event->Roi); vector data2(1440*4); // Mean, RMS, Max, Pos const double max = DrsCalibrate::GetPixelStats(data2.data(), vec, event->Roi); // Maximum above roughly 5pe if (ptr->IsTriggerPhys() && max<100 && skip<10) { skip++; return; } skip = 0; fDimRawData.setQuality(ptr->fTriggerType); fDimRawData.Update(data); fDimEventData.setQuality(ptr->fTriggerType); fDimEventData.Update(data2); } void SendFeedbackData(PEVNT_HEADER *fadhd, EVENT *event) { /* if (!DataCalib::IsValid()) return; // Workaround to find a valid header..... const FAD::EventHeader *beg = reinterpret_cast(fadhd); const FAD::EventHeader *end = reinterpret_cast(fadhd)+40; // FIMXE: Compare with target configuration const FAD::EventHeader *ptr=beg; for (; ptrfStartDelimiter!=0) break; } if (ptr==end) return; if (!ptr->HasTriggerLPext() && !ptr->HasTriggerLPint()) return; vector data(event->Roi*1440); DataCalib::Apply(data.data(), event->Adc_Data, event->StartPix, event->Roi); DrsCalibrate::RemoveSpikes(data.data(), event->Roi); vector data2(1440); // Mean, RMS, Max, Pos, first, last DrsCalibrate::GetPixelMax(data2.data(), data.data(), event->Roi, 0, event->Roi-1); fDimFeedbackData.Update(data2); */ } int subProcEvt(int threadID, PEVNT_HEADER *fadhd, EVENT *event, int16_t /*iboard*/, void */*buffer*/) { switch (threadID) { case 0: SendRawData(fadhd, event); return 100; /* case 1: SendFeedbackData(fadhd, event); return 2;*/ } return 100; } bool IsRunStarted() const { const set::const_iterator it = fIsRunStarted.find(fRunNumber-1); return it==fIsRunStarted.end();// ? true : it->second.started; } uint32_t GetRunNumber() const { return fRunNumber; } bool IsRunFileOpen() { return fLastOpened==fRunNumber-1; } bool IncreaseRunNumber(uint32_t run) { if (!InitRunNumber()) return false; if (run::iterator it = fIsRunStarted.begin(); while (it!=fIsRunStarted.end()) { if (*it::iterator i2 = fExpectedRuns.find(runnr); if (i2==fExpectedRuns.end()) { ostringstream str; str << "gotNewRun - Run " << runnr << " wasn't expected."; fMsg.Warn(str); return; } CloseRunFile(runnr, time(NULL)+i2->second.maxtime, i2->second.maxevt); // return: 0=close scheduled / >0 already closed / <0 does not exist // FIXME: Move configuration from expected runs to runs which will soon // be opened/closed fIsRunStarted.erase(it); } map fLastMessage; void factOut(int severity, int err, const char *message) { if (!fDebugLog && severity==99) return; ostringstream str; //str << boost::this_thread::get_id() << " "; str << "EventBuilder("; if (err<0) str << "---"; else str << err; str << "): " << message; string &old = fLastMessage[boost::this_thread::get_id()]; if (str.str()==old) return; old = str.str(); fMsg.Update(str, severity); } /* void factStat(int64_t *stat, int len) { if (len!=7) { fMsg.Warn("factStat received unknown number of values."); return; } vector data(1, g_maxMem); data.insert(data.end(), stat, stat+len); static vector last(8); if (data==last) return; last = data; fDimStatistics.Update(data); // len ist die Laenge des arrays. // array[4] enthaelt wieviele bytes im Buffer aktuell belegt sind; daran // kannst Du pruefen, ob die 100MB voll sind .... ostringstream str; str << "Wait=" << stat[0] << " " << "Skip=" << stat[1] << " " << "Del=" << stat[2] << " " << "Tot=" << stat[3] << " " << "Mem=" << stat[4] << "/" << g_maxMem << " " << "Read=" << stat[5] << " " << "Conn=" << stat[6]; fMsg.Info(str); } */ void factStat(const EVT_STAT &/*stat*/) { //fDimStatistics2.Update(stat); } void factStatSend(const pair &stat) { fDimStatistics1.setData(&stat.second, sizeof(GUI_STAT)); fDimStatistics1.Update(stat.first); } void factStat(const GUI_STAT &stat) { fDimQueue1.post(make_pair(Time(), stat)); } void factReportIncomplete(uint64_t rep) { fDimIncomplete.setQuality(1); fDimIncomplete.Update(rep); } array fVecHeader; template array Compare(const S *vec, const T *t) { const int offset = reinterpret_cast(t) - reinterpret_cast(vec); const T *min = NULL; const T *val = NULL; const T *max = NULL; array arr; // bool rc = true; for (int i=0; i<40; i++) { const char *base = reinterpret_cast(vec+i); const T *ref = reinterpret_cast(base+offset); arr[i] = *ref; if (gi_NumConnect[i]!=7) { arr[i] = 0; continue; } if (!val) { min = ref; val = ref; max = ref; } if (*ref<*min) min = ref; if (*ref>*max) max = ref; // if (*val!=*ref) // rc = false; } arr[40] = val ? *min : 1; arr[41] = val ? *max : 0; return arr; } template array CompareBits(const FAD::EventHeader *h, const T *t) { const int offset = reinterpret_cast(t) - reinterpret_cast(h); T val = 0; T rc = 0; array vec; bool first = true; for (int i=0; i<40; i++) { const char *base = reinterpret_cast(&fVecHeader[i]); const T *ref = reinterpret_cast(base+offset); vec[i+2] = *ref; if (gi_NumConnect[i]!=7) { vec[i+2] = 0; continue; } if (first) { first = false; val = *ref; rc = 0; } rc |= val^*ref; } vec[0] = rc; vec[1] = val; return vec; } template void Update(DimDescribedService &svc, const array &data, const Time &t=Time(), int n=N) { // svc.setQuality(vec[40]<=vec[41]); svc.setData(const_cast(data.data()), sizeof(T)*n); svc.Update(t); } template void Print(const char *name, const pair> &data) { cout << name << "|" << data.first << "|" << data.second[1] << "|" << data.second[0] << " fNumConnected; 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++) { // 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); } } void debugHead(int /*socket*/, const FAD::EventHeader &h) { const uint16_t id = h.Id(); if (id>39) return; if (fNumConnected.size()!=40) fNumConnected.resize(40); const vector con(gi_NumConnect, gi_NumConnect+40); const bool changed = con!=fNumConnected || !IsThreadRunning(); fNumConnected = con; fDimQueue2.post(make_tuple(Time(), changed, h)); //const lock_guard guard(fMutexDimQueue2); //fDimQueue2.push_back(make_tuple(Time(), changed, h)); } }; EventBuilderWrapper *EventBuilderWrapper::This = 0; // ----------- Event builder callbacks implementation --------------- extern "C" { FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len) { return EventBuilderWrapper::This->runOpen(irun, runhd, len); } int runWrite(FileHandle_t fileId, EVENT *event, size_t len) { return EventBuilderWrapper::This->runWrite(fileId, event, len); } int runClose(FileHandle_t fileId, RUN_TAIL *runth, size_t len) { return EventBuilderWrapper::This->runClose(fileId, runth, len); } // ----- //void *runStart(uint32_t /*irun*/, RUN_HEAD */*runhd*/, size_t /*len*/) //{ // return NULL; //} int subProcEvt(int threadID, PEVNT_HEADER *fadhd, EVENT *event, int16_t mboard, void *runPtr) { return EventBuilderWrapper::This->subProcEvt(threadID, fadhd, event, mboard, runPtr); } int runEnd(uint32_t, void */*runPtr*/) { return 0; } // ----- int eventCheck(uint32_t runNr, PEVNT_HEADER *fadhd, EVENT *event, int mboard) { return EventBuilderWrapper::This->eventCheck(runNr, fadhd, event, mboard); } void gotNewRun(uint32_t runnr, PEVNT_HEADER *headers) { return EventBuilderWrapper::This->gotNewRun(runnr, headers); } // ----- void factOut(int severity, int err, const char *message) { EventBuilderWrapper::This->factOut(severity, err, message); } void factStat(GUI_STAT stat) { EventBuilderWrapper::This->factStat(stat); } void factStatNew(EVT_STAT stat) { EventBuilderWrapper::This->factStat(stat); } void factReportIncomplete (uint64_t rep) { EventBuilderWrapper::This->factReportIncomplete(rep); } // ------ void debugHead(int socket, int/*board*/, void *buf) { const FAD::EventHeader &h = *reinterpret_cast(buf); EventBuilderWrapper::This->debugHead(socket, h); } void debugStream(int isock, void *buf, int len) { return EventBuilderWrapper::This->debugStream(isock, buf, len); } void debugRead(int isock, int ibyte, int32_t event, int32_t ftmevt, int32_t runno, int state, uint32_t tsec, uint32_t tusec) { EventBuilderWrapper::This->debugRead(isock, ibyte, event, ftmevt, runno, state, tsec, tusec); } } #endif