<commit_before>// @(#)root/netx:$Id$
// Author: <NAME>, <NAME>

/*************************************************************************
 * Copyright (C) 1995-2004, <NAME> and <NAME>.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TXNetFile                                                            //
//                                                                      //
// Authors: <NAME>, <NAME>                              //
//          INFN Padova, 2003                                           //
// Interfaced to the standalone client (XrdClient): <NAME>, CERN      //
//                                                                      //
// TXNetFile is an extension of TNetFile able to deal with new xrootd   //
// server. Its new features are:                                        //
//  - Automatic server kind recognition (xrootd load balancer, xrootd   //
//    data server, old rootd)                                           //
//  - Backward compatibility with old rootd server (acts as an old      //
//    TNetFile)                                                         //
//  - Fault tolerance for read/write operations (read/write timeouts    //
//    and retry)                                                        //
//  - Internal connection timeout (tunable indipendently from the OS    //
//    one) handled by threads                                           //
//  - handling of redirections from server                              //
//  - Single TCP physical channel for multiple TXNetFile's instances    //
//    inside the same application                                       //
//    So, each TXNetFile object client must send messages containing    //
//    its ID (streamid). The server, of course, will respond with       //
//    messages containing the client's ID, in order to make the client  //
//    able to recognize its message by matching its streamid with that  //
//    one contained in the server's response.                           //
//  - Tunable log verbosity level (0 = nothing, 3 = dump read/write     //
//    buffers too!)                                                     //
//  - Many parameters configurable via TEnv facility (see SetParm()     //
//    methods)                                                          //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#include "Bytes.h"

#include "TError.h"
#include "TEnv.h"
#include "TSocket.h"
#include "TXNetFile.h"
#include "TROOT.h"
#include "TVirtualMonitoring.h"
#include "TFileStager.h"
#include "TFileCacheRead.h"

#include <XrdClient/XrdClient.hh>
#include <XrdClient/XrdClientConst.hh>
#include <XrdClient/XrdClientEnv.hh>
#include <XProtocol/XProtocol.hh>
#ifdef OLDXRDOUC
#  include "XrdOuc/XrdOucPthread.hh"
#else
#  include "XrdSys/XrdSysPthread.hh"
#endif

ClassImp(TXNetFile);

Bool_t TXNetFile::fgInitDone = kFALSE;
Bool_t TXNetFile::fgRootdBC = kTRUE;


//_____________________________________________________________________________
TXNetFile::TXNetFile(const char *url, Option_t *option, const char* ftitle,
                     Int_t compress, Int_t netopt, Bool_t parallelopen,
                     const char *logicalurl) :
            TNetFile((logicalurl ? logicalurl : url), ftitle, compress, kFALSE)
{
   // Create a TXNetFile object. A TXNetFile object is the same as a TNetFile
   // (from which the former derives) except that the protocol is extended to
   // support dealing with new xrootd data server or xrootd load balancer
   // server.
   //
   // The "url" argument must be of the form
   //
   //   root://server1:port1[,server2:port2,...,serverN:portN]/pathfile,
   //
   // Note that this means that multiple servers (>= 1) can be specified in
   // the url. The connection will try to connect to the first server:port
   // and if that does not succeed, it will try the second one, and so on
   // until it finds a server that will respond.
   //
   // See the TNetFile documentation for the description of the other arguments.
   //
   // The creation consists of internal variable settings (most important is
   // the client's domain), creation of a TXUrl array containing all specified
   // urls (a single url is serverX:portX/pathfile), trying to connect to the
   // servers calling Connect() method, getting a valid access to the remote
   // server the client is connected to using GetAccessToSrv() method,
   // recognizing the remote server (if an old rootd the TNetFile's Create
   // method will be called).
   //
   // The options field of the URL can be used for the following purposes:
   //   a. open a non-ROOT generic file
   //      "root://server1:port1[,server2:port2,...]/pathfile?filetype=raw"
   //   b. re-check the environment variables
   //      "root://server1:port1[,server2:port2,...]/pathfile?checkenv"
   //
   TUrl urlnoanchor(url);
   // Set debug level
   EnvPutInt(NAME_DEBUG, gEnv->GetValue("XNet.Debug", -1));

   // Read ahead size. We need to reset it in any case
   Int_t rAheadsiz = gEnv->GetValue("XNet.ReadAheadSize",
                                     DFLT_READAHEADSIZE);
   EnvPutInt(NAME_READAHEADSIZE, rAheadsiz);

   // Set environment, if needed
   if (!fgInitDone || strstr(urlnoanchor.GetOptions(),"checkenv")) {
      SetEnv();
      fgInitDone = kTRUE;

      // Print the tag, if required (only once)
      if (gEnv->GetValue("XNet.PrintTAG",0) == 1)
         Info("TXNetFile","(C) 2005 SLAC TXNetFile (eXtended TNetFile) %s",
              gROOT->GetVersion());
   }

   // Remove anchors from the URL!
   urlnoanchor.SetAnchor("");

   // Init mutex used in the asynchronous open machinery
   fInitMtx = new XrdSysRecMutex();

   // Create an instance
   CreateXClient(urlnoanchor.GetUrl(), option, netopt, parallelopen);
}

//_____________________________________________________________________________
TXNetFile::~TXNetFile()
{
   // Destructor.

   if (IsOpen())
      Close(0);

   SafeDelete(fInitMtx);
   SafeDelete(fClient);
}

//_____________________________________________________________________________
void TXNetFile::FormUrl(TUrl uu, TString &uus)
{
   // Form url for rootd socket.

   // Protocol
   uus = "root://";

   // User, if any
   if (strlen(uu.GetUser()) > 0) {
      uus += uu.GetUser();
      uus += "@";
   }

   // Host, if any
   if (strlen(uu.GetHost()) > 0) {
      uus += uu.GetHost();
   }

   // Port, if any
   if (uu.GetPort() > 0) {
      uus += ":";
      uus += uu.GetPort();
   }

   // End of string
   uus += "/";
}

//_____________________________________________________________________________
void TXNetFile::CreateXClient(const char *url, Option_t *option, Int_t netopt,
                              Bool_t parallelopen)
{
   // The real creation work is done here.

   fClient = 0;

   // Set the timeout (default 999999999 secs, i.e. far, far in the future)
   gSystem->Setenv("XRDCLIENTMAXWAIT", Form("%d",TFile::GetOpenTimeout()));

   if (GetOnlyStaged()) {
      static TFileStager* fFileStager = 0;
      // check if the file is staged before opening it
      if (!fFileStager)
         fFileStager = TFileStager::Open(url);
      if (fFileStager) {
         if (!(fFileStager->IsStaged(url))) {
            ::Warning("TXNetFile","<%s> is not staged - StageOnly flag is set!",url);
            goto zombie;
         }
      }
   }

   // Init members
   fIsRootd = kFALSE;

   // The parallel open can be forced to true in the config
   if (gEnv->GetValue("XNet.ForceParallelOpen", 0))
      parallelopen = kTRUE;
   fAsyncOpenStatus = (parallelopen) ? kAOSInProgress : fAsyncOpenStatus ;

   Bool_t isRootd;
   isRootd = kFALSE;
   //
   // Setup a client instance
   fClient = new XrdClient(url);
   if (!fClient) {
      fAsyncOpenStatus = (parallelopen) ? kAOSFailure : fAsyncOpenStatus ;
      Error("CreateXClient","fatal error: new object creation failed -"
            " out of system resources.");
      gSystem->Abort();
      goto zombie;
   }

   //
   // Now try opening the file
   if (!Open(option, parallelopen)) {
      if (!fClient->IsOpen_wait()) {
         if (gDebug > 1)
            Info("CreateXClient", "remote file could not be open");

         // If the server is a rootd we need to create a TNetFile
         isRootd = (fClient->GetClientConn()->GetServerType() == kSTRootd);

         if (isRootd) {
            if (fgRootdBC) {

               Int_t sd = fClient->GetClientConn()->GetOpenSockFD();
               if (sd > -1) {
                  //
                  // Create a TSocket on the open connection
                  TSocket *s = new TSocket(sd);

                  s->SetOption(kNoBlock, 0);

                  // Find out the remote protocol (send the client protocol first)
                  Int_t rproto = GetRootdProtocol(s);
                  if (rproto < 0) {
                     Error("CreateXClient", "getting rootd server protocol");
                     goto zombie;
                  }

                  // Finalize TSocket initialization
                  s->SetRemoteProtocol(rproto);
                  TUrl uut((fClient->GetClientConn()
                                   ->GetCurrentUrl()).GetUrl().c_str());
                  TString uu;
                  FormUrl(uut,uu);

                  if (gDebug > 2)
                     Info("CreateXClient"," url: %s",uu.Data());
                  s->SetUrl(uu.Data());
                  s->SetService("rootd");
                  s->SetServType(TSocket::kROOTD);
                  //
                  // Set rootd flag
                  fIsRootd = kTRUE;
                  //
                  // Now we can check if we can create a TNetFile on the
                  // open connection
                  if (rproto > 13) {
                     //
                     // Remote support for reuse of open connection
                     TNetFile::Create(s, option, netopt);
                  } else {
                     //
                     // Open connection has been closed because could
                     // not be reused; TNetFile will open a new connection
                     TNetFile::Create(uu.Data(), option, netopt);
                  }

                  return;
               } else {
                  Error("CreateXClient", "rootd: underlying socket undefined");
                  goto zombie;
               }
            } else {
               if (gDebug > 0)
                  Info("CreateXClient", "rootd: fall back not enabled - closing");
               goto zombie;
            }
         } else {
            Error("CreateXClient", "open attempt failed on %s", fUrl.GetUrl());
            goto zombie;
         }
      }
   }

   return;

zombie:
   // error in file opening occured, make this object a zombie
   SafeDelete(fClient);
   MakeZombie();
   gDirectory = gROOT;
}

//_____________________________________________________________________________
Int_t TXNetFile::GetRootdProtocol(TSocket *s)
{
   // Find out the remote rootd protocol version.
   // Returns -1 in case of error.

   Int_t rproto = -1;

   UInt_t cproto = 0;
   Int_t len = sizeof(cproto);
   memcpy((char *)&cproto,
      Form(" %d", TSocket::GetClientProtocol()),len);
   Int_t ns = s->SendRaw(&cproto, len);
   if (ns != len) {
      ::Error("TXNetFile::GetRootdProtocol",
              "sending %d bytes to rootd server [%s:%d]",
              len, (s->GetInetAddress()).GetHostName(), s->GetPort());
      return -1;
   }

   // Get the remote protocol
   Int_t ibuf[2] = {0};
   len = sizeof(ibuf);
   Int_t nr = s->RecvRaw(ibuf, len);
   if (nr != len) {
      ::Error("TXNetFile::GetRootdProtocol",
              "reading %d bytes from rootd server [%s:%d]",
              len, (s->GetInetAddress()).GetHostName(), s->GetPort());
      return -1;
   }
   Int_t kind = net2host(ibuf[0]);
   if (kind == kROOTD_PROTOCOL) {
      rproto = net2host(ibuf[1]);
   } else {
      kind = net2host(ibuf[1]);
      if (kind == kROOTD_PROTOCOL) {
         len = sizeof(rproto);
         nr = s->RecvRaw(&rproto, len);
         if (nr != len) {
            ::Error("TXNetFile::GetRootdProtocol",
                    "reading %d bytes from rootd server [%s:%d]",
                    len, (s->GetInetAddress()).GetHostName(), s->GetPort());
            return -1;
         }
         rproto = net2host(rproto);
      }
   }
   if (gDebug > 2)
      ::Info("TXNetFile::GetRootdProtocol",
             "remote rootd: buf1: %d, buf2: %d rproto: %d",
             net2host(ibuf[0]),net2host(ibuf[1]),rproto);

   // We are done
   return rproto;
}

//_____________________________________________________________________________
Bool_t TXNetFile::Open(Option_t *option, Bool_t doitparallel)
{
   // The real creation work is done here.

   //
   // Parse options
   kXR_unt16 openOpt = 0;
   memset(&openOpt, 0, sizeof(openOpt));
   TString opt = option;
   opt.ToUpper();
   //
   // Check force, accepting 'f'/'F' for backward compatibility,
   // and special read syntax
   if (opt.BeginsWith("-") || opt.BeginsWith("F") || (opt == "+READ")) {
      opt.Remove(0,1);
      openOpt |= kXR_force;
   }
   //
   // Read flag
   Bool_t read = (opt == "READ");
   //
   // Create flag ("NEW" == "CREATE")
   Bool_t create = (opt == "CREATE" || opt == "NEW");
   //
   // Recreate flag
   Bool_t recreate = (opt == "RECREATE");
   //
   // Update flag
   Bool_t update = (opt == "UPDATE");
   //
   // Default is Read
   if (!create && !recreate && !update && !read) {
      read = kTRUE;
      opt = "READ";
   }
   //
   // Save effective options
   fOption = opt;
   if (create || update || recreate)
      fWritable = 1;
   //
   // Update requires the file existing: check that and switch to create,
   // if the file is not found.
   if (update) {
      if (gSystem->AccessPathName(fUrl.GetUrl(), kFileExists)) {
         update = kFALSE;
         create = kTRUE;
      }
      if (update) {
         if (gSystem->AccessPathName(fUrl.GetUrl(), kWritePermission)) {
            Error("Open", "no write permission, could not open file %s",
                          fUrl.GetUrl());
            fAsyncOpenStatus = (doitparallel) ? kAOSFailure : fAsyncOpenStatus ;
            return kFALSE;
         }
         openOpt |= kXR_open_updt;
      }
   }

   //
   // Create and Recreate are correlated
   if (recreate) {
      openOpt |= kXR_delete;
      create = kTRUE;
   }
   if (create)
      openOpt |= kXR_new;

   Bool_t mkpath = (gEnv->GetValue("XNet.Mkpath", 0) == 1) ? kTRUE : kFALSE;
   char *p = (char*)strstr(fUrl.GetOptions(), "mkpath=");
   if (p)
     mkpath = (*(p + strlen("mkpath=")) == '1') ? kTRUE : kFALSE;
   if (mkpath)
     openOpt |= kXR_mkpath;

   if (read)
      openOpt |= kXR_open_read;

   //
   // Set open mode to rw-r-r
   kXR_unt16 openMode = kXR_or | kXR_gr | kXR_ur | kXR_uw;

   //
   // Open file (FileOpenerThread disabled for the time being)
   if (!fClient->Open(openMode, openOpt, doitparallel)) {
      if (gDebug > 1)
         Info("Open", "remote file could not be open");
      fAsyncOpenStatus = (doitparallel) ? kAOSFailure : fAsyncOpenStatus ;
      return kFALSE;
   } else {
      // Initialize the file
      // If we are using the parallel open, the init phase is
      // performed later. In checking for the IsOpen or
      // asynchronously in a callback func
      if (!doitparallel) {
         // Mutex serialization is done inside
         Init(create);
         // If initialization failed close everything
         if (TFile::IsZombie()) {
            fClient->Close();
            // To avoid problems in final deletion of object not completely
            // initialized
            fWritable = 0;
            // Notify failure
            return kFALSE;
         }
      }
   }

   // We are done
   return kTRUE;
}

//_____________________________________________________________________________
Bool_t TXNetFile::ReadBuffer(char *buffer, Int_t bufferLength)
{
   // Override TNetFile::ReadBuffer to deal with the xrootd server.
   // Returns kTRUE in case of errors.

   if (IsZombie()) {
      Error("ReadBuffer", "ReadBuffer is not possible because object"
            " is in 'zombie' state");
      return kTRUE;
   }

   if (fIsRootd) {
      if (gDebug > 1)
         Info("ReadBuffer","Calling TNetFile::ReadBuffer");
      return TNetFile::ReadBuffer(buffer, bufferLength);
   }

   if (!IsOpen()) {
      Error("ReadBuffer","The remote file is not open");
      return kTRUE;
   }

   Bool_t result = kFALSE;

   if (bufferLength==0)
      return 0;

   // This returns:
   //  2 if errors
   //  1 it looks like the block has already been prefetched
   //  0 it looks like the block has not been prefetched
   // But we don't want it to return the buffer, to avoid recursion
   Int_t st = ReadBufferViaCache(0, bufferLength);
   if (st == 1) fOffset -= bufferLength;

   if (!st) {
      // Not prefetched. set the read ahead at its default value
      Int_t rAheadsiz = gEnv->GetValue("XNet.ReadAheadSize", DFLT_READAHEADSIZE);
      EnvPutInt(NAME_READAHEADSIZE, rAheadsiz);
   }

   // Read for the remote xrootd
   Int_t nr = fClient->Read(buffer, fOffset, bufferLength);

   if (!nr)
      return kTRUE;

   if (gDebug > 1)
      Info("ReadBuffer", "%d bytes of data read from offset"
                         " %lld (%d requested)", nr, fOffset, bufferLength);

   fOffset += bufferLength;

   if (!st) {
     // Update the counters only if the block has not been prefetched
     fBytesRead += nr;
#ifdef WIN32
     SetFileBytesRead(GetFileBytesRead() + nr);
     SetFileReadCalls(GetFileReadCalls() + 1);
#else
     fgBytesRead += nr;
     fgReadCalls++;
#endif
   }

   if (gMonitoringWriter)
      gMonitoringWriter->SendFileReadProgress(this);

   return result;
}

//______________________________________________________________________________
Bool_t TXNetFile::ReadBufferAsync(Long64_t offs, Int_t bufferLength)
{
   // Implementation dealing with the xrootd server.
   // Returns kTRUE in case of errors.
   // This is the same as TXNetFile::ReadBuffer but using the async
   // call from xrootd

   if (IsZombie()) {
      Error("ReadBuffer", "ReadBuffer is not possible because object"
            " is in 'zombie' state");
      return kTRUE;
   }

   if (fIsRootd) {
      if (gDebug > 1)
         Error("ReadBufferAsync","Not supported for rootd");
      return kTRUE;
   }

   if (!IsOpen()) {
      Error("ReadBuffer","The remote file is not open");
      return kTRUE;
   }

   Bool_t result = kFALSE;

   if (bufferLength==0)
      return 0;

   SynchronizeCacheSize();

   // Read for the remote xrootd
   // This doesnt return the number of bytes read...
   // and even if it did we dont want to update fBytesRead
   // because that would be updated in the real read
   XReqErrorType nr = fClient->Read_Async(offs, bufferLength);

   if (nr != kOK)
      return kTRUE;

   fBytesRead += bufferLength;
#ifdef WIN32
   SetFileBytesRead(GetFileBytesRead() + bufferLength);
   SetFileReadCalls(GetFileReadCalls() + 1);
#else
   fgBytesRead += bufferLength;
   fgReadCalls++;
#endif

   if (gDebug > 1)
      Info("ReadBufferAsync", "%d bytes of data read request from offset"
                              " %lld", bufferLength, offs);
   return result;
}



//______________________________________________________________________________
Bool_t TXNetFile::ReadBuffers(char *buf,  Long64_t *pos, Int_t *len, Int_t nbuf)<commit_msg>in raw mode don't use the read-ahead cache.<commit_after>// @(#)root/netx:$Id$
// Author: <NAME>, <NAME>

/*************************************************************************
 * Copyright (C) 1995-2004, <NAME> and <NAME>.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TXNetFile                                                            //
//                                                                      //
// Authors: <NAME>, <NAME>                              //
//          INFN Padova, 2003                                           //
// Interfaced to the standalone client (XrdClient): <NAME>, CERN      //
//                                                                      //
// TXNetFile is an extension of TNetFile able to deal with new xrootd   //
// server. Its new features are:                                        //
//  - Automatic server kind recognition (xrootd load balancer, xrootd   //
//    data server, old rootd)                                           //
//  - Backward compatibility with old rootd server (acts as an old      //
//    TNetFile)                                                         //
//  - Fault tolerance for read/write operations (read/write timeouts    //
//    and retry)                                                        //
//  - Internal connection timeout (tunable indipendently from the OS    //
//    one) handled by threads                                           //
//  - handling of redirections from server                              //
//  - Single TCP physical channel for multiple TXNetFile's instances    //
//    inside the same application                                       //
//    So, each TXNetFile object client must send messages containing    //
//    its ID (streamid). The server, of course, will respond with       //
//    messages containing the client's ID, in order to make the client  //
//    able to recognize its message by matching its streamid with that  //
//    one contained in the server's response.                           //
//  - Tunable log verbosity level (0 = nothing, 3 = dump read/write     //
//    buffers too!)                                                     //
//  - Many parameters configurable via TEnv facility (see SetParm()     //
//    methods)                                                          //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#include "Bytes.h"

#include "TError.h"
#include "TEnv.h"
#include "TSocket.h"
#include "TXNetFile.h"
#include "TROOT.h"
#include "TVirtualMonitoring.h"
#include "TFileStager.h"
#include "TFileCacheRead.h"

#include <XrdClient/XrdClient.hh>
#include <XrdClient/XrdClientConst.hh>
#include <XrdClient/XrdClientEnv.hh>
#include <XProtocol/XProtocol.hh>
#ifdef OLDXRDOUC
#  include "XrdOuc/XrdOucPthread.hh"
#else
#  include "XrdSys/XrdSysPthread.hh"
#endif

ClassImp(TXNetFile);

Bool_t TXNetFile::fgInitDone = kFALSE;
Bool_t TXNetFile::fgRootdBC = kTRUE;


//_____________________________________________________________________________
TXNetFile::TXNetFile(const char *url, Option_t *option, const char* ftitle,
                     Int_t compress, Int_t netopt, Bool_t parallelopen,
                     const char *logicalurl) :
            TNetFile((logicalurl ? logicalurl : url), ftitle, compress, kFALSE)
{
   // Create a TXNetFile object. A TXNetFile object is the same as a TNetFile
   // (from which the former derives) except that the protocol is extended to
   // support dealing with new xrootd data server or xrootd load balancer
   // server.
   //
   // The "url" argument must be of the form
   //
   //   root://server1:port1[,server2:port2,...,serverN:portN]/pathfile,
   //
   // Note that this means that multiple servers (>= 1) can be specified in
   // the url. The connection will try to connect to the first server:port
   // and if that does not succeed, it will try the second one, and so on
   // until it finds a server that will respond.
   //
   // See the TNetFile documentation for the description of the other arguments.
   //
   // The creation consists of internal variable settings (most important is
   // the client's domain), creation of a TXUrl array containing all specified
   // urls (a single url is serverX:portX/pathfile), trying to connect to the
   // servers calling Connect() method, getting a valid access to the remote
   // server the client is connected to using GetAccessToSrv() method,
   // recognizing the remote server (if an old rootd the TNetFile's Create
   // method will be called).
   //
   // The options field of the URL can be used for the following purposes:
   //   a. open a non-ROOT generic file
   //      "root://server1:port1[,server2:port2,...]/pathfile?filetype=raw"
   //   b. re-check the environment variables
   //      "root://server1:port1[,server2:port2,...]/pathfile?checkenv"
   //
   TUrl urlnoanchor(url);
   // Set debug level
   EnvPutInt(NAME_DEBUG, gEnv->GetValue("XNet.Debug", -1));

   // Read ahead size. We need to reset it in any case
   Int_t rAheadsiz = gEnv->GetValue("XNet.ReadAheadSize",
                                     DFLT_READAHEADSIZE);
   EnvPutInt(NAME_READAHEADSIZE, rAheadsiz);

   // Set environment, if needed
   if (!fgInitDone || strstr(urlnoanchor.GetOptions(),"checkenv")) {
      SetEnv();
      fgInitDone = kTRUE;

      // Print the tag, if required (only once)
      if (gEnv->GetValue("XNet.PrintTAG",0) == 1)
         Info("TXNetFile","(C) 2005 SLAC TXNetFile (eXtended TNetFile) %s",
              gROOT->GetVersion());
   }

   if (IsRaw()) {
      EnvPutInt(NAME_READAHEADSIZE, 0);
   }

   // Remove anchors from the URL!
   urlnoanchor.SetAnchor("");

   // Init mutex used in the asynchronous open machinery
   fInitMtx = new XrdSysRecMutex();

   // Create an instance
   CreateXClient(urlnoanchor.GetUrl(), option, netopt, parallelopen);
}

//_____________________________________________________________________________
TXNetFile::~TXNetFile()
{
   // Destructor.

   if (IsOpen())
      Close(0);

   SafeDelete(fInitMtx);
   SafeDelete(fClient);
}

//_____________________________________________________________________________
void TXNetFile::FormUrl(TUrl uu, TString &uus)
{
   // Form url for rootd socket.

   // Protocol
   uus = "root://";

   // User, if any
   if (strlen(uu.GetUser()) > 0) {
      uus += uu.GetUser();
      uus += "@";
   }

   // Host, if any
   if (strlen(uu.GetHost()) > 0) {
      uus += uu.GetHost();
   }

   // Port, if any
   if (uu.GetPort() > 0) {
      uus += ":";
      uus += uu.GetPort();
   }

   // End of string
   uus += "/";
}

//_____________________________________________________________________________
void TXNetFile::CreateXClient(const char *url, Option_t *option, Int_t netopt,
                              Bool_t parallelopen)
{
   // The real creation work is done here.

   fClient = 0;

   // Set the timeout (default 999999999 secs, i.e. far, far in the future)
   gSystem->Setenv("XRDCLIENTMAXWAIT", Form("%d",TFile::GetOpenTimeout()));

   if (GetOnlyStaged()) {
      static TFileStager* fFileStager = 0;
      // check if the file is staged before opening it
      if (!fFileStager)
         fFileStager = TFileStager::Open(url);
      if (fFileStager) {
         if (!(fFileStager->IsStaged(url))) {
            ::Warning("TXNetFile","<%s> is not staged - StageOnly flag is set!",url);
            goto zombie;
         }
      }
   }

   // Init members
   fIsRootd = kFALSE;

   // The parallel open can be forced to true in the config
   if (gEnv->GetValue("XNet.ForceParallelOpen", 0))
      parallelopen = kTRUE;
   fAsyncOpenStatus = (parallelopen) ? kAOSInProgress : fAsyncOpenStatus ;

   Bool_t isRootd;
   isRootd = kFALSE;
   //
   // Setup a client instance
   fClient = new XrdClient(url);
   if (!fClient) {
      fAsyncOpenStatus = (parallelopen) ? kAOSFailure : fAsyncOpenStatus ;
      Error("CreateXClient","fatal error: new object creation failed -"
            " out of system resources.");
      gSystem->Abort();
      goto zombie;
   }

   //
   // Now try opening the file
   if (!Open(option, parallelopen)) {
      if (!fClient->IsOpen_wait()) {
         if (gDebug > 1)
            Info("CreateXClient", "remote file could not be open");

         // If the server is a rootd we need to create a TNetFile
         isRootd = (fClient->GetClientConn()->GetServerType() == kSTRootd);

         if (isRootd) {
            if (fgRootdBC) {

               Int_t sd = fClient->GetClientConn()->GetOpenSockFD();
               if (sd > -1) {
                  //
                  // Create a TSocket on the open connection
                  TSocket *s = new TSocket(sd);

                  s->SetOption(kNoBlock, 0);

                  // Find out the remote protocol (send the client protocol first)
                  Int_t rproto = GetRootdProtocol(s);
                  if (rproto < 0) {
                     Error("CreateXClient", "getting rootd server protocol");
                     goto zombie;
                  }

                  // Finalize TSocket initialization
                  s->SetRemoteProtocol(rproto);
                  TUrl uut((fClient->GetClientConn()
                                   ->GetCurrentUrl()).GetUrl().c_str());
                  TString uu;
                  FormUrl(uut,uu);

                  if (gDebug > 2)
                     Info("CreateXClient"," url: %s",uu.Data());
                  s->SetUrl(uu.Data());
                  s->SetService("rootd");
                  s->SetServType(TSocket::kROOTD);
                  //
                  // Set rootd flag
                  fIsRootd = kTRUE;
                  //
                  // Now we can check if we can create a TNetFile on the
                  // open connection
                  if (rproto > 13) {
                     //
                     // Remote support for reuse of open connection
                     TNetFile::Create(s, option, netopt);
                  } else {
                     //
                     // Open connection has been closed because could
                     // not be reused; TNetFile will open a new connection
                     TNetFile::Create(uu.Data(), option, netopt);
                  }

                  return;
               } else {
                  Error("CreateXClient", "rootd: underlying socket undefined");
                  goto zombie;
               }
            } else {
               if (gDebug > 0)
                  Info("CreateXClient", "rootd: fall back not enabled - closing");
               goto zombie;
            }
         } else {
            Error("CreateXClient", "open attempt failed on %s", fUrl.GetUrl());
            goto zombie;
         }
      }
   }

   return;

zombie:
   // error in file opening occured, make this object a zombie
   SafeDelete(fClient);
   MakeZombie();
   gDirectory = gROOT;
}

//_____________________________________________________________________________
Int_t TXNetFile::GetRootdProtocol(TSocket *s)
{
   // Find out the remote rootd protocol version.
   // Returns -1 in case of error.

   Int_t rproto = -1;

   UInt_t cproto = 0;
   Int_t len = sizeof(cproto);
   memcpy((char *)&cproto,
      Form(" %d", TSocket::GetClientProtocol()),len);
   Int_t ns = s->SendRaw(&cproto, len);
   if (ns != len) {
      ::Error("TXNetFile::GetRootdProtocol",
              "sending %d bytes to rootd server [%s:%d]",
              len, (s->GetInetAddress()).GetHostName(), s->GetPort());
      return -1;
   }

   // Get the remote protocol
   Int_t ibuf[2] = {0};
   len = sizeof(ibuf);
   Int_t nr = s->RecvRaw(ibuf, len);
   if (nr != len) {
      ::Error("TXNetFile::GetRootdProtocol",
              "reading %d bytes from rootd server [%s:%d]",
              len, (s->GetInetAddress()).GetHostName(), s->GetPort());
      return -1;
   }
   Int_t kind = net2host(ibuf[0]);
   if (kind == kROOTD_PROTOCOL) {
      rproto = net2host(ibuf[1]);
   } else {
      kind = net2host(ibuf[1]);
      if (kind == kROOTD_PROTOCOL) {
         len = sizeof(rproto);
         nr = s->RecvRaw(&rproto, len);
         if (nr != len) {
            ::Error("TXNetFile::GetRootdProtocol",
                    "reading %d bytes from rootd server [%s:%d]",
                    len, (s->GetInetAddress()).GetHostName(), s->GetPort());
            return -1;
         }
         rproto = net2host(rproto);
      }
   }
   if (gDebug > 2)
      ::Info("TXNetFile::GetRootdProtocol",
             "remote rootd: buf1: %d, buf2: %d rproto: %d",
             net2host(ibuf[0]),net2host(ibuf[1]),rproto);

   // We are done
   return rproto;
}

//_____________________________________________________________________________
Bool_t TXNetFile::Open(Option_t *option, Bool_t doitparallel)
{
   // The real creation work is done here.

   //
   // Parse options
   kXR_unt16 openOpt = 0;
   memset(&openOpt, 0, sizeof(openOpt));
   TString opt = option;
   opt.ToUpper();
   //
   // Check force, accepting 'f'/'F' for backward compatibility,
   // and special read syntax
   if (opt.BeginsWith("-") || opt.BeginsWith("F") || (opt == "+READ")) {
      opt.Remove(0,1);
      openOpt |= kXR_force;
   }
   //
   // Read flag
   Bool_t read = (opt == "READ");
   //
   // Create flag ("NEW" == "CREATE")
   Bool_t create = (opt == "CREATE" || opt == "NEW");
   //
   // Recreate flag
   Bool_t recreate = (opt == "RECREATE");
   //
   // Update flag
   Bool_t update = (opt == "UPDATE");
   //
   // Default is Read
   if (!create && !recreate && !update && !read) {
      read = kTRUE;
      opt = "READ";
   }
   //
   // Save effective options
   fOption = opt;
   if (create || update || recreate)
      fWritable = 1;
   //
   // Update requires the file existing: check that and switch to create,
   // if the file is not found.
   if (update) {
      if (gSystem->AccessPathName(fUrl.GetUrl(), kFileExists)) {
         update = kFALSE;
         create = kTRUE;
      }
      if (update) {
         if (gSystem->AccessPathName(fUrl.GetUrl(), kWritePermission)) {
            Error("Open", "no write permission, could not open file %s",
                          fUrl.GetUrl());
            fAsyncOpenStatus = (doitparallel) ? kAOSFailure : fAsyncOpenStatus ;
            return kFALSE;
         }
         openOpt |= kXR_open_updt;
      }
   }

   //
   // Create and Recreate are correlated
   if (recreate) {
      openOpt |= kXR_delete;
      create = kTRUE;
   }
   if (create)
      openOpt |= kXR_new;

   Bool_t mkpath = (gEnv->GetValue("XNet.Mkpath", 0) == 1) ? kTRUE : kFALSE;
   char *p = (char*)strstr(fUrl.GetOptions(), "mkpath=");
   if (p)
     mkpath = (*(p + strlen("mkpath=")) == '1') ? kTRUE : kFALSE;
   if (mkpath)
     openOpt |= kXR_mkpath;

   if (read)
      openOpt |= kXR_open_read;

   //
   // Set open mode to rw-r-r
   kXR_unt16 openMode = kXR_or | kXR_gr | kXR_ur | kXR_uw;

   //
   // Open file (FileOpenerThread disabled for the time being)
   if (!fClient->Open(openMode, openOpt, doitparallel)) {
      if (gDebug > 1)
         Info("Open", "remote file could not be open");
      fAsyncOpenStatus = (doitparallel) ? kAOSFailure : fAsyncOpenStatus ;
      return kFALSE;
   } else {
      // Initialize the file
      // If we are using the parallel open, the init phase is
      // performed later. In checking for the IsOpen or
      // asynchronously in a callback func
      if (!doitparallel) {
         // Mutex serialization is done inside
         Init(create);
         // If initialization failed close everything
         if (TFile::IsZombie()) {
            fClient->Close();
            // To avoid problems in final deletion of object not completely
            // initialized
            fWritable = 0;
            // Notify failure
            return kFALSE;
         }
      }
   }

   // We are done
   return kTRUE;
}

//_____________________________________________________________________________
Bool_t TXNetFile::ReadBuffer(char *buffer, Int_t bufferLength)
{
   // Override TNetFile::ReadBuffer to deal with the xrootd server.
   // Returns kTRUE in case of errors.

   if (IsZombie()) {
      Error("ReadBuffer", "ReadBuffer is not possible because object"
            " is in 'zombie' state");
      return kTRUE;
   }

   if (fIsRootd) {
      if (gDebug > 1)
         Info("ReadBuffer","Calling TNetFile::ReadBuffer");
      return TNetFile::ReadBuffer(buffer, bufferLength);
   }

   if (!IsOpen()) {
      Error("ReadBuffer","The remote file is not open");
      return kTRUE;
   }

   Bool_t result = kFALSE;

   if (bufferLength==0)
      return 0;

   // This returns:
   //  2 if errors
   //  1 it looks like the block has already been prefetched
   //  0 it looks like the block has not been prefetched
   // But we don't want it to return the buffer, to avoid recursion
   Int_t st = ReadBufferViaCache(0, bufferLength);
   if (st == 1) fOffset -= bufferLength;

   if (!st) {
      // Not prefetched. set the read ahead at its default value
      Int_t rAheadsiz = gEnv->GetValue("XNet.ReadAheadSize", DFLT_READAHEADSIZE);
      EnvPutInt(NAME_READAHEADSIZE, rAheadsiz);
   }

   // Read for the remote xrootd
   Int_t nr = fClient->Read(buffer, fOffset, bufferLength);

   if (!nr)
      return kTRUE;

   if (gDebug > 1)
      Info("ReadBuffer", "%d bytes of data read from offset"
                         " %lld (%d requested)", nr, fOffset, bufferLength);

   fOffset += bufferLength;

   if (!st) {
     // Update the counters only if the block has not been prefetched
     fBytesRead += nr;
#ifdef WIN32
     SetFileBytesRead(GetFileBytesRead() + nr);
     SetFileReadCalls(GetFileReadCalls() + 1);
#else
     fgBytesRead += nr;
     fgReadCalls++;
#endif
   }

   if (gMonitoringWriter)
      gMonitoringWriter->SendFileReadProgress(this);

   return result;
}

//______________________________________________________________________________
Bool_t TXNetFile::ReadBufferAsync(Long64_t offs, Int_t bufferLength)
{
   // Implementation dealing with the xrootd server.
   // Returns kTRUE in case of errors.
   // This is the same as TXNetFile::ReadBuffer but using the async
   // call from xrootd

   if (IsZombie()) {
      Error("ReadBuffer", "ReadBuffer is not possible because object"
            " is in 'zombie' state");
      return kTRUE;
   }

   if (fIsRootd) {
      if (gDebug > 1)
         Error("ReadBufferAsync","Not supported for rootd");
      return kTRUE;
   }

   if (!IsOpen()) {
      Error("ReadBuffer","The remote file is not open");
      return kTRUE;
   }

   Bool_t result = kFALSE;

   if (bufferLength==0)
      return 0;

   SynchronizeCacheSize();

   // Read for the remote xrootd
   // This doesnt return the number of bytes read...
   // and even if it did we dont want to update fBytesRead
   // because that would be updated in the real read
   XReqErrorType nr = fClient->Read_Async(offs, bufferLength);

   if (nr != kOK)
      return kTRUE;

   fBytesRead += bufferLength;
#ifdef WIN32
   SetFileBytesRead(GetFileBytesRead() + bufferLength);
   SetFileReadCalls(GetFileReadCalls() + 1);
#else
   fgBytesRead += bufferLength;
   fgReadCalls++;
#endif

   if (gDebug > 1)
      Info("ReadBufferAsync", "%d bytes of data read request from offset"
                              " %lld", bufferLength, offs);
   return result;
}

//______________________________________________________________________________
Bool_t TXNetFile::ReadBuffers(char *buf,  Long64_t *pos, Int_t *len, Int_t nbuf)
{'fraction_non_alphanumeric': 0.12658700522778193, 'fraction_numerical': 0.005850136918098083, 'mean_word_length': 1.8679231812665096, 'pattern_counts': {'":': 2, '<': 29, '<?xml version=': 0, '>': 133, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 76, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5593503', 'n_tokens_mistral': 11645, 'n_tokens_neox': 11169, 'n_words': 3336}