crimson/osd/pg: Seperate mutate_object from submit_transaction

instead of mutating the object in each transaction being submitted,
only mutate obc for user triggered transactions.
This way, PG::submit_transaction internal users would not mutate obc
(e.g SnapTrimObjSubEvent)

Fixes: https://tracker.ceph.com/issues/67675
Signed-off-by: Xuehan Xu <[email protected]>

Author: xxhdx1985126

src/crimson/osd/pg.cc

@@ -849,6 +849,45 @@ std::ostream& operator<<(std::ostream& os, const PG& pg)
   return os;
 }
 
+void PG::mutate_object(
+  ObjectContextRef& obc,
+  ceph::os::Transaction& txn,
+  osd_op_params_t& osd_op_p)
+{
+  if (obc->obs.exists) {
+    obc->obs.oi.prior_version = obc->obs.oi.version;
+    obc->obs.oi.version = osd_op_p.at_version;
+    if (osd_op_p.user_modify)
+      obc->obs.oi.user_version = osd_op_p.at_version.version;
+    obc->obs.oi.last_reqid = osd_op_p.req_id;
+    obc->obs.oi.mtime = osd_op_p.mtime;
+    obc->obs.oi.local_mtime = ceph_clock_now();
+
+    // object_info_t
+    {
+      ceph::bufferlist osv;
+      obc->obs.oi.encode_no_oid(osv, CEPH_FEATURES_ALL);
+      // TODO: get_osdmap()->get_features(CEPH_ENTITY_TYPE_OSD, nullptr));
+      txn.setattr(coll_ref->get_cid(), ghobject_t{obc->obs.oi.soid}, OI_ATTR, osv);
+    }
+
+    // snapset
+    if (obc->obs.oi.soid.snap == CEPH_NOSNAP) {
+      logger().debug("final snapset {} in {}",
+        obc->ssc->snapset, obc->obs.oi.soid);
+      ceph::bufferlist bss;
+      encode(obc->ssc->snapset, bss);
+      txn.setattr(coll_ref->get_cid(), ghobject_t{obc->obs.oi.soid}, SS_ATTR, bss);
+      obc->ssc->exists = true;
+    } else {
+      logger().debug("no snapset (this is a clone)");
+    }
+  } else {
+    // reset cached ObjectState without enforcing eviction
+    obc->obs.oi = object_info_t(obc->obs.oi.soid);
+  }
+}
+
 std::tuple<PG::interruptible_future<>,
            PG::interruptible_future<>>
 PG::submit_transaction(
@@ -981,6 +1020,7 @@ PG::do_osd_ops_execute(
 	logger().debug(
 	  "do_osd_ops_execute: object {} submitting txn",
 	  obc->get_oid());
+        mutate_object(obc, txn, osd_op_p);
 	return submit_transaction(
           std::move(obc),
           std::move(txn),

src/crimson/osd/pg.h

@@ -848,6 +848,10 @@ class PG : public boost::intrusive_ref_counter<
   friend class SnapTrimObjSubEvent;
 private:
 
+  void mutate_object(
+    ObjectContextRef& obc,
+    ceph::os::Transaction& txn,
+    osd_op_params_t& osd_op_p);
   bool can_discard_replica_op(const Message& m, epoch_t m_map_epoch) const;
   bool can_discard_op(const MOSDOp& m) const;
   void context_registry_on_change();

src/crimson/osd/pg_backend.cc

@@ -162,38 +162,6 @@ PGBackend::mutate_object(
   std::vector<pg_log_entry_t>&& log_entries)
 {
   logger().trace("mutate_object: num_ops={}", txn.get_num_ops());
-  if (obc->obs.exists) {
-    obc->obs.oi.prior_version = obc->obs.oi.version;
-    obc->obs.oi.version = osd_op_p.at_version;
-    if (osd_op_p.user_modify)
-      obc->obs.oi.user_version = osd_op_p.at_version.version;
-    obc->obs.oi.last_reqid = osd_op_p.req_id;
-    obc->obs.oi.mtime = osd_op_p.mtime;
-    obc->obs.oi.local_mtime = ceph_clock_now();
-
-    // object_info_t
-    {
-      ceph::bufferlist osv;
-      obc->obs.oi.encode_no_oid(osv, CEPH_FEATURES_ALL);
-      // TODO: get_osdmap()->get_features(CEPH_ENTITY_TYPE_OSD, nullptr));
-      txn.setattr(coll->get_cid(), ghobject_t{obc->obs.oi.soid}, OI_ATTR, osv);
-    }
-
-    // snapset
-    if (obc->obs.oi.soid.snap == CEPH_NOSNAP) {
-      logger().debug("final snapset {} in {}",
-        obc->ssc->snapset, obc->obs.oi.soid);
-      ceph::bufferlist bss;
-      encode(obc->ssc->snapset, bss);
-      txn.setattr(coll->get_cid(), ghobject_t{obc->obs.oi.soid}, SS_ATTR, bss);
-      obc->ssc->exists = true;
-    } else {
-      logger().debug("no snapset (this is a clone)");
-    }
-  } else {
-    // reset cached ObjectState without enforcing eviction
-    obc->obs.oi = object_info_t(obc->obs.oi.soid);
-  }
   return _submit_transaction(
     std::move(pg_shards), obc->obs.oi.soid, std::move(txn),
     std::move(osd_op_p), min_epoch, map_epoch, std::move(log_entries));