DesSerPrePC Class Reference

A class definition of an input module for Sequential ROOT I/O. More...

#include <DesSerPrePC.h>

Inheritance diagram for DesSerPrePC:

Public Member Functions
	DesSerPrePC (std::string host_recv, int port_recv, const std::string &host_send, int port_send, int shmflag, const std::string &nodename, int nodeid)
	Constructor / Destructor.
void	DataAcquisition () override
	Module functions to be called from event process.
void	initialize (bool close_listen=true)
void	printData (int *buf, int nwords)
	dump error data
int *	getPreAllocBuf ()
	Getbuffer.
int *	getNewBuffer (int nwords, int *delete_flag)
	Getbuffer.
double	getTimeSec ()
	store time info.
void	recordTime (int event, double *array)
	store time info.
unsigned int	calcSimpleChecksum (int *buf, int nwords)
	calculate checksum
unsigned int	calcXORChecksum (int *buf, int nwords)
	calculate checksum
void	clearNumUsedBuf ()
void	RateMonitor (unsigned int nevt)
	monitor rate
void	shmOpen (char *path_cfg, char *path_sta)
	open shared memory
int *	shmGet (int fd, int size_words)
	Get shared memory.
void	initializeNode ()
	Module functions to be called from main process.
void	beginRun ()
	Module functions to be called from event process.
void	endRun ()
void	terminate ()
int	sendByWriteV (RawDataBlockFormat *rawdblk)
void	Accept (bool close_listen=true)
void	fillSendHeaderTrailer (SendHeader *hdr, SendTrailer *trl, RawDataBlockFormat *rawdblk)
int	Send (int socket, char *buf, int size_bytes)
	send buffer

Public Attributes
int	max_nevt
double	max_seconds
	time to stop a run
int	m_compressionLevel
	Compression Level.
int	n_basf2evt
	No. of sent events.
int	m_prev_nevt
	No. of prev sent events.
std::string	m_dump_fname
	dump filename
FILE *	m_fp_dump
	dump file descripter
int *	m_buffer
	buffer
int *	m_bufary [NUM_PREALLOC_BUF]
	buffer
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
	for time monitoring
double	m_start_time
double	m_prev_time
int	prev_event
int	m_num_usedbuf
std::string	m_nodename
	Node Name for SLC.
int	m_nodeid
	Node ID for SLC.
RunInfoBuffer	m_status
	Run info buffer.
unsigned int	m_exprunsubrun_no
	run no.
unsigned int	m_prev_exprunsubrun_no
	run no.
int	m_exp_no
	exp no.
int	m_data_type
	data type
int	m_trunc_mask
	trunc mask
int	m_shmflag
	Use shared memory.
int	m_shmfd_cfg
	file descripter for shm
int	m_shmfd_sta
	file descripter for shm
int	monitor_numeve
	buffer for shared memory
int	m_start_flag
	start flag
CprErrorMessage	print_err
	wrapper for B2LOG system
PreRawCOPPERFormat_v2	m_pre_rawcpr
	report status to SLC
int *	m_cfg_buf
	buffer for shared memory
int *	m_cfg_sta
	buffer for shared memory
int	m_run_pause
	flag to show that run-controller pauses a run
int	m_run_error
	flag to show that there is some errors with which DAQ cannot continue.

Protected Types
enum	{   COPPER = 1 ,   ROPC = 2 }

Protected Member Functions
int	Connect ()
	Accept connection.
int	recvFD (int fd, char *buf, int data_size_byte, int flag)
	receive data
int *	recvData (int *delete_flag, int *total_m_size_word, int *num_events_in_sendblock, int *num_nodes_in_sendblock)
	receive data
void	setRecvdBuffer (RawDataBlockFormat *raw_datablk, int *delete_flag)
	attach buffer to RawDataBlock
void	checkData (RawDataBlockFormat *raw_datablk, unsigned int *eve_copper_0)
	check data contents

Protected Attributes
int	m_num_connections
	check data contents
std::vector< std::string >	m_hostname_from
	Reciever basf2 Socket.
std::vector< int >	m_port_from
	port # to connect data sources
std::vector< int >	m_socket_recv
int	event_diff
unsigned int	m_prev_copper_ctr
unsigned int	m_prev_evenum
int	m_socket_send
	Reciever Socket.
std::string	m_hostname_local
	Destination Host.
int	m_port_to
	Destination port.
std::string	p_method
	EvtSocket.
int	p_method_val
RawHeader_v2	tmp_header
	which format is used

Detailed Description

A class definition of an input module for Sequential ROOT I/O.

Definition at line 26 of file DesSerPrePC.h.

Member Enumeration Documentation

anonymous enum

anonymous enum

protectedinherited

Definition at line 269 of file DesSer.h.

         {
      COPPER = 1,
      ROPC = 2
    };

Constructor & Destructor Documentation

DesSerPrePC()

DesSerPrePC	(	std::string	host_recv,
		int	port_recv,
		const std::string &	host_send,
		int	port_send,
		int	shmflag,
		const std::string &	nodename,
		int	nodeid
	)

Constructor / Destructor.

Definition at line 28 of file DesSerPrePC.cc.

{
 
  for (int i = 0 ; i < m_num_connections; i++) {
    //  m_hostname_from.push_back( "localhost");
    m_hostname_from.push_back(host_recv);
    //  m_port_from.push_back(30000);
    m_port_from.push_back(port_recv) ;
    m_socket_recv.push_back(-1);
  }
 
  //  m_port_to = 31001;
  m_port_to = port_send;
  //  m_hostname_local = "localhost";
  m_hostname_local = host_send;
  m_nodename = nodename;
 
  m_shmflag = shmflag;
 
  //  B2INFO("DeSerializerPrePC: Constructor done.");
  printf("[INFO] DeSerializerPrePC: Constructor done.\n"); fflush(stdout);
}

~DesSerPrePC()

~DesSerPrePC ( )

virtual

Definition at line 54 of file DesSerPrePC.cc.

{
}

Member Function Documentation

Accept()

void Accept ( bool  close_listen = true )

inherited

Definition at line 364 of file DesSer.cc.

