bogus_dcs.c

Go to the documentation of this file.

#include <stdio.h>
/* #include <getopt.h> seems to be part of <unistd.h> */
#include <unistd.h>
/* stdlib.h:  malloc and atoi */
#include <stdlib.h>
/* string.h: strrchr */
#include <string.h>

/* int32_t & uint32_t */
#include <stdint.h>

#include "OnlineUtil/msgLogLib/msgLog.h"
#include "OnlineUtil/rototalk.h"

/* timeval, gettimeofday */
#include <time.h>
#include <sys/time.h>

#include "OnlineUtil/rawBlockIds.h"
#include "OnlineUtil/rdChecksum.h"
#include "OnlineUtil/rotoMessages.h"

/* forward declarations */
int32_t send_bogus_stuff(int32_t* buffer, int32_t bsize, int32_t mxrec);
int32_t parse_detector(const char* detname);
int32_t config_autosave(const char* config);
int32_t config_compress(const char* config);
int32_t config_basketsize(const char* config);
int32_t build_dcs_blockid(int32_t detector, int32_t major, int32_t version);

int32_t* append_dcs_header_block(int32_t* buffer, int32_t* toofar);
int32_t* append_dcs_monitor_block(int32_t* buffer, int32_t* toofar);
int32_t* append_dcs_alarm_block(int32_t* buffer, int32_t* toofar);

/* (default) maximum buffer size for raw records (in bytes) */
#define MAXBUFFER 8388608

/* globals */
int32_t gDetector = kMdBlockSourceFarDetector; /* 1=Near, 2=Far, 4=CalDet */

