rgw: handle stripe transition when flushing final pending_data_bl

Fixes: http://tracker.ceph.com/issues/15745
When complete_writing_data() is called, if pending_data_bl is not empty
we still need to handle stripe transition correctly. If pending_data_bl
has more data that we can allow in current stripe, move to the next one.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
(cherry picked from commit b7a7d41839558b1a8786a66e27d9d80e606f61c7)

Conflicts:
	src/rgw/rgw_rados.cc (hammer write_data() takes rgw_obj reference as
	4th argument, hammer throttle_data() takes rgw_obj as 2nd argument)

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 91f7ce1e0e28e5b506db5d36100b37e731e6335e..a6fa97fee910cb4ce3eed0defcaddf756ec75d2b 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -1189,19 +1189,34 @@ int RGWPutObjProcessor_Atomic::complete_writing_data()
     first_chunk.claim(pending_data_bl);
     obj_len = (uint64_t)first_chunk.length();
   }
-  if (pending_data_bl.length()) {
+  while (pending_data_bl.length()) {
     void *handle;
+    uint64_t max_write_size = MIN(max_chunk_size, (uint64_t)next_part_ofs - data_ofs);
+    if (max_write_size > pending_data_bl.length()) {
+      max_write_size = pending_data_bl.length();
+    }
+    bufferlist bl;
+    pending_data_bl.splice(0, max_write_size, &bl);
     rgw_obj obj;
-    int r = write_data(pending_data_bl, data_ofs, &handle, &obj, false);
+    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);
     if (r < 0) {
       ldout(store->ctx(), 0) << "ERROR: throttle_data() returned " << r << dendl;
       return r;
     }
+
+    if (data_ofs >= next_part_ofs) {
+      r = prepare_next_part(data_ofs);
+      if (r < 0) {
+        ldout(store->ctx(), 0) << "ERROR: prepare_next_part() returned " << r << dendl;
+        return r;
+      }
+    }
   }
   int r = complete_parts();
   if (r < 0) {
@@ -1408,6 +1423,10 @@ int RGWRados::get_required_alignment(rgw_bucket& bucket, uint64_t *alignment)
   }
 
   *alignment = ioctx.pool_required_alignment();
+  if (*alignment != 0) {
+    ldout(cct, 20) << "required alignment=" << *alignment << dendl;
+  }
+
   return 0;
 }
 
@@ -1433,6 +1452,8 @@ int RGWRados::get_max_chunk_size(rgw_bucket& bucket, uint64_t *max_chunk_size)
 
   *max_chunk_size = config_chunk_size - (config_chunk_size % alignment);
 
+  ldout(cct, 20) << "max_chunk_size=" << *max_chunk_size << dendl;
+
   return 0;
 }