PerOutputStreamManager.cxx

Go to the documentation of this file.

//
// PerOutputStreamManager
//
// Package: Per (Persistency).
//
// S. Kasahara 04/2001
//
// Purpose: Output stream manager class used to manage PerOutputStreams.
//

#include <string>

#include "TClass.h"

#include "Persistency/PerOutputStream.h"
#include "Persistency/PerOutputStreamManager.h"
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "Record/RecMinos.h"
#include "Record/RecRecord.h"

std::ostream& operator<<(std::ostream& ms, 
                         const PerOutputStreamManager& posm) {
  return posm.Print(ms);
}

ClassImp(PerOutputStreamManager)

//   Definition of static data members
//   *********************************

CVSID("$Id: PerOutputStreamManager.cxx,v 1.27 2005/05/01 21:53:35 minoscvs Exp $");

// Definition of methods (alphabetical order)
// ******************************************

UInt_t PerOutputStreamManager::GetRecordTimeSec(TObject* object) {
  // Private method to extract record time from time stamp if possible
  // else returns 0

  UInt_t recordTime = 0;

  if ( !object ) return recordTime;

  if ( RecMinos* record = dynamic_cast<RecMinos*>(object) ) {
    recordTime = (record -> GetVldContext() -> GetTimeStamp()).GetSec();
  }
  else if ( RecRecord* record = dynamic_cast<RecRecord*>(object) ) {
    recordTime = (record->GetHeader()).GetVldContext().GetTimeStamp().GetSec();
  }

  return recordTime;
  
}
  
PerOutputStream* PerOutputStreamManager::OpenStream(string streamname, 
                                                    string treename, 
                                                    string classname, 
                                                    string username,
                                                    string inputstreamname, 
                                                    Int_t  splitlevel, 
                                                    Int_t  basketsize, 
                                                    Int_t  compress) {
  //
  //  Purpose:  Open output stream with name streamname.
  //
  //  Arguments:streamname     name of stream to be opened. Names of managed
  //                           streams must be unique.
  //            treename       name of tree to be served by this stream.
  //            classname      classname of objects to be stored in this stream
  //            username       username of objects to be stored in this stream
  //                           used as secondary id to classname (default="")
  //       inputstreamname     input strm name of origin of objs to be stored
  //                           in this stream used as tertiary id (default="")
  //            splitlevel     splitlevel of created ROOT TTree branch 
  //                           (default = 0).
  //            basketsize     basketsize with which main branch on stream is
  //                           opened (default = 64000).
  //            compress       compression level with which tree will be 
  //                           written to file. Default = -1 => set to  
  //                           default compression level of file (ROOT
  //                           sets default file compression = 1).  Valid
  //                           range is 0 -> 9 with 0 == no compression.  
  //
  //  Return:  pointer to PerOutputStream. If unable to open stream (because
  //           stream with this streamname has already been opened), returns
  //           (PerOutputStream*)0.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: PerOutputStream objects are owned by the PerOutputStreamManager 
  //         and should only be deleted through the 
  //         PerStreamManager::CloseStream method.

  bool openok = false;

  PerOutputStream* stream = (PerOutputStream*)fStreamMap[streamname];
  if (!stream) {
    // Stream not found in map, need to create new one
    stream = new PerOutputStream(treename,classname,username,inputstreamname,
                                               splitlevel,basketsize,compress);
    stream -> SetStreamName(streamname);
    fStreamMap[streamname] = stream;
    openok = true;
  }
  else {
    MSG("Per",Msg::kWarning)<<"Stream manager failed to open requested stream "
      << streamname << " because name conflicts with previously opened stream."
      << endl;
  }

  return (openok) ? stream : (PerOutputStream*)0;

}


std::ostream& PerOutputStreamManager::Print(std::ostream& ms,
                                            const char* option) const {
  //
  //  Purpose:  Print status of output stream manager on std::ostream.
  //
  //  Arguments: ms          std::ostream to display on.
  //             option      verbosity level ("" (default),or "brief")
  //  Return:  std::ostream reference.
  //
  //  Contact:   S. Kasahara
  //

  TString opt = option;
  opt.ToLower();
  if ( opt == "brief" || opt.IsNull() && fPrintOpt == "brief") {
    ms << "PerOutputStreamManager managing enabled streams: " << endl;
    Int_t nenabled = 0;
    for(StreamMapConstItr citr = fStreamMap.begin();
                          citr!= fStreamMap.end(); ++citr) {
      PerOutputStream* outstream =dynamic_cast<PerOutputStream*>(citr->second);
      if ( outstream -> IsEnabled() ) {
        ms << nenabled << ")" << citr->first 
           << " persisting records of class " 
           << outstream->GetClassName();
        string username = outstream->GetUserName();
        if ( !username.empty() ) ms << ", username " << username;
        string inputstreamname = outstream->GetInputStreamName();
        if ( !inputstreamname.empty() ) 
          ms << ", originating from input stream " << inputstreamname;
        ms << " with splitlevel " << outstream->GetSplitLevel() 
           << " and AutoSave(Int,Time,Bytes) (" << outstream->GetAutoSaveInt()
           << "," << outstream->GetAutoSaveTime() << "," 
           << outstream->GetAutoSaveBytes() << ")" << endl;
        nenabled++;
      }
    }
  }
  else {
    ms << "PerOutput";
    PerStreamManager::Print(ms);
  }

  return ms;

}


