RotoClientModule.cxx

Go to the documentation of this file.

// $Id: RotoClientModule.cxx,v 1.17 2007/07/09 00:50:06 rhatcher Exp $
//
// A JobControl Module for sending raw blocks to the Roto-Rooter
//
// rhatcher@fnal.gov

#include "Rotorooter/RotoClientModule.h"
#include "Rotorooter/RotoClientBinaryFile.h"
#include "Rotorooter/RotoClient.h"
#include "Rotorooter/RotoObjectifier.h"

#include "MinosObjectMap/MomNavigator.h"
#include "MessageService/MsgService.h"
#include "JobControl/JobCModuleRegistry.h"
#include "JobControl/JobCommand.h"

#include "RawData/RawRecord.h"
#include "RawData/RawDaqHeader.h"
#include "RawData/RawDataBlock.h"

#include "RawData/RawDaqHeaderBlock.h"
#include "RawData/RawRunStartBlock.h"
#include "RawData/RawRunEndBlock.h"

#include "TSystem.h"

const Char_t *dfltHost = "localhost";
const Int_t   dfltPort = 9011;

const Bool_t  dfltPrimary     = true;
const Int_t   dfltBufferWords = 6 * 1024 * 1024;  // 6*sizeof(int) Mbytes

const UInt_t dbg_DumpHead         = 0x0001;
const UInt_t dbg_DumpAddBlock     = 0x0002;
const UInt_t dbg_PrintAddBlock    = 0x0004;
const UInt_t dbg_PrintHexAddBlock = 0x0008;

ClassImp(RotoClientModule)

//......................................................................

CVSID("$Id: RotoClientModule.cxx,v 1.17 2007/07/09 00:50:06 rhatcher Exp $");
JOBMODULE(RotoClientModule, "RotoClientModule",
         "Sends RawRecord info to Roto-Rooter");

//......................................................................

RotoClientModule::RotoClientModule() 
   : fRotoClient(0), fBufferWords(dfltBufferWords), fBuffer(0), 
     fUseRototalk(true), fGenFakeRec(true), fDebugFlags(0)
{
   // construct a new "RotoClientModule" JobControl module

   // MSG("Roto", Msg::kInfo) << "RotoClientModule::ctor" << endl;

   // everyone has a shot at being the primary
   fPrimary = dfltPrimary;
   
   // pick up server host from env variable (if possible)
   const char *ROTOSERVER = gSystem->Getenv("ROTOSERVER");
   if (!ROTOSERVER || strlen(ROTOSERVER) == 0) {
      ROTOSERVER = dfltHost;
   }
   SetHostPort(ROTOSERVER,0);
  
   MSG("Roto", Msg::kInfo) << "  Initially configured for host \""
                           << fHost.c_str() << "\"  Port " << fPort << endl;
}

//......................................................................

RotoClientModule::~RotoClientModule() 
{
   // destruct this "RotoClientModule"

   // MSG("Roto", Msg::kInfo) << "RotoClientModule::dtor" << endl;

   DestroyClient();
   DestroyBuffer();
}

//......................................................................

JobCResult RotoClientModule::Put(const MomNavigator *mom)
{
   // Turn a RawRecord into a flat Int_t buffer
   // pass it on to RotoClient to send to the Roto-rooter (or binary file)
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::Put" << endl;

   JobCResult jobcStatus(JobCResult::kAOK); // All Ok

   int* iptr_data   = GetBuffer();  // get a buffer, create one if needed
   if (!iptr_data) return JobCResult::kError;

   RotoClient* client = GetClient();
   if (!client) return JobCResult::kError;

   Bool_t ok = true;

   // Squish the record(s) flat
   // Note there may be more than one RawRecord in MOM
   TObject *tobj = 0;
   TIter reciter = const_cast<MomNavigator*>(mom)->FragmentIter();
   while ( (tobj = reciter() ) ) {
      RawRecord *rawrec = dynamic_cast<RawRecord*>(tobj);
      if (!rawrec) continue;

      Char_t* iptr_bytes = (Char_t*) iptr_data;
      Int_t     maxbytes = fBufferWords * sizeof(Int_t)/sizeof(Char_t);
      Int_t  nbytes_used = 
         RotoObjectifier::BufferSquish(rawrec,iptr_bytes,maxbytes);

      VldContext vldc = rawrec->GetRawHeader()->GetVldContext(); // always okay

      const RawDaqHeader *rawdaqhead = 
         dynamic_cast<const RawDaqHeader *>(rawrec->GetRawHeader());
      Int_t run    = rawdaqhead->GetRun();
      Int_t subrun = rawdaqhead->GetSubRun();

      ReOpenOutputFile(vldc.GetDetector(),run,subrun,&vldc,fGenFakeRec);
   
      // send the buffer to the RotoRooter
      ok &= client->SendRecordBuffer(iptr_bytes,nbytes_used);
   }

   if (ok) return JobCResult::kPassed;  // All Ok
   return jobcStatus;
}

