PerInputStreamManager.cxx

Go to the documentation of this file.

//
// PerInputStreamManager
//
// Package: Per (Persistency).
//
// S. Kasahara 09/2001
//
// Purpose: Input stream manager class used to manage PerInputStreams.
//

#include <iostream>

#include "TTree.h"
#include "TTreeFormula.h"
#include "TMath.h"  //  max,min
#include "TRandom3.h"

#include "Persistency/PerInputStream.h"
#include "Persistency/PerInputStreamManager.h"
#include "MessageService/MsgService.h"
#include "MinosObjectMap/MomNavigator.h"
#include "Record/RecMinos.h"
#include "Record/RecMinosHdr.h"
#include "Record/RecRecord.h"

#include <Conventions/Munits.h>

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

ClassImp(PerInputStreamManager)

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

CVSID("$Id: PerInputStreamManager.cxx,v 1.64 2009/04/29 23:55:53 arms Exp $");

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

//.....................................................................
  
int PerInputStreamManager::AddFile(std::string fullfilepathname, int at,
                                   std::string streamname) {
  //
  //  Purpose:  Add file to file list at position at of specified stream(s).
  //
  //  Arguments: fullfilepathname
  //             at = position in file list (-1 => end of list)
  //             streamname (if "*" (default), apply to all streams)
  //             
  //  Return: number of streams on which action is successfully applied.
  //
  //  Contact:   S. Kasahara

  int oldnstream = this -> GetNumStreamOpen();

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      nstream += instream -> AddFile(fullfilepathname,at); 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) nstream += instream -> AddFile(fullfilepathname,at);
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }
  // reset if first, otherwise leave it where it is
  if ( oldnstream == 0 && nstream > 0 ) SetCurrentVld(Per::GetVldBegin());

  return nstream;

}

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

int PerInputStreamManager::AdvanceLowerBoundTags(const VldContext& keyVld) {
  //  Purpose:  Advance record tags in Per::kLowerBound streams to
  //            be equal to or less than the keyVld given in the argument.
  //
  //  Argument: none.
  //
  //  Return:  1 if successful, else 0 (=> waiting for new records on streams 
  //           when reading in update mode).   
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: To return successful when in update mode requires that the
  //         method is able to see at least one record set with validity
  //         equal to or greater than the specified input validity, or that
  //         the method can see that the open file has been closed by
  //         the writer.
  //

  int nFail = 0;
  for ( StreamMapItr sitr=fStreamMap.begin(); sitr!=fStreamMap.end(); ++sitr ){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if ( instream->GetSequenceMode() != Per::kLowerBound ) continue;
    if ( !instream -> AdvanceLowerBoundTags(keyVld) ) nFail++;
  }

  if ( nFail ) return 0;
  return 1;

}

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

int PerInputStreamManager::AdvanceWindowTags(const VldContext& keyVld) {
  //  Purpose: Attempt to advance record tags in Per::kWindow streams
  //  such that their VldContext is in the window around the key
  //  VldContext.  The window is defined in the individual streams.
  //
  //  Argument: VldContext of the key record.
  //
  //  Return:  1 if successful, else 0 (=> waiting for new records on streams 
  //           when reading in update mode).   
  //

  int nFail = 0;
  for ( StreamMapItr sitr=fStreamMap.begin();sitr!=fStreamMap.end(); sitr++ ) {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if (instream->GetSequenceMode() != Per::kWindow) continue;
    if ( !instream -> AdvanceWindowTags(keyVld) ) nFail++;
  }
  
  if (nFail) return 0;
  return 1;
}

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

int PerInputStreamManager::AdvanceSequentialTags() {
  //  Purpose: Attempt to advance record tags in Per::kSequential &
  //  Per::kRandom streams to make the stream active and at an acceptable
  //  record regardless of VldContext.
  //
  //  Argument: None.
  //
  //  Return:  1 if successful, else 0 (=> waiting for new records on streams 
  //           when reading in update mode).   
  //

  int nFail = 0;
  for ( StreamMapItr sitr=fStreamMap.begin();sitr!=fStreamMap.end(); sitr++ ) {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if (instream->GetSequenceMode() != Per::kSequential &&
        instream->GetSequenceMode() != Per::kRandom      ) continue;
    if ( !instream -> AdvanceTagsList() ) nFail++;
  }
  
  if (nFail) return 0;
  return 1;
}

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

int PerInputStreamManager::AdvanceRecordTags() {
  //  Purpose:  Advance record tags in managed streams to one past
  //            fCurrentVld.
  //
  //  Argument: none.
  //
  //  Return:  1 if successful, else 0 (=> end of all 
  //           streams or waiting for new records on streams when 
  //           reading in update mode).
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: This method checks each stream's current record set to see if the 
  //         set has a vld that matches or is less than fCurrentVld.  If so, 
  //         it advances one record set on that stream.  
  //         At the end of the method, if advance has been successful,
  //         fCurrentVld is set to point to the lowest validity
  //         of all the current stream record tags, or Per::GetVldEnd()
  //         if at the end of all streams. 

  int nFail = 0;
  VldContext newVld = Per::GetVldEnd();
  int  nKey = 0;
  int  nSeq = 0;
  bool allSeqDone = true;

  for ( StreamMapItr sitr=fStreamMap.begin();sitr != fStreamMap.end();++sitr ){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if ( instream->GetSequenceMode() != Per::kKey ) continue;
    nKey++;
    PerRecordTags tags = instream -> GetTags();
    VldContext lastvld = fLastServed[sitr->first];
    bool isdone = false;
    while ( !isdone ) {
      isdone = true;
      if ( tags.IsBegin() || (tags.GetVldContext() <= fCurrentVld 
                          || lastvld == tags.GetVldContext())   
                          || (tags.IsEnd() && !instream->IsEndOfFiles()) ) {
        tags = instream->NextTags();
        if ( tags.IsEnd() ) {
          if ( fUpdateMode && !instream->IsFileEnd() ) nFail++;
          else if (instream->NextFile()) isdone = false;
        }
      }
    }
    if ( tags.GetVldContext() < newVld ) newVld = tags.GetVldContext();
  }
  // if there is no kKey stream, look for a kSequential stream
  if (!nKey) {
    for ( StreamMapItr sitr=fStreamMap.begin();
          sitr != fStreamMap.end();++sitr ){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
      if ( instream->GetSequenceMode() != Per::kSequential ) continue;
      nSeq++;
      PerRecordTags tags = instream -> GetTags();
      VldContext lastvld = fLastServed[sitr->first];
      bool isdone = false;
      while ( !isdone ) {
        isdone = true;
        if ( tags.IsBegin()                               ||
             ( tags.IsEnd() && !instream->IsEndOfFiles() ) ) {
          tags = instream->NextTags();
          if ( tags.IsEnd() ) {
            if ( fUpdateMode && !instream->IsFileEnd() ) nFail++;
            else if (instream->NextFile()) isdone = false;
          }
        }
      }
      if ( tags.GetVldContext() < newVld ) newVld = tags.GetVldContext();
      if ( !( tags.IsEnd() && instream->IsEndOfFiles() ) ) allSeqDone = false;
    } // end for loop over the stream map
  }

  if ( !nKey && !nSeq) {
    // Must be at least one key stream in list of managed streams
    MSG("Per",Msg::kWarning)
      << "No stream of sequence mode Per::kKey or Per::kSequential "
      << "found.\n Must be at least one key or sequential stream to "
      << "facilitate sequencing." 
      << std::endl;
    return 0;
  }

  // Now sequence any Per::kLowerBound or Per:kWindow streams relative
  // to newVld (kSequential and kRandom streams advance themselves)
  if ( !this -> AdvanceLowerBoundTags(newVld) ) nFail++;
  if ( !this -> AdvanceWindowTags(newVld) ) nFail++;

  if ( nFail ) {
    // One or more streams failed to advance but file is not closed and 
    // update mode has been requested (e.g. client is dispatcher)
    // Check to see if we should declare a successful advance anyway because
    // one or more streams indicate that the condition of fMaxSyncDelay has
    // been met.
    MSG("Per",Msg::kVerbose) << nFail << " streams failed to advance\n" 
                             << " checking time to tree end against maxsyncdelay " 
                             << fMaxSyncDelay << "." << endl;
    
    Int_t maxTimeToTreeEnd = fMaxSyncDelay - 1;
    for( StreamMapItr sitr=fStreamMap.begin(); sitr!=fStreamMap.end();++sitr ){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
      if ( instream -> IsOpen() ) {
        Int_t timeToTreeEnd = 0;
        if ( instream -> fLastEntryVld.IsValid() ) {
          timeToTreeEnd = (Int_t)(instream->fLastEntryVld.GetTimeStamp().GetSec())
            - (Int_t)(newVld.GetTimeStamp().GetSec());
        }
        MSG("Per",Msg::kVerbose) << sitr->first << " time to end " 
                                 << timeToTreeEnd << "(sec)" << endl;
        maxTimeToTreeEnd = TMath::Max(maxTimeToTreeEnd,timeToTreeEnd);
        if ( timeToTreeEnd < (Int_t)fMaxSyncDelay ) instream -> UpdateTree();
      }
    }
    if ( maxTimeToTreeEnd < (Int_t)fMaxSyncDelay ) {
      if ( fCurrentVld != newVld ) {
        SetCurrentVld(newVld);
        this -> RewindRecordTags(); // to push back to where we were
      }
      return 0; // failure
    }  
  }

  for ( StreamMapItr sitr = fStreamMap.begin();
        sitr != fStreamMap.end(); ++sitr)  {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if ( instream->GetSequenceMode() != Per::kKey       &&
         instream->GetSequenceMode() != Per::kSequential ) continue;
    PerRecordTags tags = instream -> GetTags();
    if ( tags.GetVldContext() == newVld ) {
      fLastServed[sitr->first] = newVld;
    }
  }

  if ( (newVld == fCurrentVld && !nSeq ) ) {
    return 0; // failure
  }

  if ( nSeq && allSeqDone ) {
    return 0; // failure
  }

  SetCurrentVld(newVld); // equals Per::GetVldEnd() when end of all streams reached
  return 1; // success

}

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

