Jpp  17.3.0-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 
108 
109  for (JTreeWriter_t::iterator i = writer.begin(); i != writer.end(); ++i) {
110  buffer.add(JSubscriptionAll(i->first));
111  }
112 
114 
115  datawriter->Subscribe(buffer);
116  datawriter->SendMeAlways();
117 
118  JNoticeStream(logger) << "Established connection to " << hostname;
119  }
120  catch(const JControlHostException& exception) {
121  JErrorStream(logger) << exception;
122  }
123  }
124 
125 
126  virtual void actionConfigure(int length, const char* buffer) override
127  {
128  using namespace std;
129 
130  long long int update_s = 10;
131  long long int logger_s = 5;
132 
133  JProperties properties(JEquationParameters("=", ";", "", ""));
134 
135  properties["path"] = path;
136  properties["update_s"] = update_s;
137  properties["logger_s"] = logger_s;
138 
139  properties.read(string(buffer, length));
140 
141  if (update_s <= 0) { update_s = 1; }
142  if (logger_s <= 0) { logger_s = 1; }
143 
144  setClockInterval(update_s * 1000000LL);
145 
146  JDebugStream(logger) << "Path <" << path << ">";
147  JDebugStream(logger) << "Update period [s] " << update_s;
148 
153 
154  numberOfEvents = 0;
155  numberOfBytes = 0;
156  }
157 
158 
159  virtual void actionReset(int length, const char* buffer) override
160  {
161  datawriter.reset();
162  }
163 
164 
165  virtual void actionQuit(int length, const char* buffer) override
166  {}
167 
168 
169  virtual void actionStart(int length, const char* buffer) override
170  {
171  using namespace std;
172  using namespace JPP;
173 
174  JStatusStream(logger) << "Start run " << getDetectorID() << ' ' << getRunNumber();
175 
176  if (writer.is_open()) {
177 
178  JErrorStream (logger) << "Previous file still open -> close";
179 
180  writer.close();
181  }
182 
183  ostringstream os;
184 
185  for (int i = 0; !writer.is_open() && i != MAXIMUM_FILE_NUMBER; ++i) {
186 
187  os.str("");
188 
189  os << getFullPath(path)
190  << "KM3NeT"
191  << "_" << FILL(8,'0') << getDetectorID()
192  << "_" << FILL(8,'0') << getRunNumber();
193 
194  if (i != 0) {
195  os << "_" << i;
196  }
197 
198  os << ".root";
199 
200  try {
201  writer.open(os.str().c_str());
202  }
203  catch(JException& exception) {
204  JErrorStream(logger) << exception;
205  }
206  }
207 
208  if (writer.is_open())
209  JNoticeStream(logger) << "Output file " << os.str();
210  else
211  JErrorStream (logger) << "File not opened " << os.str();
212 
213  numberOfEvents = 0;
214  numberOfBytes = 0;
215 
217 
218  timer.reset();
219 
220  logErrorRun .reset();
221  logErrorFile .reset();
222  logErrorTag .reset();
224 
226  }
227 
228 
229  virtual void actionStop(int length, const char* buffer) override
230  {
231  typeout();
232 
233  if (timer.usec_wall > 0) {
234  JNoticeStream(logger) << "I/O " << (int) (numberOfBytes / timer.usec_wall) << " MB/s";
235  }
236 
237  if (!run_db.is_written(getRunNumber())) {
238  JErrorStream(logger) << "No trigger parameters written for run " << getRunNumber();
239  }
240 
241  writer.close();
242 
243  // Release resources.
244 
246 
247  this->buffer.swap(null);
248  }
249 
250 
251  virtual void setSelect(JFileDescriptorMask& mask) const override
252  {
253  if (datawriter.is_valid()) {
254  mask.set(*datawriter);
255  }
256  }
257 
258 
259  virtual void actionSelect(const JFileDescriptorMask& mask) override
260  {
261  using namespace std;
262  using namespace JPP;
263 
264  if (datawriter.is_valid() && mask.has(*datawriter)) {
265 
266  try {
267 
268  JPrefix prefix;
269 
270  datawriter->WaitHead(prefix);
271 
272  timer.start();
273 
274  buffer.resize(prefix.getSize());
275 
276  datawriter->GetFullData(buffer.data(), buffer.size());
277 
278 
279  if (prefix.getTag() == IO_TRIGGER_PARAMETERS) {
280 
281  try {
282  run_db.read(buffer.data(), buffer.size());
283  }
284  catch(const exception& error) {
285 
286  JErrorStream(logger) << "Fatal error reading trigger parameters " << error.what();
287 
288  ev_error();
289  }
290  }
291 
292 
293  if (isRunning()) {
294 
295  // Write trigger parameters for current run if not yet done
296 
298 
299  JTreeWriter_t::iterator i = writer.find(prefix.toString());
300 
301  if (i != writer.end()) {
302 
303  TFile* out = i->second->GetCurrentFile();
304 
305  if (out != NULL && out->IsOpen()) {
306 
307  JDAQPreamble preamble;
308  Version_t version;
309  JDAQHeader header;
310 
311  JByteArrayReader in(buffer.data(), buffer.size());
312 
313  in >> preamble >> version >> header;
314 
315  in.seekg(0); // rewind
316 
317  if (header.getRunNumber() == getRunNumber()) {
318 
319  const Int_t nb = i->second->copy(in);
320 
321  if (nb < (int) buffer.size() || in.tellg() != (int) buffer.size()) {
322  JWarningStream(logger) << "Inconsistency at copy of "
323  << prefix.toString() << ' '
324  << buffer.size() << ' '
325  << in.tellg() << ' '
326  << nb;
327  }
328 
329  if (prefix.getTag() == IO_EVENT)
330  numberOfEvents += 1;
331  numberOfBytes += buffer.size();
332 
333  if (prefix.getTag() == IO_EVENT && numberOfEvents == 1) {
334  typeout();
335  }
336 
337  } else {
338  JErrorStream(logErrorRun) << "Inconsistent run number "
339  << header.getRunNumber()
340  << " != "
341  << getRunNumber();
342  }
343  } else {
344  JErrorStream(logErrorFile) << "Output file not open";
345  }
346  } else {
347  if (prefix.getTag() != IO_TRIGGER_PARAMETERS) {
348  JErrorStream(logErrorTag) << "Unknown tag <" << prefix.toString() << ">, no data written";
349  }
350  }
351  } else {
352  JWarningStream(logErrorState) << "Not in running state <" << prefix.toString() << ">, no data written";
353  }
354 
355  timer.stop();
356  }
357  catch(const JControlHostException& exception) {
358  JErrorStream(logger) << exception;
359  }
360  }
361  }
362 
363 
364  virtual void actionRunning() override
365  {
366  typeout();
367  }
368 
369 
370  /**
371  * Report status of data writing.
372  */
373  void typeout()
374  {
375  std::ostringstream message;
376 
378 
379  logger.typeout(RC_LOG, message.str());
380  logger.status(message.str());
381  }
382 
383  JMeta meta; //!< meta data
384 
385  static const int MAXIMUM_FILE_NUMBER = 100; //!< maximum file number for overwrite protection.
386 
387  private:
388 
390  std::string path; // directory for output file
391 
392  JEEP::JTimer timer; // timer for I/O measurement
393  Long64_t numberOfEvents; // total number of events
394  long long int numberOfBytes; // total number of bytes
395 
400 
401  std::string hostname; //!< host name of data server
402  JTreeWriter_t writer; //!< TTree writer
403  std::vector<char> buffer; //!< internal buffer for incoming data
404 
405 
406  /**
407  * Auxiliary data structure for I/O of trigger parameters.
408  */
409  struct JValue_t {
410  /**
411  * Default constructor.
412  */
414  count(0),
415  is_written(false)
416  {}
417 
418  JTriggerParameters parameters; //!< trigger parameters
419  int count; //!< reader count
420  bool is_written; //!< writer status
421  };
422 
423 
424  /**
425  * Map run number to trigger parameters.
426  */
427  struct JRunDB :
428  public std::map<int, JValue_t>
429  {
430  /**
431  * Remove all entries before given run.
432  *
433  * \param run run number
434  */
435  inline void reset(const int run)
436  {
437  while (!this->empty() && this->begin()->first < run) {
438  this->erase(this->begin());
439  }
440  }
441 
442  /**
443  * Check if trigger parameters have been written for given run.
444  *
445  * \param run run number
446  * \return true if written; else false.
447  */
448  inline bool is_written(const int run) const
449  {
450  const_iterator p = this->find(run);
451 
452  return p != this->end() && p->second.is_written;
453  }
454 
455  /**
456  * Read trigger parameters.
457  *
458  * \param data data
459  * \param size size
460  */
461  void read(const char* const data, const size_t size)
462  {
463  using namespace std;
464  using namespace JPP;
465 
466 
467  const string buffer(data, size);
468 
469  istringstream in(buffer);
470 
471  int run = -1;
473 
474  in >> run;
475 
476  if (!in) {
477  THROW(JIOException, "Error reading run number for trigger parameters " << run << endl << in.rdbuf());
478  }
479 
480  in >> parameters;
481 
482  in.clear(std::ios::eofbit);
483 
484  if (!in) {
485  THROW(JIOException, "Error reading trigger parameters " << in.rdbuf());
486  }
487 
488  JValue_t& value = (*this)[run];
489 
490  if (value.count == 0) {
491  value.parameters = parameters;
492  }
493 
494  value.count += 1;
495 
496  if (!parameters.equals(value.parameters)) {
497  THROW(JException, "Inconsistent trigger parameters " << endl << value.parameters << " != " << endl << parameters);
498  }
499  }
500 
501  /**
502  * Write trigger parameters for given run if not yet done.
503  *
504  * \param run run number
505  * \param file pointer to ROOT file
506  */
507  inline void write(const int run, TFile* file)
508  {
509  if (file != NULL) {
510 
511  iterator p = this->find(run);
512 
513  if (p != this->end() && p->second.count != 0 && !p->second.is_written) {
514 
515  file->WriteTObject(&p->second.parameters);
516 
517  p->second.is_written = true;
518  }
519  }
520  }
521  };
522 
524  };
525 }
526 
527 
528 /**
529  * \file
530  *
531  * Application for writing real-time data to disk.
532  * \author mdejong
533  */
534 int main(int argc, char* argv[])
535 {
536  using namespace std;
537  using namespace JPP;
538  using namespace KM3NETDAQ;
539 
540  string server;
541  string logger;
542  string hostname;
543  string client_name;
544  bool use_cout;
545  string path;
546  int debug;
547 
548  try {
549 
550  JParser<> zap("Application for writing real-time data to disk.");
551 
552  zap['H'] = make_field(server, "host name of server for command messages") = "localhost";
553  zap['M'] = make_field(logger, "host name of server for logger messages") = "localhost";
554  zap['D'] = make_field(hostname, "host name of server for incoming data from data filter") = "localhost";
555  zap['u'] = make_field(client_name, "client name") = "%";
556  zap['c'] = make_field(use_cout, "print to terminal");
557  zap['p'] = make_field(path, "directory for permanent archival of data") = "";
558  zap['d'] = make_field(debug, "debug level") = 0;
559 
560 
561  zap(argc, argv);
562  }
563  catch(const exception &error) {
564  FATAL(error.what() << endl);
565  }
566 
567 
568  JLogger* out = NULL;
569 
570  if (use_cout)
571  out = new JStreamLogger(cout);
572  else
573  out = new JControlHostLogger(logger);
574 
575  JDataWriter dwriter(getProcessName(client_name, argv[0]), server, hostname, out, debug, path);
576 
577  dwriter.meta = JMeta(argc, argv);
578 
579  dwriter.enter();
580  dwriter.run();
581 }
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:1517
virtual void actionStop(int length, const char *buffer) override
Definition: JDataWriter.cc:229
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:403
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:385
ROOT TTree parameter settings of various packages.
virtual void actionQuit(int length, const char *buffer) override
Definition: JDataWriter.cc:165
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:461
void reset(const int run)
Remove all entries before given run.
Definition: JDataWriter.cc:435
JMessageScheduler logErrorState
Definition: JDataWriter.cc:399
#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:409
std::string hostname
host name of data server
Definition: JDataWriter.cc:401
*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:68
JTreeWriter_t writer
TTree writer.
Definition: JDataWriter.cc:402
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:493
JSharedPointer< JControlHost > server
message server
Definition: JDAQClient.hh:801
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:662
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:448
virtual void setSelect(JFileDescriptorMask &mask) const override
Set the file descriptor mask for the select call.
Definition: JDataWriter.cc:251
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:259
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:126
long long int numberOfBytes
Definition: JDataWriter.cc:394
virtual void actionRunning() override
This method is repeatedly called when this client machine is in state Running and the clock interval ...
Definition: JDataWriter.cc:364
void reset()
Reset timer.
Definition: JTimer.hh:76
Map run number to trigger parameters.
Definition: JDataWriter.cc:427
JMessageScheduler logErrorRun
Definition: JDataWriter.cc:396
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:50
JMessageScheduler logErrorFile
Definition: JDataWriter.cc:397
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:1993
Byte array binary input.
Definition: JByteArrayIO.hh:25
JValue_t()
Default constructor.
Definition: JDataWriter.cc:413
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:398
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:418
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:480
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:383
static const JNET::JTag RC_DWRITER
Definition: JDAQTags.hh:49
#define FATAL(A)
Definition: JMessage.hh:67
virtual void actionStart(int length, const char *buffer) override
Definition: JDataWriter.cc:169
void write(const int run, TFile *file)
Write trigger parameters for given run if not yet done.
Definition: JDataWriter.cc:507
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:373
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:273
Utility class to parse command line options.
virtual bool enter(const JArgs &args)
Enter the state machine.
Definition: JDAQClient.hh:364
JMessageLogger logger
message logger
Definition: JDAQClient.hh:802
Fixed parameters andd ControlHost tags for KM3NeT DAQ.
static const JNET::JTag RC_CMD
Definition: JDAQTags.hh:44
static const JNET::JTag IO_EVENT
Definition: JDAQTags.hh:66
virtual void actionReset(int length, const char *buffer) override
Definition: JDataWriter.cc:159
ControlHost tag.
Definition: JTag.hh:38
KM3NeT DAQ constants, bit handling, etc.
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:389
static void Throw(const bool option)
Enable/disable throw option.
Definition: JThrow.hh:37
esac $JPP_BIN JLogger sh $LOGGER until pgrep JGetMessage</dev/null > dev null
int debug
debug level
unsigned long long usec_wall
Definition: JTimer.hh:224
void start()
Start timer.
Definition: JTimer.hh:89
Exception for I/O.
Definition: JException.hh:324