//......................................................................
void RotoClientModule::BeginJob()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::BeginJob" << endl;
}
//......................................................................
void RotoClientModule::EndJob()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::EndJob" << endl;

   // ensure that last file is closed
   ReOpenOutputFile(-1,-1,-1,0,fGenFakeRec);
}
//......................................................................
void RotoClientModule::BeginFile()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::BeginFile" << endl;
}
//......................................................................
void RotoClientModule::EndFile()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::EndFile" << endl;
}
//......................................................................
void RotoClientModule::BeginRun()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::BeginRun" << endl;
}
//......................................................................
void RotoClientModule::EndRun()
{
   // 
   // MSG("Roto", Msg::kInfo) << "RotoClientModule::EndRun" << endl;
}
//......................................................................
void RotoClientModule::Report()
{
   // Report (to message service) current configuration/status
   MSG("Roto", Msg::kInfo) << "RotoClientModule::Report" << endl;
   
   MSG("Roto", Msg::kInfo) << "RotoClientModule has:" << endl;

   // information about the client
   if (fRotoClient) {
      RotoClient* client = GetClient();
      if (client) {
         int maxbytes = 1024;
         char* buffer = new char[maxbytes];
         client->GetRotoStatus(buffer,maxbytes);
         if (fPort >= 0) {         
            MSG("Roto", Msg::kInfo) 
               << "   connected client to \"" 
               << fHost.c_str() 
               << "\" Port "
               << fPort 
               << endl
               << "   state_report: \"" << buffer << "\""
               << endl;            
         } else {
            MSG("Roto", Msg::kInfo) 
               << "   open binary file \"" 
               << fHost.c_str() 
               << "\" " 
               << endl
               << buffer
               << endl;
         }
         delete [] buffer;
      } else {
            MSG("Roto", Msg::kInfo) 
               << "   connected client to \"" 
               << fHost.c_str() 
               << "\" Port "
               << fPort 
               << endl;
      }            
   } else {
      MSG("Roto", Msg::kInfo) 
         << "   no currently connected client" 
         << endl
         << "      configured to use to \"" 
         << fHost.c_str() 
         << "\" Port "
         << fPort 
         << endl;
   }

   // information about the buffer
   if (fBuffer) {
      MSG("Roto", Msg::kInfo) 
         << "   buffer of " 
         << fBufferWords 
         << " words"
         << endl;
   } else {
      MSG("Roto", Msg::kInfo) 
         << "   no buffer"
         << endl
         << "      configured to use "
         << fBufferWords 
         << " words"
         << endl;
   }
  
   // other info
   MSG("Roto", Msg::kInfo) 
      << "   primary flag set to " << fPrimary
      << endl
      << "   DebugFlags set to 0x" << hex << setw(8) << fDebugFlags << dec
      << endl
      << endl;
}

