Jpp  18.0.1-rc.2
the software that should make you happy
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
JDataWriter.cc
Go to the documentation of this file.
1 #include <string>
2 #include <iostream>
3 #include <sstream>
4 #include <iomanip>
5 #include <map>
6 
7 #include "Jeep/JParser.hh"
8 #include "Jeep/JProperties.hh"
9 #include "Jeep/JTimer.hh"
10 #include "Jeep/JTimekeeper.hh"
11 #include "Jeep/JPrint.hh"
12 #include "Jeep/JeepToolkit.hh"
13 #include "JNet/JControlHost.hh"
14 #include "JLang/JException.hh"
15 #include "JLang/JSharedPointer.hh"
18 #include "JDAQ/JDAQTags.hh"
19 #include "JDAQ/JDAQEventIO.hh"
20 #include "JDAQ/JDAQTimesliceIO.hh"
25 #include "JIO/JByteArrayIO.hh"
26 #include "JTools/JAutoMap.hh"
27 #include "JSupport/JMeta.hh"
28 #include "JSupport/JSupport.hh"
30 
31 
32 /**
33  * Type definition of auto map.
34  */
36 
37 
38 namespace JSUPPORT {
39 
40  /**
41  * Get key for given DAQ data type.
42  *
43  * \param type data type
44  * \return map element
45  */
46  template<>
47  template<class T>
49  {
50  return getTag<T>();
51  }
52 }
53 
54 
55 namespace KM3NETDAQ {
56 
57  /**
58  * Runcontrol client to write data to disk.
59  * In state running, this application will write ROOT formatted data from the data filters to disk.
60  */
61  class JDataWriter :
62  public JDAQClient
63  {
64  public:
65  /**
66  * Constructor.
67  *
68  * \param name name of client
69  * \param server name of command message server
70  * \param hostname name of data server
71  * \param logger pointer to logger
72  * \param level debug level
73  * \param path default path
74  */
76  const std::string& server,
77  const std::string& hostname,
78  JLogger* logger,
79  const int level,
80  const std::string& path) :
81  JDAQClient(name, server, logger, level),
82  datawriter(),
83  path (path),
84  hostname (hostname)
85  {
87 
88  JControlHost::Throw(true);
89 
90  // map ControlHost tag to TTree writer.
91 
93  }
94 
95 
96  virtual void actionInit(int length, const char* buffer) override
97  {
98  using namespace std;
99  using namespace JPP;
100 
101  // start server
102 
103  try {
104 
105  datawriter.reset(new JControlHost(hostname));
106 
107  datawriter->setReceiveBufferSize(DWRITER_RECEIVE_BUFFER_SIZE);
108 
110 
111  for (JTreeWriter_t::iterator i = writer.begin(); i != writer.end(); ++i) {
112  buffer.add(JSubscriptionAll(i->first));
113  }
114 
116 
117  datawriter->Subscribe(buffer);
118  datawriter->SendMeAlways();
119 
120  JNoticeStream(logger) << "Established connection to " << hostname;
121  }
122  catch(const JControlHostException& exception) {
123  JErrorStream(logger) << exception;
124  }
125  }
126 
127 
128  virtual void actionConfigure(int length, const char* buffer) override
129  {
130  using namespace std;
131 
132  long long int update_s = 10;
133  long long int logger_s = 5;
134 
135  JProperties properties(JEquationParameters("=", ";", "", ""));
136 
137  properties["path"] = path;
138  properties["update_s"] = update_s;
139  properties["logger_s"] = logger_s;
140 
141  properties.read(string(buffer, length));
142 
143  if (update_s <= 0) { update_s = 1; }
144  if (logger_s <= 0) { logger_s = 1; }
145 
146  setClockInterval(update_s * 1000000LL);
147 
148  JDebugStream(logger) << "Path <" << path << ">";
149  JDebugStream(logger) << "Update period [s] " << update_s;
150 
155 
156  numberOfEvents = 0;
157  numberOfBytes = 0;
158  }
159 
160 
161  virtual void actionReset(int length, const char* buffer) override
162  {
163  datawriter.reset();
164  }
165 
166 
167  virtual void actionQuit(int length, const char* buffer) override
168  {}
169 
170 
171  virtual void actionStart(int length, const char* buffer) override
172  {
173  using namespace std;
174  using namespace JPP;
175 
176  JStatusStream(logger) << "Start run " << getDetectorID() << ' ' << getRunNumber();
177 
178  if (writer.is_open()) {
179 
180  JErrorStream (logger) << "Previous file still open -> close";
181 
182  writer.close();
183  }
184 
185  ostringstream os;
186 
187  for (int i = 0; !writer.is_open() && i != MAXIMUM_FILE_NUMBER; ++i) {
188 
189  os.str("");
190 
191  os << getFullPath(path)
192  << "KM3NeT"
193  << "_" << FILL(8,'0') << getDetectorID()
194  << "_" << FILL(8,'0') << getRunNumber();
195 
196  if (i != 0) {
197  os << "_" << i;
198  }
199 
200  os << ".root";
201 
202  try {
203  writer.open(os.str().c_str());
204  }
205  catch(JException& exception) {
206  JErrorStream(logger) << exception;
207  }
208  }
209 
210  if (writer.is_open())
211  JNoticeStream(logger) << "Output file " << os.str();
212  else
213  JErrorStream (logger) << "File not opened " << os.str();
214 
215  numberOfEvents = 0;
216  numberOfBytes = 0;
217 
219 
220  timer.reset();
221 
222  logErrorRun .reset();
223  logErrorFile .reset();
224  logErrorTag .reset();
226 
228  }
229 
230 
231  virtual void actionStop(int length, const char* buffer) override
232  {
233  typeout();
234 
235  if (timer.usec_wall > 0) {
236  JStatusStream(logger) << "I/O " << (int) (numberOfBytes / timer.usec_wall) << " MB/s";
237  }
238 
239  if (!run_db.is_written(getRunNumber())) {
240  JErrorStream(logger) << "No trigger parameters written for run " << getRunNumber();
241  }
242 
243  writer.close();
244 
245  // Release resources.
246 
248 
249  this->buffer.swap(null);
250  }
251 
252 
253  virtual void setSelect(JFileDescriptorMask& mask) const override
254  {
255  if (datawriter.is_valid()) {
256  mask.set(*datawriter);
257  }
258  }
259 
260 
261  virtual void actionSelect(const JFileDescriptorMask& mask) override
262  {
263  using namespace std;
264  using namespace JPP;
265 
266  if (datawriter.is_valid() && mask.has(*datawriter)) {
267 
268  try {
269 
270  JPrefix prefix;
271 
272  datawriter->WaitHead(prefix);
273 
274  timer.start();
275 
276  buffer.resize(prefix.getSize());
277 
278  datawriter->GetFullData(buffer.data(), buffer.size());
279 
280 
281  if (prefix.getTag() == IO_TRIGGER_PARAMETERS) {
282 
283  try {
284  run_db.read(buffer.data(), buffer.size());
285  }
286  catch(const exception& error) {
287 
288  JErrorStream(logger) << "Fatal error reading trigger parameters " << error.what();
289 
290  ev_error();
291  }
292  }
293 
294 
295  if (isRunning()) {
296 
297  // Write trigger parameters for current run if not yet done
298 
300 
301  JTreeWriter_t::iterator i = writer.find(prefix.toString());
302 
303  if (i != writer.end()) {
304 
305  TFile* out = i->second->GetCurrentFile();
306 
307  if (out != NULL && out->IsOpen()) {
308 
309  JDAQPreamble preamble;
310  Version_t version;
311  JDAQHeader header;
312 
313  JByteArrayReader in(buffer.data(), buffer.size());
314 
315  in >> preamble >> version >> header;
316 
317  in.seekg(0); // rewind
318 
319  if (header.getRunNumber() == getRunNumber()) {
320 
321  const Int_t nb = i->second->copy(in);
322 
323  if (nb < (int) buffer.size() || in.tellg() != (int) buffer.size()) {
324  JWarningStream(logger) << "Inconsistency at copy of "
325  << prefix.toString() << ' '
326  << buffer.size() << ' '
327  << in.tellg() << ' '
328  << nb;
329  }
330 
331  if (prefix.getTag() == IO_EVENT)
332  numberOfEvents += 1;
333  numberOfBytes += buffer.size();
334 
335  if (prefix.getTag() == IO_EVENT && numberOfEvents == 1) {
336  typeout();
337  }
338 
339  } else {
340  JErrorStream(logErrorRun) << "Inconsistent run number "
341  << header.getRunNumber()
342  << " != "
343  << getRunNumber();
344  }
345  } else {
346  JErrorStream(logErrorFile) << "Output file not open";
347  }
348  } else {
349  if (prefix.getTag() != IO_TRIGGER_PARAMETERS) {
350  JErrorStream(logErrorTag) << "Unknown tag <" << prefix.toString() << ">, no data written";
351  }
352  }
353  } else {
354  JWarningStream(logErrorState) << "Not in running state <" << prefix.toString() << ">, no data written";
355  }
356 
357  timer.stop();
358  }
359  catch(const JControlHostException& exception) {
360  JErrorStream(logger) << exception;
361  }
362  }
363  }
364 
365 
366  virtual void actionRunning() override
367  {
368  typeout();
369  }
370 
371 
372  /**
373  * Report status of data writing.
374  */
375  void typeout()
376  {
377  std::ostringstream message;
378 
380 
381  logger.typeout(RC_LOG, message.str());
382  logger.status(message.str());
383  }
384 
385  JMeta meta; //!< meta data
386 
387  static const int MAXIMUM_FILE_NUMBER = 100; //!< maximum file number for overwrite protection.
388 
389  private:
390 
392  std::string path; // directory for output file
393 
394  JEEP::JTimer timer; // timer for I/O measurement
395  Long64_t numberOfEvents; // total number of events
396  long long int numberOfBytes; // total number of bytes
397 
402 
403  std::string hostname; //!< host name of data server
404  JTreeWriter_t writer; //!< TTree writer
405  std::vector<char> buffer; //!< internal buffer for incoming data
406 
407 
408  /**
409  * Auxiliary data structure for I/O of trigger parameters.
410  */
411  struct JValue_t {
412  /**
413  * Default constructor.
414  */
416  count(0),
417  is_written(false)
418  {}
419 
420  JTriggerParameters parameters; //!< trigger parameters
421  int count; //!< reader count
422  bool is_written; //!< writer status
423  };
424 
425 
426  /**
427  * Map run number to trigger parameters.
428  */
429  struct JRunDB :
430  public std::map<int, JValue_t>
431  {
432  /**
433  * Remove all entries before given run.
434  *
435  * \param run run number
436  */
437  inline void reset(const int run)
438  {
439  while (!this->empty() && this->begin()->first < run) {
440  this->erase(this->begin());
441  }
442  }
443 
444  /**
445  * Check if trigger parameters have been written for given run.
446  *
447  * \param run run number
448  * \return true if written; else false.
449  */
450  inline bool is_written(const int run) const
451  {
452  const_iterator p = this->find(run);
453 
454  return p != this->end() && p->second.is_written;
455  }
456 
457  /**
458  * Read trigger parameters.
459  *
460  * \param data data
461  * \param size size
462  */
463  void read(const char* const data, const size_t size)
464  {
465  using namespace std;
466  using namespace JPP;
467 
468 
469  const string buffer(data, size);
470 
471  istringstream in(buffer);
472 
473  int run = -1;
475 
476  in >> run;
477 
478  if (!in) {
479  THROW(JIOException, "Error reading run number for trigger parameters " << run << endl << in.rdbuf());
480  }
481 
482  in >> parameters;
483 
484  in.clear(std::ios::eofbit);
485 
486  if (!in) {
487  THROW(JIOException, "Error reading trigger parameters " << in.rdbuf());
488  }
489 
490  JValue_t& value = (*this)[run];
491 
492  if (value.count == 0) {
493  value.parameters = parameters;
494  }
495 
496  value.count += 1;
497 
498  if (!parameters.equals(value.parameters)) {
499  THROW(JException, "Inconsistent trigger parameters " << endl << value.parameters << " != " << endl << parameters);
500  }
501  }
502 
503  /**
504  * Write trigger parameters for given run if not yet done.
505  *
506  * \param run run number
507  * \param file pointer to ROOT file
508  */
509  inline void write(const int run, TFile* file)
510  {
511  if (file != NULL) {
512 
513  iterator p = this->find(run);
514 
515  if (p != this->end() && p->second.count != 0 && !p->second.is_written) {
516 
517  file->WriteTObject(&p->second.parameters);
518 
519  p->second.is_written = true;
520  }
521  }
522  }
523  };
524 
526  };
527 }
528 
529 
530 /**
531  * \file
532  *
533  * Application for writing real-time data to disk.
534  * \author mdejong
535  */
536 int main(int argc, char* argv[])
537 {
538  using namespace std;
539  using namespace JPP;
540  using namespace KM3NETDAQ;
541 
542  string server;
543  string logger;
544  string hostname;
545  string client_name;
546  bool use_cout;
547  string path;
548  int debug;
549 
550  try {
551 
552  JParser<> zap("Application for writing real-time data to disk.");
553 
554  zap['H'] = make_field(server, "host name of server for command messages") = "localhost";
555  zap['M'] = make_field(logger, "host name of server for logger messages") = "localhost";
556  zap['D'] = make_field(hostname, "host name of server for incoming data from data filter") = "localhost";
557  zap['u'] = make_field(client_name, "client name") = "%";
558  zap['c'] = make_field(use_cout, "print to terminal");
559  zap['p'] = make_field(path, "directory for permanent archival of data") = "";
560  zap['d'] = make_field(debug, "debug level") = 0;
561 
562 
563  zap(argc, argv);
564  }
565  catch(const exception &error) {
566  FATAL(error.what() << endl);
567  }
568 
569 
570  JLogger* out = NULL;
571 
572  if (use_cout)
573  out = new JStreamLogger(cout);
574  else
575  out = new JControlHostLogger(logger);
576 
577  JDataWriter dwriter(getProcessName(client_name, argv[0]), server, hostname, out, debug, path);
578 
579  dwriter.meta = JMeta(argc, argv);
580 
581  dwriter.enter();
582  dwriter.run();
583 }
JDataWriter(const std::string &name, const std::string &server, const std::string &hostname, JLogger *logger, const int level, const std::string &path)
Constructor.
Definition: JDataWriter.cc:75
ControlHost prefix.
Definition: JPrefix.hh:31
Message logger with time scheduler.
Auxiliary class for ROOT I/O of application specific meta data.
Definition: JMeta.hh:70
Utility class to parse command line options.
Definition: JParser.hh:1514
virtual void actionStop(int length, const char *buffer) override
Definition: JDataWriter.cc:231
General exception.
Definition: JException.hh:23
Runcontrol client to write data to disk.
Definition: JDataWriter.cc:61
bool read(const JEquation &equation)
Read equation.
Definition: JProperties.hh:677
Exceptions.
std::vector< char > buffer
internal buffer for incoming data
Definition: JDataWriter.cc:405
void status(const JMessage_t &message)
int main(int argc, char *argv[])
Definition: Main.cc:15
static const int MAXIMUM_FILE_NUMBER
maximum file number for overwrite protection.
Definition: JDataWriter.cc:387
ROOT TTree parameter settings of various packages.
virtual void actionQuit(int length, const char *buffer) override
Definition: JDataWriter.cc:167
JDAQStateMachine::ev_configure_event ev_configure
JDAQStateMachine::ev_error_event ev_error
ControlHost class.
then usage $script[< detector identifier >< run range >]< QA/QCfile > nExample script to produce data quality plots nWhen a detector identifier and run range are data are downloaded from the database nand subsequently stored in the given QA QC file
Definition: JDataQuality.sh:19
std::string getProcessName(const std::string &name, const std::string &process)
Get process name of run control client.
Message logging based on std::ostream.
version
Definition: JEditTuneHV.sh:5
bool has(const int file_descriptor) const
Has file descriptor.
void set(const int file_descriptor)
Set file descriptor.
void read(const char *const data, const size_t size)
Read trigger parameters.
Definition: JDataWriter.cc:463
void reset(const int run)
Remove all entries before given run.
Definition: JDataWriter.cc:437
JMessageScheduler logErrorState
Definition: JDataWriter.cc:401
#define THROW(JException_t, A)
Marco for throwing exception with std::ostream compatible message.
Definition: JException.hh:696
Interface for logging messages.
Definition: JLogger.hh:22
Message logging based on ControlHost.
Utility class to parse parameter values.
Definition: JProperties.hh:496
void setClockInterval(const long long int interval_us)
Set interval time.
Definition: JDAQClient.hh:165
Auxiliary data structure for I/O of trigger parameters.
Definition: JDataWriter.cc:411
std::string hostname
host name of data server
Definition: JDataWriter.cc:403
*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
Definition: diff-Tuna.sh:38
Simple data structure to support I/O of equations (see class JLANG::JEquation).
Time keeper.
Definition: JTimekeeper.hh:34
static const JNET::JTag IO_TRIGGER_PARAMETERS
Definition: JDAQTags.hh:82
JTreeWriter_t writer
TTree writer.
Definition: JDataWriter.cc:404
std::string name
Definition: JDAQCHSM.chsm:154
void replaceEvent(const JTag &oldTag, const JTag &newTag, JDAQEvent_t &event)
Replace tag of given event in event table.
Definition: JDAQClient.hh:518
JSharedPointer< JControlHost > server
message server
Definition: JDAQClient.hh:826
Auxiliary class for a type holder.
Definition: JType.hh:19
void run()
Run as run control client following command messages via JNET::JControlHost.
Definition: JDAQClient.hh:687
std::string getFullPath(const std::string &path)
Get full path, i.e. add JEEP::PATHNAME_SEPARATOR if necessary.
Definition: JeepToolkit.hh:168
bool is_written(const int run) const
Check if trigger parameters have been written for given run.
Definition: JDataWriter.cc:450
virtual void setSelect(JFileDescriptorMask &mask) const override
Set the file descriptor mask for the select call.
Definition: JDataWriter.cc:253
int getSize() const
Get size.
Definition: JPrefix.hh:62
Subscription list.
void stop()
Stop timer.
Definition: JTimer.hh:113
virtual void actionSelect(const JFileDescriptorMask &mask) override
Action method following last select call.
Definition: JDataWriter.cc:261
Utility class to parse parameter values.
virtual bool is_open() const override
Check is file is open.
Definition: JRootFile.hh:77
then echo The file $DIR KM3NeT_00000001_00000000 root already please rename or remove it first
virtual void actionConfigure(int length, const char *buffer) override
Definition: JDataWriter.cc:128
long long int numberOfBytes
Definition: JDataWriter.cc:396
virtual void actionRunning() override
This method is repeatedly called when this client machine is in state Running and the clock interval ...
Definition: JDataWriter.cc:366
void reset()
Reset timer.
Definition: JTimer.hh:76
Map run number to trigger parameters.
Definition: JDataWriter.cc:429
JMessageScheduler logErrorRun
Definition: JDataWriter.cc:398
Type list.
Definition: JTypeList.hh:22
int getDetectorID() const
Get detector identifier.
Definition: JDAQCHSM.chsm:89
static JKey_t getKey(JType< T > type)
Get key.
TFile * getFile() const
Get file.
Definition: JRootFile.hh:66
Scheduling of actions via fixed latency intervals.
virtual void actionInit(int length, const char *buffer) override
Definition: JDataWriter.cc:96
I/O formatting auxiliaries.
The template JSharedPointer class can be used to share a pointer to an object.
static const JNET::JTag RC_LOG
Definition: JDAQTags.hh:64
JMessageScheduler logErrorFile
Definition: JDataWriter.cc:399
JSUPPORT::JAutoTreeWriter< JNET::JTag > JTreeWriter_t
Type definition of auto map.
Definition: JDataWriter.cc:35
#define make_field(A,...)
macro to convert parameter to JParserTemplateElement object
Definition: JParser.hh:1989
Byte array binary input.
Definition: JByteArrayIO.hh:25
JValue_t()
Default constructor.
Definition: JDataWriter.cc:415
Auxiliary methods for handling file names, type names and environment.
Auxiliary class for method select.
Auxiliary class for CPU timing and usage.
Definition: JTimer.hh:32
ROOT I/O of application specific meta data.
then awk string
JMessageScheduler logErrorTag
Definition: JDataWriter.cc:400
virtual void typeout(const std::string &tag, const std::string &message) override
Report message.
JSubscriptionList & add(const JSubscription &subscription)
Add subscription.
Level specific message streamers.
Exception for ControlHost.
Definition: JException.hh:468
JTriggerParameters parameters
trigger parameters
Definition: JDataWriter.cc:420
bool putObject(TDirectory &dir, const TObject &object)
Write object to ROOT directory.
char getTokenDelimeter()
Get the token delimeter for command messages.
bool isRunning() const
Check if this client is in runnig state.
Definition: JDAQClient.hh:505
virtual void close() override
Close file.
Auxiliary class for all subscription.
Definition: JControlHost.hh:97
virtual void open(const char *file_name) override
Open file.
const JTag & getTag() const
Get tag.
Definition: JTag.hh:86
Auxiliary data structure for sequence of same character.
Definition: JManip.hh:328
JMeta meta
meta data
Definition: JDataWriter.cc:385
static const JNET::JTag RC_DWRITER
Definition: JDAQTags.hh:63
#define FATAL(A)
Definition: JMessage.hh:67
virtual void actionStart(int length, const char *buffer) override
Definition: JDataWriter.cc:171
void write(const int run, TFile *file)
Write trigger parameters for given run if not yet done.
Definition: JDataWriter.cc:509
std::string toString() const
Convert tag to string.
Definition: JTag.hh:171
int getRunNumber() const
Get run number.
Definition: JDAQCHSM.chsm:100
void insert()
Insert (list of) data type(s).
void typeout()
Report status of data writing.
Definition: JDataWriter.cc:375
const std::string & getFullName() const
Get full name of this run control client.
Definition: JDAQClient.hh:120
Control unit client base class.
Definition: JDAQClient.hh:298
Utility class to parse command line options.
virtual bool enter(const JArgs &args)
Enter the state machine.
Definition: JDAQClient.hh:389
JMessageLogger logger
message logger
Definition: JDAQClient.hh:827
Fixed parameters and ControlHost tags for KM3NeT DAQ.
static const JNET::JTag RC_CMD
Definition: JDAQTags.hh:58
static const JNET::JTag IO_EVENT
Definition: JDAQTags.hh:80
virtual void actionReset(int length, const char *buffer) override
Definition: JDataWriter.cc:161
ControlHost tag.
Definition: JTag.hh:38
KM3NeT DAQ constants, bit handling, etc.
static JNullStream null
Null I/O stream.
Definition: JNullStream.hh:51
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
Definition: JCanberra.sh:46
JLANG::JSharedPointer< JControlHost > datawriter
Definition: JDataWriter.cc:391
static void Throw(const bool option)
Enable/disable throw option.
Definition: JThrow.hh:37
int debug
debug level
unsigned long long usec_wall
Definition: JTimer.hh:224
static const int DWRITER_RECEIVE_BUFFER_SIZE
socket JDataWriter.cc &lt;- JDataFilter.cc
Definition: JDAQTags.hh:31
void start()
Start timer.
Definition: JTimer.hh:89
Exception for I/O.
Definition: JException.hh:324