rgw: configurable write obj window size

Fixes: http://tracker.ceph.com/issues/18623
Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
(cherry picked from commit 66a82b4266acedfdd71c64394d68d9e50ed11b20)

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index c814decb7b9e963f3d9f72f67a1745f9c8b83ac1..91acc2c7ef54da3cc71409e938657ac70745da47 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -1353,6 +1353,8 @@ OPTION(nss_db_path, OPT_STR, "") // path to nss db
 
 
 OPTION(rgw_max_chunk_size, OPT_INT, 4 * 1024 * 1024)
+OPTION(rgw_put_obj_min_window_size, OPT_INT, 16 * 1024 * 1024)
+OPTION(rgw_put_obj_max_window_size, OPT_INT, 64 * 1024 * 1024)
 OPTION(rgw_max_put_size, OPT_U64, 5ULL*1024*1024*1024)
 
 /**

diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index b852520e4c6ecb7b1f34486b1aae4b7d0f5910e8..e3acf3c2059e0a4a5feec0e1ffc0ee35397732e9 100644 (file)
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -117,8 +117,6 @@ using ceph::crypto::MD5;
 
 #define RGW_BUCKETS_OBJ_SUFFIX ".buckets"
 
-#define RGW_MAX_PENDING_CHUNKS  16
-
 #define RGW_FORMAT_PLAIN        0
 #define RGW_FORMAT_XML          1
 #define RGW_FORMAT_JSON         2

diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index 44a693c10bb805abddb49fcde8ab9efba5fae79a..d1098c341e752ad658310cc0548b6825d61bef29 100644 (file)
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -779,8 +779,8 @@ public:
   virtual int handle_data(bufferlist& bl, off_t ofs, void **phandle, rgw_obj *pobj, bool *again) override {
     return next->handle_data(bl, ofs, phandle, pobj, again);
   }
-  virtual int throttle_data(void *handle, const rgw_obj& obj, bool need_to_wait) override {
-    return next->throttle_data(handle, obj, need_to_wait);
+  virtual int throttle_data(void *handle, const rgw_obj& obj, uint64_t size, bool need_to_wait) override {
+    return next->throttle_data(handle, obj, size, need_to_wait);
   }
 }; /* RGWPutObj_Filter */
 
@@ -1606,11 +1606,13 @@ static inline int put_data_and_throttle(RGWPutObjDataProcessor *processor,
     void *handle;
     rgw_obj obj;
 
+    uint64_t size = data.length();
+
     int ret = processor->handle_data(data, ofs, &handle, &obj, &again);
     if (ret < 0)
       return ret;
 
-    ret = processor->throttle_data(handle, obj, need_to_wait);
+    ret = processor->throttle_data(handle, obj, size, need_to_wait);
     if (ret < 0)
       return ret;
 

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 47f3790b3313f8bec57941863b15ab60059f5621..9149c601e4aa8afb95f440d9da0067ee27d56683 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -2290,6 +2290,7 @@ struct put_obj_aio_info RGWPutObjProcessor_Aio::pop_pending()
   struct put_obj_aio_info info;
   info = pending.front();
   pending.pop_front();
+  pending_size -= info.size;
   return info;
 }
 
@@ -2328,7 +2329,7 @@ int RGWPutObjProcessor_Aio::drain_pending()
   return ret;
 }
 