Int_t PerOutputStreamManager::Put(const MomNavigator* mom) {
  //  Purpose:  Extract objects from mom and fill them into stream tree 
  //            designated to persist that object.
  //
  //  Argument: mom  pointer to MomNavigator
  //
  //  Return:  number of objects actually persisted.
  //
  //  Contact:   S. Kasahara
  //

  PerOutputStream* outstream;

  Int_t nobject = 0;

  // Loop over output streams managed by this stream manager
  UInt_t maxcurrentTime = 0;
  for ( StreamMapConstItr citr = fStreamMap.begin(); 
                          citr != fStreamMap.end(); ++citr ) {
    outstream = (PerOutputStream*)citr -> second;
    if (outstream -> IsOpen() && outstream -> IsEnabled()) {
    // retrieve classname & username of objects served by this stream
      string streamusername = outstream -> GetUserName(); 
      string streaminputstream = outstream -> GetInputStreamName(); 

      // Loop over objects stored in Mom.
      TIter fiter = const_cast<MomNavigator*>(mom) -> FragmentIter();
      TObject* object = 0;
      while ( (object = fiter.Next()) ) {
        // SetObject will test to see if this object is of the right type
        // for the output stream
        if ( outstream -> SetObject(object) ) {
          outstream -> Store();
          outstream -> Reset();
          MsgStream& msgPer = MSGSTREAM("Per",Msg::kVerbose);
          msgPer << "Persisted object: class " << object->IsA()->GetName();
          if ( !streamusername.empty() ) 
            msgPer << ", username " << object->GetName();
          if ( !streaminputstream.empty() )  
            msgPer << ", origin " << streaminputstream;
          msgPer << " on output stream " << citr->first << "." << endl;
          UInt_t currentTime = GetRecordTimeSec(object);
          if ( currentTime > maxcurrentTime ) maxcurrentTime = currentTime;
          nobject++;
        }
      }
    }
  }

  if ( maxcurrentTime > 0 ) {
    for ( StreamMapConstItr citr = fStreamMap.begin();
          citr != fStreamMap.end(); ++citr ) {
      outstream = (PerOutputStream*)citr -> second;
      outstream -> AutoSaveByTime(maxcurrentTime);
    }
  }

  MSG("Per",Msg::kVerbose) << "Put persisted " 
       << nobject << " total object(s). " << endl; 
  return nobject;

}

Int_t PerOutputStreamManager::SetAutoSave(std::string streamname, 
      UInt_t autosaveint, UInt_t autosavetime, UInt_t autosavebytes,
      bool basketsavelinked) {
  //  Purpose:  Set autosave interval(s) of specified stream(s).
  //
  //  Argument: streamname  std::string  name of stream.
  //                                if streamname="*" (default), change will 
  //                                affect all streams.
  //           autoSaveInt   UInt_t  TTree save entry interval     (default=0 
  //                              => no AutoSave requested at entry intervals) 
  //           autoSaveTime  UInt_t  TTree save time interval(sec) (default=0 
  //                              => no AutoSave requested at time intervals) 
  //           autoSaveBytes UInt_t  TTree save byte interval      (default=0 
  //                              => no AutoSave requested at byte intervals) 
  //        basketSaveLinked bool    tie basket dumps to TTree saves so that
  //                                 all branch baskets are dumped just before
  //                                 the tree header is saved to disk. 
  //                                 (default = true)
  //
  //  Return:  number of streams affected.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Invokes PerOutputStream::SetAutoSave for each requested 
  //          stream.
  //

  Int_t nstream = 0;

  if (streamname == "*") {
    for (StreamMapConstItr citr = fStreamMap.begin();
         citr != fStreamMap.end(); ++citr) {
      ((PerOutputStream*)(citr -> second)) 
       -> SetAutoSave(autosaveint,autosavetime,autosavebytes,basketsavelinked);
      nstream++;
    }
  }
  else {
    PerOutputStream* stream = (PerOutputStream*)GetOpenedStream(streamname);
    if (stream) {
      stream->SetAutoSave(autosaveint,autosavetime,autosavebytes,
                          basketsavelinked);
      nstream++;
    }
  }

  return nstream;
  
}

Int_t PerOutputStreamManager::Write(string streamname, bool force) {
  //  Purpose:  Write tree of specified stream(s) to file.
  //
  //  Argument: streamname  string  name of stream for which to write tree.
  //                                if streamname="*" (default), all 
  //                                streams will have their tree written.
  //            force       bool    true => tree is written to file even
  //                                if empty (default).  
  //
  //  Return:  number of trees written to file.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Invokes PerOutputStream::Write for each requested stream.
  //          This method would normally be invoked only once just before
  //          CloseFile.  

  Int_t ntree = 0;

  if (streamname == "*") {
    // Write tree on all streams
    for (StreamMapConstItr citr = fStreamMap.begin();
         citr != fStreamMap.end(); ++citr) {
      if ( force || citr -> second -> GetNumEntries() != 0 ) { 
        Int_t nbytes = ((PerOutputStream*)(citr -> second)) -> Write();
        if ( nbytes > 0 ) ntree++;
      }
    }
  }
  else {
    // Write tree on specified stream
    PerOutputStream* stream = (PerOutputStream*)GetOpenedStream(streamname);
    if (stream) {
      if ( force || stream -> GetNumEntries() != 0 ) {
        Int_t nbytes = stream -> Write();
        if ( nbytes > 0 ) ntree++;
      }
    }
  }

  return ntree;
  
}

---