Index: /drsdaq/DAQReadout.cc =================================================================== --- /drsdaq/DAQReadout.cc (revision 210) +++ /drsdaq/DAQReadout.cc (revision 211) @@ -34,5 +34,5 @@ {"take", &DAQReadout::cmd_take, false, " [n] [source]", "Start run (data, pedestal or test) with n events"}, {"events", &DAQReadout::cmd_events, false, "", "Number of events in current run"}, - {"start", &DAQReadout::cmd_start, true, "", "Start DRS and DAQ without disk writing (feedback will be called)"}, + {"start", &DAQReadout::cmd_take, false, "", "Start run without disk writing"}, {"stop", &DAQReadout::cmd_stop, false, "", "Issue soft trigger and stop DAQ"}, {"regtest", &DAQReadout::cmd_regtest, true, "", "DRS register test"}, @@ -45,4 +45,5 @@ {"disk", &DAQReadout::cmd_disk, false, "" ,"Remaining disk space"}, {"uptime", &DAQReadout::cmd_uptime, false, "", "Get DAQ uptime"}, + {"update", &DAQReadout::cmd_update, false, "", "Minimum delay between updates to DIM event service"}, {"exit", &DAQReadout::cmd_exit, false, "", "Exit program"}, {"fmode", &DAQReadout::cmd_fmode, false, "[off|active|targ]", "Set or get feedback mode"}, @@ -55,4 +56,7 @@ +// Global pointer for thread entry routines +class DAQReadout *This; + // ----------------------------------------------- // ***** Constructor: Class initialisation ***** @@ -60,19 +64,29 @@ // -DAQReadout::DAQReadout() :EvidenceServer(SERVER_NAME) { - - time(&StartTime); // Start time of DAQ +DAQReadout::DAQReadout(): + DimCommand((char *) SERVER_NAME"/Command", (char *) "C"), + EvidenceServer(SERVER_NAME) { + + // Global class pointer + This = this; + MainThread = getpid(); + + // Start time of DAQ + time(&StartTime); + + // Initialize mutex for thread synchronisation + if (pthread_mutex_init(&Mutex, NULL) != 0) { + State(FATAL, "pthread_mutex_init() failed"); + } // Initialize status structure - daq_state = stopped; - daq_runtype = data; - Socket = -1; - Exit = false; - NumEvents = 0; - NumEventsRequested = 0; - NumBoards = 0; - FirstBoard = 0; - LastBoard = -1; - CmdFromSocket = false; + daq_state = stopped; + daq_runtype = data; + NumEvents = 0; + NumEventsRequested = 0; + NumBoards = 0; + FirstBoard = 0; + LastBoard = -1; + MinDelay = 1; // Get configuration data @@ -83,5 +97,4 @@ fMinDiskSpaceMB = atoi(GetConfig("MinDiskSpaceMB")); fMaxFileSizeMB = atoi(GetConfig("MaxFileSizeMB")); - fCCPort = atoi(GetConfig("CCPort")); fDefaultFrequency = atof(GetConfig("DefaultFrequency")); @@ -106,5 +119,5 @@ TCalib = new bool [NumBoards]; - if (NumBoards == 0) PrintMessage("No DRS boards found - check VME crate and configuration file!\n"); + if (NumBoards == 0) PrintMessage("No DRS boards found\n"); for (int i=0; i*CommandList[CmdNumber].CommandPointer)(); + else if(CommandList[CmdNumber].NeedNotBusy && NumBoards==0) PrintMessage("No boards available\n"); + else { + pthread_mutex_lock(&Mutex); + (this->*CommandList[CmdNumber].CommandPointer)(); + pthread_mutex_unlock(&Mutex); + } return; } PrintMessage("Unknown command '%s'\n",Param[0]); - return; } @@ -186,5 +215,5 @@ // Print DAQ configuration void DAQReadout::cmd_config() { - PrintConfig(CmdFromSocket ? MsgToSocket : MsgToConsole); + PrintConfig(MsgToConsole); } @@ -202,5 +231,5 @@ for (int i=FirstBoard; i<=LastBoard-1; i++) { if ((GetBoard(i)->GetNumberOfChannels() != GetBoard(i+1)->GetNumberOfChannels()) || (GetBoard(i)->GetNumberOfChips() != GetBoard(i+1)->GetNumberOfChips())) { - PrintMessage("Cannot take data is not all boards have the same number of DRS chips and channels due to data format restriction\n"); + PrintMessage("Cannot take data if not all boards have the same number of DRS chips and channels due to data format restriction\n"); return; } @@ -241,40 +270,33 @@ else snprintf(RHeader->Description,sizeof(RHeader->Description),"DUMMY"); - // Request new run number - DimRpcInfo RunNumRPC((char *) "NextRunNumber", -1); - RunNumRPC.setData((char *) ""); - RunNumber = RunNumRPC.getInt(); - if(RunNumber < 1) { - PrintMessage("Error: No connection to run number dispatcher or received number smaller than 1\n"); - return; - } + // Request new run number (if command was 'start', set run number to -1 --> no disk writing) + if (Match(Param[0], "take")) { + DimRpcInfo RunNumRPC((char *) "NextRunNumber", -1); + + RunNumRPC.setData((char *) "");printf("and least here\n"); + + RunNumber = RunNumRPC.getInt(); + if(RunNumber < 1) { + PrintMessage("Error: No connection to run number dispatcher or received number smaller than 1\n"); + return; + } + } + else RunNumber = -1; // Create DAQ thread - if ((pthread_create(&thread_DAQ, NULL, (void * (*)(void *)) DAQ,(void *) this)) != 0) - PrintMessage("pthread_create failed with DAQ thread (%s)\n",strerror(errno)); + pthread_t Thread; + int Code; + if ((Code = pthread_create(&Thread, NULL, (void * (*)(void *)) ::DAQ,(void *) this)) != 0) { + PrintMessage("pthread_create() failed with DAQ thread (error code %d)\n", Code); + } else { daq_state = active; Stop = false; - pthread_detach(thread_DAQ); + if ((Code = pthread_detach(Thread)) != 0) { + PrintMessage("pthread_detach() failed with DAQ thread (error code %d)\n", Code); + } } } -// Start DRS -void DAQReadout::cmd_start() { - if (IsDRSFreqSet()) { - StartDRS(); - PrintMessage("Domino wave started\n"); - - // Create DAQ thread - if ((pthread_create(&thread_DAQ_Silent, NULL, (void * (*)(void *)) DAQ_Silent,(void *) this)) != 0) - PrintMessage("pthread_create failed with DAQ_Silent thread (%s)\n",strerror(errno)); - else { - daq_state = active; - Stop = false; - pthread_detach(thread_DAQ_Silent); - } - } -} - // EEPROM test void DAQReadout::cmd_eepromtest() { @@ -366,19 +388,11 @@ } - PrintMessage(CmdFromSocket ? MsgToSocket:MsgToConsole|MsgToLog, "==START== %d %.2f %.2f ",kNumberOfBins+2,DRSFreq[Board],GetBoard(Board)->GetPrecision()); - double mean=0,square=0.0; + PrintMessage(MsgToConsole, "==START== %d %.2f %.2f ",kNumberOfBins+2,DRSFreq[Board],GetBoard(Board)->GetPrecision()); for (int k=0; k Execute shell command\n\n" + PrintMessage(MsgToConsole,". Execute shell command\n\n" "Items in <> are mandatory, in [] optional, | indicates mutual exclusive or.\n" "Test data can also be written if no DRS boards are available.\n" @@ -718,19 +742,15 @@ } -// Exit programm +// Exit programm - SIGTERM sets ExitRequest flag +// If command comes from DimCommand thread, SIGTERM also makes readline() return void DAQReadout::cmd_exit() { - if (CmdFromSocket) { - PrintMessage("Exit command not allowed over socket.\n"); - return; - } - if (daq_state==active) PrintMessage("Issue \"stop\" first to stop daq\n"); - else { - Exit = true; - if(SocketThread != NULL) pthread_kill(*SocketThread, SIGUSR1); - } + + if (daq_state == active) PrintMessage("Issue 'stop' first to stop daq\n"); + else kill(MainThread, SIGTERM); } // Set/get mode of feedback void DAQReadout::cmd_fmode() { + if(Match(Param[1],"off")) HVFB->SetFBMode(FB_Off); if(Match(Param[1],"active")) HVFB->SetFBMode(FB_Active); @@ -741,4 +761,5 @@ // Set/get current number of events void DAQReadout::cmd_faverage() { + if(NParam==1) PrintMessage("Current number of feedback events: %u (acting when %u events are reached)\n", HVFB->GetCurrentCount(), HVFB->GetNumAverages()); @@ -749,4 +770,5 @@ // Set/get feedback gain void DAQReadout::cmd_fgain() { + if(NParam==2) HVFB->SetGain(atof(Param[1])); PrintMessage("Feedback gain is %.2f\n", HVFB->GetGain()); @@ -755,4 +777,5 @@ // Set/get target value void DAQReadout::cmd_ftarget() { + if(NParam==1) HVFB->GetTargets(); else if(NParam!=5) PrintUsage(); @@ -768,4 +791,5 @@ // Start response measurement void DAQReadout::cmd_fresponse() { + if(NParam==1) HVFB->GetResponse(); else if(atof(Param[1])) HVFB->MeasureResponse(atof(Param[1])); @@ -775,5 +799,6 @@ // Print feedback configuration void DAQReadout::cmd_fconfig() { - HVFB->PrintConfig(CmdFromSocket ? MsgToSocket : MsgToConsole); + + HVFB->PrintConfig(MsgToConsole); } @@ -940,5 +965,5 @@ } -// Open new raw data file +// Open new raw data file (if RunNumber == -1, data will be written to /dev/null) bool DAQReadout::OpenRawFile() { @@ -955,16 +980,19 @@ // Create direcory if not existing (ignore error if already existing) and change to it - snprintf(Buffer, sizeof(Buffer), "%s/%s", fRawDataPath, RunDate); - if(mkdir(Buffer, S_IRWXU|S_IRWXG)==-1 && errno!=EEXIST) { - PrintMessage("\rError: Could not create direcory \"%s\" (%s)\n", Buffer, strerror(errno)); - return false; + if (RunNumber != -1) { + snprintf(Buffer, sizeof(Buffer), "%s/%s", fRawDataPath, RunDate); + if(mkdir(Buffer, S_IRWXU|S_IRWXG)==-1 && errno!=EEXIST) { + PrintMessage("\rError: Could not create direcory \"%s\" (%s)\n", Buffer, strerror(errno)); + return false; + } } // Generate filename - snprintf(FileName,sizeof(FileName),"%s/%s/%s_D1_%.8u.%.3u_%c_%s.raw", fRawDataPath, RunDate, - RunDate,RunNumber,FileNumber,toupper(daq_runtype_str[daq_runtype][0]),RHeader->Description); + if (RunNumber == -1) snprintf(FileName, sizeof(FileName), "/dev/null"); + else snprintf(FileName, sizeof(FileName),"%s/%s/%s_D1_%.8u.%.3u_%c_%s.raw", fRawDataPath, RunDate, + RunDate, RunNumber, FileNumber, toupper(daq_runtype_str[daq_runtype][0]), RHeader->Description); // Open file with rwx right for owner and group, never overwrite file - Rawfile = open(FileName,O_WRONLY|O_CREAT|O_EXCL, S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP); + Rawfile = open(FileName,O_WRONLY|O_CREAT|(RunNumber==-1?0:O_EXCL), S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP); if(Rawfile==-1) { PrintMessage("\rError: Could not open file \"%s\" (%s)\n", FileName, strerror(errno)); @@ -1072,16 +1100,28 @@ return false; } - - // Event data (It is required that at least three chunks can be written with writev(), therefore - // IOV_MAX>=3 is checked at startup - + unsigned int Start, Count = 0; ssize_t WriteResult, Size = 0; struct iovec DataPart[IOV_MAX]; + // Write trigger cells + for (int i=FirstBoard; (i<=LastBoard + (NumBoards==0)); i++) { + for (unsigned int k=0; kNChips; k++) { + if ((WriteResult=write(Rawfile, &TriggerCell[i][k], sizeof(int))) != sizeof(int)) { + if (WriteResult == -1) PrintMessage("Error: Could not write trigger cells, terminating run (%s)\n", strerror(errno)); + else PrintMessage("Error: Could only write %u out of %u bytes of event data, terminating run\n", WriteResult, sizeof(int)); + return false; + } + } + } + + // Event data (It is required that at least three chunks can be written with writev(), therefore + // IOV_MAX>=3 is checked at startup + + // First chunk: trigger cells - DataPart[Count].iov_base = (char *) TriggerCell + FirstBoard*RHeader->NChips*sizeof(int); // TriggerCell is without cast a pointer to an 8-byte unit (two ints) ! - DataPart[Count++].iov_len = RHeader->NBoards * RHeader->NChips * sizeof(int); - Size += DataPart[Count-1].iov_len; + //DataPart[Count].iov_base = (char *) TriggerCell + FirstBoard*RHeader->NChips*sizeof(int); // TriggerCell is without cast a pointer to an 8-byte unit (two ints) ! + //DataPart[Count++].iov_len = RHeader->NBoards * RHeader->NChips * sizeof(int); + //Size += DataPart[Count-1].iov_len; // Remaining chunks: ADC data (two chucks per channel if wrap around of pipeline occurred) @@ -1123,8 +1163,8 @@ PrintMessage(Target, "RawDataPath: %s\n" "DefaultFrequency: %.2f\tFirstSample: %d\tSamples: %u\n" - "MinDiskSpaceMB: %u\tMaxFileSizeMB: %d\tCCPort: %d\n" + "MinDiskSpaceMB: %u\tMaxFileSizeMB: %d\n" "CalibDataPath: %s\n", fRawDataPath,fDefaultFrequency,fFirstSample,fSamples,fMinDiskSpaceMB, - fMaxFileSizeMB,fCCPort,fCalibDataPath); + fMaxFileSizeMB,fCalibDataPath); } @@ -1149,10 +1189,9 @@ } -// Print message to log file, and screen or socket (depending on command origin) +// Print message to console only void DAQReadout::PrintMessage(const char *Format, ...) { va_list ArgumentPointer; va_start(ArgumentPointer, Format); - if(CmdFromSocket) DoPrintMessage(Format, ArgumentPointer, MsgToSocket); - else DoPrintMessage(Format, ArgumentPointer, MsgToConsole); + DoPrintMessage(Format, ArgumentPointer, MsgToConsole); va_end(ArgumentPointer); } @@ -1163,6 +1202,6 @@ void DAQReadout::DoPrintMessage(const char *Format, va_list ArgumentPointer, int Target) { - static char Textbuffer[MAX_COM_SIZE]; // static: it is only allocated once - + static char Textbuffer[MAX_COM_SIZE]; + memset(Textbuffer, 0, sizeof(Textbuffer)); vsnprintf(Textbuffer, sizeof(Textbuffer), Format, ArgumentPointer); @@ -1176,4 +1215,8 @@ fflush(stdout); } + + // Send to DIM service + SetStdOut(Textbuffer); + // Send to log if(Target & MsgToLog) { @@ -1185,8 +1228,176 @@ else DimClient::sendCommandNB("DColl/Log", SERVER_NAME" asprintf() failed"); } - // Print to socket - if((Target & MsgToSocket) && Socket!=-1) write(Socket, Textbuffer, strlen(Textbuffer)); -} - +} + +// DIM command handler (must be non-blocking, otherwise a DIM rpc would dead-lock) +void DAQReadout::commandHandler() { + + // Copy command to new buffer (must be freed by the new thread) + char *Command; + if (asprintf(&Command, "%s", getString()) == -1) { + PrintMessage("asprintf() failed in DRSReadout::commandHandler() (%s)\n", strerror(errno)); + return; + } + + // Create detached command handling thread + pthread_t Thread; + int Code; + if ((Code = pthread_create(&Thread, NULL, (void * (*)(void *)) ::Execute,(void *) Command)) != 0) { + PrintMessage("pthread_create() failed in DRSReadout::commandHandler() (%s)\n", strerror(Code)); + } + else { + if ((Code = pthread_detach(Thread)) != 0) { + PrintMessage("pthread_detach() failed in DRSReadout::commandHandler() (%s)\n", strerror(Code)); + } + } +} + +/********************************************************************\ + + DAQ Thread + + This thread takes data until the requested number of events is reached, + until no more disk space is available or until data taking is stopped. + No mutex mechanism is used since variables will never be written + simultaneoously by this and the main thread. + +\********************************************************************/ + +void DAQReadout::DAQ() { + + struct timeval StartTime, StopTime; + unsigned int EventsInFile; + unsigned long long RunSize = 0; + bool WriteError = false; + time_t LastDIMUpdate = 0; + off_t FileSize; + int DIMSize; + + NumEvents = 0; + FileNumber = 0; + HVFB->ClearAverages(); + gettimeofday(&StartTime, NULL); + PrintMessage("\rStarting run #%d (%s) with %u event(s)\n", RunNumber, daq_runtype_str[daq_runtype], NumEventsRequested); + + do { + // Check if enough disk space is left + if (CheckDisk(fRawDataPath) <= fMinDiskSpaceMB+fMaxFileSizeMB) { + PrintMessage("\rError: Disk space after next file (max. %d MByte) below %d MByte\n", fMaxFileSizeMB, fMinDiskSpaceMB); + break; + } + + // Init run header, open raw file, write run header + if (!OpenRawFile()) break; + PrintMessage("\rData file \"%s\" opened.\n", FileName); + EventsInFile = 0; + FileSize = 0; + + WriteError |= !WriteRunHeader(); + StartDRS(); + + // Copy run header and board structures to buffer for DIM event service + memcpy(DIMEventData, RHeader, sizeof(RunHeader)); + ((RunHeader *) DIMEventData)->Events = 1; // always contains 1 event + memcpy(DIMEventData+sizeof(RunHeader), &BStruct[FirstBoard], sizeof(BoardStructure)*(LastBoard-FirstBoard+1+(NumBoards==0))); + + // Service size for DIM + DIMSize = sizeof(RunHeader) + sizeof(EventHeader) + + sizeof(BoardStructure)*RHeader->NBoards + + sizeof(int)*RHeader->NBoards*RHeader->NChips + + sizeof(short)*RHeader->NBoards*RHeader->NChips*RHeader->NChannels*RHeader->Samples; + + // Take data until finished, stopped or file too large + while ((NumEventsProcessEvent(); + + // Call routine to update DIM service (update rate is limited) + if (time(NULL) - LastDIMUpdate < MinDelay) continue; + LastDIMUpdate = time(NULL); + + // Copy new event header + char *Pnt = DIMEventData+sizeof(RunHeader)+RHeader->NBoards*sizeof(BoardStructure); + memcpy(Pnt, EHeader, sizeof(EventHeader)); + Pnt += sizeof(EventHeader); + + // Copy new trigger cells + for (int i=FirstBoard; (i<=LastBoard + (NumBoards==0)); i++) { + for (unsigned int k=0; kNChips; k++) { + *((int *) Pnt) = TriggerCell[i][k]; + Pnt += sizeof(int); + } + } + + // Copy event data + unsigned int Start; + for (int i=FirstBoard; (i<=LastBoard + (NumBoards==0)); i++) { + for (unsigned int k=0; kNChips; k++) { + // Start bin + if (GetBoard(i)->GetDRSType() == 4) Start = 0; + else Start = (TriggerCell[i][k]-fFirstSample+kNumberOfBins) % kNumberOfBins; + + for (unsigned int l=0; lNChannels; l++) { + memcpy(Pnt, &WaveForm[i][k][l][Start], (Start+fSamplesupdateService((void *) DIMEventData, DIMSize); + } + + // Write updated run header, close file + RunSize += FileSize; + WriteError |= !UpdateRunHeader(EventsInFile, WriteError); + if(close(Rawfile) == -1) PrintMessage("Error: Could not close data file (%s)\n", strerror(errno)); + else PrintMessage("Data file closed (size %lu MByte).\n", FileSize/1024/1024); + + FileNumber += 1; + } while((NumEventsSlowDataClass->NewEntry("Runinfo"); + //m->SlowDataClass->AddToEntry("%d %s %s %d %d %s", m->RunNumber, WriteError?"Error":"OK", daq_runtype_str[m->daq_runtype], m->NumEvents, m->FileNumber, m->RHeader->Description); + + // Print run statistics + if (NumEvents>0 && !WriteError) { + gettimeofday(&StopTime, NULL); + float RunTime = StopTime.tv_sec-StartTime.tv_sec + (StopTime.tv_usec-StartTime.tv_usec)*1e-6; + PrintMessage("Time for run %.2f seconds, trigger rate %.2f Hz.\n", RunTime, NumEvents/RunTime); + PrintMessage("Run size %llu MByte, data rate %.1f MByte/s.\n", RunSize/1024/1024, RunSize/1024.0/1024/RunTime); + } + + daq_state = stopped; +} // --------------------------------------- @@ -1228,129 +1439,13 @@ } - -/********************************************************************\ - - DAQ Thread - - This thread takes data until the requested number of events is reached, - until no more disk space is available or until data taking is stopped. - No mutex mechanism is used since variables will never be written - simultaneoously by two threads. - -\********************************************************************/ - +// Thread entry routines (non-static class methods cannot be directly executed as thread) void DAQ(DAQReadout *m) { - struct timeval StartTime, StopTime; - unsigned int EventsInFile; - unsigned long long RunSize = 0; - bool WriteError = false; - off_t FileSize; - - m->NumEvents = 0; - m->FileNumber = 0; - m->HVFB->ClearAverages(); - gettimeofday(&StartTime, NULL); - m->PrintMessage("\rStarting run #%d (%s) with %u event(s)\n",m->RunNumber,daq_runtype_str[m->daq_runtype],m->NumEventsRequested); - - do { - // Check if enough disk space is left - if (CheckDisk(m->fRawDataPath) <= m->fMinDiskSpaceMB+m->fMaxFileSizeMB) { - m->PrintMessage("\rError: Disk space after next file (max. %d MByte) below %d MByte\n",m->fMaxFileSizeMB,m->fMinDiskSpaceMB); - break; - } - - // Init run header, open raw file, write run header - if (!m->OpenRawFile()) break; - m->PrintMessage("\rData file \"%s\" opened.\n",m->FileName); - EventsInFile = 0; - FileSize = 0; - - WriteError |= !m->WriteRunHeader(); - - if (m->daq_runtype != test) m->StartDRS(); - - // Take data until finished, stopped or file too large - while ((m->NumEventsNumEventsRequested || m->NumEventsRequested==0) && - !m->Stop && FileSize/1024/1024fMaxFileSizeMB && !WriteError) { - - if (m->daq_runtype == data) while (m->IsDRSBusy()); // Wait for hardware trigger (if DAQ stopped, DRS will not be busy anymore) - else if (m->daq_runtype == pedestal) m->StopDRS(); // ..or for software trigger - - // Read event data via VME or generate test data (for one board if no boards available) - if (m->daq_runtype != test) { - m->ReadCalibratedDRSData(); - m->StartDRS(); // Restart here: writing data is in parallel to waiting for next trigger - } - else { - double Period = ((double) rand())/RAND_MAX*20; - for (long unsigned int i=0; iRHeader->NBoards*m->RHeader->NChips*m->RHeader->NChannels*m->RHeader->Samples; i++) - *((short *) m->WaveForm+i) = (short) (sin(i/Period)*1000); - } - - // Write event to disk and update file size - EventsInFile++; m->NumEvents++; - WriteError |= !m->WriteEvent(); - - if((FileSize = lseek(m->Rawfile, 0, SEEK_CUR)) == -1) { - m->PrintMessage("Error: Could not determine file size, terminating run (%s)\n", strerror(errno)); - WriteError = true; - } - - // Call feedback to process event - m->HVFB->ProcessEvent(); - } - - // Write updated run header, close file - RunSize += FileSize; - WriteError |= !m->UpdateRunHeader(EventsInFile, WriteError); - if(close(m->Rawfile)==-1) m->PrintMessage("Error: Could not close data file (%s)\n", strerror(errno)); - else m->PrintMessage("Data file closed (size %lu MByte).\n", FileSize/1024/1024); - - m->FileNumber += 1; - } while((m->NumEventsNumEventsRequested || m->NumEventsRequested==0) && !m->Stop && !WriteError); - - m->StopDRS(); - - // Print run summary to screen - if(!WriteError) m->PrintMessage("\rRun #%d (%s) %s, %d events\n",m->RunNumber,daq_runtype_str[m->daq_runtype],(m->NumEvents == m->NumEventsRequested) ? "completed":"stopped",m->NumEvents); - else m->PrintMessage("\rRun #%d (%s) aborted due to error after %d events\n",m->RunNumber,daq_runtype_str[m->daq_runtype],m->NumEvents); - - // Write run summary to slow data file - //m->SlowDataClass->NewEntry("Runinfo"); - //m->SlowDataClass->AddToEntry("%d %s %s %d %d %s", m->RunNumber, WriteError?"Error":"OK", daq_runtype_str[m->daq_runtype], m->NumEvents, m->FileNumber, m->RHeader->Description); - - // Print run statistics - if (m->NumEvents>0 && !WriteError) { - gettimeofday(&StopTime, NULL); - float RunTime = StopTime.tv_sec-StartTime.tv_sec + (StopTime.tv_usec-StartTime.tv_usec)*1e-6; - m->PrintMessage("Time for run %.2f seconds, trigger rate %.2f Hz.\n", RunTime, m->NumEvents/RunTime); - m->PrintMessage("Run size %llu MByte, data rate %.1f MByte/s.\n", RunSize/1024/1024, RunSize/1024.0/1024/RunTime); - } - - m->daq_state = stopped; -} - - -/********************************************************************\ - - DAQ Thread - no disk writing, only hardware trigger, for feedback tests - -\********************************************************************/ - -void DAQ_Silent(DAQReadout *m) { - - m->PrintMessage("\rData taking started\n"); - do { - // Start DRS and wait for hardware trigger - m->StartDRS(); - while (m->IsDRSBusy()); - - // Read event data via VME and call feedback - m->ReadCalibratedDRSData(); - m->HVFB->ProcessEvent(); - } while(!m->Stop); - - m->PrintMessage("\rData taking stopped\n"); - m->daq_state = stopped; -} + m->DAQ(); +} + +void Execute(char *Command) { + + This->Execute(Command); + free(Command); +} Index: /drsdaq/DAQReadout.h =================================================================== --- /drsdaq/DAQReadout.h (revision 210) +++ /drsdaq/DAQReadout.h (revision 211) @@ -1,4 +1,7 @@ #ifndef DAQREADOUT_H_SEEN #define DAQREADOUT_H_SEEN + +#define SERVER_NAME "drsdaq" // Name to use in DIM +#include "Evidence.h" #include @@ -26,26 +29,20 @@ #define MsgToConsole 1 #define MsgToLog 2 -#define MsgToSocket 4 enum state_enum {active, stopped}; enum runtype_enum {data, pedestal, reserved, test}; -class DAQReadout : public DRS, public DRSCallback, public EvidenceServer { +class DAQReadout : public DRS, public DRSCallback, public DimCommand, public EvidenceServer { + time_t StartTime; + pid_t MainThread; + DimService *EventService; + int MinDelay; + unsigned int CmdNumber; - pthread_t thread_DAQ; - pthread_t thread_DAQ_Silent; + void PrintUsage(); + void commandHandler(); - unsigned int CmdNumber; - void PrintUsage(); - - public: - RunHeader* RHeader; - EventHeader* EHeader; - - short (*WaveForm)[kNumberOfChipsMax][kNumberOfChannelsMax][kNumberOfBins]; - int (*TriggerCell)[kNumberOfChipsMax]; - - pthread_mutex_t control_mutex; + pthread_mutex_t Mutex; int Rawfile; class HVFeedback* HVFB; @@ -58,11 +55,29 @@ int fFirstSample; unsigned int fSamples; - int fCCPort; double fDefaultFrequency; + + state_enum daq_state; + runtype_enum daq_runtype; + bool Stop; // Set to true to stop run + unsigned int NumEvents; // Number of event taken + unsigned int NumEventsRequested; // Number of events requested + int RunNumber; + unsigned int FileNumber; + char FileName[MAX_PATH]; + char CalibInfoFilename[MAX_PATH]; + + public: + RunHeader* RHeader; + EventHeader* EHeader; + BoardStructure *BStruct; - // Status variables + short (*WaveForm)[kNumberOfChipsMax][kNumberOfChannelsMax][kNumberOfBins]; + int (*TriggerCell)[kNumberOfChipsMax]; + char *DIMEventData; + + char Prompt[MAX_COM_SIZE]; + int NParam; // Number of command parameters const char *Param[MAX_NUM_TOKEN]; // Pointers to parameters - bool CmdFromSocket; // Current command issued via socket int NumBoards; int FirstBoard; @@ -72,21 +87,5 @@ double *ACalibTemp; bool *TCalib; - - BoardStructure *BStruct; - state_enum daq_state; - runtype_enum daq_runtype; - int Socket; // -1 if not connected - pthread_t *SocketThread; // exit function sends signal to this thread - bool Exit; - bool Stop; // Set to true to stop run - unsigned int NumEvents; // Number of event taken - unsigned int NumEventsRequested; // Number of events requested - int RunNumber; - unsigned int FileNumber; - char FileName[MAX_PATH]; - char Prompt[MAX_COM_SIZE]; - char CalibInfoFilename[MAX_PATH]; - - // Public functions + DAQReadout(); ~DAQReadout(); @@ -94,5 +93,5 @@ void cmd_exit(); void cmd_help(); void cmd_board(); void cmd_status(); - void cmd_led(); void cmd_freq(); + void cmd_led(); void cmd_freq(); void cmd_acalib(); void cmd_serial(); void cmd_trigger(); void cmd_centre(); @@ -103,5 +102,5 @@ void cmd_eepromtest(); void cmd_tcalib(); void cmd_regtest(); void cmd_ramtest(); - void cmd_start(); void cmd_take(); + void cmd_take(); void cmd_update(); void cmd_config(); void cmd_events(); void cmd_disk(); void cmd_uptime(); @@ -111,5 +110,5 @@ void cmd_fresponse(); void cmd_fconfig(); - void CommandControl(char*); + void Execute(char*); void StartDRS(); void StopDRS(); @@ -133,4 +132,5 @@ bool UpdateRunHeader(unsigned int, bool); bool WriteEvent(); + void DAQ(); void Progress(int); @@ -138,5 +138,5 @@ void DAQ(DAQReadout *); -void DAQ_Silent(DAQReadout *); +void Execute(char *); bool Match(const char*, const char*); Index: /drsdaq/HVFeedback.cc =================================================================== --- /drsdaq/HVFeedback.cc (revision 210) +++ /drsdaq/HVFeedback.cc (revision 211) @@ -57,16 +57,25 @@ char *Token = strtok(m->GetConfig("DefaultResponse"), " \t"); - for (int i=0; iNumBoards*fNumberOfChips*fNumberOfChannels; i++) { - if (Token == NULL) break; - *(&Response[0][0][0]+i) = (float) atof(Token); - Token = strtok(NULL, " \t"); - } + for (int i=0; iNumBoards; i++) { + for (int j=0; jGetConfig("DefaultTarget"), " \t"); - for (int i=0; iNumBoards*fNumberOfChips*fNumberOfChannels; i++) { - if (Token == NULL) break; - *(&Target[0][0][0]+i) = (float) atof(Token); - Token = strtok(NULL, " \t"); - } - + for (int i=0; iNumBoards; i++) { + for (int j=0; jGetConfig("DefaultGain"));printf("Gain %d\n", Gain); + Gain = atof(m->GetConfig("DefaultGain")); SetFBMode(FB_Off); SetNumAverages(fDefaultNumAverage); Index: /drsdaq/HVFeedback.h =================================================================== --- /drsdaq/HVFeedback.h (revision 210) +++ /drsdaq/HVFeedback.h (revision 211) @@ -6,7 +6,4 @@ #include #include - -#define SERVER_NAME "Feedback" // Name to use in DIM -#include "Evidence.h" #include "RawDataCTX.h" Index: /drsdaq/History.txt =================================================================== --- /drsdaq/History.txt (revision 210) +++ /drsdaq/History.txt (revision 211) @@ -61,2 +61,8 @@ 11/3/2010 Removed SlowData class. 12/3/2010 Removed local configuration and logging. +30/4/2010 Command 'start' now starts a normal run as 'take', but data is written to + /dev/null, the run number is -1. +4/5/2010 Removed socket server, added DIM command handler (thus cannot communicate + with ddd over socket anymore). Added DIM service containing full event. +18/5/2010 Rate of event service adjustable with command 'update'. +21/5/2010 Fix so that 'exit' command also works as DimCommand. Index: /drsdaq/drsdaq.cpp =================================================================== --- /drsdaq/drsdaq.cpp (revision 210) +++ /drsdaq/drsdaq.cpp (revision 211) @@ -1,22 +1,18 @@ /**************************************************************\ - drsdaq.cpp + drsdaq - Main program for DRS CTX DAQ system. Global initialization, - starts threads for console input and socket interface. + Main program for data acquisition, starts threads for console input. - Sebastian Commichau, Oliver Grimm + Original program by S. Commichau. + + Oliver Grimm, May 2010 \**************************************************************/ -#define DEFAULT_CONFIG "../config/DRSDAQ.conf" // Default configuration file #define LOCKFILE "/tmp/CT3_DAQ_LOCK" #include #include -#include -#include -#include -#include #include #include @@ -26,7 +22,4 @@ // Function prototypes -void ConsoleCommand(DAQReadout *); -void CCCommand(DAQReadout *); -void SignalHandler(int); void CrashHandler(int); void ExitFunction(); @@ -38,27 +31,24 @@ // Several unlikely system call failures are handled via throwing an exception. -int main(int argc, char *argv[]) { +int main() { - char str[MAX_COM_SIZE]; - pthread_t thread_ConsoleCommand, thread_CCCommand; + char str[MAX_COM_SIZE], *Command; int LockDescriptor; - - // writev() in DAQ thread needs to be able to write at least 3 chunks - if(IOV_MAX < 3) { - printf("Fatal error: IOV_MAX is less than 3, cannot use writev() to write event data.\n"); - exit(EXIT_FAILURE); - } + + // Readline library uses getc() (allows interruption by signal) + rl_getc_function = getc; + + // writev() in DAQ thread needs to be able to write at least 3 chunks + if(IOV_MAX < 3) { + printf("Fatal error: IOV_MAX is less than 3, cannot use writev() to write event data.\n"); + exit(EXIT_FAILURE); + } - // Interpret command line (do before lockfile creation in case of exit()) - if((argc==3 && strcmp(argv[1],"-c")!=0) || argc==2) { - printf("Usage: %s [-c ] Default file is \"%s\"\n", argv[0], DEFAULT_CONFIG); - exit(EXIT_SUCCESS); - } - // Assure only one instance of program runs (lock creator written to log file) + // Lock file deleted by ExitFunction() if((LockDescriptor = open(LOCKFILE,O_WRONLY|O_CREAT|O_EXCL, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH)) == -1) { if(errno==EEXIST) { printf("Error: Lock file already existing\n"); - sprintf(str,"paste %s -s -d ' '",LOCKFILE); + snprintf(str, sizeof(str), "paste %s -s -d ' '",LOCKFILE); system(str); } @@ -71,10 +61,11 @@ system("clear"); - printf("\n*** DRS readout built %s, %s (revision %s) *** \n\n",__DATE__, __TIME__, REVISION); + printf("\n*** DRS readout (built %s, %s, revision %s) *** \n\n",__DATE__, __TIME__, REVISION); - // Set signal handlers - signal(SIGUSR1, &SignalHandler); - siginterrupt (SIGUSR1, true); // Set SIGUSR1 to interrupt (and not restart) blocking system calls - signal(SIGQUIT, &CrashHandler); + // Construct main instance (static ensures destructor is called with exit()) + static DAQReadout M; + + // Set signal and exit handlers + atexit(&ExitFunction); signal(SIGILL, &CrashHandler); signal(SIGABRT, &CrashHandler); @@ -82,190 +73,28 @@ signal(SIGSEGV, &CrashHandler); signal(SIGBUS, &CrashHandler); - signal(SIGTERM, &CrashHandler); - signal(SIGINT, &CrashHandler); - signal(SIGHUP, &CrashHandler); - atexit(&ExitFunction); - // Construct main instance and create mutex for thread synchronization - DAQReadout dreadout; - if (pthread_mutex_init(&dreadout.control_mutex, NULL) != 0) { - perror("pthread_mutex_init failed"); - throw; - } - - // Create threads - if (pthread_mutex_init(&dreadout.control_mutex, NULL) != 0) { - perror("pthread_mutex_init failed"); - throw; - } - if ((pthread_create(&thread_ConsoleCommand, NULL, (void * (*)(void *)) ConsoleCommand,(void *) &dreadout)) != 0) { - perror("pthread_create failed with console thread"); - throw; - } - if ((pthread_create(&thread_CCCommand, NULL, (void * (*)(void *)) CCCommand,(void *) &dreadout)) != 0) { - perror("pthread_create failed with socket thread"); - dreadout.SocketThread = NULL; - } - else dreadout.SocketThread = &thread_CCCommand; // Thread should be accessible for sending signals - - // Wait for threads to quit - pthread_join(thread_ConsoleCommand, NULL); - if(dreadout.SocketThread != NULL) pthread_join(thread_CCCommand, NULL); - - // Destruct mutex and main instance - pthread_mutex_destroy (&dreadout.control_mutex); - dreadout.~DAQReadout(); - - // Remove lockfile - if (remove(LOCKFILE)==-1) { - printf("Could not remove lock file %s (%s)\n", LOCKFILE, strerror(errno)); - exit(EXIT_FAILURE); - } - exit(EXIT_SUCCESS); -} - - -/********************************************************************\ - - ConsoleCommand thread - - Handle console input using readline library functions to allow - line editing and history capability - -\********************************************************************/ - -void ConsoleCommand(DAQReadout *m) { - - char *Command; - - while (!m->Exit) { - + // Command loop + while (!M.ExitRequest) { // Assemble prompt - if (m->NumBoards == 0) snprintf(m->Prompt,sizeof(m->Prompt),"\rDAQ> "); - else if (m->FirstBoard == m->LastBoard) snprintf(m->Prompt,sizeof(m->Prompt),"\rDAQ|B%d> ",m->FirstBoard); - else snprintf(m->Prompt,sizeof(m->Prompt),"\rDAQ|B%d-%d> ",m->FirstBoard,m->LastBoard); + if (M.NumBoards == 0) snprintf(M.Prompt,sizeof(M.Prompt),"\rDAQ> "); + else if (M.FirstBoard == M.LastBoard) snprintf(M.Prompt,sizeof(M.Prompt),"\rDAQ|B%d> ",M.FirstBoard); + else snprintf(M.Prompt,sizeof(M.Prompt),"\rDAQ|B%d-%d> ",M.FirstBoard,M.LastBoard); // Read Command - Command = readline(m->Prompt); - if (Command == NULL) { - m->PrintMessage("Error reading command line input\n"); - continue; - } + Command = readline(M.Prompt); + if (Command == NULL) continue; if(strlen(Command)>0) add_history(Command); - // Log command - m->PrintMessage(MsgToLog, "CONSOLE> %s\n", Command); - // Process command - pthread_mutex_lock(&m->control_mutex); - m->CommandControl(Command); - pthread_mutex_unlock(&m->control_mutex); + DimClient::sendCommand(SERVER_NAME"/Command", Command); free(Command); - } + } } - -/********************************************************************\ - - CCCommand thread - - Listen to commands from socket (Central Control) - - This thread will block execution in the accept() and read() socket function - while waiting for a connection or data. If the exit function is invoked through - keyboard command, these blocking functions are interrupted by raising the signal - SIGUSR1. Testing on errno=EINTR indicates this termination. The dummy signal - handler below is needed to prevent the standard thread termination occurring - when this signal is received. - -\********************************************************************/ - -void CCCommand(DAQReadout *m) { - - int ServerSocket,ConnectionSocket,ReadResult; - struct sockaddr_in SocketAddress, ClientAddress; - struct hostent *ClientName; - socklen_t SizeClientAddress=sizeof(ClientAddress); - char Command[MAX_COM_SIZE]; - - // Set up server socket - if ((ServerSocket = socket(PF_INET, SOCK_STREAM, 0)) == -1) { - m->PrintMessage("Could not open server socket, no remote connection possible (%s).\n", strerror(errno)); - return; - } - // Allows immediate reuse of socket after closing (circumvents TIME_WAIT) - int Value=1; - if (setsockopt(ServerSocket, SOL_SOCKET, SO_REUSEADDR, (char *) &Value, sizeof (Value)) == -1) { - m->PrintMessage("Warning: Could not set server socket option SO_REUSEADDR (%s)\n", strerror(errno)); - } - - SocketAddress.sin_family = PF_INET; - SocketAddress.sin_port = htons((unsigned short) m->fCCPort); - SocketAddress.sin_addr.s_addr = INADDR_ANY; - - if (bind(ServerSocket, (struct sockaddr *) &SocketAddress, sizeof(SocketAddress)) == -1) - { - m->PrintMessage("Could not bind port to socket (%s)\n", strerror(errno)); - close(ServerSocket); - return; - } - if (listen(ServerSocket, 0) == -1) { - m->PrintMessage("Could not set socket to listening (%s)\n", strerror(errno)); - close(ServerSocket); - return; - } - - // Looping to wait for incoming connection - while (!m->Exit) { - if ((ConnectionSocket = accept(ServerSocket, (struct sockaddr *) &ClientAddress, &SizeClientAddress)) == -1) { - if (errno!=EINTR) m->PrintMessage("Failed to accept incoming connection (%s)\n", strerror(errno)); - close(ServerSocket); - return; - } - - ClientName = gethostbyaddr((char *) &ClientAddress.sin_addr ,sizeof(struct sockaddr_in),AF_INET); - m->PrintMessage("Connected to client at %s (%s).\n", inet_ntoa(ClientAddress.sin_addr), ClientName!=NULL ? ClientName->h_name:"name unknown"); - m->Socket = ConnectionSocket; - - // Looping as long as client exists and program not terminated - while (!m->Exit) { - - // Try to read command from socket - memset(Command,0,sizeof(Command)); - ReadResult = read(ConnectionSocket, Command, MAX_COM_SIZE); - if (ReadResult==0) break; // Client not exisiting anymore - if (ReadResult==-1) { - if (errno!=EINTR) m->PrintMessage("Error read from socket (%s)\n", strerror(errno)); - break; - } - if (Command[strlen(Command)-1]=='\n') Command[strlen(Command)-1]='\0'; // Remove trailing newline - - // Log command - m->PrintMessage(MsgToConsole|MsgToLog, "SOCKET> %s\n", Command); - - // Process command - pthread_mutex_lock(&m->control_mutex); - m->CmdFromSocket = true; - m->CommandControl(Command); - m->CmdFromSocket = false; - pthread_mutex_unlock(&m->control_mutex); - } - - m->Socket = -1; - m->PrintMessage("Disconnected from client.\n"); - close(ConnectionSocket); - } - close(ServerSocket); - m->PrintMessage("Server socket closed.\n"); -} - - -/********************************************************************\ - - Signal handlers - -\********************************************************************/ +//***************** +// Signal handlers +//***************** // Remove lock file before running default signal code @@ -277,12 +106,11 @@ } -// Dummy signal handler to return from blocking syscalls -void SignalHandler(int Signal) { - return; -} - // This function will be implicitly called by exit() void ExitFunction() { - remove(LOCKFILE); + + if (remove(LOCKFILE)==-1) { + printf("Could not remove lock file %s (%s)\n", LOCKFILE, strerror(errno)); + } + return; }