IoOutputModule.cxx

Go to the documentation of this file.

// $Id: IoOutputModule.cxx,v 1.26 2005/11/28 04:06:11 schubert Exp $
//
// messier@huhepl.harvard.edu
#include "IoModules/IoOutputModule.h"

#include <vector>
#include <algorithm>

#include "MessageService/MsgService.h"
#include "JobControl/JobCEnv.h"
#include "JobControl/JobCommand.h"
#include "JobControl/JobCModuleRegistry.h"
#include "Persistency/PerOutputStreamManager.h"
#include "Persistency/PerOutputStream.h"
#include "Registry/Registry.h"
#include "Util/UtilString.h"  // StringTok method

CVSID("$Id: IoOutputModule.cxx,v 1.26 2005/11/28 04:06:11 schubert Exp $");
JOBMODULE(IoOutputModule,"Output","Write data to output streams");

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

IoOutputModule::IoOutputModule() :
  fOutputStreamManager(new PerOutputStreamManager),
  fAutoSaveInt(0),
  fAutoSaveTime(0),
  fAutoSaveBytes(10000000), // default is to save tree every 10 Mbytes
  fFileName(""),
  fDefaultFileName(""),
  fAccessMode(Per::kNew),
  fOutputFileOpen(false)
{ 
  // Initialize fStreamList with default list of output streams 
  this->SetStreams("DaqSnarl,DaqMonitor,LightInjection,Cand,Config,SimSnarl");
}

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

IoOutputModule::~IoOutputModule() 
{
  // If we have output stream manager, delete it
  if (fOutputStreamManager) {
    delete fOutputStreamManager;
    fOutputStreamManager = 0;
  }
}

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

void IoOutputModule::BeginJob() 
{
  // Open and configure the default output streams.

  // Define and enable default list of output streams, but don't override
  // user definitions if specified
  if (!fOutputStreamManager->GetOpenedStream("DaqSnarl"))
    this -> DefineStream("DaqSnarl","RawRecord","","DaqSnarl");
  if (!fOutputStreamManager->GetOpenedStream("DaqMonitor"))
    this -> DefineStream("DaqMonitor","RawRecord","","DaqMonitor");
  if (!fOutputStreamManager->GetOpenedStream("LightInjection"))
    this -> DefineStream("LightInjection","RawRecord","","LightInjection");
  if (!fOutputStreamManager->GetOpenedStream("Cand"))
    this -> DefineStream("Cand","CandRecord","PrimaryCandidateRecord");
  if (!fOutputStreamManager->GetOpenedStream("SimSnarl"))
    this -> DefineStream("SimSnarl","SimSnarlRecord");
  if (!fOutputStreamManager->GetOpenedStream("Config"))
    this -> DefineStream("Config","ConfigRecord");
  if (!fOutputStreamManager->GetOpenedStream("NtpSR"))
    this -> DefineStream("NtpSR","NtpSRRecord");
  if (!fOutputStreamManager->GetOpenedStream("NtpMC"))
    this -> DefineStream("NtpMC","NtpMCRecord");
  if (!fOutputStreamManager->GetOpenedStream("NtpTH"))
    this -> DefineStream("NtpTH","NtpTHRecord");
  if (!fOutputStreamManager->GetOpenedStream("NtpSt"))
    this -> DefineStream("NtpSt","NtpStRecord");
  
  this -> EnableStreamList();
  
  const char* saveOpt = fOutputStreamManager->GetPrintOpt();
  fOutputStreamManager->SetPrintOpt("brief"); // must be a better way
  MSG("Io",Msg::kDebug) << "BeginJob. Status of output stream manager:\n"
                        << *fOutputStreamManager << endl;
  fOutputStreamManager->SetPrintOpt(saveOpt);
}

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

const Registry& IoOutputModule::DefaultConfig() const
{
//======================================================================
// Configure the module given the values held in registry r
//======================================================================
  static Registry r;
  std::string name = this->GetName();
  name += ".config.default";
  r.SetName(name.c_str());

  MSG("Io",Msg::kDebug) << "Loading default config\n";

  r.UnLockValues();
  r.Set("AutoSaveInt",    0);
  r.Set("AutoSaveTime",   0);
  r.Set("AutoSaveBytes", 10000000);
  r.Set("AccessMode",     "Recreate");
  r.Set("Streams",  "DaqSnarl,DaqMonitor,LightInjection,Cand,Config,SimSnarl");

  // Grab the default output file from the command line
  r.Set("FileName",       JobCEnv::Instance().GetDefaultOutputFileName());

  // Set a fall back if nothing is given on the command line
  r.Set("DefaultFileName","out.root");

  r.LockValues();

  return r;
}

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

