ceph/src/rgw/rgw_http_client.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include "include/compat.h"
   5 #include "common/errno.h"
   6
   7
   8 #include <curl/curl.h>
   9 #include <curl/easy.h>
  10 #include <curl/multi.h>
  11
  12 #include "rgw_common.h"
  13 #include "rgw_http_client.h"
  14 #include "rgw_http_errors.h"
  15 #include "common/async/completion.h"
  16 #include "common/RefCountedObj.h"
  17
  18 #include "rgw_coroutine.h"
  19 #include "rgw_tools.h"
  20
  21 #include <atomic>
  22 #include <string_view>
  23
  24 #define dout_context g_ceph_context
  25 #define dout_subsys ceph_subsys_rgw
  26
  27 RGWHTTPManager *rgw_http_manager;
  28
  29 struct RGWCurlHandle;
  30
  31 static void do_curl_easy_cleanup(RGWCurlHandle *curl_handle);
  32
  33 struct rgw_http_req_data : public RefCountedObject {
  34   RGWCurlHandle *curl_handle{nullptr};
  35   curl_slist *h{nullptr};
  36   uint64_t id;
  37   int ret{0};
  38   std::atomic<bool> done = { false };
  39   RGWHTTPClient *client{nullptr};
  40   rgw_io_id control_io_id;
  41   void *user_info{nullptr};
  42   bool registered{false};
  43   RGWHTTPManager *mgr{nullptr};
  44   char error_buf[CURL_ERROR_SIZE];
  45   bool write_paused{false};
  46   bool read_paused{false};
  47
  48   optional<int> user_ret;
  49
  50   ceph::mutex lock = ceph::make_mutex("rgw_http_req_data::lock");
  51   ceph::condition_variable cond;
  52
  53   using Signature = void(boost::system::error_code);
  54   using Completion = ceph::async::Completion<Signature>;
  55   std::unique_ptr<Completion> completion;
  56
  57   rgw_http_req_data() : id(-1) {
  58     // FIPS zeroization audit 20191115: this memset is not security related.
  59     memset(error_buf, 0, sizeof(error_buf));
  60   }
  61
  62   template <typename ExecutionContext, typename CompletionToken>
  63   auto async_wait(ExecutionContext& ctx, CompletionToken&& token) {
  64     boost::asio::async_completion<CompletionToken, Signature> init(token);
  65     auto& handler = init.completion_handler;
  66     {
  67       std::unique_lock l{lock};
  68       completion = Completion::create(ctx.get_executor(), std::move(handler));
  69     }
  70     return init.result.get();
  71   }
  72
  73   int wait(optional_yield y) {
  74     if (done) {
  75       return ret;
  76     }
  77     if (y) {
  78       auto& context = y.get_io_context();
  79       auto& yield = y.get_yield_context();
  80       boost::system::error_code ec;
  81       async_wait(context, yield[ec]);
  82       return -ec.value();
  83     }
  84     // work on asio threads should be asynchronous, so warn when they block
  85     if (is_asio_thread) {
  86       dout(20) << "WARNING: blocking http request" << dendl;
  87     }
  88     std::unique_lock l{lock};
  89     cond.wait(l, [this]{return done==true;});
  90     return ret;
  91   }
  92
  93   void set_state(int bitmask);
  94
  95   void finish(int r, long http_status = -1) {
  96     std::lock_guard l{lock};
  97     if (http_status != -1) {
  98       if (client) {
  99         client->set_http_status(http_status);
 100       }
 101     }
 102     ret = r;
 103     if (curl_handle)
 104       do_curl_easy_cleanup(curl_handle);
 105
 106     if (h)
 107       curl_slist_free_all(h);
 108
 109     curl_handle = NULL;
 110     h = NULL;
 111     done = true;
 112     if (completion) {
 113       boost::system::error_code ec(-ret, boost::system::system_category());
 114       Completion::post(std::move(completion), ec);
 115     } else {
 116       cond.notify_all();
 117     }
 118   }
 119
 120   bool is_done() {
 121     return done;
 122   }
 123
 124   int get_retcode() {
 125     std::lock_guard l{lock};
 126     return ret;
 127   }
 128
 129   RGWHTTPManager *get_manager() {
 130     std::lock_guard l{lock};
 131     return mgr;
 132   }
 133
 134   CURL *get_easy_handle() const;
 135 };
 136
 137 struct RGWCurlHandle {
 138   int uses;
 139   mono_time lastuse;
 140   CURL* h;
 141
 142   explicit RGWCurlHandle(CURL* h) : uses(0), h(h) {};
 143   CURL* operator*() {
 144     return this->h;
 145   }
 146 };
 147
 148 void rgw_http_req_data::set_state(int bitmask) {
 149   /* no need to lock here, moreover curl_easy_pause() might trigger
 150    * the data receive callback :/
 151    */
 152   CURLcode rc = curl_easy_pause(**curl_handle, bitmask);
 153   if (rc != CURLE_OK) {
 154     dout(0) << "ERROR: curl_easy_pause() returned rc=" << rc << dendl;
 155   }
 156 }
 157
 158 #define MAXIDLE 5
 159 class RGWCurlHandles : public Thread {
 160 public:
 161   ceph::mutex cleaner_lock = ceph::make_mutex("RGWCurlHandles::cleaner_lock");
 162   std::vector<RGWCurlHandle*> saved_curl;
 163   int cleaner_shutdown;
 164   ceph::condition_variable cleaner_cond;
 165
 166   RGWCurlHandles() :
 167     cleaner_shutdown{0} {
 168   }
 169
 170   RGWCurlHandle* get_curl_handle();
 171   void release_curl_handle_now(RGWCurlHandle* curl);
 172   void release_curl_handle(RGWCurlHandle* curl);
 173   void flush_curl_handles();
 174   void* entry();
 175   void stop();
 176 };
 177
 178 RGWCurlHandle* RGWCurlHandles::get_curl_handle() {
 179   RGWCurlHandle* curl = 0;
 180   CURL* h;
 181   {
 182     std::lock_guard lock{cleaner_lock};
 183     if (!saved_curl.empty()) {
 184       curl = *saved_curl.begin();
 185       saved_curl.erase(saved_curl.begin());
 186     }
 187   }
 188   if (curl) {
 189   } else if ((h = curl_easy_init())) {
 190     curl = new RGWCurlHandle{h};
 191   } else {
 192     // curl = 0;
 193   }
 194   return curl;
 195 }
 196
 197 void RGWCurlHandles::release_curl_handle_now(RGWCurlHandle* curl)
 198 {
 199   curl_easy_cleanup(**curl);
 200   delete curl;
 201 }
 202
 203 void RGWCurlHandles::release_curl_handle(RGWCurlHandle* curl)
 204 {
 205   if (cleaner_shutdown) {
 206     release_curl_handle_now(curl);
 207   } else {
 208     curl_easy_reset(**curl);
 209     std::lock_guard lock{cleaner_lock};
 210     curl->lastuse = mono_clock::now();
 211     saved_curl.insert(saved_curl.begin(), 1, curl);
 212   }
 213 }
 214
 215 void* RGWCurlHandles::entry()
 216 {
 217   RGWCurlHandle* curl;
 218   std::unique_lock lock{cleaner_lock};
 219
 220   for (;;) {
 221     if (cleaner_shutdown) {
 222       if (saved_curl.empty())
 223         break;
 224     } else {
 225       cleaner_cond.wait_for(lock, std::chrono::seconds(MAXIDLE));
 226     }
 227     mono_time now = mono_clock::now();
 228     while (!saved_curl.empty()) {
 229       auto cend = saved_curl.end();
 230       --cend;
 231       curl = *cend;
 232       if (!cleaner_shutdown && now - curl->lastuse < std::chrono::seconds(MAXIDLE))
 233         break;
 234       saved_curl.erase(cend);
 235       release_curl_handle_now(curl);
 236     }
 237   }
 238   return nullptr;
 239 }
 240
 241 void RGWCurlHandles::stop()
 242 {
 243   std::lock_guard lock{cleaner_lock};
 244   cleaner_shutdown = 1;
 245   cleaner_cond.notify_all();
 246 }
 247
 248 void RGWCurlHandles::flush_curl_handles()
 249 {
 250   stop();
 251   join();
 252   if (!saved_curl.empty()) {
 253     dout(0) << "ERROR: " << __func__ << " failed final cleanup" << dendl;
 254   }
 255   saved_curl.shrink_to_fit();
 256 }
 257
 258 CURL *rgw_http_req_data::get_easy_handle() const
 259 {
 260   return **curl_handle;
 261 }
 262
 263 static RGWCurlHandles *handles;
 264
 265 static RGWCurlHandle *do_curl_easy_init()
 266 {
 267   return handles->get_curl_handle();
 268 }
 269
 270 static void do_curl_easy_cleanup(RGWCurlHandle *curl_handle)
 271 {
 272   handles->release_curl_handle(curl_handle);
 273 }
 274
 275 // XXX make this part of the token cache?  (but that's swift-only;
 276 //      and this especially needs to integrates with s3...)
 277
 278 void rgw_setup_saved_curl_handles()
 279 {
 280   handles = new RGWCurlHandles();
 281   handles->create("rgw_curl");
 282 }
 283
 284 void rgw_release_all_curl_handles()
 285 {
 286   handles->flush_curl_handles();
 287   delete handles;
 288 }
 289
 290 void RGWIOProvider::assign_io(RGWIOIDProvider& io_id_provider, int io_type)
 291 {
 292   if (id == 0) {
 293     id = io_id_provider.get_next();
 294   }
 295 }
 296
 297 /*
 298  * the following set of callbacks will be called either on RGWHTTPManager::process(),
 299  * or via the RGWHTTPManager async processing.
 300  */
 301 size_t RGWHTTPClient::receive_http_header(void * const ptr,
 302                                           const size_t size,
 303                                           const size_t nmemb,
 304                                           void * const _info)
 305 {
 306   rgw_http_req_data *req_data = static_cast<rgw_http_req_data *>(_info);
 307   size_t len = size * nmemb;
 308
 309   std::lock_guard l{req_data->lock};
 310
 311   if (!req_data->registered) {
 312     return len;
 313   }
 314
 315   int ret = req_data->client->receive_header(ptr, size * nmemb);
 316   if (ret < 0) {
 317     dout(5) << "WARNING: client->receive_header() returned ret=" << ret << dendl;
 318     req_data->user_ret = ret;
 319     return CURLE_WRITE_ERROR;
 320   }
 321
 322   return len;
 323 }
 324
 325 size_t RGWHTTPClient::receive_http_data(void * const ptr,
 326                                         const size_t size,
 327                                         const size_t nmemb,
 328                                         void * const _info)
 329 {
 330   rgw_http_req_data *req_data = static_cast<rgw_http_req_data *>(_info);
 331   size_t len = size * nmemb;
 332
 333   bool pause = false;
 334
 335   RGWHTTPClient *client;
 336
 337   {
 338     std::lock_guard l{req_data->lock};
 339     if (!req_data->registered) {
 340       return len;
 341     }
 342
 343     client = req_data->client;
 344   }
 345
 346   size_t& skip_bytes = client->receive_pause_skip;
 347
 348   if (skip_bytes >= len) {
 349     skip_bytes -= len;
 350     return len;
 351   }
 352
 353   int ret = client->receive_data((char *)ptr + skip_bytes, len - skip_bytes, &pause);
 354   if (ret < 0) {
 355     dout(5) << "WARNING: client->receive_data() returned ret=" << ret << dendl;
 356     req_data->user_ret = ret;
 357     return CURLE_WRITE_ERROR;
 358   }
 359
 360   if (pause) {
 361     dout(20) << "RGWHTTPClient::receive_http_data(): pause" << dendl;
 362     skip_bytes = len;
 363     std::lock_guard l{req_data->lock};
 364     req_data->read_paused = true;
 365     return CURL_WRITEFUNC_PAUSE;
 366   }
 367
 368   skip_bytes = 0;
 369
 370   return len;
 371 }
 372
 373 size_t RGWHTTPClient::send_http_data(void * const ptr,
 374                                      const size_t size,
 375                                      const size_t nmemb,
 376                                      void * const _info)
 377 {
 378   rgw_http_req_data *req_data = static_cast<rgw_http_req_data *>(_info);
 379
 380   RGWHTTPClient *client;
 381
 382   {
 383     std::lock_guard l{req_data->lock};
 384
 385     if (!req_data->registered) {
 386       return 0;
 387     }
 388
 389     client = req_data->client;
 390   }
 391
 392   bool pause = false;
 393
 394   int ret = client->send_data(ptr, size * nmemb, &pause);
 395   if (ret < 0) {
 396     dout(5) << "WARNING: client->send_data() returned ret=" << ret << dendl;
 397     req_data->user_ret = ret;
 398     return CURLE_READ_ERROR;
 399   }
 400
 401   if (ret == 0 &&
 402       pause) {
 403     std::lock_guard l{req_data->lock};
 404     req_data->write_paused = true;
 405     return CURL_READFUNC_PAUSE;
 406   }
 407
 408   return ret;
 409 }
 410
 411 ceph::mutex& RGWHTTPClient::get_req_lock()
 412 {
 413   return req_data->lock;
 414 }
 415
 416 void RGWHTTPClient::_set_write_paused(bool pause)
 417 {
 418   ceph_assert(ceph_mutex_is_locked(req_data->lock));
 419
 420   RGWHTTPManager *mgr = req_data->mgr;
 421   if (pause == req_data->write_paused) {
 422     return;
 423   }
 424   if (pause) {
 425     mgr->set_request_state(this, SET_WRITE_PAUSED);
 426   } else {
 427     mgr->set_request_state(this, SET_WRITE_RESUME);
 428   }
 429 }
 430
 431 void RGWHTTPClient::_set_read_paused(bool pause)
 432 {
 433   ceph_assert(ceph_mutex_is_locked(req_data->lock));
 434
 435   RGWHTTPManager *mgr = req_data->mgr;
 436   if (pause == req_data->read_paused) {
 437     return;
 438   }
 439   if (pause) {
 440     mgr->set_request_state(this, SET_READ_PAUSED);
 441   } else {
 442     mgr->set_request_state(this, SET_READ_RESUME);
 443   }
 444 }
 445
 446 static curl_slist *headers_to_slist(param_vec_t& headers)
 447 {
 448   curl_slist *h = NULL;
 449
 450   param_vec_t::iterator iter;
 451   for (iter = headers.begin(); iter != headers.end(); ++iter) {
 452     pair<string, string>& p = *iter;
 453     string val = p.first;
 454
 455     if (strncmp(val.c_str(), "HTTP_", 5) == 0) {
 456       val = val.substr(5);
 457     }
 458
 459     /* we need to convert all underscores into dashes as some web servers forbid them
 460      * in the http header field names
 461      */
 462     for (size_t i = 0; i < val.size(); i++) {
 463       if (val[i] == '_') {
 464         val[i] = '-';
 465       }
 466     }
 467
 468     val = camelcase_dash_http_attr(val);
 469
 470     // curl won't send headers with empty values unless it ends with a ; instead
 471     if (p.second.empty()) {
 472       val.append(1, ';');
 473     } else {
 474       val.append(": ");
 475       val.append(p.second);
 476     }
 477     h = curl_slist_append(h, val.c_str());
 478   }
 479
 480   return h;
 481 }
 482
 483 static bool is_upload_request(const string& method)
 484 {
 485   return method == "POST" || method == "PUT";
 486 }
 487
 488 /*
 489  * process a single simple one off request
 490  */
 491 int RGWHTTPClient::process(optional_yield y)
 492 {
 493   return RGWHTTP::process(this, y);
 494 }
 495
 496 string RGWHTTPClient::to_str()
 497 {
 498   string method_str = (method.empty() ? "<no-method>" : method);
 499   string url_str = (url.empty() ? "<no-url>" : url);
 500   return method_str + " " + url_str;
 501 }
 502
 503 int RGWHTTPClient::get_req_retcode()
 504 {
 505   if (!req_data) {
 506     return -EINVAL;
 507   }
 508
 509   return req_data->get_retcode();
 510 }
 511
 512 /*
 513  * init request, will be used later with RGWHTTPManager
 514  */
 515 int RGWHTTPClient::init_request(rgw_http_req_data *_req_data)
 516 {
 517   ceph_assert(!req_data);
 518   _req_data->get();
 519   req_data = _req_data;
 520
 521   req_data->curl_handle = do_curl_easy_init();
 522
 523   CURL *easy_handle = req_data->get_easy_handle();
 524
 525   dout(20) << "sending request to " << url << dendl;
 526
 527   curl_slist *h = headers_to_slist(headers);
 528
 529   req_data->h = h;
 530
 531   curl_easy_setopt(easy_handle, CURLOPT_CUSTOMREQUEST, method.c_str());
 532   curl_easy_setopt(easy_handle, CURLOPT_URL, url.c_str());
 533   curl_easy_setopt(easy_handle, CURLOPT_NOPROGRESS, 1L);
 534   curl_easy_setopt(easy_handle, CURLOPT_NOSIGNAL, 1L);
 535   curl_easy_setopt(easy_handle, CURLOPT_HEADERFUNCTION, receive_http_header);
 536   curl_easy_setopt(easy_handle, CURLOPT_WRITEHEADER, (void *)req_data);
 537   curl_easy_setopt(easy_handle, CURLOPT_WRITEFUNCTION, receive_http_data);
 538   curl_easy_setopt(easy_handle, CURLOPT_WRITEDATA, (void *)req_data);
 539   curl_easy_setopt(easy_handle, CURLOPT_ERRORBUFFER, (void *)req_data->error_buf);
 540   curl_easy_setopt(easy_handle, CURLOPT_LOW_SPEED_TIME, cct->_conf->rgw_curl_low_speed_time);
 541   curl_easy_setopt(easy_handle, CURLOPT_LOW_SPEED_LIMIT, cct->_conf->rgw_curl_low_speed_limit);
 542   curl_easy_setopt(easy_handle, CURLOPT_READFUNCTION, send_http_data);
 543   curl_easy_setopt(easy_handle, CURLOPT_READDATA, (void *)req_data);
 544   curl_easy_setopt(easy_handle, CURLOPT_BUFFERSIZE, cct->_conf->rgw_curl_buffersize);
 545   if (send_data_hint || is_upload_request(method)) {
 546     curl_easy_setopt(easy_handle, CURLOPT_UPLOAD, 1L);
 547   }
 548   if (has_send_len) {
 549     // TODO: prevent overflow by using curl_off_t
 550     // and: CURLOPT_INFILESIZE_LARGE, CURLOPT_POSTFIELDSIZE_LARGE
 551     const long size = send_len;
 552     curl_easy_setopt(easy_handle, CURLOPT_INFILESIZE, size);
 553     if (method == "POST") {
 554       curl_easy_setopt(easy_handle, CURLOPT_POSTFIELDSIZE, size);
 555       // TODO: set to size smaller than 1MB should prevent the "Expect" field
 556       // from being sent. So explicit removal is not needed
 557       h = curl_slist_append(h, "Expect:");
 558     }
 559   }
 560   if (h) {
 561     curl_easy_setopt(easy_handle, CURLOPT_HTTPHEADER, (void *)h);
 562   }
 563   if (!verify_ssl) {
 564     curl_easy_setopt(easy_handle, CURLOPT_SSL_VERIFYPEER, 0L);
 565     curl_easy_setopt(easy_handle, CURLOPT_SSL_VERIFYHOST, 0L);
 566     dout(20) << "ssl verification is set to off" << dendl;
 567   }
 568   curl_easy_setopt(easy_handle, CURLOPT_PRIVATE, (void *)req_data);
 569   curl_easy_setopt(easy_handle, CURLOPT_TIMEOUT, req_timeout);
 570
 571   return 0;
 572 }
 573
 574 bool RGWHTTPClient::is_done()
 575 {
 576   return req_data->is_done();
 577 }
 578
 579 /*
 580  * wait for async request to complete
 581  */
 582 int RGWHTTPClient::wait(optional_yield y)
 583 {
 584   return req_data->wait(y);
 585 }
 586
 587 void RGWHTTPClient::cancel()
 588 {
 589   if (req_data) {
 590     RGWHTTPManager *http_manager = req_data->mgr;
 591     if (http_manager) {
 592       http_manager->remove_request(this);
 593     }
 594   }
 595 }
 596
 597 RGWHTTPClient::~RGWHTTPClient()
 598 {
 599   cancel();
 600   if (req_data) {
 601     req_data->put();
 602   }
 603 }
 604
 605
 606 int RGWHTTPHeadersCollector::receive_header(void * const ptr, const size_t len)
 607 {
 608   const std::string_view header_line(static_cast<const char *>(ptr), len);
 609
 610   /* We're tokening the line that way due to backward compatibility. */
 611   const size_t sep_loc = header_line.find_first_of(" \t:");
 612
 613   if (std::string_view::npos == sep_loc) {
 614     /* Wrongly formatted header? Just skip it. */
 615     return 0;
 616   }
 617
 618   header_name_t name(header_line.substr(0, sep_loc));
 619   if (0 == relevant_headers.count(name)) {
 620     /* Not interested in this particular header. */
 621     return 0;
 622   }
 623
 624   const auto value_part = header_line.substr(sep_loc + 1);
 625
 626   /* Skip spaces and tabs after the separator. */
 627   const size_t val_loc_s = value_part.find_first_not_of(' ');
 628   const size_t val_loc_e = value_part.find_first_of("\r\n");
 629
 630   if (std::string_view::npos == val_loc_s ||
 631       std::string_view::npos == val_loc_e) {
 632     /* Empty value case. */
 633     found_headers.emplace(name, header_value_t());
 634   } else {
 635     found_headers.emplace(name, header_value_t(
 636         value_part.substr(val_loc_s, val_loc_e - val_loc_s)));
 637   }
 638
 639   return 0;
 640 }
 641
 642 int RGWHTTPTransceiver::send_data(void* ptr, size_t len, bool* pause)
 643 {
 644   int length_to_copy = 0;
 645   if (post_data_index < post_data.length()) {
 646     length_to_copy = min(post_data.length() - post_data_index, len);
 647     memcpy(ptr, post_data.data() + post_data_index, length_to_copy);
 648     post_data_index += length_to_copy;
 649   }
 650   return length_to_copy;
 651 }
 652
 653
 654 static int clear_signal(int fd)
 655 {
 656   // since we're in non-blocking mode, we can try to read a lot more than
 657   // one signal from signal_thread() to avoid later wakeups. non-blocking reads
 658   // are also required to support the curl_multi_wait bug workaround
 659   std::array<char, 256> buf;
 660   int ret = ::read(fd, (void *)buf.data(), buf.size());
 661   if (ret < 0) {
 662     ret = -errno;
 663     return ret == -EAGAIN ? 0 : ret; // clear EAGAIN
 664   }
 665   return 0;
 666 }
 667
 668 #if HAVE_CURL_MULTI_WAIT
 669
 670 static std::once_flag detect_flag;
 671 static bool curl_multi_wait_bug_present = false;
 672
 673 static int detect_curl_multi_wait_bug(CephContext *cct, CURLM *handle,
 674                                       int write_fd, int read_fd)
 675 {
 676   int ret = 0;
 677
 678   // write to write_fd so that read_fd becomes readable
 679   uint32_t buf = 0;
 680   ret = ::write(write_fd, &buf, sizeof(buf));
 681   if (ret < 0) {
 682     ret = -errno;
 683     ldout(cct, 0) << "ERROR: " << __func__ << "(): write() returned " << ret << dendl;
 684     return ret;
 685   }
 686
 687   // pass read_fd in extra_fds for curl_multi_wait()
 688   int num_fds;
 689   struct curl_waitfd wait_fd;
 690
 691   wait_fd.fd = read_fd;
 692   wait_fd.events = CURL_WAIT_POLLIN;
 693   wait_fd.revents = 0;
 694
 695   ret = curl_multi_wait(handle, &wait_fd, 1, 0, &num_fds);
 696   if (ret != CURLM_OK) {
 697     ldout(cct, 0) << "ERROR: curl_multi_wait() returned " << ret << dendl;
 698     return -EIO;
 699   }
 700
 701   // curl_multi_wait should flag revents when extra_fd is readable. if it
 702   // doesn't, the bug is present and we can't rely on revents
 703   if (wait_fd.revents == 0) {
 704     curl_multi_wait_bug_present = true;
 705     ldout(cct, 0) << "WARNING: detected a version of libcurl which contains a "
 706         "bug in curl_multi_wait(). enabling a workaround that may degrade "
 707         "performance slightly." << dendl;
 708   }
 709
 710   return clear_signal(read_fd);
 711 }
 712
 713 static bool is_signaled(const curl_waitfd& wait_fd)
 714 {
 715   if (wait_fd.fd < 0) {
 716     // no fd to signal
 717     return false;
 718   }
 719
 720   if (curl_multi_wait_bug_present) {
 721     // we can't rely on revents, so we always return true if a wait_fd is given.
 722     // this means we'll be trying a non-blocking read on this fd every time that
 723     // curl_multi_wait() wakes up
 724     return true;
 725   }
 726
 727   return wait_fd.revents > 0;
 728 }
 729
 730 static int do_curl_wait(CephContext *cct, CURLM *handle, int signal_fd)
 731 {
 732   int num_fds;
 733   struct curl_waitfd wait_fd;
 734
 735   wait_fd.fd = signal_fd;
 736   wait_fd.events = CURL_WAIT_POLLIN;
 737   wait_fd.revents = 0;
 738
 739   int ret = curl_multi_wait(handle, &wait_fd, 1, cct->_conf->rgw_curl_wait_timeout_ms, &num_fds);
 740   if (ret) {
 741     ldout(cct, 0) << "ERROR: curl_multi_wait() returned " << ret << dendl;
 742     return -EIO;
 743   }
 744
 745   if (is_signaled(wait_fd)) {
 746     ret = clear_signal(signal_fd);
 747     if (ret < 0) {
 748       ldout(cct, 0) << "ERROR: " << __func__ << "(): read() returned " << ret << dendl;
 749       return ret;
 750     }
 751   }
 752   return 0;
 753 }
 754
 755 #else
 756
 757 static int do_curl_wait(CephContext *cct, CURLM *handle, int signal_fd)
 758 {
 759   fd_set fdread;
 760   fd_set fdwrite;
 761   fd_set fdexcep;
 762   int maxfd = -1;
 763
 764   FD_ZERO(&fdread);
 765   FD_ZERO(&fdwrite);
 766   FD_ZERO(&fdexcep);
 767
 768   /* get file descriptors from the transfers */
 769   int ret = curl_multi_fdset(handle, &fdread, &fdwrite, &fdexcep, &maxfd);
 770   if (ret) {
 771     ldout(cct, 0) << "ERROR: curl_multi_fdset returned " << ret << dendl;
 772     return -EIO;
 773   }
 774
 775   if (signal_fd > 0) {
 776     FD_SET(signal_fd, &fdread);
 777     if (signal_fd >= maxfd) {
 778       maxfd = signal_fd + 1;
 779     }
 780   }
 781
 782   /* forcing a strict timeout, as the returned fdsets might not reference all fds we wait on */
 783   uint64_t to = cct->_conf->rgw_curl_wait_timeout_ms;
 784 #define RGW_CURL_TIMEOUT 1000
 785   if (!to)
 786     to = RGW_CURL_TIMEOUT;
 787   struct timeval timeout;
 788   timeout.tv_sec = to / 1000;
 789   timeout.tv_usec = to % 1000;
 790
 791   ret = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
 792   if (ret < 0) {
 793     ret = -errno;
 794     ldout(cct, 0) << "ERROR: select returned " << ret << dendl;
 795     return ret;
 796   }
 797
 798   if (signal_fd > 0 && FD_ISSET(signal_fd, &fdread)) {
 799     ret = clear_signal(signal_fd);
 800     if (ret < 0) {
 801       ldout(cct, 0) << "ERROR: " << __func__ << "(): read() returned " << ret << dendl;
 802       return ret;
 803     }
 804   }
 805
 806   return 0;
 807 }
 808
 809 #endif
 810
 811 void *RGWHTTPManager::ReqsThread::entry()
 812 {
 813   manager->reqs_thread_entry();
 814   return NULL;
 815 }
 816
 817 /*
 818  * RGWHTTPManager has two modes of operation: threaded and non-threaded.
 819  */
 820 RGWHTTPManager::RGWHTTPManager(CephContext *_cct, RGWCompletionManager *_cm) : cct(_cct),
 821                                                     completion_mgr(_cm)
 822 {
 823   multi_handle = (void *)curl_multi_init();
 824   thread_pipe[0] = -1;
 825   thread_pipe[1] = -1;
 826 }
 827
 828 RGWHTTPManager::~RGWHTTPManager() {
 829   stop();
 830   if (multi_handle)
 831     curl_multi_cleanup((CURLM *)multi_handle);
 832 }
 833
 834 void RGWHTTPManager::register_request(rgw_http_req_data *req_data)
 835 {
 836   std::unique_lock rl{reqs_lock};
 837   req_data->id = num_reqs;
 838   req_data->registered = true;
 839   reqs[num_reqs] = req_data;
 840   num_reqs++;
 841   ldout(cct, 20) << __func__ << " mgr=" << this << " req_data->id=" << req_data->id << ", curl_handle=" << req_data->curl_handle << dendl;
 842 }
 843
 844 bool RGWHTTPManager::unregister_request(rgw_http_req_data *req_data)
 845 {
 846   std::unique_lock rl{reqs_lock};
 847   if (!req_data->registered) {
 848     return false;
 849   }
 850   req_data->get();
 851   req_data->registered = false;
 852   unregistered_reqs.push_back(req_data);
 853   ldout(cct, 20) << __func__ << " mgr=" << this << " req_data->id=" << req_data->id << ", curl_handle=" << req_data->curl_handle << dendl;
 854   return true;
 855 }
 856
 857 void RGWHTTPManager::complete_request(rgw_http_req_data *req_data)
 858 {
 859   std::unique_lock rl{reqs_lock};
 860   _complete_request(req_data);
 861 }
 862
 863 void RGWHTTPManager::_complete_request(rgw_http_req_data *req_data)
 864 {
 865   map<uint64_t, rgw_http_req_data *>::iterator iter = reqs.find(req_data->id);
 866   if (iter != reqs.end()) {
 867     reqs.erase(iter);
 868   }
 869   {
 870     std::lock_guard l{req_data->lock};
 871     req_data->mgr = nullptr;
 872   }
 873   if (completion_mgr) {
 874     completion_mgr->complete(NULL, req_data->control_io_id, req_data->user_info);
 875   }
 876
 877   req_data->put();
 878 }
 879
 880 void RGWHTTPManager::finish_request(rgw_http_req_data *req_data, int ret, long http_status)
 881 {
 882   req_data->finish(ret, http_status);
 883   complete_request(req_data);
 884 }
 885
 886 void RGWHTTPManager::_finish_request(rgw_http_req_data *req_data, int ret)
 887 {
 888   req_data->finish(ret);
 889   _complete_request(req_data);
 890 }
 891
 892 void RGWHTTPManager::_set_req_state(set_state& ss)
 893 {
 894   ss.req->set_state(ss.bitmask);
 895 }
 896 /*
 897  * hook request to the curl multi handle
 898  */
 899 int RGWHTTPManager::link_request(rgw_http_req_data *req_data)
 900 {
 901   ldout(cct, 20) << __func__ << " req_data=" << req_data << " req_data->id=" << req_data->id << ", curl_handle=" << req_data->curl_handle << dendl;
 902   CURLMcode mstatus = curl_multi_add_handle((CURLM *)multi_handle, req_data->get_easy_handle());
 903   if (mstatus) {
 904     dout(0) << "ERROR: failed on curl_multi_add_handle, status=" << mstatus << dendl;
 905     return -EIO;
 906   }
 907   return 0;
 908 }
 909
 910 /*
 911  * unhook request from the curl multi handle, and finish request if it wasn't finished yet as
 912  * there will be no more processing on this request
 913  */
 914 void RGWHTTPManager::_unlink_request(rgw_http_req_data *req_data)
 915 {
 916   if (req_data->curl_handle) {
 917     curl_multi_remove_handle((CURLM *)multi_handle, req_data->get_easy_handle());
 918   }
 919   if (!req_data->is_done()) {
 920     _finish_request(req_data, -ECANCELED);
 921   }
 922 }
 923
 924 void RGWHTTPManager::unlink_request(rgw_http_req_data *req_data)
 925 {
 926   std::unique_lock wl{reqs_lock};
 927   _unlink_request(req_data);
 928 }
 929
 930 void RGWHTTPManager::manage_pending_requests()
 931 {
 932   reqs_lock.lock_shared();
 933   if (max_threaded_req == num_reqs &&
 934       unregistered_reqs.empty() &&
 935       reqs_change_state.empty()) {
 936     reqs_lock.unlock_shared();
 937     return;
 938   }
 939   reqs_lock.unlock_shared();
 940
 941   std::unique_lock wl{reqs_lock};
 942
 943   if (!reqs_change_state.empty()) {
 944     for (auto siter : reqs_change_state) {
 945       _set_req_state(siter);
 946     }
 947     reqs_change_state.clear();
 948   }
 949
 950   if (!unregistered_reqs.empty()) {
 951     for (auto& r : unregistered_reqs) {
 952       _unlink_request(r);
 953       r->put();
 954     }
 955
 956     unregistered_reqs.clear();
 957   }
 958
 959   map<uint64_t, rgw_http_req_data *>::iterator iter = reqs.find(max_threaded_req);
 960
 961   list<std::pair<rgw_http_req_data *, int> > remove_reqs;
 962
 963   for (; iter != reqs.end(); ++iter) {
 964     rgw_http_req_data *req_data = iter->second;
 965     int r = link_request(req_data);
 966     if (r < 0) {
 967       ldout(cct, 0) << "ERROR: failed to link http request" << dendl;
 968       remove_reqs.push_back(std::make_pair(iter->second, r));
 969     } else {
 970       max_threaded_req = iter->first + 1;
 971     }
 972   }
 973
 974   for (auto piter : remove_reqs) {
 975     rgw_http_req_data *req_data = piter.first;
 976     int r = piter.second;
 977
 978     _finish_request(req_data, r);
 979   }
 980 }
 981
 982 int RGWHTTPManager::add_request(RGWHTTPClient *client)
 983 {
 984   rgw_http_req_data *req_data = new rgw_http_req_data;
 985
 986   int ret = client->init_request(req_data);
 987   if (ret < 0) {
 988     req_data->put();
 989     req_data = NULL;
 990     return ret;
 991   }
 992
 993   req_data->mgr = this;
 994   req_data->client = client;
 995   req_data->control_io_id = client->get_io_id(RGWHTTPClient::HTTPCLIENT_IO_CONTROL);
 996   req_data->user_info = client->get_io_user_info();
 997
 998   register_request(req_data);
 999
1000   if (!is_started) {
1001     ret = link_request(req_data);
1002     if (ret < 0) {
1003       req_data->put();
1004       req_data = NULL;
1005     }
1006     return ret;
1007   }
1008   ret = signal_thread();
1009   if (ret < 0) {
1010     finish_request(req_data, ret);
1011   }
1012
1013   return ret;
1014 }
1015
1016 int RGWHTTPManager::remove_request(RGWHTTPClient *client)
1017 {
1018   rgw_http_req_data *req_data = client->get_req_data();
1019
1020   if (!is_started) {
1021     unlink_request(req_data);
1022     return 0;
1023   }
1024   if (!unregister_request(req_data)) {
1025     return 0;
1026   }
1027   int ret = signal_thread();
1028   if (ret < 0) {
1029     return ret;
1030   }
1031
1032   return 0;
1033 }
1034
1035 int RGWHTTPManager::set_request_state(RGWHTTPClient *client, RGWHTTPRequestSetState state)
1036 {
1037   rgw_http_req_data *req_data = client->get_req_data();
1038
1039   ceph_assert(ceph_mutex_is_locked(req_data->lock));
1040
1041   /* can only do that if threaded */
1042   if (!is_started) {
1043     return -EINVAL;
1044   }
1045
1046   bool suggested_wr_paused = req_data->write_paused;
1047   bool suggested_rd_paused = req_data->read_paused;
1048
1049   switch (state) {
1050     case SET_WRITE_PAUSED:
1051       suggested_wr_paused = true;
1052       break;
1053     case SET_WRITE_RESUME:
1054       suggested_wr_paused = false;
1055       break;
1056     case SET_READ_PAUSED:
1057       suggested_rd_paused = true;
1058       break;
1059     case SET_READ_RESUME:
1060       suggested_rd_paused = false;
1061       break;
1062     default:
1063       /* shouldn't really be here */
1064       return -EIO;
1065   }
1066   if (suggested_wr_paused == req_data->write_paused &&
1067       suggested_rd_paused == req_data->read_paused) {
1068     return 0;
1069   }
1070
1071   req_data->write_paused = suggested_wr_paused;
1072   req_data->read_paused = suggested_rd_paused;
1073
1074   int bitmask = CURLPAUSE_CONT;
1075
1076   if (req_data->write_paused) {
1077     bitmask |= CURLPAUSE_SEND;
1078   }
1079
1080   if (req_data->read_paused) {
1081     bitmask |= CURLPAUSE_RECV;
1082   }
1083
1084   reqs_change_state.push_back(set_state(req_data, bitmask));
1085   int ret = signal_thread();
1086   if (ret < 0) {
1087     return ret;
1088   }
1089
1090   return 0;
1091 }
1092
1093 int RGWHTTPManager::start()
1094 {
1095   if (pipe_cloexec(thread_pipe, 0) < 0) {
1096     int e = errno;
1097     ldout(cct, 0) << "ERROR: pipe(): " << cpp_strerror(e) << dendl;
1098     return -e;
1099   }
1100
1101   // enable non-blocking reads
1102   if (::fcntl(thread_pipe[0], F_SETFL, O_NONBLOCK) < 0) {
1103     int e = errno;
1104     ldout(cct, 0) << "ERROR: fcntl(): " << cpp_strerror(e) << dendl;
1105     TEMP_FAILURE_RETRY(::close(thread_pipe[0]));
1106     TEMP_FAILURE_RETRY(::close(thread_pipe[1]));
1107     return -e;
1108   }
1109
1110 #ifdef HAVE_CURL_MULTI_WAIT
1111   // on first initialization, use this pipe to detect whether we're using a
1112   // buggy version of libcurl
1113   std::call_once(detect_flag, detect_curl_multi_wait_bug, cct,
1114                  static_cast<CURLM*>(multi_handle),
1115                  thread_pipe[1], thread_pipe[0]);
1116 #endif
1117
1118   is_started = true;
1119   reqs_thread = new ReqsThread(this);
1120   reqs_thread->create("http_manager");
1121   return 0;
1122 }
1123
1124 void RGWHTTPManager::stop()
1125 {
1126   if (is_stopped) {
1127     return;
1128   }
1129
1130   is_stopped = true;
1131
1132   if (is_started) {
1133     going_down = true;
1134     signal_thread();
1135     reqs_thread->join();
1136     delete reqs_thread;
1137     TEMP_FAILURE_RETRY(::close(thread_pipe[1]));
1138     TEMP_FAILURE_RETRY(::close(thread_pipe[0]));
1139   }
1140 }
1141
1142 int RGWHTTPManager::signal_thread()
1143 {
1144   uint32_t buf = 0;
1145   int ret = write(thread_pipe[1], (void *)&buf, sizeof(buf));
1146   if (ret < 0) {
1147     ret = -errno;
1148     ldout(cct, 0) << "ERROR: " << __func__ << ": write() returned ret=" << ret << dendl;
1149     return ret;
1150   }
1151   return 0;
1152 }
1153
1154 void *RGWHTTPManager::reqs_thread_entry()
1155 {
1156   int still_running;
1157   int mstatus;
1158
1159   ldout(cct, 20) << __func__ << ": start" << dendl;
1160
1161   while (!going_down) {
1162     int ret = do_curl_wait(cct, (CURLM *)multi_handle, thread_pipe[0]);
1163     if (ret < 0) {
1164       dout(0) << "ERROR: do_curl_wait() returned: " << ret << dendl;
1165       return NULL;
1166     }
1167
1168     manage_pending_requests();
1169
1170     mstatus = curl_multi_perform((CURLM *)multi_handle, &still_running);
1171     switch (mstatus) {
1172       case CURLM_OK:
1173       case CURLM_CALL_MULTI_PERFORM:
1174         break;
1175       default:
1176         dout(10) << "curl_multi_perform returned: " << mstatus << dendl;
1177         break;
1178     }
1179     int msgs_left;
1180     CURLMsg *msg;
1181     while ((msg = curl_multi_info_read((CURLM *)multi_handle, &msgs_left))) {
1182       if (msg->msg == CURLMSG_DONE) {
1183         int result = msg->data.result;
1184         CURL *e = msg->easy_handle;
1185         rgw_http_req_data *req_data;
1186         curl_easy_getinfo(e, CURLINFO_PRIVATE, (void **)&req_data);
1187         curl_multi_remove_handle((CURLM *)multi_handle, e);
1188
1189         long http_status;
1190         int status;
1191         if (!req_data->user_ret) {
1192           curl_easy_getinfo(e, CURLINFO_RESPONSE_CODE, (void **)&http_status);
1193
1194           status = rgw_http_error_to_errno(http_status);
1195           if (result != CURLE_OK && status == 0) {
1196             dout(0) << "ERROR: curl error: " << curl_easy_strerror((CURLcode)result) << ", maybe network unstable" << dendl;
1197             status = -EAGAIN;
1198           }
1199         } else {
1200           status = *req_data->user_ret;
1201           rgw_err err;
1202           set_req_state_err(err, status, 0);
1203           http_status = err.http_ret;
1204         }
1205         int id = req_data->id;
1206         finish_request(req_data, status, http_status);
1207         switch (result) {
1208           case CURLE_OK:
1209             break;
1210           case CURLE_OPERATION_TIMEDOUT:
1211             dout(0) << "WARNING: curl operation timed out, network average transfer speed less than "
1212               << cct->_conf->rgw_curl_low_speed_limit << " Bytes per second during " << cct->_conf->rgw_curl_low_speed_time << " seconds." << dendl;
1213           default:
1214             dout(20) << "ERROR: msg->data.result=" << result << " req_data->id=" << id << " http_status=" << http_status << dendl;
1215             dout(20) << "ERROR: curl error: " << curl_easy_strerror((CURLcode)result) << dendl;
1216             break;
1217         }
1218       }
1219     }
1220   }
1221
1222
1223   std::unique_lock rl{reqs_lock};
1224   for (auto r : unregistered_reqs) {
1225     _unlink_request(r);
1226   }
1227
1228   unregistered_reqs.clear();
1229
1230   auto all_reqs = std::move(reqs);
1231   for (auto iter : all_reqs) {
1232     _unlink_request(iter.second);
1233   }
1234
1235   reqs.clear();
1236
1237   if (completion_mgr) {
1238     completion_mgr->go_down();
1239   }
1240
1241   return 0;
1242 }
1243
1244 void rgw_http_client_init(CephContext *cct)
1245 {
1246   curl_global_init(CURL_GLOBAL_ALL);
1247   rgw_http_manager = new RGWHTTPManager(cct);
1248   rgw_http_manager->start();
1249 }
1250
1251 void rgw_http_client_cleanup()
1252 {
1253   rgw_http_manager->stop();
1254   delete rgw_http_manager;
1255   curl_global_cleanup();
1256 }
1257
1258
1259 int RGWHTTP::send(RGWHTTPClient *req) {
1260   if (!req) {
1261     return 0;
1262   }
1263   int r = rgw_http_manager->add_request(req);
1264   if (r < 0) {
1265     return r;
1266   }
1267
1268   return 0;
1269 }
1270
1271 int RGWHTTP::process(RGWHTTPClient *req, optional_yield y) {
1272   if (!req) {
1273     return 0;
1274   }
1275   int r = send(req);
1276   if (r < 0) {
1277     return r;
1278   }
1279
1280   return req->wait(y);
1281 }
1282