objecter: use read policy configurable

for localized and balanced reads.

Signed-off-by: Yehuda Sadeh <ysadehwe@ibm.com>

diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index 087b623333bb7e8ada68eabc51f04d46f8cf1b1a..367956cb3f555128e00f4d6808de9d8e05a0aa1e 100644 (file)
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -242,6 +242,13 @@ void Objecter::handle_conf_change(const ConfigProxy& conf,
   if (changed.count("rados_osd_op_timeout")) {
     osd_timeout = conf.get_val<std::chrono::seconds>("rados_osd_op_timeout");
   }
+
+  auto read_policy = conf.get_val<std::string>("rados_replica_read_policy");
+  if (read_policy == "localize") {
+    extra_read_flags = CEPH_OSD_FLAG_LOCALIZE_READS;
+  } else if (read_policy == "balance") {
+    extra_read_flags = CEPH_OSD_FLAG_BALANCE_READS;
+  }
 }
 
 void Objecter::update_crush_location()
@@ -5111,6 +5118,13 @@ Objecter::Objecter(CephContext *cct,
 {
   mon_timeout = cct->_conf.get_val<std::chrono::seconds>("rados_mon_op_timeout");
   osd_timeout = cct->_conf.get_val<std::chrono::seconds>("rados_osd_op_timeout");
+
+  auto read_policy = cct->_conf.get_val<std::string>("rados_replica_read_policy");
+  if (read_policy == "localize") {
+    extra_read_flags = CEPH_OSD_FLAG_LOCALIZE_READS;
+  } else if (read_policy == "balance") {
+    extra_read_flags = CEPH_OSD_FLAG_BALANCE_READS;
+  }
 }
 
 Objecter::~Objecter()

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 927c7e413296ffc9bef72900c7bfffb2fb253397..2ed8c3f27db036fb00547803dd835f4f3ffd4a3a 100644 (file)
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -1723,6 +1723,8 @@ private:
   bool keep_balanced_budget = false;
   bool honor_pool_full = true;
 
+  std::atomic<int> extra_read_flags{0};
+
   // If this is true, accumulate a set of blocklisted entities
   // to be drained by consume_blocklist_events.
   bool blocklist_events_enabled = false;
@@ -3089,7 +3091,7 @@ public:
     int *data_offset = NULL,
     uint64_t features = 0,
     ZTracer::Trace *parent_trace = nullptr) {
-    Op *o = new Op(oid, oloc, std::move(op.ops), flags | global_op_flags |
+    Op *o = new Op(oid, oloc, std::move(op.ops), flags | global_op_flags | extra_read_flags |
                   CEPH_OSD_FLAG_READ, onack, objver,
                   data_offset, parent_trace);
     o->priority = op.priority;
@@ -3125,7 +3127,7 @@ public:
            int flags, Op::OpComp onack,
            version_t *objver = nullptr, int *data_offset = nullptr,
            uint64_t features = 0, ZTracer::Trace *parent_trace = nullptr) {
-    Op *o = new Op(oid, oloc, std::move(op.ops), flags | global_op_flags |
+    Op *o = new Op(oid, oloc, std::move(op.ops), flags | global_op_flags | extra_read_flags |
                   CEPH_OSD_FLAG_READ, std::move(onack), objver,
                   data_offset, parent_trace);
     o->priority = op.priority;
@@ -3164,7 +3166,7 @@ public:
     int *ctx_budget) {
     Op *o = new Op(object_t(), oloc,
                   std::move(op.ops),
-                  flags | global_op_flags | CEPH_OSD_FLAG_READ |
+                  flags | global_op_flags | CEPH_OSD_FLAG_READ | extra_read_flags |
                   CEPH_OSD_FLAG_IGNORE_OVERLAY,
                   onack, NULL);
     o->target.precalc_pgid = true;
@@ -3305,7 +3307,7 @@ public:
     ops[i].op.op = CEPH_OSD_OP_STAT;
     C_Stat *fin = new C_Stat(psize, pmtime, onfinish);
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, fin, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, fin, objver);
     o->snapid = snap;
     o->outbl = &fin->bl;
     return o;
@@ -3337,7 +3339,7 @@ public:
     ops[i].op.extent.truncate_seq = 0;
     ops[i].op.flags = op_flags;
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, onfinish, objver,
+                  CEPH_OSD_FLAG_READ | extra_read_flags, onfinish, objver,
                   nullptr, parent_trace);
     o->snapid = snap;
     o->outbl = pbl;
@@ -3370,7 +3372,7 @@ public:
     ops[i].indata = cmp_bl;
     ops[i].op.flags = op_flags;
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, onfinish, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, onfinish, objver);
     o->snapid = snap;
     return o;
   }
@@ -3402,7 +3404,7 @@ public:
     ops[i].op.extent.truncate_seq = trunc_seq;
     ops[i].op.flags = op_flags;
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, onfinish, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, onfinish, objver);
     o->snapid = snap;
     o->outbl = pbl;
     ceph_tid_t tid;
@@ -3421,7 +3423,7 @@ public:
     ops[i].op.extent.truncate_size = 0;
     ops[i].op.extent.truncate_seq = 0;
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, onfinish, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, onfinish, objver);
     o->snapid = snap;
     o->outbl = pbl;
     ceph_tid_t tid;
@@ -3440,7 +3442,7 @@ public:
     if (name)
       ops[i].indata.append(name, ops[i].op.xattr.name_len);
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, onfinish, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, onfinish, objver);
     o->snapid = snap;
     o->outbl = pbl;
     ceph_tid_t tid;
@@ -3457,7 +3459,7 @@ public:
     ops[i].op.op = CEPH_OSD_OP_GETXATTRS;
     C_GetAttrs *fin = new C_GetAttrs(attrset, onfinish);
     Op *o = new Op(oid, oloc, std::move(ops), flags | global_op_flags |
-                  CEPH_OSD_FLAG_READ, fin, objver);
+                  CEPH_OSD_FLAG_READ | extra_read_flags, fin, objver);
     o->snapid = snap;
     o->outbl = &fin->bl;
     ceph_tid_t tid;