27 inline int xtod(
char c) {
return (c>=
'0' && c<=
'9') ? c-
'0' : ((c>=
'A' && c<=
'F') ? c-
'A'+10 : ((c>=
'a' && c<=
'f') ? c-
'a'+10 : 0)); }
58 fPendingBlocks =
new TList();
59 fReadBlocks =
new TList();
61 fPendingBlocks->SetOwner();
62 fReadBlocks->SetOwner();
90 std::lock_guard<std::mutex> lk(fMutexPendingList);
91 fPrefetchFinished =
kTRUE;
93 fNewBlockAdded.notify_one();
96 fThreadJoined =
kTRUE;
97 fPrefetchFinished =
kFALSE;
108 if (CheckBlockInCache(path, block)){
114 if (fFile->GetArchive()) {
116 block->
SetPos(i, block->
GetPos(i) - fFile->GetArchiveOffset());
131 while((block = GetPendingBlock())){
132 ReadAsync(block, inCache);
135 SaveBlockInCache(block);
147 while (first <= last){
148 mid = first + (last - first) / 2;
155 else if (blockObj->
GetPos(
mid) < offset){
170 return Long64_t(fWaitTime.RealTime()*1.e+6);
182 std::unique_lock<std::mutex> lk(fMutexReadList);
184 TIter iter(fReadBlocks);
187 if (BinarySearchReadList(blockObj, offset, len, &index)){
196 fReadBlockAdded.wait(lk);
203 pBuff += (offset - blockObj->
GetPos(index));
204 memcpy(buf, pBuff, len);
214 TFPBlock* block = CreateBlockObj(offset, len, nblock);
215 AddPendingBlock(block);
223 fMutexPendingList.lock();
224 fPendingBlocks->Add(block);
225 fMutexPendingList.unlock();
227 fNewBlockAdded.notify_one();
239 fSemChangeFile->Post();
240 std::unique_lock<std::mutex> lk(fMutexPendingList);
242 fNewBlockAdded.wait(lk, [&]{
return fPendingBlocks->GetSize() > 0 || fPrefetchFinished; });
244 fSemChangeFile->Wait();
247 if (fPendingBlocks->GetSize()){
248 block = (
TFPBlock*)fPendingBlocks->First();
249 block = (
TFPBlock*)fPendingBlocks->Remove(block);
259 fMutexReadList.lock();
263 movedBlock = (
TFPBlock*)fReadBlocks->Remove(movedBlock);
268 fReadBlocks->Add(block);
269 fMutexReadList.unlock();
272 fReadBlockAdded.notify_one();
283 fMutexReadList.lock();
286 blockObj =
static_cast<TFPBlock*
>(fReadBlocks->First());
287 fReadBlocks->Remove(blockObj);
288 fMutexReadList.unlock();
292 fMutexReadList.unlock();
293 blockObj =
new TFPBlock(offset, len, noblock);
318 if (!fThreadJoined) {
319 fSemChangeFile->Wait();
324 fMutexPendingList.lock();
325 fPendingBlocks->Clear();
326 fMutexPendingList.unlock();
328 fMutexReadList.lock();
329 fReadBlocks->Clear();
330 fMutexReadList.unlock();
334 if (!fThreadJoined) {
335 fSemChangeFile->Post();
339 assert((fFile == file) &&
"kDoNotDisconnect must reattach to the same file");
352 rc = fConsumer->Run();
382 const char* ptr = hex;
385 result +=
xtod(ptr[i]);
395 if (fPathCache ==
"")
399 TString fullPath(fPathCache);
411 concatStr.Form(
"%lld", block->
GetPos(i));
417 value = SumHex(fileName);
420 dirName.Form(
"%i", value);
422 fullPath +=
"/" + dirName +
"/" + fileName;
426 path =
new char[fullPath.Length() + 1];
427 strlcpy(path, fullPath,fullPath.Length() + 1);
442 TString strPath = path;
444 strPath +=
"?filetype=raw";
445 TFile*
file =
new TFile(strPath);
450 buffer = (
char*)
calloc(length,
sizeof(
char));
451 file->ReadBuffer(buffer, 0, length);
453 fFile->fBytesRead += length;
454 fFile->fgBytesRead += length;
455 fFile->SetReadCalls(fFile->GetReadCalls() + 1);
456 fFile->fgReadCalls++;
461 gPerfStats->FileReadEvent(fFile, length, start);
474 if (fPathCache ==
"")
482 concatStr.Form(
"%lld", block->
GetPos(i));
488 Int_t value = SumHex(fileName);
491 TString fullPath( fPathCache );
493 dirName.Form(
"%i", value);
494 fullPath += (
"/" + dirName);
500 fullPath += (
"/" + fileName);
503 fullPath +=
"?filetype=raw";
506 fullPath +=
"?filetype=raw";
void AddPendingBlock(TFPBlock *)
Safe method to add a block to the pendingList.
void *(* VoidRtnFunc_t)(void *)
void ReadAsync(TFPBlock *, Bool_t &)
Read one block and insert it in prefetchBuffers list.
Bool_t BinarySearchReadList(TFPBlock *, Long64_t, Int_t, Int_t *)
Search for a requested element in a block and return the index.
you should not use this method at all Int_t Int_t Double_t Double_t Double_t Int_t mid
void ReallocBlock(Long64_t *, Int_t *, Int_t)
Reallocate the block's buffer based on the length of the elements it will contain.
Bool_t CheckBlockInCache(char *&, TFPBlock *)
Test if the block is in cache.
void Final()
MD5 finalization, ends an MD5 message-digest operation, writing the the message digest and zeroizing ...
Bool_t IsPrefetchFinished() const
static TThread::VoidRtnFunc_t ThreadProc(void *)
Execution loop of the consumer thread.
int GetPathInfo(const char *path, Long_t *id, Long_t *size, Long_t *flags, Long_t *modtime)
Get info about a file: id, size, flags, modification time.
static const int kMAX_READ_SIZE
Long64_t GetWaitTime()
Return the time spent wating for buffer to be read in microseconds.
const char * AsString() const
Return message digest as string.
Int_t GetNoElem() const
Return number of elements in the block.
char * GetPtrToPiece(Int_t index) const
Get block buffer.
Int_t ThreadStart()
Used to start the consumer thread.
virtual int mkdir(const char *name, Bool_t recursive=kFALSE)
Make a file system directory.
char * GetBuffer() const
Get block buffer.
void SaveBlockInCache(TFPBlock *)
Save the block content in cache.
static TFile * Open(const char *name, Option_t *option="", const char *ftitle="", Int_t compress=1, Int_t netopt=0)
Create / open a file.
void SetPos(Int_t, Long64_t)
Set pos value for index idx.
This code implements the MD5 message-digest algorithm.
R__EXTERN TVirtualMonitoringWriter * gMonitoringWriter
void SetBuffer(char *)
Set block buffer.
This class represents the encapsulation of a block request.
void AddReadBlock(TFPBlock *)
Safe method to add a block to the readList.
void Update(const UChar_t *buf, UInt_t len)
Update TMD5 object to reflect the concatenation of another buffer full of bytes.
R__EXTERN TSystem * gSystem
void SetFile(TFile *file, TFile::ECacheAction action=TFile::kDisconnect)
Change the file.
void WaitFinishPrefetch()
Killing the async prefetching thread.
Bool_t ReadBuffer(char *, Long64_t, Int_t)
Return a prefetched element.
Long64_t GetPos(Int_t) const
Get position of the element at index i.
Int_t SumHex(const char *)
Sum up individual hex values to obtain a decimal value.
Long64_t GetDataSize() const
Return size of the data in the block.
void ReadListOfBlocks()
Get blocks specified in prefetchBlocks.
virtual ~TFilePrefetch()
Destructor.
TThread * GetThread() const
Return reference to the consumer thread.
The TTimeStamp encapsulates seconds and ns since EPOCH.
The prefetching mechanism uses two classes (TFilePrefetch and TFPBlock) to prefetch in advance a bloc...
Int_t GetLen(Int_t) const
Get length of the element at index i.
virtual Bool_t SendFileReadProgress(TFile *)
char * GetBlockFromCache(const char *, Int_t)
Return a buffer from cache.
Bool_t SetCache(const char *)
Set the path of the cache directory.
void ReadBlock(Long64_t *, Int_t *, Int_t)
Create a TFPBlock object or recycle one and add it to the prefetchBlocks list.
virtual void * OpenDirectory(const char *name)
Open a directory. Returns 0 if directory does not exist.
TFPBlock * GetPendingBlock()
Safe method to remove a block from the pendingList.
TFPBlock * CreateBlockObj(Long64_t *, Int_t *, Int_t)
Create a new block or recycle an old one.
ECacheAction
TTreeCache flushing semantics.