void PerInputStreamManager::CloseFile(string streamName) {
  //  Purpose:  Close current file serving specified stream(s).
  //
  //  Argument: streamName string  name of stream on which to close file.
  //                               if streamname="*" (default), all
  //                               streams will have their files closed.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  PerStreamManager::CloseFile(streamName);
  // When all files have been closed, reset fCurrentVld to reflect this
  if ( this->GetNumStreamOpen() <= 0 ) SetCurrentVld(Per::GetVldEnd()); 

}

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

void PerInputStreamManager::CloseStream(string streamName) {
  //  Purpose:  Close specified stream(s).
  //
  //  Argument: streamName string  name of stream to close.
  //                               if streamName="*" (default), all
  //                               streams will be closed.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  PerStreamManager::CloseStream(streamName);
  // When all files have been closed, reset fCurrentVld to reflect this
  if ( this->GetNumStreamOpen() <= 0 ) SetCurrentVld(Per::GetVldEnd()); 
  if ( streamName == "*" ) fLastServed.clear();
  else {
    std::map<std::string,VldContext>::iterator itr 
                                            = fLastServed.find(streamName);
    if ( itr != fLastServed.end() ) fLastServed.erase(itr);
  }
  
}

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

Int_t PerInputStreamManager::Get(MomNavigator* mom) {
  //  Purpose:  Retrieve current record set from managed record streams
  //            with no advancing.  
  //
  //  Argument: mom        pointer to MomNavigator
  //
  //  Return:  number of records retrieved.  
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  If one record in a record set (sharing common vldcontext) is 
  //          rejected due to a selection cut, then entire set is rejected.
  //

  if ( !mom ) return 0;

  Int_t nrecord = 0;
  if ( IsSelectedSet() ) {
    nrecord = this -> LoadRecord(mom);
  }

  return nrecord;

}

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

VldContext PerInputStreamManager::GetLastEntryVld() const {
  //
  //  Purpose:  Return last entry vld of any open stream.  This is useful
  //            when reading record sets from an open file, e.g. when
  //            used in context of dispatcher, to determine how close we
  //            are to the file end.
  //

  VldContext lastentryvld = Per::GetVldBegin();
  for (StreamMapConstItr itr=fStreamMap.begin(); itr!=fStreamMap.end(); itr++){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
    instream -> UpdateTree();
    if ( instream -> GetLastEntryVld() > lastentryvld ) 
      lastentryvld = instream->GetLastEntryVld();
  }

  return lastentryvld;

}

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

int PerInputStreamManager::GoToFile(int n, std::string streamname) {
  //
  //  Purpose:  Move to nth files in file list for requested stream(s).
  //
  //  Arguments: n = file number (0 is first)
  //             streamname = name of stream to apply movement. if "*"(default)
  //                          will apply to all managed streams.
  //             
  //  Return: number of streams on which action is successfully applied.
  //
  //  Contact:   S. Kasahara
  // 

  VldContext newVld = Per::GetVldEnd();

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      if ( instream -> GoToFile(n) ) {
        nstream++;
        instream -> NextTags(); // must advance to first record of file
        newVld = instream->GetTags().GetVldContext();
      } 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      if ( instream -> GoToFile(n) ) {
        nstream++;
        instream -> NextTags(); // must adavnce to first record of file
        newVld = instream->GetTags().GetVldContext();
      }
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }

  if ( nstream ) {
    // Setting vld to lower or upper bound is determined by movement direction
    if ( newVld > fCurrentVld ) {
      VldContext minVld = this -> GetCurrentKeyVld(true);
      if ( minVld > fCurrentVld ) SetCurrentVld(minVld);
    }
    else {
      VldContext maxVld = this -> GetCurrentKeyVld(false);
      if ( maxVld < fCurrentVld ) SetCurrentVld(maxVld);
    }
  }

  return nstream;

}

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

int PerInputStreamManager::GoToFile(std::string fullfilepathname, 
                                    std::string streamname) {
  //
  //  Purpose:  Move to specified file in file list for requested stream(s).
  //
  //  Arguments: fullfilepathname = file name(if "" move to first file in list)
  //             streamname = name of stream to apply movement. if "*"(default)
  //                          will apply to all managed streams.
  //             
  //  Return: number of streams on which action is successfully applied.
  //
  //  Contact:   S. Kasahara
  // 
  //  Notes: After moving to file, the current record ptr is positioned at
  //         the first record of the file.

  VldContext newVld = Per::GetVldEnd();

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      if ( instream -> GoToFile(fullfilepathname) ) {
        nstream++;
        instream -> NextTags(); // must advance to first record of file
        newVld = instream->GetTags().GetVldContext();
      } 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      if ( instream -> GoToFile(fullfilepathname) ) {
        nstream++;
        instream -> NextTags(); // must advance to first record of file
        newVld = instream->GetTags().GetVldContext();
      }
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }

  if ( nstream ) {
    // Setting vld to lower or upper bound is determined by movement direction
    if ( newVld > fCurrentVld ) {
      VldContext minVld = this -> GetCurrentKeyVld(true);
      if ( minVld > fCurrentVld ) SetCurrentVld(minVld);
    }
    else {
      VldContext maxVld = this -> GetCurrentKeyVld(false);
      if ( maxVld < fCurrentVld ) SetCurrentVld(maxVld);
    }
  }

  return nstream;

}

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