{
  //
  // Connect to cprtb01
  //
  struct hostent* host;
  host = gethostbyname(m_hostname_local.c_str());
  if (host == NULL) {
    char temp_buf[500];
    sprintf(temp_buf, "[FATAL] hostname(%s) cannot be resolved(%s). Check /etc/hosts. Exiting...\n",
            m_hostname_local.c_str(), strerror(errno));
    print_err.PrintError(temp_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    exit(1);
  }
 
  //
  // Bind and listen
  //
  int fd_listen;
  struct sockaddr_in sock_listen;
  sock_listen.sin_family = AF_INET;
  sock_listen.sin_addr.s_addr = *(unsigned int*)host->h_addr_list[0];
 
  socklen_t addrlen = sizeof(sock_listen);
  sock_listen.sin_port = htons(m_port_to);
  fd_listen = socket(PF_INET, SOCK_STREAM, 0);
 
  int flags = 1;
  int ret = setsockopt(fd_listen, SOL_SOCKET, SO_REUSEADDR, &flags, (socklen_t)sizeof(flags));
  if (ret < 0) {
    perror("Failed to set REUSEADDR");
  }
 
  if (bind(fd_listen, (struct sockaddr*)&sock_listen, sizeof(struct sockaddr)) < 0) {
    printf("[FATAL] Failed to bind. Maybe other programs have already occupied this port(%d). Exiting...\n",
           m_port_to); fflush(stdout);
    // Check the process occupying the port 30000.
    FILE* fp;
    char buf[256];
    char cmdline[500];
    sprintf(cmdline, "/usr/sbin/ss -ap | grep %d", m_port_to);
    if ((fp = popen(cmdline, "r")) == NULL) {
      printf("[WARNING] Failed to run %s\n", cmdline);
    }
    while (fgets(buf, 256, fp) != NULL) {
      printf("[ERROR] Failed to bind. output of ss(port %d) : %s\n", m_port_to, buf); fflush(stdout);
    }
    // Error message
    fclose(fp);
    char temp_char[500];
    sprintf(temp_char, "[FATAL] Failed to bind.(%s) Maybe other programs have already occupied this port(%d). Exiting...",
            strerror(errno), m_port_to);
    print_err.PrintError(temp_char, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    exit(1);
  }
 
  int val1 = 0;
  setsockopt(fd_listen, IPPROTO_TCP, TCP_NODELAY, &val1, (socklen_t)sizeof(val1));
  int backlog = 1;
  if (listen(fd_listen, backlog) < 0) {
    char err_buf[500];
    sprintf(err_buf, "Failed in listen(%s). Exting...", strerror(errno));
    print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    exit(-1);
  }
 
  //
  // Accept
  //
  int fd_accept;
  struct sockaddr_in sock_accept;
  printf("[DEBUG] Accepting... : port %d server %s\n", m_port_to, m_hostname_local.c_str());
  fflush(stdout);
 
  if ((fd_accept = accept(fd_listen, (struct sockaddr*) & (sock_accept), &addrlen)) == 0) {
    char err_buf[500];
    sprintf(err_buf, "[FATAL] Failed to accept(%s). Exiting...", strerror(errno));
    print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    exit(-1);
  } else {
    //    B2INFO("Done.");
    printf("[DEBUG] Done.\n"); fflush(stdout);
 
    //    set timepout option
    struct timeval timeout;
    timeout.tv_sec = 1;
    timeout.tv_usec = 0;
    ret = setsockopt(fd_accept, SOL_SOCKET, SO_SNDTIMEO, &timeout, (socklen_t)sizeof(timeout));
    if (ret < 0) {
      char temp_char[100] = "[FATAL] Failed to set TIMEOUT. Exiting...";
      print_err.PrintError(temp_char, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      exit(-1);
    }
  }
 
  if (close_listen) {
    close(fd_listen);
  }
 
  //   int flag = 1;
  //   ret = setsockopt(fd_accept, IPPROTO_TCP, TCP_NODELAY, (char*)&flag, sizeof(flag) );
  m_socket_send = fd_accept;
 
  if (m_status.isAvailable()) {
    m_status.setOutputPort(ntohs(sock_listen.sin_port));
    m_status.setOutputAddress(sock_listen.sin_addr.s_addr);
    //    B2INFO("Accepted " << (int)ntohs(sock_listen.sin_port) << " " << (int)sock_listen.sin_addr.s_addr);
    printf("Accepted. port %d address %d\n", (int)ntohs(sock_listen.sin_port), (int)sock_listen.sin_addr.s_addr); fflush(stdout);
  }
 
  return;
 
}

calcXORChecksum()

unsigned int calcXORChecksum ( int *  buf, int  nwords  )

inherited

calculate checksum

Definition at line 495 of file DesSer.cc.

{
  unsigned int checksum = 0;
  for (int i = 0; i < nwords; i++) {
    checksum = checksum ^ buf[ i ];
  }
  return checksum;
}

checkData()

void checkData ( RawDataBlockFormat *  raw_datablk, unsigned int *  eve_copper_0  )

protected

check data contents

Definition at line 381 of file DesSerPrePC.cc.

{
  //  int data_size_copper_0 = -1;
  //  int data_size_copper_1 = -1;
 
  //
  // Data check
  //
  int* temp_buf = raw_datablk->GetBuffer(0);
  int cpr_num = 0;
  unsigned int cur_evenum = 0, cur_copper_ctr = 0;
  unsigned int eve_array[32]; // # of noeds is less than 17
  unsigned int utime_array[32];// # of noeds is less than 17
  unsigned int ctime_type_array[32];// # of noeds is less than 17
 
#ifdef DUMHSLB
  unsigned int exp_run_ftsw = 0, ctime_trgtype_ftsw = 0, utime_ftsw = 0;
#endif
 
  for (int k = 0; k < raw_datablk->GetNumEvents(); k++) {
    memset(eve_array, 0, sizeof(eve_array));
    memset(utime_array, 0, sizeof(utime_array));
    memset(ctime_type_array, 0, sizeof(ctime_type_array));
 
    int num_nodes_in_sendblock = raw_datablk->GetNumNodes();
    for (int l = 0; l < num_nodes_in_sendblock; l++) {
      int entry_id = l + k * num_nodes_in_sendblock;
 
      //
      // RawFTSW
      //
      if (raw_datablk->CheckFTSWID(entry_id)) {
        RawFTSWFormat_latest* temp_rawftsw = new RawFTSWFormat_latest;
        int block_id = 0;
        temp_rawftsw->SetBuffer((int*)temp_buf + raw_datablk->GetBufferPos(entry_id),
                                raw_datablk->GetBlockNwords(entry_id), 0, 1, 1);
        if (temp_rawftsw->GetEveNo(block_id) < 10) {
          printf("[DEBUG] ######FTSW#########\n");
          printData((int*)temp_buf + raw_datablk->GetBufferPos(entry_id), raw_datablk->GetBlockNwords(entry_id));
        }
 
#ifdef DUMHSLB
        exp_run_ftsw = temp_rawftsw->GetExpRunSubrun(block_id);
        ctime_trgtype_ftsw = temp_rawftsw->GetTTCtimeTRGType(block_id);
        utime_ftsw = temp_rawftsw->GetTTUtime(block_id);
#endif
 
 
#ifndef NO_DATA_CHECK
        try {
          temp_rawftsw->CheckData(0, m_prev_evenum, &cur_evenum, m_prev_exprunsubrun_no, &m_exprunsubrun_no);
          eve_array[ entry_id ] = cur_evenum;
        } catch (const string& err_str) {
          char err_buf[500];
          strcpy(err_buf, err_str.c_str());
          print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
          exit(1);
        }
#endif
        utime_array[ entry_id ] = temp_rawftsw->GetTTUtime(block_id);
        ctime_type_array[ entry_id ] = temp_rawftsw->GetTTCtimeTRGType(block_id);
        delete temp_rawftsw;
 
        //
        // RawTLU
        //
      } else if (raw_datablk->CheckTLUID(entry_id)) {
 
        RawTLUFormat* temp_rawtlu = new RawTLUFormat;
        temp_rawtlu->SetBuffer((int*)temp_buf + raw_datablk->GetBufferPos(entry_id),
                               raw_datablk->GetBlockNwords(entry_id), 0, 1, 1);
        if (temp_rawtlu->GetEveNo(0) < 10
           ) {
          printf("[DEBUG] ######TLU#########\n");
          printData((int*)temp_buf + raw_datablk->GetBufferPos(entry_id), raw_datablk->GetBlockNwords(entry_id));
        }
 
#ifndef NO_DATA_CHECK
        try {
          temp_rawtlu->CheckData(0, m_prev_evenum, &cur_evenum);
          eve_array[ entry_id ] = cur_evenum;
        } catch (const string& err_str) {
          char err_buf[500];
          strcpy(err_buf, err_str.c_str());
          print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
          exit(1);
        }
#endif
        delete temp_rawtlu;
      } else {
        //
        // RawCOPPER
        //
        PreRawCOPPERFormat_v2* pre_rawcpr_fmt = new PreRawCOPPERFormat_v2;
        pre_rawcpr_fmt->SetBuffer((int*)temp_buf + raw_datablk->GetBufferPos(entry_id),
                                  raw_datablk->GetBlockNwords(entry_id), 0, 1, 1);
 
#ifdef DUMHSLB
        int block_id = 0;
        "do not use the following for actual DAQ"
        (pre_rawcpr_fmt->GetBuffer(block_id))[ RawHeader_v2::POS_EXP_RUN_NO ] = exp_run_ftsw;
        (pre_rawcpr_fmt->GetBuffer(block_id))[ RawHeader_v2::POS_TTCTIME_TRGTYPE ] = ctime_trgtype_ftsw;
        (pre_rawcpr_fmt->GetBuffer(block_id))[ RawHeader_v2::POS_TTUTIME ] = utime_ftsw;
#endif
 
#ifndef NO_DATA_CHECK
        try {
          pre_rawcpr_fmt->CheckData(0, m_prev_evenum, &cur_evenum,
                                    m_prev_copper_ctr, &cur_copper_ctr,
                                    m_prev_exprunsubrun_no, &m_exprunsubrun_no);
          eve_array[ entry_id ] = cur_evenum;
        } catch (const string& err_str) {
          exit(1); // Error in the contents of an event was detected
        }
#endif
 
        utime_array[ entry_id ] = pre_rawcpr_fmt->GetTTUtime(0);
        ctime_type_array[ entry_id ] = pre_rawcpr_fmt->GetTTCtimeTRGType(0);
 
        if (cpr_num == 0) {
          //          data_size_copper_0 = raw_datablk->GetBlockNwords(entry_id);
          *eve_copper_0 = (raw_datablk->GetBuffer(entry_id))[ 3 ];
        } else if (cpr_num == 1) {
          //          data_size_copper_1 = raw_datablk->GetBlockNwords(entry_id);
        }
        cpr_num++;
        delete pre_rawcpr_fmt;
      }
    }
 
#ifndef NO_DATA_CHECK
    // event #, ctime, utime over nodes
    for (int l = 1; l < num_nodes_in_sendblock; l++) {
      if (eve_array[ 0 ] != eve_array[ l ] ||
          utime_array[ 0 ] != utime_array[ l ] ||
          ctime_type_array[ 0 ] != ctime_type_array[ l ]) {
        char err_buf[500];
        for (int m = 0; m < num_nodes_in_sendblock; m++) {
          printf("[DEBUG] node %d eve # %x utime %x ctime %x\n",
                 m,  eve_array[ m ], utime_array[ m ], ctime_type_array[ m ]);
        }
        sprintf(err_buf, "[FATAL] CORRUPTED DATA: Event or Time record mismatch. Exiting...");
        print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
        sleep(1234567);
        exit(-1);
      }
    }
#endif
 
    // Event # monitor in runchange
//     if (m_prev_runsubrun_no != m_runsubrun_no) {
//       printf("[DEBUG] ##############################################\n");
//       for (int m = 0; m < raw_datablk->GetNumEntries(); m++) {
//         printf("[DEBUG] %d eve %u prev %u\n", m, eve_array[ m ], m_prev_evenum);
//       }
//       printf("[DEBUG] ##############################################\n");
//       fflush(stderr);
//     }
    m_prev_evenum = cur_evenum;
    m_prev_copper_ctr = cur_copper_ctr;
    m_prev_exprunsubrun_no = m_exprunsubrun_no;
  }
  return;
}

clearNumUsedBuf()

void clearNumUsedBuf ( )

inlineinherited

Definition at line 121 of file DesSer.h.

    {
      m_num_usedbuf = 0;
      return ;
    }

Connect()

int Connect ( )

protected

Accept connection.

Definition at line 114 of file DesSerPrePC.cc.

{
 
  for (int i = 0; i < m_num_connections; i++) {
 
    if (m_socket_recv[ i ] >= 0) continue;   // Already have an established socket
 
    //
    // Connect to a downstream node
    //
    struct hostent* host;
    host = gethostbyname(m_hostname_from[ i ].c_str());
    if (host == NULL) {
      char err_buf[100];
      sprintf(err_buf, "[FATAL] hostname(%s) cannot be resolved(%s). Check /etc/hosts. Exiting...", m_hostname_from[ i ].c_str(),
              strerror(errno));
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(1);
    }
 
    struct sockaddr_in socPC;
    socPC.sin_family = AF_INET;
    socPC.sin_addr.s_addr = *(unsigned int*)host->h_addr_list[0];
    socPC.sin_port = htons(m_port_from[ i ]);
    int sd = socket(PF_INET, SOCK_STREAM, 0);
    int val1 = 0;
    setsockopt(sd, IPPROTO_TCP, TCP_NODELAY, &val1, sizeof(val1));
 
    struct timeval timeout;
    timeout.tv_sec = 1;
    timeout.tv_usec = 0;
    setsockopt(sd, SOL_SOCKET, SO_RCVTIMEO, &timeout, (socklen_t)sizeof(timeout));
 
    printf("[DEBUG] Connecting to %s port %d\n", m_hostname_from[ i ].c_str(), m_port_from[ i ]); fflush(stdout);
 
    while (1) {
      if (connect(sd, (struct sockaddr*)(&socPC), sizeof(socPC)) < 0) {
        perror("Failed to connect. Retrying...");
        usleep(500000);
      } else {
        //        B2INFO("Done");
        printf("[DEBUG] Done\n"); fflush(stdout);
        break;
      }
    }
    //    m_socket_recv.push_back(sd);
    m_socket_recv[ i ] = sd;
 
    // check socket paramters
    int val, len;
    len = sizeof(val);
    getsockopt(m_socket_recv[ i ], SOL_SOCKET, SO_RCVBUF, &val, (socklen_t*)&len);
    //    B2INFO("SO_RCVBUF" << val);
    getsockopt(m_socket_recv[ i ], SOL_SOCKET, SO_SNDBUF, &val, (socklen_t*)&len);
    //    B2DEBUG("SO_SNDBUF" <<  val);
    getsockopt(m_socket_recv[ i ], IPPROTO_TCP, TCP_MAXSEG, &val, (socklen_t*)&len);
    //    B2DEBUG("TCP_MAXSEG" <<  val);
    getsockopt(m_socket_recv[ i ], IPPROTO_TCP, TCP_NODELAY, &val, (socklen_t*)&len);
    //    B2DEBUG("TCP_NODELAY" <<  val);
 
    if (m_status.isAvailable()) {
      sockaddr_in sa;
      memset(&sa, 0, sizeof(sockaddr_in));
      socklen_t sa_len = sizeof(sa);
      if (getsockname(m_socket_recv[i], (struct sockaddr*)&sa, (socklen_t*)&sa_len) == 0) {
        m_status.setInputPort(ntohs(sa.sin_port));
        m_status.setInputAddress(sa.sin_addr.s_addr);
      }
    }
 
  }
  //  B2INFO("[DEBUG] Initialization finished");
  printf("[DEBUG] Initialization finished\n"); fflush(stdout);
  return 0;
}

DataAcquisition()

void DataAcquisition ( )

overridevirtual

Module functions to be called from event process.

Implements DesSer.

Definition at line 548 of file DesSerPrePC.cc.

{
  // For data check
  unsigned int eve_copper_0 = 0;
  //  B2INFO("initializing...");
  printf("[DEBUG] initializing...\n"); fflush(stdout);
  initialize();
 
  //  B2INFO("Done.");
  printf("[DEBUG] Done.\n"); fflush(stdout);
 
  if (m_start_flag == 0) {
    //
    // Connect to eb0: This should be here because we want Serializer first to accept connection from eb1tx
    //
    Connect();
    if (m_status.isAvailable()) {
      //      B2INFO("DeSerializerPrePC: Waiting for Start...\n");
      printf("[DEBUG] DeSerializerPrePC: Waiting for Start...\n"); fflush(stdout);
      m_status.reportRunning();
    }
    m_start_time = getTimeSec();
    n_basf2evt = 0;
  }
 
  //
  // Main loop
  //
  while (1) {
    //
    // Stand-by loop
    //
#ifdef NONSTOP
    if (m_run_pause > 0 || m_run_error > 0) {
      waitResume();
    }
#endif
 
    clearNumUsedBuf();
    //    RawDataBlock raw_datablk[ NUM_EVT_PER_BASF2LOOP_PC ];
    RawDataBlockFormat raw_datablk[ NUM_EVT_PER_BASF2LOOP_PC ];
 
    //
    // Recv loop
    //
    for (int j = 0; j < NUM_EVT_PER_BASF2LOOP_PC; j++) {
      //
      // Receive data from COPPER
      //
      eve_copper_0 = 0;
      int delete_flag_from =
        0; // Delete flag for temp_rawdatablk.It can be set to 1 by setRecvdBuffer if the buffer size is larger than that of pre-allocated buffer.
      int delete_flag_to =
        0; // Delete flag for raw_datablk[i]. It can be set to 1 by getNewBuffer if the buffer size is larger than that of pre-allocated buffer.
      RawDataBlockFormat temp_rawdatablk;
      try {
        setRecvdBuffer(&temp_rawdatablk, &delete_flag_from);
        checkData(&temp_rawdatablk, &eve_copper_0);
      } catch (const string& err_str) {
        printf("Error was detected\n"); fflush(stdout);
        break;
      }
      //     PreRawCOPPERFormat_v2 pre_rawcopper_v2;
      //     pre_rawcopper_v2.SetBuffer((int*)temp_rawdatablk.GetWholeBuffer(), temp_rawdatablk.TotalBufNwords(),
      //                                    0, temp_rawdatablk.GetNumEvents(), temp_rawdatablk.GetNumNodes());
 
      int temp_num_events, temp_num_nodes;
      int temp_nwords_to;
      int* buf_to = NULL;
#ifdef REDUCED_RAWCOPPER
      //
      // Copy reduced buffer
      //
      //      int* buf_to = getNewBuffer(m_pre_rawcpr.CalcReducedDataSize(&temp_rawdatablk), &delete_flag_to); // basf2-dependent style
      int* temp_bufin = temp_rawdatablk.GetWholeBuffer();
      int temp_nwords_from = temp_rawdatablk.TotalBufNwords();
      temp_num_events = temp_rawdatablk.GetNumEvents();
      temp_num_nodes = temp_rawdatablk.GetNumNodes();
      int calced_temp_nwords_to = m_pre_rawcpr.CalcReducedDataSize(temp_bufin, temp_nwords_from, temp_num_events, temp_num_nodes);
      buf_to = getNewBuffer(calced_temp_nwords_to, &delete_flag_to);
 
      //      m_pre_rawcpr.CopyReducedData(&temp_rawdatablk, buf_to, delete_flag_from); // basf2-dependent style
      m_pre_rawcpr.CopyReducedData(temp_bufin, temp_nwords_from, temp_num_events, temp_num_nodes, buf_to, &temp_nwords_to);
      if (calced_temp_nwords_to != temp_nwords_to) {
        char err_buf[500];
        sprintf(err_buf,
                "[FATAL] CORRUPTED DATA: Estimations of reduced event size are inconsistent. CalcReducedDataSize = %d. CopyReducedData = %d. Exiting...",
                calced_temp_nwords_to, temp_nwords_to);
        print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
        exit(1);
      }
      m_status.copyEventHeader(buf_to);
#else
      buf_to = temp_rawdatablk.GetWholeBuffer();
      temp_nwords_to = temp_rawdatablk.TotalBufNwords();
      temp_num_events = temp_rawdatablk.GetNumEvents();
      temp_num_nodes = temp_rawdatablk.GetNumNodes();
      delete_flag_to = delete_flag_from;
      delete_flag_from = 0; // to avoid double delete
#endif
 
      //
      // Set buffer to the RawData class stored in DataStore
      //
//       raw_datablk[ j ].SetBuffer( (int*)temp_rawdatablk.GetWholeBuffer(), temp_rawdatablk.TotalBufNwords(),
//                                  delete_flag_to, temp_rawdatablk.GetNumEvents(),
//                                  temp_rawdatablk.GetNumNodes());
      raw_datablk[ j ].SetBuffer(buf_to, temp_nwords_to, delete_flag_to, temp_num_events, temp_num_nodes);
 
 
      //
      // CRC16 check after data reduction
      //
#ifdef REDUCED_RAWCOPPER
      PostRawCOPPERFormat_v2 post_rawcopper_v2; // Should be the latest version before ver.4(PCIe40)
 
//       post_rawcopper_v2.SetBuffer((int*)temp_rawdatablk.GetWholeBuffer(), temp_rawdatablk.TotalBufNwords(),
//                                       0, temp_rawdatablk.GetNumEvents(), temp_rawdatablk.GetNumNodes());
      post_rawcopper_v2.SetBuffer(raw_datablk[ j ].GetWholeBuffer(), raw_datablk[ j ].TotalBufNwords(),
                                  0, raw_datablk[ j ].GetNumEvents(), raw_datablk[ j ].GetNumNodes());
 
      for (int i_finesse_num = 0; i_finesse_num < 4; i_finesse_num ++) {
        int block_num = 0;
        if (post_rawcopper_v2.GetFINESSENwords(block_num, i_finesse_num) > 0) {
          post_rawcopper_v2.CheckCRC16(block_num, i_finesse_num);
        }
      }
 
#endif
    }
 
#ifdef NONSTOP
    // Goto Stand-by loop when run is paused or stopped by error
    if (m_run_pause != 0 || m_run_error != 0) continue;
#endif
 
 
    // From Serializer.cc
    if (m_start_flag == 0) {
      m_start_time = getTimeSec();
      n_basf2evt = 0;
    }
    //  StoreArray<RawCOPPER> rawcprarray;
    //  StoreArray<RawDataBlock> raw_dblkarray;
 
    for (int j = 0; j < NUM_EVT_PER_BASF2LOOP_PC; j++) {
 
      //
      // Send data
      //
      if (m_start_flag == 0) {
        //        B2INFO("SerializerPC: Sending the 1st packet...");
        printf("[DEBUG] SerializerPC: Sending the 1st packet...\n"); fflush(stdout);
      }
 
      try {
        m_sent_totbytes += sendByWriteV(&(raw_datablk[ j ]));
      } catch (const string& err_str) {
#ifdef NONSTOP
        break;
#endif
        print_err.PrintError((char*)err_str.c_str(), __FILE__, __PRETTY_FUNCTION__, __LINE__);
        exit(1);
      }
      if (m_start_flag == 0) {
        //        B2INFO("Done. ");
        printf("[DEBUG] Done.\n"); fflush(stdout);
        m_start_flag = 1;
      }
    }
 
#ifdef NONSTOP
    // Goto Stand-by loop when run is paused or stopped by error
    if (m_run_pause != 0 || m_run_error != 0) continue;
#endif
 
    //
    // Monitor
    //
    if (max_nevt >= 0 || max_seconds >= 0.) {
#ifdef AIUEO
      if (n_basf2evt % 10000 == 0) {
//     if ((n_basf2evt * NUM_EVT_PER_BASF2LOOP_PC >= max_nevt && max_nevt > 0)
//         || (getTimeSec() - m_start_time > max_seconds && max_seconds > 0.)) {
        printf("[DEBUG] RunStop was detected. ( Setting:  Max event # %d MaxTime %lf ) Processed Event %d Elapsed Time %lf[s]\n",
               max_nevt, max_seconds, n_basf2evt * NUM_EVT_PER_BASF2LOOP_PC, getTimeSec() - m_start_time);
      }
#endif
    }
 
    if ((n_basf2evt * NUM_EVT_PER_BASF2LOOP_PC) % 100000 == 0) {
      double interval = cur_time - m_prev_time;
      double total_time = cur_time - m_start_time;
      printf("[DEBUG] Event %12d Rate %6.2lf[kHz] Recvd %6.2lf[MB/s] sent %6.2lf[MB/s] RunTime %8.2lf[s] interval %8.4lf[s]\n",
             n_basf2evt * NUM_EVT_PER_BASF2LOOP_PC,
             (n_basf2evt  - m_prev_nevt)*NUM_EVT_PER_BASF2LOOP_PC / interval / 1.e3,
             (m_recvd_totbytes - m_recvd_prev_totbytes) / interval / 1.e6,
             (m_sent_totbytes - m_sent_prev_totbytes) / interval / 1.e6,
             total_time,
             interval);
      fflush(stdout);
 
      m_prev_time = cur_time;
      m_recvd_prev_totbytes = m_recvd_totbytes;
      m_sent_prev_totbytes = m_sent_totbytes;
      m_prev_nevt = n_basf2evt;
      cur_time = getTimeSec();
    }
 
    n_basf2evt++;
 
    if (m_status.isAvailable()) {
      m_status.setOutputNBytes(m_sent_totbytes);
      m_status.setOutputCount(n_basf2evt * NUM_EVT_PER_BASF2LOOP_PC);
    }
 
  }
 
  return;
}

fillSendHeaderTrailer()

void fillSendHeaderTrailer ( SendHeader *  hdr, SendTrailer *  trl, RawDataBlockFormat *  rawdblk  )

inherited

Definition at line 171 of file DesSer.cc.

{
 
  int total_send_nwords =
    hdr->GetHdrNwords() +
    rawdblk->TotalBufNwords() +
    //    rawhdr.GetNwords() +
    trl->GetTrlNwords();
 
  hdr->SetNwords(total_send_nwords);
  hdr->SetNumEventsinPacket(rawdblk->GetNumEvents());
  hdr->SetNumNodesinPacket(rawdblk->GetNumNodes());
 
  //
  // For bug check
  //
  if (rawdblk->GetNumEntries() == 1) {
    if (total_send_nwords != (rawdblk->GetBuffer(0))[ 0 ] + 8) {
      char err_buf[500];
      sprintf(err_buf, "Length error. total length %d rawdblk length %d. Exting...\n",
              total_send_nwords, (rawdblk->GetBuffer(0))[ 0 ]);
      printData(rawdblk->GetBuffer(0), rawdblk->TotalBufNwords());
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(-1);
    }
  }
 
 
  for (int i = 0; i < rawdblk->GetNumEntries(); i++) {
 
    //copy event # from a tonp COPPER block
    if (!(rawdblk->CheckFTSWID(i)) && !(rawdblk->CheckTLUID(i))) {
      tmp_header.SetBuffer(rawdblk->GetBuffer(i));
      hdr->SetEventNumber(tmp_header.GetEveNo());
      hdr->SetNodeID(tmp_header.GetNodeID());
      hdr->SetExpRunWord(tmp_header.GetExpRunSubrun());
      break;
    }
 
    //Error if you cannot find any COPPER block
    if (i == (rawdblk->GetNumEntries() - 1)) {
      printf("[DEBUG] i= %d : num entries %d : Tot words %d\n", i, rawdblk->GetNumEntries(), rawdblk->TotalBufNwords());
      printData(rawdblk->GetBuffer(0), rawdblk->TotalBufNwords());
 
      char err_buf[500] = "[FATAL] CORRUPTED DATA: No COPPER blocks in RawDataBlock. Exiting...";
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(-1);
    }
  }
  return;
}

getNewBuffer()

int * getNewBuffer ( int  nwords, int *  delete_flag  )

inherited

Getbuffer.

Definition at line 77 of file DesSer.cc.

{
 
  int* temp_buf = NULL;
  // Prepare buffer
  if (nwords >  BUF_SIZE_WORD) {
    *delete_flag = 1;
    temp_buf = new int[ nwords ];
  } else {
    if ((temp_buf = getPreAllocBuf()) == 0x0) {
      char err_buf[500];
      sprintf(err_buf, "Null pointer from GetPreALlocBuf(). Exting...\n");
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(1);
    } else {
      *delete_flag = 0;
    }
  }
 
  return temp_buf;
 
}

getPreAllocBuf()

int * getPreAllocBuf ( )

inherited

Getbuffer.

Definition at line 57 of file DesSer.cc.

{
  int* tempbuf = 0;
  if (m_num_usedbuf < NUM_PREALLOC_BUF) {
    tempbuf = m_bufary[ m_num_usedbuf  ];
    m_num_usedbuf++;
  } else {
    char err_buf[500];
    sprintf(err_buf,
            "No pre-allocated buffers are left. %d > %d. Not enough buffers are allocated or "
            "memory leak or forget to call ClearNumUsedBuf every event loop. Exting...",
            m_num_usedbuf, NUM_PREALLOC_BUF);
    print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    sleep(1234567);
    exit(1);
  }
  return tempbuf;
}

getTimeSec()

double getTimeSec ( )

inherited

store time info.

Definition at line 478 of file DesSer.cc.

{
  struct timeval t;
  gettimeofday(&t, NULL);
  return (t.tv_sec + t.tv_usec * 1.e-6);
}

initialize()

void initialize ( bool  close_listen = true )

inherited

Definition at line 101 of file DesSer.cc.

{
  printf("[DEBUG] DesSer: initialize() started.\n"); fflush(stdout);
  signal(SIGPIPE, SIG_IGN);
 
  //
  // initialize Rx part from DeSerializer**.cc
  //
 
  // allocate buffer
  for (int i = 0 ; i < NUM_PREALLOC_BUF; i++) {
    m_bufary[i] = new int[ BUF_SIZE_WORD ];
  }
  m_buffer = new int[ BUF_SIZE_WORD ];
 
  // initialize buffer
  for (int i = 0 ; i < NUM_PREALLOC_BUF; i++) {
    memset(m_bufary[i], 0,  BUF_SIZE_WORD * sizeof(int));
  }
 
  // Open message handler
  clearNumUsedBuf();
  // Shared memory
  if (m_shmflag > 0) {
    if (m_nodename.size() == 0 || m_nodeid < 0) {
      m_shmflag = 0;
    } else {
      m_status.open(m_nodename, m_nodeid);
      m_status.reportReady();
    }
  }
 
  event_diff = 0;
 
  m_prev_copper_ctr = 0xFFFFFFFF;
  m_prev_evenum = 0xFFFFFFFF;
 
 
  //
  // initialize Rx part from DeSerializer**.cc
  //
  m_start_flag = 0;
  n_basf2evt = -1;
  m_compressionLevel = 0;
 
#ifdef DUMMY
  m_buffer = new int[ BUF_SIZE_WORD ];
#endif
  Accept(close_listen);
#ifdef NONSTOP
  openRunPauseNshm();
#endif
 
 
  // For monitor
  if (m_status.isAvailable()) {
    m_status.setOutputNBytes(0);
    m_status.setOutputCount(0);
  }
  //  B2INFO("DesSer: initialize() was done.");
  printf("[DEBUG] DesSer: initialize() was done.\n"); fflush(stdout);
 
}

printData()

void printData ( int *  buf, int  nwords  )

inherited

dump error data

Definition at line 505 of file DesSer.cc.

{
  printf("[DEBUG]");
  for (int i = 0; i < nwords; i++) {
    printf("%.8x ", buf[ i ]);
    if (i % 10 == 9) printf("\n[DEBUG]");
  }
  printf("\n[DEBUG]");
  printf("\n");
  return;
}

recordTime()

void recordTime ( int  event, double *  array  )

inherited

store time info.

Definition at line 486 of file DesSer.cc.

{
  if (event >= 50000 && event < 50500) {
    array[ event - 50000 ] = getTimeSec() - m_start_time;
  }
  return;
}

recvData()

int * recvData ( int *  delete_flag, int *  total_m_size_word, int *  num_events_in_sendblock, int *  num_nodes_in_sendblock  )

protected

receive data

Definition at line 193 of file DesSerPrePC.cc.

{
  int* temp_buf = NULL; // buffer for data-body
  int flag = 0;
 
  vector <int> each_buf_nwords;
  each_buf_nwords.clear();
  vector <int> each_buf_nodes;
  each_buf_nodes.clear();
  vector <int> each_buf_events;
  each_buf_events.clear();
 
  *total_buf_nwords = 0;
  *num_nodes_in_sendblock = 0;
  *num_events_in_sendblock = 0;
 
  //
  // Read Header and obtain data size
  //
  int send_hdr_buf[ SendHeader::SENDHDR_NWORDS ];
 
  // Read header
  for (int i = 0; i < (int)(m_socket_recv.size()); i++) {
 
    recvFD(m_socket_recv[ i ], (char*)send_hdr_buf, sizeof(int)*SendHeader::SENDHDR_NWORDS, flag);
 
    SendHeader send_hdr;
    send_hdr.SetBuffer(send_hdr_buf);
 
    int temp_num_events = send_hdr.GetNumEventsinPacket();
    int temp_num_nodes = send_hdr.GetNumNodesinPacket();
 
    if (i == 0) {
      *num_events_in_sendblock = temp_num_events;
    } else if (*num_events_in_sendblock != temp_num_events) {
#ifndef NO_DATA_CHECK
      char err_buf[500];
      sprintf(err_buf,
              "[FATAL] CORRUPTED DATA: Different # of events or nodes in SendBlocks( # of eve : %d(socket 0) %d(socket %d), # of nodes: %d(socket 0) %d(socket %d). Exiting...\n",
              *num_events_in_sendblock, temp_num_events, i,  *num_nodes_in_sendblock, temp_num_nodes, i);
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(1);
#endif
    }
 
    *num_nodes_in_sendblock += temp_num_nodes;
 
    int rawblk_nwords = send_hdr.GetTotalNwords()
                        - SendHeader::SENDHDR_NWORDS
                        - SendTrailer::SENDTRL_NWORDS;
    *total_buf_nwords += rawblk_nwords;
 
    //
    // Data size check1
    //
    if (rawblk_nwords > (int)(2.5e6) || rawblk_nwords <= 0) {
      printData(send_hdr_buf, SendHeader::SENDHDR_NWORDS);
      char err_buf[500];
      sprintf(err_buf, "CORRUPTED DATA: Too large event : Header %d %d %d %d :block size %d words\n", i, temp_num_events, temp_num_nodes,
              send_hdr.GetTotalNwords(), rawblk_nwords);
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(123456);
      exit(1);
 
    }
 
    each_buf_nwords.push_back(rawblk_nwords);
    each_buf_events.push_back(temp_num_events);
    each_buf_nodes.push_back(temp_num_nodes);
 
  }
 
 
  temp_buf = getNewBuffer(*total_buf_nwords, delete_flag); // this include only data body
  //
  // Read body
  //
  int total_recvd_byte = 0;
  for (int i = 0; i < (int)(m_socket_recv.size()); i++) {
 
    try {
      total_recvd_byte += recvFD(m_socket_recv[ i ], (char*)temp_buf + total_recvd_byte,
                                 each_buf_nwords[ i ] * sizeof(int), flag);
    } catch (const string& err_str) {
      if (*delete_flag) {
        //        B2WARNING("Delete buffer before going to Run-pause state");
        printf("[WARNING] Delete buffer before going to Run-pause state\n"); fflush(stdout);
        delete temp_buf;
      }
      throw (std::move(err_str));
    }
    //
    // Data length check
    //
    int temp_length = 0;
    for (int j = 0; j < each_buf_nodes[ i ] * each_buf_events[ i ]; j++) {
      int this_length = *((int*)((char*)temp_buf + total_recvd_byte - each_buf_nwords[ i ] * sizeof(int) + temp_length));
      temp_length += this_length * sizeof(int);
    }
    if (temp_length != (int)(each_buf_nwords[ i ] * sizeof(int))) {
      printf("[DEBUG]*******SENDHDR*********** \n");
      printData(send_hdr_buf, SendHeader::SENDHDR_NWORDS);
      printf("[DEBUG]*******BODY***********\n ");
      printData(temp_buf, (int)(total_recvd_byte / sizeof(int)));
      char err_buf[500];
      sprintf(err_buf, "CORRUPTED DATA: Length written on SendHeader(%d) is invalid. Actual data size is %d. Exting...",
              (int)(*total_buf_nwords * sizeof(int)), temp_length);
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      sleep(1234567);
      exit(-1);
    }
 
  }
 
  if ((int)(*total_buf_nwords * sizeof(int)) != total_recvd_byte) {
    char err_buf[500];
    sprintf(err_buf, "CORRUPTED DATA: Received data size (%d byte) is not same as expected one (%d) from Sendheader. Exting...",
            total_recvd_byte, (int)(*total_buf_nwords * sizeof(int)));
    print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    sleep(1234567);
    exit(-1);
  }
 
  // Read Traeiler
  int send_trl_buf[(unsigned int)(SendTrailer::SENDTRL_NWORDS) ];
  for (int i = 0; i < (int)(m_socket_recv.size()); i++) {
    try {
      recvFD(m_socket_recv[ i ], (char*)send_trl_buf, SendTrailer::SENDTRL_NWORDS * sizeof(int), flag);
    } catch (const string& err_str) {
      if (*delete_flag) {
        //        B2WARNING("Delete buffer before going to Run-pause state");
        printf("[WARNING] Delete buffer before going to Run-pause state\n"); fflush(stdout);
        delete temp_buf;
      }
      throw (std::move(err_str));
    }
  }
 
  return temp_buf;
}

recvFD()

int recvFD ( int  fd, char *  buf, int  data_size_byte, int  flag  )

protected

receive data

Definition at line 60 of file DesSerPrePC.cc.

{
  int n = 0;
  while (1) {
    int read_size = 0;
    if ((read_size = recv(sock, (char*)buf + n, data_size_byte - n, flag)) < 0) {
      if (errno == EINTR) {
        continue;
      } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
#ifdef NONSTOP
        string err_str;
        callCheckRunPause(err_str);
#endif
        continue;
      } else {
        perror("[WARNING]");
        char err_buf[500];
        sprintf(err_buf, "recv() returned error; ret = %d. : %s %s %d",
                read_size, __FILE__, __PRETTY_FUNCTION__, __LINE__);
#ifdef NONSTOP
        m_run_error = 1;
        //        B2ERROR(err_buf);
        printf("[WARNING] %s\n", err_buf); fflush(stdout);
        string err_str = "RUN_ERROR";
        printf("AIUEO********************\n"); fflush(stdout);
        throw (err_str);
#endif
        print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
        exit(-1);
      }
    } else if (read_size == 0) {
      // Connection is closed ( error )
      char err_buf[500];
      sprintf(err_buf, "[WARNING] Connection is closed by peer(%s). readsize = %d %d. : %s %s %d",
              strerror(errno), read_size, errno, __FILE__, __PRETTY_FUNCTION__, __LINE__);
#ifdef NONSTOP
      m_run_error = 1;
      //      B2ERROR(err_buf);
      printf("%s\n", err_buf); fflush(stdout);
      string err_str = "RUN_ERROR";
      throw (err_str);
#else
      print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
      exit(-1);
#endif
    } else {
      n += read_size;
      if (n == data_size_byte)break;
    }
  }
  return n;
}

Send()

int Send ( int  socket, char *  buf, int  size_bytes  )

inherited

send buffer

Definition at line 328 of file DesSer.cc.

{
  int sent_bytes = 0;
  while (true) {
    int ret = 0;
    if ((ret = send(socket,
                    buf + sent_bytes, size_bytes - sent_bytes,  MSG_NOSIGNAL)) < 0) {
      if (errno == EINTR) {
        continue;
      } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
#ifdef NONSTOP
        string err_str;
        callCheckRunPause(err_str);
#endif
        continue;
      } else {
        char err_buf[500];
        sprintf(err_buf, "[WARNING] SEND ERROR.(%s)", strerror(errno));
#ifdef NONSTOP
        m_run_error = 1;
        //        B2ERROR(err_buf);
        printf("%s\n", err_buf); fflush(stdout);
        string err_str = "RUN_ERROR";
        throw (err_str);  // To exit this module, go to DeSerializer** and wait for run-resume.
#else
        print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
        exit(1);
#endif
      }
    }
    sent_bytes += ret;
    if (sent_bytes == size_bytes) break;
  }
  return sent_bytes;
}

sendByWriteV()

int sendByWriteV ( RawDataBlockFormat *  rawdblk )

inherited

Definition at line 227 of file DesSer.cc.

{
  SendHeader send_header;
  SendTrailer send_trailer;
  fillSendHeaderTrailer(&send_header, &send_trailer, rawdblk);
 
  enum {
    NUM_BUFFER = 3
  };
  struct iovec iov[ NUM_BUFFER ];
 
  // check Body data size
  int rawcopper_nwords = rawdblk->TotalBufNwords();
 
  //Fill iov info.
  iov[0].iov_base = (char*)send_header.GetBuffer();
  iov[0].iov_len = sizeof(int) * send_header.GetHdrNwords();
 
  iov[1].iov_base = (char*)rawdblk->GetWholeBuffer();
  iov[1].iov_len = sizeof(int) * rawcopper_nwords;
 
  iov[2].iov_base = (char*)send_trailer.GetBuffer();
  iov[2].iov_len = sizeof(int) * send_trailer.GetTrlNwords();
 
  // Send Multiple buffers
  int n = 0;
  while (true) {
    if ((n = writev(m_socket_send, iov, NUM_BUFFER)) < 0) {
      if (errno == EINTR) {
        continue;
      } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
#ifdef NONSTOP
        string err_str;
        callCheckRunPause(err_str);
#endif
        continue;
      } else {
        char err_buf[500];
        sprintf(err_buf, "[WARNING] WRITEV error.(%s) : sent %d bytes, header %lu bytes body %lu trailer %lu\n",
                strerror(errno), n, iov[0].iov_len, iov[1].iov_len, iov[2].iov_len);
#ifdef NONSTOP
        m_run_error = 1;
        //        B2ERROR(err_buf);
        printf("%s\n", err_buf); fflush(stdout);
        string err_str = "RUN_ERROR";
        throw (err_str);  // To exit this module, go to DeSerializer** and wait for run-resume.
#else
        print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
        exit(1);
#endif
      }
    }
    break;
  }
 
#ifdef DEBUG
  printf("[DEBUG] *******BODY**********\n");
  printf("[DEBUG] \n%.8d : ", 0);
  printData((int*)(iov[1].iov_base), iov[1].iov_len);
#endif
 
 
  int total_send_bytes = sizeof(int) * send_header.GetTotalNwords();
 
 
  //
  // Retry sending
  //
  if (n != total_send_bytes) {
    //    B2WARNING("Serializer: Sent byte(" << n << "bytes) is not same as the event size (" << total_send_bytes << "bytes). Retrying...");
    printf("[WARNING] Serializer: Sent byte( %d bytes) is not same as the event size ( %d bytes). Retrying...\n", n,
           total_send_bytes); // back to WARNING due to request from Konno-san on 2018/7/3
    fflush(stdout);
 
    double retry_start = getTimeSec();
    // Send Header
    if (n < (int)(iov[ 0 ].iov_len)) {
      n += Send(m_socket_send, (char*)iov[ 0 ].iov_base + n, iov[ 0 ].iov_len - n);
    }
 
    if (n < (int)(iov[ 0 ].iov_len + iov[ 1 ].iov_len)) {
      n += Send(m_socket_send, (char*)iov[ 1 ].iov_base + (n - iov[ 0 ].iov_len), iov[ 1 ].iov_len - (n - iov[ 0 ].iov_len));
    }
 
    if (n < (int)(iov[ 0 ].iov_len + iov[ 1 ].iov_len + iov[ 2 ].iov_len)) {
      n += Send(m_socket_send, (char*)iov[ 2 ].iov_base + (n - iov[ 0 ].iov_len - iov[ 1 ].iov_len),
                iov[ 2 ].iov_len - (n - iov[ 0 ].iov_len - iov[ 1 ].iov_len));
    }
    double retry_end = getTimeSec();
    //    B2WARNING("Resending ends. It takes " << retry_end - retry_start << "(s)");
    printf("[WARNING] Resending ends. It takes %lf (s)\n", retry_end - retry_start);
    fflush(stdout); // back to WARNING due to request from Konno-san on 2018/7/3
  }
  //   printf( "[DEBUG] n %d total %d\n", n, total_send_bytes);
  //  delete temp_buf;
 
  return total_send_bytes;
 
}

