EOS-26465: Non-batch put kv

Signed-off-by: Dattaprasad Govekar <dattaprasad.govekar@seagate.com>

Author: DPG17

server/s3_motr_kvs_writer.cc

@@ -46,6 +46,11 @@ S3MotrKVSWriter::S3MotrKVSWriter(std::shared_ptr<RequestObject> req,
   } else {
     s3_motr_api = std::make_shared<ConcreteMotrAPI>();
   }
+  on_success_for_parallelkv =
+      std::bind(&S3MotrKVSWriter::parallel_put_kv_successful, this,
+                std::placeholders::_1);
+  on_failure_for_parallelkv = std::bind(
+      &S3MotrKVSWriter::parallel_put_kv_failed, this, std::placeholders::_1);
 }
 
 S3MotrKVSWriter::S3MotrKVSWriter(std::string req_id,
@@ -61,6 +66,11 @@ S3MotrKVSWriter::S3MotrKVSWriter(std::string req_id,
   } else {
     s3_motr_api = std::make_shared<ConcreteMotrAPI>();
   }
+  on_success_for_parallelkv =
+      std::bind(&S3MotrKVSWriter::parallel_put_kv_successful, this,
+                std::placeholders::_1);
+  on_failure_for_parallelkv = std::bind(
+      &S3MotrKVSWriter::parallel_put_kv_failed, this, std::placeholders::_1);
 }
 
 S3MotrKVSWriter::~S3MotrKVSWriter() {
@@ -479,11 +489,88 @@ void S3MotrKVSWriter::delete_indices_failed() {
   s3_log(S3_LOG_DEBUG, "", "%s Exit", __func__);
 }
 
+void S3MotrKVSWriter::parallel_put_kv_successful(unsigned int processed_count) {
+  s3_log(S3_LOG_DEBUG, request_id, "%s Entry\n", __func__);
+  next_key_offset += processed_count;
+  kvop_keys_in_flight -= 1;
+  unsigned int total_keys = kv_list.size();
+  s3_log(S3_LOG_DEBUG, request_id, "Total writer contexts = %zu",
+         parallel_writer_contexts.size());
+#if 0
+  // Early free of writer context
+  if ((next_key_offset - 1) < parallel_writer_contexts.size()) {
+    s3_log(S3_LOG_DEBUG, request_id, "Freeing writer context at index = %d",
+           (next_key_offset - 1));
+    parallel_writer_contexts[next_key_offset - 1].reset(NULL);
+    s3_log(S3_LOG_DEBUG, request_id, "Outstanding writer contexts = %zu",
+           parallel_writer_contexts.size() - next_key_offset);
+  }
+#endif
+
+  if ((next_key_offset + kvop_keys_in_flight) < total_keys &&
+      !atleast_one_failed) {
+    // Still more to launch PUT KV
+    if (kvop_keys_in_flight < max_parallel_kv) {
+      // Current number of parallel KVs is less than max allowed.
+      // Launch next PUT KV operation
+      put_partial_keyval(idx_los[0], kv_list, on_success_for_parallelkv,
+                         on_failure_for_parallelkv,
+                         next_key_offset + kvop_keys_in_flight, 1, false);
+      ++kvop_keys_in_flight;
+      // Save writer context
+      parallel_writer_contexts.push_back(std::move(this->get_writer_context()));
+    }
+  } else {
+    // All PUT KVs done(all success or atleast one fail). Perform next step
+    if (atleast_one_failed && kvop_keys_in_flight == 0) {
+      // One of PUT KV failed previously, and there is no outstanding PUT KV
+      // that we need to wait from the batch of parallel PUT KV.
+      // Perform next step: Failure
+      // TODO: Need to rollback all successfull PUT KV
+      handler_on_failed();
+      return;
+    }
+    if (!atleast_one_failed && kvop_keys_in_flight == 0) {
+      // All keys are done, and all PUT KV in parallel
+      // operation are done. Perform next step: Success
+      handler_on_success();
+      return;
+    }
+    // Need to wait for reminaing PUT KV from the batch of parallel
+    // PUT KV
+  }
+  s3_log(S3_LOG_DEBUG, "", "%s Exit", __func__);
+}
+
+void S3MotrKVSWriter::parallel_put_kv_failed(unsigned int processed_count) {
+  s3_log(S3_LOG_DEBUG, request_id, "%s Entry\n", __func__);
+  next_key_offset += processed_count;
+  atleast_one_failed = true;
+  kvop_keys_in_flight -= 1;
+  s3_log(S3_LOG_DEBUG, request_id, "Total writer contexts = %zu",
+         parallel_writer_contexts.size());
+#if 0
+  // Early free of writer context
+  if ((next_key_offset - 1) < parallel_writer_contexts.size()) {
+    s3_log(S3_LOG_DEBUG, request_id, "Freeing writer context at index = %d",
+           (next_key_offset - 1));
+    parallel_writer_contexts[next_key_offset - 1].reset(NULL);
+    s3_log(S3_LOG_DEBUG, request_id, "Outstanding writer contexts = %zu",
+           parallel_writer_contexts.size() - next_key_offset);
+  }
+#endif
+  if (kvop_keys_in_flight == 0) {
+    // All the keys in parallel batch are done, perform next step
+    handler_on_failed();
+  }
+  s3_log(S3_LOG_DEBUG, "", "%s Exit", __func__);
+}
+
 void S3MotrKVSWriter::put_keyval(
     const struct s3_motr_idx_layout &idx_lo,
     const std::map<std::string, std::string> &kv_list,
     std::function<void(void)> on_success, std::function<void(void)> on_failed,
-    S3MotrKVSWriter::CallbackType callback) {
+    CallbackType callback, bool parallel_mode) {
 
   s3_log(S3_LOG_INFO, stripped_request_id,
          "%s Entry with oid = %" SCNx64 " : %" SCNx64
@@ -499,24 +586,47 @@ void S3MotrKVSWriter::put_keyval(
       s3_log(S3_LOG_ERROR, request_id, "Empty key in PUT KV\n");
     }
   }
-  idx_los.clear();
-  idx_los.push_back(idx_lo);
 
   this->handler_on_success = std::move(on_success);
   this->handler_on_failed = std::move(on_failed);
 
-  if (idx_ctx) {
-    // clean up any old allocations
-    clean_up_contexts();
-  }
-  idx_ctx = create_idx_context(1);
+  if (!this->parallel_run || parallel_mode) {
+    // Initialize once in parallel mode
+    idx_los.clear();
+    idx_los.push_back(idx_lo);
 
+    if (idx_ctx) {
+      // clean up any old allocations
+      clean_up_contexts();
+    }
+    idx_ctx = create_idx_context(1);
+  }
+  if (parallel_mode && kv_list.size() > 1) {
+    // Parallel mode PUT kv, provided list contains at least 2 keys
+    this->kv_list = kv_list;
+    next_key_offset = 0;
+    parallel_writer_contexts.clear();
+    const int parallel_kvs =
+        (kv_list.size() <= max_parallel_kv) ? kv_list.size() : max_parallel_kv;
+    this->parallel_run = parallel_mode;
+    s3_log(S3_LOG_DEBUG, request_id, "Parallel PUT kv, with %d kv in parallel",
+           parallel_kvs);
+    for (int i = 0; i < parallel_kvs; ++i) {
+      // Call below in non-parallel mode
+      put_partial_keyval(idx_lo, kv_list, on_success_for_parallelkv,
+                         on_failure_for_parallelkv, i, 1, false);
+      ++kvop_keys_in_flight;
+      // Save writer context
+      parallel_writer_contexts.push_back(std::move(this->get_writer_context()));
+    }
+    // End of Parallel mode
+    return;
+  }
   writer_context.reset(new S3AsyncMotrKVSWriterContext(
       request, std::bind(&S3MotrKVSWriter::put_keyval_successful, this),
       std::bind(&S3MotrKVSWriter::put_keyval_failed, this), 1, s3_motr_api));
 
   writer_context->init_kvs_write_op_ctx(kv_list.size());
-
   // Ret code can be ignored as its already handled in async case.
   put_keyval_impl(kv_list, true, false, 0, 0, callback);
 }
@@ -526,7 +636,7 @@ void S3MotrKVSWriter::put_partial_keyval(
     const std::map<std::string, std::string> &kv_list,
     std::function<void(unsigned int)> on_success,
     std::function<void(unsigned int)> on_failed, unsigned int offset,
-    unsigned int how_many) {
+    unsigned int how_many, bool parallel_mode) {
   std::map<std::string, std::string>::const_iterator it;
   unsigned int record_count = 0;
   assert(how_many != 0);
@@ -557,17 +667,41 @@ void S3MotrKVSWriter::put_partial_keyval(
       break;
     }
   }
-  idx_los.clear();
-  idx_los.push_back(idx_lo);
-
   this->on_success_for_extends = std::move(on_success);
   this->on_failure_for_extends = std::move(on_failed);
 
-  if (idx_ctx) {
-    // clean up any old allocations
-    clean_up_contexts();
+  if (!this->parallel_run || parallel_mode) {
+    // Initialize once in parallel mode
+    idx_los.clear();
+    idx_los.push_back(idx_lo);
+
+    if (idx_ctx) {
+      // clean up any old allocations
+      clean_up_contexts();
+    }
+    idx_ctx = create_idx_context(1);
+  }
+  if (parallel_mode && how_many > 1) {
+    // Parallel mode PUT partial kv, provided list contains at least 2 keys
+    this->kv_list = kv_list;
+    next_key_offset = 0;
+    parallel_writer_contexts.clear();
+    const int parallel_kvs =
+        (how_many <= max_parallel_kv) ? how_many : max_parallel_kv;
+    this->parallel_run = parallel_mode;
+    s3_log(S3_LOG_DEBUG, request_id, "Parallel PUT kv, with %d kv in parallel",
+           parallel_kvs);
+    for (int i = 0; i < parallel_kvs; ++i) {
+      // Call below in non-parallel mode
+      put_partial_keyval(idx_lo, kv_list, on_success_for_parallelkv,
+                         on_failure_for_parallelkv, i, 1, false);
+      ++kvop_keys_in_flight;
+      // Save writer context
+      parallel_writer_contexts.push_back(std::move(this->get_writer_context()));
+    }
+    // End of Parallel mode
+    return;
   }
-  idx_ctx = create_idx_context(1);
   writer_context.reset(new S3AsyncMotrKVSWriterContext(
       request, std::bind(&S3MotrKVSWriter::put_partial_keyval_successful, this),
       std::bind(&S3MotrKVSWriter::put_partial_keyval_failed, this), 1,
@@ -720,7 +854,7 @@ void S3MotrKVSWriter::put_keyval(const struct s3_motr_idx_layout &idx_lo,
                                  const std::string &key, const std::string &val,
                                  std::function<void(void)> on_success,
                                  std::function<void(void)> on_failed,
-                                 S3MotrKVSWriter::CallbackType callback) {
+                                 CallbackType callback) {
   s3_log(S3_LOG_INFO, stripped_request_id,
          "%s Entry with oid = %" SCNx64 " : %" SCNx64
          " key = %s and value = %s\n",

server/s3_motr_kvs_writer.h

@@ -26,12 +26,14 @@
 #include <gtest/gtest_prod.h>
 #include <functional>
 #include <memory>
+#include <deque>
 
 #include "s3_asyncop_context_base.h"
 #include "s3_motr_context.h"
 #include "s3_motr_wrapper.h"
 #include "s3_log.h"
 #include "s3_request_object.h"
+#define MAX_PARALLEL_KVS 30
 
 class S3SyncMotrKVSWriterContext {
   // Basic Operation context.
@@ -139,7 +141,18 @@ class S3MotrKVSWriter {
   std::unique_ptr<S3AsyncMotrKVSWriterContext> sync_context;
   std::string kvs_key;
   std::string kvs_value;
+
+  // Following are used for parallel KV operation
   unsigned int how_many_being_processed;
+  // Number of Keys/values in KV operation in flight (in parallel)
+  unsigned int kvop_keys_in_flight = 0;
+  bool parallel_run = false;
+  bool atleast_one_failed = false;
+  std::map<std::string, std::string> kv_list;
+  unsigned int next_key_offset = 0;
+  std::deque<std::unique_ptr<S3AsyncMotrKVSWriterContext>>
+      parallel_writer_contexts;
+  unsigned int max_parallel_kv = MAX_PARALLEL_KVS;
 
   std::string request_id;
   std::string stripped_request_id;
@@ -155,6 +168,9 @@ class S3MotrKVSWriter {
   struct s3_motr_idx_context* idx_ctx;
 
   void clean_up_contexts();
+  std::unique_ptr<S3AsyncMotrKVSWriterContext>& get_writer_context() {
+    return writer_context;
+  }
 
   void create_index_successful();
   void create_index_failed();
@@ -179,6 +195,12 @@ class S3MotrKVSWriter {
       S3MotrKVSWriter::CallbackType
           callback = S3MotrKVSWriter::CallbackType::STABLE);
 
+ protected:
+  void parallel_put_kv_successful(unsigned int);
+  void parallel_put_kv_failed(unsigned int);
+  std::function<void(unsigned int)> on_success_for_parallelkv;
+  std::function<void(unsigned int)> on_failure_for_parallelkv;
+
  public:
   S3MotrKVSWriter(std::shared_ptr<RequestObject> req,
                   std::shared_ptr<MotrAPI> motr_api = nullptr);
@@ -223,26 +245,30 @@ class S3MotrKVSWriter {
       std::function<void(void)> on_success,
       std::function<void(void)> on_failed);
 
-  virtual void put_keyval(const struct s3_motr_idx_layout& idx_lo,
-                          const std::map<std::string, std::string>& kv_list,
-                          std::function<void(void)> on_success,
-                          std::function<void(void)> on_failed,
-                          S3MotrKVSWriter::CallbackType
-                              callback = S3MotrKVSWriter::CallbackType::STABLE);
+  // When parallel_mode = true (default), the function will perform several
+  // single
+  // PUT kv for keys in the list in parallel. When all PUT kv are successfull,
+  // on_success would be called, else on_failed.
+  virtual void put_keyval(
+      const struct s3_motr_idx_layout& idx_lo,
+      const std::map<std::string, std::string>& kv_list,
+      std::function<void(void)> on_success, std::function<void(void)> on_failed,
+      CallbackType callback = S3MotrKVSWriter::CallbackType::STABLE,
+      bool parallel_mode = true);
 
   virtual void put_partial_keyval(
       const struct s3_motr_idx_layout& idx_lo,
       const std::map<std::string, std::string>& kv_list,
       std::function<void(unsigned int)> on_success,
       std::function<void(unsigned int)> on_failed, unsigned int offset = 0,
-      unsigned int how_many = 30);
+      unsigned int how_many = 30, bool parallel_mode = true);
 
   // Async save operation.
   virtual void put_keyval(
       const struct s3_motr_idx_layout& idx_lo, const std::string& key,
       const std::string& val, std::function<void(void)> on_success,
       std::function<void(void)> on_failed,
-      enum CallbackType callback = S3MotrKVSWriter::CallbackType::STABLE);
+      CallbackType callback = S3MotrKVSWriter::CallbackType::STABLE);
   // Sync save operation.
   virtual int put_keyval_sync(
       const struct s3_motr_idx_layout& idx_lo,
@@ -307,6 +333,8 @@ class S3MotrKVSWriter {
   FRIEND_TEST(S3PartMetadataTest, CreatePartIndexSuccessfulSaveMetadata);
   FRIEND_TEST(S3NewAccountRegisterNotifyActionTest,
               CreateBucketListIndexSuccessful);
+  FRIEND_TEST(S3MotrKVSWritterTest, ParallelPutKeyValSuccessful);
+  FRIEND_TEST(S3MotrKVSWritterTest, ParallelPutKeyValFail);
 };
 
 #endif

ut/mock_s3_motr_kvs_writer.h

@@ -59,11 +59,11 @@ class MockS3MotrKVSWriter : public S3MotrKVSWriter {
                     const std::string&, std::function<void(void)> on_success,
                     std::function<void(void)> on_failed,
                     CallbackType callback));
-  MOCK_METHOD5(put_keyval,
+  MOCK_METHOD6(put_keyval,
                void(const struct s3_motr_idx_layout&,
                     const std::map<std::string, std::string>& kv_list,
                     std::function<void(void)> on_success,
-                    std::function<void(void)> on_failed,
-                    CallbackType callback));
+                    std::function<void(void)> on_failed, CallbackType callback,
+                    bool parallel));
 };
 #endif

ut/s3_abort_multipart_test.cc

@@ -298,8 +298,9 @@ TEST_F(S3AbortMultipartActionTest, Send200SuccessToS3Client) {
   // expectations for mark_oid_for_deletion()
   EXPECT_CALL(*prob_rec, set_force_delete(true)).Times(1);
   EXPECT_CALL(*prob_rec, to_json()).Times(1);
+  bool parallel = true;
   EXPECT_CALL(*(motr_kvs_writer_factory->mock_motr_kvs_writer),
-              put_keyval(_, _, _, _, _)).Times(1);
+              put_keyval(_, _, _, _, _, parallel)).Times(1);
 
   EXPECT_CALL(*ptr_mock_request, send_response(200, _)).Times(1);
   action_under_test->send_response_to_s3_client();