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) )
243 using namespace XrdCl;
246 if (
gDebug > 1)
Info(
"Init",
"TFile::Init already called once");
282 using namespace XrdCl;
288 bool forceStat =
true;
289 if(
fMode == XrdCl::OpenFlags::Read )
293 if( !
fFile->Stat( forceStat, info ).IsOK() )
328 XrdCl::XRootDStatus status =
fFile->Close();
329 if (!status.IsOK()) {
330 Error(
"Close",
"%s", status.ToStr().c_str());
348 using namespace XrdCl;
350 OpenFlags::Flags mode;
355 if (parseres<0 || (mode != OpenFlags::Read && mode != OpenFlags::Update)) {
356 Error(
"ReOpen",
"mode must be either READ or UPDATE, not %s", modestr);
361 if (mode ==
fMode || (mode == OpenFlags::Update
362 &&
fMode == OpenFlags::New)) {
366 XRootDStatus st =
fFile->Close();
368 Error(
"ReOpen",
"%s", st.ToStr().c_str());
376 Error(
"ReOpen",
"%s", st.ToStr().c_str());
405 using namespace XrdCl;
407 Info(
"ReadBuffer",
"offset: %lld length: %d", position, length);
426 uint32_t bytesRead = 0;
427 XRootDStatus st =
fFile->Read(
fOffset, length, buffer, bytesRead);
429 Info(
"ReadBuffer",
"%s bytes read: %u", st.ToStr().c_str(), bytesRead);
432 Error(
"ReadBuffer",
"%s", st.ToStr().c_str());
436 if ((
Int_t)bytesRead != length) {
437 Error(
"ReadBuffer",
"error reading all requested bytes, got %u of %d",
473 using namespace XrdCl;
479 std::vector<ChunkList> chunkLists;
481 std::vector<XRootDStatus*> *statuses;
483 Int_t totalBytes = 0;
485 char *cursor = buffer;
491 for (
Int_t i = 0; i < nbuffs; i++)
495 for (
Int_t i = 0; i < nbuffs; ++i) {
496 totalBytes += length[i];
505 for (j = 0; j < nsplit; ++j) {
507 chunks.push_back(ChunkInfo(offset,
fReadvIorMax, cursor));
513 chunks.push_back(ChunkInfo(offset, rem, cursor));
516 chunks.push_back(ChunkInfo(position[i], length[i], cursor));
522 chunkLists.push_back(chunks);
523 chunks = ChunkList();
525 chunkLists.push_back(ChunkList(chunks.begin(),
527 chunks = ChunkList(chunks.begin() +
fReadvIovMax, chunks.end());
532 if( !chunks.empty() )
533 chunkLists.push_back(chunks);
535 TAsyncReadvHandler *handler;
538 statuses =
new std::vector<XRootDStatus*>(chunkLists.size());
541 std::vector<ChunkList>::iterator it;
542 for (it = chunkLists.begin(); it != chunkLists.end(); ++it)
544 handler =
new TAsyncReadvHandler(statuses, it - chunkLists.begin(),
546 status =
fFile->VectorRead(*it, 0, handler);
548 if (!status.IsOK()) {
549 Error(
"ReadBuffers",
"%s", status.ToStr().c_str());
555 for (it = chunkLists.begin(); it != chunkLists.end(); ++it) {
560 for (it = chunkLists.begin(); it != chunkLists.end(); ++it) {
561 XRootDStatus *st = statuses->at(it - chunkLists.begin());
564 Error(
"ReadBuffers",
"%s", st->ToStr().c_str());
565 for( ; it != chunkLists.end(); ++it )
567 st = statuses->at( it - chunkLists.begin() );
586 gPerfStats->FileReadEvent(
this, totalBytes, start);
606 using namespace XrdCl;
614 Info(
"WriteBuffer",
"file not writable");
627 XRootDStatus st =
fFile->Write(
fOffset, length, buffer);
629 Error(
"WriteBuffer",
"%s", st.ToStr().c_str());
650 Info(
"Flush",
"file not writable - do nothing");
658 XrdCl::XRootDStatus status =
fFile->Sync();
660 Error(
"Flush",
"%s", status.ToStr().c_str());
663 Info(
"Flush",
"XrdClient::Sync succeeded.");
689 XrdCl::OpenFlags::Flags &mode,
692 using namespace XrdCl;
695 if (modestr ==
"NEW" || modestr ==
"CREATE") mode = OpenFlags::New;
696 else if (modestr ==
"RECREATE") mode = OpenFlags::Delete;
697 else if (modestr ==
"UPDATE") mode = OpenFlags::Update;
698 else if (modestr ==
"READ") mode = OpenFlags::Read;
704 mode = OpenFlags::Read;
716 Error(
"TNetXNGFile",
"Object is in 'zombie' state");
721 Error(
"TNetXNGFile",
"The remote file is not open");
734 using namespace XrdCl;
743#if XrdVNUMBER >= 40000
745 fFile->GetProperty(
"LastURL",lasturl);
749 if(lrl.GetProtocol().compare(
"file") == 0 &&
750 lrl.GetHostId().compare(
"localhost") == 0){
752 Info(
"GetVectorReadLimits",
"Local redirect, using default values");
756 std::string dataServerStr;
757 if( !
fFile->GetProperty(
"DataServer", dataServerStr ) )
759 URL dataServer(dataServerStr);
761 URL dataServer(
fFile->GetDataServer());
763 FileSystem fs(dataServer);
766 arg.FromString(std::string(
"readv_ior_max readv_iov_max"));
768 XRootDStatus status = fs.Query(QueryCode::Config, arg, response);
775 std::vector<TString> resps;
777 resps.push_back(token);
779 if (resps.size() != 2)
782 if (resps[0].IsDigit())
785 if (resps[1].IsDigit())
806 XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
807 const char *cenv = 0;
812 || strlen(cenv) <= 0))
813 env->PutInt(
"ConnectionWindow", val.
Atoi());
817 || strlen(cenv) <= 0))
818 env->PutInt(
"RequestTimeout", val.
Atoi());
822 || strlen(cenv) <= 0))
823 env->PutInt(
"RequestTimeout", val.
Atoi());
825 val =
gEnv->
GetValue(
"NetXNG.SubStreamsPerChannel",
"");
827 || strlen(cenv) <= 0))
828 env->PutInt(
"SubStreamsPerChannel", val.
Atoi());
832 || strlen(cenv) <= 0))
833 env->PutInt(
"TimeoutResolution", val.
Atoi());
837 || strlen(cenv) <= 0))
838 env->PutInt(
"StreamErrorWindow", val.
Atoi());
842 || strlen(cenv) <= 0))
843 env->PutInt(
"RunForkHandler", val.
Atoi());
847 || strlen(cenv) <= 0))
848 env->PutInt(
"RedirectLimit", val.
Atoi());
852 || strlen(cenv) <= 0))
853 env->PutInt(
"WorkerThreads", val.
Atoi());
857 || strlen(cenv) <= 0))
858 env->PutInt(
"CPChunkSize", val.
Atoi());
862 || strlen(cenv) <= 0))
863 env->PutInt(
"CPParallelChunks", val.
Atoi());
867 || strlen(cenv) <= 0))
868 env->PutString(
"PollerPreference", val.
Data());
872 || strlen(cenv) <= 0))
873 env->PutString(
"ClientMonitor", val.
Data());
877 || strlen(cenv) <= 0))
878 env->PutString(
"ClientMonitorParam", val.
Data());
881 env->PutInt(
"MultiProtocol",
gEnv->
GetValue(
"TFile.CrossProtocolRedirects", 1));
931 || strlen(cenv) <= 0))
936 || strlen(cenv) <= 0))
941 || strlen(cenv) <= 0))
946 || strlen(cenv) <= 0))
951 || strlen(cenv) <= 0))
956 || strlen(cenv) <= 0))
void Info(const char *location, const char *msgfmt,...)
Use this function for informational messages.
void Error(const char *location, const char *msgfmt,...)
Use this function in case an error occurred.
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
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 void Setenv(const char *name, const char *value)
Set environment variable.
virtual const char * HomeDirectory(const char *userName=nullptr)
Return the user's home directory.
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 *)