Fix HeapChunkSelector::get_most_available_blk_chunk

1. Add check to ensure max_it points to an available chunk.
   If not, it indicates there are no available chunks left.
2. Add safeguards in the create and recover pg paths to ensure
   the pg size cannot be zero.

conanfile.py

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

src/lib/homestore_backend/heap_chunk_selector.cpp

@@ -129,6 +129,10 @@ std::optional< uint32_t > HeapChunkSelector::select_chunks_for_pg(pg_id_t pg_id,
         LOGWARNMOD(homeobject, "PG had already created, pg_id {}", pg_id);
         return std::nullopt;
     }
+    if (pg_size == 0) {
+        LOGWARNMOD(homeobject, "Not supported to create empty PG, pg_id {}, pg_size {}", pg_id, pg_size);
+        return std::nullopt;
+    }
 
     const auto chunk_size = get_chunk_size();
     const uint32_t num_chunk = sisl::round_down(pg_size, chunk_size) / chunk_size;
@@ -178,6 +182,10 @@ bool HeapChunkSelector::recover_pg_chunks(pg_id_t pg_id, std::vector< chunk_num_
         LOGWARNMOD(homeobject, "PG {} had been recovered", pg_id);
         return false;
     }
+    if (p_chunk_ids.size() == 0) {
+        LOGWARNMOD(homeobject, "Unexpected empty PG {}", pg_id);
+        return false;
+    }
 
     // check chunks valid, must belong to m_chunks and have same pdev_id
     std::optional< uint32_t > last_pdev_id;
@@ -277,23 +285,18 @@ std::optional< homestore::chunk_num_t > HeapChunkSelector::get_most_available_bl
         LOGWARNMOD(homeobject, "No pg found for pg_id {}", pg_id);
         return std::nullopt;
     }
-    if (pg_it->second->available_num_chunks == 0) {
-        LOGWARNMOD(homeobject, "No available chunk for pg {}", pg_id);
-        return std::nullopt;
-    }
-
     std::scoped_lock lock(pg_it->second->mtx);
     auto pg_chunk_collection = pg_it->second;
     auto& pg_chunks = pg_chunk_collection->m_pg_chunks;
     auto max_it =
         std::max_element(pg_chunks.begin(), pg_chunks.end(),
                          [](const std::shared_ptr< ExtendedVChunk >& a, const std::shared_ptr< ExtendedVChunk >& b) {
-                             if (a->available() && b->available()) { return a->available_blks() < b->available_blks(); }
-                             if (!a->available() && b->available()) { return true; }
-                             if (a->available() && !b->available()) { return false; }
-                             return false;
+                             return !a->available() || (b->available() && a->available_blks() < b->available_blks());
                          });
-
+    if (!(*max_it)->available()) {
+        LOGWARNMOD(homeobject, "No available chunk for PG {}", pg_id);
+        return std::nullopt;
+    }
     auto v_chunk_id = std::distance(pg_chunks.begin(), max_it);
     pg_chunks[v_chunk_id]->m_state = ChunkState::INUSE;
     --pg_chunk_collection->available_num_chunks;

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -60,6 +60,11 @@ PGManager::NullAsyncResult HSHomeObject::_create_pg(PGInfo&& pg_info, std::set<
     auto pg_id = pg_info.id;
     if (auto lg = std::shared_lock(_pg_lock); _pg_map.end() != _pg_map.find(pg_id)) return folly::Unit();
 
+    if (pg_info.size == 0) {
+        LOGW("Not supported to create empty PG, pg_id {}, pg_size {}", pg_id, pg_info.size);
+        return folly::makeUnexpected(PGError::INVALID_ARG);
+    }
+
     const auto most_avail_num_chunks = chunk_selector()->most_avail_num_chunks();
     const auto chunk_size = chunk_selector()->get_chunk_size();
     const auto needed_num_chunks = sisl::round_down(pg_info.size, chunk_size) / chunk_size;

src/lib/homestore_backend/tests/test_heap_chunk_selector.cpp

@@ -108,7 +108,9 @@ class HeapChunkSelectorTest : public ::testing::Test {
         const uint32_t chunk_size = HCS.get_chunk_size();
         const u_int64_t pg_size = chunk_size * 3;
         for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
-            ASSERT_EQ(HCS.select_chunks_for_pg(pg_id, pg_size), 3);
+            // not supported to create empty pg
+            ASSERT_FALSE(HCS.select_chunks_for_pg(pg_id, 0).has_value());
+            ASSERT_EQ(HCS.select_chunks_for_pg(pg_id, pg_size).value(), 3);
             uint32_t last_pdev_id = 0;
             // test pg heap
             auto pg_it = HCS.m_per_pg_chunks.find(pg_id);
@@ -287,6 +289,7 @@ TEST_F(HeapChunkSelectorTest, test_recovery) {
     // on_pg_meta_blk_found
     for (uint16_t pg_id = 1; pg_id < 4; ++pg_id) {
         std::vector< chunk_num_t > chunk_ids{1, 2};
+        std::vector< chunk_num_t > empty_chunk_ids{};
         std::vector< chunk_num_t > chunk_ids_for_twice{1, 2};
         std::vector< chunk_num_t > chunk_ids_not_valid{1, 20};
         std::vector< chunk_num_t > chunk_ids_not_same_pdev{1, 6};
@@ -298,6 +301,7 @@ TEST_F(HeapChunkSelectorTest, test_recovery) {
         }
 
         // test recover chunk map
+        ASSERT_FALSE(HCS_recovery.recover_pg_chunks(pg_id, std::move(empty_chunk_ids)));
         ASSERT_FALSE(HCS_recovery.recover_pg_chunks(pg_id, std::move(chunk_ids_not_valid)));
         ASSERT_FALSE(HCS_recovery.recover_pg_chunks(pg_id, std::move(chunk_ids_not_same_pdev)));