enhance gc

conanfile.py

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

src/lib/homestore_backend/gc_manager.hpp

@@ -42,6 +42,7 @@ class GCManager {
     inline static auto const _gc_actor_meta_name = std::string("GCActor");
     inline static auto const _gc_task_meta_name = std::string("GCTask");
     inline static auto const _gc_reserved_chunk_meta_name = std::string("GCReservedChunk");
+    inline static atomic_uint64_t _gc_task_id{1}; // 0 is used for crash recovery
 
 #pragma pack(1)
     struct gc_actor_superblk {
@@ -110,19 +111,22 @@ class GCManager {
         void stop();
 
     private:
-        void process_gc_task(chunk_id_t move_from_chunk, uint8_t priority, folly::Promise< bool > task);
+        void process_gc_task(chunk_id_t move_from_chunk, uint8_t priority, folly::Promise< bool > task,
+                             const uint64_t task_id);
 
         // this should be called only after gc_task meta blk is persisted. it will update the pg index table according
         // to the gc index table. return the move_to_chunk to chunkselector and put move_from_chunk to reserved chunk
         // queue.
         bool
         replace_blob_index(chunk_id_t move_from_chunk, chunk_id_t move_to_chunk,
-                           const std::vector< std::pair< BlobRouteByChunkKey, BlobRouteValue > >& valid_blob_indexes);
+                           const std::vector< std::pair< BlobRouteByChunkKey, BlobRouteValue > >& valid_blob_indexes,
+                           const uint64_t task_id);
 
         // copy all the valid data from the move_from_chunk to move_to_chunk. valid data means those blobs that are not
         // tombstone in the pg index table
         // return true if the data copy is successful, false otherwise.
-        bool copy_valid_data(chunk_id_t move_from_chunk, chunk_id_t move_to_chunk, bool is_emergent = false);
+        bool copy_valid_data(chunk_id_t move_from_chunk, chunk_id_t move_to_chunk, const uint64_t task_id,
+                             bool is_emergent = false);
 
         // before we select a reserved chunk and start gc, we need:
         //  1 clear all the entries of this chunk in the gc index table
@@ -136,7 +140,8 @@ class GCManager {
         // case and recvoery case
         bool process_after_gc_metablk_persisted(
             homestore::superblk< GCManager::gc_task_superblk >& gc_task_sb,
-            const std::vector< std::pair< BlobRouteByChunkKey, BlobRouteValue > >& valid_blob_indexes);
+            const std::vector< std::pair< BlobRouteByChunkKey, BlobRouteValue > >& valid_blob_indexes,
+            const uint64_t task_id);
 
     private:
         // utils

src/lib/homestore_backend/heap_chunk_selector.cpp

@@ -73,8 +73,14 @@ bool HeapChunkSelector::try_mark_chunk_to_gc_state(const chunk_num_t chunk_id, b
     }
 
     auto& chunk_state = chunk_it->second->m_state;
+
+    if (chunk_state == ChunkState::GC) {
+        LOGWARNMOD(homeobject, "gc: chunk is already in gc state, chunk_id={}", chunk_id);
+        return false; // already in gc state, no need to change
+    }
+
     if (chunk_state == ChunkState::INUSE && !force) {
-        LOGWARNMOD(homeobject, "Chunk is inuse, chunk_id={}", chunk_id);
+        LOGWARNMOD(homeobject, "gc: chunk is inuse, chunk_id={}", chunk_id);
         return false;
     }
 
@@ -148,6 +154,7 @@ bool HeapChunkSelector::reset_pg_chunks(pg_id_t pg_id) {
         auto pg_chunk_collection = pg_it->second;
         std::scoped_lock lock(pg_chunk_collection->mtx);
         for (auto& chunk : pg_chunk_collection->m_pg_chunks) {
+            LOGDEBUGMOD(homeobject, "reset chunk={} in pg={} for destruction", chunk->get_chunk_id(), pg_id);
             chunk->reset();
         }
     }
@@ -271,7 +278,7 @@ std::optional< uint32_t > HeapChunkSelector::select_chunks_for_pg(pg_id_t pg_id,
 
 void HeapChunkSelector::switch_chunks_for_pg(const pg_id_t pg_id, const chunk_num_t old_chunk_id,
                                              const chunk_num_t new_chunk_id) {
-    LOGDEBUGMOD(homeobject, "switch chunks for pg_id={}, old_chunk={}, new_chunk={}", pg_id, old_chunk_id,
+    LOGDEBUGMOD(homeobject, "gc: switch chunks for pg_id={}, old_chunk={}, new_chunk={}", pg_id, old_chunk_id,
                 new_chunk_id);
 
     auto EXVchunk_old = get_extend_vchunk(old_chunk_id);
@@ -295,27 +302,32 @@ void HeapChunkSelector::switch_chunks_for_pg(const pg_id_t pg_id, const chunk_nu
     std::unique_lock lk(pg_chunk_collection->mtx);
     auto& pg_chunks = pg_chunk_collection->m_pg_chunks;
 
+    // LOGDEBUGMOD(homeobject, "gc: before switch chunks for pg_id={}, pg_chunks={}", pg_chunks);
+
     if (sisl_unlikely(pg_chunks[v_chunk_id]->get_chunk_id() == new_chunk_id)) {
         // this might happens when crash recovery. the crash happens after pg metablk is updated but before gc task
         // metablk is destroyed.
         LOGDEBUGMOD(
             homeobject,
-            "the pchunk_id for vchunk_id={} in chunkselector for pg_id={} is already {},  skip switching chunks!",
+            "gc: the pchunk_id for vchunk_id={} in chunkselector for pg_id={} is already {},  skip switching chunks!",
             v_chunk_id, pg_id, new_chunk_id);
 
         return;
     } else {
         RELEASE_ASSERT(pg_chunks[v_chunk_id]->get_chunk_id() == old_chunk_id,
-                       "vchunk={} for pg={} in chunkselector should have a pchunk={} , but have a pchunk={}",
+                       "gc: vchunk={} for pg={} in chunkselector should have a pchunk={} , but have a pchunk={}",
                        v_chunk_id, pg_id, old_chunk_id, pg_chunks[v_chunk_id]->get_chunk_id());
 
         pg_chunks[v_chunk_id] = EXVchunk_new;
 
-        LOGDEBUGMOD(homeobject,
-                    "vchunk={} in pg_chunk_collection for pg_id={} has been update from pchunk_id={} to pchunk_id={}",
-                    v_chunk_id, pg_id, old_chunk_id, new_chunk_id);
+        LOGDEBUGMOD(
+            homeobject,
+            "gc: vchunk={} in pg_chunk_collection for pg_id={} has been update from pchunk_id={} to pchunk_id={}",
+            v_chunk_id, pg_id, old_chunk_id, new_chunk_id);
     }
 
+    // LOGDEBUGMOD(homeobject, "gc: after switch chunks for pg_id={}, pg_chunks={}", pg_chunks);
+
     pg_chunk_collection->available_blk_count += new_available_blks - old_available_blks;
 }
 

src/lib/homestore_backend/hs_blob_manager.cpp

@@ -301,7 +301,7 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob(ShardInfo const& shard,
         return folly::makeUnexpected(BlobError(BlobErrorCode::RETRY_REQUEST));
     }
 
-    BLOGD(tid, shard.id, blob_id, "Blob Get request: pd={}, group={}, shard=0x{:x}, blob={}, offset={}, len={}", pg_id,
+    BLOGD(tid, shard.id, blob_id, "Blob Get request: pg={}, group={}, shard=0x{:x}, blob={}, offset={}, len={}", pg_id,
           repl_dev->group_id(), shard.id, blob_id, req_offset, req_len);
     auto r = get_blob_from_index_table(index_table, shard.id, blob_id);
     if (!r) {
@@ -378,7 +378,7 @@ BlobManager::AsyncResult< Blob > HSHomeObject::_get_blob_data(const shared< home
             auto body = sisl::io_blob_safe(res_len);
             std::memcpy(body.bytes(), blob_bytes + req_offset, res_len);
 
-            BLOGD(tid, blob_id, shard_id, "Blob get success: blkid={}", blkid.to_string());
+            BLOGD(tid, shard_id, blob_id, "Blob get success: blkid={}", blkid.to_string());
             decr_pending_request_num();
             return Blob(std::move(body), std::move(user_key), header->object_offset, repl_dev->get_leader_id());
         });
@@ -423,9 +423,9 @@ HSHomeObject::blob_put_get_blk_alloc_hints(sisl::blob const& header, cintrusive<
         return folly::makeUnexpected(homestore::ReplServiceError::RESULT_NOT_EXIST_YET);
     }
 
-    homestore::blk_alloc_hints hints;
-
     auto hs_shard = d_cast< HS_Shard* >((*shard_iter->second).get());
+
+    homestore::blk_alloc_hints hints;
     hints.chunk_id_hint = hs_shard->sb_->p_chunk_id;
     if (hs_ctx->is_proposer()) { hints.reserved_blks = get_reserved_blks(); }
     BLOGD(tid, msg_header->shard_id, msg_header->blob_id, "Picked p_chunk_id={}, reserved_blks={}",

src/lib/homestore_backend/hs_shard_manager.cpp

@@ -41,6 +41,8 @@ ShardError toShardError(ReplServiceError const& e) {
         [[fallthrough]];
     case ReplServiceError::TERM_MISMATCH:
         return ShardError::PG_NOT_READY;
+    case ReplServiceError::NO_SPACE_LEFT:
+        return ShardError::NO_SPACE_LEFT;
     case ReplServiceError::NOT_LEADER:
         return ShardError::NOT_LEADER;
     case ReplServiceError::TIMEOUT:
@@ -153,7 +155,8 @@ ShardManager::AsyncResult< ShardInfo > HSHomeObject::_create_shard(pg_id_t pg_ow
         decr_pending_request_num();
         return folly::makeUnexpected(ShardError::NO_SPACE_LEFT);
     }
-    SLOGD(tid, new_shard_id, "vchunk_id={}", v_chunkID.value());
+    const auto v_chunk_id = v_chunkID.value();
+    SLOGD(tid, new_shard_id, "vchunk_id={}", v_chunk_id);
 
     // Prepare the shard info block
     sisl::io_blob_safe sb_blob(sisl::round_up(sizeof(shard_info_superblk), repl_dev->get_blk_size()), io_align);
@@ -168,7 +171,7 @@ ShardManager::AsyncResult< ShardInfo > HSHomeObject::_create_shard(pg_id_t pg_ow
                          .available_capacity_bytes = size_bytes,
                          .total_capacity_bytes = size_bytes};
     sb->p_chunk_id = 0;
-    sb->v_chunk_id = v_chunkID.value();
+    sb->v_chunk_id = v_chunk_id;
 
     auto req = repl_result_ctx< ShardManager::Result< ShardInfo > >::make(
         sizeof(shard_info_superblk) /* header_extn_size */, 0u /* key_size */);
@@ -192,20 +195,33 @@ ShardManager::AsyncResult< ShardInfo > HSHomeObject::_create_shard(pg_id_t pg_ow
 
     // replicate this create shard message to PG members;
     repl_dev->async_alloc_write(req->cheader_buf(), sisl::blob{}, req->data_sgs(), req, false /* part_of_batch */, tid);
-    return req->result().deferValue(
-        [this, req, repl_dev, tid](const auto& result) -> ShardManager::AsyncResult< ShardInfo > {
-            if (result.hasError()) {
-                auto err = result.error();
-                // FIXME: RETURNING CORRECT LEADER
-                // if (err == ShardError::NOT_LEADER) { err.current_leader = repl_dev->get_leader_id(); }
-                decr_pending_request_num();
-                return folly::makeUnexpected(err);
+    return req->result().deferValue([this, req, repl_dev, tid, pg_owner, new_shard_id,
+                                     v_chunk_id](const auto& result) -> ShardManager::AsyncResult< ShardInfo > {
+        if (result.hasError()) {
+            auto err = result.error();
+            // FIXME: RETURNING CORRECT LEADER
+            // if (err == ShardError::NOT_LEADER) { err.current_leader = repl_dev->get_leader_id(); }
+
+            bool res = chunk_selector()->release_chunk(pg_owner, v_chunk_id);
+            RELEASE_ASSERT(res, "Failed to release v_chunk_id={}, pg={}", v_chunk_id, pg_owner);
+
+            SLOGE(tid, new_shard_id, "got {} when creating shard at leader, failed to create shard {}!", err,
+                  new_shard_id);
+
+            if (err == ShardError::NO_SPACE_LEFT) {
+                gc_manager()->submit_gc_task(task_priority::normal,
+                                             chunk_selector()->get_pg_vchunk(pg_owner, v_chunk_id)->get_chunk_id());
+                SLOGD(tid, new_shard_id, "got no space left error when creating shard {} at leader", new_shard_id);
             }
-            auto shard_info = result.value();
-            SLOGD(tid, shard_info.id, "Shard created success.");
+
             decr_pending_request_num();
-            return shard_info;
-        });
+            return folly::makeUnexpected(err);
+        }
+        auto shard_info = result.value();
+        SLOGD(tid, shard_info.id, "Shard created success.");
+        decr_pending_request_num();
+        return shard_info;
+    });
 }
 
 ShardManager::AsyncResult< ShardInfo > HSHomeObject::_seal_shard(ShardInfo const& info, trace_id_t tid) {

src/lib/homestore_backend/replication_state_machine.cpp

@@ -674,11 +674,15 @@ folly::Future< std::error_code > ReplicationStateMachine::on_fetch_data(const in
             .via(folly::getGlobalIOExecutor())
             .thenValue([this, lsn, blob_id, shard_id, given_buffer, total_size](auto&& err) {
                 // io error
-                if (err) throw std::system_error(err);
+                if (err) {
+                    LOGE("FetchData fails to read blob, lsn={}, blob_id={}, shard_id={}, err_value={}, error={}", lsn,
+                         blob_id, shard_id, err.value(), err.message());
+                    throw std::system_error(err);
+                }
 
-                // folly future has no machenism to bypass the later thenValue in the then value chain. so for all
-                // the case that no need to schedule the later async_read, we throw a system_error with no error
-                // code to bypass the next thenValue.
+            // folly future has no machenism to bypass the later thenValue in the then value chain. so for all
+            // the case that no need to schedule the later async_read, we throw a system_error with no error
+            // code to bypass the next thenValue.
 #ifdef _PRERELEASE
                 if (iomgr_flip::instance()->test_flip("local_blk_data_invalid")) {
                     LOGI("Simulating forcing to read by indextable");
@@ -734,6 +738,10 @@ folly::Future< std::error_code > ReplicationStateMachine::on_fetch_data(const in
 
                 RELEASE_ASSERT(pbas.blk_count() * repl_dev()->get_blk_size() == total_size,
                                "pbas blk size does not match total_size");
+
+                LOGD("on_fetch_data: read data with blob_id={}, shardID=0x{:x}, pg={}, shard=0x{:x} from pbas={}",
+                     blob_id, shard_id, pg_id, (shard_id & homeobject::shard_mask), pbas.to_string());
+
                 return homestore::data_service().async_read(pbas, given_buffer, total_size);
             })
             .thenValue([this, lsn, blob_id, shard_id, given_buffer, total_size](auto&& err) {
@@ -785,8 +793,7 @@ bool ReplicationStateMachine::validate_blob(shard_id_t shard_id, blob_id_t blob_
         return false;
     }
     if (header->shard_id != shard_id) {
-        LOGD("shard_id does not match , expected shardID=0x{:x} , shard_id in header=0x{:x}", shard_id,
-             header->shard_id);
+        LOGD("shard_id does not match , expected shardID={} , shard_id in header={}", shard_id, header->shard_id);
         return false;
     }
     if (header->blob_id != blob_id) {
@@ -975,11 +982,13 @@ void ReplicationStateMachine::handle_no_space_left(homestore::repl_lsn_t lsn, ho
             .via(&folly::InlineExecutor::instance())
             .thenValue([this, lsn, chunk_id](auto&& res) {
                 if (!res) {
-                    RELEASE_ASSERT(false,
-                                   "failed to submit emergent gc task for chunk_id={} , lsn={} - fatal error, aborting",
-                                   chunk_id, lsn);
+                    LOGERROR("failed to submit emergent gc task for chunk_id={} , lsn={}, will retry again if new "
+                             "no_space_left happens",
+                             chunk_id, lsn);
+                } else {
+                    LOGD("successfully handle no_space_left error for chunk_id={} , lsn={}", chunk_id, lsn);
                 }
-                LOGD("successfully handle no_space_left error for chunk_id={} , lsn={}", chunk_id, lsn);
+
                 // start accepting new requests again.
                 repl_dev()->resume_accepting_reqs();
             });