int PerInputStreamManager::IsBeginOfFiles(std::string streamname) const {
  //  Purpose:  Return number of streams that have reached begin of filelist.
  //
  //  Argument: name of stream to check status. If "*" (default), check all 
  //            managed streams
  //
  //  Contact:   S. Kasahara
  //

  int nbegin = 0;  

  if ( streamname != "*" ) {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      nbegin = instream -> IsBeginOfFiles();
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }
  else { 
    for ( StreamMapConstItr citr = fStreamMap.begin();
                            citr!= fStreamMap.end(); ++citr ) { 
      PerInputStream* instream = dynamic_cast<PerInputStream*>(citr->second);
      nbegin += instream -> IsBeginOfFiles();
    }
  }

  return nbegin;

}

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

int PerInputStreamManager::IsEndOfFiles(std::string streamname) const {
  //  Purpose:  Return number of streams that have reached end of filelist.
  //
  //  Argument: name of stream to check status. If "*" (default), check all 
  //            managed streams
  //
  //  Contact:   S. Kasahara
  //

 
  int nend = 0;  

  if ( streamname != "*" ) {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      nend = instream -> IsEndOfFiles();
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }
  else { 
    for ( StreamMapConstItr citr = fStreamMap.begin();
                            citr!= fStreamMap.end(); ++citr ) { 
      PerInputStream* instream = dynamic_cast<PerInputStream*>(citr->second);
      nend += instream -> IsEndOfFiles();
    }
  }

  return nend;

}

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

bool PerInputStreamManager::IsFileEnd() const {
  //  Purpose:  Check if input file has been closed by writer on all
  //            managed streams.
  //
  //  Return:  true if file end detected on all streams.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Invokes PerInputStream::IsFileEnd for all streams.
  //          This method is useful for a user (e.g. dispatcher)
  //          reading open files.  Note that detecting file closure
  //          by the writer does not mean that the user has reached
  //          the end of the entries in the current record map.
  //          Use IsEnd()&&IsFileEnd() to determine if the user has
  //          reached the end of the current record map *and* all
  //          all files have been closed by the writer.
  //

  bool isFileEnd = true;  

  for(StreamMapConstItr citr=fStreamMap.begin();citr!=fStreamMap.end();++citr){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(citr->second);
    if (!instream->IsFileEnd()) isFileEnd = false;
  }

  return isFileEnd;

}

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

bool PerInputStreamManager::IsSelectedSet() {
  //  Purpose:  Check to see if record set corresponding to fCurrentVld
  //            has passed user specified selection cuts.
  //
  //  Arguments: none.
  //
  //  Return:  true if all records passed selection cut.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: If multiple streams have selection cuts applied, the record set
  //         is rejected based on an AND of the different
  //         selection cuts, i.e. rejecting one record rejects the entire
  //         set.

  if ( !fIsNewCurrentVldToSelect ) return fIsSelectedSet;
  
  fIsNewCurrentVldToSelect = false;

  if ( this -> IsEnd() || this -> IsBegin() ) {
    fIsSelectedSet = false;
    return fIsSelectedSet;
  }
  
  fIsSelectedSet = true;  

  if ( !fGlobalSelection.empty() ) {
    MSG("Per",Msg::kVerbose) << "Testing globally applied selection cut " 
                             << fGlobalSelection << endl;
    // Selection cut was applied globally
    this -> UpdateTreeFormula();
    if ( fTTreeFormula ) {
      for(StreamMapConstItr itr =fStreamMap.begin();
                            itr!=fStreamMap.end();++itr) {
        PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
        const PerRecordTags& tags = instream->GetTags();
        Int_t idx = -1;
        if ( tags.GetVldContext() == fCurrentVld ) idx = tags.GetIndexLo();
        TTree* ttree = instream->fTTree;
        if ( ttree ) {
          ttree -> LoadTree(idx);
          if ( !instream->fTObject ) {
            instream -> fTBranch -> SetAddress(&(instream->fTObject));
          }
        }
      }
      Int_t arraySize = fTTreeFormula->GetNdata();
      bool ispass = false;
      for ( int jent = 0; jent < arraySize; jent++ ) {
        if ( fTTreeFormula -> EvalInstance(jent) != 0 ) ispass = true;  
      }
      if ( !ispass) fIsSelectedSet = false;
      for (StreamMapConstItr itr = fStreamMap.begin();
                             itr!= fStreamMap.end(); ++itr) {
        PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
        instream -> Reset(true);
      }
    }
  }
  else {
    for(StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();++itr) {
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      const PerRecordTags& tags = instream->GetTags();
      TTreeFormula* treeformula = instream->fTTreeFormula;
      if ( tags.GetVldContext() == fCurrentVld ) {
        if ( treeformula ) {
          MSG("Per",Msg::kVerbose) << "Testing locally applied selection cut " 
                << instream->fSelection << " on stream " << itr->first << endl;
          TTree* ttree = instream -> fTTree;
          if ( ttree ) {
            for (Int_t idx=tags.GetIndexLo();idx <= tags.GetIndexHi(); idx++) {
              ttree -> LoadTree(idx);
              if ( !instream->fTObject ) {
                instream -> fTBranch -> SetAddress(&(instream->fTObject));
              }
              Int_t arraySize = treeformula->GetNdata();
              bool ispass = false;
              for ( int jent = 0; jent < arraySize; jent++ ) {
                if ( treeformula -> EvalInstance(jent) != 0 ) ispass = true;
              }
              if ( !ispass) fIsSelectedSet = false;
              instream->Reset(true);
            }
          }
        }
      }
      // Check to see if record set is missing a record from a required stream 
      else if ( instream->IsRequired() ) fIsSelectedSet = false;  
    }

  }
  
  if (!fIsSelectedSet) 
     MSG("Per",Msg::kVerbose) << "Record set failed selection cut" << endl;
  else 
     MSG("Per",Msg::kVerbose) << "Record set passed selection cut" << endl;

  return fIsSelectedSet;

}

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

bool PerInputStreamManager::IsValidSelectionString() const {
  // OR of PerInputStream::IsValidSelectionString() results, e.g.
  // if any one stream's application of selection string failed, return
  // false

  for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
    bool isValid = instream -> IsValidSelectionString();
    if ( !isValid ) return false;
  }

  return true;
  
}

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

std::ostream& PerInputStreamManager::ListFile(std::ostream& os, 
                                       std::string streamname) const {
 //
 //  Purpose:  Print the contents of the fFileList for requested stream(s).
 //
 //  Arguments: ostream reference to print on and name of stream(s). If
 //             streamname == "*" (default), lists files on all streams.
 //
 //  Contact:   S. Kasahara
 // 

  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      os << "Stream " << itr->first << ":" << endl;
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      instream -> ListFile(os);
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      os << "Stream " << streamname << ":" << endl;
      instream -> ListFile(os);
    }
    else os << "Stream " << streamname << " not open." << endl;
  }

  return os;

}

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