setRecvdBuffer()

void setRecvdBuffer ( RawDataBlockFormat *  raw_datablk, int *  delete_flag  )

protected

attach buffer to RawDataBlock

Definition at line 336 of file DesSerPrePC.cc.

{
  //
  // Get data from socket
  //
  int total_buf_nwords = 0 ;
  int num_events_in_sendblock = 0;
  int num_nodes_in_sendblock = 0;
 
  if (m_start_flag == 0) {
    //    B2INFO("DeSerializerPrePC: Reading the 1st packet from eb0...");
    printf("DeSerializerPrePC: Reading the 1st packet from eb0...\n"); fflush(stdout);
  }
  int* temp_buf = recvData(delete_flag, &total_buf_nwords, &num_events_in_sendblock,
                           &num_nodes_in_sendblock);
  if (m_start_flag == 0) {
    //    B2INFO("DeSerializerPrePC: Done. the size of the 1st packet " << total_buf_nwords << " words");
    printf("DeSerializerPrePC: Done. the size of the 1st packet %d words\n", total_buf_nwords); fflush(stdout);
    m_start_flag = 1;
  }
  m_recvd_totbytes += total_buf_nwords * sizeof(int);
 
  temp_raw_datablk->SetBuffer((int*)temp_buf, total_buf_nwords, *delete_flag,
                              num_events_in_sendblock, num_nodes_in_sendblock);
 
  //
  // check even # and node # in one Sendblock
  //
  int num_entries = temp_raw_datablk->GetNumEntries();
  if (num_entries != num_events_in_sendblock * num_nodes_in_sendblock) {
    char err_buf[500];
    sprintf(err_buf,
            "[FATAL] CORRUPTED DATA: Inconsistent SendHeader value. # of nodes(%d) times # of events(%d) differs from # of entries(%d). Exiting...",
            num_nodes_in_sendblock, num_events_in_sendblock, num_entries);
    print_err.PrintError(err_buf, __FILE__, __PRETTY_FUNCTION__, __LINE__);
    sleep(1234567);
    exit(-1);
  }
  return;
 
}

