crimson/osd/pg: introduce projected_log

Signed-off-by: Matan Breizman <[email protected]>

Author: Matan-B

src/crimson/osd/backfill_facades.h

@@ -36,6 +36,10 @@ struct PeeringFacade final : BackfillState::PeeringFacade {
     return peering_state.get_info().log_tail;
   }
 
+  const PGLog& get_pg_log() const override {
+    return peering_state.get_pg_log();
+  }
+
   void scan_log_after(eversion_t v, scan_log_func_t f) const override {
     peering_state.get_pg_log().get_log().scan_log_after(v, std::move(f));
   }
@@ -73,6 +77,10 @@ struct PGFacade final : BackfillState::PGFacade {
     return pg.projected_last_update;
   }
 
+  const PGLog::IndexedLog& get_projected_log() const override {
+    return pg.projected_log;
+  }
+
   PGFacade(PG& pg) : pg(pg) {}
 };
 

src/crimson/osd/backfill_state.cc

@@ -125,7 +125,6 @@ void BackfillState::Enqueuing::maybe_update_range()
     logger().info("{}: bi is current", __func__);
     ceph_assert(primary_bi.version == pg().get_projected_last_update());
   } else if (primary_bi.version >= peering_state().get_log_tail()) {
-#if 0
     if (peering_state().get_pg_log().get_log().empty() &&
         pg().get_projected_log().empty()) {
       /* Because we don't move log_tail on split, the log might be
@@ -137,13 +136,11 @@ void BackfillState::Enqueuing::maybe_update_range()
       ceph_assert(primary_bi.version == eversion_t());
       return;
     }
-#endif
     logger().debug("{}: bi is old, ({}) can be updated with log to {}",
                    __func__,
                    primary_bi.version,
                    pg().get_projected_last_update());
-    logger().debug("{}: scanning pg log first", __func__);
-    peering_state().scan_log_after(primary_bi.version,
+    auto func =
       [&](const pg_log_entry_t& e) {
         logger().debug("maybe_update_range(lambda): updating from version {}",
                        e.version);
@@ -160,7 +157,11 @@ void BackfillState::Enqueuing::maybe_update_range()
             primary_bi.objects.erase(e.soid);
           }
         }
-      });
+      };
+    logger().debug("{}: scanning pg log first", __func__);
+    peering_state().scan_log_after(primary_bi.version, func);
+    logger().debug("{}: scanning projected log", __func__);
+    pg().get_projected_log().scan_log_after(primary_bi.version, func);
     primary_bi.version = pg().get_projected_last_update();
   } else {
     ceph_abort_msg(

src/crimson/osd/backfill_state.h

@@ -14,6 +14,7 @@
 #include <boost/statechart/transition.hpp>
 
 #include "osd/recovery_types.h"
+#include "osd/PGLog.h"
 
 namespace crimson::osd {
 
@@ -367,6 +368,7 @@ struct BackfillState::PeeringFacade {
   virtual hobject_t earliest_backfill() const = 0;
   virtual const std::set<pg_shard_t>& get_backfill_targets() const = 0;
   virtual const hobject_t& get_peer_last_backfill(pg_shard_t peer) const = 0;
+  virtual const PGLog& get_pg_log() const = 0;
   virtual const eversion_t& get_last_update() const = 0;
   virtual const eversion_t& get_log_tail() const = 0;
 
@@ -392,6 +394,8 @@ struct BackfillState::PeeringFacade {
 // of behaviour that must be provided by a unit test's mock.
 struct BackfillState::PGFacade {
   virtual const eversion_t& get_projected_last_update() const = 0;
+  virtual const PGLog::IndexedLog& get_projected_log() const = 0;
+
   virtual ~PGFacade() {}
 };
 

src/crimson/osd/pg.cc

@@ -932,6 +932,10 @@ PG::submit_transaction(
   ceph_assert(log_entries.rbegin()->version >= projected_last_update);
   projected_last_update = log_entries.rbegin()->version;
 
+  for (const auto& entry: log_entries) {
+    projected_log.add(entry);
+  }
+
   auto [submitted, all_completed] = co_await backend->submit_transaction(
       peering_state.get_acting_recovery_backfill(),
       obc->obs.oi.soid,
@@ -1333,19 +1337,15 @@ void PG::log_operation(
   if (is_primary()) {
     ceph_assert(trim_to <= peering_state.get_pg_committed_to());
   }
-  /* TODO: when we add snap mapper and projected log support,
-   * we'll likely want to update them here.
-   *
-   * See src/osd/PrimaryLogPG.h:log_operation for how classic
-   * handles these cases.
-   */
-#if 0
   auto last = logv.rbegin();
   if (is_primary() && last != logv.rend()) {
+    logger().debug("{} on primary, trimming projected log",
+                   __func__);
     projected_log.skip_can_rollback_to_to_head();
-    projected_log.trim(cct, last->version, nullptr, nullptr, nullptr);
+    projected_log.trim(shard_services.get_cct(), last->version,
+                       nullptr, nullptr, nullptr);
   }
-#endif
+
   if (!is_primary()) { // && !is_ec_pg()
     replica_clear_repop_obc(logv);
   }
@@ -1651,8 +1651,8 @@ PG::already_complete(const osd_reqid_t& reqid)
   int ret;
   std::vector<pg_log_op_return_item_t> op_returns;
 
-  if (peering_state.get_pg_log().get_log().get_request(
-	reqid, &version, &user_version, &ret, &op_returns)) {
+  if (check_in_progress_op(
+        reqid, &version, &user_version, &ret, &op_returns)) {
     complete_op_t dupinfo{
       user_version,
       version,
@@ -1717,4 +1717,19 @@ void PG::C_PG_FinishRecovery::finish(int r) {
     DEBUGDPP("stale recovery finsher", pg);
   }
 }
+bool PG::check_in_progress_op(
+  const osd_reqid_t& reqid,
+  eversion_t *version,
+  version_t *user_version,
+  int *return_code,
+  std::vector<pg_log_op_return_item_t> *op_returns
+  ) const
+{
+  return (
+    projected_log.get_request(reqid, version, user_version, return_code,
+                              op_returns) ||
+    peering_state.get_pg_log().get_log().get_request(
+      reqid, version, user_version, return_code, op_returns));
+}
+
 }

src/crimson/osd/pg.h

@@ -376,6 +376,7 @@ class PG : public boost::intrusive_ref_counter<
   void check_blocklisted_watchers() final;
   void clear_primary_state() final {
     recovery_finisher = nullptr;
+    projected_log = PGLog::IndexedLog();
   }
 
   void queue_check_readable(epoch_t last_peering_reset,
@@ -826,8 +827,15 @@ class PG : public boost::intrusive_ref_counter<
     const eversion_t version;
     const int err;
   };
+  PGLog::IndexedLog projected_log;
   interruptible_future<std::optional<complete_op_t>>
   already_complete(const osd_reqid_t& reqid);
+  bool check_in_progress_op(
+    const osd_reqid_t& reqid,
+    eversion_t *version,
+    version_t *user_version,
+    int *return_code,
+    std::vector<pg_log_op_return_item_t> *op_returns) const;
   int get_recovery_op_priority() const {
     int64_t pri = 0;
     get_pgpool().info.opts.get(pool_opts_t::RECOVERY_OP_PRIORITY, &pri);