PerOutputStream.cxx

Go to the documentation of this file.

//
// PerOutputStream
//
// Package: Per (Persistency).
//
// S. Kasahara 04/2001
//
// Purpose: Class for managing an individual output data stream.
//          The user should use the PerOutputStreamManager to open and
//          close PerOutputStreams.
//
#include "TROOT.h"
#include "TFile.h"
#include "TTree.h"
#include "TClass.h"
#include <iostream>
using std::endl;

#include "Persistency/PerOutputStream.h"
#include "Persistency/PerFile.h"
#include "Persistency/PerFileManager.h"
#include "MessageService/MsgService.h"
#include "Record/RecMinos.h"
#include "Record/RecRecord.h"
#include "Validity/VldContext.h"

std::ostream& operator<<(std::ostream& os, const PerOutputStream& pos) {
  return pos.Print(os);
}

ClassImp(PerOutputStream)

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

CVSID("$Id: PerOutputStream.cxx,v 1.41 2007/01/15 20:57:28 rhatcher Exp $");


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

void PerOutputStream::AutoSaveByTime(UInt_t currentVldTime) {
  //
  // Purpose: AutoSave stream if new entries on this tree and currentvldtime
  //          passed from manager satisfies autosavetime requirement. 
  //


  if ( !IsOpen() || fEntry == -1 ) return;

  TDirectory* savedir = gDirectory; // save current working directory

  if ( IsFileChange() ) ChangeFile();

  bool isAutoSave = false;
  
  if (!fLastTimeSaved) fLastTimeSaved = currentVldTime;
  // Save tree to file if requested on this entry and/or time interval
  bool isAutoSaveTime
     = (fAutoSaveTime > 0 && (currentVldTime - fLastTimeSaved) >= fAutoSaveTime);
  if ( isAutoSaveTime && fLastEntrySaved != fEntry ) isAutoSave = true;
  
  if ( isAutoSave && fBasketSaveLinked ) {
    // Reset basket size so baskets will be dumped on next TTree::Fill
    // if more than 1 kbyte of data in that buffer
    fTBranch -> SetBasketSize(1000);
  } else {
    fTBranch -> SetBasketSize(fBasketSize); // reset to default basketsize
  }

  if ( isAutoSave ) {
    Int_t savecompress = 9;
    if ( fCompress >= 0 ) {
      savecompress = fTFile -> GetCompressionLevel();
      fTFile -> SetCompressionLevel(fCompress);
    }

    // Save tree to file if requested on this entry interval
    MSG("Per",Msg::kDebug) << "AutoSaving tree " << fTreeName << " with " 
       << fTTree->GetEntries() << " entries to file " 
       << fFullFilePathName << " ByTime(sec) " 
       << currentVldTime << "." << endl;  
    fTFile -> cd();
    fTTree -> AutoSave();  // save tree to file
    fTFile -> SaveSelf();  // save file directory containing this tree
    fLastTimeSaved = currentVldTime;
    fLastEntrySaved = fEntry;
    fLastBytesSaved = fTotalBytes;
    if ( fCompress >= 0 )fTFile -> SetCompressionLevel(savecompress);
  }
    
  savedir -> cd(); // return to original working directory

  return;

}
  
