Merge pull request ceph#65771 from aainscow/ec_direct_reads_pr_1

EC Direct Reads: First PR, background work

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

Author: aainscow

src/CMakeLists.txt

@@ -532,6 +532,7 @@ set(libcommon_files
   osdc/Striper.cc
   osdc/Objecter.cc
   osdc/error_code.cc
+  osdc/SplitOp.cc
   librbd/Features.cc
   librbd/io/IoOperations.cc
   ${mds_files})

src/crimson/osd/osd_operations/client_request.cc

@@ -209,8 +209,8 @@ ClientRequest::interruptible_future<> ClientRequest::with_pg_process_interruptib
       pg.get_perf_logger().inc(l_osd_replica_read_redirect_missing);
       co_await reply_op_error(pgref, -EAGAIN);
       co_return;
-    } else if (!pg.get_peering_state().can_serve_replica_read(m->get_hobj())) {
-      // Note: can_serve_replica_read checks for writes on the head object
+    } else if (!pg.get_peering_state().can_serve_read(m->get_hobj())) {
+      // Note: can_serve_read checks for writes on the head object
       //       as writes can only occur to head.
       DEBUGDPP("{}.{}: unstable write on replica, bouncing to primary",
 	       pg, *this, this_instance_id);

src/crimson/osd/pg.cc

@@ -1388,7 +1388,7 @@ void PG::log_operation(
 
   if (!is_primary()) { // && !is_ec_pg()
     DEBUGDPP("on replica, clearing obc", *this);
-    replica_clear_repop_obc(logv);
+    clear_repop_obc(logv);
   }
   if (!logv.empty()) {
     scrubber.on_log_update(logv.rbegin()->version);
@@ -1402,9 +1402,9 @@ void PG::log_operation(
                            false);
 }
 
-void PG::replica_clear_repop_obc(
+void PG::clear_repop_obc(
   const std::vector<pg_log_entry_t> &logv) {
-  LOG_PREFIX(PG::replica_clear_repop_obc);
+  LOG_PREFIX(PG::clear_repop_obc);
   DEBUGDPP("clearing obc for {} log entries", *this, logv.size());
   for (auto &&e: logv) {
     DEBUGDPP("clearing entry for {} from: {} to: {}",
@@ -1615,8 +1615,7 @@ bool PG::can_discard_op(const MOSDOp& m) const {
     return true;
   }
 
-  if ((m.get_flags() & (CEPH_OSD_FLAG_BALANCE_READS |
-                        CEPH_OSD_FLAG_LOCALIZE_READS))
+  if ((m.get_flags() & CEPH_OSD_FLAGS_DIRECT_READ)
     && !is_primary()
     && (m.get_map_epoch() <
         peering_state.get_info().history.same_interval_since))

src/crimson/osd/pg.h

@@ -675,7 +675,7 @@ class PG : public boost::intrusive_ref_counter<
     bool transaction_applied,
     ObjectStore::Transaction &txn,
     bool async = false);
-  void replica_clear_repop_obc(
+  void clear_repop_obc(
     const std::vector<pg_log_entry_t> &logv);
   void handle_rep_op_reply(const MOSDRepOpReply& m);
   interruptible_future<> do_update_log_missing(

src/erasure-code/ErasureCodeInterface.h

@@ -685,6 +685,11 @@ namespace ceph {
        * to decode a parity CRC to get the CRC of a data shard.
        */
       FLAG_EC_PLUGIN_CRC_ENCODE_DECODE_SUPPORT = 1<<7,
+      /* This plugin supports the ability for the client to read directly from
+       * the OSD containing a shard. This currently requires that raw shard ==
+       * shard and that the data shards are simply striped.
+       */
+      FLAG_EC_PLUGIN_DIRECT_READS = 1<<8,
     };
     static const char *get_optimization_flag_name(const plugin_flags flag) {
       switch (flag) {
@@ -697,6 +702,8 @@ namespace ceph {
       case FLAG_EC_PLUGIN_OPTIMIZED_SUPPORTED: return "optimizedsupport";
       case FLAG_EC_PLUGIN_CRC_ENCODE_DECODE_SUPPORT:
         return "crcencodedecode";
+      case FLAG_EC_PLUGIN_DIRECT_READS:
+        return "directreads";
       default: return "???";
       }
     }

src/erasure-code/isa/ErasureCodeIsa.h

@@ -69,7 +69,8 @@ class ErasureCodeIsa : public ceph::ErasureCode {
             FLAG_EC_PLUGIN_PARTIAL_READ_OPTIMIZATION |
             FLAG_EC_PLUGIN_PARTIAL_WRITE_OPTIMIZATION |
             FLAG_EC_PLUGIN_ZERO_INPUT_ZERO_OUTPUT_OPTIMIZATION |
-            FLAG_EC_PLUGIN_PARITY_DELTA_OPTIMIZATION;
+            FLAG_EC_PLUGIN_PARITY_DELTA_OPTIMIZATION |
+            FLAG_EC_PLUGIN_DIRECT_READS;
 
     if (technique == "reed_sol_van"sv) {
        flags |= FLAG_EC_PLUGIN_CRC_ENCODE_DECODE_SUPPORT;

src/erasure-code/jerasure/ErasureCodeJerasure.h

@@ -51,7 +51,8 @@ class ErasureCodeJerasure : public ceph::ErasureCode {
     flags = FLAG_EC_PLUGIN_PARTIAL_READ_OPTIMIZATION |
       FLAG_EC_PLUGIN_PARTIAL_WRITE_OPTIMIZATION |
       FLAG_EC_PLUGIN_ZERO_INPUT_ZERO_OUTPUT_OPTIMIZATION |
-      FLAG_EC_PLUGIN_PARITY_DELTA_OPTIMIZATION;
+      FLAG_EC_PLUGIN_PARITY_DELTA_OPTIMIZATION |
+      FLAG_EC_PLUGIN_DIRECT_READS;
 
     if (technique == "reed_sol_van"sv) {
       flags |= FLAG_EC_PLUGIN_OPTIMIZED_SUPPORTED;

src/include/rados.h

@@ -481,8 +481,12 @@ enum {
 	CEPH_OSD_FLAG_IGNORE_REDIRECT = 0x2000000,  /* ignore redirection */
 	CEPH_OSD_FLAG_RETURNVEC = 0x4000000, /* allow overall result >= 0, and return >= 0 and buffer for each op in opvec */
 	CEPH_OSD_FLAG_SUPPORTSPOOLEIO = 0x8000000,   /* client understands pool EIO flag */
+        CEPH_OSD_FLAG_EC_DIRECT_READ = 0x10000000,  /* Erasure code doing a partial read direct to OSD. */
 };
 
+// Indicates an IO which is direct-to-OSD and may not be on the primary.
+#define CEPH_OSD_FLAGS_DIRECT_READ (CEPH_OSD_FLAG_BALANCE_READS | CEPH_OSD_FLAG_LOCALIZE_READS | CEPH_OSD_FLAG_EC_DIRECT_READ)
+
 enum {
 	CEPH_OSD_OP_FLAG_EXCL = 0x1,      /* EXCL object create */
 	CEPH_OSD_OP_FLAG_FAILOK = 0x2,    /* continue despite failure */

src/messages/MOSDRepOp.h

@@ -65,7 +65,7 @@ class MOSDRepOp final : public MOSDFastDispatchOp {
    * Because updates <= pg_committed_to cannot become divergent, replicas
    * may safely serve reads on objects which do not have more recent updates.
    *
-   * See PeeringState::pg_committed_to, PeeringState::can_serve_replica_read
+   * See PeeringState::pg_committed_to, PeeringState::can_serve_read
    *
    * Historical note: Prior to early 2024, this field was named
    * min_last_complete_ondisk.  The replica, however, only actually relied on

src/mon/OSDMonitor.cc

@@ -8357,6 +8357,8 @@ int OSDMonitor::prepare_new_pool(string& name,
     enable_pool_ec_optimizations(*pi, nullptr, true);
   }
 
+  enable_pool_ec_direct_reads(*pi);
+
   pending_inc.new_pool_names[pool] = name;
   return 0;
 }
@@ -8451,6 +8453,29 @@ int OSDMonitor::enable_pool_ec_optimizations(pg_pool_t &p,
   return 0;
 }
 
+void OSDMonitor::enable_pool_ec_direct_reads(pg_pool_t &p) {
+  if (p.is_erasure()) {
+    ErasureCodeInterfaceRef erasure_code;
+    stringstream tmp;
+    int err = get_erasure_code(p.erasure_code_profile, &erasure_code, &tmp);
+
+    // Once this feature is finished, we will replace this with upgrade code.
+    // The upgrade code will enable the split read flag once all OSDs are at
+    // Umbrella. For now, if the plugin does not support direct reads, we just
+    // disable it.  All plugins and techniques should be capable of supporting
+    // direct reads, but we put in place this capability to reduce the test
+    // matrix for less important plugins/techniques.
+    //
+    // To enable direct reads in development, set the osd_pool_default_flags to
+    // 1<<20 = 0x100000 = 1048576
+    if (err != 0 || !p.allows_ecoptimizations() ||
+          (erasure_code->get_supported_optimizations() &
+            ErasureCodeInterface::FLAG_EC_PLUGIN_DIRECT_READS) == 0) {
+      p.flags &= ~pg_pool_t::FLAG_CLIENT_SPLIT_READS;
+    }
+  }
+}
+
 int OSDMonitor::prepare_command_pool_set(const cmdmap_t& cmdmap,
                                          stringstream& ss)
 {