Merge branch 'eBay:main' into main

Author: Hooper9973

src/lib/homestore_backend/CMakeLists.txt

@@ -127,18 +127,12 @@ add_test(NAME HomestoreResyncTestWithLeaderRestart
 # GC tests
 add_test(NAME FetchDataWithOriginatorGC
         COMMAND homestore_test_dynamic -csv error --executor immediate --config_path ./
-        --override_config hs_backend_config.enable_gc=true 
-        # remove this until gc supports baseline resync
-        --override_config homestore_config.consensus.snapshot_freq_distance:0
         --gtest_filter=HomeObjectFixture.FetchDataWithOriginatorGC)
 
 add_executable(homestore_test_gc)
 target_sources(homestore_test_gc PRIVATE $<TARGET_OBJECTS:homestore_tests_gc>)
 target_link_libraries(homestore_test_gc PUBLIC homeobject_homestore ${COMMON_TEST_DEPS})
 add_test(NAME HomestoreTestGC COMMAND homestore_test_gc -csv error --executor immediate --config_path ./
-        --override_config hs_backend_config.enable_gc=true 
         --override_config hs_backend_config.gc_garbage_rate_threshold=0 
-        --override_config hs_backend_config.gc_scan_interval_sec=5
-        # remove this until gc supports baseline resync
-        --override_config homestore_config.consensus.snapshot_freq_distance:0)
+        --override_config hs_backend_config.gc_scan_interval_sec=5)
 

src/lib/homestore_backend/gc_manager.cpp

