Move snapshot data write to async mode.

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "2.3.21"
+    version = "2.3.22"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"

src/lib/homestore_backend/hs_homeobject.hpp

@@ -441,14 +441,20 @@ class HSHomeObject : public HomeObjectImpl {
     inline const static homestore::MultiBlkId tombstone_pbas{0, 0, 0};
 
     struct PGBlobIterator {
+        struct blob_read_result {
+            blob_id_t blob_id_;
+            sisl::io_blob_safe blob_;
+            ResyncBlobState state_;
+            blob_read_result(blob_id_t blob_id, sisl::io_blob_safe&& blob, ResyncBlobState state) :
+                    blob_id_(blob_id), blob_(std::move(blob)), state_(state) {}
+        };
         PGBlobIterator(HSHomeObject& home_obj, homestore::group_id_t group_id, uint64_t upto_lsn = 0);
         PG* get_pg_metadata();
         bool update_cursor(objId id);
         void reset_cursor();
         objId expected_next_obj_id();
         bool generate_shard_blob_list();
-        BlobManager::AsyncResult< sisl::io_blob_safe > load_blob_data(const BlobInfo& blob_info,
-                                                                      ResyncBlobState& state);
+        BlobManager::AsyncResult< blob_read_result > load_blob_data(const BlobInfo& blob_info);
         bool create_pg_snapshot_data(sisl::io_blob_safe& meta_blob);
         bool create_shard_snapshot_data(sisl::io_blob_safe& meta_blob);
         bool create_blobs_snapshot_data(sisl::io_blob_safe& data_blob);

src/lib/homestore_backend/pg_blob_iterator.cpp

@@ -90,8 +90,8 @@ void HSHomeObject::PGBlobIterator::reset_cursor() {
     cur_obj_id_ = {0, 0};
     cur_shard_idx_ = -1;
     std::vector< BlobInfo > cur_blob_list_{0};
-    cur_start_blob_idx_=0;
-    cur_batch_blob_count_=0;
+    cur_start_blob_idx_ = 0;
+    cur_batch_blob_count_ = 0;
     cur_batch_start_time_ = Clock::time_point{};
 }
 
@@ -187,8 +187,8 @@ bool HSHomeObject::PGBlobIterator::create_shard_snapshot_data(sisl::io_blob_safe
     return true;
 }
 
-BlobManager::AsyncResult< sisl::io_blob_safe > HSHomeObject::PGBlobIterator::load_blob_data(const BlobInfo& blob_info,
-                                                                                            ResyncBlobState& state) {
+typedef HSHomeObject::PGBlobIterator::blob_read_result blob_read_result;
+BlobManager::AsyncResult< blob_read_result > HSHomeObject::PGBlobIterator::load_blob_data(const BlobInfo& blob_info) {
     auto shard_id = blob_info.shard_id;
     auto blob_id = blob_info.blob_id;
     auto blkid = blob_info.pbas;
@@ -202,8 +202,8 @@ BlobManager::AsyncResult< sisl::io_blob_safe > HSHomeObject::PGBlobIterator::loa
     LOGD("Blob get request: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, blkid={}", shard_id,
          (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id, blkid.to_string());
     return repl_dev_->async_read(blkid, sgs, total_size)
-        .thenValue([this, blob_id, shard_id, &state, read_buf = std::move(read_buf)](
-                       auto&& result) mutable -> BlobManager::AsyncResult< sisl::io_blob_safe > {
+        .thenValue([this, blob_id, shard_id, read_buf = std::move(read_buf)](auto&& result) mutable
+                       -> BlobManager::AsyncResult< HSHomeObject::PGBlobIterator::blob_read_result > {
             if (result) {
                 LOGE("Failed to get blob, shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, err={}", shard_id,
                      (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id,
@@ -217,17 +217,15 @@ BlobManager::AsyncResult< sisl::io_blob_safe > HSHomeObject::PGBlobIterator::loa
                 LOGE("Invalid header found, shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, [header={}]", shard_id,
                      (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id,
                      header->to_string());
-                state = ResyncBlobState::CORRUPTED;
-                return std::move(read_buf);
+                return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::CORRUPTED);
             }
 
             if (header->shard_id != shard_id) {
                 // The metrics for corrupted blob is handled on the follower side.
                 LOGE("Invalid shard_id in header, shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, [header={}]",
                      shard_id, (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id,
                      header->to_string());
-                state = ResyncBlobState::CORRUPTED;
-                return std::move(read_buf);
+                return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::CORRUPTED);
             }
 
             std::string user_key = header->user_key_size
@@ -244,12 +242,12 @@ BlobManager::AsyncResult< sisl::io_blob_safe > HSHomeObject::PGBlobIterator::loa
                      "[{}] [computed={:np}]",
                      shard_id, (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id,
                      header->to_string(), spdlog::to_hex(computed_hash, computed_hash + BlobHeader::blob_max_hash_len));
-                state = ResyncBlobState::CORRUPTED;
+                return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::CORRUPTED);
             }
 
             LOGD("Blob get success: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}", shard_id,
                  (shard_id >> homeobject::shard_width), (shard_id & homeobject::shard_mask), blob_id);
-            return std::move(read_buf);
+            return blob_read_result(blob_id, std::move(read_buf), ResyncBlobState::NORMAL);
         });
 }
 
@@ -261,24 +259,27 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
     uint64_t total_bytes = 0;
     auto idx = cur_start_blob_idx_;
 
+    std::vector< BlobManager::AsyncResult< blob_read_result > > futs;
+    auto total_blobs = 0;
+    auto skipped_blobs = 0;
     while (total_bytes < max_batch_size_ && idx < cur_blob_list_.size()) {
         auto info = cur_blob_list_[idx++];
-        ResyncBlobState state = ResyncBlobState::NORMAL;
+        total_blobs++;
         // handle deleted object
         if (info.pbas == tombstone_pbas) {
             LOGT("Blob is deleted: shardID=0x{:x}, pg={}, shard=0x{:x}, blob_id={}, blkid={}", info.shard_id,
                  (info.shard_id >> homeobject::shard_width), (info.shard_id & homeobject::shard_mask), info.blob_id,
                  info.pbas.to_string());
             // ignore
+            skipped_blobs++;
             continue;
         }
 
-        auto blob_start = Clock::now();
-
 #ifdef _PRERELEASE
         if (iomgr_flip::instance()->test_flip("pg_blob_iterator_load_blob_data_error")) {
             LOGW("Simulating loading blob data error");
-            return false;
+            futs.emplace_back(folly::makeUnexpected(BlobError(BlobErrorCode::READ_FAILED)));
+            continue;
         }
         auto delay = iomgr_flip::instance()->get_test_flip< long >("simulate_read_snapshot_load_blob_delay",
                                                                    static_cast< long >(info.blob_id));
@@ -288,32 +289,51 @@ bool HSHomeObject::PGBlobIterator::create_blobs_snapshot_data(sisl::io_blob_safe
             std::this_thread::sleep_for(std::chrono::milliseconds(delay.get()));
         }
 #endif
-        sisl::io_blob_safe blob;
-        uint8_t retries = HS_BACKEND_DYNAMIC_CONFIG(snapshot_blob_load_retry);
-        for (int i = 0; i < retries; i++) {
-            auto result = load_blob_data(info, state).get();
-            if (result.hasError() && result.error().code == BlobErrorCode::READ_FAILED) {
-                LOGW("Failed to retrieve blob for shardID=0x{:x}, pg={}, shard=0x{:x} blob={} pbas={}, err={}, "
-                     "attempt={}",
-                     info.shard_id, (info.shard_id >> homeobject::shard_width),
-                     (info.shard_id & homeobject::shard_mask), info.blob_id, info.pbas.to_string(), result.error(), i);
-            } else {
-                blob = std::move(result.value());
-                break;
-            }
-        }
-        if (blob.size() == 0) {
-            LOGE("Failed to retrieve blob for shardID=0x{:x}, pg={}, shard=0x{:x} blob={} pbas={}", info.shard_id,
-                 (info.shard_id >> homeobject::shard_width), (info.shard_id & homeobject::shard_mask), info.blob_id,
-                 info.pbas.to_string());
-            COUNTER_INCREMENT(*metrics_, snp_dnr_error_count, 1);
-            return false;
-        }
+        auto blob_start = Clock::now();
 
-        HISTOGRAM_OBSERVE(*metrics_, snp_dnr_blob_process_latency, get_elapsed_time_us(blob_start));
-        std::vector< uint8_t > data(blob.cbytes(), blob.cbytes() + blob.size());
-        blob_entries.push_back(CreateResyncBlobDataDirect(builder_, info.blob_id, (uint8_t)state, &data));
-        total_bytes += blob.size();
+        LOGT("submitting io for blob {}", info.blob_id);
+        // Fixme: Re-enable retries uint8_t retries = HS_BACKEND_DYNAMIC_CONFIG(snapshot_blob_load_retry);
+        futs.emplace_back(
+            std::move(load_blob_data(info))
+                .via(folly::getKeepAliveToken(folly::InlineExecutor::instance()))
+                .thenValue(
+                    [&, info, blob_start](auto&& result) mutable -> BlobManager::AsyncResult< blob_read_result > {
+                        if (result.hasError() && result.error().code == BlobErrorCode::READ_FAILED) {
+                            LOGE("Failed to retrieve blob for shardID=0x{:x}, pg={}, shard=0x{:x} blob={} pbas={}",
+                                 info.shard_id, (info.shard_id >> homeobject::shard_width),
+                                 (info.shard_id & homeobject::shard_mask), info.blob_id, info.pbas.to_string());
+                            COUNTER_INCREMENT(*metrics_, snp_dnr_error_count, 1);
+                        } else {
+                            LOGT("retrieved blob,  blob={} pbas={}", info.blob_id, info.pbas.to_string());
+                            HISTOGRAM_OBSERVE(*metrics_, snp_dnr_blob_process_latency, get_elapsed_time_us(blob_start));
+                        }
+                        return result;
+                    }));
+
+        auto const expect_blob_size = info.pbas.blk_count() * repl_dev_->get_blk_size();
+        total_bytes += expect_blob_size;
+    }
+    // collect futs and add to flatbuffer.
+    bool hit_error = false;
+    if (skipped_blobs + (long)futs.size() != total_blobs) {
+        LOGE("total {} blobs, skipped {}, expect {} but only has {} futs", total_blobs, skipped_blobs,
+             total_blobs - skipped_blobs, futs.size());
+        hit_error = true;
+    }
+    for (auto it = futs.begin(); it != futs.end(); it++) {
+        auto res = std::move(*it).get();
+        // Although we fail the whole batch if single blob failed, however we still need to wait till all in-flight ops
+        // finished, because the iterator might be released by nuraft after returning an error, however the "thenValue"
+        // still access the metrics ptr.
+        if (res.hasError()) { hit_error = true; }
+        if (!hit_error) {
+            std::vector< uint8_t > data(res->blob_.cbytes(), res->blob_.cbytes() + res->blob_.size());
+            blob_entries.push_back(CreateResyncBlobDataDirect(builder_, res->blob_id_, (uint8_t)res->state_, &data));
+        }
+    }
+    if (hit_error) {
+        builder_.Clear();
+        return false;
     }
     // should include the deleted blobs
     cur_batch_blob_count_ = idx - cur_start_blob_idx_;
@@ -347,4 +367,4 @@ void HSHomeObject::PGBlobIterator::pack_resync_message(sisl::io_blob_safe& dest_
     builder_.Clear();
 }
 
-} // namespace homeobject
+} // namespace homeobject

src/lib/homestore_backend/replication_state_machine.cpp

@@ -486,7 +486,7 @@ void ReplicationStateMachine::write_snapshot_obj(std::shared_ptr< homestore::sna
         }
         snp_obj->offset =
             objId(HSHomeObject::get_sequence_num_from_shard_id(m_snp_rcv_handler->get_next_shard()), 0).value;
-        LOGD("Write snapshot, processed PG data pg={} {}", pg_data->pg_id(), log_suffix);
+        LOGI("Write snapshot, processed PG data pg={} {}", pg_data->pg_id(), log_suffix);
         return;
     }