rotorooter.cc

Go to the documentation of this file.

// $ID:  rotorooter.C,v 1.2 1998/09/14 16:25 rhatcher Exp $
//
// rotorooter: server for MINOS ROOT file writing

#include <iostream>
#include <stdlib.h>
/* #include <getopt.h> seems to be part of <unistd.h> */
#include <unistd.h>

#include "OnlineUtil/msgLogLib/msgLog.h"
#include "OnlineUtil/minosDaq.h" /* msgLog node ids defined here */

#include <signal.h>
#include <string>
#include <map>

#include "TROOT.h"
#include "TRint.h"
#include "TVirtualX.h"
#include "TGuiFactory.h"

#include "TServerSocket.h"
#include "TSocket.h"
#include "TMessage.h"

#include "Rotorooter/RotoServer.h"
#include "Rotorooter/RotoObjectifier.h"
#include "Rotorooter/MsgLogTErrorHandler.h"

#include "RawData/RawBlockId.h"
#include "RawData/RawBlockRegistry.h"
#include "RawData/RawBlockProxy.h"
#include "RawData/RawDataBlock.h"

#include "RawData/RawRecord.h"

//---- Global sever ----------------------------------------------------

RotoServer* server = 0;

//---- Signal handler --------------------------------------------------
static void roto_sighandler(int dummy)
{
   if (server) {
      logNotice("roto_sighandler signal %d; attempt shutdown",dummy);

      // make sure the clean up code is called
      delete server;
      logNotice("roto_sighandler exit");

   } else {
      logNotice("roto_sighandler signal %d, but no server",dummy);

   }

   msgLogCleanup();

   exit(1);
}

//---- Forward declarations --------------------------------------------
int config_autosave(const char* config);
int config_compress(const char* config);
int config_basketsize(const char* config);
int parse_port(const char* opt, string& inputFileName);

//---- Main program ----------------------------------------------------