shmOpen()

void shmOpen ( char *  path_cfg, char *  path_sta  )

inherited

open shared memory

Definition at line 679 of file DesSer.cc.

{
  errno = 0;
  /*m_shmfd_cfg = shm_open( "/cpr_config2", O_CREAT | O_EXCL | O_RDWR, 0666);
  if (m_shmfd_cfg < 0) {
    if (errno != EEXIST) {
      perror("shm_open1");
      exit(1);
    }
  m_shmfd_cfg = shm_open(path_cfg, O_RDWR, 0666);
  if (m_shmfd_cfg < 0) {
    printf( "[DEBUG] %s\n", path_cfg);
    perror("[ERROR] shm_open2");
    exit(1);
  }
  */
  //}
  /*
  m_shmfd_sta = shm_open( "/cpr_status2", O_CREAT | O_EXCL | O_RDWR, 0666);
  if (m_shmfd_sta < 0) {
    if (errno != EEXIST) {
      perror("shm_open1");
      exit(1);
    }
  m_shmfd_sta = shm_open(path_sta , O_RDWR, 0666);
  if (m_shmfd_sta < 0) {
    printf( "[DEBUG] %s\n", path_sta);
    perror("[ERROR] shm_open2");
    exit(1);
  }
  //}
  int size = 4 * sizeof(int);
  ftruncate(m_shmfd_cfg, size);
  ftruncate(m_shmfd_sta, size);
    */
}