void IoOutputModule::Config(const Registry& r)
{
//======================================================================
// Configure the module given the values held in registry r
//======================================================================
  int         tmpi;
  const char* tmps;

  MSG("Io",Msg::kDebug) << "Config IoOutputModule with r=" << r << "\n";

  bool doAutoSaveConfig = false;
  if (r.Get("AutoSaveInt",tmpi)) {
    fAutoSaveInt = tmpi;
    doAutoSaveConfig = true;
  }
  if (r.Get("AutoSaveTime",tmpi)) {
    fAutoSaveTime = tmpi;
    doAutoSaveConfig = true;
  }
  if (r.Get("AutoSaveBytes",tmpi)) {
    fAutoSaveBytes = tmpi;
    doAutoSaveConfig = true;
  }

  if ( doAutoSaveConfig && fOutputStreamManager ) fOutputStreamManager
          ->SetAutoSave("*",fAutoSaveInt,fAutoSaveTime,fAutoSaveBytes);

  bool doFileConfig = false;
  if (r.Get("FileName", tmps)) {
    fFileName = tmps;
    doFileConfig = true;
  }
  if (r.Get("AccessMode", tmps)) {
    tmpi = Per::GetAccessMode(tmps);
    if ( tmpi > 0 ) {
      fAccessMode = static_cast<Per::EAccessMode>(tmpi);
      doFileConfig = true;
    }
    else {
      MSG("Io",Msg::kWarning) << "Invalid file accessmode " << tmps
                              << " will be ignored. " << endl;
    }
  }
  if ( doFileConfig ) {
    // Close file because filename or accessmode has changed.  Wait until
    // IoOutputModule::Put call to reopen
    this -> CloseFile();
  }
  if ( r.Get("Streams",tmps) ) {
    this -> SetStreams(tmps);
  }
  if (r.Get("DefaultFileName", tmps)) {
    fDefaultFileName = tmps;
  }
}

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

void IoOutputModule::EndJob()
{
  // Close current output file
  this->CloseFile();

  // Close all streams
  fOutputStreamManager->CloseStream();
}

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

JobCResult IoOutputModule::Put(const MomNavigator* mom)
{ 
  if (fOutputFileOpen==false) this->OpenFile();
  fOutputStreamManager->Put(mom);
  return JobCResult::kAOK;
}

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

void IoOutputModule::HandleCommand(JobCommand* cmd) 
{
  if (!cmd->HaveCmd()) return;
  string s = cmd->PopCmd();
  if (s == "Open") {
    if (cmd->HaveOpt()) {
      fFileName = cmd->PopOpt(); // Set the file name
      // Determine the access mode
      string mode = "New";
      fAccessMode = Per::kNew;
      if (cmd->HaveOpt()) {
        mode = cmd->PopOpt();
        if      (mode == "New")      fAccessMode = Per::kRead;
        else if (mode == "Recreate") fAccessMode = Per::kRecreate;
        else if (mode == "Update")   fAccessMode = Per::kUpdate;
        else {
          MSG("Io",Msg::kError) << 
            mode.c_str() << " is not a valid output mode\n";
          return;
        }
      }
    }
    else {
      MSG("Io",Msg::kError) <<
        "Open requires at least a file name argument.\n";
    }
    return;
  }
  if (s == "Close") { 
    this->CloseFile(); 
    return; 
  }
  if (s == "AutoSave") {
    if (cmd->HaveOpt()) {
      fAutoSaveInt = cmd->PopIntOpt();
      return;
    }
    else {
      MSG("Io",Msg::kWarning) << "AutoSave requires one argument.\n";
      return;
    }
  }
  if (s == "AddStreams") {
    string streamlist;
    if (cmd->HaveOpt()) streamlist = cmd->PopOpt();
    this -> AddStreams(streamlist);
    return;
  }
  if (s == "DefineStream") {
    string streamname,classname,username,instreamnm;
    Int_t splitlevel=Per::kRecSplit;
    if (cmd->HaveOpt()) {
      streamname = cmd->PopOpt(); 
      if (cmd->HaveOpt()) {
        classname = cmd->PopOpt(); if (classname == "<null>") classname="";
        if (cmd->HaveOpt()) {
          username = cmd->PopOpt(); if (username == "<null>") username="";
          if ( cmd->HaveOpt() ) {
            instreamnm = cmd->PopOpt(); if (instreamnm=="<null>")instreamnm="";
            if ( cmd->HaveOpt() ) {
              splitlevel = cmd->PopIntOpt();
            }
          }
        }
        this -> DefineStream(streamname,classname,username,instreamnm,
                             splitlevel);
      }
    }
    if (streamname.empty() || classname.empty()) {
      MSG("Io",Msg::kError) <<
      "DefineStream requires at least streamname,recordclassname arguments.\n";
    }
    return;
  }

  if (s == "RemoveStreams") {
    string streamlist;
    if (cmd->HaveOpt()) streamlist = cmd->PopOpt();
    this -> RemoveStreams(streamlist);
    return;
  }
  if (s == "SetStreams") {
    string streamlist;
    if (cmd->HaveOpt()) streamlist = cmd->PopOpt();
    this -> SetStreams(streamlist);
    return;
  }
  if (s == "Set") {
    const char* optc = cmd->PopOpt();
    if (optc) {
      string opt(optc);
      if ( opt == "Streams") {
        string streamlist;
        if (cmd->HaveOpt()) streamlist = cmd->PopOpt();
        this -> SetStreams(streamlist);
        return;
      }
      MSG("Io",Msg::kWarning) << "Unknown option '" << opt << "'.\n";
    }
    else {
      MSG("Io",Msg::kWarning) << "Set requires additional arguments.\n";
    }
  }
  // Errors fall through to here
  MSG("Io",Msg::kError) << "Illegal command: " << s.c_str() << "\n";
}

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