void PerOutputStream::ChangeFile() {
  //
  //  Purpose:  This private method is called internally when it is detected
  //            that TTree::Fill has closed the current file and opened
  //            a new one with extension _n because it has detected that
  //            the file limit of 2 GByte is near.
  //
  //  Arguments: none.
  //
  //  Return: none.
  //
  //  Contact:   S. Kasahara
  //

                         
  TFile* newFile = fTTree->GetCurrentFile();
  if ( !newFile ) {
    MSG("Per",Msg::kWarning) 
      << "ChangeFile requested for tree " << fTreeName << 
      " from old file " << fFullFilePathName << " but TTree has null file ptr."
      << endl;
    return;
  }
  std::string newFileName = newFile -> GetName();
  MSG("Per",Msg::kInfo) << "PerOutputStream::ChangeFile requested for tree " 
    << fTreeName << "\nfrom old file " << fFullFilePathName 
    << " total entries " << fEntry+1 
    << ", to new file " << newFileName << "." << endl; 
  PerFileManager& perfilemanager = PerFileManager::Instance();
  perfilemanager.ChangeFile(fFullFilePathName,newFileName,newFile);

  fFullFilePathName = newFileName;
  fTFile = newFile;
     
  // All tree parameters get reset upon start of new file
  fEntry = static_cast<Int_t>(fTTree->GetEntries()-1);
  // All counts start fresh in new file
  fLastEntrySaved = fEntry;
  fLastTimeSaved = 0; // will be set to 1st record time
  fLastBytesSaved = 0; fTotalBytes = 0;

  return;

}

PerOutputStream::PerOutputStream(string treename, string classname, 
  string username, string inputstreamname, Int_t splitlevel, Int_t basketsize, 
  Int_t compress) :
  PerStream(treename),fUserName(username),fInputStreamName(inputstreamname),
  fSplitLevel(splitlevel),fBasketSize(basketsize),fCompress(compress),
  fAutoSaveInt(0),fAutoSaveTime(0),fAutoSaveBytes(0),fLastEntrySaved(-1),
  fLastTimeSaved(0),fLastBytesSaved(0),fTotalBytes(0),fBasketSaveLinked(false){
  //
  //  Purpose:  Default constructor.
  //
  //  Arguments: 
  //      treename   string name of tree served by this output stream.
  //     classname   string classname of object to be stored in this stream -
  //                        used to retrieve objects from MomNavigator.
  //      username   string username of object to be stored in this stream -
  //                        secondary identifier used to retrieve object from 
  //                        MomNavigator.
  // inputstreamname string input streamname of origin of object to be stored in this
  //                        stream - tertiary identifier used to retrieve object from
  //                        MomNavigator.
  //    splitlevel   Int_t  splitlevel of created ROOT TTree branch (default = 0)
  //    basketsize   Int_t  basketsize with which main branch is opened
  //                        (default = 64000).
  //      compress   Int_t  compression level of tree. If -1 (default),
  //                        tree compression level is set to that of
  //                        file (default file compression = 1).  ROOT defines 
  //                        compression levels 0-9, with level 0 == no 
  //                        compression.
  //
  //  Return:    n/a.
  //
  //  Contact:   S. Kasahara
  // 
  //  Notes: This constructor doesn't do anything except fill in data members.
  //         PerOutputStream does not become active until PerOutputStream::
  //         SetFile is invoked.
  // 

  fClassName = classname;

}

PerOutputStream::~PerOutputStream() {
  //
  //  Purpose:  PerOutputStream destructor.  
  //
  //  Contact:   S. Kasahara
  // 

}

std::ostream& PerOutputStream::Print(std::ostream& ms) const {
  //
  //  Purpose:  Print status of stream on std::ostream.
  //
  //  Arguments: ms std::ostream to display on.
  //
  //  Return:  std::ostream reference.
  //
  //  Contact:   S. Kasahara
  //

  ms << "  PerOutput"; 
  PerStream::Print(ms);

  if (!fUserName.empty()) ms << "  This stream serves objects of username " 
                             << fUserName << "." << endl;
 
  ms << "  TTree SplitLevel = " << fSplitLevel << endl;
 
  if ( fAutoSaveInt && fAutoSaveTime ) {
    ms << "  AutoSaves were requested in intervals of " << fAutoSaveInt
       << " entries or " << fAutoSaveTime 
       << " seconds,\n  whichever comes first." << endl;
  }
  else if (fAutoSaveInt) {
    ms << "  AutoSaves were requested in intervals of " << fAutoSaveInt 
       << " entries." << endl;
  }
  else if (fAutoSaveTime) {
    ms << "  AutoSaves were requested in time intervals of " << fAutoSaveTime
       << " seconds." << endl;
  }
  else if (fAutoSaveBytes) {
    ms << "  AutoSaves were requested in byte intervals of " << fAutoSaveBytes
       << " bytes." << endl;
  }
  else {
    ms << "  AutoSaves were not requested." << endl;
  }

  if (fBasketSaveLinked) {
    ms << "  Branch basket dumps are requested to occur at the same\n"
       << "  interval as the tree AutoSaves or when the branch basket size\n  "
       << fBasketSize << " is reached, whichever comes first." << endl;
  }
  else {
    ms << "  Branch basket dumps were not requested at regular intervals.\n" 
       << "  Basket dumps will occur when the basket size "
       << fBasketSize << " is reached." << endl;
  }
  if (fCompress >= 0) {
    ms << "  Compression level " << fCompress 
       << " was requested for this stream." << endl;
  }

  return ms;

}