/*
 *=========================================================================
 * main(int argc, char **argv)
 * 
 * Purpose:
 *    Generate and send a series of fake DCS records to the Rotorooter
 *
 *   usage: bogus_dcs -i<hostname> -b<buffer size> 
 *           -w<whoami> -p<port #> 
 *           -c"autosave_configstring" 
 *           -C"compress_configstring" 
 *           -B"basketsize_configstring" 
 *
 *     -i: hostname where rotorooter is running
 *     -b: buffer size to use
 *     -d: detector ("near","far","caldet")
 *     -c: autosave config    "streamName,nrec,nsec[;streamName,nrec,nsec]" 
 *     -C: compression config "streamName,level[;streamName,level]"
 *     -B: basketsize config "streamName,size[;streamName,size]"
 *            both -c,-C,-B use "*" as streamName to set all legal streams
 *     -w: DCP or DCS
 *     -p: port number rotorooter is listening on
 *     -v: how verbose to be
 *     -n: maximum number of records from each file
 *     -h: print this message
 *
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int main(int argc, char **argv)
{

   /* this is C ... all declarations must come first */
   const char* iphost = "localhost";
   const char* default_producer = "DCS";       /* default producer */
   const int   default_port     = 9012;        /* default DCS port */
   const char* producer = default_producer;
   const char* autosave_config = 0;
   const char* compress_config = 0;
   const char* basketsize_config = 0;
   int32_t port = default_port;
   int32_t whoami = MINOS_ROOTER_DCS;       
   int32_t bsize = MAXBUFFER;
   int32_t* buffer = 0;
   int32_t nerr = 0;
   int32_t mxrec = 1000;  /* no input file, so limit how much stuff to send */
   int32_t nodelay_flag = 1;  /* on by default */
   int32_t echoMsgLog = 0;
   int32_t copt;


   /*
    * parse the options and filenames
    */
   while ((copt = getopt(argc, argv, "i:b:d:c:C:B:w:p:ev:n:D:h")) != EOF) {
      switch (copt) {
      case 'i':  /* internet host */
         iphost = optarg;
         break;
      case 'b':  /* buffer size */
         bsize = atoi(optarg);
         break;
      case 'd':  /* detector */
         gDetector = parse_detector(optarg);
         break;
      case 'c':  /* autosave config string */
         autosave_config = optarg;
         break;
      case 'C':  /* compression config string */
         compress_config = optarg;
         break;
      case 'B':  /* basketsize config string */
         basketsize_config = optarg;
         break;
      case 'w':  /* whoami:  DCP, DCS, other */
         producer = optarg;
         break;
      case 'p':  /* port # */
         port = atoi(optarg);
         break;
      case 'e':  /* msgLog messages to stdout as well */
         echoMsgLog = 1;
         break;
      case 'v':  /* verbosity */
         roto_verbose = atoi(optarg);
         break;
      case 'n':  /* maximum number of records */
         mxrec = atoi(optarg);
         break;
      case 'D':  /* set nodelay_flag? */
         nodelay_flag = atoi(optarg);
         break;
      case 'h':  /* help */
         printf(" usage: %s -i<hostname> -b<buffer size> -w<whoami> -p<port #> <filenames..>\n", argv[0]);
         printf("   -i: hostname where rotorooter is running\n");
         printf("   -b: buffer size to use\n");
         printf("   -d: detector ('near','far','caldet')\n");
         printf("   -w: DCP, DCS or BeamMon\n");
         printf("   -p: port number rotorooter is listening on\n");
         printf("   -v: how verbose to be\n");
         printf("   -c: autosave config \"streamName,nrec,nsec[;streamName,nrec,nsec]\"\n");
         printf("   -C: compression config \"streamName,level[;streamName,level]\"\n");
         printf("   -B: basketsize config \"streamName,size[;streamName,size]\"\n");

         printf("   -n: maximum number of records from each file\n");

         printf("   -h: print this message\n");
         exit(1);
      default:
         printf(" unrecognized option '%c' ignored\n",(char)optopt);
         break;
      }
   }

   msgLogInit(argv[0]);
   msgLogNodeIdSet(whoami);
   msgLogLocalEchoSet(echoMsgLog);
   logDebugLevelSet(3);
   logNotice("starting %s",argv[0]);

   /* allocate a buffer to use */
   buffer = (int32_t*) malloc(bsize);
   if (!buffer) {
      printf("failed to allocate buffer of size %d\n",bsize);
      exit(1);
   }
   if (roto_verbose>1) printf("allocated buffer of size %d\n",bsize);

   /* open a connection to the Rotorooter */
   if (0 == strcmp(producer,default_producer) && 
       port == default_port) {
      /* standard connection */
      if (roto_verbose>1) 
         printf("connect via roto_open_DCP_connection\n");
      nerr += roto_open_DCS_connection(iphost);
   } else {
      /* non-standard connection requested (either producer or port) */
      if (0 == strcmp(producer,"DCP")) {
         whoami  = MINOS_ROOTER_DCP;
      }
      else if (0 == strcmp(producer,"DCS")) {
         whoami = MINOS_ROOTER_DCS;
      }
      else if (0 == strcmp(producer,"BeamMon")) {
         whoami = MINOS_ROOTER_BEAMMON;
      }
      else 
         whoami = MINOS_ROOTER_UNIDENTIFIED_CLIENT;

      if (roto_verbose>1) 
         printf("connect via roto_open_connection: port %d, whoami %s (0x%2.2x)\n",port,producer,whoami);
      nerr += roto_open_connection(iphost,port,whoami);
   }
   if (nerr) exit(nerr);
   
   nerr += roto_set_connection_nodelay(nodelay_flag);

   if (autosave_config) nerr += config_autosave(autosave_config);
   if (compress_config) nerr += config_compress(compress_config);
   if (basketsize_config) nerr += config_basketsize(basketsize_config);

   nerr += send_bogus_stuff(buffer,bsize,mxrec);

   /* close the connection */
   nerr += roto_close_connection();

   /* set my buffer free */
   free(buffer);

   logNotice("stopping %s",argv[0]);
   msgLogCleanup();

   exit(nerr);
}

/*
 *=========================================================================
 * send_bogus_stuff(long* buffer, int bsize, int mxrec) 
 * 
 * Purpose:
 *    Send a series of bogus DCS records to the rotorooter
 *
 * Arguments:
 *    buffer:  ptr to pre-allocated buffer
 *    bsize:   buffer size (bytes)
 *    mxrec:   maximum number of records to send (0=all)
 *
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int32_t send_bogus_stuff(int32_t* buffer, int32_t bsize, int32_t mxrec) 
{

   int32_t   nRecords=0;
   int32_t   nbytes;

   int32_t* current;
   int32_t* endofbuf = buffer + bsize/sizeof(int32_t);

   int32_t nerr = 0;

   struct timeval now;
   gettimeofday(&now,0);

   nerr += roto_open_dcsfile(gDetector,now.tv_sec,now.tv_usec*1000);
   if (nerr) {
      printf("Rotorooter failed to open output file\n");
      return 1;
   }

   while (1) {
      if (mxrec>0 && nRecords==mxrec) break;
      nRecords++;

      /* 
       * fill a new record 
       */

      current = buffer; /* start at beginning of allocated buffer */

      /* all records start with a header block */
      current = append_dcs_header_block(current,endofbuf);

      /* 
       * add additional blocks
       * each function decides whether there is something to add
       */
      current = append_dcs_alarm_block(current,endofbuf);
      current = append_dcs_monitor_block(current,endofbuf);

      /* 
       * send the record
       */
      nbytes = (current-buffer)*sizeof(int32_t); /* calc used size */
      nerr += roto_send_record(buffer,nbytes);

      logDebug(5,"sent record %d of length %d bytes",nRecords,nbytes);

   }

   nerr += roto_close_dcsfile();
   return nerr;
}