bool IoOutputModule::OpenFile() 
{
  // If no file name has been given by this time, use the default as a
  // fall back
  if (fFileName=="") fFileName = fDefaultFileName;

  fOutputFileOpen = false;

  // set new file on all managed output streams
  if (fOutputStreamManager->SetFile("*",fFileName,fAccessMode)) {
    MSG("Io",Msg::kDebug) << "OpenFile " << fFileName 
      << " with accessmode " << Per::AsString(fAccessMode) << "." << endl;
    fOutputFileOpen = true;
  }
  return fOutputFileOpen;
}

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

bool IoOutputModule::CloseFile()
{
  MSG("Io",Msg::kDebug) << "CloseFile." << endl;
  fOutputStreamManager->Write("*",false);
  fOutputStreamManager->CloseFile();
  fOutputFileOpen = false;
  return true;
}

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

int IoOutputModule::AddStreams(std::string streamList)
{
  //
  // Purpose: Add output streams on the input streamList to the list of
  //          enabled streams.
  // 
  // Argument: streamlist  list of streams separated by delimiters
  //
  // Return: Number of streams remaining open.
  //
  // Contact: S.Kasahara
  // 
  // Notes: Streams must be defined via the DefineStream method.
  //


  // Parse input streamList into vector of requested output streams.
  std::vector<std::string> addList;
  UtilString::StringTok(addList,streamList,":,;/ ");
  
  vector<string>::iterator addItr,stItr;
  for ( addItr = addList.begin(); addItr != addList.end(); addItr++ ) {
    string addStreamName = *addItr;
    stItr = find(fStreamList.begin(),fStreamList.end(),addStreamName);
    if (stItr == fStreamList.end())fStreamList.push_back(addStreamName);
  }
  
  Int_t nenabled = this -> EnableStreamList();
  MSG("Io",Msg::kVerbose) << "AddStreams " << streamList << "." << endl;
 
  return nenabled;

}

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

int IoOutputModule::DefineStream(std::string streamName,std::string className, 
          std::string userName,std::string inputStreamName,Int_t splitLevel)
{
  //
  // Purpose: Define and open output stream.
  // 
  // Argument:   streamName   name of output stream
  //              className   class of records to persist
  //               userName   TNamed name of record (optional, default="")
  //        inputStreamName   origin of record  (optional, default="")
  //             splitLevel   splitLevel of tree (optional, def=Per::kRecSplit)
  //
  // Return: 1.
  //
  // Contact: S.Kasahara
  // 
  // Notes: If stream with name streamName has already been defined it will
  //        be overridden with new definition.  classname,username,
  //        and inputstreamname are the key used to extract the record(s) 
  //        from mom.
  //

  if ( fOutputStreamManager->GetOpenedStream(streamName) ) {
    // Close existing stream before proceeding
    fOutputStreamManager->CloseStream(streamName);
  }

  // Open output stream
  PerOutputStream* outStream = fOutputStreamManager ->
    OpenStream(streamName,streamName,className,userName,inputStreamName,
                                                        splitLevel);

  outStream->SetAutoSave(fAutoSaveInt,fAutoSaveTime,fAutoSaveBytes);

  return 1;

}

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

