178 static const int MAXIMUM_FILE_NUMBER = 100;
209 gErrorIgnoreLevel = kFatal;
211 std::ostringstream os;
216 std::remove(os.str().c_str());
219 this->
open(os.str().c_str());
237 if (this->is_open()) {
239 const string file_name = this->
getFile()->GetName();
245 for (
int i = 0; i != MAXIMUM_FILE_NUMBER; ++i) {
251 <<
"_" << cal.getYear() <<
'-' <<
FILL(2,
'0') << cal.getMonth() <<
'-' <<
FILL(2,
'0') << cal.getDay()
262 if (std::rename(file_name.c_str(), os.str().c_str()) == 0)
271 std::remove(file_name.c_str());
296 return (sizeL0 > 0 ||
312 if (
object.is_open())
313 out <<
object.getFile()->GetName();
319 out <<
object.sizeL0 <<
'/'
320 <<
object.sizeL1 <<
'/'
321 <<
object.sizeL2 <<
'/'
322 <<
object.sizeSN <<
'/';
384 current_slice_index = -1;
397 if (c_buffer.is_open()) {
399 JStatusStream(logger) <<
"Close and remove circular buffer " << c_buffer;
401 c_buffer.close(
false);
408 virtual void actionInit(
int length,
const char* buffer)
override
416 if (serversocket.is_valid()) {
417 serversocket->shutdown();
422 catch(
const std::exception& error) {
423 JErrorStream(logger) <<
"Error \"" << error.what() <<
"\"; trigger ev_error.";
444 dumpLimit = numeric_limits<int>::max();
450 properties[
"dataWriter"] = hostname;
451 properties[
"numberOfFramesPerSlice"] = frames_per_slice = 1;
454 properties[
"queueSize"] = maxQueueSize = (totalCPURAM -
GIGABYTE);
455 properties[
"queueDepth"] = maxQueueDepth = 20;
456 properties[
"frameIndex"] = maximal_frame_index = 100000;
459 properties[
"JDataFilter"] = dataFilters;
460 properties[
"DataQueue"] = dataQueues;
461 properties[
"path"] = c_buffer.path;
462 properties[
"archive"] = c_buffer.archive;
463 properties[
"c_sizeL0"] = c_buffer.sizeL0;
464 properties[
"c_sizeL1"] = c_buffer.sizeL1;
465 properties[
"c_sizeL2"] = c_buffer.sizeL2;
466 properties[
"c_sizeSN"] = c_buffer.sizeSN;
467 properties[
"dumpLimit"] = dumpLimit;
470 properties.
read(
string(buffer, length));
472 catch(
const exception& error) {
479 setClockInterval(
update_s * 1000000LL);
481 hostname =
trim(hostname);
486 throw JException(
"Undefined data writer host name.");
490 maximum_frames_per_slice = frames_per_slice;
494 if (dataFilters.empty()) {
495 JNoticeStream(logger) <<
"No DataFilters in process list, or no process list. "
496 <<
"Assuming that this process is the only process on this CPU and setting parameters accordingly.";
499 sort(dataFilters.begin(), dataFilters.end(), compare);
501 unsigned int numberOfDataFiltersOnThisMachine = 0;
509 notice <<
"My IP addresses:";
518 JDebugStream(logger) <<
"Test IP address \"" << i->hostname <<
"\" " << (find(IP.begin(), IP.end(), i->hostname) != IP.end());
520 if (find(IP.begin(), IP.end(), i->hostname) != IP.end()) {
522 numberOfDataFiltersOnThisMachine++;
524 if (i->port ==
this->port) {
530 if (numberOfDataFiltersOnThisMachine == 0) {
531 JNoticeStream(logger) <<
"Zero data filters on this machine according to process list (if it exists). "
532 <<
"Assuming one datafilter on this machine.";
533 numberOfDataFiltersOnThisMachine = 1;
536 if (thisProcess == dataFilters.end()) {
537 JErrorStream(logger) <<
"This process cannot be found in the process list. Why do I exist?";
540 if (thisProcess != dataFilters.end() && thisProcess->index !=
getName()) {
541 JErrorStream(logger) <<
"Mismatch between given process names: "
543 <<
", but in the process list I am referred to as " << thisProcess->index;
546 if (dataFilters.begin() == thisProcess || dataFilters.empty()) {
550 if (maxQueueSize > (totalCPURAM-
GIGABYTE)/(numberOfDataFiltersOnThisMachine)) {
552 maxQueueSize = (totalCPURAM-
GIGABYTE)/(numberOfDataFiltersOnThisMachine);
554 JNoticeStream(logger) <<
"Maximum queue size is too large given the number of processes on this machine. "
555 <<
"Queue size reduced to "
556 << maxQueueSize <<
" bytes." ;
563 JNoticeStream(logger) <<
"Disabling high-rate veto of all PMTs.";
580 JNoticeStream(logger) <<
"This data filter process will report.";
584 JNoticeStream(logger) <<
"Update period [s]: " << getClockInterval();
596 if (buildL1.get() == NULL) {
JErrorStream(logger) <<
"Failed to allocate buildL1."; }
597 if (buildL2.get() == NULL) {
JErrorStream(logger) <<
"Failed to allocate buildL2."; }
598 if (buildSN.get() == NULL) {
JErrorStream(logger) <<
"Failed to allocate buildSN."; }
599 if (buildNB.get() == NULL) {
JErrorStream(logger) <<
"Failed to allocate buildSN."; }
606 if (c_buffer.is_enabled()) {
608 if (!c_buffer.is_open()) {
612 if (c_buffer.is_open()) {
616 JStatusStream(logger) <<
"Created circular buffer " << c_buffer;
620 JErrorStream (logger) <<
"Failed to create circular buffer in directory <" << c_buffer.path <<
">; disable functionality.";
625 JNoticeStream(logger) <<
"Continue using circular buffer " << c_buffer;
629 if (c_buffer.is_open()) {
650 current_slice_index = -1;
655 numberOfTimeslicesProcessed = 0;
656 numberOfIncompleteTimeslicesProcessed = 0;
658 number_of_packets_received = 0;
659 number_of_packets_discarded = 0;
660 number_of_bytes_received = 0;
663 minFrameNumber = numeric_limits<int>::max();
664 maxFrameNumber = numeric_limits<int>::min();
670 logErrorRun .reset();
671 logErrorDetector .reset();
672 logErrorIndex .reset();
673 logErrorIncomplete.reset();
694 if (!timeslices.empty()) {
696 JNoticeStream(logger) <<
"Flushing " << timeslices.size() <<
" slices.";
698 for (deque<JDAQTimesliceL0>::const_iterator i = timeslices.begin(); i != timeslices.end(); ++i) {
707 deque<JDAQTimesliceL0> buffer;
709 timeslices.swap(buffer);
716 current_slice_index = -1;
729 virtual void actionStop(
int length,
const char* buffer)
override
739 if (serversocket.is_valid()) {
740 serversocket->shutdown();
743 serversocket.reset();
747 virtual void actionQuit(
int length,
const char* buffer)
override
755 if (serversocket.is_valid()) {
756 mask.
set(*serversocket);
759 for (JChannelList_t::const_iterator channel = channelList.begin(); channel != channelList.end(); ++channel) {
760 if (!channel->isReady()) {
761 mask.
set(channel->getFileDescriptor());
771 for (JChannelList_t::iterator channel = channelList.begin(); channel != channelList.end(); ) {
775 if (mask.
has(channel->getFileDescriptor())) {
779 if (channel->isReady()) {
781 number_of_packets_received += 1;
782 number_of_reads += channel->getCounter();
783 number_of_bytes_received += channel->size();
787 updateFrameQueue(channel);
791 JErrorStream(logErrorRun) <<
"Receiving data while not running.";
793 number_of_packets_discarded += 1;
801 catch(
const exception& error) {
803 JNoticeStream(logger) <<
"Disconnecting channel " << channel->getFileDescriptor() <<
' ' << error.what();
807 channel = channelList.erase(channel);
812 if (serversocket.is_valid()) {
814 if (mask.
has(*serversocket)) {
818 socket.
accept(serversocket->getFileDescriptor());
832 if (!timeslices.empty() && ((timeslices[0].size() >= frames_per_slice) ||
833 (timeslices.size() >= 2
u &&
834 timeslices[1].size() >= frames_per_slice) ||
835 (timeslices.size() >= maxQueueDepth) ||
842 minFrameNumber = min(minFrameNumber, pending_slice.
getFrameIndex());
843 maxFrameNumber = max(maxFrameNumber, pending_slice.
getFrameIndex());
845 if (pending_slice.size() > frames_per_slice) {
847 JErrorStream(logger) <<
"More frames in timeslice than expected "
848 << pending_slice.size() <<
" > " << frames_per_slice;
850 if (pending_slice.size() <= maximum_frames_per_slice) {
852 JErrorStream(logger) <<
"Adjusting expected frames per timeslice to " << pending_slice.size();
854 frames_per_slice = pending_slice.size();
858 if (!pending_slice.empty()) {
862 processTimeSlice(pending_slice);
866 numberOfTimeslicesProcessed += 1;
870 if (pending_slice.size() < frames_per_slice) {
872 numberOfIncompleteTimeslicesProcessed += 1;
876 error <<
"Timeout -> processed incomplete timeslice: "
878 <<
"Size of timeslice = " << pending_slice.size() <<
';'
879 <<
"Queue depth = " << timeslices.size() <<
';'
882 if ((timeslices.size() >= 2
u &&
883 timeslices[1].size() >= frames_per_slice)) {
885 error <<
" intermittent problem -> continues as-is";
889 error <<
" adjusting expected frames per timeslice from " << frames_per_slice <<
" to " << pending_slice.size();
891 frames_per_slice = pending_slice.size();
898 timeslices.pop_front();
930 if (preamble.getLength() != channel->size()) {
932 JErrorStream(logErrorRun) <<
"Size of received data does not match size reported by preamble: "
933 <<
"preamble.getLength() = " << preamble.getLength() <<
';'
934 <<
"channel->size(): " << channel->size() <<
';';
936 number_of_packets_discarded += 1;
943 JErrorStream(logErrorRun) <<
"Run number " << header.getRunNumber()
945 <<
" -> Dropping frame.";
947 number_of_packets_discarded += 1;
952 if (header.getFrameIndex() <= current_slice_index) {
954 JErrorStream(logErrorIndex) <<
"Frame index " << header.getFrameIndex() <<
" <= " << current_slice_index
955 <<
" module " << header.getModuleID()
956 <<
" -> Dropping frame.";
958 number_of_packets_discarded += 1;
960 if (frames_per_slice < maximum_frames_per_slice) {
964 JErrorStream(logErrorIndex) <<
"Increase number of frames expected to: " << frames_per_slice;
970 if (current_slice_index != -1 && header.getFrameIndex() > current_slice_index + maximal_frame_index) {
972 JErrorStream(logErrorIndex) <<
"Frame index " << header.getFrameIndex() <<
" > " << current_slice_index + maximal_frame_index
973 <<
" module " << header.getModuleID()
974 <<
" -> Dropping frame.";
976 number_of_packets_discarded += 1;
981 deque<JDAQTimesliceL0>::iterator timesliceIterator = timeslices.begin();
983 while (timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() < header.getFrameIndex()) {
987 if (timesliceIterator != timeslices.end() && timesliceIterator->getFrameIndex() == header.getFrameIndex()) {
995 timesliceIterator = timeslices.insert(timesliceIterator,
JDAQTimesliceL0());
997 timesliceIterator->setDAQChronometer(header.getDAQChronometer());
1004 in >>
static_cast<JDAQFrame&
>(*(timesliceIterator->rbegin()));
1017 using namespace std;
1021 timesliceRouter->configure(timeslice);
1035 c_buffer.is_enabled()) {
1044 for (JDAQTimesliceL0::const_iterator frame = timeslice.begin(); frame != timeslice.end(); ++frame) {
1046 if (moduleRouter->hasModule(frame->getModuleID())) {
1053 <<
"module = " << frame->getModuleID() <<
";"
1054 <<
"discard" << (dumpCount < dumpLimit ?
" and dump" :
"");
1056 timesliceTX.push_back(*frame);
1061 const JModule& module = moduleRouter->getModule(frame->getModuleID());
1062 JSuperFrame2D_t& buffer = JSuperFrame2D_t::demultiplex(*frame, module);
1078 if (buffer.begin() != __end) {
1081 frame->getModuleIdentifier(),
1086 (*buildL1)(buffer.begin(), __end , back_inserter(zbuf));
1088 (*buildNB)(buffer.begin() , __end, zbuf, back_inserter(*timesliceNB.rbegin()));
1095 frame->getModuleIdentifier(),
1098 (*buildL1)(*timesliceL0.rbegin(), back_inserter(*timesliceL1.rbegin()));
1103 frame->getModuleIdentifier(),
1106 (*buildL2)(buffer, *timesliceL1.rbegin(), back_inserter(*timesliceL2.rbegin()));
1111 frame->getModuleIdentifier(),
1114 (*buildSN)(buffer, *timesliceL1.rbegin(), back_inserter(*timesliceSN.rbegin()));
1118 JErrorStream(logErrorDetector) <<
"No detector information for module " << frame->getModuleID();
1122 if (!timesliceTX.empty()) {
1124 if (dumpCount < dumpLimit) {
1126 this->put(timesliceTX);
1138 for (JTriggerInput::const_iterator hit = trigger_input.begin(); hit != trigger_input.end(); ++hit) {
1159 (*trigger3DMuon) (trigger_input, back_inserter(trigger_output));
1160 (*trigger3DShower)(trigger_input, back_inserter(trigger_output));
1161 (*triggerMXShower)(trigger_input, timesliceL0, back_inserter(trigger_output));
1165 for (JTriggerOutput::const_iterator event = trigger_output.begin();
event != trigger_output.end(); ++event) {
1171 numberOfEvents += 1;
1174 if (
parameters.writeL1() || c_buffer.sizeL1 > 0) {
1178 if (
parameters.writeL1) { this->put(
object); }
1179 if (c_buffer.sizeL1 > 0) { c_buffer.put(
object); }
1182 if (
parameters.writeL2() || c_buffer.sizeL2 > 0) {
1186 if (
parameters.writeL2) { this->put(
object); }
1187 if (c_buffer.sizeL2 > 0) { c_buffer.put(
object); }
1190 if (
parameters.writeSN() || c_buffer.sizeSN > 0) {
1194 if (
parameters.writeSN) { this->put(
object); }
1195 if (c_buffer.sizeSN > 0) { c_buffer.put(
object); }
1198 if (
parameters.writeL0() || c_buffer.sizeL0 > 0) {
1200 if (
parameters.writeL0) { this->put(timeslice); }
1201 if (c_buffer.sizeL0 > 0) { c_buffer.put(timeslice); }
1205 }
catch(
const exception& error) {
1207 JErrorStream(logger) <<
"Error = " << error.what() <<
";"
1210 <<
"time slice not correctly processed;"
1211 <<
"discard" << (dumpCount < dumpLimit ?
" and dump" :
"");
1213 if (dumpCount < dumpLimit) {
1215 this->put(static_cast<const JDAQTimeslice&>(timeslice));
1221 timesliceRouter->reset();
1232 const double T_us = (double) timer.usec_wall;
1234 JStatusStream(logger) <<
"Elapsed real (wall) time [s] " << T_us / 1e6;
1235 JStatusStream(logger) <<
"Elapsed user CPU time [s] " << (double) timer.usec_ucpu/ 1e6;
1236 JStatusStream(logger) <<
"Elapsed system CPU time [s] " << (double) timer.usec_scpu/ 1e6;
1238 JStatusStream(logger) <<
"Elapsed real time per time slice [ms] " << Qt.getMean() * 1.0e-3 <<
" +/- " << Qt.getDeviation() * 1.0e-3;
1240 catch(
const std::exception&) {}
1241 JStatusStream(logger) <<
"Number of packets received/discarded " << number_of_packets_received <<
"/" << number_of_packets_discarded;
1242 JStatusStream(logger) <<
"Number of events/MB sent " << numberOfEvents <<
"/" << numberOfBytes/1e6;
1244 if (number_of_packets_received > 0) {
1245 JStatusStream(logger) <<
"Number of reads/packet " << (double) number_of_reads / (
double) number_of_packets_received;
1249 JStatusStream(logger) <<
"Current number of frames per slice expected: " << frames_per_slice;
1251 JStatusStream(logger) <<
"Number of timeslices processed total/incomplete " << numberOfTimeslicesProcessed <<
"/" << numberOfIncompleteTimeslicesProcessed;
1253 if (numberOfTimeslicesProcessed > 0) {
1254 JStatusStream(logger) <<
"Real time per timeslice [ms] " << timer.usec_wall / 1000 / numberOfTimeslicesProcessed;
1255 JStatusStream(logger) <<
"User CPU time per timeslice [ms] " << timer.usec_ucpu / 1000 / numberOfTimeslicesProcessed;
1256 JStatusStream(logger) <<
"System CPU time per timeslice [ms] " << timer.usec_scpu / 1000 / numberOfTimeslicesProcessed;
1259 const double processedSlicesTime_us = numberOfTimeslicesProcessed *
getFrameTime() / 1000;
1260 const double processedDetectorTime_us = (maxFrameNumber - minFrameNumber) *
getFrameTime() / 1000;
1262 if (processedSlicesTime_us > 0) {
1263 JStatusStream(logger) <<
"Performance factor (inaccurate estimate): " << T_us / processedSlicesTime_us;
1265 if (processedDetectorTime_us > 0) {
1266 JStatusStream(logger) <<
"Performance factor whole detector (inaccurate estimate): " << T_us / processedDetectorTime_us;
1282 using namespace std;
1288 if (c_buffer.is_open()) {
1290 JStatusStream(logger) <<
"Archive circular buffer in <" << c_buffer.archive <<
">";
1292 c_buffer.close(
true);
1295 if (c_buffer.is_enabled()) {
1299 if (c_buffer.is_open()) {
1301 JStatusStream(logger) <<
"Created circular buffer " << c_buffer;
1307 JErrorStream (logger) <<
"Failed to create circular buffer in directory <" << c_buffer.path <<
">; disable functionality.";
1337 datawriter->put(
object);
1341 catch(
const std::exception& error) {
1342 JErrorStream(logger) <<
"Error \"" << error.what() <<
"\"; trigger ev_error.";
1435 using namespace std;
1436 using namespace JPP;
1437 using namespace KM3NETDAQ;
1453 JParser<> zap(
"Application for real-time filtering of data.");
1455 zap[
'H'] =
make_field(server,
"host name of server for command messages") =
"localhost";
1456 zap[
'M'] =
make_field(logger,
"host name of server for logger messages") =
"localhost";
1457 zap[
'D'] =
make_field(hostname,
"host name of server of data writer") =
"";
1458 zap[
'u'] =
make_field(client_name,
"client name") =
"%";
1459 zap[
'P'] =
make_field(port,
"port to listen for incoming data from data queue");
1460 zap[
'q'] =
make_field(backlog,
"back log") = 1024;
1461 zap[
'c'] =
make_field(use_cout,
"print to terminal");
1462 zap[
'p'] =
make_field(
path,
"directory for temporary storage of circular buffer") =
"/tmp/";
1463 zap[
'A'] =
make_field(archive,
"directory for permanent archival of circular buffer") =
"/tmp/";
1468 catch(
const exception& error) {
1469 FATAL(error.what() << endl);
virtual void actionRunning() override
This method is repeatedly called when this client machine is in state Running and the clock interval ...
Exception for opening of file.
Message logger with time scheduler.
static bool compare(const JDAQProcess &first, const JDAQProcess &second)
Sort DAQ process by index.
Utility class to parse command line options.
void processTimeSlice(const JDAQTimesliceL0 ×lice)
Process time slice.
int getSizeOfPacket(const KM3NETDAQ::JDAQAbstractPreamble &preamble)
Get size of packeet.
bool read(const JEquation &equation)
Read equation.
JTriggerMask_t getTriggerMask(const unsigned int bit)
Convert trigger bit to trigger mask.
int main(int argc, char *argv[])
void close(const bool option)
Close file.
ROOT TTree parameter settings of various packages.
Data structure for a composite optical module.
JSinglePointer< JTriggerNB > triggerNB
Router for fast addressing of hits in KM3NETDAQ::JDAQTimeslice data structure as a function of the op...
long long int numberOfTimeslicesProcessed
JBuildL2< hit_type > JBuildL2_t
Message logging based on std::ostream.
std::deque< JDAQTimesliceL0 > timeslices
buffer with pending time slice data
bool has(const int file_descriptor) const
Has file descriptor.
void set(const int file_descriptor)
Set file descriptor.
JMessageScheduler logErrorIndex
JSuperFrame2D< hit_type > JSuperFrame2D_t
JSocketInputChannel< JDAQAbstractPreamble > JSocketInputChannel_t
Long64_t sizeL1
Number of L1 time slices.
JTag getUniqueTag(const std::string &hostname, const std::string &name)
Get unique tag of run control client.
static const JNET::JTag RC_DFILTER
static const int DFILTER_SEND_BUFFER_SIZE
socket JDataFilter.cc -> JDataWriter.cc
Router for direct addressing of module data in detector data structure.
#define THROW(JException_t, A)
Marco for throwing exception with std::ostream compatible message.
Interface for logging messages.
Message logging based on ControlHost.
then echo Enter input within $TIMEOUT_S seconds echo n User name
Utility class to parse parameter values.
JSinglePointer< JServerSocket > serversocket
server for data queue connections
static const int HIGH_RATE_VETO_DISABLE
Enable (disable) use of high-rate veto test if this status bit is 0 (1);.
unsigned int maximum_frames_per_slice
ROOT TTree object output.
JMessageScheduler logErrorIncomplete
*fatal Wrong number of arguments esac JCookie sh typeset Z DETECTOR typeset Z SOURCE_RUN typeset Z TARGET_RUN set_variable PARAMETERS_FILE $WORKDIR parameters
Simple data structure to support I/O of equations (see class JLANG::JEquation).
then usage $script< detector file >< detectorfile > nIf the range of floors is the first detector file is aligned to the second before the comparison nIn this
static const JNET::JTag IO_TRIGGER_PARAMETERS
JTriggerParameters parameters
Main class for real-time filtering of data.
void run()
Run as run control client following command messages via JNET::JControlHost.
std::string getFullPath(const std::string &path)
Get full path, i.e. add JEEP::PATHNAME_SEPARATOR if necessary.
Basic data structure for time and time over threshold information of hit.
long long int number_of_packets_discarded
int getRunNumber() const
Get run number.
Data structure for detector geometry and calibration.
T * open(const std::string &file_name)
Open file.
Utility class to parse parameter values.
static const JChecksum checksum
Function object to perform check-sum of raw data.
then echo The file $DIR KM3NeT_00000001_00000000 root already please rename or remove it first
JCircularBuffer_t(const std::string &path, const std::string &archive, const JTag &tag)
Constructor.
std::string path
Directory for temporary storage.
void disable()
Disable writing.
int getFrameIndex() const
Get frame index.
JBuildL1< hit_type > JBuildL1_t
1-dimensional frame with time calibrated data from one optical module.
JCircularBuffer_t c_buffer
JTimeRange getTimeRange(const Evt &event)
Get time range (i.e. time between earliest and latest hit) of Monte Carlo event.
void typeout()
Report status to message logger.
bool is_enabled() const
Check whether writing of data is enabled.
Basic data structure for L0 hit.
std::string index
index in process list
JSuperFrame1D< hit_type > JSuperFrame1D_t
std::string trim(const std::string &buffer)
Trim string.
JSinglePointer< JBuildL2_t > buildL2
int getLength() const
Get length.
JSinglePointer< JTriggerMXShower > triggerMXShower
JSinglePointer< JBuildL1_t > buildL1
static const int DFILTER_RECEIVE_BUFFER_SIZE
socket JDataFilter.cc <- JLigier.cc
int getFileDescriptor() const
Get file descriptor.
long long int totalCPURAM
JTimeslice< hit_type > JTimeslice_t
JSinglePointer< JTrigger3DShower > trigger3DShower
Scheduling of actions via fixed latency intervals.
then echo Variable JPP_DIR undefined exit fi source $JPP_DIR setenv sh $JPP_DIR set_variable DIR $JPP_DIR examples JPhysics set_variable DEBUG set_variable darkcount_Hz set_variable rateL0_Hz set_array rateL1_Hz if do_usage *then usage $script[working directory] fi function typeout()
TFile * getFile(const std::string &file_name, const std::string &option="exist")
Get TFile pointer corresponding to give file name.
I/O formatting auxiliaries.
void accept(const int server)
Accept connection from a server.
virtual void actionContinue(int length, const char *buffer) override
long long int numberOfBytes
Auxiliary class for itemization of process list.
std::vector< JDAQProcess > dataFilters
void merge(const JMatch_t &match)
Merge events.
std::vector< JSocketInputChannel_t > JChannelList_t
friend std::ostream & operator<<(std::ostream &out, const JCircularBuffer_t &object)
Write circular buffer to output stream.
static const JLocalTime getLocalTime
Function object to get local time in micro seconds.
event< ev_daq > ev_configure
#define make_field(A,...)
macro to convert parameter to JParserTemplateElement object
void close(std::istream *pf)
Close file.
static const int DFILTER_DQUEUE_BUFFER_SIZE
socket JDataFilter.cc <- DataQueue.cc
double getFrameTime()
Get frame time duration.
do set_variable OUTPUT_DIRECTORY $WORKDIR T
int getRunNumber(const std::string &file_name)
Get run number for given file name of data taking run.
JMessageScheduler logErrorDetector
Auxiliary class for method select.
void put(const T &object)
Auxiliary method to send object to data server.
JSinglePointer< JBuildL2_t > buildSN
const JDAQChronometer & getDAQChronometer() const
Get DAQ chronometer.
long long int numberOfIncompleteTimeslicesProcessed
Auxiliary class to build JDAQEvent for a triggered event.
void updateFrameQueue(const JChannelList_t::const_iterator channel)
Update queue with data frames.
JTreeWriter object output.
JSinglePointer< JBuildL2_t > buildNB
virtual void actionEnter() override
Interface methods for actions corresponding to state transitions.
JTag tag
Unique tag of this process.
Level specific message streamers.
long long int number_of_packets_received
$WORKDIR driver txt done cat $WORKDIR driver txt<< EOFprocess ${DATAFILTER}$FILTER_HOST csh-c '(setenv ROOTSYS $ROOTSYS &&source $JPP_DIR/setenv.csh $JPP_DIR &&(JDataFilter-H\$SERVER\$-M\$LOGGER\$-d $DEBUG-u ${DATAFILTER}-P $PORT</dev/null > &/dev/null &))';process ${DATAWRITER}$WRITER_HOST csh-c '(setenv ROOTSYS $ROOTSYS &&source $JPP_DIR/setenv.csh $JPP_DIR &&(JDataWriter-H\$SERVER\$-M\$LOGGER\$-d $DEBUG-u ${DATAWRITER}</dev/null > &/dev/null &))';print enterevent ev_init{RC_CMD}event ev_reset{RC_CMD}event ev_init{RC_CMD}event ev_configure{RC_DFLTR%<$WORKDIR/ev_configure_datafilter.txt > RC_DQSIM<$WORKDIR/ev_configure_dqsimulator.txt > RC_DWRT path
bool putObject(TDirectory &dir, const TObject &object)
Write object to ROOT directory.
const JPosition3D & getPosition() const
Get position.
Simple data structure for the DAQ preamble required for a correct calculation of the object size for ...
virtual void actionQuit(int length, const char *buffer) override
Long64_t sizeSN
Number of SN time slices.
Auxiliary class for all subscription.
int port
server socket port
static void reset()
Reset counter of unique instance of this class object.
General purpose messaging.
Auxiliary data structure for sequence of same character.
long long int localtime_t
Type definition of local time.
void setReceiveBufferSize(const int size)
Set receive buffer size.
std::string hostname
host name of data server
unsigned int maxQueueDepth
JControlHostObjectOutput< JDAQTypes_t > JControlHost_t
static JStat getFileStatus
Function object for file status.
std::vector< JDAQProcess > dataQueues
long long int maxQueueSize
virtual void actionStart(int length, const char *buffer) override
std::string toString() const
Convert tag to string.
std::vector< value_type >::iterator iterator
virtual void actionReset(int length, const char *buffer) override
virtual void actionExit() override
JSinglePointer< JTimesliceRouter > timesliceRouter
static const long long int GIGABYTE
Number of bytes in a mega-byte.
Auxiliary class to build JDAQTimeslice for L1 timeslice.
JChannelList_t channelList
connections to data queue
Control unit client base class.
Auxiliary class for date and time.
Utility class to parse command line options.
static const JNET::JTag RC_ALERT
Implemenation of object output through ControlHost.
virtual void actionSelect(const JFileDescriptorMask &mask) override
Action method following last select call.
void applyHighRateVeto(const double rate_Hz)
Apply high-rate veto.
std::vector< std::string > getListOfIPaddresses()
Get list of IP address (decimal-dot notation).
unsigned long long int getRAM()
Get RAM of this CPU.
Long64_t sizeL0
Number of L0 time slices.
JSinglePointer< JTrigger3DMuon > trigger3DMuon
JSinglePointer< JControlHost_t > datawriter
controlhost of data server (to which data writer should be connected)
double getMaximalDistance(const JDetector &detector, const bool option=false)
Get maximal distance between modules in detector.
JMessageScheduler logErrorRun
virtual bool enter(const JArgs &args)
Enter the state machine.
virtual void actionTagged(const JTag &tag, int length, const char *buffer) override
Tagged action to handle alerts.
JSinglePointer< JModuleRouter > moduleRouter
2-dimensional frame with time calibrated data from one optical module.
void setNonBlocking(const bool on)
Set non-blocking of I/O.
JSocketInputChannel< JPrefix_t > JSocketInputChannel_t
const char * getName()
Get ROOT name of given data type.
static const JNET::JTag RC_CMD
void setKeepAlive(const bool on)
Set keep alive of socket.
long long int number_of_bytes_received
virtual void actionStop(int length, const char *buffer) override
do set_variable DETECTOR_TXT $WORKDIR detector
KM3NeT DAQ constants, bit handling, etc.
JDataFilter(const std::string &name, const std::string &server, const std::string &hostname, JLogger *logger, const int level, const int port, const int backlog, const std::string &path, const std::string &archive)
Constructor.
virtual void actionInit(int length, const char *buffer) override
then fatal Wrong number of arguments fi set_variable DETECTOR $argv[1] set_variable INPUT_FILE $argv[2] eval JPrintDetector a $DETECTOR O IDENTIFIER eval JPrintDetector a $DETECTOR O SUMMARY JAcoustics sh $DETECTOR_ID source JAcousticsToolkit sh CHECK_EXIT_CODE typeset A EMITTERS get_tripods $WORKDIR tripod txt EMITTERS get_transmitters $WORKDIR transmitter txt EMITTERS for EMITTER in
Match of two events considering overlap in time.
size_t getSizeof(const JDAQEvent &object)
Get size of object.
virtual void actionPause(int length, const char *buffer) override
Hostname and IP address functions.
Long64_t sizeL2
Number of L2 time slices.
Data frame of one optical module.
long long int number_of_reads
long long int numberOfEvents
unsigned int frames_per_slice
Timeslice data structure for L0 data.
Basic data structure for L1 hit.
std::string archive
Directory for permanent archival.
virtual void setSelect(JFileDescriptorMask &mask) const override
Set the file descriptor mask for the select call.
virtual void actionConfigure(int length, const char *buffer) override
Time slice with calibrated data.