void PerOutputStream::SetAutoSave(UInt_t autoSaveInt, UInt_t autoSaveTime, 
                                UInt_t autoSaveBytes, bool basketSaveLinked) {
  //
  //  Purpose:  Set AutoSave interval for this stream.  The TTree will
  //            be saved to file every fAutoSaveInt entries OR 
  //            OR every fAutoSaveTime (sec) OR every fAutoSaveBytes
  //            bytes filled to the tree, whichever comes first.
  //            Any or all of the autosave intervals may be set. An autosave
  //            parameter set to zero will be ignored.
  //
  //  Arguments: 
  //           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:  none
  //
  //  Contact:   S. Kasahara
  //

  fAutoSaveInt   = autoSaveInt;
  fAutoSaveTime  = autoSaveTime;
  fAutoSaveBytes = autoSaveBytes;
  fBasketSaveLinked = basketSaveLinked;
  
  return;

}

void PerOutputStream::Close() {
  //
  //  Purpose:  Close the output stream.
  //
  //  Arguments: none.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  // 

  CloseFile();

}

void PerOutputStream::CloseFile() {
  //
  //  Purpose:  Close the file attached to the output stream.
  //
  //  Arguments: none.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  // 

  if ( IsFileChange() ) ChangeFile();
  fLastEntrySaved = -1;
  fLastTimeSaved = 0;
  fLastBytesSaved = 0;
  fTotalBytes = 0;
  PerStream::CloseFile();

}