-int RGWPutObjProcessor_Aio::throttle_data(void *handle, const rgw_obj& obj, bool need_to_wait)
+int RGWPutObjProcessor_Aio::throttle_data(void *handle, const rgw_obj& obj, uint64_t size, bool need_to_wait)
 {
   bool _wait = need_to_wait;
 
@@ -2336,9 +2337,11 @@ int RGWPutObjProcessor_Aio::throttle_data(void *handle, const rgw_obj& obj, bool
     struct put_obj_aio_info info;
     info.handle = handle;
     info.obj = obj;
+    info.size = size;
+    pending_size += size;
     pending.push_back(info);
   }
-  size_t orig_size = pending.size();
+  size_t orig_size = pending_size;
 
   /* first drain complete IOs */
   while (pending_has_completed()) {
@@ -2350,12 +2353,16 @@ int RGWPutObjProcessor_Aio::throttle_data(void *handle, const rgw_obj& obj, bool
   }
 
   /* resize window in case messages are draining too fast */
-  if (orig_size - pending.size() >= max_chunks) {
-    max_chunks++;
+  if (orig_size - pending_size >= window_size) {
+    window_size += store->ctx()->_conf->rgw_max_chunk_size;
+    uint64_t max_window_size = store->ctx()->_conf->rgw_put_obj_max_window_size;
+    if (window_size > max_window_size) {
+      window_size = max_window_size;
+    }
   }
 
   /* now throttle. Note that need_to_wait should only affect the first IO operation */
-  if (pending.size() > max_chunks || _wait) {
+  if (pending_size > window_size || _wait) {
     int r = wait_pending_front();
     if (r < 0)
       return r;
@@ -2377,6 +2384,15 @@ int RGWPutObjProcessor_Atomic::write_data(bufferlist& bl, off_t ofs, void **phan
   return RGWPutObjProcessor_Aio::handle_obj_data(cur_obj, bl, ofs - cur_part_ofs, ofs, phandle, exclusive);
 }
 
+int RGWPutObjProcessor_Aio::prepare(RGWRados *store, string *oid_rand)
+{
+  RGWPutObjProcessor::prepare(store, oid_rand);
+
+  window_size = store->ctx()->_conf->rgw_put_obj_min_window_size;
+
+  return 0;
+}
+
 int RGWPutObjProcessor_Atomic::handle_data(bufferlist& bl, off_t ofs, void **phandle, rgw_obj *pobj, bool *again)
 {
   *phandle = NULL;
@@ -2418,7 +2434,7 @@ int RGWPutObjProcessor_Atomic::handle_data(bufferlist& bl, off_t ofs, void **pha
 
 int RGWPutObjProcessor_Atomic::prepare_init(RGWRados *store, string *oid_rand)
 {
-  RGWPutObjProcessor::prepare(store, oid_rand);
+  RGWPutObjProcessor_Aio::prepare(store, oid_rand);
 
   int r = store->get_max_chunk_size(bucket, &max_chunk_size);
   if (r < 0) {
@@ -2495,13 +2511,14 @@ int RGWPutObjProcessor_Atomic::complete_writing_data()
     }
     bufferlist bl;
     pending_data_bl.splice(0, max_write_size, &bl);
+    uint64_t write_len = bl.length();
     int r = write_data(bl, data_ofs, &handle, &obj, false);
     if (r < 0) {
       ldout(store->ctx(), 0) << "ERROR: write_data() returned " << r << dendl;
       return r;
     }
-    data_ofs += bl.length();
-    r = throttle_data(handle, obj, false);
+    data_ofs += write_len;
+    r = throttle_data(handle, obj, write_len, false);
     if (r < 0) {
       ldout(store->ctx(), 0) << "ERROR: throttle_data() returned " << r << dendl;
       return r;
@@ -6719,6 +6736,7 @@ public:
     do {
       void *handle = NULL;
       rgw_obj obj;
+      uint64_t size = bl.length();
       int ret = filter->handle_data(bl, ofs, &handle, &obj, &again);
       if (ret < 0)
         return ret;
@@ -6730,7 +6748,7 @@ public:
         ret = opstate->renew_state();
         if (ret < 0) {
           ldout(cct, 0) << "ERROR: RGWRadosPutObj::handle_data(): failed to renew op state ret=" << ret << dendl;
-          int r = filter->throttle_data(handle, obj, false);
+          int r = filter->throttle_data(handle, obj, size, false);
           if (r < 0) {
             ldout(cct, 0) << "ERROR: RGWRadosPutObj::handle_data(): processor->throttle_data() returned " << r << dendl;
           }
@@ -6741,7 +6759,7 @@ public:
         need_opstate = false;
       }
 
-      ret = filter->throttle_data(handle, obj, false);
+      ret = filter->throttle_data(handle, obj, size, false);
       if (ret < 0)
         return ret;
     } while (again);
@@ -7654,7 +7672,7 @@ int RGWRados::copy_obj_data(RGWObjectCtx& obj_ctx,
       if (ret < 0) {
         return ret;
       }
-      ret = processor.throttle_data(handle, obj, false);
+      ret = processor.throttle_data(handle, obj, end - ofs + 1, false);
       if (ret < 0)
         return ret;
     } while (again);

diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index c44cd12c67bd4220ac61d2dbfa3557500aeabad9..d25156decf0d135761421d13b1fbd1490a43a970 100644 (file)
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -3328,7 +3328,7 @@ public:
   RGWPutObjDataProcessor(){}
   virtual ~RGWPutObjDataProcessor(){}
   virtual int handle_data(bufferlist& bl, off_t ofs, void **phandle, rgw_obj *pobj, bool *again) = 0;
-  virtual int throttle_data(void *handle, const rgw_obj& obj, bool need_to_wait) = 0;
+  virtual int throttle_data(void *handle, const rgw_obj& obj, uint64_t size, bool need_to_wait) = 0;
 }; /* RGWPutObjDataProcessor */
 
 
@@ -3371,12 +3371,16 @@ public:
 struct put_obj_aio_info {
   void *handle;
   rgw_obj obj;
+  uint64_t size;
 };
 
+#define RGW_PUT_OBJ_MIN_WINDOW_SIZE_DEFAULT (16 * 1024 * 1024)
+
 class RGWPutObjProcessor_Aio : public RGWPutObjProcessor
 {
   list<struct put_obj_aio_info> pending;
-  size_t max_chunks;
+  uint64_t window_size{RGW_PUT_OBJ_MIN_WINDOW_SIZE_DEFAULT};
+  uint64_t pending_size{0};
 
   struct put_obj_aio_info pop_pending();
   int wait_pending_front();
@@ -3385,7 +3389,7 @@ class RGWPutObjProcessor_Aio : public RGWPutObjProcessor
   rgw_obj last_written_obj;
 
 protected:
-  uint64_t obj_len;
+  uint64_t obj_len{0};
 
   set<rgw_obj> written_objs;
 
@@ -3397,9 +3401,10 @@ protected:
   int handle_obj_data(rgw_obj& obj, bufferlist& bl, off_t ofs, off_t abs_ofs, void **phandle, bool exclusive);
 
 public:
-  int throttle_data(void *handle, const rgw_obj& obj, bool need_to_wait);
+  int prepare(RGWRados *store, string *oid_rand);
+  int throttle_data(void *handle, const rgw_obj& obj, uint64_t size, bool need_to_wait);
 
-  RGWPutObjProcessor_Aio(RGWObjectCtx& obj_ctx, RGWBucketInfo& bucket_info) : RGWPutObjProcessor(obj_ctx, bucket_info), max_chunks(RGW_MAX_PENDING_CHUNKS), obj_len(0) {}
+  RGWPutObjProcessor_Aio(RGWObjectCtx& obj_ctx, RGWBucketInfo& bucket_info) : RGWPutObjProcessor(obj_ctx, bucket_info) {}
   virtual ~RGWPutObjProcessor_Aio();
 }; /* RGWPutObjProcessor_Aio */