/*
 *=========================================================================
 * parse_detector(const char* detname)
 * 
 * Purpose:
 *    Turn detector name into detector source.
 *
 * Arguments:
 *    detname:  detector name should be one of 'near','far',caldet'
 * 
 * Return Value:
 *    detector enum value
 *=========================================================================
 */
int32_t parse_detector(const char* detname)
{

  switch (detname[0]) {
  case 'n':
  case 'N':
  case '1':
    return kMdBlockSourceNearDetector;
    break;
  case 'f':
  case 'F':
  case '2':
    return kMdBlockSourceFarDetector;
    break;
  case 'c':
  case 'C':
  case '4':
    return kMdBlockSourceCalDetector;
    break;
  default:
    logWarn("parse_detector: could not decipher '%s', using value %d",
            detname,gDetector);
    return gDetector;
  }
}

/*
 *=========================================================================
 * config_autosave(const char* config)
 * 
 * Purpose:
 *    Send autosave config to the rotorooter, parsed from string
 *
 * Arguments:
 *    config:  configuration string of form:
 *             "streamName,nrec,nsec[;streamName,nrec,nsec]"
 *             use "*" as streamName to set all legal streams
 * 
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int32_t config_autosave(const char* config)
{
   int32_t nerr = 0;
   int32_t len = strlen(config);
   char *wcopy = (char*)malloc(len+1);  /* a writable copy of config string */
   char stream[1024];
   const char *p1;
   char *p2;
   char achar;
   int32_t nrec,nsec, nitems;

   /* make a copy, replacing all the commas with blanks*/
   p1 = config; p2 = wcopy;
   while (p2 < wcopy+len) {
      achar = *p1;
      if (achar != ',') *p2 = achar;
      else              *p2 = ' ';
      p1++; p2++;
   }

   p2 = wcopy;
   while (p2  < wcopy+len) {
      nitems = sscanf(p2,"%s %d %d",stream,&nrec,&nsec);
      if (nitems != 3) {
         logInfo("config_autosave: only %d items from \"%s\"\n",nitems,p2);
         nerr += rototalk_err_badconfig;
      }
      else 
         nerr += roto_send_autosave_config(stream,nrec,nsec);

      p2 = strchr(p2,';'); /* move to the semicolon         */
      if (!p2) break;      /* in case of no final semicolon */
      p2++;                /* start just beyond semicolon   */
   }


   free(wcopy);  /* return the allocated space */
   wcopy = 0;

   return nerr;
}

/*
 *=========================================================================
 * config_compress(const char* config)
 * 
 * Purpose:
 *    Send compression config to the rotorooter, parsed from string
 *
 * Arguments:
 *    config:  configuration string of form:
 *             "streamName,level[;streamName,level]"
 *             use "*" as streamName to set all legal streams
 * 
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int32_t config_compress(const char* config)
{
   int32_t nerr = 0;
   int32_t len = strlen(config);
   char *wcopy = (char*)malloc(len+1);  /* a writable copy of config string */
   char stream[1024];
   const char *p1;
   char *p2;
   char achar;
   int32_t level, nitems;

   /* make a copy, replacing all the commas with blanks*/
   p1 = config; p2 = wcopy;
   while (p2 < wcopy+len) {
      achar = *p1;
      if (achar != ',') *p2 = achar;
      else              *p2 = ' ';
      p1++; p2++;
   }

   p2 = wcopy;
   while (p2  < wcopy+len) {
      nitems = sscanf(p2,"%s %d",stream,&level);
      if (nitems != 2) {
         logInfo("config_compress: only %d items from \"%s\"\n",nitems,p2);
         nerr += rototalk_err_badconfig;
      }
      else 
         nerr += roto_send_compress_config(stream,level);

      p2 = strchr(p2,';'); /* move to the semicolon         */
      if (!p2) break;      /* in case of no final semicolon */
      p2++;                /* start just beyond semicolon   */
   }


   free(wcopy);  /* return the allocated space */
   wcopy = 0;

   return nerr;
}

