// @(#)root/proofplayer:$Id: TPacketizerAdaptive.h 20882 2007-11-19 11:31:26Z rdm $ // Author: Jan Iwaszkiewicz 11/12/06 /************************************************************************* * Copyright (C) 1995-2006, Rene Brun and Fons Rademakers. * * All rights reserved. * * * * For the licensing terms see $ROOTSYS/LICENSE. * * For the list of contributors see $ROOTSYS/README/CREDITS. * *************************************************************************/ #ifndef ROOT_TPacketizerAdaptive #define ROOT_TPacketizerAdaptive ////////////////////////////////////////////////////////////////////////// // // // TPacketizerAdaptive // // // // This packetizer is based on TPacketizer but uses different // // load-balancing algorithms and data structures. // // Two main improvements in the load-balancing strategy: // // - First one was to change the order in which the files are assigned // // to the computing nodes in such a way that network transfers are // // evenly distributed in the query time. Transfer of the remote files // // was often becoming a bottleneck at the end of a query. // // - The other improvement is the use of time-based packet size. We // // measure the processing rate of all the nodes and calculate the // // packet size, so that it takes certain amount of time. In this way // // packetizer prevents the situation where the query can't finish // // because of one slow node. // // // // The data structures: TFileStat, TFileNode and TSlaveStat are // // enriched + changed and TFileNode::Compare method is changed. // // // ////////////////////////////////////////////////////////////////////////// #ifndef ROOT_TVirtualPacketizer #include "TVirtualPacketizer.h" #endif class TMessage; class TTimer; class TTree; class TMap; class TNtupleD; class TProofStats; class TRandom; class TPacketizerAdaptive : public TVirtualPacketizer { public: // public because of Sun CC bug class TFileNode; class TFileStat; class TSlaveStat; private: TList *fFileNodes; // nodes with files TList *fUnAllocated; // nodes with unallocated files TList *fActive; // nodes with unfinished files TMap *fSlaveStats; // slave status, keyed by correspondig TSlave Int_t fMaxPerfIdx; // maximum of our slaves' performance index Float_t fFractionOfRemoteFiles; // fraction of TDSetElements // that are on non slaves Long64_t fNEventsOnRemLoc; // number of events in currently // unalloc files on non-worker loc. Float_t fCumProcTime; // sum of proc time of all packets so far Float_t fBaseLocalPreference; // indicates how much more likely // the nodes will be to open their local files (1 means indifferent) Bool_t fForceLocal; // if 1 - eliminate the remote processing TPacketizerAdaptive(); TPacketizerAdaptive(const TPacketizerAdaptive&); // no implementation, will generate void operator=(const TPacketizerAdaptive&); // error on accidental usage TFileNode *NextNode(); void RemoveUnAllocNode(TFileNode *); TFileNode *NextActiveNode(); void RemoveActiveNode(TFileNode *); TFileStat *GetNextUnAlloc(TFileNode *node = 0); TFileStat *GetNextActive(); void RemoveActive(TFileStat *file); void Reset(); void ValidateFiles(TDSet *dset, TList *slaves); public: static Int_t fgMaxSlaveCnt; // maximum number of slaves per filenode static Int_t fgPacketAsAFraction;// used to calculate the packet size // fPacketSize = fTotalEntries / (fPacketAsAFraction * nslaves) // fPacketAsAFraction can be interpreted as follows: // assuming all slaves have equal processing rate, packet size // is (#events processed by 1 slave) / fPacketSizeAsAFraction. // It can be set with PROOF_PacketAsAFraction in input list. static Double_t fgMinPacketTime; // minimum packet time static Int_t fgNetworkFasterThanHD; // 1 if network faster than hard disk TPacketizerAdaptive(TDSet *dset, TList *slaves, Long64_t first, Long64_t num, TList *input); virtual ~TPacketizerAdaptive(); Long64_t GetEntriesProcessed(TSlave *sl) const; Int_t GetEstEntriesProcessed(Float_t t, Long64_t &ent, Long64_t &bytes); Int_t CalculatePacketSize(TObject *slstat); TDSetElement *GetNextPacket(TSlave *sl, TMessage *r); ClassDef(TPacketizerAdaptive,0) //Generate work packets for parallel processing }; #endif