Int_t PerInputStreamManager::LoadRecord(MomNavigator* mom) {
  //  Purpose:  Loads records with fCurrentVld in input MomNavigator
  //            object.
  //
  //  Argument: pointer to MomNavigator.  
  //
  //  Return:  Number of records loaded. 
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: LoadRecord does not apply user selection cuts to the record set.
  //

  MSG("Per",Msg::kVerbose) << "Load Record Set w/currentVld " << fCurrentVld 
                           << endl;
  
  Int_t nrecord = 0;
  Int_t lastNRecord = 0;
  if ( !mom || this -> IsBegin() || this -> IsEnd() ) return nrecord;

  for ( StreamMapItr itr=fStreamMap.begin(); itr!=fStreamMap.end(); ++itr ) {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
    std::string streamName = itr->first;

    if ( instream -> GetSequenceMode() == Per::kKey ) {
      const PerRecordTags& tags = instream -> GetTags();
      if ( tags.GetVldContext() == fCurrentVld ) {
        nrecord += this -> LoadRecordWithTag(*instream,tags,mom);
      }
    }
    else if ( instream -> GetSequenceMode() == Per::kLowerBound ) {
      const PerRecordTags& tags = instream -> GetTags();
      // First clean Mom of previously loaded object from this stream if not this tag
      VldContext loVld = tags.GetVldContext();
      VldContext hiVld = tags.GetVldContext();
      this -> RemoveFragmentsNotInWindow(mom,streamName,loVld,hiVld);
      if ( !tags.IsBegin() && !tags.IsEnd() ) {
        nrecord += this -> LoadRecordWithTag(*instream,tags,mom);
      }
    }
    else if ( instream -> GetSequenceMode() == Per::kWindow ) {
      std::vector<PerRecordTags> tagslist = instream -> GetWindowTags();
      // First clean Mom of any objects not in range of tags on this list
      VldContext loVld = Per::GetVldEnd(); // set in this way all records will be removed
      VldContext hiVld = Per::GetVldBegin();
      if ( !tagslist.empty() ) {
        loVld = (tagslist[0]).GetVldContext();
        hiVld = (tagslist[tagslist.size()-1]).GetVldContext();
      }
      this ->  RemoveFragmentsNotInWindow(mom,streamName,loVld,hiVld);
      if ( !tagslist.empty() ) {
        std::vector<PerRecordTags>::iterator itr;
        for ( itr = tagslist.begin(); itr != tagslist.end(); itr++ ) {
          const PerRecordTags& tags = *itr;
          nrecord += this -> LoadRecordWithTag(*instream,tags,mom);
        }
      }
    }
    else if ( instream -> GetSequenceMode() == Per::kSequential ||
              instream -> GetSequenceMode() == Per::kRandom      ){
      // clear Mom of all entries of this sort
      this->RemoveAllFragments(mom,streamName);

      double meanMom = instream->GetMeanMom();

      if (!fRandomOverride) {
        fRandomSeed =
          int(10000*fStreamMap.size()+100*meanMom+(instream->fSumTags)+42);
        fRanGen->SetSeed(fRandomSeed);
      }

      int    numEvts =
        ( instream->GetPushRandom() )?
        fRanGen->Poisson(meanMom) : int(ceil(meanMom)) ;

      int iEvt = 0;
      while (iEvt < numEvts) {
        const PerRecordTags& tags = instream -> GetTags();
        int iSuccess = this -> LoadRecordWithTag(*instream,tags,mom);
        iEvt    += iSuccess;
        nrecord += iSuccess;
        if ( !instream->AdvanceTags() ) break;
      } // end loop over # events to push
    } // end if Per::kSequential || Per::kRandom
    lastNRecord = nrecord;
  } // end loop over stream map

  return nrecord;

}

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

void PerInputStreamManager::RemoveFragmentsNotInWindow(
                MomNavigator* mom,std::string streamName, 
                const VldContext& loVld, const VldContext& hiVld) {
  //  Purpose:  Remove fragments < loVld || > hiVld
  //
  //  Argument: pointer to MomNavigator.  
  //
  //  Return:  none
  //
  //  Contact:   S. Kasahara
  //

  TObjArray* fragarray=const_cast<TObjArray*>(mom->GetFragmentArray());
  for (int idx = fragarray->GetEntriesFast()-1; idx >= 0; idx--) {
    TObject* oldobject = fragarray -> At(idx);
    Registry* oldiotags = 0;
    const VldContext* oldvldc = 0;
    if ( RecMinos* oldrecord = dynamic_cast<RecMinos*>(oldobject) ) {
      oldiotags = &(oldrecord -> GetTempTags());
      oldvldc = oldrecord -> GetVldContext();
    }
    else if(RecRecord* oldrecord = dynamic_cast<RecRecord*>(oldobject)){
      oldiotags = &(oldrecord -> GetTempTags());
      oldvldc = &(oldrecord -> GetHeader().GetVldContext());
    }
    if ( oldiotags ) {
      const char* oldtagstream = 0;
      if ( oldiotags -> Get("stream",oldtagstream) 
        && strcmp(streamName.c_str(),oldtagstream) == 0 ) {
        if ( oldvldc && (*oldvldc < loVld || *oldvldc > hiVld) ) {
          fragarray->RemoveAt(idx);
          delete oldobject; oldobject = 0;
        }
      }
    }
  }
  
}

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

void PerInputStreamManager::RemoveAllFragments(
                 MomNavigator* mom,std::string streamName ) {
  //  Purpose:  Remove fragments of stream streamName from mom
  //
  //  Argument: pointer to MomNavigator & streamName  
  //
  //  Return:  none
  //
  //  Contact:   S. Kasahara & K. Arms
  //

  TObjArray* fragarray=const_cast<TObjArray*>(mom->GetFragmentArray());
  for (int idx = fragarray->GetEntriesFast()-1; idx >= 0; idx--) {
    TObject* oldobject = fragarray -> At(idx);
    Registry* oldiotags = 0;
    if ( RecMinos* oldrecord = dynamic_cast<RecMinos*>(oldobject) ) {
      oldiotags = &(oldrecord -> GetTempTags());
    }
    else if(RecRecord* oldrecord = dynamic_cast<RecRecord*>(oldobject)){
      oldiotags = &(oldrecord -> GetTempTags());
    }
    if ( oldiotags ) {
      const char* oldtagstream = 0;
      if ( oldiotags -> Get("stream",oldtagstream)     &&
           strcmp(streamName.c_str(),oldtagstream) == 0 ) {
          fragarray->RemoveAt(idx);
          delete oldobject; oldobject = 0;
      } // end if tags are from stream streamName
    } // end if tags exist
  } // end for loop over mom fragment array
  
}

//.....................................................................
  
Int_t PerInputStreamManager::LoadRecordWithTag(const PerInputStream& instream,
                                               const PerRecordTags& tags, 
                                               MomNavigator* mom) {
  //  Purpose:  Loads records from this tag in input MomNavigator
  //            object.
  //
  //  Argument: pointer to MomNavigator.  
  //
  //  Return:  Number of records loaded. 
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: LoadRecord does not apply user selection cuts to the record set.
  //

  int nrecord = 0;

  // Load the records in the tag set
  for (Int_t idx = tags.GetIndexLo(); idx <= tags.GetIndexHi(); idx++) {
    std::string streamName = instream.GetStreamName();
    TObject* object = mom -> GetFragmentByInputTag(streamName.c_str(),
            (tags.GetTreeName()).c_str(),idx,(tags.GetFileName()).c_str());
    if ( object ) continue;  // object already in Mom don't bother
    object = (const_cast<PerInputStream&>(instream)).GetObject(idx,false);
    if ( object ) {
      Registry* temptags = 0;
      if ( RecMinos* record = dynamic_cast<RecMinos*>(object) ) {
        temptags = &(record->GetTempTags());
        if ( instream.GetSequenceMode() != Per::kKey        &&
             instream.GetSequenceMode() != Per::kSequential &&
             instream.GetSequenceMode() != Per::kRandom      )
          record -> SetTransient(false);
      }
      else if ( RecRecord* record = dynamic_cast<RecRecord*>(object) ) {
        temptags = &(record->GetTempTags());
        if ( instream.GetSequenceMode() != Per::kKey        &&
             instream.GetSequenceMode() != Per::kSequential &&
             instream.GetSequenceMode() != Per::kRandom      )
          record -> SetTransient(false);
      }
        
      if ( temptags ) {
        // Stamp record with i/o tags
        temptags->UnLockValues();
        std::string tagname = instream.GetClassName()+".fTempTags";
        temptags->SetName(tagname.c_str());
        temptags->Set("stream",streamName.c_str());
        temptags->Set("tree",(tags.GetTreeName()).c_str());
        temptags->Set("index",idx);
        temptags->Set("file",(tags.GetFileName()).c_str());
        temptags->LockValues();
      }
      // And add record to Mom, mom now owns record
      if ( nrecord == 0) {
        MSG("Per",Msg::kVerbose) << "..Loading records with Vld" 
                                 <<  fCurrentVld << " from stream " 
                                 << streamName.c_str() << ":" << endl;
      }
      MSG("Per",Msg::kVerbose) << "....Record " << nrecord << " from"
                               << " stream " << streamName
                               << " tree " << tags.GetTreeName()
                               << " index " << idx
                               << " file "<< tags.GetFileName()
                               << endl;
      mom -> AdoptFragment(object);
        
      nrecord++;
    } // object retrieved
  } // loop over tag tree indices

  return nrecord; 

}

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