/*
 *=========================================================================
 * config_basketsize(const char* config)
 * 
 * Purpose:
 *    Send basketsize config to the rotorooter, parsed from string
 *
 * Arguments:
 *    config:  configuration string of form:
 *             "streamName,bsize[;streamName,bsize]"
 *             use "*" as streamName to set all legal streams
 * 
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int32_t config_basketsize(const char* config)
{
   int32_t nerr = 0;
   int32_t len = strlen(config);
   char *wcopy = (char*)malloc(len+1);  /* a writable copy of config string */
   char stream[1024];
   const char *p1;
   char *p2;
   char achar;
   int32_t bsize, nitems;

   /* make a copy, replacing all the commas with blanks*/
   p1 = config; p2 = wcopy;
   while (p2 < wcopy+len) {
      achar = *p1;
      if (achar != ',') *p2 = achar;
      else              *p2 = ' ';
      p1++; p2++;
   }

   p2 = wcopy;
   while (p2  < wcopy+len) {
      nitems = sscanf(p2,"%s %d",stream,&bsize);
      if (nitems != 2) {
         logInfo("config_basketsize: only %d items from \"%s\"\n",nitems,p2);
         nerr += rototalk_err_badconfig;
      }
      else 
         nerr += roto_send_basketsize_config(stream,bsize);

      p2 = strchr(p2,';'); /* move to the semicolon         */
      if (!p2) break;      /* in case of no final semicolon */
      p2++;                /* start just beyond semicolon   */
   }


   free(wcopy);  /* return the allocated space */
   wcopy = 0;

   return nerr;
}

/*
 *=========================================================================
 * long build_dcs_blockid(int detector, int major, int version)
 * 
 * Purpose:
 *    Generate a block id  (force DCS, bogus simflag)
 *
 * Arguments:
 *    detector : 1=near, 2=far, 4=caldet
 *    major    : major block id
 *    version  : minor version #
 * 
 * Return Value:
 *    return constructed block id
 *
 * The details of the encoding of RawBlockId are here in RawBlockId.h;
 * this is so that the compiler can inline the code.
 * Users should NOT rely on intimate knowledge of these details.
 *
 *    MSB                          LSB
 *     3         2         1         0
 *    10987654321098765432109876543210
 *    --xxCFNSMMMMMMMMMMMMMMMMmmmmmmmm
 *
 *    88887777666655554444333322221111 
 *
 *         xx: compactSimFlag: {00=data, 01=fakedaqdata, 10=MC, 11=Reroot}
 *        CFN: DetectorType & 0x07 (CalDet,Far,Near bits)
 *          S: Source (0=DAQ, 1=DCS)
 *   MMMMMMMM: major block id
 *   mmmmmmmm: minor block id
 *
 *=========================================================================
 */
int32_t build_dcs_blockid(int32_t detector, int32_t major, int32_t version)
{
   const int32_t  shiftRawBlkIdMinor    =  0;
   const int32_t  shiftRawBlkIdMajor    =  8;
   const int32_t  shiftRawBlkIdDetector = 25;
   const int32_t  shiftRawBlkIdCSimFlag = 28;
   
   const int32_t  maskRawBlkIdMinor     = 0x000000ff;
   const int32_t  maskRawBlkIdMajor     = 0x00ffff00;
   const int32_t  maskRawBlkIdIsDCS     = 0x01000000;
   const int32_t  maskRawBlkIdDetector  = 0x0e000000;
   const int32_t  maskRawBlkIdCSimFlag  = 0x30000000;

   int32_t cmptSimFlag = 0;  /* compact simflag for real Data */
   int32_t isDCS = 1;

   return
      ((cmptSimFlag << shiftRawBlkIdCSimFlag) & maskRawBlkIdCSimFlag) |
      ((   detector << shiftRawBlkIdDetector) & maskRawBlkIdDetector) |
      ((isDCS) ? maskRawBlkIdIsDCS : 0)                               |
      ((         major   << shiftRawBlkIdMajor) & maskRawBlkIdMajor)  |
      ((         version << shiftRawBlkIdMinor) & maskRawBlkIdMinor);

}

/*
 *=========================================================================
 * long* append_dcs_header_block(long* buffer, long* toofar)
 * 
 * Purpose:
 *    Append into buffer a DCS Header block
 *
 * Arguments:
 *    buffer   : pointer to start of where add data
 *    toofar   : pointer to position beyond end of buffer
 * 
 * Return Value:
 *    return starting position for new additional blocks
 *=========================================================================
 */
