ROOT logo
ROOT » TREE » TREE » TTreeCacheUnzip

class TTreeCacheUnzip: public TTreeCache

Parallel Unzipping

TTreeCache has been specialised in order to let additional threads
free to unzip in advance its content. In this implementation we
support up to 10 threads, but right now it makes more sense to
limit their number to 1-2

The application reading data is carefully synchronized, in order to:
- if the block it wants is not unzipped, it self-unzips it without
- if the block is being unzipped in parallel, it waits only
for that unzip to finish
- if the block has already been unzipped, it takes it

This is supposed to cancel a part of the unzipping latency, at the
expenses of cpu time.

The default parameters are the same of the prev version, i.e. 20%
of the TTreeCache cache size. To change it use
TTreeCache::SetUnzipBufferSize(Long64_t bufferSize)
where bufferSize must be passed in bytes.

Function Members (Methods)

TTreeCacheUnzip(TTree* tree, Int_t buffersize = 0)
voidTObject::AbstractMethod(const char* method) const
virtual voidAddBranch(TBranch* b, Bool_t subbranches = kFALSE)
virtual voidAddBranch(const char* branch, Bool_t subbranches = kFALSE)
virtual voidTObject::AppendPad(Option_t* option = "")
virtual voidTObject::Browse(TBrowser* b)
static TClass*Class()
virtual const char*TObject::ClassName() const
virtual voidTObject::Clear(Option_t* = "")
virtual TObject*TObject::Clone(const char* newname = "") const
virtual Int_tTObject::Compare(const TObject* obj) const
virtual voidTObject::Copy(TObject& object) const
virtual voidTObject::Delete(Option_t* option = "")MENU
virtual Int_tTObject::DistancetoPrimitive(Int_t px, Int_t py)
virtual voidTObject::Draw(Option_t* option = "")
virtual voidTObject::DrawClass() constMENU
virtual TObject*TObject::DrawClone(Option_t* option = "") constMENU
virtual voidTObject::Dump() constMENU
virtual voidTObject::Error(const char* method, const char* msgfmt) const
virtual voidTObject::Execute(const char* method, const char* params, Int_t* error = 0)
virtual voidTObject::Execute(TMethod* method, TObjArray* params, Int_t* error = 0)
virtual voidTObject::ExecuteEvent(Int_t event, Int_t px, Int_t py)
virtual voidTObject::Fatal(const char* method, const char* msgfmt) const
virtual Bool_tFillBuffer()
virtual TObject*TObject::FindObject(const char* name) const
virtual TObject*TObject::FindObject(const TObject* obj) const
virtual Int_tTFileCacheRead::GetBufferSize() const
virtual Option_t*TObject::GetDrawOption() const
static Long_tTObject::GetDtorOnly()
virtual const char*TObject::GetIconName() const
static Int_tTTreeCache::GetLearnEntries()
virtual const char*TObject::GetName() const
virtual char*TObject::GetObjectInfo(Int_t px, Int_t py) const
static Bool_tTObject::GetObjectStat()
virtual Option_t*TObject::GetOption() const
TTree*TTreeCache::GetOwner() const
static TTreeCacheUnzip::EParUnzipModeGetParallelUnzip()
Int_tGetRecordHeader(char* buf, Int_t maxbytes, Int_t& nbytes, Int_t& objlen, Int_t& keylen)
virtual const char*TObject::GetTitle() const
TTree*TTreeCache::GetTree() const
virtual UInt_tTObject::GetUniqueID() const
virtual Int_tGetUnzipBuffer(char** buf, Long64_t pos, Int_t len, Bool_t* free)
virtual Bool_tTObject::HandleTimer(TTimer* timer)
virtual ULong_tTObject::Hash() const
virtual voidTObject::Info(const char* method, const char* msgfmt) const
virtual Bool_tTObject::InheritsFrom(const char* classname) const
virtual Bool_tTObject::InheritsFrom(const TClass* cl) const
virtual voidTObject::Inspect() constMENU
voidTObject::InvertBit(UInt_t f)
virtual TClass*IsA() const
virtual Bool_tTFileCacheRead::IsAsyncReading() const
virtual Bool_tTObject::IsEqual(const TObject* obj) const
virtual Bool_tTObject::IsFolder() const
virtual Bool_tTTreeCache::IsLearning() const
Bool_tTObject::IsOnHeap() const
static Bool_tIsParallelUnzip()
virtual Bool_tTObject::IsSortable() const
Bool_tTObject::IsZombie() const
virtual voidTObject::ls(Option_t* option = "") const
voidTObject::MayNotUse(const char* method) const
virtual Bool_tTObject::Notify()
static voidTObject::operator delete(void* ptr)
static voidTObject::operator delete(void* ptr, void* vp)
static voidTObject::operator delete[](void* ptr)
static voidTObject::operator delete[](void* ptr, void* vp)
void*TObject::operator new(size_t sz)
void*TObject::operator new(size_t sz, void* vp)
void*TObject::operator new[](size_t sz)
void*TObject::operator new[](size_t sz, void* vp)
virtual voidTObject::Paint(Option_t* option = "")
virtual voidTObject::Pop()
virtual voidTFileCacheRead::Prefetch(Long64_t pos, Int_t len)
virtual voidPrint(Option_t* option = "") const
virtual Int_tTObject::Read(const char* name)
virtual Int_tTTreeCache::ReadBuffer(char* buf, Long64_t pos, Int_t len)
virtual Int_tReadBufferExt(char* buf, Long64_t pos, Int_t len, Int_t& loc)
virtual voidTObject::RecursiveRemove(TObject* obj)
voidTObject::ResetBit(UInt_t f)
virtual voidResetCache()
virtual voidTObject::SaveAs(const char* filename = "", Option_t* option = "") constMENU
virtual voidTObject::SavePrimitive(basic_ostream<char,char_traits<char> >& out, Option_t* option = "")
voidSendUnzipStartSignal(Bool_t broadcast)
voidTObject::SetBit(UInt_t f)
voidTObject::SetBit(UInt_t f, Bool_t set)
virtual voidTObject::SetDrawOption(Option_t* option = "")MENU
static voidTObject::SetDtorOnly(void* obj)
virtual voidSetEntryRange(Long64_t emin, Long64_t emax)
virtual voidTFileCacheRead::SetFile(TFile* file)
static voidTTreeCache::SetLearnEntries(Int_t n = 10)
static voidTObject::SetObjectStat(Bool_t stat)
static Int_tSetParallelUnzip(TTreeCacheUnzip::EParUnzipMode option = TTreeCacheUnzip::kEnable)
virtual voidTFileCacheRead::SetSkipZip(Bool_t = kTRUE)
virtual voidTObject::SetUniqueID(UInt_t uid)
voidSetUnzipBufferSize(Long64_t bufferSize)
static voidSetUnzipRelBufferSize(Float_t relbufferSize)
virtual voidShowMembers(TMemberInspector& insp, char* parent)
virtual voidTFileCacheRead::Sort()
virtual voidStopLearningPhase()
virtual voidStreamer(TBuffer& b)
voidStreamerNVirtual(TBuffer& b)
virtual voidTObject::SysError(const char* method, const char* msgfmt) const
Bool_tTObject::TestBit(UInt_t f) const
Int_tTObject::TestBits(UInt_t f) const
Int_tUnzipBuffer(char** dest, char* src)
Int_tUnzipCache(Int_t& startindex, Int_t& locbuffsz, char*& locbuff)
static void*UnzipLoop(void* arg)
virtual voidUpdateBranches(TTree* tree, Bool_t owner = kFALSE)
virtual voidTObject::UseCurrentStyle()
virtual voidTObject::Warning(const char* method, const char* msgfmt) const
virtual Int_tTObject::Write(const char* name = 0, Int_t option = 0, Int_t bufsize = 0)
virtual Int_tTObject::Write(const char* name = 0, Int_t option = 0, Int_t bufsize = 0) const
virtual voidTObject::DoError(int level, const char* location, const char* fmt, va_list va) const

