xrootd
XrdTpcTPC.hh
Go to the documentation of this file.
1 
2 #include <memory>
3 #include <string>
4 #include <vector>
5 #include <sys/time.h>
6 
8 
10 #include "XrdHttp/XrdHttpUtils.hh"
11 
12 #include "XrdTls/XrdTlsTempCA.hh"
13 
14 #include <curl/curl.h>
15 
16 class XrdOucErrInfo;
17 class XrdOucStream;
18 class XrdSfsFile;
19 class XrdSfsFileSystem;
20 class XrdXrootdTpcMon;
21 typedef void CURL;
22 
23 namespace TPC {
24 class State;
25 
26 enum LogMask {
27  Debug = 0x01,
28  Info = 0x02,
29  Warning = 0x04,
30  Error = 0x08,
31  All = 0xff
32 };
33 
34 
35 struct CurlDeleter {
36  void operator()(CURL *curl);
37 };
38 using ManagedCurlHandle = std::unique_ptr<CURL, CurlDeleter>;
39 
40 
41 class TPCHandler : public XrdHttpExtHandler {
42 public:
43  TPCHandler(XrdSysError *log, const char *config, XrdOucEnv *myEnv);
44  virtual ~TPCHandler();
45 
46  virtual bool MatchesPath(const char *verb, const char *path);
47  virtual int ProcessReq(XrdHttpExtReq &req);
48  // Abstract method in the base class, but does not seem to be used
49  virtual int Init(const char *cfgfile) {return 0;}
50 
51 private:
52 
53  static int sockopt_setcloexec_callback(void * clientp, curl_socket_t curlfd, curlsocktype purpose);
54  static int opensocket_callback(void *clientp,
55  curlsocktype purpose,
56  struct curl_sockaddr *address);
57 
58  struct TPCLogRecord {
59 
61  tpc_status(-1), streams( 1 ), isIPv6(false)
62  {
63  gettimeofday(&begT, 0); // Set effective start time
64  }
66 
67  std::string log_prefix;
68  std::string local;
69  std::string remote;
70  std::string name;
71  std::string clID;
73  timeval begT;
75  int status;
77  unsigned int streams;
78  bool isIPv6;
79  };
80 
82 
83  static std::string GetAuthz(XrdHttpExtReq &req);
84 
85  // Configure curl handle's CA settings. The CA files present here should
86  // be valid for the lifetime of the process.
87  void ConfigureCurlCA(CURL *curl);
88 
89  // Redirect the transfer according to the contents of an XrdOucErrInfo object.
90  int RedirectTransfer(CURL *curl, const std::string &redirect_resource, XrdHttpExtReq &req,
91  XrdOucErrInfo &error, TPCLogRecord &);
92 
93  int OpenWaitStall(XrdSfsFile &fh, const std::string &resource, int mode,
94  int openMode, const XrdSecEntity &sec,
95  const std::string &authz);
96 
97 #ifdef XRD_CHUNK_RESP
98  int DetermineXferSize(CURL *curl, XrdHttpExtReq &req, TPC::State &state,
99  bool &success, TPCLogRecord &, bool shouldReturnErrorToClient = true);
100 
101  int GetContentLengthTPCPull(CURL *curl, XrdHttpExtReq &req, uint64_t & contentLength, bool & success, TPCLogRecord &rec);
102 
103  // Send a 'performance marker' back to the TPC client, informing it of our
104  // progress. The TPC client will use this information to determine whether
105  // the transfer is making sufficient progress and/or other monitoring info
106  // (such as whether the transfer is happening over IPv4, IPv6, or both).
107  int SendPerfMarker(XrdHttpExtReq &req, TPCLogRecord &rec, TPC::State &state);
108  int SendPerfMarker(XrdHttpExtReq &req, TPCLogRecord &rec, std::vector<State*> &state,
109  off_t bytes_transferred);
110 
111  // Perform the libcurl transfer, periodically sending back chunked updates.
112  int RunCurlWithUpdates(CURL *curl, XrdHttpExtReq &req, TPC::State &state,
113  TPCLogRecord &rec);
114 
115  // Experimental multi-stream version of RunCurlWithUpdates
116  int RunCurlWithStreams(XrdHttpExtReq &req, TPC::State &state,
117  size_t streams, TPCLogRecord &rec);
118  int RunCurlWithStreamsImpl(XrdHttpExtReq &req, TPC::State &state,
119  size_t streams, std::vector<TPC::State*> &streams_handles,
120  std::vector<ManagedCurlHandle> &curl_handles,
121  TPCLogRecord &rec);
122 #else
123  int RunCurlBasic(CURL *curl, XrdHttpExtReq &req, TPC::State &state,
124  TPCLogRecord &rec);
125 #endif
126 
127  int ProcessPushReq(const std::string & resource, XrdHttpExtReq &req);
128  int ProcessPullReq(const std::string &resource, XrdHttpExtReq &req);
129 
130  bool ConfigureFSLib(XrdOucStream &Config, std::string &path1, bool &path1_alt,
131  std::string &path2, bool &path2_alt);
132  bool Configure(const char *configfn, XrdOucEnv *myEnv);
134 
135  // Generate a consistently-formatted log message.
136  void logTransferEvent(LogMask lvl, const TPCLogRecord &record,
137  const std::string &event, const std::string &message="");
138 
139  static int m_marker_period;
140  static size_t m_block_size;
141  static size_t m_small_block_size;
143  int m_timeout; // the 'timeout interval'; if no bytes have been received during this time period, abort the transfer.
144  int m_first_timeout; // the 'first timeout interval'; the amount of time we're willing to wait to get the first byte.
145  // Unless explicitly specified, this is 2x the timeout interval.
146  std::string m_cadir; // The directory to use for CAs.
147  std::string m_cafile; // The file to use for CAs in libcurl
149  static uint64_t m_monid;
152  std::shared_ptr<XrdTlsTempCA> m_ca_file;
153 
154  // 16 blocks in flight at 16 MB each, meaning that there will be up to 256MB
155  // in flight; this is equal to the bandwidth delay product of a 200ms transcontinental
156  // connection at 10Gbps.
157 #ifdef USE_PIPELINING
158  static const int m_pipelining_multiplier = 16;
159 #else
160  static const int m_pipelining_multiplier = 1;
161 #endif
162 
163  bool usingEC; // indicate if XrdEC is used
164 };
165 }
Utility functions for XrdHTTP.
void CURL
Definition: XrdTpcState.hh:13
void CURL
Definition: XrdTpcTPC.hh:20
Definition: XrdTpcState.hh:20
Definition: XrdTpcTPC.hh:41
int ProcessOptionsReq(XrdHttpExtReq &req)
std::string m_cadir
Definition: XrdTpcTPC.hh:146
TPCHandler(XrdSysError *log, const char *config, XrdOucEnv *myEnv)
bool ConfigureLogger(XrdOucStream &Config)
void logTransferEvent(LogMask lvl, const TPCLogRecord &record, const std::string &event, const std::string &message="")
static XrdSysMutex m_monid_mutex
Definition: XrdTpcTPC.hh:148
bool usingEC
Definition: XrdTpcTPC.hh:163
virtual ~TPCHandler()
static size_t m_small_block_size
Definition: XrdTpcTPC.hh:141
std::string m_cafile
Definition: XrdTpcTPC.hh:147
void ConfigureCurlCA(CURL *curl)
int m_timeout
Definition: XrdTpcTPC.hh:143
int ProcessPullReq(const std::string &resource, XrdHttpExtReq &req)
virtual int ProcessReq(XrdHttpExtReq &req)
static uint64_t m_monid
Definition: XrdTpcTPC.hh:149
static const int m_pipelining_multiplier
Definition: XrdTpcTPC.hh:160
bool m_desthttps
Definition: XrdTpcTPC.hh:142
static std::string GetAuthz(XrdHttpExtReq &req)
static int m_marker_period
Definition: XrdTpcTPC.hh:139
static int sockopt_setcloexec_callback(void *clientp, curl_socket_t curlfd, curlsocktype purpose)
int OpenWaitStall(XrdSfsFile &fh, const std::string &resource, int mode, int openMode, const XrdSecEntity &sec, const std::string &authz)
int RedirectTransfer(CURL *curl, const std::string &redirect_resource, XrdHttpExtReq &req, XrdOucErrInfo &error, TPCLogRecord &)
virtual bool MatchesPath(const char *verb, const char *path)
Tells if the incoming path is recognized as one of the paths that have to be processed.
virtual int Init(const char *cfgfile)
Initializes the external request handler.
Definition: XrdTpcTPC.hh:49
int ProcessPushReq(const std::string &resource, XrdHttpExtReq &req)
XrdSysError m_log
Definition: XrdTpcTPC.hh:150
static size_t m_block_size
Definition: XrdTpcTPC.hh:140
int RunCurlBasic(CURL *curl, XrdHttpExtReq &req, TPC::State &state, TPCLogRecord &rec)
bool ConfigureFSLib(XrdOucStream &Config, std::string &path1, bool &path1_alt, std::string &path2, bool &path2_alt)
XrdSfsFileSystem * m_sfs
Definition: XrdTpcTPC.hh:151
std::shared_ptr< XrdTlsTempCA > m_ca_file
Definition: XrdTpcTPC.hh:152
bool Configure(const char *configfn, XrdOucEnv *myEnv)
static int opensocket_callback(void *clientp, curlsocktype purpose, struct curl_sockaddr *address)
int m_first_timeout
Definition: XrdTpcTPC.hh:144
Definition: XrdHttpExtHandler.hh:82
Definition: XrdHttpExtHandler.hh:45
Definition: XrdOucEnv.hh:42
Definition: XrdOucErrInfo.hh:101
Definition: XrdOucStream.hh:47
Definition: XrdSecEntity.hh:64
Definition: XrdSfsInterface.hh:841
Definition: XrdSfsInterface.hh:365
Definition: XrdSysError.hh:90
Definition: XrdSysPthread.hh:165
Definition: XrdXrootdTpcMon.hh:41
Definition: XrdTpcState.hh:17
std::unique_ptr< CURL, CurlDeleter > ManagedCurlHandle
Definition: XrdTpcTPC.hh:38
LogMask
Definition: XrdTpcTPC.hh:26
@ All
Definition: XrdTpcTPC.hh:31
@ Info
Definition: XrdTpcTPC.hh:28
@ Error
Definition: XrdTpcTPC.hh:30
@ Debug
Definition: XrdTpcTPC.hh:27
@ Warning
Definition: XrdTpcTPC.hh:29
XrdCmsConfig Config
Definition: XrdTpcTPC.hh:35
void operator()(CURL *curl)
Definition: XrdTpcTPC.hh:58
timeval begT
Definition: XrdTpcTPC.hh:73
int tpc_status
Definition: XrdTpcTPC.hh:76
std::string remote
Definition: XrdTpcTPC.hh:69
TPCLogRecord()
Definition: XrdTpcTPC.hh:60
int status
Definition: XrdTpcTPC.hh:75
off_t bytes_transferred
Definition: XrdTpcTPC.hh:74
std::string name
Definition: XrdTpcTPC.hh:70
bool isIPv6
Definition: XrdTpcTPC.hh:78
std::string clID
Definition: XrdTpcTPC.hh:71
std::string local
Definition: XrdTpcTPC.hh:68
static XrdXrootdTpcMon * tpcMonitor
Definition: XrdTpcTPC.hh:72
std::string log_prefix
Definition: XrdTpcTPC.hh:67
unsigned int streams
Definition: XrdTpcTPC.hh:77