28#include <XrdCl/XrdClURL.hh>
29#include <XrdCl/XrdClFile.hh>
30#include <XrdCl/XrdClXRootDResponses.hh>
31#include <XrdCl/XrdClDefaultEnv.hh>
32#include <XrdVersion.hh>
39class TAsyncOpenHandler:
public XrdCl::ResponseHandler
56 virtual void HandleResponse(XrdCl::XRootDStatus *status,
57 XrdCl::AnyObject *response)
81class TAsyncReadvHandler:
public XrdCl::ResponseHandler
88 TAsyncReadvHandler(std::vector<XrdCl::XRootDStatus*> *statuses,
91 fStatuses(statuses), fStatusIndex(statusIndex), fSemaphore(semaphore) {}
98 virtual void HandleResponse(XrdCl::XRootDStatus *status,
99 XrdCl::AnyObject *response)
101 fStatuses->at(fStatusIndex) = status;
108 std::vector<XrdCl::XRootDStatus*> *fStatuses;
132 TNetXNGFile(url,0,mode,title,compress,netopt,parallelopen){}
141 TFile((lurl ? lurl : url),
"NET", title, compress)
143 using namespace XrdCl;
148 if (!val.
IsNull()) XrdCl::DefaultEnv::SetLogLevel(val.
Data());
154 TUrl urlnoanchor(url);
156 fUrl =
new URL(std::string(urlnoanchor.
GetUrl()));
161 fUrl->SetProtocol(std::string(
"root"));
167 Error(
"Open",
"could not parse open mode %s", mode);
188 TAsyncOpenHandler *handler =
new TAsyncOpenHandler(
this);
189 status =
fFile->Open(
fUrl->GetURL(),
fMode, Access::None, handler);
190 if (!status.IsOK()) {
191 Error(
"Open",
"%s", status.ToStr().c_str());
199 if (!status.IsOK()) {
200#if XrdVNUMBER >= 40000
201 if( status.code == errRedirect )
202 fNewUrl = status.GetErrorMessage().c_str();
204 Error(
"Open",
"%s", status.ToStr().c_str());
206 Error(
"Open",
"%s", status.ToStr().c_str());
212 if( (
fMode & OpenFlags::New) || (
fMode & OpenFlags::Delete) ||
213 (
fMode & OpenFlags::Update) )
218 if( (
fMode & OpenFlags::New) || (
fMode & OpenFlags::Delete) )
244 using namespace XrdCl;
247 if (
gDebug > 1)
Info(
"Init",
"TFile::Init already called once");
283 using namespace XrdCl;
289 bool forceStat =
true;
290 if(
fMode == XrdCl::OpenFlags::Read )
294 if( !
fFile->Stat( forceStat, info ).IsOK() )
306 return fFile->IsOpen();
329 XrdCl::XRootDStatus status =
fFile->Close();
330 if (!status.IsOK()) {
331 Error(
"Close",
"%s", status.ToStr().c_str());
347 using namespace XrdCl;
349 OpenFlags::Flags mode;
354 if (parseres<0 || (mode != OpenFlags::Read && mode != OpenFlags::Update)) {
355 Error(
"ReOpen",
"mode must be either READ or UPDATE, not %s", modestr);
360 if (mode ==
fMode || (mode == OpenFlags::Update
361 &&
fMode == OpenFlags::New)) {
365 XRootDStatus st =
fFile->Close();
367 Error(
"ReOpen",
"%s", st.ToStr().c_str());
375 Error(
"ReOpen",
"%s", st.ToStr().c_str());
404 using namespace XrdCl;
406 Info(
"ReadBuffer",
"offset: %lld length: %d", position, length);
425 uint32_t bytesRead = 0;
426 XRootDStatus st =
fFile->Read(
fOffset, length, buffer, bytesRead);
428 Info(
"ReadBuffer",
"%s bytes read: %u", st.ToStr().c_str(), bytesRead);
431 Error(
"ReadBuffer",
"%s", st.ToStr().c_str());
435 if ((
Int_t)bytesRead != length) {
436 Error(
"ReadBuffer",
"error reading all requested bytes, got %u of %d",
472 using namespace XrdCl;
478 std::vector<ChunkList> chunkLists;
480 std::vector<XRootDStatus*> *statuses;
482 Int_t totalBytes = 0;
484 char *cursor = buffer;
490 for (
Int_t i = 0; i < nbuffs; i++)
494 for (
Int_t i = 0; i < nbuffs; ++i) {
495 totalBytes += length[i];
504 for (j = 0; j < nsplit; ++j) {
506 chunks.push_back(ChunkInfo(offset,
fReadvIorMax, cursor));
512 chunks.push_back(ChunkInfo(offset, rem, cursor));
515 chunks.push_back(ChunkInfo(position[i], length[i], cursor));
521 chunkLists.push_back(chunks);
522 chunks = ChunkList();
524 chunkLists.push_back(ChunkList(chunks.begin(),
526 chunks = ChunkList(chunks.begin() +
fReadvIovMax, chunks.end());
531 if( !chunks.empty() )
532 chunkLists.push_back(chunks);
534 TAsyncReadvHandler *handler;
537 statuses =
new std::vector<XRootDStatus*>(chunkLists.size());
540 std::vector<ChunkList>::iterator it;
541 for (it = chunkLists.begin(); it != chunkLists.end(); ++it)
543 handler =
new TAsyncReadvHandler(statuses, it - chunkLists.begin(),
545 status =
fFile->VectorRead(*it, 0, handler);
547 if (!status.IsOK()) {
548 Error(
"ReadBuffers",
"%s", status.ToStr().c_str());
554 for (it = chunkLists.begin(); it != chunkLists.end(); ++it) {
559 for (it = chunkLists.begin(); it != chunkLists.end(); ++it) {
560 XRootDStatus *st = statuses->at(it - chunkLists.begin());
563 Error(
"ReadBuffers",
"%s", st->ToStr().c_str());
564 for( ; it != chunkLists.end(); ++it )
566 st = statuses->at( it - chunkLists.begin() );
585 gPerfStats->FileReadEvent(
this, totalBytes, start);
605 using namespace XrdCl;
613 Info(
"WriteBuffer",
"file not writable");
626 XRootDStatus st =
fFile->Write(
fOffset, length, buffer);
628 Error(
"WriteBuffer",
"%s", st.ToStr().c_str());
649 Info(
"Flush",
"file not writable - do nothing");
657 XrdCl::XRootDStatus status =
fFile->Sync();
659 Error(
"Flush",
"%s", status.ToStr().c_str());
662 Info(
"Flush",
"XrdClient::Sync succeeded.");
688 XrdCl::OpenFlags::Flags &mode,
691 using namespace XrdCl;
694 if (modestr ==
"NEW" || modestr ==
"CREATE") mode = OpenFlags::New;
695 else if (modestr ==
"RECREATE") mode = OpenFlags::Delete;
696 else if (modestr ==
"UPDATE") mode = OpenFlags::Update;
697 else if (modestr ==
"READ") mode = OpenFlags::Read;
703 mode = OpenFlags::Read;
715 Error(
"TNetXNGFile",
"Object is in 'zombie' state");
720 Error(
"TNetXNGFile",
"The remote file is not open");
733 using namespace XrdCl;
742#if XrdVNUMBER >= 40000
744 fFile->GetProperty(
"LastURL",lasturl);
748 if(lrl.GetProtocol().compare(
"file") == 0 &&
749 lrl.GetHostId().compare(
"localhost") == 0){
751 Info(
"GetVectorReadLimits",
"Local redirect, using default values");
755 std::string dataServerStr;
756 if( !
fFile->GetProperty(
"DataServer", dataServerStr ) )
758 URL dataServer(dataServerStr);
760 URL dataServer(
fFile->GetDataServer());
762 FileSystem fs(dataServer);
765 arg.FromString(std::string(
"readv_ior_max readv_iov_max"));
767 XRootDStatus status = fs.Query(QueryCode::Config, arg, response);
774 std::vector<TString> resps;
776 resps.push_back(token);
778 if (resps.size() != 2)
781 if (resps[0].IsDigit())
784 if (resps[1].IsDigit())
805 XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
806 const char *cenv = 0;
811 || strlen(cenv) <= 0))
812 env->PutInt(
"ConnectionWindow", val.
Atoi());
816 || strlen(cenv) <= 0))
817 env->PutInt(
"RequestTimeout", val.
Atoi());
821 || strlen(cenv) <= 0))
822 env->PutInt(
"RequestTimeout", val.
Atoi());
824 val =
gEnv->
GetValue(
"NetXNG.SubStreamsPerChannel",
"");
826 || strlen(cenv) <= 0))
827 env->PutInt(
"SubStreamsPerChannel", val.
Atoi());
831 || strlen(cenv) <= 0))
832 env->PutInt(
"TimeoutResolution", val.
Atoi());
836 || strlen(cenv) <= 0))
837 env->PutInt(
"StreamErrorWindow", val.
Atoi());
841 || strlen(cenv) <= 0))
842 env->PutInt(
"RunForkHandler", val.
Atoi());
846 || strlen(cenv) <= 0))
847 env->PutInt(
"RedirectLimit", val.
Atoi());
851 || strlen(cenv) <= 0))
852 env->PutInt(
"WorkerThreads", val.
Atoi());
856 || strlen(cenv) <= 0))
857 env->PutInt(
"CPChunkSize", val.
Atoi());
861 || strlen(cenv) <= 0))
862 env->PutInt(
"CPParallelChunks", val.
Atoi());
866 || strlen(cenv) <= 0))
867 env->PutString(
"PollerPreference", val.
Data());
871 || strlen(cenv) <= 0))
872 env->PutString(
"ClientMonitor", val.
Data());
876 || strlen(cenv) <= 0))
877 env->PutString(
"ClientMonitorParam", val.
Data());
880 env->PutInt(
"MultiProtocol",
gEnv->
GetValue(
"TFile.CrossProtocolRedirects", 1));
930 || strlen(cenv) <= 0))
935 || strlen(cenv) <= 0))
940 || strlen(cenv) <= 0))
945 || strlen(cenv) <= 0))
950 || strlen(cenv) <= 0))
955 || strlen(cenv) <= 0))
TString ToUpper(const TString &s)
Return an upper-case version of str.
R__EXTERN TSystem * gSystem
R__EXTERN TVirtualMonitoringWriter * gMonitoringWriter
TArchiveMember * GetMember() const
Long64_t GetDecompressedSize() const
virtual void SetOwner(Bool_t enable=kTRUE)
Set whether this collection is the owner (enable==true) of its content.
Bool_t fWritable
True if directory is writable.
virtual Int_t GetValue(const char *name, Int_t dflt) const
Returns the integer value for a resource.
A ROOT file is a suite of consecutive data records (TKey instances) with a well defined format.
static std::atomic< Long64_t > fgBytesRead
Number of bytes read by all TFile objects.
Int_t fReadCalls
Number of read calls ( not counting the cache calls )
Long64_t fBytesRead
Number of bytes read from this file.
TArchiveFile * fArchive
!Archive file from which we read this file
TList * fOpenPhases
!Time info about open phases
Int_t WriteBufferViaCache(const char *buf, Int_t len)
Write buffer via cache.
Int_t ReadBufferViaCache(char *buf, Int_t len)
Read buffer via cache.
Long64_t fArchiveOffset
!Offset at which file starts in archive
virtual void Init(Bool_t create)
Initialize a TFile object.
Long64_t GetRelOffset() const
TString fOption
File options.
EAsyncOpenStatus
Asynchronous open request status.
Bool_t FlushWriteCache()
Flush the write cache if active.
Long64_t fBytesWrite
Number of bytes written to this file.
Bool_t fInitDone
!True if the file has been initialized
virtual void SetOffset(Long64_t offset, ERelativeTo pos=kBeg)
Set position from where to start reading.
Long64_t fOffset
!Seek offset cache
static std::atomic< Long64_t > fgBytesWrite
Number of bytes written by all TFile objects.
EAsyncOpenStatus fAsyncOpenStatus
!Status of an asynchronous open request
void Close(Option_t *option="") override
Close a file.
static std::atomic< Int_t > fgReadCalls
Number of bytes read from all TFile objects.
virtual void SetEnv()
Map ROOT and xrootd environment variables.
virtual Bool_t IsUseable() const
Check the file is open and isn't a zombie.
virtual void SetAsyncOpenStatus(EAsyncOpenStatus status)
Set the status of an asynchronous file open.
XrdCl::OpenFlags::Flags fMode
virtual void Close(const Option_t *option="")
Close the file.
Int_t ParseOpenMode(Option_t *in, TString &modestr, XrdCl::OpenFlags::Flags &mode, Bool_t assumeRead)
Parse a file open mode given as a string into a canonically formatted output mode string and an integ...
virtual Int_t ReOpen(Option_t *modestr)
Reopen the file with the new access mode.
virtual Bool_t WriteBuffer(const char *buffer, Int_t length)
Write a data chunk.
virtual Bool_t ReadBuffer(char *buffer, Int_t length)
Read a data chunk of the given size.
virtual Bool_t ReadBuffers(char *buffer, Long64_t *position, Int_t *length, Int_t nbuffs)
Read scattered data chunks in one operation.
virtual Long64_t GetSize() const
Get the file size.
XrdSysCondVar * fInitCondVar
virtual Bool_t IsOpen() const
Check if the file is open.
virtual void Init(Bool_t create)
Initialize the file.
virtual ~TNetXNGFile()
Destructor.
virtual void Flush()
Synchronize a file's in-memory and on-disk states.
virtual Bool_t GetVectorReadLimits()
Find the server-specific readv config params.
virtual void Seek(Long64_t offset, ERelativeTo position=kBeg)
Set the position within the file.
R__ALWAYS_INLINE Bool_t IsZombie() const
virtual void Error(const char *method, const char *msgfmt,...) const
Issue error message.
virtual void Info(const char *method, const char *msgfmt,...) const
Issue info message.
Int_t Wait()
If the semaphore value is > 0 then decrement it and carry on, else block, waiting on the condition un...
Int_t Atoi() const
Return integer value of string.
const char * Data() const
TObjArray * Tokenize(const TString &delim) const
This function is used to isolate sequential tokens in a TString.
void Form(const char *fmt,...)
Formats a string using a printf style format descriptor.
virtual const char * Getenv(const char *env)
Get environment variable.
virtual const char * HomeDirectory(const char *userName=0)
Return the user's home directory.
virtual void Setenv(const char *name, const char *value)
Set environment variable.
The TTimeStamp encapsulates seconds and ns since EPOCH.
This class represents a WWW compatible URL.
const char * GetUrl(Bool_t withDeflt=kFALSE) const
Return full URL.
void SetAnchor(const char *anchor)
virtual Bool_t SendFileOpenProgress(TFile *, TList *, const char *, Bool_t=kFALSE)
virtual Bool_t SendFileReadProgress(TFile *)