int main(int argc, char **argv)
{
   int   copt;

   Bool_t allow_overwrite = kFALSE;
   string inputFileName;
   Int_t  port   = ROOTER_PORT_DAQ;
   Int_t  nwords = 2*1024*1024;
   Int_t  tcp_nodelay_flag = 1;  /* on by default */
   const Char_t* autosave_config = 0;
   const Char_t* compress_config = 0;
   const Char_t* basketsize_config = 0;
   const Char_t* bin_output_file = 0;
   const Char_t* symlink_config = 0;

   // install error handler for Info(),Warning(),Error(),SysError()
   // that redirects messages to msgLog
   //ErrorHandlerFunc_t old_errhndlr = 
   SetErrorHandler(MsgLogTErrorHandler);

#ifdef ALLOW_FORCE
   Detector::Detector_t forceDetector = Detector::kUnknown;
   SimFlag::SimFlag_t forceSimFlag = SimFlag::kUnknown;
   const char* optlist = "aep:b:c:C:B:D:v:z:hd:sl:";
#else
   const char* optlist = "aep:b:c:C:B:D:v:z:hl:";
#endif

   /*
    * parse the options and filenames
    *     -a: allow overwrite
    *     -p: port # (or "dcp"/"daq", "dcs", "beammon")
    *     -b: initial buffer size
    *     -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
    *     -D: tcp_nodelay flag
    *     -e: msgLog local echo
    *     -v: verbosity level
    *     -d: force detector
    *     -s: force simflag
    *     -l: Make symlink to current file
    *     -z: write a binary copy of received data to this file
    *     -h: help
    */

   msgLogInit(argv[0]);
   msgLogNodeIdSet(MINOS_ROOTER);
   logInfo("rotorooter command line configuration\n");

   while ((copt = getopt(argc, argv, optlist)) != EOF) {
      switch (copt) {
      case 'a':
         allow_overwrite = kTRUE;
         break;
      case 'e':
         msgLogLocalEchoSet(1);
         break;
      case 'p':
//       Handle port number or file name
         port = parse_port(optarg, inputFileName);
         break;
      case 'b':
         nwords = atoi(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 'D':
         tcp_nodelay_flag = atoi(optarg);
         break;
      case 'v':
         logDebugLevelSet(atoi(optarg));
         break;
      case 'z':
        bin_output_file = optarg;
        break;
      case 'h':
         printf(" usage: %s -a -p<port#> -b<buffersize> -e\n", argv[0]);
         printf("   -a: allow file overwriting\n");
         printf("   -p: port number rotorooter is listening on\n");
         printf("       or file name or \"dcs\"/\"daq\", \"dcs\", \"beammon\"\n");
         printf("   -b: initial buffer size in long words\n");
         printf("   -e: direct msgLog errors to stdout as well\n");
         printf("   -v: msgLog debug level\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("   -D: TCP_NODELAY flag\n");
         printf("   -z: make copy of flat data to binary file\n");
#ifdef ALLOW_FORCE
         printf("   -d: detector override\n");
         printf("   -s: simflag override\n");
#endif
         printf("   -l: symlink to this file on each successful open\n");
         printf("   -h: print this message\n");
         exit(1);
         break;
      default:
         printf(" unrecognized option '%c'\n",(char)optopt);
         exit(1);
         break;
#ifdef ALLOW_FORCE
      case 'd':
         {
            // be very strict!
            Detector::Detector_t dlist[] = { Detector::kNear, 
                                             Detector::kFar, 
                                             Detector::kCalDet };
            int n = sizeof(dlist)/sizeof(forceDetector);
            string user = optarg;
            // look for a match
            for (int i = 0; i < n; i++) {
               if ( Detector::AsString(dlist[i]) == user) 
                  forceDetector = dlist[i];
            }
            // complain if value didn't match
            if (Detector::kUnknown == forceDetector) {
               printf(" -d flag (force detector type) unknown value: %s\n",
                      user.c_str());
               printf("   must be followed by exactly one of: %s",
                      Detector::AsString(dlist[0]));
               for (int i = 1; i < n; i++ ) printf(", %s",
                      Detector::AsString(dlist[i]));
               printf("\n");
               forceDetector = Detector::kUnknown;
               exit(1);
            }
         }
         break;
      case 's':
         {
            // be very strict!
            SimFlag::SimFlag_t slist[] = { SimFlag::kData, 
                                           SimFlag::kDaqFakeData, 
                                           SimFlag::kMC, 
                                           SimFlag::kReroot };
            int n = sizeof(slist)/sizeof(forceSimFlag);
            string user = optarg;
            // look for a match
            for (int i = 0; i < n; i++) {
               if ( SimFlag::AsString(slist[i]) == user) 
                  forceSimFlag = slist[i];
            }
            // complain if value didn't match
            if (SimFlag::kUnknown == forceSimFlag) {
               printf(" -s flag (force simflag) unknown value: %s\n",
                      user.c_str());
               printf("   must be followed by exactly one of: %s",
                      SimFlag::AsString(slist[0]));
               for (int i = 1; i < n; i++ ) printf(", %s",
                      SimFlag::AsString(slist[i]));
               printf("\n");
               forceSimFlag = SimFlag::kUnknown;
               exit(1);
            }
         }
         break;
#endif
      case 'l':
          symlink_config = optarg;
          break;
      }
   }

   logNotice("starting %s port %d, inputFileName '%s'",
             argv[0],port,inputFileName.c_str());

   // use global so handler can do a clean shutdown
   if ( port ) 
     server = new RotoServer(port,nwords,allow_overwrite,tcp_nodelay_flag);
   else 
     server = new RotoServer(inputFileName.c_str(),nwords,allow_overwrite);

   /* Trap HUP, INT and TERM signals so that we close down gracefully */
   struct sigaction termAction ;
   termAction.sa_handler = roto_sighandler ;
   sigemptyset (&termAction.sa_mask) ;
   termAction.sa_flags = 0 ;
   sigaction (SIGHUP, &termAction, NULL) ;
   sigaction (SIGINT, &termAction, NULL) ;
   sigaction (SIGTERM, &termAction, NULL) ;

#ifdef ALLOW_FORCE
   if (Detector::kUnknown != forceDetector) {
      // DANGER! DANGER! DANGER!
      logNotice("DANGER! overriding detector in all block IDs: %s",
                Detector::AsString(forceDetector));
      RotoObjectifier::SetForceDetector(forceDetector);
   }
   if (SimFlag::kUnknown != forceSimFlag) {
      // DANGER! DANGER! DANGER!
      logNotice("DANGER! overriding simflag in all block IDs: %s",
                SimFlag::AsString(forceSimFlag));
      RotoObjectifier::SetForceSimFlag(forceSimFlag);
   }
#endif

   // run ROOT in batch mode (ie. no X11)
   // based on what root does when passed -b flag
   gROOT->SetBatch();
   if (gGuiFactory != gBatchGuiFactory) delete gGuiFactory;
   gGuiFactory = gBatchGuiFactory;
#ifndef WIN32
   if (gVirtualX != gGXBatch) delete gVirtualX;
#endif
   gVirtualX = gGXBatch;

   // configure initial autosave and compression values
   Int_t nerr = 0;
   if (autosave_config)   nerr += config_autosave(autosave_config);
   if (compress_config)   nerr += config_compress(compress_config);
   if (basketsize_config) nerr += config_basketsize(basketsize_config);

   if (bin_output_file)   nerr += server->SetFlatBinaryOutputFile(bin_output_file);

   if (symlink_config) server->SetSymlink(symlink_config);

   server->Run();
   delete server;

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

   return 0;
}

/*
 *=========================================================================
 * 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
 *=========================================================================
 */
int config_autosave(const char* config)
{
   int nerr = 0;
   int 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;
   int 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) {
         logNotice("config_autosave: only %d items from \"%s\"\n",nitems,p2);
         nerr += 1;
      }
      else 
         nerr += server->SetAutoSaveConfig(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
 *=========================================================================
 */
int config_compress(const char* config)
{
   int nerr = 0;
   int 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;
   int 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) {
         logNotice("config_compress: only %d items from \"%s\"\n",nitems,p2);
         nerr += 1;
      }
      else 
         nerr += server->SetCompressConfig(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,size[;streamName,size]"
 *             use "*" as streamName to set all legal streams
 * 
 * Return Value:
 *    non-zero indicates there was an error
 *=========================================================================
 */
int config_basketsize(const char* config)
{
   int nerr = 0;
   int 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;
   int basketsize, 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,&basketsize);
      if (nitems != 2) {
         logNotice("config_basketsize: only %d items from \"%s\"\n",nitems,p2);
         nerr += 1;
      }
      else 
         nerr += server->SetBasketSizeConfig(stream,basketsize);

      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;
}

/*
 *=========================================================================
 * parse_port(const char* optarg, string& inputFileName)
 * 
 * Purpose:
 *    Parse the -p argument for port # or file name
 *       numeric values get converted and returned
 *       "daq", "dcp", "dcs", "beammon" return the correct numeric #
 *       others are treated as filenames
 *
 * Arguments:
 *    optarg:          option value associated w/ -p flag
 *    inputFileName:   file to use if value doesn't resolve to numeric port #
 * 
 * Return Value:
 *    port # to use, 0 if one is to use inputFileName.
 *=========================================================================
 */
int parse_port(const char* optarg, string& inputFileName)
{
  int  port = ROOTER_PORT_DAQ;

  enum ptype { kNumeric, kSpecial, kIsFile };
  ptype pType = kNumeric;  // start by assuming numeric value
  // bool input_from_file = kFALSE;

  // look for non-digit characters
  const char* c = optarg;
  while ( *c != '\0' ) {
    if ( ! isdigit(*c) ) {
      //input_from_file = kTRUE;
      pType = kIsFile;
      inputFileName = optarg;
      break;
    }
    c++;
  }

  // if (provisionally) from a file (ie. not all digits) 
  // look for special cases
  //if ( input_from_file ) {
  if ( pType != kNumeric ) {

    std::map<string,int> specialNames;
    specialNames["dcp"]     = ROOTER_PORT_DAQ;
    specialNames["DCP"]     = ROOTER_PORT_DAQ;
    specialNames["daq"]     = ROOTER_PORT_DAQ;
    specialNames["DAQ"]     = ROOTER_PORT_DAQ;
    specialNames["dcs"]     = ROOTER_PORT_DCS;
    specialNames["DCS"]     = ROOTER_PORT_DCS;
    specialNames["beammon"] = ROOTER_PORT_BEAMMON;
    specialNames["BEAMMON"] = ROOTER_PORT_BEAMMON;
    specialNames["BeamMon"] = ROOTER_PORT_BEAMMON;

    int specialPort = specialNames[inputFileName];
    if (specialPort) { 
      port = specialPort;
      //input_from_file = kFALSE;
      pType = kSpecial;
    }
  }

  if      ( pType == kNumeric) port = atoi(optarg);
  else if ( pType == kIsFile ) port = 0;

  return port;
}

---