Merge pull request ceph#56332 from cyx1231st/wip-fix-racing-get-or-create-pg-mapping

cyx1231st · web-flow · commit 9555086eaf51 · 2024-03-26T11:41:42.000+08:00
crimson/os/pg_map: allow multiple shards to create new pg mappings at the same time

Reviewed-by: Samuel Just &lt;sjust@redhat.com&gt;
Reviewed-by: Matan Breizman &lt;mbreizma@redhat.com&gt;
diff --git a/src/crimson/osd/pg_map.cc b/src/crimson/osd/pg_map.cc
@@ -14,75 +14,144 @@ namespace crimson::osd {
 
 seastar::future<core_id_t> PGShardMapping::get_or_create_pg_mapping(
   spg_t pgid,
-  core_id_t core)
+  core_id_t core_expected)
 {
   LOG_PREFIX(PGShardMapping::get_or_create_pg_mapping);
   auto find_iter = pg_to_core.find(pgid);
   if (find_iter != pg_to_core.end()) {
-    ceph_assert_always(find_iter->second != NULL_CORE);
-    if (core != NULL_CORE) {
-      ceph_assert_always(find_iter->second == core);
+    auto core_found = find_iter->second;
+    assert(core_found != NULL_CORE);
+    if (core_expected != NULL_CORE && core_expected != core_found) {
+      ERROR("the mapping is inconsistent for pg {}: core {}, expected {}",
+            pgid, core_found, core_expected);
+      ceph_abort("The pg mapping is inconsistent!");
     }
-    return seastar::make_ready_future<core_id_t>(find_iter->second);
+    return seastar::make_ready_future<core_id_t>(core_found);
   } else {
-    return container().invoke_on(0,[pgid, core, FNAME]
-      (auto &primary_mapping) {
-      auto [insert_iter, inserted] = primary_mapping.pg_to_core.emplace(pgid, core);
-      ceph_assert_always(inserted);
-      ceph_assert_always(primary_mapping.core_to_num_pgs.size() > 0);
-      std::map<core_id_t, unsigned>::iterator core_iter;
-      if (core == NULL_CORE) {
-        core_iter = std::min_element(
-          primary_mapping.core_to_num_pgs.begin(),
-          primary_mapping.core_to_num_pgs.end(),
+    DEBUG("calling primary to add mapping for pg {} to the expected core {}",
+          pgid, core_expected);
+    return container().invoke_on(
+        0, [pgid, core_expected, FNAME](auto &primary_mapping) {
+      auto core_to_update = core_expected;
+      auto find_iter = primary_mapping.pg_to_core.find(pgid);
+      if (find_iter != primary_mapping.pg_to_core.end()) {
+        // this pgid was already mapped within primary_mapping, assert that the
+        // mapping is consistent and avoid emplacing once again.
+        auto core_found = find_iter->second;
+        assert(core_found != NULL_CORE);
+        if (core_expected != NULL_CORE) {
+          if (core_expected != core_found) {
+            ERROR("the mapping is inconsistent for pg {} (primary): core {}, expected {}",
+                  pgid, core_found, core_expected);
+            ceph_abort("The pg mapping is inconsistent!");
+          }
+          // core_expected == core_found
+          DEBUG("mapping pg {} to core {} (primary): already mapped and expected",
+                pgid, core_to_update);
+        } else { // core_expected == NULL_CORE
+          core_to_update = core_found;
+          DEBUG("mapping pg {} to core {} (primary): already mapped",
+                pgid, core_to_update);
+        }
+        // proceed to broadcast core_to_update
+      } else { // find_iter == primary_mapping.pg_to_core.end()
+        // this pgid isn't mapped within primary_mapping,
+        // add the mapping and ajust core_to_num_pgs
+        ceph_assert_always(primary_mapping.core_to_num_pgs.size() > 0);
+        std::map<core_id_t, unsigned>::iterator count_iter;
+        if (core_expected == NULL_CORE) {
+          count_iter = std::min_element(
+            primary_mapping.core_to_num_pgs.begin(),
+            primary_mapping.core_to_num_pgs.end(),
             [](const auto &left, const auto &right) {
-            return left.second < right.second;
-        });
-      } else {
-        core_iter = primary_mapping.core_to_num_pgs.find(core);
+              return left.second < right.second;
+            }
+          );
+          core_to_update = count_iter->first;
+        } else { // core_expected != NULL_CORE
+          count_iter = primary_mapping.core_to_num_pgs.find(core_to_update);
+        }
+        ceph_assert_always(primary_mapping.core_to_num_pgs.end() != count_iter);
+        ++(count_iter->second);
+        auto [insert_iter, inserted] =
+          primary_mapping.pg_to_core.emplace(pgid, core_to_update);
+        assert(inserted);
+        DEBUG("mapping pg {} to core {} (primary): num_pgs {}",
+              pgid, core_to_update, count_iter->second);
       }
-      ceph_assert_always(primary_mapping.core_to_num_pgs.end() != core_iter);
-      insert_iter->second = core_iter->first;
-      core_iter->second++;
-      DEBUG("mapping pg {} to core: {} with num_pgs of: {}",
-            pgid, insert_iter->second, core_iter->second);
+      assert(core_to_update != NULL_CORE);
       return primary_mapping.container().invoke_on_others(
-        [pgid = insert_iter->first, core = insert_iter->second, FNAME]
-        (auto &other_mapping) {
-        ceph_assert_always(core != NULL_CORE);
-        auto [insert_iter, inserted] = other_mapping.pg_to_core.emplace(pgid, core);
-        ceph_assert_always(inserted);
-        DEBUG("mapping pg {} to core: {}", pgid, core);
+          [pgid, core_to_update, FNAME](auto &other_mapping) {
+        auto find_iter = other_mapping.pg_to_core.find(pgid);
+        if (find_iter == other_mapping.pg_to_core.end()) {
+          DEBUG("mapping pg {} to core {} (others)",
+                pgid, core_to_update);
+          auto [insert_iter, inserted] =
+            other_mapping.pg_to_core.emplace(pgid, core_to_update);
+          assert(inserted);
+        } else {
+          auto core_found = find_iter->second;
+          if (core_found != core_to_update) {
+            ERROR("the mapping is inconsistent for pg {} (others): core {}, expected {}",
+                  pgid, core_found, core_to_update);
+            ceph_abort("The pg mapping is inconsistent!");
+          }
+          DEBUG("mapping pg {} to core {} (others): already mapped",
+                pgid, core_to_update);
+        }
       });
-    }).then([this, pgid, FNAME] {
+    }).then([this, pgid, core_expected, FNAME] {
       auto find_iter = pg_to_core.find(pgid);
-      ceph_assert_always(find_iter != pg_to_core.end());
-      DEBUG("returning pg {} mapping to core {}", pgid, find_iter->second);
-      return seastar::make_ready_future<core_id_t>(find_iter->second);
+      if (find_iter == pg_to_core.end()) {
+        ERROR("the mapping is inconsistent for pg {}: core not found, expected {}",
+              pgid, core_expected);
+        ceph_abort("The pg mapping is inconsistent!");
+      }
+      auto core_found = find_iter->second;
+      if (core_expected != NULL_CORE && core_found != core_expected) {
+        ERROR("the mapping is inconsistent for pg {}: core {}, expected {}",
+              pgid, core_found, core_expected);
+        ceph_abort("The pg mapping is inconsistent!");
+      }
+      DEBUG("returning pg {} mapping to core {} after broadcasted",
+            pgid, core_found);
+      return seastar::make_ready_future<core_id_t>(core_found);
     });
   }
 }
 
 seastar::future<> PGShardMapping::remove_pg_mapping(spg_t pgid) {
   LOG_PREFIX(PGShardMapping::remove_pg_mapping);
-  DEBUG("{}", pgid);
-  return container().invoke_on(0, [pgid, FNAME](auto &primary_mapping) {
-    auto iter = primary_mapping.pg_to_core.find(pgid);
-    ceph_assert_always(iter != primary_mapping.pg_to_core.end());
-    ceph_assert_always(iter->second != NULL_CORE);
-    auto count_iter = primary_mapping.core_to_num_pgs.find(iter->second);
-    ceph_assert_always(count_iter != primary_mapping.core_to_num_pgs.end());
-    ceph_assert_always(count_iter->second > 0);
+  auto find_iter = pg_to_core.find(pgid);
+  if (find_iter == pg_to_core.end()) {
+    ERROR("trying to remove non-exist mapping for pg {}", pgid);
+    ceph_abort("The pg mapping is inconsistent!");
+  }
+  DEBUG("calling primary to remove mapping for pg {}", pgid);
+  return container().invoke_on(
+      0, [pgid, FNAME](auto &primary_mapping) {
+    auto find_iter = primary_mapping.pg_to_core.find(pgid);
+    if (find_iter == primary_mapping.pg_to_core.end()) {
+      ERROR("trying to remove non-exist mapping for pg {} (primary)", pgid);
+      ceph_abort("The pg mapping is inconsistent!");
+    }
+    assert(find_iter->second != NULL_CORE);
+    auto count_iter = primary_mapping.core_to_num_pgs.find(find_iter->second);
+    assert(count_iter != primary_mapping.core_to_num_pgs.end());
+    assert(count_iter->second > 0);
     --(count_iter->second);
-    primary_mapping.pg_to_core.erase(iter);
-    DEBUG("pg {} mapping erased", pgid);
+    primary_mapping.pg_to_core.erase(find_iter);
+    DEBUG("pg {} mapping erased (primary)", pgid);
     return primary_mapping.container().invoke_on_others(
       [pgid, FNAME](auto &other_mapping) {
-      auto iter = other_mapping.pg_to_core.find(pgid);
-      ceph_assert_always(iter != other_mapping.pg_to_core.end());
-      ceph_assert_always(iter->second != NULL_CORE);
-      other_mapping.pg_to_core.erase(iter);
-      DEBUG("pg {} mapping erased", pgid);
+      auto find_iter = other_mapping.pg_to_core.find(pgid);
+      if (find_iter == other_mapping.pg_to_core.end()) {
+        ERROR("trying to remove non-exist mapping for pg {} (others)", pgid);
+        ceph_abort("The pg mapping is inconsistent!");
+      }
+      assert(find_iter->second != NULL_CORE);
+      other_mapping.pg_to_core.erase(find_iter);
+      DEBUG("pg {} mapping erased (others)", pgid);
     });
   });
 }
diff --git a/src/crimson/osd/pg_map.h b/src/crimson/osd/pg_map.h
@@ -37,7 +37,7 @@ class PGShardMapping : public seastar::peering_sharded_service<PGShardMapping> {
   /// Returns mapping for pgid, creates new one if it doesn't already exist
   seastar::future<core_id_t> get_or_create_pg_mapping(
     spg_t pgid,
-    core_id_t core = NULL_CORE);
+    core_id_t core_expected = NULL_CORE);
 
   /// Remove pgid mapping
   seastar::future<> remove_pg_mapping(spg_t pgid);
@@ -60,7 +60,9 @@ class PGShardMapping : public seastar::peering_sharded_service<PGShardMapping> {
   }
 
 private:
+  // only in shard 0
   std::map<core_id_t, unsigned> core_to_num_pgs;
+  // per-shard, updated by shard 0
   std::map<spg_t, core_id_t> pg_to_core;
 };
 
diff --git a/src/crimson/osd/shard_services.cc b/src/crimson/osd/shard_services.cc
@@ -183,7 +183,7 @@ seastar::future<> OSDSingletonState::osdmap_subscribe(
   version_t epoch, bool force_request)
 {
   LOG_PREFIX(OSDSingletonState::osdmap_subscribe);
-  INFO("epoch {}");
+  INFO("epoch {}", epoch);
   if (monc.sub_want_increment("osdmap", epoch, CEPH_SUBSCRIBE_ONETIME) ||
       force_request) {
     return monc.renew_subs();

Original file line number	Diff line number	Diff line change
`@@ -183,7 +183,7 @@ seastar::future<> OSDSingletonState::osdmap_subscribe(`
`183`	`183`	`version_t epoch, bool force_request)`
`184`	`184`	`{`
`185`	`185`	`LOG_PREFIX(OSDSingletonState::osdmap_subscribe);`
`186`		`- INFO("epoch {}");`
	`186`	`+ INFO("epoch {}", epoch);`
`187`	`187`	`if (monc.sub_want_increment("osdmap", epoch, CEPH_SUBSCRIBE_ONETIME) \|\|`
`188`	`188`	`force_request) {`
`189`	`189`	`return monc.renew_subs();`