int IoOutputModule::EnableStreamList() 
{
  //
  // Purpose: Enable all streams on fStreamList.  
  // 
  // Argument: none.
  //
  // Return: Number of streams remaining open.
  //
  // Contact: S.Kasahara
  // 
  // Notes: Streams must have been previously defined via the 
  //        DefineStream method.
  //

  // Attach "Associated" streams. An example is the Config stream is
  // associated with the Cand stream.
  this -> AttachAssociatedStreams();

  // Begin by enabling all defined streams
  fOutputStreamManager->SetEnable();
  Int_t nenabled = fOutputStreamManager->GetNumStream();

  // Disable streams that are not on requested list
  const PerStreamManager::StreamMap& smap 
                            = fOutputStreamManager->GetStreamMap();
  for ( PerStreamManager::StreamMapConstItr citr  = smap.begin();
                                            citr != smap.end(); ++citr ) {
    string streamName = citr->first;
    typedef vector<string>::const_iterator vs_citer;
    vs_citer vpos = find(fStreamList.begin(),fStreamList.end(),streamName);
    if ( vpos == fStreamList.end() ) {
      fOutputStreamManager->SetEnable(streamName,false);
      nenabled--;
    }
  }

  return nenabled;
}

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

int IoOutputModule::RemoveStreams(std::string streamList)
{
  //
  // Purpose: Remove output streams on the streamList from the list of
  //          enabled streams.
  // 
  // Argument: streamlist  list of streams separated by delimiters
  //
  // Return: Number of streams remaining open.
  //
  // Contact: S.Kasahara
  // 
  // Notes: Streams must be defined via the DefineStream method.
  //


  // Parse input streamList into vector of requested output streams.
  std::vector<std::string> removeList;
  UtilString::StringTok(removeList,streamList,":,;/ ");
  
  vector<string>::iterator rmItr,stItr;
  for ( rmItr = removeList.begin(); rmItr != removeList.end(); rmItr++ ) {
    stItr = find(fStreamList.begin(),fStreamList.end(),*rmItr);
    if (stItr != fStreamList.end())fStreamList.erase(stItr);
  }

  Int_t nenabled = this -> EnableStreamList();
  MSG("Io",Msg::kVerbose) << "RemoveStreams " << streamList << "." << endl;

  return nenabled;

}

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

int IoOutputModule::SetStreams(string streamList)
{
  //
  // Purpose: Modify enabled output streams to match those on the streamlist
  // 
  // Argument: streamlist  list of streams separated by delimiters
  //
  // Return: Number of streams remaining open.
  //
  // Contact: S.Kasahara
  // 
  // Notes: Streams must be defined via the DefineStream method.
  //


  MSG("Io",Msg::kVerbose) << "SetStreams " << streamList << "." << endl;

  // Parse input streamList into vector of requested output streams.
  fStreamList.clear();
  UtilString::StringTok(fStreamList,streamList,":,;/ ");

  Int_t nenabled = this -> EnableStreamList();
 
  return nenabled;

}

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

int IoOutputModule::AttachAssociatedStreams() {
  //
  // Purpose: Private method to attach "Associated" streams to output
  //          stream list
  //
  // Return: Number of attached associated streams.
  // 

  int nassoc = 0;

  // Since associated streams may in turn have associated streams,
  // this is done as a while loop until no more streams are added
  bool isNewStream = true;
  while (isNewStream) {
    isNewStream = false;
    std::vector<std::string>::const_iterator s_citer;
    for (s_citer = fStreamList.begin();s_citer != fStreamList.end();s_citer++){
      std::string streamname = *s_citer;
      std::string assocstreams 
                          = Per::GetAssociatedStreamList(streamname.c_str());
      if ( !assocstreams.empty() ) {
        // parse delimiter separated streamlist into a vector of streamnames
        std::vector<std::string> assocstreamlist;
        UtilString::StringTok(assocstreamlist,
                              std::string(assocstreams),":,;/ ");
        std::vector<std::string>::const_iterator a_citer;
        // Add associated streams to fStreamList that aren't already present
        for ( a_citer = assocstreamlist.begin(); 
              a_citer != assocstreamlist.end(); a_citer++ ) {
          std::string assocstreamname = *a_citer;
          if ( find(fStreamList.begin(),fStreamList.end(),assocstreamname)
               == fStreamList.end() ) {
            MSG("Io",Msg::kVerbose) << "Attach associated stream " 
                                    << assocstreamname.c_str() << "." << endl;
            fStreamList.push_back(assocstreamname);
            nassoc++;
            isNewStream = true;
          }
        }
      }
    }
  }
  
  return nassoc;

}

---