os/Transaction: page align write data buffers to improve performance

Align write data in Objectstore::Transaction encoded buffers so that
RepOp and ECSubOpWrite messages align data on the receiving OSD to
avoid a later memmove. Also fix ECSubOpReadReply messages in a
similar way so that read data is aligned on the receiving OSD.

Signed-off-by: Radoslaw Zarzynski <[email protected]>
Signed-off-by: Bill Scales <[email protected]>

Author: bill-scales

src/crimson/osd/ops_executer.cc

@@ -1102,17 +1102,18 @@ void OpsExecuter::apply_stats()
   pg->apply_stats(get_target(), delta_stats);
 }
 
-OpsExecuter::OpsExecuter(Ref<PG> pg,
+OpsExecuter::OpsExecuter(Ref<PG> _pg,
                          ObjectContextRef _obc,
                          const OpInfo& op_info,
                          abstracted_msg_t&& msg,
                          crimson::net::ConnectionXcoreRef conn,
                          const SnapContext& _snapc)
-  : pg(std::move(pg)),
+  : pg(std::move(_pg)),
     obc(std::move(_obc)),
     op_info(op_info),
     msg(std::move(msg)),
     conn(conn),
+    txn(pg->min_peer_features()),
     snapc(_snapc)
 {
   if (op_info.may_write() && should_clone(*obc, snapc)) {

src/crimson/osd/pg.cc

@@ -1221,8 +1221,10 @@ PG::handle_rep_op_fut PG::handle_rep_op(Ref<MOSDRepOp> req)
   DEBUGDPP("{}", *this, *req);
 
   ceph::os::Transaction txn;
-  auto encoded_txn = req->get_data().cbegin();
-  decode(txn, encoded_txn);
+  auto encoded_txn_p = req->get_middle().cbegin();
+  auto encoded_txn_d = req->get_data().cbegin();
+  txn.decode(req->get_middle().length() != 0 ? encoded_txn_p : encoded_txn_d,
+             encoded_txn_d);
   auto p = req->logbl.cbegin();
   std::vector<pg_log_entry_t> log_entries;
   decode(log_entries, p);

src/crimson/osd/replicated_backend.cc

@@ -39,7 +39,8 @@ ReplicatedBackend::_read(const hobject_t& hoid,
 MURef<MOSDRepOp> ReplicatedBackend::new_repop_msg(
   const pg_shard_t &pg_shard,
   const hobject_t &hoid,
-  const bufferlist &encoded_txn,
+  bufferlist &encoded_txn_p_bl,
+  bufferlist &encoded_txn_d_bl,
   const osd_op_params_t &osd_op_p,
   epoch_t min_epoch,
   epoch_t map_epoch,
@@ -59,7 +60,13 @@ MURef<MOSDRepOp> ReplicatedBackend::new_repop_msg(
     tid,
     osd_op_p.at_version);
   if (send_op) {
-    m->set_data(encoded_txn);
+    if (encoded_txn_d_bl.length() != 0) {
+      m->set_txn_payload(encoded_txn_p_bl);
+      m->set_data(encoded_txn_d_bl);
+    } else {
+      // Pre-tentacle format - everything in data
+      m->set_data(encoded_txn_p_bl);
+    }
   } else {
     ceph::os::Transaction t;
     bufferlist bl;
@@ -97,8 +104,8 @@ ReplicatedBackend::submit_transaction(
       pg_shards.size(),
       osd_op_p.at_version,
       pg.get_last_complete()).first;
-  bufferlist encoded_txn;
-  encode(txn, encoded_txn);
+  bufferlist encoded_txn_p_bl, encoded_txn_d_bl;
+  txn.encode(encoded_txn_p_bl, encoded_txn_d_bl, pg.min_peer_features());
 
   bool is_delete = false;
   for (auto &le : log_entries) {
@@ -120,11 +127,11 @@ ReplicatedBackend::submit_transaction(
     MURef<MOSDRepOp> m;
     if (pg.should_send_op(pg_shard, hoid)) {
       m = new_repop_msg(
-	pg_shard, hoid, encoded_txn, osd_op_p,
+	pg_shard, hoid, encoded_txn_p_bl, encoded_txn_d_bl, osd_op_p,
 	min_epoch, map_epoch, log_entries, true, tid);
     } else {
       m = new_repop_msg(
-	pg_shard, hoid, encoded_txn, osd_op_p,
+	pg_shard, hoid, encoded_txn_p_bl, encoded_txn_d_bl, osd_op_p,
 	min_epoch, map_epoch, log_entries, false, tid);
       if (pg.is_missing_on_peer(pg_shard, hoid)) {
 	if (_new_clone) {

src/crimson/osd/replicated_backend.h

@@ -70,7 +70,8 @@ class ReplicatedBackend : public PGBackend
   MURef<MOSDRepOp> new_repop_msg(
     const pg_shard_t &pg_shard,
     const hobject_t &hoid,
-    const bufferlist &encoded_txn,
+    bufferlist &encoded_txn_p_bl,
+    bufferlist &encoded_txn_d_bl,
     const osd_op_params_t &osd_op_p,
     epoch_t min_epoch,
     epoch_t map_epoch,

src/messages/MOSDECSubOpReadReply.h

@@ -48,9 +48,10 @@ class MOSDECSubOpReadReply : public MOSDFastDispatchOp {
   void decode_payload() override {
     using ceph::decode;
     auto p = payload.cbegin();
+    auto d = data.cbegin();
     decode(pgid, p);
     decode(map_epoch, p);
-    decode(op, p);
+    op.decode(p, d);
     if (header.version >= 2) {
       decode(min_epoch, p);
       decode_trace(p);
@@ -63,7 +64,7 @@ class MOSDECSubOpReadReply : public MOSDFastDispatchOp {
     using ceph::encode;
     encode(pgid, payload);
     encode(map_epoch, payload);
-    encode(op, payload);
+    op.encode(payload, data, features);
     encode(min_epoch, payload);
     encode_trace(payload, features);
   }

src/messages/MOSDECSubOpWrite.h

@@ -52,9 +52,10 @@ class MOSDECSubOpWrite : public MOSDFastDispatchOp {
   void decode_payload() override {
     using ceph::decode;
     auto p = payload.cbegin();
+    auto d = data.cbegin();
     decode(pgid, p);
     decode(map_epoch, p);
-    decode(op, p);
+    op.decode(p, d);
     if (header.version >= 2) {
       decode(min_epoch, p);
       decode_trace(p);
@@ -67,7 +68,7 @@ class MOSDECSubOpWrite : public MOSDFastDispatchOp {
     using ceph::encode;
     encode(pgid, payload);
     encode(map_epoch, payload);
-    encode(op, payload);
+    op.encode(payload, data, features);
     encode(min_epoch, payload);
     encode_trace(payload, features);
   }

src/messages/MOSDRepOp.h

@@ -85,6 +85,8 @@ class MOSDRepOp final : public MOSDFastDispatchOp {
   /// non-empty if this transaction involves a hit_set history update
   std::optional<pg_hit_set_history_t> updated_hit_set_history;
 
+  bufferlist txn_payload;
+
   epoch_t get_map_epoch() const override {
     return map_epoch;
   }
@@ -99,6 +101,11 @@ class MOSDRepOp final : public MOSDFastDispatchOp {
     return data.length();
   }
 
+  void set_txn_payload(bufferlist bl)
+  {
+    txn_payload = bl;
+  }
+
   void decode_payload() override {
     using ceph::decode;
     p = payload.cbegin();
@@ -159,6 +166,8 @@ class MOSDRepOp final : public MOSDFastDispatchOp {
     encode(from, payload);
     encode(updated_hit_set_history, payload);
     encode(pg_committed_to, payload);
+    bufferlist middle(txn_payload);
+    set_middle(middle);
   }
 
   MOSDRepOp()

src/msg/Message.h

@@ -441,6 +441,7 @@ class Message : public RefCountedObject {
       byte_throttler->take(middle.length());
   }
   ceph::buffer::list& get_middle() { return middle; }
+  const ceph::buffer::list& get_middle() const { return middle; }
 
   void set_data(const ceph::buffer::list &bl) {
     if (byte_throttler)