Member Data Documentation

cur_time

double cur_time

inherited

for time monitoring

Definition at line 102 of file DesSer.h.

event_diff

int event_diff

protectedinherited

Definition at line 256 of file DesSer.h.

m_bufary

int* m_bufary[NUM_PREALLOC_BUF]

inherited

buffer

Definition at line 85 of file DesSer.h.

m_buffer

int* m_buffer

inherited

buffer

Definition at line 82 of file DesSer.h.

m_cfg_buf

int* m_cfg_buf

inherited

buffer for shared memory

Definition at line 221 of file DesSer.h.

m_cfg_sta

int* m_cfg_sta

inherited

buffer for shared memory

Definition at line 224 of file DesSer.h.

m_compressionLevel

int m_compressionLevel

inherited

Compression Level.

Definition at line 67 of file DesSer.h.

m_data_type

int m_data_type

inherited

data type

Definition at line 153 of file DesSer.h.

m_dump_fname

std::string m_dump_fname

inherited

dump filename

Definition at line 76 of file DesSer.h.

m_exp_no

int m_exp_no

inherited

exp no.

Definition at line 150 of file DesSer.h.

m_exprunsubrun_no

unsigned int m_exprunsubrun_no

inherited

run no.

Definition at line 144 of file DesSer.h.