bool PerOutputStream::SetFile(string fullfilepathname, 
                          Per::EAccessMode accessmode) {
  //
  //  Purpose:  Set new output file for this stream.
  //
  //  Arguments: fullfilepathname string      new filename containing stream.
  //             accessmode Per::EAccessMode  accessmode in which to open file.
  //                                          (must be Per::kNew or
  //                                           Per::kRecreate (default)).
  //
  //  Return:  true if file was opened successfully and stream tree of
  //           name fTreeName was set to that file, else false.  See Note.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  This method enforces the uniqueness of trees of a given 
  //          treename in the output file.  This is done to avoid overwriting
  //          trees in the same file.  If treename is not unique, the new file
  //          is not set for this stream, and false is returned.
  //          If SetFile fails, the method PerStream::GetErrorCode can be
  //          used to determine reason for failure.

  if (!IsEnabled()) return false;

  // Save current working directory
  TDirectory* savedir = gDirectory;

  // Check to see if file is already set, if so close it before proceeding
  if ( fTFile ) {
    CloseFile();
  }

  fErrorCode = Per::kErrSuccess;

  if (accessmode != Per::kNew && accessmode != Per::kRecreate &&
      accessmode != Per::kUpdate) {
    MSG("Per",Msg::kWarning) 
      << "PerOutputStream::SetFile called w/ invalid accessmode " 
      << Per::AsString(accessmode) << endl;
    fErrorCode = Per::kErrInvalidAccessMode;
    savedir -> cd();
    return false;
  }

  // Use file manager singleton to open file in requested mode.
  PerFileManager& perfilemanager = PerFileManager::Instance();
  const PerFile* file = perfilemanager.OpenFile(fullfilepathname,accessmode);
  if (!file) {
    // file open failed
    fErrorCode = perfilemanager.GetErrorCode();
    savedir -> cd();
    return false;
  }
  fFullFilePathName = fullfilepathname;
  fTFile = file -> GetTFile();
  // Check to see if a previous existing tree of fTreeName exists in this file
  TTree* tree = dynamic_cast<TTree*> (fTFile -> Get(fTreeName.c_str()));
  if (tree) {
    // tree of duplicate treename already exists in file
    MSG("Per",Msg::kWarning) 
      << "PerOutputStream::SetFile failed because requested tree "
      << fTreeName << " already exists in file " << fullfilepathname << endl;
    fErrorCode = Per::kErrTreeExists;
    delete tree; 
    CloseFile();
    savedir -> cd();
    return false;
  }
     
  fTFile -> cd(); // make current file the working directory

  // This funniness is required because otherwise must manually
  // set compression level of all subbranches when tree is split
  Int_t savecompress = 9;
  if (fCompress >= 0) {
    savecompress = fTFile -> GetCompressionLevel();
    fTFile -> SetCompressionLevel(fCompress);
  }

  // Create new tree with requested treename, may want to specify job
  // related title for this tree
  fTTree = new TTree(fTreeName.c_str(),"Persistency stream tree");
  
  fTObject = 0;
  TClass* tclass = gROOT -> GetClass(fClassName.c_str());
  if ( tclass ) fTObject = (TObject*)tclass->New();
  
  if ( ! tclass ) {
    MSG("Per",Msg::kError) 
      << "TClass not found for requested output stream of type " 
      << fClassName.c_str() << "\nor class does not derive from TObject."
      << endl;
    CloseFile();
    savedir -> cd();
    return false;
  }
  
  fTBranch = fTTree -> Branch(fClassName.c_str(),fClassName.c_str(),
                               &fTObject,fBasketSize,fSplitLevel);
  if ( fTObject ) delete fTObject; fTObject = 0;
  
  if (fCompress >= 0) fTFile -> SetCompressionLevel(savecompress);

  // Autosave immediately to avoid the situation in which the open file
  // has basket dumps but no keys which would produce an error during the
  // read of the open file.
  fTTree -> AutoSave();  // save tree to file
  fTFile -> SaveSelf();  // save file directory containing this tree
  
  // Effectively turn ROOT autosave off.  AutoSave is controlled instead
  // through PerOutputStream AutoSavexxx parameters
  Int_t rootautosize = 2000000000;
  fTTree->SetAutoSave(rootautosize);

  // Switch back to original working directory
  savedir -> cd();

  return true;

}

