XRootD
XrdClHttpUtil.cc
Go to the documentation of this file.
1 /******************************************************************************/
2 /* Copyright (C) 2025, Pelican Project, Morgridge Institute for Research */
3 /* */
4 /* This file is part of the XrdClHttp client plugin for XRootD. */
5 /* */
6 /* XRootD is free software: you can redistribute it and/or modify it under */
7 /* the terms of the GNU Lesser General Public License as published by the */
8 /* Free Software Foundation, either version 3 of the License, or (at your */
9 /* option) any later version. */
10 /* */
11 /* XRootD is distributed in the hope that it will be useful, but WITHOUT */
12 /* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
13 /* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
14 /* License for more details. */
15 /* */
16 /* The copyright holder's institutional names and contributor's names may not */
17 /* be used to endorse or promote products derived from this software without */
18 /* specific prior written permission of the institution or contributor. */
19 /******************************************************************************/
20 
21 #include "XrdClHttpFile.hh"
22 #include "XrdClHttpOps.hh"
23 #include "XrdClHttpOptionsCache.hh"
24 #include "XrdClHttpUtil.hh"
25 #include "XrdClHttpWorker.hh"
26 
27 #include <XProtocol/XProtocol.hh>
28 #include <XrdCl/XrdClDefaultEnv.hh>
29 #include <XrdCl/XrdClLog.hh>
30 #include <XrdCl/XrdClURL.hh>
31 #include <XrdCl/XrdClXRootDResponses.hh>
32 #include <XrdOuc/XrdOucCRC.hh>
33 #include <XrdOuc/XrdOucPrivateUtils.hh>
34 #include <XrdSys/XrdSysPageSize.hh>
35 #include <XrdVersion.hh>
36 
37 #include <curl/curl.h>
38 #include <openssl/bio.h>
39 #include <openssl/evp.h>
40 
41 #include <fcntl.h>
42 #include <fstream>
43 #ifdef __APPLE__
44 #include <pthread.h>
45 #else
46 #include <sys/syscall.h>
47 #include <sys/types.h>
48 #endif
49 #include <unistd.h>
50 
51 #include <charconv>
52 #include <sstream>
53 #include <stdexcept>
54 #include <utility>
55 
56 using namespace XrdClHttp;
57 
58 thread_local std::vector<CURL*> HandlerQueue::m_handles;
59 std::atomic<unsigned> CurlWorker::m_maintenance_period = 5;
60 std::vector<std::unique_ptr<XrdClHttp::CurlWorker>> CurlWorker::m_workers;
61 std::mutex CurlWorker::m_workers_mutex;
62 
63 // Performance statistics for the worker
64 std::atomic<uint64_t> CurlWorker::m_conncall_errors = 0;
65 std::atomic<uint64_t> CurlWorker::m_conncall_req = 0;
66 std::atomic<uint64_t> CurlWorker::m_conncall_success = 0;
67 std::atomic<uint64_t> CurlWorker::m_conncall_timeout = 0;
68 decltype(CurlWorker::m_ops) CurlWorker::m_ops = {};
69 std::vector<std::atomic<std::chrono::system_clock::rep>*> CurlWorker::m_workers_last_completed_cycle;
70 std::vector<std::atomic<std::chrono::system_clock::rep>*> CurlWorker::m_workers_oldest_op;
71 std::mutex CurlWorker::m_worker_stats_mutex;
72 
73 // Performance statistics for the queue
74 std::atomic<uint64_t> HandlerQueue::m_ops_consumed = 0; // Count of operations consumed from the queue.
75 std::atomic<uint64_t> HandlerQueue::m_ops_produced = 0; // Count of operations added to the queue.
76 std::atomic<uint64_t> HandlerQueue::m_ops_rejected = 0; // Count of operations rejected by the queue.
77 
78 // shutdown + init trigger, must be last of the static members
79 CurlWorker::initcontrol CurlWorker::m_initcontrol;
80 
81 struct WaitingForBroker {
82  CURL *curl{nullptr};
83  time_t expiry{0};
84 };
85 
86 namespace {
87 
88 pid_t getthreadid() {
89 #if defined(__APPLE__)
90  uint64_t pth_threadid;
91  pthread_threadid_np(pthread_self(), &pth_threadid);
92  return pth_threadid;
93 #elif defined(__linux__)
94  // NOTE: glibc 2.30 finally provides a gettid() wrapper; however,
95  // we currently support RHEL 8, which is based on glibc 2.28. Until
96  // we drop that platform, it's easier to do the syscall directly on Linux
97  // instead of additional ifdef calls.
98  return syscall(SYS_gettid);
99 #else
100  return getpid();
101 #endif
102 }
103 
104 }
105 
106 bool XrdClHttp::HTTPStatusIsError(unsigned status) {
107  return (status < 100) || (status >= 400);
108 }
109 
110 std::pair<uint16_t, uint32_t> XrdClHttp::HTTPStatusConvert(unsigned status) {
111  switch (status) {
112  case 400: // Bad Request
113  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
114  case 401: // Unauthorized (needs authentication)
115  return std::make_pair(XrdCl::errErrorResponse, kXR_NotAuthorized);
116  case 402: // Payment Required
117  case 403: // Forbidden (failed authorization)
118  return std::make_pair(XrdCl::errErrorResponse, kXR_NotAuthorized);
119  case 404:
120  return std::make_pair(XrdCl::errErrorResponse, kXR_NotFound);
121  case 405: // Method not allowed
122  return std::make_pair(XrdCl::errErrorResponse, kXR_Unsupported);
123  case 406: // Not acceptable
124  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
125  case 407: // Proxy Authentication Required
126  return std::make_pair(XrdCl::errErrorResponse, kXR_NotAuthorized);
127  case 408: // Request timeout
128  return std::make_pair(XrdCl::errErrorResponse, kXR_ReqTimedOut);
129  case 409: // Conflict
130  return std::make_pair(XrdCl::errErrorResponse, kXR_Conflict);
131  case 410: // Gone
132  return std::make_pair(XrdCl::errErrorResponse, kXR_NotFound);
133  case 411: // Length required
134  case 412: // Precondition failed
135  case 413: // Payload too large
136  case 414: // URI too long
137  case 415: // Unsupported Media Type
138  case 416: // Range Not Satisfiable
139  case 417: // Expectation Failed
140  case 418: // I'm a teapot
141  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
142  case 421: // Misdirected Request
143  case 422: // Unprocessable Content
144  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
145  case 423: // Locked
146  return std::make_pair(XrdCl::errErrorResponse, kXR_FileLocked);
147  case 424: // Failed Dependency
148  case 425: // Too Early
149  case 426: // Upgrade Required
150  case 428: // Precondition Required
151  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
152  case 429: // Too Many Requests
153  return std::make_pair(XrdCl::errErrorResponse, kXR_Overloaded);
154  case 431: // Request Header Fields Too Large
155  return std::make_pair(XrdCl::errErrorResponse, kXR_InvalidRequest);
156  case 451: // Unavailable For Legal Reasons
157  return std::make_pair(XrdCl::errErrorResponse, kXR_Impossible);
158  case 500: // Internal Server Error
159  return std::make_pair(XrdCl::errErrorResponse, kXR_ServerError);
160  case 501: // Not Implemented
161  return std::make_pair(XrdCl::errErrorResponse, kXR_Unsupported);
162  case 502: // Bad Gateway
163  return std::make_pair(XrdCl::errErrorResponse, kXR_ServerError);
164  case 503: // Service Unavailable
165  return std::make_pair(XrdCl::errErrorResponse, kXR_Overloaded);
166  case 504: // Gateway Timeout
167  return std::make_pair(XrdCl::errErrorResponse, kXR_ReqTimedOut);
168  case 507: // Insufficient Storage
169  return std::make_pair(XrdCl::errErrorResponse, kXR_overQuota);
170  case 508: // Loop Detected
171  case 510: // Not Extended
172  case 511: // Network Authentication Required
173  return std::make_pair(XrdCl::errErrorResponse, kXR_ServerError);
174  }
175  return std::make_pair(XrdCl::errUnknown, status);
176 }
177 
178 std::pair<uint16_t, uint32_t> CurlCodeConvert(CURLcode res) {
179  switch (res) {
180  case CURLE_OK:
181  return std::make_pair(XrdCl::errNone, 0);
182  case CURLE_COULDNT_RESOLVE_PROXY:
183  case CURLE_COULDNT_RESOLVE_HOST:
184  return std::make_pair(XrdCl::errInvalidAddr, 0);
185  case CURLE_LOGIN_DENIED:
186  // Commented-out cases are for platforms (RHEL7) where the error
187  // codes are undefined.
188  //case CURLE_AUTH_ERROR:
189  //case CURLE_SSL_CLIENTCERT:
190  case CURLE_REMOTE_ACCESS_DENIED:
191  return std::make_pair(XrdCl::errLoginFailed, EACCES);
192  case CURLE_SSL_CONNECT_ERROR:
193  case CURLE_SSL_ENGINE_NOTFOUND:
194  case CURLE_SSL_ENGINE_SETFAILED:
195  case CURLE_SSL_CERTPROBLEM:
196  case CURLE_SSL_CIPHER:
197  case 51: // In old curl versions, this is CURLE_PEER_FAILED_VERIFICATION; that constant was changed to be 60 / CURLE_SSL_CACERT
198  case CURLE_SSL_SHUTDOWN_FAILED:
199  case CURLE_SSL_CRL_BADFILE:
200  case CURLE_SSL_ISSUER_ERROR:
201  case CURLE_SSL_CACERT: // value is 60; merged with CURLE_PEER_FAILED_VERIFICATION
202  //case CURLE_SSL_PINNEDPUBKEYNOTMATCH:
203  //case CURLE_SSL_INVALIDCERTSTATUS:
204  return std::make_pair(XrdCl::errTlsError, 0);
205  case CURLE_SEND_ERROR:
206  case CURLE_RECV_ERROR:
207  return std::make_pair(XrdCl::errSocketError, EIO);
208  case CURLE_COULDNT_CONNECT:
209  case CURLE_GOT_NOTHING:
210  return std::make_pair(XrdCl::errConnectionError, ECONNREFUSED);
211  case CURLE_OPERATION_TIMEDOUT:
212 #ifdef HAVE_XPROTOCOL_TIMEREXPIRED
213  return std::make_pair(XrdCl::errErrorResponse, XErrorCode::kXR_TimerExpired);
214 #else
215  return std::make_pair(XrdCl::errOperationExpired, ESTALE);
216 #endif
217  case CURLE_UNSUPPORTED_PROTOCOL:
218  case CURLE_NOT_BUILT_IN:
219  return std::make_pair(XrdCl::errNotSupported, ENOSYS);
220  case CURLE_FAILED_INIT:
221  return std::make_pair(XrdCl::errInternal, 0);
222  case CURLE_URL_MALFORMAT:
223  return std::make_pair(XrdCl::errInvalidArgs, res);
224  //case CURLE_WEIRD_SERVER_REPLY:
225  //case CURLE_HTTP2:
226  //case CURLE_HTTP2_STREAM:
227  return std::make_pair(XrdCl::errCorruptedHeader, res);
228  case CURLE_PARTIAL_FILE:
229  return std::make_pair(XrdCl::errDataError, res);
230  // These two errors indicate a failure in the callback. That
231  // should generate their own failures, meaning this should never
232  // get use.
233  case CURLE_READ_ERROR:
234  case CURLE_WRITE_ERROR:
235  return std::make_pair(XrdCl::errInternal, res);
236  case CURLE_RANGE_ERROR:
237  case CURLE_BAD_CONTENT_ENCODING:
238  return std::make_pair(XrdCl::errNotSupported, res);
239  case CURLE_TOO_MANY_REDIRECTS:
240  return std::make_pair(XrdCl::errRedirectLimit, res);
241  default:
242  return std::make_pair(XrdCl::errUnknown, res);
243  }
244 }
245 
246 bool HeaderParser::Base64Decode(std::string_view input, std::array<unsigned char, 32> &output) {
247  if (input.size() > 44 || input.size() % 4 != 0) return false;
248  if (input.size() == 0) return true;
249 
250  std::unique_ptr<BIO, decltype(&BIO_free_all)> b64(BIO_new(BIO_f_base64()), &BIO_free_all);
251  BIO_set_flags(b64.get(), BIO_FLAGS_BASE64_NO_NL);
252  std::unique_ptr<BIO, decltype(&BIO_free_all)> bmem(
253  BIO_new_mem_buf(const_cast<char *>(input.data()), input.size()), &BIO_free_all);
254  bmem.reset(BIO_push(b64.release(), bmem.release()));
255 
256  // Compute expected length of output; used to verify BIO_read consumes all input
257  size_t expectedLen = static_cast<size_t>(input.size() * 0.75);
258  if (input[input.size() - 1] == '=') {
259  expectedLen -= 1;
260  if (input[input.size() - 2] == '=') {
261  expectedLen -= 1;
262  }
263  }
264 
265  auto len = BIO_read(bmem.get(), &output[0], output.size());
266 
267  if (len == -1 || static_cast<size_t>(len) != expectedLen) return false;
268 
269  return true;
270 }
271 
272 // Parse a single header line.
273 //
274 // Curl promises for its callbacks "The header callback is
275 // called once for each header and only complete header lines
276 // are passed on to the callback".
277 bool HeaderParser::Parse(const std::string &header_line)
278 {
279  if (m_recv_all_headers) {
280  m_recv_all_headers = false;
281  m_recv_status_line = false;
282  }
283 
284  if (!m_recv_status_line) {
285  m_recv_status_line = true;
286 
287  std::stringstream ss(header_line);
288  std::string item;
289  if (!std::getline(ss, item, ' ')) return false;
290  m_resp_protocol = item;
291  if (!std::getline(ss, item, ' ')) return false;
292  try {
293  m_status_code = std::stol(item);
294  } catch (...) {
295  return false;
296  }
297  if (m_status_code < 100 || m_status_code >= 600) {
298  return false;
299  }
300  if (!std::getline(ss, item, '\n')) return false;
301  auto cr_loc = item.find('\r');
302  if (cr_loc != std::string::npos) {
303  m_resp_message = item.substr(0, cr_loc);
304  } else {
305  m_resp_message = item;
306  }
307  return true;
308  }
309 
310  if (header_line.empty() || header_line == "\n" || header_line == "\r\n") {
311  m_recv_all_headers = true;
312  return true;
313  }
314 
315  auto found = header_line.find(":");
316  if (found == std::string::npos) {
317  return false;
318  }
319 
320  std::string header_name = header_line.substr(0, found);
321  if (!Canonicalize(header_name)) {
322  return false;
323  }
324 
325  found += 1;
326  while (found < header_line.size()) {
327  if (header_line[found] != ' ') {break;}
328  found += 1;
329  }
330  std::string header_value = header_line.substr(found);
331  // Note: ignoring the fact headers are only supposed to contain ASCII.
332  // We should trim out UTF-8.
333  header_value.erase(header_value.find_last_not_of(" \r\n\t") + 1);
334 
335  // Record the line in our header structure. Will be returned as part
336  // of the response info object.
337  auto iter = m_headers.find(header_name);
338  if (iter == m_headers.end()) {
339  m_headers.insert(iter, {header_name, {header_value}});
340  } else {
341  iter->second.push_back(header_value);
342  }
343 
344  if (header_name == "Allow") {
345  std::string_view val(header_value);
346  while (!val.empty()) {
347  auto found = val.find(',');
348  auto method = val.substr(0, found);
349  if (method == "PROPFIND") {
350  auto new_verbs = static_cast<unsigned>(m_allow_verbs) | static_cast<unsigned>(VerbsCache::HttpVerb::kPROPFIND);
351  m_allow_verbs = static_cast<VerbsCache::HttpVerb>(new_verbs);
352  }
353  if (found == std::string_view::npos) break;
354  val = val.substr(found + 1);
355  }
356  if (static_cast<unsigned>(m_allow_verbs) & ~static_cast<unsigned>(VerbsCache::HttpVerb::kUnknown)) {
357  m_allow_verbs = static_cast<VerbsCache::HttpVerb>(static_cast<unsigned>(m_allow_verbs) & ~static_cast<unsigned>(VerbsCache::HttpVerb::kUnknown));
358  }
359  } else if (header_name == "Content-Length") {
360  try {
361  m_content_length = std::stoll(header_value);
362  } catch (...) {
363  return false;
364  }
365  }
366  else if (header_name == "Content-Type") {
367  std::string_view val(header_value);
368  auto found = val.find(";");
369  auto first_type = val.substr(0, found);
370  m_multipart_byteranges = first_type == "multipart/byteranges";
371  if (m_multipart_byteranges) {
372  auto remainder = val.substr(found + 1);
373  found = remainder.find("boundary=");
374  if (found != std::string_view::npos) {
375  SetMultipartSeparator(remainder.substr(found + 9));
376  }
377  }
378  }
379  else if (header_name == "Content-Range") {
380  auto found = header_value.find(" ");
381  if (found == std::string::npos) {
382  return false;
383  }
384  std::string range_unit = header_value.substr(0, found);
385  if (range_unit != "bytes") {
386  return false;
387  }
388  auto range_resp = header_value.substr(found + 1);
389  found = range_resp.find("/");
390  if (found == std::string::npos) {
391  return false;
392  }
393  auto incl_range = range_resp.substr(0, found);
394  found = incl_range.find("-");
395  if (found == std::string::npos) {
396  return false;
397  }
398  auto first_pos = incl_range.substr(0, found);
399  try {
400  m_response_offset = std::stoll(first_pos);
401  } catch (...) {
402  return false;
403  }
404  auto last_pos = incl_range.substr(found + 1);
405  size_t last_byte;
406  try {
407  last_byte = std::stoll(last_pos);
408  } catch (...) {
409  return false;
410  }
411  m_content_length = last_byte - m_response_offset + 1;
412  }
413  else if (header_name == "Location") {
414  m_location = header_value;
415  } else if (header_name == "Digest") {
416  ParseDigest(header_value, m_checksums);
417  }
418  else if (header_name == "Etag")
419  {
420  // Note, the original hader name is ETag, renamed to Etag in parsing
421  // remove additional quotes
422  m_etag = header_value;
423  m_etag.erase(remove(m_etag.begin(), m_etag.end(), '\"'), m_etag.end());
424  }
425  else if (header_name == "Cache-Control")
426  {
427  m_cache_control = header_value;
428  }
429 
430  return true;
431 }
432 
433 // Parse a RFC 3230 header into the checksum info structure
434 //
435 // If the parsing fails, the second element of the tuple will be false.
436 void HeaderParser::ParseDigest(const std::string &digest, XrdClHttp::ChecksumInfo &info) {
437  std::string_view view(digest);
438  std::array<unsigned char, 32> checksum_value;
439  std::string digest_lower;
440  while (!view.empty()) {
441  auto nextsep = view.find(',');
442  auto entry = view.substr(0, nextsep);
443  if (nextsep == std::string_view::npos) {
444  view = "";
445  } else {
446  view = view.substr(nextsep + 1);
447  }
448  nextsep = entry.find('=');
449  auto name = entry.substr(0, nextsep);
450  auto value = entry.substr(nextsep + 1);
451  digest_lower.clear();
452  digest_lower.resize(name.size());
453  std::transform(name.begin(), name.end(), digest_lower.begin(), [](unsigned char c) {
454  return std::tolower(c);
455  });
456  if (digest_lower == "md5") {
457  if (value.size() != 24) {
458  continue;
459  }
460  if (Base64Decode(value, checksum_value)) {
461  info.Set(XrdClHttp::ChecksumType::kMD5, checksum_value);
462  }
463  } else if (digest_lower == "crc32c") {
464  // XRootD currently incorrectly base64-encodes crc32c checksums; see
465  // https://github.com/xrootd/xrootd/issues/2456
466  // For backward comaptibility, if this looks like base64 encoded (8
467  // bytes long and last two bytes are padding), then we base64 decode.
468  if (value.size() == 8 && value[6] == '=' && value[7] == '=') {
469  if (Base64Decode(value, checksum_value)) {
470  info.Set(XrdClHttp::ChecksumType::kCRC32C, checksum_value);
471  }
472  continue;
473  }
474  std::size_t pos{0};
475  unsigned long val;
476  try {
477  val = std::stoul(value.data(), &pos, 16);
478  } catch (...) {
479  continue;
480  }
481  if (pos == value.size()) {
482  checksum_value[0] = (val >> 24) & 0xFF;
483  checksum_value[1] = (val >> 16) & 0xFF;
484  checksum_value[2] = (val >> 8) & 0xFF;
485  checksum_value[3] = val & 0xFF;
486  info.Set(XrdClHttp::ChecksumType::kCRC32C, checksum_value);
487  }
488  }
489  }
490 }
491 
492 // Convert the checksum type to a RFC 3230 digest name as recorded by IANA here:
493 // https://www.iana.org/assignments/http-dig-alg/http-dig-alg.xhtml
494 std::string HeaderParser::ChecksumTypeToDigestName(XrdClHttp::ChecksumType type) {
495  switch (type) {
496  case XrdClHttp::ChecksumType::kMD5:
497  return "MD5";
498  case XrdClHttp::ChecksumType::kCRC32C:
499  return "CRC32c";
500  case XrdClHttp::ChecksumType::kSHA1:
501  return "SHA";
502  case XrdClHttp::ChecksumType::kSHA256:
503  return "SHA-256";
504  default:
505  return "";
506  }
507 }
508 
509 // This clever approach was inspired by golang's net/textproto
510 bool HeaderParser::validHeaderByte(unsigned char c)
511 {
512  const static uint64_t mask_lower = 0 |
513  uint64_t((1<<10)-1) << '0' |
514  uint64_t(1) << '!' |
515  uint64_t(1) << '#' |
516  uint64_t(1) << '$' |
517  uint64_t(1) << '%' |
518  uint64_t(1) << '&' |
519  uint64_t(1) << '\'' |
520  uint64_t(1) << '*' |
521  uint64_t(1) << '+' |
522  uint64_t(1) << '-' |
523  uint64_t(1) << '.';
524 
525  const static uint64_t mask_upper = 0 |
526  uint64_t((1<<26)-1) << ('a'-64) |
527  uint64_t((1<<26)-1) << ('A'-64) |
528  uint64_t(1) << ('^'-64) |
529  uint64_t(1) << ('_'-64) |
530  uint64_t(1) << ('`'-64) |
531  uint64_t(1) << ('|'-64) |
532  uint64_t(1) << ('~'-64);
533 
534  if (c >= 128) return false;
535  if (c >= 64) return (uint64_t(1)<<(c-64)) & mask_upper;
536  return (uint64_t(1) << c) & mask_lower;
537 }
538 
539 bool HeaderParser::Canonicalize(std::string &headerName)
540 {
541  auto upper = true;
542  const static int toLower = 'a' - 'A';
543  for (size_t idx=0; idx<headerName.size(); idx++) {
544  char c = headerName[idx];
545  if (!validHeaderByte(c)) {
546  return false;
547  }
548  if (upper && 'a' <= c && c <= 'z') {
549  c -= toLower;
550  } else if (!upper && 'A' <= c && c <= 'Z') {
551  c += toLower;
552  }
553  headerName[idx] = c;
554  upper = c == '-';
555  }
556  return true;
557 }
558 
559 HandlerQueue::HandlerQueue(unsigned max_pending_ops) :
560  m_max_pending_ops(max_pending_ops)
561 {
562  int filedes[2];
563  auto result = pipe(filedes);
564  if (result == -1) {
565  throw std::runtime_error(strerror(errno));
566  }
567  if (fcntl(filedes[0], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1 || fcntl(filedes[1], F_SETFL, O_NONBLOCK | O_CLOEXEC) == -1) {
568  close(filedes[0]);
569  close(filedes[1]);
570  throw std::runtime_error(strerror(errno));
571  }
572  m_read_fd = filedes[0];
573  m_write_fd = filedes[1];
574 };
575 
576 namespace {
577 
578 bool EnableCurlHeaderDump() {
579  auto *log = XrdCl::DefaultEnv::GetLog();
580  if (log && log->GetLevel() >= XrdCl::Log::DumpMsg)
581  return true;
582 
583  return false;
584 }
585 
586 // Debug callback for libcurl headers; enabled with XRD_LOGLEVEL=Dump
587 int DumpHeader(CURL *handle, curl_infotype type, char *data, size_t size, void *clientp) {
588  (void)handle;
589  auto *logger = static_cast<XrdCl::Log *>(clientp);
590  if (!logger || !data || size == 0) {
591  return 0;
592  }
593 
594  const char *direction = nullptr;
595  switch (type) {
596  case CURLINFO_HEADER_OUT:
597  direction = ">";
598  break;
599  case CURLINFO_HEADER_IN:
600  direction = "<";
601  break;
602  default:
603  return 0;
604  }
605 
606  const std::string redacted = obfuscateAuth(std::string(data, size));
607  logger->Debug(kLogXrdClHttp, "%s %s", direction, redacted.c_str());
608  return 0;
609 }
610 
611 }
612 
613 // Trim left and right side of a string_view for space characters
614 std::string_view XrdClHttp::trim_view(const std::string_view &input_view) {
615  auto view = XrdClHttp::ltrim_view(input_view);
616  for (size_t idx = 0; idx < input_view.size(); idx++) {
617  if (!isspace(view[view.size() - 1 - idx])) {
618  return view.substr(0, view.size() - idx);
619  }
620  }
621  return "";
622 }
623 
624 // Trim the left side of a string_view for space
625 std::string_view XrdClHttp::ltrim_view(const std::string_view &input_view) {
626  for (size_t idx = 0; idx < input_view.size(); idx++) {
627  if (!isspace(input_view[idx])) {
628  return input_view.substr(idx);
629  }
630  }
631  return "";
632 }
633 
634 CURL *
635 XrdClHttp::GetHandle(bool verbose) {
636  auto result = curl_easy_init();
637  if (result == nullptr) {
638  return result;
639  }
640 
641  curl_easy_setopt(result, CURLOPT_USERAGENT, "xrdcl-http/" XrdVERSION);
642  curl_easy_setopt(result, CURLOPT_DEBUGFUNCTION, DumpHeader);
643  curl_easy_setopt(result, CURLOPT_DEBUGDATA, XrdCl::DefaultEnv::GetLog());
644  if (verbose)
645  curl_easy_setopt(result, CURLOPT_VERBOSE, 1L);
646 
647  auto env = XrdCl::DefaultEnv::GetEnv();
648  std::string ca_file;
649  if (!env->GetString("HttpCertFile", ca_file) || ca_file.empty()) {
650  char *x509_ca_file = getenv("X509_CERT_FILE");
651  if (x509_ca_file) {
652  ca_file = std::string(x509_ca_file);
653  }
654  }
655  if (!ca_file.empty()) {
656  curl_easy_setopt(result, CURLOPT_CAINFO, ca_file.c_str());
657  }
658  std::string ca_dir;
659  if (!env->GetString("HttpCertDir", ca_dir) || ca_dir.empty()) {
660  char *x509_ca_dir = getenv("X509_CERT_DIR");
661  if (x509_ca_dir) {
662  ca_dir = std::string(x509_ca_dir);
663  }
664  }
665  if (!ca_dir.empty()) {
666  curl_easy_setopt(result, CURLOPT_CAPATH, ca_dir.c_str());
667  }
668 
669  curl_easy_setopt(result, CURLOPT_BUFFERSIZE, 32*1024);
670 
671  return result;
672 }
673 
674 CURL *
675 HandlerQueue::GetHandle() {
676  if (m_handles.size()) {
677  auto result = m_handles.back();
678  m_handles.pop_back();
679  return result;
680  }
681 
682  return ::GetHandle(EnableCurlHeaderDump());
683 }
684 
685 void
686 HandlerQueue::RecycleHandle(CURL *curl) {
687  m_handles.push_back(curl);
688 }
689 
690 void
691 HandlerQueue::Expire()
692 {
693  std::unique_lock<std::mutex> lk(m_mutex);
694  auto now = std::chrono::steady_clock::now();
695 
696  // Iterate through the paused transfers, checking if they are done.
697  for (auto &op : m_ops) {
698  if (!op->IsPaused()) continue;
699 
700  if (op->TransferStalled(0, now)) {
701  op->ContinueHandle();
702  }
703  }
704 
705  std::vector<decltype(m_ops)::value_type> expired_ops;
706  unsigned expired_count = 0;
707  auto it = std::remove_if(m_ops.begin(), m_ops.end(),
708  [&](const std::shared_ptr<CurlOperation> &handler) {
709  auto expired = handler->GetOperationExpiry() < now;
710  if (expired) {
711  expired_ops.push_back(handler);
712  expired_count++;
713  }
714  return expired;
715  });
716  m_ops.erase(it, m_ops.end());
717 
718  // The contents of our pipe and the in-memory queue are now off by expired_count.
719  // Read exactly that many bytes from the pipe and throw them away.
720  char throwaway[64];
721  unsigned bytes_to_read = expired_count;
722  while (bytes_to_read > 0) {
723  size_t chunk = std::min<size_t>(sizeof(throwaway), bytes_to_read);
724  ssize_t n = read(m_read_fd, throwaway, chunk);
725  if (n > 0) {
726  bytes_to_read -= n;
727  } else if (n == -1) {
728  if (errno == EINTR) {
729  continue;
730  } else {
731  // EWOULDBLOCK is a possibility if there's a synchronization error;
732  // for now, just continue on as if we were successful in reading out
733  // the missing bytes
734  break;
735  }
736  } else {
737  break;
738  }
739  }
740 
741  // Note: the failure handler may trigger new operations submitted to the queue
742  // (which requires the lock to be held) such as a prefetch operation that gets split
743  // into multiple sub-operations.
744  //
745  // Thus, we must unlock the mutex protecting the queue and avoid touching the shared state of
746  // m_ops.
747  lk.unlock();
748  for (auto &handler : expired_ops) {
749  if (handler) handler->Fail(XrdCl::errOperationExpired, 0, "Operation expired while in queue");
750  }
751 }
752 
753 void
754 HandlerQueue::Produce(std::shared_ptr<CurlOperation> handler)
755 {
756  auto handler_expiry = handler->GetOperationExpiry();
757  std::unique_lock<std::mutex> lk{m_mutex};
758  m_producer_cv.wait_until(lk,
759  handler_expiry,
760  [&]{return m_ops.size() < m_max_pending_ops;}
761  );
762  if (std::chrono::steady_clock::now() > handler_expiry) {
763  lk.unlock();
764  handler->Fail(XrdCl::errOperationExpired, 0, "Operation expired while waiting for worker");
765  m_ops_rejected.fetch_add(1, std::memory_order_relaxed);
766  return;
767  }
768 
769  m_ops.push_back(handler);
770  char ready[] = "1";
771  while (true) {
772  auto result = write(m_write_fd, ready, 1);
773  if (result == -1) {
774  if (errno == EINTR) {
775  continue;
776  } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
777  // This should never happen, but if it does, just continue
778  // as if we successfully wrote the notification to the pipe.
779  break;
780  }
781  throw std::runtime_error(strerror(errno));
782  }
783  break;
784  }
785 
786  lk.unlock();
787  m_consumer_cv.notify_one();
788  m_ops_produced.fetch_add(1, std::memory_order_relaxed);
789 }
790 
791 std::shared_ptr<CurlOperation>
792 HandlerQueue::Consume(std::chrono::steady_clock::duration dur)
793 {
794  std::unique_lock<std::mutex> lk(m_mutex);
795  m_consumer_cv.wait_for(lk, dur, [&]{return m_ops.size() > 0 || m_shutdown;});
796  if (m_shutdown || m_ops.empty()) {
797  return {};
798  }
799 
800  std::shared_ptr<CurlOperation> result = m_ops.front();
801  m_ops.pop_front();
802 
803  char ready[1];
804  while (true) {
805  auto result = read(m_read_fd, ready, 1);
806  if (result == -1) {
807  if (errno == EINTR) {
808  continue;
809  } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
810  // This should never happen, but if it does, just continue
811  // as if we successfully read the byte.
812  break;
813  }
814  throw std::runtime_error(strerror(errno));
815  }
816  break;
817  }
818 
819  lk.unlock();
820  m_producer_cv.notify_one();
821  m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
822 
823  return result;
824 }
825 
826 std::string
827 HandlerQueue::GetMonitoringJson()
828 {
829  auto consumed = m_ops_consumed.load(std::memory_order_relaxed);
830  auto produced = m_ops_produced.load(std::memory_order_relaxed);
831  return "{"
832  "\"produced\":" + std::to_string(produced) + ","
833  "\"consumed\":" + std::to_string(consumed) + ","
834  "\"pending\":" + std::to_string(produced - consumed) + ","
835  "\"rejected\":" + std::to_string(m_ops_rejected.load(std::memory_order_relaxed)) +
836  "}";
837 }
838 
839 std::shared_ptr<CurlOperation>
840 HandlerQueue::TryConsume()
841 {
842  std::unique_lock<std::mutex> lk(m_mutex);
843  if (m_ops.size() == 0) {
844  std::shared_ptr<CurlOperation> result;
845  return result;
846  }
847 
848  std::shared_ptr<CurlOperation> result = m_ops.front();
849  m_ops.pop_front();
850 
851  char ready[1];
852  while (true) {
853  auto result = read(m_read_fd, ready, 1);
854  if (result == -1) {
855  if (errno == EINTR) {
856  continue;
857  } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
858  // This should never happen, but if it does, just continue
859  // as if we successfully read the byte.
860  break;
861  }
862  throw std::runtime_error(strerror(errno));
863  }
864  break;
865  }
866 
867  lk.unlock();
868  m_producer_cv.notify_one();
869  m_ops_consumed.fetch_add(1, std::memory_order_relaxed);
870 
871  return result;
872 }
873 
874 void
875 HandlerQueue::Shutdown()
876 {
877  std::unique_lock lock(m_mutex);
878  m_shutdown = true;
879  m_consumer_cv.notify_all();
880 }
881 
882 void
883 HandlerQueue::ReleaseHandles()
884 {
885  for (auto handle : m_handles) {
886  curl_easy_cleanup(handle);
887  }
888  m_handles.clear();
889 }
890 
891 CurlWorker::CurlWorker(std::shared_ptr<HandlerQueue> queue, VerbsCache &cache, XrdCl::Log* logger) :
892  m_cache(cache),
893  m_queue(queue),
894  m_logger(logger)
895 {
896  {
897  std::unique_lock lk(m_worker_stats_mutex);
898  m_stats_offset = m_workers_last_completed_cycle.size();
899  m_workers_last_completed_cycle.push_back(&m_last_completed_cycle);
900  m_workers_oldest_op.push_back(&m_oldest_op);
901  }
902  int pipeInfo[2];
903  if ((pipe(pipeInfo) == -1) || (fcntl(pipeInfo[0], F_SETFD, FD_CLOEXEC)) || (fcntl(pipeInfo[1], F_SETFD, FD_CLOEXEC))) {
904  throw std::runtime_error("Failed to create shutdown monitoring pipe for curl worker");
905  }
906  m_shutdown_pipe_r = pipeInfo[0];
907  m_shutdown_pipe_w = pipeInfo[1];
908 
909  // Handle setup of the X509 authentication
910  auto env = XrdCl::DefaultEnv::GetEnv();
911  env->GetString("HttpClientCertFile", m_x509_client_cert_file);
912  env->GetString("HttpClientKeyFile", m_x509_client_key_file);
913 }
914 
915 std::tuple<std::string, std::string> CurlWorker::ClientX509CertKeyFile() const
916 {
917  return std::make_tuple(m_x509_client_cert_file, m_x509_client_key_file);
918 }
919 
920 std::string
921 CurlWorker::GetMonitoringJson()
922 {
923  auto now = std::chrono::system_clock::now().time_since_epoch().count();
924  auto oldest_op = now;
925  auto oldest_cycle = now;
926  {
927  std::unique_lock lk(m_worker_stats_mutex);
928  for (const auto &entry : m_workers_last_completed_cycle) {
929  if (!entry) {continue;}
930  auto cycle = entry->load(std::memory_order_relaxed);
931  if (cycle < oldest_cycle) oldest_cycle = cycle;
932  }
933  for (const auto &entry : m_workers_oldest_op) {
934  if (!entry) {continue;}
935  auto op = entry->load(std::memory_order_relaxed);
936  if (op < oldest_op) oldest_op = op;
937  }
938  }
939  auto oldest_op_dbl = std::chrono::duration<double>(std::chrono::system_clock::time_point(std::chrono::system_clock::duration(oldest_op)).time_since_epoch()).count();
940  auto oldest_cycle_dbl = std::chrono::duration<double>(std::chrono::system_clock::time_point(std::chrono::system_clock::duration(oldest_cycle)).time_since_epoch()).count();
941  std::string retval = "{"
942  "\"oldest_op\":" + std::to_string(oldest_op_dbl) + ","
943  "\"oldest_cycle\":" + std::to_string(oldest_cycle_dbl) + ","
944  ;
945 
946  for (size_t verb_idx = 0; verb_idx < static_cast<int>(XrdClHttp::CurlOperation::HttpVerb::Count); verb_idx++) {
947  const auto &verb_str = XrdClHttp::CurlOperation::GetVerbString(static_cast<XrdClHttp::CurlOperation::HttpVerb>(verb_idx));
948  for (size_t op_idx = 0; op_idx < 402; op_idx++) {
949  if (op_idx == 401) continue;
950 
951  auto &op_stats = m_ops[verb_idx][op_idx];
952  auto duration = op_stats.m_duration.load(std::memory_order_relaxed);
953  if (duration == 0) continue;
954 
955  std::string prefix = "http_" + verb_str + "_" + ((op_idx == 402) ? "invalid" : std::to_string(200 + op_idx)) + "_";
956 
957  auto duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
958  retval += "\"" + prefix + "duration\":" + std::to_string(duration_dbl) + ",";
959 
960  duration = op_stats.m_pause_duration.load(std::memory_order_relaxed);
961  if (duration > 0) {
962  duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
963  retval += "\"" + prefix + "pause_duration\":" + std::to_string(duration_dbl) + ",";
964  }
965 
966  auto count = op_stats.m_bytes.load(std::memory_order_relaxed);
967  if (count) retval += "\"" + prefix + "bytes\":" + std::to_string(count) + ",";
968  count = op_stats.m_error.load(std::memory_order_relaxed);
969  if (count) retval += "\"" + prefix + "error\":" + std::to_string(count) + ",";
970  count = op_stats.m_finished.load(std::memory_order_relaxed);
971  if (count) retval += "\"" + prefix + "finished\":" + std::to_string(count) + ",";
972  count = op_stats.m_client_timeout.load(std::memory_order_relaxed);
973  if (count) retval += "\"" + prefix + "client_timeout\":" + std::to_string(count) + ",";
974  count = op_stats.m_server_timeout.load(std::memory_order_relaxed);
975  if (count) retval += "\"" + prefix + "server_timeout\":" + std::to_string(count) + ",";
976  }
977  {
978  auto &op_stats = m_ops[verb_idx][401];
979  auto duration = op_stats.m_duration.load(std::memory_order_relaxed);
980  if (duration == 0) continue;
981 
982  std::string prefix = "http_" + verb_str + "_";
983 
984  auto duration_dbl = std::chrono::duration<double>(std::chrono::steady_clock::duration(duration)).count();
985  retval += "\"" + prefix + "preheader_duration\":" + std::to_string(duration_dbl) + ",";
986 
987  auto count = op_stats.m_started.load(std::memory_order_relaxed);
988  if (count) retval += "\"" + prefix + "started\":" + std::to_string(count) + ",";
989  count = op_stats.m_error.load(std::memory_order_relaxed);
990  if (count) retval += "\"" + prefix + "preheader_error\":" + std::to_string(count) + ",";
991  count = op_stats.m_finished.load(std::memory_order_relaxed);
992  if (count) retval += "\"" + prefix + "preheader_finished\":" + std::to_string(count) + ",";
993  count = op_stats.m_server_timeout.load(std::memory_order_relaxed);
994  if (count) retval += "\"" + prefix + "preheader_timeout\":" + std::to_string(count) + ",";
995  count = op_stats.m_conncall_timeout.load(std::memory_order_relaxed);
996  if (count) retval += "\"" + prefix + "conncall_timeout\":" + std::to_string(count) + ",";
997  }
998  }
999 
1000  retval +=
1001  "\"conncall_error\":" + std::to_string(m_conncall_errors.load(std::memory_order_relaxed)) + ","
1002  "\"conncall_started\":" + std::to_string(m_conncall_req.load(std::memory_order_relaxed)) + ","
1003  "\"conncall_success\":" + std::to_string(m_conncall_success.load(std::memory_order_relaxed)) + ","
1004  "\"conncall_timeout\":" + std::to_string(m_conncall_timeout.load(std::memory_order_relaxed)) +
1005  "}";
1006 
1007  return retval;
1008 }
1009 
1010 void
1011 CurlWorker::OpRecord(XrdClHttp::CurlOperation &op, OpKind kind)
1012 {
1013  int sc = op.GetStatusCode();
1014  // - We encode everything pre-header as integer "401". We include a 100-continue request as "pre-header".
1015  // - Status codes out of the acceptable range are labeled "402"
1016  // - Otherwise, we store it in the array shifted by 200 (to avoid more sparsity)
1017  if (sc < 0 || kind == OpKind::Start || sc == 100) {
1018  sc = 401;
1019  } else if (sc < 200 || sc >= 600) {
1020  sc = 402;
1021  } else {
1022  sc -= 200;
1023  }
1024  auto [bytes, pre_headers, post_headers, pause_duration] = op.StatisticsReset();
1025  auto &op_stats = m_ops[static_cast<int>(op.GetVerb())][sc];
1026  op_stats.m_bytes.fetch_add(bytes, std::memory_order_relaxed);
1027  op_stats.m_duration.fetch_add((sc == 401) ? pre_headers.count() : post_headers.count(), std::memory_order_relaxed);
1028  op_stats.m_pause_duration.fetch_add(pause_duration.count(), std::memory_order_relaxed);
1029  if (pre_headers != std::chrono::steady_clock::duration::zero() && sc != 401) {
1030  auto &old_stats = m_ops[static_cast<int>(op.GetVerb())][401];
1031  old_stats.m_duration.fetch_add(pre_headers.count(), std::memory_order_relaxed);
1032  }
1033  switch (kind) {
1034  case OpKind::ConncallTimeout:
1035  op_stats.m_conncall_timeout.fetch_add(1, std::memory_order_relaxed);
1036  break;
1037  case OpKind::ClientTimeout:
1038  op_stats.m_client_timeout.fetch_add(1, std::memory_order_relaxed);
1039  break;
1040  case OpKind::Error:
1041  op_stats.m_error.fetch_add(1, std::memory_order_relaxed);
1042  break;
1043  case OpKind::Finish:
1044  op_stats.m_finished.fetch_add(1, std::memory_order_relaxed);
1045  break;
1046  case OpKind::Start:
1047  op_stats.m_started.fetch_add(1, std::memory_order_relaxed);
1048  break;
1049  case OpKind::ServerTimeout:
1050  op_stats.m_server_timeout.fetch_add(1, std::memory_order_relaxed);
1051  break;
1052  case OpKind::Update:
1053  break;
1054  }
1055 }
1056 
1057 void
1058 CurlWorker::Start(std::unique_ptr<XrdClHttp::CurlWorker> self, std::thread tid)
1059 {
1060  {
1061  std::unique_lock lock(m_workers_mutex);
1062  m_workers.emplace_back(std::move(self));
1063  m_self_tid = std::move(tid);
1064  }
1065  std::unique_lock lock(m_start_lock);
1066  m_start_complete = true;
1067  m_start_complete_cv.notify_one();
1068 }
1069 
1070 void
1071 CurlWorker::RunStatic(CurlWorker *myself)
1072 {
1073  {
1074  std::unique_lock lock(myself->m_start_lock);
1075  myself->m_start_complete_cv.wait(lock, [&]{return myself->m_start_complete;});
1076  }
1077  try {
1078  myself->Run();
1079  } catch (...) {
1080  myself->m_logger->Warning(kLogXrdClHttp, "Curl worker got an exception");
1081  {
1082  std::unique_lock lock(m_workers_mutex);
1083  auto iter = std::remove_if(m_workers.begin(), m_workers.end(), [&](std::unique_ptr<XrdClHttp::CurlWorker> &worker){return worker.get() == myself;});
1084  m_workers.erase(iter);
1085  }
1086  }
1087 }
1088 
1089 void
1090 CurlWorker::Run() {
1091  int max_pending = 50;
1092  XrdCl::DefaultEnv::GetEnv()->GetInt("HttpMaxPendingOps", max_pending);
1093  m_continue_queue.reset(new HandlerQueue(max_pending));
1094  auto &queue = *m_queue.get();
1095  m_logger->Debug(kLogXrdClHttp, "Started a curl worker");
1096 
1097  CURLM *multi_handle = curl_multi_init();
1098  if (multi_handle == nullptr) {
1099  throw std::runtime_error("Failed to create curl multi-handle");
1100  }
1101 
1102  int running_handles = 0;
1103  time_t last_maintenance = time(NULL);
1104  CURLMcode mres = CURLM_OK;
1105 
1106  // Map from a file descriptor that has an outstanding broker request
1107  // to the corresponding CURL handle.
1108  std::unordered_map<int, WaitingForBroker> broker_reqs;
1109  std::vector<struct curl_waitfd> waitfds;
1110 
1111  bool want_shutdown = false;
1112  while (!want_shutdown) {
1113  m_last_completed_cycle.store(std::chrono::system_clock::now().time_since_epoch().count());
1114  auto oldest_op = std::chrono::system_clock::now();
1115  for (const auto &entry : m_op_map) {
1116  OpRecord(*entry.second.first, OpKind::Update);
1117  if (entry.second.second < oldest_op) {
1118  oldest_op = entry.second.second;
1119  }
1120  }
1121  m_oldest_op.store(oldest_op.time_since_epoch().count());
1122 
1123  // Try continuing any available handles that have more data
1124  while (true) {
1125  auto op = m_continue_queue->TryConsume();
1126  if (!op) {
1127  break;
1128  }
1129  // Avoid race condition where external thread added a continue operation to queue
1130  // while the curl worker thread failed the transfer.
1131  if (op->IsDone()) {
1132  m_logger->Debug(kLogXrdClHttp, "Ignoring continuation of operation that has already completed");
1133  continue;
1134  }
1135  m_logger->Debug(kLogXrdClHttp, "Continuing the curl handle from op %p on thread %d", op.get(), getthreadid());
1136  auto curl = op->GetCurlHandle();
1137  if (!op->ContinueHandle()) {
1138  op->Fail(XrdCl::errInternal, 0, "Failed to continue the curl handle for the operation");
1139  OpRecord(*op, OpKind::Error);
1140  op->ReleaseHandle();
1141  if (curl) {
1142  curl_multi_remove_handle(multi_handle, curl);
1143  curl_easy_cleanup(curl);
1144  m_op_map.erase(curl);
1145  }
1146  running_handles -= 1;
1147  continue;
1148  } else {
1149  auto iter = m_op_map.find(curl);
1150  if (iter != m_op_map.end()) iter->second.second = std::chrono::system_clock::now();
1151  }
1152  }
1153  // Consume from the shared new operation queue
1154  while (running_handles < static_cast<int>(m_max_ops)) {
1155  auto op = running_handles == 0 ? queue.Consume(std::chrono::seconds(1)) : queue.TryConsume();
1156  if (!op) {
1157  break;
1158  }
1159  auto curl = queue.GetHandle();
1160  if (curl == nullptr) {
1161  m_logger->Debug(kLogXrdClHttp, "Unable to allocate a curl handle");
1162  op->Fail(XrdCl::errInternal, ENOMEM, "Unable to get allocate a curl handle");
1163  continue;
1164  }
1165  try {
1166  auto rv = op->Setup(curl, *this);
1167  if (!rv) {
1168  m_logger->Debug(kLogXrdClHttp, "Failed to setup the curl handle");
1169  op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the operation");
1170  continue;
1171  }
1172  if (!op->FinishSetup(curl)) {
1173  m_logger->Debug(kLogXrdClHttp, "Failed to finish setup of the curl handle");
1174  op->Fail(XrdCl::errInternal, ENOMEM, "Failed to finish setup of the curl handle for the operation");
1175  continue;
1176  }
1177  } catch (...) {
1178  m_logger->Debug(kLogXrdClHttp, "Unable to setup the curl handle");
1179  op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the operation");
1180  continue;
1181  }
1182  op->SetContinueQueue(m_continue_queue);
1183 
1184  if (op->IsDone()) {
1185  continue;
1186  }
1187  m_op_map[curl] = {op, std::chrono::system_clock::now()};
1188 
1189  // If the operation requires the result of the OPTIONS verb to function, then
1190  // we add that to the multi-handle instead, chaining the two calls together.
1191  if (op->RequiresOptions()) {
1192  std::string modified_url;
1193  std::shared_ptr<CurlOptionsOp> options_op(
1194  new CurlOptionsOp(
1195  curl, op,
1196  std::string(
1197  VerbsCache::GetUrlKey(op->GetUrl(), modified_url)
1198  ),
1199  m_logger, op->GetConnCalloutFunc()
1200  )
1201  );
1202  // Note this `curl` variable is not local to the conditional; it is the curl handle of the
1203  // CurlOptionsOp and will be added below to the multi-handle, causing it - not the parent's
1204  // curl handle - to be executed.
1205  curl = queue.GetHandle();
1206  if (curl == nullptr) {
1207  m_logger->Debug(kLogXrdClHttp, "Unable to allocate a curl handle");
1208  op->Fail(XrdCl::errInternal, ENOMEM, "Unable to get allocate a curl handle");
1209  OpRecord(*op, OpKind::Error);
1210  continue;
1211  }
1212  auto rv = options_op->Setup(curl, *this);
1213  if (!rv) {
1214  m_logger->Debug(kLogXrdClHttp, "Failed to allocate a curl handle for OPTIONS");
1215  continue;
1216  }
1217  m_op_map[curl] = {options_op, std::chrono::system_clock::now()};
1218  OpRecord(*options_op, OpKind::Start);
1219  running_handles += 1;
1220  } else {
1221  OpRecord(*op, OpKind::Start);
1222  }
1223 
1224  auto mres = curl_multi_add_handle(multi_handle, curl);
1225  if (mres != CURLM_OK) {
1226  m_logger->Debug(kLogXrdClHttp, "Unable to add operation to the curl multi-handle");
1227  op->Fail(XrdCl::errInternal, mres, "Unable to add operation to the curl multi-handle");
1228  OpRecord(*op, OpKind::Error);
1229  continue;
1230  }
1231  m_logger->Debug(kLogXrdClHttp, "Added request for URL %s to worker thread for processing", op->GetUrl().c_str());
1232  running_handles += 1;
1233  }
1234 
1235  // Maintain the periodic reporting of thread activity and fail any operations
1236  // that have expired / timed out.
1237  time_t now = time(NULL);
1238  time_t next_maintenance = last_maintenance + m_maintenance_period.load(std::memory_order_relaxed);
1239  if (now >= next_maintenance) {
1240  m_queue->Expire();
1241  m_continue_queue->Expire();
1242  m_logger->Debug(kLogXrdClHttp, "Curl worker thread %d is running %d operations",
1243  getthreadid(), running_handles);
1244  last_maintenance = now;
1245 
1246  // Timeout all the pending broker requests.
1247  std::vector<std::pair<int, CURL *>> expired_ops;
1248  for (const auto &entry : broker_reqs) {
1249  if (entry.second.expiry < now) {
1250  expired_ops.emplace_back(entry.first, entry.second.curl);
1251  }
1252  }
1253  for (const auto &entry : expired_ops) {
1254  auto iter = m_op_map.find(entry.second);
1255  if (iter == m_op_map.end()) {
1256  m_logger->Warning(kLogXrdClHttp, "Found an expired curl handle with no corresponding operation!");
1257  } else {
1258 
1259  CurlOptionsOp *options_op = nullptr;
1260  if ((options_op = dynamic_cast<CurlOptionsOp*>(iter->second.first.get())) != nullptr) {
1261  auto parent_op = options_op->GetOperation();
1262  bool parent_op_failed = false;
1263  if (parent_op->IsRedirect()) {
1264  std::string target;
1265  if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
1266  auto iter = m_op_map.find(options_op->GetParentCurlHandle());
1267  if (iter != m_op_map.end()) {
1268  OpRecord(*iter->second.first, OpKind::Error);
1269  iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
1270  m_op_map.erase(iter);
1271  running_handles -= 1;
1272  }
1273  parent_op_failed = true;
1274  } else {
1275  OpRecord(*parent_op, OpKind::Start);
1276  }
1277  } else {
1278  OpRecord(*parent_op, OpKind::Start);
1279  }
1280  if (!parent_op_failed){
1281  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1282  }
1283  }
1284 
1285  iter->second.first->Fail(XrdCl::errConnectionError, 1, "Timeout: connection never provided for request");
1286  iter->second.first->ReleaseHandle();
1287  OpRecord(*(iter->second.first), OpKind::ConncallTimeout);
1288  m_op_map.erase(entry.second);
1289  curl_easy_cleanup(entry.second);
1290  running_handles -= 1;
1291  }
1292  broker_reqs.erase(entry.first);
1293  m_conncall_timeout.fetch_add(1, std::memory_order_relaxed);
1294  }
1295 
1296  // Cleanup the fake connection cache entries.
1297  XrdClHttp::CurlOperation::CleanupDnsCache();
1298  }
1299 
1300  waitfds.clear();
1301  waitfds.resize(3 + broker_reqs.size());
1302 
1303  waitfds[0].fd = queue.PollFD();
1304  waitfds[0].events = CURL_WAIT_POLLIN;
1305  waitfds[0].revents = 0;
1306  waitfds[1].fd = m_continue_queue->PollFD();
1307  waitfds[1].events = CURL_WAIT_POLLIN;
1308  waitfds[1].revents = 0;
1309  waitfds[2].fd = m_shutdown_pipe_r;
1310  waitfds[2].revents = 0;
1311  waitfds[2].events = CURL_WAIT_POLLIN | CURL_WAIT_POLLPRI;
1312 
1313  int idx = 3;
1314  for (const auto &entry : broker_reqs) {
1315  waitfds[idx].fd = entry.first;
1316  waitfds[idx].events = CURL_WAIT_POLLIN|CURL_WAIT_POLLPRI;
1317  waitfds[idx].revents = 0;
1318  idx += 1;
1319  }
1320 
1321  long timeo;
1322  curl_multi_timeout(multi_handle, &timeo);
1323  // These commented-out lines are purposely left; will need to revisit after the 0.9.1 release;
1324  // for now, they are too verbose on RHEL7.
1325  //m_logger->Debug(kLogXrdClHttp, "Curl advises a timeout of %ld ms", timeo);
1326  if (running_handles && timeo == -1) {
1327  // Bug workaround: we've seen RHEL7 libcurl have a race condition where it'll not
1328  // set a timeout while doing the DNS lookup; assume that if there are running handles
1329  // but no timeout, we've hit this bug.
1330  //m_logger->Debug(kLogXrdClHttp, "Will sleep for up to 50ms");
1331  mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), 50, nullptr);
1332  } else {
1333  //m_logger->Debug(kLogXrdClHttp, "Will sleep for up to %d seconds", max_sleep_time);
1334  //mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), max_sleep_time*1000, nullptr);
1335  // Temporary test: we've been seeing DNS lookups timeout on additional platforms. Switch to always
1336  // poll as curl_multi_wait doesn't seem to get notified when DNS lookups are done.
1337  mres = curl_multi_wait(multi_handle, &waitfds[0], waitfds.size(), 50, nullptr);
1338  }
1339  if (mres != CURLM_OK) {
1340  m_logger->Warning(kLogXrdClHttp, "Failed to wait on multi-handle: %d", mres);
1341  }
1342 
1343  // Iterate through the waiting broker callbacks.
1344  for (const auto &entry : waitfds) {
1345  // Ignore the queue's poll fd.
1346  if (waitfds[0].fd == entry.fd || waitfds[1].fd == entry.fd) {
1347  continue;
1348  }
1349  // Handle shutdown requests
1350  if ((waitfds[2].fd == entry.fd) && entry.revents) {
1351  want_shutdown = true;
1352  break;
1353  }
1354  if ((entry.revents & CURL_WAIT_POLLIN) != CURL_WAIT_POLLIN) {
1355  continue;
1356  }
1357  auto handle = broker_reqs[entry.fd].curl;
1358  auto iter = m_op_map.find(handle);
1359  if (iter == m_op_map.end()) {
1360  m_logger->Warning(kLogXrdClHttp, "Internal error: broker responded on FD %d but no corresponding curl operation", entry.fd);
1361  broker_reqs.erase(entry.fd);
1362  m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
1363  continue;
1364  }
1365  std::string err;
1366  auto result = iter->second.first->WaitSocketCallback(err);
1367  if (result == -1) {
1368  m_logger->Warning(kLogXrdClHttp, "Error when invoking the broker callback: %s", err.c_str());
1369 
1370  CurlOptionsOp *options_op = nullptr;
1371  if ((options_op = dynamic_cast<CurlOptionsOp*>(iter->second.first.get())) != nullptr) {
1372  auto parent_op = options_op->GetOperation();
1373  bool parent_op_failed = false;
1374  if (parent_op->IsRedirect()) {
1375  std::string target;
1376  if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
1377  auto iter = m_op_map.find(options_op->GetParentCurlHandle());
1378  if (iter != m_op_map.end()) {
1379  OpRecord(*iter->second.first, OpKind::Error);
1380  iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
1381  m_op_map.erase(iter);
1382  running_handles -= 1;
1383  }
1384  parent_op_failed = true;
1385  } else {
1386  OpRecord(*parent_op, OpKind::Start);
1387  }
1388  } else {
1389  OpRecord(*parent_op, OpKind::Start);
1390  }
1391  if (!parent_op_failed){
1392  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1393  }
1394  }
1395 
1396  iter->second.first->Fail(XrdCl::errErrorResponse, 1, err);
1397  OpRecord(*iter->second.first, OpKind::Error);
1398  m_op_map.erase(handle);
1399  broker_reqs.erase(entry.fd);
1400  m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
1401  running_handles -= 1;
1402  } else {
1403  broker_reqs.erase(entry.fd);
1404  curl_multi_add_handle(multi_handle, handle);
1405  m_conncall_success.fetch_add(1, std::memory_order_relaxed);
1406  }
1407  }
1408 
1409  // Do maintenance on the multi-handle
1410  int still_running;
1411  auto mres = curl_multi_perform(multi_handle, &still_running);
1412  if (mres == CURLM_CALL_MULTI_PERFORM) {
1413  continue;
1414  } else if (mres != CURLM_OK) {
1415  m_logger->Warning(kLogXrdClHttp, "Failed to perform multi-handle operation: %d", mres);
1416  break;
1417  }
1418 
1419  CURLMsg *msg;
1420  do {
1421  int msgq = 0;
1422  msg = curl_multi_info_read(multi_handle, &msgq);
1423  if (msg && (msg->msg == CURLMSG_DONE)) {
1424  if (!msg->easy_handle) {
1425  m_logger->Warning(kLogXrdClHttp, "Logic error: got a callback for a null handle");
1426  mres = CURLM_BAD_EASY_HANDLE;
1427  break;
1428  }
1429  auto iter = m_op_map.find(msg->easy_handle);
1430  if (iter == m_op_map.end()) {
1431  m_logger->Error(kLogXrdClHttp, "Logic error: got a callback for an entry that doesn't exist");
1432  mres = CURLM_BAD_EASY_HANDLE;
1433  break;
1434  }
1435  auto op = iter->second.first;
1436  auto res = msg->data.result;
1437  bool keep_handle = false;
1438  bool waiting_on_callout = false;
1439  if (res == CURLE_OK) {
1440  auto sc = op->GetStatusCode();
1441  OpRecord(*op, OpKind::Finish);
1442  if (HTTPStatusIsError(sc)) {
1443  auto httpErr = HTTPStatusConvert(sc);
1444  op->Fail(httpErr.first, httpErr.second, op->GetStatusMessage());
1445  op->ReleaseHandle();
1446  // If this was a failed CurlOptionsOp, then we re-activate the parent handle.
1447  // If the parent handle was stopped at a redirect that now returns failure, then
1448  // we'll clean it up.
1449  CurlOptionsOp *options_op = nullptr;
1450  if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
1451  auto parent_op = options_op->GetOperation();
1452  bool parent_op_failed = false;
1453  if (parent_op->IsRedirect()) {
1454  std::string target;
1455  if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
1456  OpRecord(*parent_op, OpKind::Error);
1457  m_op_map.erase(options_op->GetParentCurlHandle());
1458  running_handles -= 1;
1459  parent_op_failed = true;
1460  } else {
1461  OpRecord(*parent_op, OpKind::Start);
1462  }
1463  } else {
1464  OpRecord(*parent_op, OpKind::Start);
1465  }
1466  // Have curl execute the parent operation
1467  if (!parent_op_failed) {
1468  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1469  }
1470  }
1471  // The curl operation was successful, it's just the HTTP request failed; recycle the handle.
1472  queue.RecycleHandle(iter->first);
1473  } else {
1474  CurlOptionsOp *options_op = nullptr;
1475  // If this was a successful OPTIONS op, invoke the parent operation.
1476  if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get()))) {
1477  options_op->Success();
1478  options_op->ReleaseHandle();
1479  // Note: op is scoped external to the conditional block
1480  op = options_op->GetOperation();
1481  op->OptionsDone();
1482  OpRecord(*op, OpKind::Start);
1483  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1484  curl_multi_remove_handle(multi_handle, iter->first);
1485  queue.RecycleHandle(iter->first);
1486  }
1487  // Check to see if the operation ended in a redirect (note: this might)
1488  // be invoked a second time if this was the parent operation of an OPTIONS
1489  // op.
1490  if (op->IsRedirect()) {
1491  std::string target;
1492  switch (op->Redirect(target)) {
1493  case CurlOperation::RedirectAction::Fail:
1494  if (options_op) {
1495  // In this case, we failed immediately after an OPTIONS finished.
1496  // Since there's a Start recorded after the OPTIONS processing, we
1497  // must record an error.
1498  // In the non-OPTIONS case, we never recorded a second start and
1499  // don't need a matching failure.
1500  OpRecord(*op, OpKind::Error);
1501  }
1502  keep_handle = false;
1503  break;
1504  case CurlOperation::RedirectAction::Reinvoke:
1505  if (!options_op) {
1506  // In this case, the redirect occurred without any prior
1507  // OPTIONS call. This implies that `op` is the original call
1508  // and we need to restart it later and record another op start.
1509  keep_handle = true;
1510  OpRecord(*op, OpKind::Start);
1511  }
1512  break;
1513  case CurlOperation::RedirectAction::ReinvokeAfterAllow:
1514  {
1515  // The redirect resulted in a new endpoint where the cache lookup failed;
1516  // we need to know what HTTP verbs are in the server's Allow list before this
1517  // operation can continue. Inject a new CurlOptionsOp and chain it to the one
1518  // being processed. Once the OPTIONS request is done, then we'll restart this
1519  // operation.
1520  std::string modified_url;
1521  target = VerbsCache::GetUrlKey(target, modified_url);
1522  options_op = new CurlOptionsOp(iter->first, op, target, m_logger, op->GetConnCalloutFunc());
1523  std::shared_ptr<CurlOperation> new_op(options_op);
1524  auto curl = queue.GetHandle();
1525  if (curl == nullptr) {
1526  m_logger->Debug(kLogXrdClHttp, "Unable to allocate a curl handle");
1527  op->Fail(XrdCl::errInternal, ENOMEM, "Unable to get allocate a curl handle");
1528  keep_handle = false;
1529  options_op = nullptr;
1530  break;
1531  }
1532  OpRecord(*new_op, OpKind::Start);
1533  try {
1534  auto rv = new_op->Setup(curl, *this);
1535  if (!rv) {
1536  m_logger->Debug(kLogXrdClHttp, "Unable to configure a curl handle for OPTIONS");
1537  keep_handle = false;
1538  options_op = nullptr;
1539  break;
1540  }
1541  } catch (...) {
1542  m_logger->Debug(kLogXrdClHttp, "Unable to setup the curl handle for the OPTIONS operation");
1543  new_op->Fail(XrdCl::errInternal, ENOMEM, "Failed to setup the curl handle for the OPTIONS operation");
1544  OpRecord(*new_op, OpKind::Error);
1545  keep_handle = false;
1546  break;
1547  }
1548  new_op->SetContinueQueue(m_continue_queue);
1549  m_op_map[curl] = {new_op, std::chrono::system_clock::now()};
1550  auto mres = curl_multi_add_handle(multi_handle, curl);
1551  if (mres != CURLM_OK) {
1552  m_logger->Debug(kLogXrdClHttp, "Unable to add OPTIONS operation to the curl multi-handle: %s", curl_multi_strerror(mres));
1553  op->Fail(XrdCl::errInternal, mres, "Unable to add OPTIONS operation to the curl multi-handle");
1554  OpRecord(*new_op, OpKind::Error);
1555  break;
1556  }
1557  running_handles += 1;
1558  m_logger->Debug(kLogXrdClHttp, "Invoking the OPTIONS operation before redirect to %s", target.c_str());
1559  // The original curl operation needs to be kept around. Note that because options_op
1560  // is non-nil, we won't re-add the handle to the multi-handle.
1561  keep_handle = true;
1562  }
1563  }
1564  int callout_socket = op->WaitSocket();
1565  if ((waiting_on_callout = callout_socket >= 0)) {
1566  auto expiry = time(nullptr) + 20;
1567  m_logger->Debug(kLogXrdClHttp, "Creating a callout wait request on socket %d", callout_socket);
1568  broker_reqs[callout_socket] = {iter->first, expiry};
1569  m_conncall_req.fetch_add(1, std::memory_order_relaxed);
1570  }
1571  } else if (options_op) {
1572  // In this case, the OPTIONS call happened before the parent operation was started.
1573  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1574  }
1575  if (keep_handle) {
1576  curl_multi_remove_handle(multi_handle, iter->first);
1577  if (!waiting_on_callout && !options_op) {
1578  curl_multi_add_handle(multi_handle, iter->first);
1579  }
1580  } else if (!options_op) {
1581  op->Success();
1582  op->ReleaseHandle();
1583  // If the handle was successful, then we can recycle it.
1584  queue.RecycleHandle(iter->first);
1585  }
1586  }
1587  } else if (res == CURLE_COULDNT_CONNECT && op->UseConnectionCallout() && !op->GetTriedBoker()) {
1588  // In this case, we need to use the broker and the curl handle couldn't reuse
1589  // an existing socket.
1590  keep_handle = true;
1591  op->SetTriedBoker(); // Flag to ensure we try a connection only once per operation.
1592  std::string err;
1593  int wait_socket = -1;
1594  if (!op->StartConnectionCallout(err) || (wait_socket=op->WaitSocket()) == -1) {
1595  m_logger->Error(kLogXrdClHttp, "Failed to start broker-based connection: %s", err.c_str());
1596  op->ReleaseHandle();
1597  keep_handle = false;
1598  } else {
1599  curl_multi_remove_handle(multi_handle, iter->first);
1600  auto expiry = time(nullptr) + 20;
1601  m_logger->Debug(kLogXrdClHttp, "Curl operation requires a new TCP socket; waiting for callout to respond on socket %d", wait_socket);
1602  broker_reqs[wait_socket] = {iter->first, expiry};
1603  m_conncall_req.fetch_add(1, std::memory_order_relaxed);
1604  }
1605  } else {
1606  if (res == CURLE_ABORTED_BY_CALLBACK || res == CURLE_WRITE_ERROR) {
1607  // We cannot invoke the failure from within a callback as the curl thread and
1608  // original thread of execution may fight over the ownership of the handle memory.
1609  switch (op->GetError()) {
1610  case CurlOperation::OpError::ErrHeaderTimeout:
1611 #ifdef HAVE_XPROTOCOL_TIMEREXPIRED
1612  op->Fail(XrdCl::errOperationExpired, 0, "Origin did not respond with headers within timeout");
1613 #else
1614  op->Fail(XrdCl::errOperationExpired, 0, "Origin did not respond within timeout");
1615 #endif
1616  OpRecord(*op, OpKind::Error);
1617  break;
1618  case CurlOperation::OpError::ErrCallback: {
1619  auto [ecode, emsg] = op->GetCallbackError();
1620  op->Fail(XrdCl::errErrorResponse, ecode, emsg);
1621  OpRecord(*op, OpKind::Error);
1622  break;
1623  }
1624  case CurlOperation::OpError::ErrOperationTimeout:
1625  op->Fail(XrdCl::errOperationExpired, 0, "Operation timed out");
1626  OpRecord(*op, op->IsPaused() ? OpKind::ClientTimeout : OpKind::ServerTimeout);
1627  break;
1628  case CurlOperation::OpError::ErrTransferSlow:
1629  op->Fail(XrdCl::errOperationExpired, 0, "Transfer speed below minimum threshold");
1630  OpRecord(*op, OpKind::ServerTimeout);
1631  break;
1632  case CurlOperation::OpError::ErrTransferClientStall:
1633  op->Fail(XrdCl::errOperationExpired, 0, "Transfer stalled for too long");
1634  OpRecord(*op, OpKind::ClientTimeout);
1635  break;
1636  case CurlOperation::OpError::ErrTransferStall:
1637  op->Fail(XrdCl::errOperationExpired, 0, "Transfer stalled for too long");
1638  OpRecord(*op, OpKind::ServerTimeout);
1639  break;
1640  case CurlOperation::OpError::ErrNone:
1641  op->Fail(XrdCl::errInternal, 0, "Operation was aborted without recording an abort reason");
1642  OpRecord(*op, OpKind::Error);
1643  break;
1644  };
1645  CurlOptionsOp *options_op = nullptr;
1646  if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
1647  auto parent_op = options_op->GetOperation();
1648  bool parent_op_failed = false;
1649  if (parent_op->IsRedirect()) {
1650  std::string target;
1651  if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
1652  auto iter = m_op_map.find(options_op->GetParentCurlHandle());
1653  if (iter != m_op_map.end()) {
1654  OpRecord(*iter->second.first, OpKind::Error);
1655  iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
1656  m_op_map.erase(iter);
1657  running_handles -= 1;
1658  }
1659  parent_op_failed = true;
1660  } else {
1661  OpRecord(*parent_op, OpKind::Start);
1662  }
1663  } else {
1664  OpRecord(*parent_op, OpKind::Start);
1665  }
1666  if (!parent_op_failed){
1667  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1668  }
1669  }
1670  } else {
1671  auto xrdCode = CurlCodeConvert(res);
1672  const auto curl_err = op->GetCurlErrorMessage();
1673  const char *curl_easy_err = curl_easy_strerror(res);
1674  const std::string fail_err = !curl_err.empty() ? curl_err : curl_easy_err;
1675  m_logger->Debug(kLogXrdClHttp, "Curl generated an error: %s (%d)", fail_err.c_str(), res);
1676  op->Fail(xrdCode.first, xrdCode.second, fail_err);
1677  OpRecord(*op, OpKind::Error);
1678  CurlOptionsOp *options_op = nullptr;
1679  if ((options_op = dynamic_cast<CurlOptionsOp*>(op.get())) != nullptr) {
1680  auto parent_op = options_op->GetOperation();
1681  bool parent_op_failed = false;
1682  if (parent_op->IsRedirect()) {
1683  std::string target;
1684  if (parent_op->Redirect(target) == CurlOperation::RedirectAction::Fail) {
1685  auto iter = m_op_map.find(options_op->GetParentCurlHandle());
1686  if (iter != m_op_map.end()) {
1687  OpRecord(*iter->second.first, OpKind::Error);
1688  iter->second.first->Fail(XrdCl::errErrorResponse, 0, "Failed to send OPTIONS to redirect target");
1689  m_op_map.erase(iter);
1690  running_handles -= 1;
1691  }
1692  parent_op_failed = true;
1693  }
1694  }
1695  if (!parent_op_failed){
1696  curl_multi_add_handle(multi_handle, options_op->GetParentCurlHandle());
1697  }
1698  }
1699  }
1700  op->ReleaseHandle();
1701  }
1702  if (!keep_handle) {
1703  curl_multi_remove_handle(multi_handle, iter->first);
1704  if (res != CURLE_OK) {
1705  curl_easy_cleanup(iter->first);
1706  }
1707  for (auto &req : broker_reqs) {
1708  if (req.second.curl == iter->first) {
1709  m_logger->Warning(kLogXrdClHttp, "Curl handle finished while a broker operation was outstanding");
1710  m_conncall_errors.fetch_add(1, std::memory_order_relaxed);
1711  }
1712  }
1713  m_op_map.erase(iter);
1714  running_handles -= 1;
1715  }
1716  }
1717  } while (msg);
1718  }
1719 
1720  for (auto map_entry : m_op_map) {
1721  if (mres) {
1722  map_entry.second.first->Fail(XrdCl::errInternal, mres, curl_multi_strerror(mres));
1723  OpRecord(*map_entry.second.first, OpKind::Error);
1724  }
1725  if (multi_handle && map_entry.first) curl_multi_remove_handle(multi_handle, map_entry.first);
1726  }
1727 
1728  m_queue->ReleaseHandles();
1729  curl_multi_cleanup(multi_handle);
1730 }
1731 
1732 void
1733 CurlWorker::Shutdown()
1734 {
1735  m_queue->Shutdown();
1736  if (m_shutdown_pipe_w == -1) {
1737  m_logger->Debug(kLogXrdClHttp, "Curl worker shutdown prior to launch of thread");
1738  return;
1739  }
1740  close(m_shutdown_pipe_w);
1741  m_shutdown_pipe_w = -1;
1742 
1743  // wait for worker thread to exit
1744  m_self_tid.join();
1745 
1746  {
1747  std::unique_lock lk(m_worker_stats_mutex);
1748  m_workers_last_completed_cycle[m_stats_offset] = nullptr;
1749  m_workers_oldest_op[m_stats_offset] = nullptr;
1750  }
1751  m_logger->Debug(kLogXrdClHttp, "Curl worker thread shutdown has completed.");
1752 }
1753 
1754 void
1755 CurlWorker::ShutdownAll()
1756 {
1757  std::unique_lock lock(m_workers_mutex);
1758  for (auto &worker : m_workers) {
1759  worker->Shutdown();
1760  }
1761 }
1762 
1763 CurlWorker::initcontrol::initcontrol()
1764 {
1765  curl_global_init(CURL_GLOBAL_DEFAULT);
1766 }
1767 
1768 CurlWorker::initcontrol::~initcontrol()
1769 {
1770  ShutdownAll();
1771  curl_global_cleanup();
1772 }
kXR_InvalidRequest
@ kXR_InvalidRequest
Definition: XProtocol.hh:1038
kXR_Impossible
@ kXR_Impossible
Definition: XProtocol.hh:1063
kXR_TimerExpired
@ kXR_TimerExpired
Definition: XProtocol.hh:1067
kXR_NotAuthorized
@ kXR_NotAuthorized
Definition: XProtocol.hh:1042
kXR_NotFound
@ kXR_NotFound
Definition: XProtocol.hh:1043
kXR_FileLocked
@ kXR_FileLocked
Definition: XProtocol.hh:1035
kXR_overQuota
@ kXR_overQuota
Definition: XProtocol.hh:1053
kXR_Unsupported
@ kXR_Unsupported
Definition: XProtocol.hh:1045
kXR_Conflict
@ kXR_Conflict
Definition: XProtocol.hh:1064
kXR_ServerError
@ kXR_ServerError
Definition: XProtocol.hh:1044
kXR_Overloaded
@ kXR_Overloaded
Definition: XProtocol.hh:1056
kXR_ReqTimedOut
@ kXR_ReqTimedOut
Definition: XProtocol.hh:1066
CurlCodeConvert
std::pair< uint16_t, uint32_t > CurlCodeConvert(CURLcode res)
Definition: XrdClHttpUtil.cc:178
CURL
void CURL
Definition: XrdClHttpUtil.hh:38
obfuscateAuth
std::string obfuscateAuth(const std::string &input)
Definition: XrdOucUtils.cc:1610
read
ssize_t read(int fildes, void *buf, size_t nbyte)
Definition: XrdPosixPreload.cc:602
close
#define close(a)
Definition: XrdPosix.hh:48
write
#define write(a, b, c)
Definition: XrdPosix.hh:121
getline
void getline(uchar *buff, int blen)
Definition: XrdSectestServer.cc:291
emsg
int emsg(int rc, char *msg)
Definition: XrdSectestServer.cc:309
Error
@ Error
Definition: XrdTlsNotaryUtils.hh:42
XrdClHttp::ChecksumInfo::Set
bool Set(ChecksumType ctype, const std::array< unsigned char, g_max_checksum_length > &value)
Definition: XrdClHttpChecksum.hh:128
XrdClHttp::CurlOperation::Success
virtual void Success()=0
XrdClHttp::CurlOperation::OptionsDone
virtual void OptionsDone()
Definition: XrdClHttpOps.hh:174
XrdClHttp::CurlOperation::FinishSetup
bool FinishSetup(CURL *curl)
Definition: XrdClHttpOps.cc:216
XrdClHttp::CurlOperation::GetUrl
const std::string & GetUrl() const
Definition: XrdClHttpOps.hh:177
XrdClHttp::CurlOperation::GetCallbackError
std::pair< XErrorCode, std::string > GetCallbackError() const
Definition: XrdClHttpOps.hh:226
XrdClHttp::CurlOperation::Fail
virtual void Fail(uint16_t errCode, uint32_t errNum, const std::string &)
XrdClHttp::CurlOperation::GetVerbString
static const std::string GetVerbString(HttpVerb)
Definition: XrdClHttpOps.cc:247
XrdClHttp::CurlOperation::GetVerb
virtual HttpVerb GetVerb() const =0
XrdClHttp::CurlOperation::GetCurlErrorMessage
std::string GetCurlErrorMessage() const
Definition: XrdClHttpOps.hh:252
XrdClHttp::CurlOperation::ReleaseHandle
virtual void ReleaseHandle()
Definition: XrdClHttpOps.cc:603
XrdClHttp::CurlOperation::RequiresOptions
virtual bool RequiresOptions() const
Definition: XrdClHttpOps.hh:171
XrdClHttp::CurlOperation::CleanupDnsCache
static void CleanupDnsCache()
Definition: XrdClHttpOps.cc:694
XrdClHttp::CurlOperation::StatisticsReset
std::tuple< uint64_t, std::chrono::steady_clock::duration, std::chrono::steady_clock::duration, std::chrono::steady_clock::duration > StatisticsReset()
Definition: XrdClHttpOps.cc:422
XrdClHttp::CurlOperation::ContinueHandle
virtual bool ContinueHandle()
Definition: XrdClHttpOps.hh:129
XrdClHttp::CurlOperation::GetStatusMessage
std::string GetStatusMessage() const
Definition: XrdClHttpOps.hh:232
XrdClHttp::CurlOperation::GetCurlHandle
CURL * GetCurlHandle() const
Definition: XrdClHttpOps.hh:217
XrdClHttp::CurlOperation::GetConnCalloutFunc
CreateConnCalloutType GetConnCalloutFunc() const
Definition: XrdClHttpOps.hh:99
XrdClHttp::CurlOperation::GetError
OpError GetError() const
Definition: XrdClHttpOps.hh:220
XrdClHttp::CurlOperation::Redirect
virtual RedirectAction Redirect(std::string &target)
Definition: XrdClHttpOps.cc:305
XrdClHttp::CurlOperation::SetContinueQueue
virtual void SetContinueQueue(std::shared_ptr< XrdClHttp::HandlerQueue > queue)
Definition: XrdClHttpOps.hh:133
XrdClHttp::CurlOperation::StartConnectionCallout
bool StartConnectionCallout(std::string &err)
Definition: XrdClHttpOps.cc:411
XrdClHttp::CurlOperation::Setup
virtual bool Setup(CURL *curl, CurlWorker &)
Definition: XrdClHttpOps.cc:523
XrdClHttp::CurlOptionsOp::GetParentCurlHandle
CURL * GetParentCurlHandle() const
Definition: XrdClHttpOps.hh:432
XrdClHttp::CurlOptionsOp::GetOperation
std::shared_ptr< CurlOperation > GetOperation() const
Definition: XrdClHttpOps.hh:428
XrdClHttp::CurlWorker::ClientX509CertKeyFile
std::tuple< std::string, std::string > ClientX509CertKeyFile() const
Definition: XrdClHttpUtil.cc:915
XrdClHttp::CurlWorker::RunStatic
static void RunStatic(CurlWorker *myself)
Definition: XrdClHttpUtil.cc:1071
XrdClHttp::CurlWorker::Start
void Start(std::unique_ptr< XrdClHttp::CurlWorker > self, std::thread tid)
Definition: XrdClHttpUtil.cc:1058
XrdClHttp::CurlWorker::GetMonitoringJson
static std::string GetMonitoringJson()
Definition: XrdClHttpUtil.cc:921
XrdClHttp::HandlerQueue::HandlerQueue
HandlerQueue(unsigned max_pending_ops)
Definition: XrdClHttpUtil.cc:559
XrdClHttp::HeaderParser::SetMultipartSeparator
void SetMultipartSeparator(const std::string_view &sep)
Definition: XrdClHttpUtil.hh:105
XrdClHttp::HeaderParser::Base64Decode
static bool Base64Decode(std::string_view input, std::array< unsigned char, 32 > &output)
Definition: XrdClHttpUtil.cc:246
XrdClHttp::HeaderParser::ParseDigest
static void ParseDigest(const std::string &digest, XrdClHttp::ChecksumInfo &info)
Definition: XrdClHttpUtil.cc:436
XrdClHttp::HeaderParser::Canonicalize
static bool Canonicalize(std::string &headerName)
Definition: XrdClHttpUtil.cc:539
XrdClHttp::HeaderParser::Parse
bool Parse(const std::string &headers)
Definition: XrdClHttpUtil.cc:277
XrdClHttp::HeaderParser::ChecksumTypeToDigestName
static std::string ChecksumTypeToDigestName(XrdClHttp::ChecksumType type)
Definition: XrdClHttpUtil.cc:494
XrdClHttp::VerbsCache::GetUrlKey
static std::string_view GetUrlKey(const std::string &url, std::string &modified_url)
Definition: XrdClHttpOptionsCache.hh:120
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition: XrdClDefaultEnv.cc:493
XrdCl::DefaultEnv::GetEnv
static Env * GetEnv()
Get default client environment.
Definition: XrdClDefaultEnv.cc:445
XrdCl::Env::GetInt
bool GetInt(const std::string &key, int &value)
Definition: XrdClEnv.cc:115
XrdCl::Log
Handle diagnostics.
Definition: XrdClLog.hh:101
XrdCl::Log::DumpMsg
@ DumpMsg
print details of the request and responses
Definition: XrdClLog.hh:113
XrdCl::Log::Error
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition: XrdClLog.cc:231
XrdCl::Log::Warning
void Warning(uint64_t topic, const char *format,...)
Report a warning.
Definition: XrdClLog.cc:248
XrdCl::Log::Debug
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition: XrdClLog.cc:282
XrdClHttp::HTTPStatusConvert
std::pair< uint16_t, uint32_t > HTTPStatusConvert(unsigned status)
Definition: XrdClHttpUtil.cc:110
XrdClHttp::GetHandle
CURL * GetHandle(bool verbose)
XrdClHttp::HTTPStatusIsError
bool HTTPStatusIsError(unsigned status)
Definition: XrdClHttpUtil.cc:106
XrdClHttp::ltrim_view
std::string_view ltrim_view(const std::string_view &input_view)
Definition: XrdClHttpUtil.cc:625
XrdClHttp::trim_view
std::string_view trim_view(const std::string_view &input_view)
Definition: XrdClHttpUtil.cc:614
XrdCl::errUnknown
const uint16_t errUnknown
Unknown error.
Definition: XrdClStatus.hh:50
XrdCl::errInvalidAddr
const uint16_t errInvalidAddr
Definition: XrdClStatus.hh:71
XrdCl::errRedirectLimit
const uint16_t errRedirectLimit
Definition: XrdClStatus.hh:102
XrdCl::errErrorResponse
const uint16_t errErrorResponse
Definition: XrdClStatus.hh:105
XrdCl::errTlsError
const uint16_t errTlsError
Definition: XrdClStatus.hh:80
XrdCl::errOperationExpired
const uint16_t errOperationExpired
Definition: XrdClStatus.hh:90
XrdCl::errLoginFailed
const uint16_t errLoginFailed
Definition: XrdClStatus.hh:87
XrdCl::errDataError
const uint16_t errDataError
data is corrupted
Definition: XrdClStatus.hh:63
XrdCl::errInternal
const uint16_t errInternal
Internal error.
Definition: XrdClStatus.hh:56
XrdCl::errInvalidArgs
const uint16_t errInvalidArgs
Definition: XrdClStatus.hh:58
XrdCl::errConnectionError
const uint16_t errConnectionError
Definition: XrdClStatus.hh:78
XrdCl::errNotSupported
const uint16_t errNotSupported
Definition: XrdClStatus.hh:62
XrdCl::errSocketError
const uint16_t errSocketError
Definition: XrdClStatus.hh:72
XrdCl::errCorruptedHeader
const uint16_t errCorruptedHeader
Definition: XrdClStatus.hh:103
XrdCl::errNone
const uint16_t errNone
No error.
Definition: XrdClStatus.hh:48
XrdClHttp::kLogXrdClHttp
const uint64_t kLogXrdClHttp
Definition: XrdClHttpUtil.hh:52