m_fp_dump

FILE* m_fp_dump

inherited

dump file descripter

Definition at line 79 of file DesSer.h.

m_hostname_from

std::vector<std::string> m_hostname_from

protectedinherited

Reciever basf2 Socket.

hostname of upstream Data Sources

Definition at line 243 of file DesSer.h.

m_hostname_local

std::string m_hostname_local

protectedinherited

Destination Host.

Definition at line 275 of file DesSer.h.

m_ncycle

int m_ncycle

inherited

Definition at line 100 of file DesSer.h.

m_nodeid

int m_nodeid

inherited

Node ID for SLC.

Definition at line 138 of file DesSer.h.

m_nodename

std::string m_nodename

inherited

Node Name for SLC.

Definition at line 135 of file DesSer.h.

m_num_connections

int m_num_connections

protectedinherited

check data contents

of connections

Definition at line 237 of file DesSer.h.

m_num_usedbuf

int m_num_usedbuf

inherited

of already used buffers

Definition at line 132 of file DesSer.h.

m_port_from

std::vector<int> m_port_from

protectedinherited

port # to connect data sources

Definition at line 246 of file DesSer.h.

m_port_to

int m_port_to

protectedinherited

Destination port.

Definition at line 278 of file DesSer.h.

m_pre_rawcpr