bool PerOutputStream::SetObject(TObject* object) {
  //
  //  Purpose:  Set object to be persisted to ROOT TTree main branch.
  //            The object is not actually filled into the branch until the
  //            PerOutputStream::Store() method is invoked.
  //
  //  Arguments: object:  pointer to object to be stored.  Must be of same
  //                      type as "ClassName" with which PerOutputStream
  //                      was created.  
  //
  //  Return:  true if stream is successfully updated.
  //
  //  Contact:   S. Kasahara
  // 

  if ( !IsEnabled() ) return false;

  if( IsModified() ) {
    // Reset object cache managed by this stream
    Reset();
  }

  if ( !IsOpen() ) {
    MSG("Per",Msg::kWarning) 
      << "Attempting to SetObject in an unopened output stream." << endl;
    return false;
  }

  if ( !object ) {
    MSG("Per",Msg::kWarning)
      << "PerOutputStream called with null object pointer." << endl;
    return false;
  }

  if (strcmp(fClassName.c_str(),object->IsA()->GetName())) {
    // object is of wrong classname for this stream
    return false;
  }

  if (!fUserName.empty() && strcmp(fUserName.c_str(),object->GetName())) {
    // object is of wrong username for this stream
    return false;
  }

  Registry* temptags = 0;
  bool isPersisted = false;
  bool isInputStream = true;
  if ( !fInputStreamName.empty() ) isInputStream = false;
  if ( RecMinos* record = dynamic_cast<RecMinos*>(object) ) {
    temptags = &(record->GetTempTags());
    if ( !isInputStream && temptags ) {
      const char* objectInputStream = 0;
      if ( temptags->Get("stream",objectInputStream) ) {
        if ( !strcmp(fInputStreamName.c_str(),objectInputStream) )
                                              isInputStream = true;
      }
    }
    // test if object is of wrong input stream name for this stream
    if ( !isInputStream ) return false;
    // test to see if object has already been persisted to this stream
    isPersisted = record->IsPersistedToOutputStream(
             this->GetStreamName().c_str(),this->GetFullFilePathName().c_str(),
             this->GetTreeName().c_str());
  }
  else if ( RecRecord* record = dynamic_cast<RecRecord*>(object) ) {
    temptags = &(record->GetTempTags());
    if ( !isInputStream && temptags ) {
      const char* objectInputStream = 0;
      if ( temptags->Get("stream",objectInputStream) ) {
        if ( !strcmp(fInputStreamName.c_str(),objectInputStream) )
                                              isInputStream=true;
      }
    }
    // test if object is of wrong input stream name for this stream
    if ( !isInputStream ) return false;
    // test if object has already been persisted to this stream
    isPersisted = record->IsPersistedToOutputStream(
             this->GetStreamName().c_str(),this->GetFullFilePathName().c_str(),
             this->GetTreeName().c_str());
  }

  if ( !isInputStream || isPersisted ) return false; 

  fTObject = object;
  return true; 

}