//......................................................................
void RotoClientModule::HandleCommand(JobCommand *command)
{
   //
   // Process configuration commands
   //

   // MSG("Roto", Msg::kDebug) << "RotoClientModule::HandleCommand" << endl;

   TString cmd = command->PopCmd();
   if (cmd == "Set") {
      TString opt = command->PopOpt();
      if      (opt == "HostPort")  {
         string newHost  = command->PopOpt();
         Int_t  newPort  = command->PopIntOpt();
         if (newHost != fHost || newPort != fPort) {
            DestroyClient();  // close any existing client
            SetHostPort(newHost.c_str(),newPort);
         }
      }
      else if (opt == "BufferWords") {
         Int_t newSize = command->PopIntOpt();
         if (newSize != fBufferWords) {
            DestroyBuffer();
            fBufferWords = newSize;
         }
      }
      else if (opt == "Primary") {
         fPrimary = (command->PopIntOpt()) ? 1 : 0;
      }
      else if (opt == "GenFakeRec") {
         fGenFakeRec = (command->PopIntOpt()) ? 1 : 0;
      }
      else if (opt == "DebugFlags") {
         fDebugFlags = command->PopIntOpt();
      }
      else if (opt == "ClientDebugFlags") {
         Int_t flags = command->PopIntOpt();
         RotoClient::SetDebugFlags(flags);
      }
      else if (opt == "ObjectifierDebugFlags") {
         Int_t flags = command->PopIntOpt();
         RotoObjectifier::SetDebugFlags(flags);
      }
      else {
         MSG("Roto", Msg::kWarning)
            << "RotoClientModule: Unrecognized Set option " << opt << endl;
      }
   } else if (cmd == "Report") {
      Report();
   } else if (cmd == "CrashAndBurn") {
      MSG("Roto", Msg::kFatal)
            << "RotoClientModule: CrashAndBurn command " << endl;
      exit(0); // kill without core dumping
      // assert(0);
   } else {
      MSG("Roto", Msg::kWarning)
         << "RotoClientModule: Unrecognized command " << cmd << endl;
   }
   
}

//......................................................................
void RotoClientModule::Config(const Registry& r)
{
  // Configure the module using the registry r

  string newHost = fHost;
  int    newPort = fPort;
  int    newSize = fBufferWords;
  const char* tmpcs;
  int         tmpi;

  if (r.Get("Host:Port",tmpcs))  {
    char host[255];
    char* hostptr = host;
    // copy up to colon or end-of-string
    while (*tmpcs != ':' && *tmpcs != '\0') *hostptr++ = *tmpcs++;
    if (host != hostptr) {
      *hostptr = '\0';  // trailing end-of-string marker
      newHost = host;   // saw new host name (not just port), convert to string
    }
    if (*tmpcs == ':') {
      tmpcs++; // move beyond colon
      newPort = atoi(tmpcs);
    }
  }

  if (r.Get("BufferWords",tmpi)) newSize      = tmpi;
  if (r.Get("UseRototalk",tmpi)) fUseRototalk = tmpi;
  if (r.Get("Primary",tmpi))     fPrimary     = tmpi;
  if (r.Get("GenFakeRec",tmpi))  fGenFakeRec  = tmpi;
  if (r.Get("DebugFlags",tmpi))  fDebugFlags  = tmpi;
  if (r.Get("ClientDebugFlags",tmpi)) RotoClient::SetDebugFlags(tmpi);

  if ( newHost != fHost || newPort != fPort ) {
    DestroyClient();  // close any existing client
    SetHostPort(newHost.c_str(),newPort);
  }

  if (newSize != fBufferWords) {
    DestroyBuffer();
    fBufferWords = newSize;
  }

}
//......................................................................
const Registry& RotoClientModule::DefaultConfig() const
{
  // The default configuration for this module
  static Registry r;

  std::string name = this->GetName();
  name += ".config.default";
  r.SetName(name.c_str());

  r.UnLockValues();

  // server IP host name and port (or file name)
  // if port=0 use default, if <0 "host" is binary file name
  char dfltHostPort[255];
  sprintf(dfltHostPort,"%s:%d",dfltHost,dfltPort);
  r.Set("Host:Port",dfltHostPort);    

  r.Set("UseRototalk",true);   // use DAQ Rototalk vs. duplicate code
  r.Set("BufferWords",dfltBufferWords);  // size of client's buffer
  r.Set("Primary",true);      // client is data source (false not supported)
  r.Set("GenFakeRec",false);  // generate fake RunStart/RunEnd if missing
  //  r.Set("DebugFlags",0);     // various bits
  //  r.Set("ClientDebugFlags",0);     // various bits
  r.LockValues();

  return r;
}