PreRawCOPPERFormat_v2 m_pre_rawcpr

inherited

report status to SLC

Use ver.2 for the moment(ver.4 -> PCI40)

Definition at line 191 of file DesSer.h.

m_prev_copper_ctr

unsigned int m_prev_copper_ctr

protectedinherited

Definition at line 258 of file DesSer.h.

m_prev_evenum

unsigned int m_prev_evenum

protectedinherited

Definition at line 260 of file DesSer.h.

m_prev_exprunsubrun_no

unsigned int m_prev_exprunsubrun_no

inherited

run no.

Definition at line 147 of file DesSer.h.

m_prev_nevt

int m_prev_nevt

inherited

No. of prev sent events.

Definition at line 73 of file DesSer.h.

m_prev_time

double m_prev_time

inherited

Definition at line 105 of file DesSer.h.

m_recvd_prev_totbytes

double m_recvd_prev_totbytes

inherited

Definition at line 97 of file DesSer.h.

m_recvd_totbytes

double m_recvd_totbytes

inherited

Definition at line 96 of file DesSer.h.

m_run_error

int m_run_error

inherited

flag to show that there is some errors with which DAQ cannot continue.

Definition at line 230 of file DesSer.h.

m_run_pause

int m_run_pause

inherited

flag to show that run-controller pauses a run