Int_t PerInputStreamManager::Next(MomNavigator* mom, UInt_t advanceby) {
  //  Purpose:  Retrieve next set of records from managed streams of
  //            common VldContext which satisfy user's  
  //            selection cuts (if specified).  The records are 
  //            extracted from the input streams and loaded into the
  //            input MomNavigator object.    
  //
  //  Argument: mom        pointer to MomNavigator. (If set 0 (default), the 
  //                       method will advance to the record set of interest
  //                       but not load the records.  The user can then
  //                       retrieve the record set with the Get method.)
  //            advanceby  number of selected record sets to advance by
  //                       (default = 1).  Can be used to skip over record
  //                       sets.
  //
  //  Return:  number of selected record sets advanced.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Entries are subject to any user-specified selection strings
  //          as described under PerInputStream::GetObject and 
  //          PerInputStreamManager::IsSelectedSet.
  //

  MSG("Per",Msg::kVerbose) << "Next advance by " << advanceby << " sets." 
                           << " Start at vld " << fCurrentVld << endl;
  
  if ( this -> IsEnd() ) return 0;
  VldContext saveCurrentVld = fCurrentVld; 

  // Search for entry satisfying selection cuts
  UInt_t nselect = 0;
  while ( nselect < advanceby && this -> AdvanceRecordTags() ) {
    MSG("Per",Msg::kVerbose) << "Advanced to vld " << fCurrentVld 
                             << endl;
    if ( IsSelectedSet() ) {
      nselect++;
      if ( nselect >= advanceby ) {
        // Load record in Mom
        if ( mom ) this -> LoadRecord(mom);
      }
    }
  }

  if ( advanceby > 1 ) {
    // current vld is left at last record served if in updatemode
    if ( nselect < advanceby && 
         fUpdateMode         && 
         !IsFileEnd()         ) {
      while ( fCurrentVld > saveCurrentVld ) {
        this -> RewindRecordTags();
      }
      // At this point fCurrentVld should be back to original
      if ( saveCurrentVld != fCurrentVld ) {
        MSG("Per",Msg::kWarning) 
          << "Failure to rewind to original state in Next method." << endl;
      }
      nselect = 0; 
    }
  }

  return (Int_t)nselect;

}

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

int PerInputStreamManager::NextFile(int n, std::string streamname) {
  //
  //  Purpose:  Move stream(s) forward n files in file list.  If n exceeds 
  //            number of files to end of list, ptr is left at eof marker.  
  //
  //  Arguments: n = number of files to advance (default = 1)
  //             streamname = name of stream to apply movement. if "*"(default)
  //                          will apply to all managed streams.
  //             
  //  Return: number of streams on which action is successfully applied.
  //
  //  Contact:   S. Kasahara
  // 
  //  Notes: After moving to file, the current record ptr is positioned at
  //         the first record of the file.

  MSG("Per",Msg::kVerbose) << "PerInputStreamManager::NextFile("
                           << n << "," << streamname << ") called"
                           << endl;

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      if ( instream -> NextFile(n) ) {
        nstream++;
        instream -> NextTags(); // must advance to first record of file
      } 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) { 
      if ( instream -> NextFile(n) ) {
        nstream++;
        instream -> NextTags(); // must advance to first record of file
      }
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }

  // Recalculate fCurrentVld to point to lower bound of new set of tags
  VldContext minVld = this -> GetCurrentKeyVld(true);
  if ( minVld > fCurrentVld ) SetCurrentVld(minVld);

  return nstream;

}

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

VldContext PerInputStreamManager::GetCurrentKeyVld(bool isMin) const {
  // 
  // Purpose:  Determine current minimum or maximum validity of managed 
  //           key stream record block tags
  //
  // Arguments: isMin = true (default) => return minimum validity, else
  //            returns max validity
  //
  // Contact: S. Kasahara
  //

  VldContext currentVld = Per::GetVldEnd();
  if ( !isMin ) currentVld = Per::GetVldBegin();

  for( StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++ ) {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
    if ( instream && instream->IsOpen() ) {
      VldContext tagVld = instream->GetTags().GetVldContext();
      if ( isMin && tagVld < currentVld ) currentVld = tagVld;
      else if ( !isMin && tagVld > currentVld ) currentVld = tagVld;
    } 
  }

  return currentVld;

}

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

PerInputStream* PerInputStreamManager::OpenStream(std::string streamName, 
                                                  std::string treeName) {
  //
  //  Purpose:  Open input stream with name streamName serving tree treeName.
  //
  //  Arguments: streamName  string   name of stream to be opened. Names of 
  //                                  managed streams must be unique.
  //             treeName    string   name of tree served by this stream.
  //
  //  Return:  pointer to PerInputStream. If unable to open stream (because
  //           stream with this streamName has already been opened), returns
  //           (PerInputStream*)0.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: PerInputStream objects are owned by the PerInputStreamManager 
  //         and should only be deleted through the 
  //         PerStreamManager::CloseStream method.

  bool openok = false;

  PerInputStream* stream=dynamic_cast<PerInputStream*>(fStreamMap[streamName]);
  if ( !stream ) {
    // Stream not found in map, need to create new one
    stream = new PerInputStream(treeName);
    stream -> SetStreamName(streamName);
    stream -> SetUpdateMode(fUpdateMode);
    Per::ESequenceMode seqMode = Per::GetDefSequenceMode(streamName.c_str());
    stream -> SetSequenceMode(seqMode);
    fStreamMap[streamName] = stream;
    fLastServed[streamName] = Per::GetVldEnd();
    openok = true;
  }
  else {
    MSG("Per",Msg::kWarning)
      <<"Stream manager failed to open\nrequested stream "
      << streamName << " because name conflicts with previously opened stream."
      << endl;
  }

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

}

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

PerInputStreamManager::PerInputStreamManager() : fCurrentVld(Per::GetVldEnd()),
      fGlobalSelection(""),fTTreeFormula(0),fFormulaMap(),
      fUpdateMode(false),fMaxSyncDelay(15), fIsNewCurrentVldToSelect(true), 
      fIsSelectedSet(false), fRandomSeed(0), fRandomOverride(false) {
  //
  //  Purpose:  Default constructor.
  //
  //  Arguments: none.
  //
  //  Return:  n/a.
  //
  //  Contact:   S. Kasahara
  //
  fRanGen = new TRandom3(0);
  MSG("Per",Msg::kDebug)
    << "PerInputStreamManager ctor : Random Seed set to "
    << fRandomSeed << endl;
}

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

