class TPacketizerAdaptive: public TVirtualPacketizer
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.
Function Members (Methods)
public:
protected:
| virtual void | TObject::DoError(int level, const char* location, const char* fmt, va_list va) const |
| Long64_t | TVirtualPacketizer::GetEntries(Bool_t tree, TDSetElement* e) |
| void | TObject::MakeZombie() |
private:
| TPacketizerAdaptive() | |
| TPacketizerAdaptive(const TPacketizerAdaptive&) | |
| virtual Int_t | AddProcessed(TSlave* sl, TProofProgressStatus* st, Double_t latency, TList** listOfMissingFiles = 0) |
| TPacketizerAdaptive::TFileStat* | GetNextActive() |
| TPacketizerAdaptive::TFileStat* | GetNextUnAlloc(TPacketizerAdaptive::TFileNode* node = 0) |
| void | InitStats() |
| TPacketizerAdaptive::TFileNode* | NextActiveNode() |
| TPacketizerAdaptive::TFileNode* | NextNode() |
| void | operator=(const TPacketizerAdaptive&) |
| Int_t | ReassignPacket(TDSetElement* e, TList** listOfMissingFiles) |
| void | RemoveActive(TPacketizerAdaptive::TFileStat* file) |
| void | RemoveActiveNode(TPacketizerAdaptive::TFileNode*) |
| void | RemoveUnAllocNode(TPacketizerAdaptive::TFileNode*) |
| void | Reset() |
| void | SplitPerHost(TList* elements, TList** listOfMissingFiles) |
| void | ValidateFiles(TDSet* dset, TList* slaves) |
Data Members
public:
| enum TVirtualPacketizer::EUseEstOpt { | kEstOff | |
| kEstCurrent | ||
| kEstAverage | ||
| }; | ||
| enum TVirtualPacketizer::EStatusBits { | kIsInitializing | |
| kIsDone | ||
| }; | ||
| enum TObject::EStatusBits { | kCanDelete | |
| kMustCleanup | ||
| kObjInCanvas | ||
| kIsReferenced | ||
| kHasUUID | ||
| kCannotPick | ||
| kNoContextMenu | ||
| kInvalidObject | ||
| }; | ||
| enum TObject::[unnamed] { | kIsOnHeap | |
| kNotDeleted | ||
| kZombie | ||
| kBitMask | ||
| kSingleKey | ||
| kOverwrite | ||
| kWriteDelete | ||
| }; |
public:
| static Double_t | fgMaxPacketTime | maximum packet time |
| static Long_t | fgMaxSlaveCnt | maximum number of workers per filenode (Long_t to avoid |
| static Double_t | fgMinPacketTime | minimum packet time |
| static Int_t | fgPacketAsAFraction | used to calculate the packet size |
| static Int_t | fgStrategy | 0 means the classic and 1 (default) - the adaptive strategy |
protected:
| Bool_t | TVirtualPacketizer::fStop | Termination of Process() requested? |
| Bool_t | TVirtualPacketizer::fValid | Constructed properly? |
private:
| TList* | fActive | nodes with unfinished files |
| Float_t | fBaseLocalPreference | indicates how much more likely |
| TList* | fFileNodes | nodes with files |
| Bool_t | fForceLocal | if 1 - eliminate the remote processing |
| Float_t | fFractionOfRemoteFiles | fraction of TDSetElements |
| Int_t | fMaxPerfIdx | maximum of our slaves' performance index |
| Long64_t | fNEventsOnRemLoc | number of events in currently |
| TMap* | fSlaveStats | slave status, keyed by correspondig TSlave |
| TList* | fUnAllocated | nodes with unallocated files |
Class Charts
Function documentation
TPacketizerAdaptive(TDSet* dset, TList* slaves, Long64_t first, Long64_t num, TList* input, TProofProgressStatus* st)
Constructor
void ValidateFiles(TDSet* dset, TList* slaves)
Check existence of file/dir/tree an get number of entries. Assumes the files have been setup.
Int_t CalculatePacketSize(TObject* slstat, Long64_t cachesz, Int_t learnent)
The result depends on the fgStrategy
Int_t AddProcessed(TSlave* sl, TProofProgressStatus* st, Double_t latency, TList** listOfMissingFiles = 0)
To be used by GetNextPacket but also in reaction to kPROOF_STOPPROCESS message (when the worker was asked to stop processing during a packet). returns the #entries intended in the last packet - #processed entries
TDSetElement * GetNextPacket(TSlave* sl, TMessage* r)
Get next packet; A meaningfull difference to TPacketizer is the fact that this packetizer, for each worker, tries to predict whether the worker will finish processing it's local files before the end of the query. If yes, it allocates, to those workers, files from non-slave filenodes or from slaves that are overloaded. The check is done every time a new file needs to be assigned.
Float_t GetCurrentRate(Bool_t& all)
Get Estimation of the current rate; just summing the current rates of the active workers
Int_t GetEstEntriesProcessed(Float_t , Long64_t& ent, Long64_t& bytes, Long64_t& calls)
Get estimation for the number of processed entries and bytes read at time t, based on the numbers already processed and the latests worker measured speeds. If t <= 0 the cutrent time is used. Only the estimation for the entries is currently implemented. This is needed to smooth the instantaneous rate plot.
void MarkBad(TSlave* s, TProofProgressStatus* status, TList** missingFiles)
This method can be called at any time during processing as an effect of handling kPROOF_STOPPROCESS If the output list from this worker is going to be sent back to the master, the 'status' includes the number of entries processed by the slave. From this we calculate the remaining part of the packet. 0 indicates that the results from that worker were lost completely. Assume that the filenodes for which we have a TFileNode object are still up and running.
Int_t ReassignPacket(TDSetElement* e, TList** listOfMissingFiles)
The file in the listOfMissingFiles can appear several times; in order to fix that, a TDSetElement::Merge method is needed.
void SplitPerHost(TList* elements, TList** listOfMissingFiles)
Split into per host entries The files in the listOfMissingFiles can appear several times; in order to fix that, a TDSetElement::Merge method is needed.
TPacketizerAdaptive(const TPacketizerAdaptive& )
void operator=(const TPacketizerAdaptive& )
TFileNode * NextNode()
TFileNode * NextActiveNode()
TFileStat * GetNextUnAlloc(TPacketizerAdaptive::TFileNode* node = 0)
TFileStat * GetNextActive()