Data Members

enum EParUnzipMode { kEnable
enum TObject::EStatusBits { kCanDelete
enum TObject::[unnamed] { kIsOnHeap
queue<Int_t>fActiveBlksThe blocks which are active now
Bool_tfActiveThreadUsed to terminate gracefully the unzippers
TList*TTreeCache::fBrNames! list of branch names in the cache
TObjArray*TTreeCache::fBranches! List of branches to be stored in the cache
char*TFileCacheRead::fBuffer[fBufferSize] buffer of contiguous prefetched blocks
Int_tTFileCacheRead::fBufferLenCurrent buffer length (<= fBufferSize)
Int_tTFileCacheRead::fBufferSizeAllocated size of fBuffer (at a given time)
Int_tTFileCacheRead::fBufferSizeMinOriginal size of fBuffer
Long64_tTTreeCache::fEntryCurrent! current lowest entry number in the cache
Long64_tTTreeCache::fEntryMax! last entry in the cache
Long64_tTTreeCache::fEntryMin! first entry in the cache
Long64_tTTreeCache::fEntryNext! next entry number where cache must be filled
TFile*TFileCacheRead::fFilePointer to file
Bool_tTTreeCache::fIsLearning! true if cache is in learning mode
Bool_tTTreeCache::fIsManual! true if cache is StopLearningPhase was used
Bool_tTFileCacheRead::fIsSortedTrue if fSeek array is sorted
Bool_tTFileCacheRead::fIsTransferredTrue when fBuffer contains something valid
Int_t*TFileCacheRead::fLen[fNb] Length of long buffers
TMutex*fMutexListMutex to protect the various lists. Used by the condvars.
Int_tfNFound! number of blocks that were found in the cache
Int_tfNMissed! number of blocks that were not found in the cache and were unzipped
Int_tTTreeCache::fNReadMissNumber of blocks read and not found in the chache
Int_tTTreeCache::fNReadOkNumber of blocks read and found in the cache
Int_tTTreeCache::fNReadPrefNumber of blocks that were prefetched
Int_tfNStalls! number of hits which caused a stall
Int_tfNUnzip! number of blocks that were unzipped
Int_tTFileCacheRead::fNbNumber of long buffers
Int_tTTreeCache::fNbranches! Number of branches in the cache
Int_tTFileCacheRead::fNseekNumber of blocks to be prefetched
Int_tfNseekMax! fNseek can change so we need to know its max size
Int_tTFileCacheRead::fNtotTotal size of prefetched blocks
TTree*TTreeCache::fOwner! pointer to the owner Tree/chain
Bool_tfParallelIndicate if we want to activate the parallelism (for this instance)
Long64_t*TFileCacheRead::fPos[fNb] start of long buffers
Long64_t*TFileCacheRead::fSeek[fNseek] Position on file of buffers to be prefetched
Int_t*TFileCacheRead::fSeekIndex[fNseek] sorted index table of fSeek
Int_t*TFileCacheRead::fSeekLen[fNseek] Length of buffers to be prefetched
Int_t*TFileCacheRead::fSeekPos[fNseek] Position of sorted blocks in fBuffer
Int_tTFileCacheRead::fSeekSizeAllocated size of fSeek
Long64_t*TFileCacheRead::fSeekSort[fNseek] Position on file of buffers to be prefetched (sorted)
Int_t*TFileCacheRead::fSeekSortLen[fNseek] Length of buffers to be prefetched (sorted)
Long64_tfTotalUnzipBytes! The total sum of the currently unzipped blks
TTree*TTreeCache::fTree! pointer to the current Tree
Long64_tfUnzipBufferSize! Max Size for the ready unzipped blocks (default is 2*fBufferSize)
char**fUnzipChunks! [fNseek] Individual unzipped chunks. Their summed size is kept under control.
TCondition*fUnzipDoneConditionUsed to wait for an unzip tour to finish. Gives the Async feel.
Int_t*fUnzipLen! [fNseek] Length of the unzipped buffers
TCondition*fUnzipStartConditionUsed to signal the threads to start.
Byte_t*fUnzipStatus! [fNSeek] For each blk, tells us if it's unzipped or pending
Long64_tTTreeCache::fZipBytes! Total compressed size of branches in cache
static Int_tTTreeCache::fgLearnEntriesNumber of entries used for learning mode
static TTreeCacheUnzip::EParUnzipModefgParallelIndicate if we want to activate the parallelism
static Double_tfgRelBuffSizeThis is the percentage of the TTreeCacheUnzip that will be used

Class Charts

Inheritance Inherited Members Includes Libraries
Class Charts

Function documentation

