DesSer.h

//+
// File : DesSer.h
// Description : Module to receive data from outside and store it to DataStore
//
// Author : Satoru Yamada, IPNS, KEK
// Date : 4 - Sep - 2014
//-
 
#ifndef DESSER_H
#define DESSER_H
 
#include <string>
#include <vector>
 
#include <sys/time.h>
 
#include <daq/slc/readout/RunInfoBuffer.h>
#include <daq/rawdata/CprErrorMessage.h>
#include <daq/rawdata/modules/DAQConsts.h>
#include <rawdata/dataobjects/PreRawCOPPERFormat_latest.h>
 
#include <daq/dataobjects/SendHeader.h>
#include <daq/dataobjects/SendTrailer.h>
 
#define BUF_SIZE_WORD 4800
 
// From Serializer.cc
 
 
namespace Belle2 {
  class DesSer {
 
    // Public functions
  public:
 
    DesSer();
 
    //    DesSer();
    virtual ~DesSer();
 
    void initialize(bool close_listen = true);
 
    virtual void DataAcquisition() = 0;
 
    void printData(int* buf, int nwords);
 
    int max_nevt;
 
    double max_seconds;
 
    int m_compressionLevel;
 
    int n_basf2evt;
 
    int m_prev_nevt;
 
    std::string m_dump_fname;
 
    FILE* m_fp_dump;
 
    int* m_buffer;
 
    int* m_bufary[ NUM_PREALLOC_BUF ];
    //    int* m_bufary[NUM_EVT_PER_BASF2LOOP];
 
    int* getPreAllocBuf();
 
    int* getNewBuffer(int nwords, int* delete_flag);
 
    // For monitoring
    timeval m_t0;
    double m_recvd_totbytes;
    double m_recvd_prev_totbytes;
    double m_sent_totbytes;
    double m_sent_prev_totbytes;
    int m_ncycle;
    double cur_time;
 
    double m_start_time;
    double m_prev_time;
    int prev_event;
 
    double getTimeSec();
 
    void recordTime(int event, double* array);
 
    unsigned int calcSimpleChecksum(int* buf, int nwords);
 
    unsigned int calcXORChecksum(int* buf, int nwords);
 
 
    void clearNumUsedBuf()
    {
      m_num_usedbuf = 0;
      return ;
    }
 
    void RateMonitor(unsigned int nevt);
 
 
    int m_num_usedbuf;
 
    std::string m_nodename;
 
    int m_nodeid;
 
    RunInfoBuffer m_status;
 
    unsigned int m_exprunsubrun_no;
 
    unsigned int m_prev_exprunsubrun_no;
 
    int m_exp_no;
 
    int m_data_type;
 
    int m_trunc_mask;
 
    int m_shmflag;
 
    void shmOpen(char* path_cfg, char* path_sta);
 
    int* shmGet(int fd, int size_words);
 
    int m_shmfd_cfg;
 
    int m_shmfd_sta;
 
    //int* m_cfg_buf;
    //int* m_cfg_sta;
 
    int monitor_numeve;
 
    int m_start_flag;
 
    CprErrorMessage print_err;
 
    //    static RunInfoBuffer g_status;
    //    RunInfoBuffer g_status;
 
    PreRawCOPPERFormat_latest m_pre_rawcpr;
 
    // From Serializer.cc
 
    void initializeNode();
 
    void beginRun();
 
    void endRun();
    void terminate();
 
    //    int sendByWriteV(RawDataBlock* rawdblk);
    int sendByWriteV(RawDataBlockFormat* rawdblk);
    //    void Connect(const char* hostname, const int port);
    void Accept(bool close_listen = true);
 
    //    void fillSendHeaderTrailer(SendHeader* hdr, SendTrailer* trl, RawDataBlock* rawdblk);
    void fillSendHeaderTrailer(SendHeader* hdr, SendTrailer* trl, RawDataBlockFormat* rawdblk);
 
    // Data members
 
    int Send(int socket, char* buf, int size_bytes);
 
 
    int* m_cfg_buf;
 
    int* m_cfg_sta;
 
    int m_run_pause;
 
    int m_run_error;
 
  protected :
    //   StoreArray<RawDataBlock> raw_datablkarray;
 
    int m_num_connections;
 
    //    std::vector<EvtSocketRecv*> m_recv;
 
    std::vector<std::string> m_hostname_from;
 
    std::vector<int> m_port_from;
 
    std::vector<int> m_socket_recv;
 
    //    StoreArray<RawFTSW> raw_ftswarray;
 
    //    StoreArray<RawCOPPER> rawcprarray;
 
    //    StoreArray<ReducedRawCOPPER> rawcprarray;
 
    int event_diff;
 
    unsigned int m_prev_copper_ctr;
 
    unsigned int m_prev_evenum;
 
    // From Serializer.cc
 
    int m_socket_send;
 
    enum {
      COPPER = 1,
      ROPC = 2
    };
 
    std::string m_hostname_local;
 
    int m_port_to;
 
    //    EvtSocketSend* m_sock;
 
    std::string p_method;
    int p_method_val;
 
#ifdef NONSTOP
    int* m_ptr;
 
    int m_run_no;
 
    int m_prev_run_no;
 
    int checkRunPause();
 
    int checkRunRecovery();
 
    void openRunPauseNshm();
 
    void resumeRun();
 
    void pauseRun();
 
    void waitResume();
 
    void callCheckRunPause(std::string& err_str);
 
    int CheckConnection(int socket);
#endif
 
    RawHeader_latest tmp_header;
 
  };
 
} // end namespace Belle2
 
#endif // DESSERCOPPER_H