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:
| TDSetElement* | TVirtualPacketizer::CreateNewPacket(TDSetElement* base, Long64_t first, Long64_t num) |
| 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) |
| virtual Bool_t | TVirtualPacketizer::HandleTimer(TTimer* timer) |
| void | TObject::MakeZombie() |
private:
| TPacketizerAdaptive() | |
| TPacketizerAdaptive(const TPacketizerAdaptive&) | |
| TPacketizerAdaptive::TFileStat* | GetNextActive() |
| TPacketizerAdaptive::TFileStat* | GetNextUnAlloc(TPacketizerAdaptive::TFileNode* node = 0, const char* nodeHostName = 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, Long64_t maxent = -1, Bool_t byfile = kFALSE) |
Data Members
public:
| enum TVirtualPacketizer::EUseEstOpt { | kEstOff | |
| kEstCurrent | ||
| kEstAverage | ||
| }; | ||
| enum TVirtualPacketizer::EStatusBits { | kIsInitializing | |
| kIsDone | ||
| kIsTree | ||
| }; | ||
| enum TObject::EStatusBits { | kCanDelete | |
| kMustCleanup | ||
| kObjInCanvas | ||
| kIsReferenced | ||
| kHasUUID | ||
| kCannotPick | ||
| kNoContextMenu | ||
| kInvalidObject | ||
| }; | ||
| enum TObject::[unnamed] { | kIsOnHeap | |
| kNotDeleted | ||
| kZombie | ||
| kBitMask | ||
| kSingleKey | ||
| kOverwrite | ||
| kWriteDelete | ||
| }; |
protected:
private:
| TList* | fActive | nodes with unfinished files |
| Float_t | fBaseLocalPreference | indicates how much more likely the nodes will be |
| Bool_t | fCachePacketSync | control synchronization of cache and packet sizes |
| TList* | fFileNodes | nodes with files |
| TSortedList* | fFilesToProcess | Global list of files (TFileStat) to be processed |
| Bool_t | fForceLocal | if 1 - eliminate the remote processing |
| Float_t | fFractionOfRemoteFiles | fraction of TDSetElements that are on non-workers |
| Double_t | fMaxEntriesRatio | max file entries to avg allowed ratio for cache-to-packet sync |
| Int_t | fMaxPerfIdx | maximum of our slaves' performance index |
| Long_t | fMaxSlaveCnt | maximum number of workers per filenode (Long_t to avoid |
| Long64_t | fNEventsOnRemLoc | number of events in currently |
| Int_t | fPacketAsAFraction | used to calculate the packet size |
| TList* | fPartitions | list of partitions on nodes |
| Int_t | fStrategy | 0 means the classic and 1 (default) - the adaptive strategy |
| 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, Long64_t maxent = -1, Bool_t byfile = kFALSE)
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 fStrategy
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 current 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, const char* nodeHostName = 0)
TFileStat * GetNextActive()