 Default Constructor.
TTreeCacheUnzip(TTree* tree, Int_t buffersize = 0)
void Init()
 Initialization procedure common to all the constructors
 destructor. (in general called by the TFile destructor
 destructor. (in general called by the TFile destructor)
void AddBranch(TBranch *b, Bool_t subbranches /*= kFALSE*/)
add a branch to the list of branches to be stored in the cache
this function is called by TBranch::GetBasket
void AddBranch(const char *branch, Bool_t subbranches /*= kFALSE*/)
add a branch to the list of branches to be stored in the cache
this function is called by TBranch::GetBasket
Bool_t FillBuffer()
void SetEntryRange(Long64_t emin, Long64_t emax)
 Set the minimum and maximum entry number to be processed
 this information helps to optimize the number of baskets to read
 when prefetching the branch buffers.
void StopLearningPhase()
 It's the same as TTreeCache::StopLearningPhase but we guarantee that
 we start the unzipping just after getting the buffers
void UpdateBranches(TTree* tree, Bool_t owner = kFALSE)
update pointer to current Tree and recompute pointers to the branches in the cache
Bool_t IsParallelUnzip()
 Static function that tells wether the multithreading unzipping
 is activated
Bool_t IsActiveThread()
 This indicates if the thread is active in this moment...
 this variable is very important because if we change it from true to
 false the thread will stop... ( see StopThreadTreeCacheUnzip() )
Bool_t IsQueueEmpty()
 It says if the queue is empty... useful to see if we have to process
void WaitUnzipStartSignal()
 Here the threads sleep waiting for some blocks to unzip
void SendUnzipStartSignal(Bool_t broadcast)
 This will send the signal corresponfing to the queue... normally used
 when we want to start processing the list of buffers.
Int_t SetParallelUnzip(TTreeCacheUnzip::EParUnzipMode option = TTreeCacheUnzip::kEnable)
 Static function that(de)activates multithreading unzipping
 The possible options are:
 kEnable _Enable_ it, which causes an automatic detection and launches the
 additional thread if the number of cores in the machine is greater than one
 kDisable _Disable_ will not activate the additional thread.
 kForce _Force_ will start the additional thread even if there is only one core.
 the default will be taken as kEnable.
 returns 0 if there was an error, 1 otherwise.
Int_t StartThreadUnzip(Int_t nthreads)
 The Thread is only a part of the TTreeCache but it is the part that
 waits for info in the queue and process it... unfortunatly, a Thread is
 not an object an we have to deal with it in the old C-Style way
 Returns 0 if the thread was initialized or 1 if it was already running
Int_t StopThreadUnzip()
 To stop the thread we only need to change the value of the variable
 fActiveThread to false and the loop will stop (of course, we will have)
 to do the cleaning after that.
 Note: The syncronization part is important here or we will try to delete
       teh object while it's still processing the queue
void* UnzipLoop(void* arg)
 This is a static function.
 This is the call that will be executed in the Thread generated by
 StartThreadTreeCacheUnzip... what we want to do is to inflate the next
 series of buffers leaving them in the second cache.
 Returns 0 when it finishes
Int_t GetRecordHeader(char* buf, Int_t maxbytes, Int_t& nbytes, Int_t& objlen, Int_t& keylen)
 Read the logical record header from the buffer buf.
 That must be the pointer tho the header part not the object by itself and
 must contain data of at least maxbytes
 Returns nread;
 In output arguments:
    nbytes : number of bytes in record
             if negative, this is a deleted record
             if 0, cannot read record, wrong value of argument first
    objlen : uncompressed object size
    keylen : length of logical record header
 Note that the arguments objlen and keylen are returned only
 if maxbytes >=16
 Note: This was adapted from TFile... so some things dont apply
void ResetCache()
 This will delete the list of buffers that are in the unzipping cache
 and will reset certain values in the cache.
 This name is ambiguos because the method doesn't reset the whole cache,
 only the part related to the unzipping
 Note: This method is completely different from TTreeCache::ResetCache(),
 in that method we were cleaning the prefetching buffer while here we
 delete the information about the unzipped buffers
Int_t GetUnzipBuffer(char** buf, Long64_t pos, Int_t len, Bool_t* free)
 We try to read a buffer that has already been unzipped
 Returns -1 in case of read failure, 0 in case it's not in the
 cache and n>0 in case read from cache (number of bytes copied).
 pos and len are the original values as were passed to ReadBuffer
 but instead we will return the inflated buffer.
 Note!! : If *buf == 0 we will allocate the buffer and it will be the
 responsability of the caller to free it... it is useful for example
 to pass it to the creator of TBuffer
void SetUnzipRelBufferSize(Float_t relbufferSize)
 static function: Sets the unzip relatibe buffer size
 FABRIZIO: PLEASE DOCUMENT and also in TTree::Set...
void SetUnzipBufferSize(Long64_t bufferSize)
 Sets the size for the unzipping cache... by default it should be
 two times the size of the prefetching cache
Int_t UnzipBuffer(char** dest, char* src)
 UNzips a ROOT specific buffer... by reading the header at the beginning.
 returns the size of the inflated buffer or -1 if error
 Note!! : If *dest == 0 we will allocate the buffer and it will be the
 responsability of the caller to free it... it is useful for example
 to pass it to the creator of TBuffer
 src is the original buffer with the record (header+compressed data)
 *dest is the inflated buffer (including the header)
Int_t UnzipCache(Int_t& startindex, Int_t& locbuffsz, char*& locbuff)
 This inflates all the buffers in the cache.. passing the data to a new
 buffer that will only wait there to be read...
 We can not inflate all the buffers in the cache so we will try to do
 it until the cache gets full... there is a member called fUnzipBufferSize which will
 tell us the max size we can allocate for this cache.

 note that we will  unzip in the order they were put into the cache not
 the order of the transference so it has to be read in that order or the
 pre-unzipping will be useless.

 startindex is used as start index to check for blks to be unzipped

 returns 0 in normal conditions or -1 if error, 1 if it would like to sleep

 This func is supposed to compete among an indefinite number of threads to get a chunk to inflate
 in order to accommodate multiple unzippers
 Since everything is so async, we cannot use a fixed buffer, we are forced to keep
 the individual chunks as separate blocks, whose summed size does not exceed the maximum
 allowed. The pointers are kept globally in the array fUnzipChunks
void Print(Option_t* option = "") const
Int_t ReadBufferExt(char* buf, Long64_t pos, Int_t len, Int_t& loc)
TTreeCacheUnzip(const TTreeCacheUnzip& )
TTreeCacheUnzip& operator=(const TTreeCacheUnzip& )
EParUnzipMode GetParallelUnzip()
 Methods related to the thread
Int_t GetNUnzip()
 Methods to get stats
{ return fNUnzip; }
Int_t GetNFound()
{ return fNFound; }
Int_t GetNMissed()
{ return fNMissed; }