Definition at line 227 of file DesSer.h.

m_sent_prev_totbytes

double m_sent_prev_totbytes

inherited

Definition at line 99 of file DesSer.h.

m_sent_totbytes

double m_sent_totbytes

inherited

Definition at line 98 of file DesSer.h.

m_shmfd_cfg

int m_shmfd_cfg

inherited

file descripter for shm

Definition at line 168 of file DesSer.h.

m_shmfd_sta

int m_shmfd_sta

inherited

file descripter for shm

Definition at line 171 of file DesSer.h.

m_shmflag

int m_shmflag

inherited

Use shared memory.

Definition at line 159 of file DesSer.h.

m_socket_recv

std::vector<int> m_socket_recv

protectedinherited

Definition at line 248 of file DesSer.h.

m_socket_send

int m_socket_send

protectedinherited

Reciever Socket.

Definition at line 267 of file DesSer.h.

m_start_flag

int m_start_flag

inherited

start flag

Definition at line 181 of file DesSer.h.

m_start_time

double m_start_time

inherited

Definition at line 104 of file DesSer.h.

m_status

RunInfoBuffer m_status

inherited

Run info buffer.

Definition at line 141 of file DesSer.h.

m_t0

timeval m_t0

inherited

Definition at line 95 of file DesSer.h.

m_trunc_mask

int m_trunc_mask

inherited

trunc mask

Definition at line 156 of file DesSer.h.

max_nevt

int max_nevt

inherited

of events in a run

Definition at line 61 of file DesSer.h.

max_seconds

double max_seconds

inherited

time to stop a run

Definition at line 64 of file DesSer.h.

monitor_numeve

int monitor_numeve

inherited

buffer for shared memory

buffer for shared memory

Definition at line 178 of file DesSer.h.

n_basf2evt

int n_basf2evt

inherited

No. of sent events.

Definition at line 70 of file DesSer.h.

p_method

std::string p_method

protectedinherited

EvtSocket.

How to handle data

Definition at line 284 of file DesSer.h.

p_method_val

int p_method_val

protectedinherited

Definition at line 285 of file DesSer.h.

prev_event

int prev_event

inherited

Definition at line 106 of file DesSer.h.

print_err

CprErrorMessage print_err

inherited

wrapper for B2LOG system

Definition at line 184 of file DesSer.h.

tmp_header

RawHeader_v2 tmp_header

protectedinherited

which format is used

Definition at line 323 of file DesSer.h.

---

The documentation for this class was generated from the following files:

• daq/rawdata/include/DesSerPrePC.h
• daq/rawdata/src/DesSerPrePC.cc