rgw: call appropriate curl calls for waiting on sockets

If libcurl supports curl_multi_wait() then use it, otherwise
use select() and force a timeout, even if it has been disabled.
Otherwise we may wait forever for events that we can't wait for
as select() only uses fds < 1024.

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index f8e888b56b0b369eb667b0b616eac053cd072f9b..45f0bb4f56a606f472039b856ab12bf5e07a1596 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -670,6 +670,7 @@ OPTION(rgw_list_buckets_max_chunk, OPT_INT, 1000) // max buckets to retrieve in
 OPTION(rgw_md_log_max_shards, OPT_INT, 64) // max shards for metadata log
 OPTION(rgw_num_zone_opstate_shards, OPT_INT, 128) // max shards for keeping inter-region copy progress info
 OPTION(rgw_opstate_ratelimit_sec, OPT_INT, 30) // min time between opstate updates on a single upload (0 for disabling ratelimit)
+OPTION(rgw_curl_wait_timeout_ms, OPT_INT, 1000) // timeout for certain curl calls
 
 OPTION(rgw_data_log_window, OPT_INT, 30) // data log entries window (in seconds)
 OPTION(rgw_data_log_changes_size, OPT_INT, 1000) // number of in-memory entries to hold for data changes log

diff --git a/src/rgw/rgw_http_client.cc b/src/rgw/rgw_http_client.cc
index 0a000bca40cf88395acca1399bdefb3a58cb9b47..314e80b9ef26cf9f5ad4ad13b4de5a130b5b6bad 100644 (file)
--- a/src/rgw/rgw_http_client.cc
+++ b/src/rgw/rgw_http_client.cc
@@ -178,6 +178,59 @@ int RGWHTTPClient::init_async(const char *method, const char *url, void **handle
   return 0;
 }
 