PerInputStreamManager::~PerInputStreamManager() {
  //
  //  Purpose:  Default constructor.
  //
  //  Arguments: none.
  //
  //  Return:  n/a.
  //
  //  oCntact:   S. Kasahara
  //

 if ( fTTreeFormula ) delete fTTreeFormula; fTTreeFormula = 0;
 fFormulaMap.clear();

 if ( fRanGen ) delete fRanGen; fRanGen = 0;

}

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

Int_t PerInputStreamManager::Previous(MomNavigator* mom, UInt_t rewindby) {
  //  Purpose:  Retrieve previous set of records from managed streams of
  //            common VldContext which satisfy user's  
  //            selection cuts (if specified).  The records are 
  //            extracted from the input streams and loaded into the
  //            input MomNavigator object.    
  //
  //  Argument: mom        pointer to MomNavigator (default=0 means 
  //                       streams are rewound but no record set is loaded.
  //                       The user can then use Get to retrieve the record 
  //                       set.)
  //            rewindby   number of selected record sets to rewind by
  //                       (default = 1).  Can be used to skip over record
  //                       sets.
  //
  //  Return:  number of record sets rewound.  
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Entries are subject to any user-specified selection strings
  //          as described under PerInputStream::GetObject and 
  //          PerInputStreamManager::IsSelectedSet.
  //

  if ( this -> IsBegin() ) return 0;

  UInt_t nselect = 0;
  // Search for entry satisfying selection cuts
  while ( nselect < rewindby && this -> RewindRecordTags() ) {
    if ( IsSelectedSet() ) {
      nselect++;
      if ( nselect >= rewindby ) {
        // Load record in Mom
        if ( mom ) this -> LoadRecord(mom);
      }
    }
  }

  return (Int_t)nselect;

}

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

int PerInputStreamManager::PrevFile(int n, std::string streamname) {
  //
  //  Purpose:  Move stream(s) back n files in file list.  If n exceeds 
  //            number of files to beginning of list, ptr is left at first
  //            file.
  //
  //  Arguments: n = number of files to rewind (default = 1)
  //             streamname = name of stream to apply movement. if "*"(default)
  //                          will apply to all managed streams.
  //             
  //  Return: number of streams on which action is successfully applied.
  //
  //  Contact:   S. Kasahara
  // 
  //  Notes: After moving to file, the current record ptr is positioned at
  //         the first record of the file.

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      if ( instream -> PrevFile(n) ) {
        nstream++;
        instream -> NextTags(); // position at first record of file
      } 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) {
      if ( instream -> PrevFile(n) ) {
        nstream++;
        instream -> NextTags(); // position at first record of file
      }
    }
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }

  // Recalculate fCurrentVld to point to upper bound of new set of tags
  VldContext maxVld = this -> GetCurrentKeyVld(false);
  if ( maxVld < fCurrentVld ) SetCurrentVld(maxVld);

  return nstream;

}

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

std::ostream& PerInputStreamManager::Print(std::ostream& os, 
                                           const char* option) const {
  //
  //  Purpose:  Print status of input stream manager on std::ostream.
  //
  //  Arguments: os           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" ) {
    Int_t nenabled = 0;
    for ( StreamMapConstItr citr  = fStreamMap.begin();
                            citr != fStreamMap.end(); ++ citr ) {
      PerInputStream* instream = dynamic_cast<PerInputStream*>(citr->second);
      if ( instream -> IsEnabled() && instream -> IsOpen() ) {
        os << "    " << nenabled << ")" << citr->first << " serving tree " 
           << instream -> GetTreeName()
           << " w/" << instream->GetNumEntries() 
           << " entries (current entry = " << instream->GetEntry()
           << ") using sequence mode " 
           << Per::AsString(instream->GetSequenceMode())
           << "\n      from the " 
           << ((instream->IsFileEnd()) ? "closed file " : "open file ")  
           << instream->GetFullFilePathName() << "." << endl;
        nenabled++;
      }
    }
  }
  else {
    os << "PerInput";
    PerStreamManager::Print(os);
  }
  return os;

}

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

VldContext PerInputStreamManager::RecordsAt(MomNavigator* mom,
                                            const VldContext& vld) {
  //  Purpose:  Retrieve record set with requested VldContext from managed
  //            record streams.  Record set is subject to selection
  //            cuts as specified by user.  The records are extracted from
  //            the input streams and loaded into the input MomNavigator
  //            object.  If exact match not found, the record set corresponding
  //            to the validity just beyond that requested is loaded.
  //
  //  Argument: mom        pointer to MomNavigator
  //                       (if == 0, streams are moved to vld but records are
  //                        not loaded)
  //            vld        VldContext of requested record set
  //
  //  Return:  VldContext of new record set (= Per::GetVldEnd() or 
  //           Per::GetVldBegin() if end-of-input-streams or 
  //           begin-of-input-streams respectively).
  //
  //  Contact:   S. Kasahara
  //

  if ( vld == fCurrentVld && IsSelectedSet() ) {
    if ( mom ) this -> LoadRecord(mom);
    return fCurrentVld;
  }

  if ( vld < fCurrentVld ) {
    while ( RewindRecordTags() ) {
      if ( vld >= fCurrentVld && IsSelectedSet() ) {
        if ( mom ) this -> LoadRecord(mom);
        return fCurrentVld;
      }
    }
  }
  else {
    while ( AdvanceRecordTags() ) {
      if ( vld <= fCurrentVld && IsSelectedSet() ) {
        if ( mom ) this -> LoadRecord(mom);
        return fCurrentVld;
      }
    }
  }

  return fCurrentVld; // begin-or-end-of-streams

}

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

int PerInputStreamManager::RemoveFile(std::string fullfilepathname, 
                                      std::string streamname) {
  //
  //  Purpose:  Remove file from file list of specified stream(s).
  //
  //  Arguments: fullfilepathname (if "*" (default), remove all files.)
  //             streamname (if "*" (default), apply to all streams)
  //             
  //  Return: number of streams on which action is successfully applied.
  //         
  //  Contact:   S. Kasahara
  // 

  int nstream = 0;
  if ( streamname == "*" ) {
    for (StreamMapConstItr itr=fStreamMap.begin();itr!=fStreamMap.end();itr++){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(itr->second);
      nstream += instream -> RemoveFile(fullfilepathname); 
    }
  }
  else {
    PerInputStream* instream = this -> GetOpenedStream(streamname);
    if ( instream ) nstream += instream -> RemoveFile(fullfilepathname);
    else MSG("Per",Msg::kWarning) << "Stream " << streamname << " not open." 
                                  << endl;
  }

  return nstream;

}

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

