Merge pull request ceph#62146 from Matan-B/wip-matanb-crimson-wip-69439-v2

crimson: Handle peer replies handling once received 

Reviewed-by: Samuel Just <[email protected]>
Reviewed-by: Xuehan Xu <[email protected]>

Author: Matan-B

src/crimson/osd/ec_backend.cc

@@ -33,6 +33,5 @@ ECBackend::submit_transaction(const std::set<pg_shard_t> &pg_shards,
 			      std::vector<pg_log_entry_t>&& log_entries)
 {
   // todo
-  return make_ready_future<rep_op_ret_t>(seastar::now(),
-	  seastar::make_ready_future<crimson::osd::acked_peers_t>());
+  return make_ready_future<rep_op_ret_t>(seastar::now(), seastar::now());
 }

src/crimson/osd/osd_operations/client_request.h

@@ -107,8 +107,7 @@ class ClientRequest final : public PhasedOperationT<ClientRequest>,
       CommonOBCPipeline::Process::BlockingEvent,
       scrub::PGScrubber::BlockingEvent,
       CommonOBCPipeline::WaitRepop::BlockingEvent,
-      CommonOBCPipeline::SendReply::BlockingEvent,
-      CompletionEvent
+      CommonOBCPipeline::SendReply::BlockingEvent
       > pg_tracking_events;
 
     template <class BlockingEventT>

src/crimson/osd/pg.cc

@@ -890,9 +890,7 @@ void PG::enqueue_delete_for_backfill(
   backfill_state->enqueue_standalone_delete(obj, v, peers);
 }
 
-PG::interruptible_future<
-  std::tuple<PG::interruptible_future<>,
-             PG::interruptible_future<>>>
+PG::interruptible_future<PG::rep_op_fut_t>
 PG::submit_transaction(
   ObjectContextRef&& obc,
   ObjectContextRef&& new_clone,
@@ -932,16 +930,7 @@ PG::submit_transaction(
       std::move(log_entries));
   co_return std::make_tuple(
     std::move(submitted),
-    all_completed.then_interruptible(
-      [this, last_complete=peering_state.get_info().last_complete, at_version]
-      (auto acked) {
-      for (const auto& peer : acked) {
-        peering_state.update_peer_last_complete_ondisk(
-          peer.shard, peer.last_complete_ondisk);
-      }
-      peering_state.complete_write(at_version, last_complete);
-      return seastar::now();
-    })
+    std::move(all_completed)
   );
 }
 

src/crimson/osd/pg.h

@@ -560,6 +560,20 @@ class PG : public boost::intrusive_ref_counter<
     return peering_state.get_primary();
   }
 
+  eversion_t get_last_complete() const {
+    return peering_state.get_info().last_complete;
+  }
+
+  void complete_write(eversion_t v, eversion_t lc) {
+    peering_state.complete_write(v, lc);
+  }
+
+  void update_peer_last_complete_ondisk(
+    pg_shard_t fromosd,
+    eversion_t lcod) {
+    peering_state.update_peer_last_complete_ondisk(fromosd, lcod);
+  }
+
   /// initialize created PG
   seastar::future<> init(
     int role,
@@ -686,8 +700,9 @@ class PG : public boost::intrusive_ref_counter<
   interruptible_future<MURef<MOSDOpReply>> do_pg_ops(Ref<MOSDOp> m);
 
 public:
-  interruptible_future<
-    std::tuple<interruptible_future<>, interruptible_future<>>>
+  using rep_op_fut_t = std::tuple<interruptible_future<>,
+                                  interruptible_future<>>;
+  interruptible_future<rep_op_fut_t>
   submit_transaction(
     ObjectContextRef&& obc,
     ObjectContextRef&& new_clone,

src/crimson/osd/pg_backend.h

@@ -62,7 +62,7 @@ class PGBackend
       ::crimson::osd::IOInterruptCondition, T>;
   using rep_op_ret_t = 
     std::tuple<interruptible_future<>,
-	       interruptible_future<crimson::osd::acked_peers_t>>;
+	       interruptible_future<>>;
   using rep_op_fut_t = interruptible_future<rep_op_ret_t>;
   PGBackend(shard_id_t shard, CollectionRef coll,
             crimson::osd::ShardServices &shard_services,

src/crimson/osd/replicated_backend.cc

@@ -92,7 +92,11 @@ ReplicatedBackend::submit_transaction(
 
   const ceph_tid_t tid = shard_services.get_tid();
   auto pending_txn =
-    pending_trans.try_emplace(tid, pg_shards.size(), osd_op_p.at_version).first;
+    pending_trans.try_emplace(
+      tid,
+      pg_shards.size(),
+      osd_op_p.at_version,
+      pg.get_last_complete()).first;
   bufferlist encoded_txn;
   encode(txn, encoded_txn);
 
@@ -162,10 +166,13 @@ ReplicatedBackend::submit_transaction(
       assert(0 == "impossible");
     }
     if (--peers->pending == 0) {
+      // no peers other than me, replication size is 1
+      pg.complete_write(peers->at_version, peers->last_complete);
       peers->all_committed.set_value();
       peers->all_committed = {};
       return seastar::now();
     }
+    // wait for all peers to ack (ReplicatedBackend::got_rep_op_reply)
     return peers->all_committed.get_shared_future();
   }).then_interruptible([pending_txn, this, _new_clone, &hoid,
 			to_push_delete=std::move(to_push_delete),
@@ -181,8 +188,7 @@ ReplicatedBackend::submit_transaction(
     if (!to_push_delete.empty()) {
       pg.enqueue_delete_for_backfill(hoid, {}, to_push_delete);
     }
-    return seastar::make_ready_future<
-      crimson::osd::acked_peers_t>(std::move(acked_peers));
+    return seastar::now();
   });
 
   auto sends_complete = seastar::when_all_succeed(
@@ -212,7 +218,10 @@ void ReplicatedBackend::got_rep_op_reply(const MOSDRepOpReply& reply)
   for (auto& peer : peers.acked_peers) {
     if (peer.shard == reply.from) {
       peer.last_complete_ondisk = reply.get_last_complete_ondisk();
+      pg.update_peer_last_complete_ondisk(
+	peer.shard, peer.last_complete_ondisk);
       if (--peers.pending == 0) {
+        pg.complete_write(peers.at_version, peers.last_complete);
         peers.all_committed.set_value();
         peers.all_committed = {};
       }

src/crimson/osd/replicated_backend.h

@@ -44,8 +44,13 @@ class ReplicatedBackend : public PGBackend
   const pg_shard_t whoami;
   class pending_on_t : public seastar::weakly_referencable<pending_on_t> {
   public:
-    pending_on_t(size_t pending, const eversion_t& at_version)
-      : pending{static_cast<unsigned>(pending)}, at_version(at_version)
+    pending_on_t(
+      size_t pending,
+      const eversion_t& at_version,
+      const eversion_t& last_complete)
+      : pending{static_cast<unsigned>(pending)},
+	at_version(at_version),
+	last_complete(last_complete)
     {}
     unsigned pending;
     // The order of pending_txns' at_version must be the same as their
@@ -54,6 +59,7 @@ class ReplicatedBackend : public PGBackend
     // way, client requests at_version must be updated synchorously/simultaneously
     // with ceph_tid_t.
     const eversion_t at_version;
+    const eversion_t last_complete;
     crimson::osd::acked_peers_t acked_peers;
     seastar::shared_promise<> all_committed;
   };