+#if HAVE_CURL_MULTI_WAIT
+
+static int do_curl_wait(CephContext *cct, CURLM *handle)
+{
+  int num_fds;
+  int ret = curl_multi_wait(handle, NULL, 0, cct->_conf->rgw_curl_wait_timeout_ms, &num_fds);
+  if (ret) {
+    dout(0) << "ERROR: curl_multi_wait() returned " << ret << dendl;
+    return -EIO;
+  }
+  return 0;
+}
+
+#else
+
+static int do_curl_wait(CephContext *cct, CURLM *handle)
+{
+  fd_set fdread;
+  fd_set fdwrite;
+  fd_set fdexcep;
+  int maxfd = -1;
+ 
+  FD_ZERO(&fdread);
+  FD_ZERO(&fdwrite);
+  FD_ZERO(&fdexcep);
+
+  /* get file descriptors from the transfers */ 
+  int ret = curl_multi_fdset(handle, &fdread, &fdwrite, &fdexcep, &maxfd);
+  if (ret) {
+    dout(0) << "ERROR: curl_multi_fdset returned " << ret << dendl;
+    return -EIO;
+  }
+
+  /* forcing a strict timeout, as the returned fdsets might not reference all fds we wait on */
+  uint64_t to = cct->_conf->rgw_curl_wait_timeout_ms;
+#define RGW_CURL_TIMEOUT 1000
+  if (!to)
+    to = RGW_CURL_TIMEOUT;
+  struct timeval timeout;
+  timeout.tv_sec = to / 1000;
+  timeout.tv_usec = to % 1000;
+
+  ret = select(maxfd+1, &fdread, &fdwrite, &fdexcep, &timeout);
+  if (ret < 0) {
+    ret = -errno;
+    dout(0) << "ERROR: select returned " << ret << dendl;
+    return ret;
+  }
+
+  return 0;
+}
+
+#endif
 
 int RGWHTTPClient::process_request(void *handle, bool wait_for_data, bool *done)
 {
@@ -187,45 +240,8 @@ int RGWHTTPClient::process_request(void *handle, bool wait_for_data, bool *done)
 
   do {
     if (wait_for_data) {
-      struct timeval timeout;
- 
-      fd_set fdread;
-      fd_set fdwrite;
-      fd_set fdexcep;
-      int maxfd = -1;
- 
-      long curl_timeo = -1;
- 
-      FD_ZERO(&fdread);
-      FD_ZERO(&fdwrite);
-      FD_ZERO(&fdexcep);
-#if 0 
-    /* set a suitable timeout to play around with */ 
-    timeout.tv_sec = 1;
-    timeout.tv_usec = 0;
- 
-    curl_multi_timeout(multi_handle, &curl_timeo);
-    if(curl_timeo >= 0) {
-      timeout.tv_sec = curl_timeo / 1000;
-      if(timeout.tv_sec > 1)
-        timeout.tv_sec = 1;
-      else
-        timeout.tv_usec = (curl_timeo % 1000) * 1000;
-    }
-#endif
- 
-      /* get file descriptors from the transfers */ 
-      int ret = curl_multi_fdset(req_data->multi_handle, &fdread, &fdwrite, &fdexcep, &maxfd);
-      if (ret) {
-        dout(0) << "ERROR: curl_multi_fdset returned " << ret << dendl;
-        return -EIO;
-      }
-
-#warning FIXME: replace select with poll 
-      ret = select(maxfd+1, &fdread, &fdwrite, &fdexcep, NULL);
+      int ret = do_curl_wait(cct, req_data->multi_handle);
       if (ret < 0) {
-        ret = -errno;
-        dout(0) << "ERROR: select returned " << ret << dendl;
         return ret;
       }
     }
@@ -243,8 +259,13 @@ int RGWHTTPClient::process_request(void *handle, bool wait_for_data, bool *done)
     CURLMsg *msg;
     while ((msg = curl_multi_info_read(req_data->multi_handle, &msgs_left))) {
       if (msg->msg == CURLMSG_DONE) {
-#warning FIXME: check result
-        dout(20) << "msg->data.result=" << msg->data.result << dendl;
+        switch (msg->data.result) {
+          case CURLE_OK:
+            break;
+          default:
+            dout(20) << "ERROR: msg->data.result=" << msg->data.result << dendl;
+            return -EIO;
+        }
       }
     }
   } while (mstatus == CURLM_CALL_MULTI_PERFORM);

diff --git a/src/rgw/rgw_http_client.h b/src/rgw/rgw_http_client.h
index d55230095fb8f925b8683ed17e07dbcfec3d5de6..cc69383b62de4ebfa7f7b87a57288ad3a7d94c1b 100644 (file)
--- a/src/rgw/rgw_http_client.h
+++ b/src/rgw/rgw_http_client.h
@@ -10,10 +10,12 @@ class RGWHTTPClient
   size_t send_len;
   bool has_send_len;
 protected:
+  CephContext *cct;
+
   list<pair<string, string> > headers;
 public:
   virtual ~RGWHTTPClient() {}
-  RGWHTTPClient(): send_len (0), has_send_len(false) {}
+  RGWHTTPClient(CephContext *_cct): send_len (0), has_send_len(false), cct(_cct) {}
 
   void append_header(const string& name, const string& val) {
     headers.push_back(pair<string, string>(name, val));

diff --git a/src/rgw/rgw_rest_client.h b/src/rgw/rgw_rest_client.h
index b709fd49a1aa8d4ee0ef535e2367812f9aee984c..a12c3dd7ff2115f153c6b2066805a3bfc6f5ae7a 100644 (file)
--- a/src/rgw/rgw_rest_client.h
+++ b/src/rgw/rgw_rest_client.h
@@ -9,8 +9,6 @@ class RGWGetDataCB;
 
 class RGWRESTSimpleRequest : public RGWHTTPClient {
 protected:
-  CephContext *cct;
-
   int http_status;
   int status;
 
@@ -31,7 +29,7 @@ protected:
   int sign_request(RGWAccessKey& key, RGWEnv& env, req_info& info);
 public:
   RGWRESTSimpleRequest(CephContext *_cct, string& _url, list<pair<string, string> > *_headers,
-                list<pair<string, string> > *_params) : cct(_cct), http_status(0), status(0),
+                list<pair<string, string> > *_params) : RGWHTTPClient(_cct), http_status(0), status(0),
                 url(_url), send_iter(NULL),
                 max_response(0) {
     if (_headers)

diff --git a/src/rgw/rgw_swift.cc b/src/rgw/rgw_swift.cc
index d42bc579e762663342e0e1d37f846bed88a87cd7..9be9757e5f89a29d490132b2bd7a09b4f95f307b 100644 (file)
--- a/src/rgw/rgw_swift.cc
+++ b/src/rgw/rgw_swift.cc
@@ -21,13 +21,12 @@ static list<string> roles_list;
 class RGWKeystoneTokenCache;
 
 class RGWValidateSwiftToken : public RGWHTTPClient {
-  CephContext *cct;
   struct rgw_swift_auth_info *info;
 
 protected:
-  RGWValidateSwiftToken() : cct(NULL), info(NULL) {}
+  RGWValidateSwiftToken() : RGWHTTPClient(NULL), info(NULL) {}
 public:
-  RGWValidateSwiftToken(CephContext *_cct, struct rgw_swift_auth_info *_info) : cct(_cct), info(_info) {}
+  RGWValidateSwiftToken(CephContext *_cct, struct rgw_swift_auth_info *_info) : RGWHTTPClient(_cct), info(_info) {}
 
   int read_header(void *ptr, size_t len);
 
@@ -290,7 +289,7 @@ void RGWKeystoneTokenCache::invalidate(const string& token_id)
 class RGWValidateKeystoneToken : public RGWHTTPClient {
   bufferlist *bl;
 public:
-  RGWValidateKeystoneToken(bufferlist *_bl) : bl(_bl) {}
+  RGWValidateKeystoneToken(CephContext *_cct, bufferlist *_bl) : RGWHTTPClient(_cct), bl(_bl) {}
 
   int read_data(void *ptr, size_t len) {
     bl->append((char *)ptr, len);
@@ -303,7 +302,7 @@ static RGWKeystoneTokenCache *keystone_token_cache = NULL;
 class RGWGetRevokedTokens : public RGWHTTPClient {
   bufferlist *bl;
 public:
-  RGWGetRevokedTokens(bufferlist *_bl) : bl(_bl) {}
+  RGWGetRevokedTokens(CephContext *_cct, bufferlist *_bl) : RGWHTTPClient(_cct), bl(_bl) {}
 
   int read_data(void *ptr, size_t len) {
     bl->append((char *)ptr, len);
@@ -383,7 +382,7 @@ static int decode_b64_cms(CephContext *cct, const string& signed_b64, bufferlist
 int RGWSwift::check_revoked()
 {
   bufferlist bl;
-  RGWGetRevokedTokens req(&bl);
+  RGWGetRevokedTokens req(cct, &bl);
 
   string url = g_conf->rgw_keystone_url;
   if (url.empty()) {
@@ -590,7 +589,7 @@ int RGWSwift::validate_keystone_token(RGWRados *store, const string& token, stru
 
     /* can't decode, just go to the keystone server for validation */
 
-    RGWValidateKeystoneToken validate(&bl);
+    RGWValidateKeystoneToken validate(cct, &bl);
 
     string url = g_conf->rgw_keystone_url;
     if (url.empty()) {