int PerInputStreamManager::RewindRecordTags() {
  //  Purpose:  Rewind record tags in managed streams to one previous to
  //            fCurrentVld.
  //
  //  Argument: none.
  //
  //  Return:  1 if success, else 0 (end of all 
  //           streams reached)
  //
  //  Contact:   S. Kasahara
  //
  //  Notes: This method checks each stream's current record set to see if the 
  //         set has a vld that matches or exceeds fCurrentVld.  If so, it 
  //         rewinds one record set on that stream.  
  //         At the end of the method, if rewind has been successful,
  //         fCurrentVld is set to point to the highest validity
  //         of all the current stream record tags, or Per::GetVldBegin()
  //         if at the beginning of all streams. 

  VldContext newVld = Per::GetVldBegin();
  int nKey = 0;
  int nSeq = 0;
  bool allSeqDone = true;

  for ( StreamMapItr sitr=fStreamMap.begin();sitr != fStreamMap.end();++sitr ){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if ( instream->GetSequenceMode() != Per::kKey ) continue;
    nKey++;
    PerRecordTags tags = instream->GetTags();
    bool isdone = false;
    while ( !isdone ) {
      isdone = true;
      if ( tags.IsEnd() || tags.GetVldContext() >= fCurrentVld 
                        || (tags.IsBegin() && !instream->IsBeginOfFiles()) ) {
        tags = instream->PrevTags();
        if ( tags.IsBegin() ) {
          if ( instream -> PrevFile() ) isdone = false;
        }
      }
    }
    if ( tags.GetVldContext() > newVld ) newVld = tags.GetVldContext();
  }

  // if there is no kKey stream, look for a kSequential stream
  if ( !nKey ) {
    for ( StreamMapItr sitr=fStreamMap.begin();
          sitr != fStreamMap.end();++sitr ){
      PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
      if ( instream->GetSequenceMode() != Per::kSequential ) continue;
      nSeq++;
      PerRecordTags tags = instream->GetTags();
      bool isdone = false;
      while ( !isdone ) {
        isdone = true;
        if ( tags.IsEnd()                                     ||
             ( tags.IsBegin() && !instream->IsBeginOfFiles() ) ) {
          tags = instream->PrevTags();
          if ( tags.IsBegin() ) {
            if ( instream -> PrevFile() ) isdone = false;
          }
        }
      }
      if ( tags.GetVldContext() > newVld ) newVld = tags.GetVldContext();
      if ( !( tags.IsBegin() && instream->IsBeginOfFiles() ) ) allSeqDone = false;
    } // end loop over the stream map
  } // end if no kKey stream

  if ( !nKey && !nSeq) {
    // Must be at least one key stream in list of managed streams
    MSG("Per",Msg::kWarning)
      << "No stream of sequence mode Per::kKey or Per::kSequential "
      << "found.\n Must be at least one key or sequential stream to "
      << "facilitate sequencing."
      << std::endl;
    return 0;
  }

  if ( fCurrentVld == newVld && !nSeq ) {
    return 0; // failure
  }

  if ( nSeq && allSeqDone ) {
    return 0; // failure
  }

  for ( StreamMapItr sitr = fStreamMap.begin();
        sitr != fStreamMap.end(); ++sitr)  {
    PerInputStream* instream = dynamic_cast<PerInputStream*>(sitr->second);
    if ( instream->GetSequenceMode() != Per::kKey       &&
         instream->GetSequenceMode() != Per::kSequential ) continue;
    fLastServed[sitr->first] = Per::GetVldBegin();
  }

  SetCurrentVld(newVld); //equals Per::GetVldBegin() when begin of all strms reached
  this -> AdvanceLowerBoundTags(fCurrentVld);
  this -> AdvanceWindowTags(fCurrentVld);
  // kSequential and kRandom streams advance themselves as appropriate

  return 1;

}

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

bool PerInputStreamManager::SetFile(string streamName,string fullFilePathName,
                                    Per::EAccessMode accessMode) {
  //  Purpose:  Sets new file for specified stream(s).
  //
  //  Argument: streamName  string  name of stream to set file.
  //                                if streamName="*" (default), all
  //                                streams will have their file set.
  //            fullFilePathName string  new filename.
  //            accessMode          Per::EAccessMode accessMode in which to
  //                                open file.
  //
  //  Return:  bool  set true if PerInputStream::SetFile returned true for at
  //                 least one of the requested streams.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Invokes PerInputStream::SetFile for each requested stream.
  //

   Int_t oldNumStream = this -> GetNumStreamOpen();
   bool openok 
   = PerStreamManager::SetFile(streamName,fullFilePathName,accessMode);
   // reset if first, otherwise leave it where it is
   if ( openok && oldNumStream == 0) {
     SetCurrentVld(Per::GetVldBegin());
   } 

   return openok;

}

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

void PerInputStreamManager::SetCurrentVld(const VldContext& vld) {
  // Set fCurrentVld to vld, and mark that IsSelected needs to be
  // applied to this validity
 
  if ( vld != fCurrentVld ) fIsNewCurrentVldToSelect = true;
  fCurrentVld = vld;

}

//.....................................................................
  
void PerInputStreamManager::SetFileEnd(bool isFileEnd) {
  //  Purpose:  Force file end on input file.  Can be used for recovering
  //            aborted files.
  //
  //  Arguments: isFileEnd  true (default) to force file end
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  for(StreamMapConstItr citr=fStreamMap.begin();citr!=fStreamMap.end();++citr){
    PerInputStream* instream = dynamic_cast<PerInputStream*>(citr -> second);
    instream -> SetFileEnd(isFileEnd);
  }

  return;

}

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

void PerInputStreamManager::SetSelection(string streamName, string selection,
                                         bool isRequired) {
  //  Purpose:  Set selection string for specified stream(s).
  //
  //  Argument: streamName  string  name of stream to set selection..
  //                                if streamName="*" (default), all
  //                                streams will have selection set.
  //            selection   string  selection cut to be applied to TTree
  //                                served by this stream. See PerInputStream::
  //                                SetSelection for more information.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //
  //  Notes:  Invokes PerInputStream::SetSelection for each requested stream.
  //

  if (streamName == "*") {
    // Apply selection string globally
    fGlobalSelection = selection;
    fFormulaMap.clear();
    if ( fTTreeFormula ) delete fTTreeFormula; fTTreeFormula = 0;
    // New TTreeFormula will be built at time of application
  }
  else {
    // SetSelection on stream of specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    if (stream) { 
      stream -> SetSelection(selection);
      stream -> SetRequired(isRequired);
    }
  }

}

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

void PerInputStreamManager::SetSequenceMode(string streamName, 
                                            Per::ESequenceMode sequenceMode) {
  //  Purpose:  Set sequence mode for specified stream(s).
  //            For kSequential and kRandom streams, a pointer to the
  //            Manager's random numner generator is passed as well
  //
  //  Argument: streamName  string  name of stream to set selection..
  //                                if streamName="*" (default), all
  //                                streams will have selection set.
  //            sequenceMode        Per::kKey (default), Per::kLowerBound,
  //                                Per::kWindow, Per::kSequential, or
  //                                Per::kRandom
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara & K. Arms
  //
  //  Notes:  Invokes PerInputStream::SetSequenceMode for each requested 
  //          stream.
  //


  if (streamName == "*") {
    // Set selection on all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
          citr!= fStreamMap.end(); ++citr ) {
      ((PerInputStream*)(citr -> second)) -> SetSequenceMode(sequenceMode);
      if (sequenceMode == Per::kSequential ||
          sequenceMode == Per::kRandom      )
        ((PerInputStream*)(citr -> second)) -> SetRandomGenerator(fRanGen);
    } // end for loop
  } else {
    // SetSelection on stream of specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    if (stream) {
      stream -> SetSequenceMode(sequenceMode);
      if (sequenceMode == Per::kSequential ||
          sequenceMode == Per::kRandom      )
        stream -> SetRandomGenerator(fRanGen);
    } // end if stream is OK
  } // end else
}

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

void PerInputStreamManager::SetMaxFileRepeat(std::string streamName,
                                             int         numRepeat )
{
  // Purpose: Set the maximum number of times to reuse files
  //          from this stream before moving to the next file
  //          (only used for kSequential & kRandom stream sequence
  //          modes)
  //
  // Argument: stream name of interest, max number (int >= 0)
  //
  // Return: none
  //
  // Note: Invokes PerInPutStream::SetMaxFileRepeat() for each stream
  //

  if (numRepeat < 0) numRepeat = 0;

  if (streamName == "*") {
    // Set info for all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
          citr!= fStreamMap.end(); ++citr ) {
      ((PerInputStream*)(citr -> second)) -> SetMaxFileRepeat(numRepeat);
    } // end loop over streams
  }
  else {
    // Set info for specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    stream -> SetMaxFileRepeat(numRepeat);
  }
}

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