Int_t PerOutputStream::Store() {
  //
  //  Purpose:  Fill data in stream tree.
  //
  //  Arguments: none.
  //
  //  Return:  number of bytes filled.
  //
  //  Contact:   S. Kasahara
  // 
  //  Notes:  This method fills the data into the stream tree. If fAutoSaveInt
  //          or fAutoSaveTime interval has been set, it will also
  //          invoke the proper methods to save the baskets and tree to file
  //          at correct entry and/or time interval.
  //

  Int_t nBytes = 0;

  TDirectory* savedir = gDirectory; // save current working directory

  if ( !IsEnabled() || !IsModified() ) {
    // disabled streams or streams in which an object has not been loaded
    // are not persisted
    return nBytes;
  }

  if ( !IsOpen() ) {
    MSG("Per",Msg::kWarning) 
      << "Attempting to store data on unopened output stream." << endl;
    return nBytes;
  } 

  if ( IsFileChange() ) ChangeFile();

  fEntry++;  // increment current entry number for this stream

  // Determine record validity time in secs
  UInt_t recordTime = 0;
  if ( RecMinos* record = dynamic_cast<RecMinos*>(fTObject) ) {
    recordTime = (record -> GetVldContext() -> GetTimeStamp()).GetSec();
    record->PersistedToOutputStream(this->GetStreamName().c_str(),
     this->GetFullFilePathName().c_str(),this->GetTreeName().c_str(),fEntry);
  }
  else if ( RecRecord* record = dynamic_cast<RecRecord*>(fTObject) ) {
    recordTime = (record->GetHeader()).GetVldContext().GetTimeStamp().GetSec();
    record->PersistedToOutputStream(this->GetStreamName().c_str(),
       this->GetFullFilePathName().c_str(),this->GetTreeName().c_str(),fEntry);
  }

  if ( !recordTime && fAutoSaveTime ) {
    // Records are required to derive from RecMinos class to use autosave
    // by time interval.
    MSG("Per",Msg::kWarning) << "Using AutoSaveTime requires records\n"
    << "to derive from RecMinos or RecRecord class.  Option will be ignored." 
    << endl;
    fAutoSaveTime = 0;
  }

  bool isAutoSave = false;
  if (!fLastTimeSaved) fLastTimeSaved = recordTime;
  // Save tree to file if requested on this entry and/or time interval
  bool isAutoSaveInt 
     = (fAutoSaveInt > 0 && (fEntry - fLastEntrySaved) >= (int)fAutoSaveInt);
  bool isAutoSaveTime
     = (fAutoSaveTime > 0 && (recordTime - fLastTimeSaved) >= fAutoSaveTime);
  bool isAutoSaveBytes
     = (fAutoSaveBytes > 0 && (fTotalBytes - fLastBytesSaved)>=fAutoSaveBytes);
  
  if ( isAutoSaveInt || isAutoSaveTime || isAutoSaveBytes ) {
     isAutoSave = true;
  }
 
  if ( isAutoSave && fBasketSaveLinked ) {
    // Reset basket size so baskets will be dumped on next TTree::Fill
    // if more than 1 kbyte of data in that buffer
    fTBranch -> SetBasketSize(1000);
  } else {
    fTBranch -> SetBasketSize(fBasketSize); // reset to default basketsize
  }

  // Must do this so that if AutoSave is invoked it knows the compression
  // level of the tree
  Int_t savecompress = 9;
  if ( fCompress >= 0 ) {
    savecompress = fTFile -> GetCompressionLevel();
    fTFile -> SetCompressionLevel(fCompress);
  }

  // Fill tree
  nBytes = fTTree -> Fill();

  fTotalBytes += nBytes;
  // Test for new file pointer in case ChangeFile was invoked
  if ( IsFileChange() ) {
    if ( savedir->GetFile() == fTFile ) {
      // This shouldn't happen
      MSG("Per",Msg::kWarning) << "gDirectory points to old file and\n"
        << "ChangeFile requested, will switch to:" << gDirectory->GetPath()
                               << endl;
      savedir = gDirectory; 
    }
    ChangeFile();
  }
  else if ( isAutoSave ) {
    // Save tree to file if requested on this entry interval
    MSG("Per",Msg::kDebug) << "AutoSaving tree " << fTreeName << " with " 
       << fTTree->GetEntries() << " entries to file " 
       << fFullFilePathName << " IsAutoSaveInt/Time/Bytes " 
       << isAutoSaveInt << "/" << isAutoSaveTime << "/" << isAutoSaveBytes 
       << endl;  
    fTFile -> cd();
    fTTree -> AutoSave();  // save tree to file
    fTFile -> SaveSelf();  // save file directory containing this tree
    fLastTimeSaved = recordTime;
    fLastEntrySaved = fEntry;
    fLastBytesSaved = fTotalBytes;
  }

  if ( fCompress >= 0 )fTFile -> SetCompressionLevel(savecompress);
    
  savedir -> cd(); // return to original working directory

  return nBytes;

}

Int_t PerOutputStream::Write() {
  //
  //  Purpose:  Write the tree to file.  By default this overwrites any
  //            previous version of the tree in the file.
  //
  //  Arguments: none.
  //
  //  Return:  number of bytes written to file.
  //
  //  Contact:   S. Kasahara
  // 

  Int_t nBytes = 0;

  if ( !IsOpen() || !IsEnabled() ) return nBytes;

  TDirectory* savedir = gDirectory; // save current working directory

  // Check to see if root has changed the file from underneath us
  if ( IsFileChange() ) ChangeFile();

  fTFile -> cd(); // move to directory of current file

  Int_t savecompress = 9;
  if (fCompress >= 0) {
    savecompress = fTFile -> GetCompressionLevel();
    fTFile -> SetCompressionLevel(fCompress);
  }

  nBytes = fTTree -> Write(NULL,TObject::kOverwrite); // save tree to this file
  MSG("Per",Msg::kDebug) << "Wrote tree " << fTreeName 
    << " with " << fTTree->GetEntries() << " entries " 
    << " to file " << fFullFilePathName << "." << endl;

  if (fCompress >= 0) fTFile -> SetCompressionLevel(savecompress);

  savedir -> cd();  // return to original directory

  return nBytes;

}

---