@@ -132,10 +132,11 @@ void GCManager::stop() {
 
 folly::SemiFuture< bool > GCManager::submit_gc_task(task_priority priority, chunk_id_t chunk_id) {
     if (!is_started()) return folly::makeFuture< bool >(false);
+
     auto pdev_id = m_chunk_selector->get_extend_vchunk(chunk_id)->get_pdev_id();
     auto it = m_pdev_gc_actors.find(pdev_id);
     if (it == m_pdev_gc_actors.end()) {
-        LOGINFO("pdev gc actor not found for pdev_id: {}", pdev_id);
+        LOGINFO("pdev gc actor not found for pdev_id={}, chunk={}", pdev_id, chunk_id);
         return folly::makeFuture< bool >(false);
     }
     auto& actor = it->second;
@@ -178,22 +179,14 @@ bool GCManager::is_eligible_for_gc(chunk_id_t chunk_id) {
         return false;
     }
 
-    // it does not belong to any pg, so we don't need to gc it.
-    if (!chunk->m_pg_id.has_value()) {
-        LOGDEBUG("chunk_id={} belongs to no pg, not eligible for gc", chunk_id)
-        return false;
-    }
-
     const auto total_blk_num = chunk->get_total_blks();
 
     const auto gc_garbage_rate_threshold = HS_BACKEND_DYNAMIC_CONFIG(gc_garbage_rate_threshold);
 
     bool should_gc = 100 * defrag_blk_num >= total_blk_num * gc_garbage_rate_threshold;
 
-    LOGDEBUG("gc scan chunk_id={}, belongs to pg {}, use_blks={}, available_blks={}, total_blks={}, defrag_blks={}, "
-             "should_gc={}",
-             chunk_id, chunk->m_pg_id.value(), chunk->get_used_blks(), chunk->available_blks(), total_blk_num,
-             defrag_blk_num, should_gc);
+    LOGDEBUG("gc scan chunk_id={}, use_blks={}, available_blks={}, total_blks={}, defrag_blks={}, should_gc={}",
+             chunk_id, chunk->get_used_blks(), chunk->available_blks(), total_blk_num, defrag_blk_num, should_gc);
 
     return should_gc;
 }
@@ -292,10 +285,27 @@ void GCManager::pdev_gc_actor::add_reserved_chunk(
 }
 
 folly::SemiFuture< bool > GCManager::pdev_gc_actor::add_gc_task(uint8_t priority, chunk_id_t move_from_chunk) {
+    auto EXvchunk = m_chunk_selector->get_extend_vchunk(move_from_chunk);
+    // it does not belong to any pg, so we don't need to gc it.
+    if (!EXvchunk->m_pg_id.has_value()) {
+        LOGDEBUG("chunk_id={} belongs to no pg, not eligible for gc", move_from_chunk)
+        return folly::makeSemiFuture< bool >(false);
+    }
+
+    const auto pg_id = EXvchunk->m_pg_id.value();
+
+    m_hs_home_object->gc_manager()->incr_pg_pending_gc_task(pg_id);
+
+    if (!m_hs_home_object->can_chunks_in_pg_be_gc(pg_id)) {
+        LOGDEBUG("chunk_id={} belongs to pg {}, which is not eligible for gc at this moment!", move_from_chunk, pg_id)
+        m_hs_home_object->gc_manager()->decr_pg_pending_gc_task(pg_id);
+        return folly::makeSemiFuture< bool >(false);
+    }
+
     if (m_chunk_selector->try_mark_chunk_to_gc_state(move_from_chunk,
                                                      priority == static_cast< uint8_t >(task_priority::emergent))) {
         auto [promise, future] = folly::makePromiseContract< bool >();
-        auto gc_task_id = GCManager::_gc_task_id.fetch_add(1);
+        const auto gc_task_id = GCManager::_gc_task_id.fetch_add(1);
 
         if (sisl_unlikely(priority == static_cast< uint8_t >(task_priority::emergent))) {
             m_egc_executor->add([this, gc_task_id, priority, move_from_chunk, promise = std::move(promise)]() mutable {
@@ -313,9 +323,49 @@ folly::SemiFuture< bool > GCManager::pdev_gc_actor::add_gc_task(uint8_t priority
     }
 
     LOGWARN("fail to submit gc task for chunk_id={}, priority={}", move_from_chunk, priority);
+    m_hs_home_object->gc_manager()->decr_pg_pending_gc_task(pg_id);
     return folly::makeSemiFuture< bool >(false);
 }
 
+void GCManager::drain_pg_pending_gc_task(const pg_id_t pg_id) {
+    while (true) {
+        uint64_t pending_gc_task_num{0};
+        {
+            std::unique_lock lock(m_pending_gc_task_mtx);
+            pending_gc_task_num = m_pending_gc_task_num_per_pg.try_emplace(pg_id, 0).first->second.load();
+        }
+
+        if (pending_gc_task_num) {
+            LOGDEBUG("{} pending gc tasks to be completed for pg={}, wait for 2 seconds!", pending_gc_task_num, pg_id);
+        } else {
+            break;
+        }
+        // wait until all the pending gc tasks for this pg are completed
+        std::this_thread::sleep_for(std::chrono::seconds(2));
+    }
+
+    LOGDEBUG("all pending gc tasks for pg_id={} are completed", pg_id);
+}
+
+void GCManager::decr_pg_pending_gc_task(const pg_id_t pg_id) {
+    std::unique_lock lock(m_pending_gc_task_mtx);
+    auto& pending_gc_task_num = m_pending_gc_task_num_per_pg.try_emplace(pg_id, 0).first->second;
+    if (pending_gc_task_num.load()) {
+        // TODO::avoid overflow here.
+        --pending_gc_task_num;
+        LOGDEBUG("decrease pending gc task num for pg_id={}, now it is {}", pg_id, pending_gc_task_num.load());
+        return;
+    }
+    LOGDEBUG("pending gc task num for pg_id={} is already 0, no need to decrease it", pg_id);
+}
+
+void GCManager::incr_pg_pending_gc_task(const pg_id_t pg_id) {
+    std::unique_lock lock(m_pending_gc_task_mtx);
+    auto& pending_gc_task_num = m_pending_gc_task_num_per_pg.try_emplace(pg_id, 0).first->second;
+    ++pending_gc_task_num;
+    LOGDEBUG("increase pending gc task num for pg_id={}, now it is {}", pg_id, pending_gc_task_num.load());
+}
+
 // this method is expected to be called sequentially when replaying metablk, so we don't need to worry about the
 // concurrency issue.
 void GCManager::pdev_gc_actor::handle_recovered_gc_task(
@@ -448,10 +498,11 @@ bool GCManager::pdev_gc_actor::replace_blob_index(
                 }
 
                 if (existing_pbas.chunk_num() != move_from_chunk) {
-                    LOGERROR("gc task_id={}, existing pbas chunk={} should be equal to move_from_chunk={}, pg_id={}, "
-                             "shard_id={}, blob_id={}, move_to_chunk={} , existing_pbas={}, new_pbas={}",
-                             task_id, existing_pbas.chunk_num(), move_from_chunk, pg_id, shard, blob, move_to_chunk,
-                             existing_pbas.to_string(), new_pbas.to_string());
+                    LOGWARN("gc task_id={}, existing pbas chunk={} should be equal to move_from_chunk={}, pg_id={}, "
+                            "shard_id={}, blob_id={}, move_to_chunk={} , existing_pbas={}, new_pbas={}, this case "
+                            "might happen when crash recovery",
+                            task_id, existing_pbas.chunk_num(), move_from_chunk, pg_id, shard, blob, move_to_chunk,
+                            existing_pbas.to_string(), new_pbas.to_string());
                     return homestore::put_filter_decision::keep;
                 }
 
@@ -676,8 +727,9 @@ bool GCManager::pdev_gc_actor::copy_valid_data(chunk_id_t move_from_chunk, chunk
                     RELEASE_ASSERT(header_sgs.iovs.size() == 1, "header_sgs.iovs.size() should be 1, but not!");
                     iomanager.iobuf_free(reinterpret_cast< uint8_t* >(header_sgs.iovs[0].iov_base));
                     if (err) {
-                        LOGERROR("gc task_id={}, Failed to write shard header for move_to_chunk={} shard_id={}, err={}",
-                                 task_id, move_to_chunk, shard_id, err.value());
+                        LOGERROR("gc task_id={}, Failed to write shard header for move_to_chunk={} shard_id={}, "
+                                 "err={}, err_category={}, err_message={}",
+                                 task_id, move_to_chunk, shard_id, err.value(), err.category().name(), err.message());
                         return folly::makeFuture< bool >(false);
                     }
 
@@ -713,8 +765,9 @@ bool GCManager::pdev_gc_actor::copy_valid_data(chunk_id_t move_from_chunk, chunk
                                     if (err) {
                                         LOGERROR(
                                             "gc task_id={}, Failed to read blob from move_from_chunk={}, shard_id={}, "
-                                            "blob_id={}: err={}",
-                                            task_id, move_from_chunk, k.key().shard, k.key().blob, err.value());
+                                            "blob_id={}, err={}, err_category={}, err_message={}",
+                                            task_id, move_from_chunk, k.key().shard, k.key().blob, err.value(),
+                                            err.category().name(), err.message());
                                         iomanager.iobuf_free(reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base));
                                         return folly::makeFuture< bool >(false);
                                     }
@@ -747,10 +800,11 @@ bool GCManager::pdev_gc_actor::copy_valid_data(chunk_id_t move_from_chunk, chunk
                                             iomanager.iobuf_free(
                                                 reinterpret_cast< uint8_t* >(data_sgs.iovs[0].iov_base));
                                             if (err) {
-                                                LOGERROR("gc task_id={}, Failed to write blob to move_to_chunk={}, "
-                                                         "shard_id={}, blob_id={}, err={}",
-                                                         task_id, move_to_chunk, k.key().shard, k.key().blob,
-                                                         err.value());
+                                                LOGERROR(
+                                                    "gc task_id={}, Failed to write blob to move_to_chunk={}, "
+                                                    "shard_id={}, blob_id={}, err={}, err_category={}, err_message={}",
+                                                    task_id, move_to_chunk, k.key().shard, k.key().blob, err.value(),
+                                                    err.category().name(), err.message());
                                                 return false;
                                             }
 
@@ -811,10 +865,10 @@ bool GCManager::pdev_gc_actor::copy_valid_data(chunk_id_t move_from_chunk, chunk
                             iomanager.iobuf_free(reinterpret_cast< uint8_t* >(footer_sgs.iovs[0].iov_base));
 
                             if (err) {
-                                LOGERROR(
-                                    "gc task_id={}, Failed to write shard footer for move_to_chunk={} shard_id={}, "
-                                    "err={}",
-                                    task_id, move_to_chunk, shard_id, err.value());
+                                LOGERROR("gc task_id={}, Failed to write shard footer for move_to_chunk={} "
+                                         "shard_id={}, err={}, error_category={}, error_message={}",
+                                         task_id, move_to_chunk, shard_id, err.value(), err.category().name(),
+                                         err.message());
                                 return false;
                             }
                             return true;
@@ -1043,6 +1097,7 @@ void GCManager::pdev_gc_actor::process_gc_task(chunk_id_t move_from_chunk, uint8
 
     task.setValue(true);
     m_reserved_chunk_queue.blockingWrite(move_from_chunk);
+    m_hs_home_object->gc_manager()->decr_pg_pending_gc_task(pg_id);
     LOGINFO("gc task_id={} for move_from_chunk={} to move_to_chunk={} with priority={} is completed", task_id,
             move_from_chunk, move_to_chunk, priority);
 }

src/lib/homestore_backend/gc_manager.hpp

@@ -199,6 +199,9 @@ class GCManager {
 
     void scan_chunks_for_gc();
     void handle_all_recovered_gc_tasks();
+    void drain_pg_pending_gc_task(const pg_id_t pg_id);
+    void decr_pg_pending_gc_task(const pg_id_t pg_id);
+    void incr_pg_pending_gc_task(const pg_id_t pg_id);
 
 private:
     void on_gc_task_meta_blk_found(sisl::byte_view const& buf, void* meta_cookie);
@@ -212,6 +215,8 @@ class GCManager {
     iomgr::timer_handle_t m_gc_timer_hdl{iomgr::null_timer_handle};
     HSHomeObject* m_hs_home_object{nullptr};
     std::list< homestore::superblk< GCManager::gc_task_superblk > > m_recovered_gc_tasks;
+    std::unordered_map< pg_id_t, atomic_uint64_t > m_pending_gc_task_num_per_pg;
+    std::mutex m_pending_gc_task_mtx;
 };
 
 } // namespace homeobject

src/lib/homestore_backend/heap_chunk_selector.cpp

@@ -139,7 +139,7 @@ csharedChunk HeapChunkSelector::select_specific_chunk(const pg_id_t pg_id, const
         }
 
         // if the chunk is not available, probably being gc. we wait for a while and retry
-        std::this_thread::sleep_for(std::chrono::seconds(5));
+        std::this_thread::sleep_for(std::chrono::seconds(1));
     }
 
     LOGDEBUGMOD(homeobject, "chunk={} is selected for v_chunk_id={}, pg={}", chunk->get_chunk_id(), v_chunk_id, pg_id);

src/lib/homestore_backend/hs_backend_config.fbs

@@ -21,7 +21,7 @@ table HSBackendSettings {
 
     //Enable GC
     //TODO: make this hotswap after gc is well tested
-    enable_gc: bool = false;
+    enable_gc: bool = true;
 
     //Total reserved chunk num (dedicated for gc/egc) per pdev 
     reserved_chunk_num_per_pdev: uint8 = 6;
@@ -33,7 +33,7 @@ table HSBackendSettings {
     gc_scan_interval_sec: uint64 = 60;
 
     //GC garbage rate threshold, upon which a chunk will be selected for gc
-    gc_garbage_rate_threshold: uint8 = 80;
+    gc_garbage_rate_threshold: uint8 = 50;
 
     //max read/write block count per second, which is used by ratelimiter to limit the io resource taken by gc
     max_read_write_block_count_per_second: uint16 = 7680;

src/lib/homestore_backend/hs_homeobject.hpp

@@ -856,6 +856,14 @@ class HSHomeObject : public HomeObjectImpl {
      */
     bool release_chunk_based_on_create_shard_message(sisl::blob const& header);
 
+    /**
+     * @brief check whether the chunks in a given pg can be gc.
+     *
+     * @param pg_id The ID of the PG whose shards are to be destroyed.
+     * @return True if the chunks in the PG can be garbage collected, false otherwise.
+     */
+    bool can_chunks_in_pg_be_gc(pg_id_t pg_id) const;
+
     bool pg_exists(pg_id_t pg_id) const;
 
     uint32_t get_reserved_blks() const { return _hs_reserved_blks; }

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -450,6 +450,11 @@ bool HSHomeObject::pg_destroy(pg_id_t pg_id, bool need_to_pause_pg_state_machine
     }
     LOGI("Destroying pg={}", pg_id);
     mark_pg_destroyed(pg_id);
+
+    // we have the assumption that after pg is marked as destroyed, it will not be marked as alive again.
+    // TODO:: if this assumption is broken, we need to handle it.
+    gc_mgr_->drain_pg_pending_gc_task(pg_id);
+    
     destroy_shards(pg_id);
     destroy_hs_resources(pg_id);
     destroy_pg_index_table(pg_id);
@@ -505,6 +510,17 @@ void HSHomeObject::mark_pg_destroyed(pg_id_t pg_id) {
     LOGD("pg={} is marked as destroyed", pg_id);
 }
 
+bool HSHomeObject::can_chunks_in_pg_be_gc(pg_id_t pg_id) const {
+    auto lg = std::scoped_lock(_pg_lock);
+    auto hs_pg = const_cast< HS_PG* >(_get_hs_pg_unlocked(pg_id));
+    if (hs_pg == nullptr) {
+        LOGW("unknown pg={}", pg_id);
+        return false;
+    }
+
+    return hs_pg->pg_sb_->state == PGState::ALIVE;
+}
+
 void HSHomeObject::destroy_hs_resources(pg_id_t pg_id) { chunk_selector_->reset_pg_chunks(pg_id); }
 
 void HSHomeObject::destroy_pg_index_table(pg_id_t pg_id) {

src/lib/homestore_backend/replication_state_machine.cpp

@@ -683,7 +683,6 @@ folly::Future< std::error_code > ReplicationStateMachine::on_fetch_data(const in
         LOGD("fetch data with blob_id={}, shard=0x{:x}", blob_id, shard_id);
         // we first try to read data according to the local_blk_id to see if it matches the blob_id
         return std::move(homestore::data_service().async_read(local_blk_id, given_buffer, total_size))
-            .via(folly::getGlobalIOExecutor())
             .thenValue([this, lsn, blob_id, shard_id, given_buffer, total_size](auto&& err) {
                 // io error
                 if (err) {