include/root/TFilePrefetch.h

0001 // @(#)root/io:$Id$
0002 // Author: Elvin Sindrilaru   19/05/2011
0003
0004 /*************************************************************************
0005  * Copyright (C) 1995-2011, Rene Brun and Fons Rademakers.               *
0006  * All rights reserved.                                                  *
0007  *                                                                       *
0008  * For the licensing terms see $ROOTSYS/LICENSE.                         *
0009  * For the list of contributors see $ROOTSYS/README/CREDITS.             *
0010  *************************************************************************/
0011
0012 #ifndef ROOT_TFilePrefetch
0013 #define ROOT_TFilePrefetch
0014
0015 #include "TObject.h"
0016 #include "TString.h"
0017 #include "TStopwatch.h"
0018 #include "TThread.h"
0019 #include "TFile.h"
0020
0021 #include <atomic>
0022 #include <condition_variable>
0023 #include <mutex>
0024
0025 #ifdef R__LESS_INCLUDES
0026 class TSemaphore;
0027 class TFPBlock;
0028 #else
0029 #include "TSemaphore.h"
0030 #include "TFPBlock.h"
0031 #endif
0032
0033 class TFilePrefetch : public TObject {
0034
0035 private:
0036    TFile      *fFile;                       ///< reference to the file
0037    TList      *fPendingBlocks;              ///< list of pending blocks to be read
0038    TList      *fReadBlocks;                 ///< list of blocks read
0039    TThread    *fConsumer;                   ///< consumer thread
0040    std::mutex fMutexPendingList;            ///< mutex for the pending list
0041    std::mutex fMutexReadList;               ///< mutex for the list of read blocks
0042    std::condition_variable fNewBlockAdded;  ///< signal the addition of a new pending block
0043    std::condition_variable fReadBlockAdded; ///< signal the addition of a new red block
0044    TSemaphore *fSemChangeFile;              ///< semaphore used when changing a file in TChain
0045    TString     fPathCache;                  ///< path to the cache directory
0046    TStopwatch  fWaitTime;                   ///< time waiting to prefetch a buffer (in usec)
0047    Bool_t      fThreadJoined;               ///< mark if async thread was joined
0048    std::atomic<Bool_t> fPrefetchFinished;   ///< true if prefetching is over
0049
0050    static TThread::VoidRtnFunc_t ThreadProc(void*);  //create a joinable worker thread
0051
0052 public:
0053    TFilePrefetch(TFile*);
0054    ~TFilePrefetch() override;
0055
0056    void      ReadAsync(TFPBlock*, Bool_t&);
0057    void      ReadListOfBlocks();
0058
0059    void      AddPendingBlock(TFPBlock*);
0060    TFPBlock *GetPendingBlock();
0061
0062    void      AddReadBlock(TFPBlock*);
0063    Bool_t    ReadBuffer(char*, Long64_t, Int_t);
0064    void      ReadBlock(Long64_t*, Int_t*, Int_t);
0065    TFPBlock *CreateBlockObj(Long64_t*, Int_t*, Int_t);
0066
0067    TThread  *GetThread() const;
0068    Int_t     ThreadStart();
0069
0070    Bool_t    SetCache(const char*);
0071    Bool_t    CheckBlockInCache(char*&, TFPBlock*);
0072    char     *GetBlockFromCache(const char*, Int_t);
0073    void      SaveBlockInCache(TFPBlock*);
0074
0075    Int_t     SumHex(const char*);
0076    Bool_t    BinarySearchReadList(TFPBlock*, Long64_t, Int_t, Int_t*);
0077    Long64_t  GetWaitTime();
0078
0079    void      SetFile(TFile* file, TFile::ECacheAction action = TFile::kDisconnect);
0080    std::condition_variable &GetCondNewBlock() { return fNewBlockAdded; };
0081    void      WaitFinishPrefetch();
0082    Bool_t    IsPrefetchFinished() const { return fPrefetchFinished; }
0083
0084    ClassDefOverride(TFilePrefetch, 0);  // File block prefetcher
0085 };
0086
0087 #endif