void PerInputStreamManager::SetMeanMom(std::string streamName,double mm)
{
  // Purpose: Set the mean number of events to push to Mom from this stream
  //
  // Argument: stream name of interest, mean number (double >= 0.)
  //
  // Return: none
  //
  // Note: Invokes PerInPutStream::SetMeanMom() for each stream
  //

  if (mm < 0.) return;

  if (streamName == "*") {
    // Set info for all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
          citr!= fStreamMap.end(); ++citr ) {
      ((PerInputStream*)(citr -> second)) -> SetMeanMom(mm);
    } // end loop over streams
  }
  else {
    // Set info for specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    stream -> SetMeanMom(mm);
  }
}

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

void PerInputStreamManager::SetPushRandom(std::string streamName,bool tf)
{
  // Purpose: Set whether to push a constant (false) or random (true)
  //          number of events to Mom from this stream
  //
  // Argument: stream name of interest, bool
  //
  // Return: none
  //
  // Note: Invokes PerInPutStream::SetPushRandom() for each stream
  //

  if (streamName == "*") {
    // Set info for all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
          citr!= fStreamMap.end(); ++citr ) {
      ((PerInputStream*)(citr -> second)) -> SetPushRandom(tf);
    } // end loop over streams
  }
  else {
    // Set info for specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    stream -> SetPushRandom(tf);
  }
}

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

void PerInputStreamManager::SetRandomSeed(int rSeed)
{ 
  // User input method. If code modifies the random seed, use fRanGen->SetSeed()
  fRanGen->SetSeed(rSeed);
  fRandomSeed = rSeed;
  fRandomOverride = true;
  MSG("Per",Msg::kInfo)
    << "PerInputStreamManager::SetRandomSeed : Random Seed set to "
    << ((rSeed==0)? fRanGen->GetSeed() : fRandomSeed) << endl;

}

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

void PerInputStreamManager::SetWindow(string streamName, 
                                      double lower, double upper)
{
  //  Purpose:  Set window to use if kWindow sequence mode is used.
  //
  //  Argument: streamName  string  name of stream to set window.
  //                                if streamName="*" (default), all
  //                                streams will have selection set.
  //            lower       double  define window relative to 
  //            upper       double  the key stream VldContext
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara, B. Viren
  //
  //  Notes: Invokes PerInputStream::SetWindow for each requested
  //  stream.
  //


  if (streamName == "*") {
    // Set selection on all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
                            citr!= fStreamMap.end(); ++citr ) 
      ((PerInputStream*)(citr -> second)) -> SetWindow(lower,upper);
  }
  else {
    // SetSelection on stream of specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    if (stream) stream -> SetWindow(lower,upper);
  }

}

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

void PerInputStreamManager::SetPerOwnedDisabled(string streamName, 
                                                bool perowneddisabled)
{
  //  Purpose:  Set PerOwnedDisabled for requested stream(s).
  //
  //  Argument: streamName  string  name of stream. if streamName="*"(default),
  //                                apply to all streams.
  //            perowneddisabled  bool  (default = true). Argument passed to 
  //                                     SetPerOwnedDisabled for each
  //                                     requested stream.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  if (streamName == "*") {
    // Invoke SetPerOwnedDisabled on all streams
    for ( StreamMapConstItr citr = fStreamMap.begin(); 
                            citr!= fStreamMap.end(); ++citr ) 
      ((PerInputStream*)(citr -> second)) 
                      -> SetPerOwnedDisabled(perowneddisabled);
  }
  else {
    // Invoke SetPerOwnedDisabled on stream of specified streamName
    PerInputStream* stream = (PerInputStream*)GetOpenedStream(streamName);
    if (stream) stream -> SetPerOwnedDisabled(perowneddisabled);
  }

}

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

void PerInputStreamManager::SetUpdateMode(bool updateMode) {
  //  Purpose:  Set update mode true or false. true should be set
  //            when reading open files (e.g. user is dispatcher).
  //
  //  Argument: updateMode  bool  true => stream tree will be updated when
  //                              end of entries is reached. Appropriate
  //                              for read of open files.  Default when
  //                              stream is opened is false.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  fUpdateMode = updateMode;
  for ( StreamMapConstItr citr = fStreamMap.begin(); 
                          citr!= fStreamMap.end(); ++citr ) 
    ((PerInputStream*)(citr -> second)) -> SetUpdateMode(updateMode);

}

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

void PerInputStreamManager::UpdateTreeFormula() {
  //  Purpose:  Private method, called by IsSelectedSet when selection
  //            cut has been applied globally.
  //
  //  Return:  none.
  //
  //  Contact:   S. Kasahara
  //

  if ( fGlobalSelection.empty() ) return;
  if ( this -> GetNumStreamOpen() <= 0 ) {
    if ( fTTreeFormula ) delete fTTreeFormula; fTTreeFormula = 0;
    fFormulaMap.clear();
    return;
  }

  bool rebuild = false;
  if ( !fTTreeFormula ) rebuild = true;
  else {
    // Loop through formula map checking for differences
    if ( this -> GetNumStream() != fFormulaMap.size() ) rebuild = true;
    else {
      for ( StreamMapConstItr citr = fStreamMap.begin();
                              citr!= fStreamMap.end(); ++citr) {
        PerInputStream* stream = (PerInputStream*)(citr->second);
        std::string streamname = citr->first;
        if ( fFormulaMap[streamname] != stream->GetFullFilePathName() ) {
          rebuild = true;
          break;
        }
      }
    }
  }

  if ( !rebuild ) return;  
  //  if ( fTTreeFormula ) delete fTTreeFormula; fTTreeFormula = 0;
  // If file boundary has been crossed on one of the streams, must
  // update treeformula
  fFormulaMap.clear();
  std::string minstreamnm = "";
  Int_t minentries = -1; 
  for(StreamMapConstItr citr=fStreamMap.begin();citr!=fStreamMap.end();++citr){
    PerInputStream* stream = (PerInputStream*)(citr -> second);
    if ( stream->fTTree != 0  && 
        (stream->GetNumEntries() < minentries || minstreamnm.empty()) ) {
      minentries  = stream->GetNumEntries();
      minstreamnm = citr->first;
    }
    fFormulaMap.insert(make_pair(citr->first,stream->GetFullFilePathName()));
  }


  // Smallest tree becomes master, else root will complain
  TTree* mastertree = ((PerInputStream*)fStreamMap[minstreamnm])->fTTree;
  for(StreamMapConstItr citr=fStreamMap.begin();citr!=fStreamMap.end();++citr){
    if ( citr->first == minstreamnm ) continue;
    PerInputStream* stream = (PerInputStream*)(citr->second);
    if ( stream->fTTree != 0 ) mastertree -> AddFriend(stream->fTTree);
  }
  if ( !fTTreeFormula ) {
    fTTreeFormula 
    = new TTreeFormula("PerMgrSelection",fGlobalSelection.c_str(),mastertree);
  }
  else {
    fTTreeFormula -> SetTree(mastertree);
    fTTreeFormula -> UpdateFormulaLeaves();
  }
  if ( fTTreeFormula -> GetNdim() <= 0 ) {
    // TTreeformula build failed
    delete fTTreeFormula; fTTreeFormula = 0;
    fFormulaMap.clear();
  }

  for(StreamMapConstItr citr=fStreamMap.begin();citr!=fStreamMap.end();++citr){
    if ( citr->first == minstreamnm ) continue;
    PerInputStream* stream = (PerInputStream*)(citr->second);
    if ( stream->fTTree != 0 ) mastertree -> RemoveFriend(stream->fTTree);
  }

  return;

}

---