int32_t* append_dcs_header_block(int32_t* buffer, int32_t* toofar)
{
  /*
   * 0 DcsHeaderBlock size
   * 1                checksum
   * 2                blockid
   * 3                sec
   * 4                ns
   */

  int32_t  version = 0;
  int32_t  blksize = 5;
  struct timeval now;

  /* check that adding this won't overflow buffer */
  if (buffer+blksize > toofar) {
    logError("append_dcs_header_block: need %ld have %d",
             blksize,toofar-buffer);
    return buffer;
  }

  /* fill in the data */

  buffer[0] = blksize;
  /* skip checksum */
  buffer[2] = build_dcs_blockid(gDetector,kMdBlockDcsHeader,version);

  /* for now tag record with the current system time */
  gettimeofday(&now,0);
  buffer[3] = now.tv_sec;
  buffer[4] = now.tv_usec*1000;

  /* now fill in checksum */
  buffer[1] = rdxsum_calc(buffer+0,1);

  return buffer + buffer[0];
}

/*
 *=========================================================================
 * long* append_dcs_alarm_block(long* buffer, long* toofar)
 * 
 * Purpose:
 *    Append into buffer a (generic, ie. bogus) DCS Alarm block
 *
 * Arguments:
 *    buffer   : pointer to start of where add data
 *    toofar   : pointer to position beyond end of buffer
 * 
 * Return Value:
 *    return starting position for new additional blocks
 *=========================================================================
 */
int32_t* append_dcs_alarm_block(int32_t* buffer, int32_t* toofar)
{
  /*
   * 0 DcsAlarmBlock  size
   * 1                checksum
   * 2                blockid
   * ...              <stuff>
   */

  int32_t  version = 0;
  int32_t blksize, payload, j;

  /* 
   * determine whether this block is to go out on this record
   */

  /* for this case pick a random frequency (see Linux "man -s 3 rand")*/
  /* here pick 10% chance of adding one */
  if ( rand()/(RAND_MAX+1.0) > 0.10 ) return buffer;

  /* 
   * determine how big this block is going to be when done
   */

  /* for this case pick a random size (1 to 30 words) for the data payload */
  payload = 1 + (int32_t)(30.*rand()/(RAND_MAX+1.0));
  blksize = 3 + payload;

  /* check that adding this won't overflow buffer */
  if (buffer+blksize > toofar) { 
    logError("append_dcs_alarm_block: need %ld have %d",
             blksize,toofar-buffer);
    return buffer;
  }

  /* fill in the data */

  buffer[0] = blksize;
  /* skip checksum */
  buffer[2] = build_dcs_blockid(gDetector,kMdBlockDcsAlarm,version);

  /* payload of random numbers */
  for (j=0; j<payload; ++j) buffer[j+3] = rand();

  /* now fill in checksum */
  buffer[1] = rdxsum_calc(buffer+0,1);

  return buffer + buffer[0];
}

/*
 *=========================================================================
 * long* append_dcs_monitor_block(long* buffer, long* toofar)
 * 
 * Purpose:
 *    Append into buffer a (generic, ie. bogus) DCS Monitor block
 *
 * Arguments:
 *    buffer   : pointer to start of where add data
 *    toofar   : pointer to position beyond end of buffer
 * 
 * Return Value:
 *    return starting position for new additional blocks
 *=========================================================================
 */
int32_t* append_dcs_monitor_block(int32_t* buffer, int32_t* toofar)
{
  /*
   * 0 DcsMonitorBlock  size
   * 1                checksum
   * 2                blockid
   * ...              <stuff>
   */

  int32_t  version = 0;
  int32_t blksize, payload, j;

  /* 
   * determine whether this block is to go out on this record
   */

  /* for this case pick a random frequency (see Linux "man -s 3 rand")*/
  /* here pick 90% chance of adding one */
  if ( rand()/(RAND_MAX+1.0) > 0.90 ) return buffer;

  /* 
   * determine how big this block is going to be when done
   */

  /* for this case pick a random size (30 to 300 words) for the data payload */
  payload = 30 + (int32_t)(300.*rand()/(RAND_MAX+1.0));
  blksize = 3 + payload;

  /* check that adding this won't overflow buffer */
  if (buffer+blksize > toofar) { 
    logError("append_dcs_monitor_block: need %ld have %d",
             blksize,toofar-buffer);
    return buffer;
  }

  /* fill in the data */

  buffer[0] = blksize;
  /* skip checksum */
  buffer[2] = build_dcs_blockid(gDetector,kMdBlockDcsMonitor,version);

  /* payload of random numbers */
  for (j=0; j<payload; ++j) buffer[j+3] = rand();

  /* now fill in checksum */
  buffer[1] = rdxsum_calc(buffer+0,1);

  return buffer + buffer[0];
}

---