//......................................................................
void RotoClientModule::Help()
{
   MSG("Roto", Msg::kInfo)
    << "Help for 'RotoClientModule':" << endl
    << "  RotoClientModule is a JobCModule for flattening a RawRecord" << endl
    << "  to look like one that would come from the DAQ and then to" << endl
    << "  write it out, either by sending the buffer to a Roto-rooter" << endl
    << "  or writing it to a binary file" << endl
    << endl
    << "Commands implemented:" << endl
    << endl
    << "  /RotoClientModule/Set HostPort <host> <port>" << endl
    << "     set the output host (string) and port (int)" << endl
    << "     if (port == -1) host is the name of the binary file" << endl
    << "     if (port ==  0) use default port #" << endl
    << endl
    << "  /RotoClientModule/Set BufferWords <nwords>" << endl
    << "     sets the buffer size (32 bit words) to use" << endl
    << endl
    << "  /RotoClientModule/Set Primary {0|1}" << endl
    << "     if not primary then only no writing is done" << endl
    << "     but Report can be requested from server" << endl
    << endl
    << "  /RotoClientModule/Set DebugFlags <bits>" << endl
    << "     sets the debug flags bits" << endl
    << endl;
}

//......................................................................
void RotoClientModule::SetHostPort(const Char_t* host, Int_t port)
{
   // Set Host and Port
   // if port == 0 use :<port> from *host, if none use dfltPort

   char* hostBuffer = new Char_t[strlen(host)+1];
   strcpy(hostBuffer,host);

   // pick off port if given
   char *colon = strchr(hostBuffer,':');
   int bport = 0;
   if (colon) {
      sscanf(colon,":%d",&bport);
      *colon = 0;  // truncate hostBuffer so there is no ':'
   }
   fHost = hostBuffer;
   if (fHost == "") fHost = dfltHost;
   fPort = port;
   if (!fPort) fPort = bport;
   if (!fPort) fPort = dfltPort;

   delete [] hostBuffer;

}

//......................................................................
RotoClient* RotoClientModule::GetClient()
{
   // return ptr to up-and-running client
   // create one if necessary
   // return 0 if unable to connect

   if (fRotoClient) return fRotoClient; // if a client is in place just move on

   if (fPort >= 0) {
      fRotoClient = new RotoClient(fHost.c_str(),fPort,
                                   MINOS_ROOTER_DCP,fUseRototalk);
   } else {
      fRotoClient = new RotoClientBinaryFile(fHost.c_str(),fPort);
   }
   if (fRotoClient->Connected()) return fRotoClient;
   // couldn't connect during construction
   delete fRotoClient;
   fRotoClient = 0;

   MSG("Roto", Msg::kWarning) 
      << "RotoClientModule::GetClient for host " << fHost.c_str()
      << " port " << fPort << " failed " << endl;
   return 0;

}

//......................................................................
void RotoClientModule::DestroyClient()
{
   // destroy any current client
   // client's dtor should do a clean disconnect
   // closing socket or file
   if (fRotoClient) { delete fRotoClient; fRotoClient=0; }
}

//......................................................................
Int_t* RotoClientModule::GetBuffer()
{
   // return ptr to buffer
   // create one if necessary

   if (fBuffer) return fBuffer; // if a buffer is in place just move on

   fBuffer = new Int_t[fBufferWords];
   if (fBuffer) return fBuffer;

   MSG("Roto", Msg::kWarning) 
      << "RotoClientModule::GetBuffer for size " << fBufferWords
      << " failed " << endl;
   return 0;

}

//......................................................................
void RotoClientModule::DestroyBuffer()
{
   // destroy any current buffer

   if (fBuffer) { delete [] fBuffer; fBuffer=0; }
}

//......................................................................
void RotoClientModule::ReOpenOutputFile(Int_t detector, Int_t run, Int_t subrun,
                                        VldContext* pvldc, Bool_t fakeRunRec)
                                                         
{
   // open an output file on the rotorooter if necessary
   // close the last one if there was one
   // this assumes a strictly sequential ordering

   static Int_t last_det = Detector::kUnknown;
   static Int_t last_run = -1;
   static Int_t last_subrun = -1;
   static VldContext last_vldc;
   static VldContext start_vldc;

   RotoClient* client = GetClient();

   if (!client) return;

   bool newrun =  (run != last_run || subrun != last_subrun);

   if (newrun) {
      // old file exists?
      if (last_run != -1 || last_subrun != -1) {

         if (fakeRunRec) {
            // pump over a fake RunEnd record

            // advance timestamp by 1 ns so it comes after last event
            VldTimeStamp end_timestamp(last_vldc.GetTimeStamp());
            end_timestamp.Add(VldTimeStamp((time_t)0,+1));
            VldContext end_vldc(last_vldc.GetDetector(),
                                last_vldc.GetSimFlag(),
                                end_timestamp);

            int runtype = 0;
            int timeframe = -1;
            RawDaqHeaderBlock* daqHdr =
               new RawDaqHeaderBlock(end_vldc,last_run,last_subrun,
                                     runtype,timeframe);
            RawDataBlock* runEnd = 
               new RawRunEndBlock(start_vldc,end_timestamp,
                                  last_run,last_subrun,runtype);
            SendFakeStartEndRecord(daqHdr,runEnd);
            delete daqHdr;
            delete runEnd;
         }
         // close the file
         client->CloseDAQFile(last_det,last_run,last_subrun);
      }
   }

   // update values 
   // do this no matter what so that vldc can timestamp
   // fake runend with the time of the last event
   last_det    = detector;
   last_run    = run;
   last_subrun = subrun;
   if (pvldc) last_vldc = *pvldc;

   if (newrun) {
      // new file sensible?
      if (last_run != -1 || last_subrun != -1) {
         // open file
         client->OpenDAQFile(last_det,last_run,last_subrun);

         if (fakeRunRec) {
            // pump over a fake RunStart record
            // backup timestamp by 1 ns so it comes before first event
            VldTimeStamp start_timestamp(last_vldc.GetTimeStamp());
            start_timestamp.Add(VldTimeStamp((time_t)0,-1));
            start_vldc = VldContext(last_vldc.GetDetector(),
                                    last_vldc.GetSimFlag(),
                                    start_timestamp);

            int runtype = 0;
            int timeframe = -1;
            RawDaqHeaderBlock* daqHdr =
               new RawDaqHeaderBlock(start_vldc,last_run,last_subrun,
                                     runtype,timeframe);
            RawDataBlock* runStart = 
               new RawRunStartBlock(start_vldc,last_run,last_subrun,runtype);
            SendFakeStartEndRecord(daqHdr,runStart);
            delete daqHdr;
            delete runStart;
         }
      }
   }

}
//......................................................................
void RotoClientModule::SendFakeStartEndRecord(RawDaqHeaderBlock* headerblk,
                                              RawDataBlock* startend)
                                                         
{
   // send these two blocks to the client
   // WITHOUT overwriting what's in the main buffer

   RotoClient* client = GetClient();

   if (!client) return;

   int nwords = headerblk->GetSize() + startend->GetSize();
   int nbytes = sizeof(Int_t) * nwords;

   // allocate a temporary buffer
   Int_t* buffer = new Int_t [nwords];
   Int_t* iptr   = buffer;

   // fill the buffer
   iptr = headerblk->AppendToBuffer(iptr);
   iptr = startend->AppendToBuffer(iptr);

   // send this "record"
   client->SendRecordBuffer(buffer,nbytes);
            

   // delete temporary buffer
   delete [] buffer;
}
//......................................................................

---