CORTX-33652: Add min_success parameter for index operations (#1991)

Problem:
This PR adds a per-operation setting for the minimum number of successful CAS operations for distributed 
index operations. Initially, all CAS operations were required to succeed for the operation as a whole to 
succeed. A previous commit (f2ba0d1) changed the behavior to reduce IO errors during degraded mode. 
Currently, a single successful operation is sufficient to consider the parent operation successful. This can 
lead to consistency issues, however. In the read-after-write case, stale data may be returned from successful
index operations if the child operations succeed on disjoint sets of CAS services.
	
Solution:
This PR changes adds a set sockopt-like function (m0_idx_op_setoption) for tuning parameters of index 
operations. The only option initially available is M0_DIX_MIN_REPLICA_QUORUM. This defaults to (N+K)/2 + 1 
to prevent the situation above (disjoint sets of CAS services). Clients may choose a more lax requirement if they 
care more about availability than consistency. Clients (RGW) may want to add configuration options for controlling 
this quorum requirement, or could use this functionality as part of S3 storage tiers, e.g. setting a bucket to use 
reduced consistency operations. Note that ensuring correctness will also require a mechanism to compare versions 
of replies from CAS services, which is not in the scope of this PR.

Follows is taken care of:
 
* Fail dix operations that don't meet the specified min_success
* Update unit test to check dix min_success
* Add m0_idx_op_setoption with M0_OIO_MIN_SUCCESS
* More detailed comment on min_success handling
* Use M0_DIX_MIN_REPLICA_QUORUM by default for dix operations
* Fix variable alignment
* Make dix_item_version_cmp consistent with rest of codebase
* Only send a single CAS GET for meta requests
* Handle M0_DIX_MIN_REPLICA_QUORUM in dix_rop_ctx_init

Signed-off-by: Tim Shaffer <[email protected]>

Author: Tim Shaffer

dix/req.c

@@ -46,6 +46,7 @@
 #include "dix/fid_convert.h"
 #include "dix/dix_addb.h"
 #include "dtm0/dtx.h"   /* m0_dtx0_* API */
+#include "motr/idx.h" /* M0_DIX_MIN_REPLICA_QUORUM */
 
 static struct m0_sm_state_descr dix_req_states[] = {
 	[DIXREQ_INIT] = {
@@ -209,11 +210,15 @@ M0_INTERNAL int m0_dix_req_wait(struct m0_dix_req *req, uint64_t states,
 static void dix_req_init(struct m0_dix_req  *req,
 			 struct m0_dix_cli  *cli,
 			 struct m0_sm_group *grp,
+			 int64_t             min_success,
 			 bool                meta)
 {
+	M0_PRE(ergo(min_success < 1,
+		    min_success == M0_DIX_MIN_REPLICA_QUORUM));
 	M0_SET0(req);
 	req->dr_cli = cli;
 	req->dr_is_meta = meta;
+	req->dr_min_success = min_success;
 	m0_sm_init(&req->dr_sm, &dix_req_sm_conf, DIXREQ_INIT, grp);
 	m0_sm_addb2_counter_init(&req->dr_sm);
 }
@@ -222,14 +227,15 @@ M0_INTERNAL void m0_dix_mreq_init(struct m0_dix_req  *req,
 				  struct m0_dix_cli  *cli,
 				  struct m0_sm_group *grp)
 {
-	dix_req_init(req, cli, grp, true);
+	dix_req_init(req, cli, grp, 1, true);
 }
 
 M0_INTERNAL void m0_dix_req_init(struct m0_dix_req  *req,
 				 struct m0_dix_cli  *cli,
-				 struct m0_sm_group *grp)
+				 struct m0_sm_group *grp,
+				 int64_t             min_success)
 {
-	dix_req_init(req, cli, grp, false);
+	dix_req_init(req, cli, grp, min_success, false);
 }
 
 static enum m0_dix_req_state dix_req_state(const struct m0_dix_req *req)
@@ -1304,12 +1310,13 @@ static int dix_rop_ctx_init(struct m0_dix_req      *req,
 			    const struct m0_bufvec *keys,
 			    uint64_t               *indices)
 {
-	struct m0_dix       *dix = &req->dr_indices[0];
-	struct m0_dix_ldesc *ldesc;
-	uint32_t             keys_nr;
-	struct m0_buf        key;
-	uint32_t             i;
-	int                  rc = 0;
+	struct m0_dix          *dix = &req->dr_indices[0];
+	struct m0_dix_ldesc    *ldesc;
+	struct m0_pool_version *pver;
+	uint32_t                keys_nr;
+	struct m0_buf           key;
+	uint32_t                i;
+	int                     rc = 0;
 
 	M0_ENTRY();
 	M0_PRE(M0_IS0(rop));
@@ -1320,6 +1327,13 @@ static int dix_rop_ctx_init(struct m0_dix_req      *req,
 	M0_PRE(keys_nr != 0);
 	ldesc = &dix->dd_layout.u.dl_desc;
 	rop->dg_pver = dix_pver_find(req, &ldesc->ld_pver);
+	M0_ASSERT(ergo(req->dr_min_success < 1,
+		       req->dr_min_success == M0_DIX_MIN_REPLICA_QUORUM));
+	if (req->dr_min_success == M0_DIX_MIN_REPLICA_QUORUM) {
+		pver = m0_dix_pver(req->dr_cli, &req->dr_indices[0]);
+		req->dr_min_success = (pver->pv_attr.pa_N +
+				       pver->pv_attr.pa_K)/2 + 1;
+	}
 	M0_ALLOC_ARR(rop->dg_rec_ops, keys_nr);
 	M0_ALLOC_ARR(rop->dg_target_rop, rop->dg_pver->pv_attr.pa_P);
 	if (rop->dg_rec_ops == NULL || rop->dg_target_rop == NULL)
@@ -1402,6 +1416,20 @@ static void dix_rop(struct m0_dix_req *req)
 	M0_LEAVE();
 }
 
+/** Checks if the given cas get reply has a newer version of the value */
+static int dix_item_version_cmp(const struct m0_dix_item *ditem,
+				const struct m0_cas_get_reply *get_rep) {
+	/*
+	 * TODO: once cas versions are propagated, check if the get reply
+	 * has a newer version than seen previously. Will need to add
+	 * version info to struct m0_dix_item. This function should return
+	 * true if no previous value is set, or if the previous value has
+	 * an older version. For now, always return true so the last
+	 * reply in the array wins.
+	 */
+	return -1;
+}
+
 static void dix_item_rc_update(struct m0_dix_req  *req,
 			       struct m0_cas_req  *creq,
 			       uint64_t            key_idx,
@@ -1418,7 +1446,8 @@ static void dix_item_rc_update(struct m0_dix_req  *req,
 		case DIX_GET:
 			m0_cas_get_rep(creq, key_idx, &get_rep);
 			rc = get_rep.cge_rc;
-			if (rc == 0) {
+			if (rc == 0 && dix_item_version_cmp(ditem, &get_rep) < 0) {
+				m0_buf_free(&ditem->dxi_val);
 				ditem->dxi_val = get_rep.cge_val;
 				/* Value will be freed at m0_dix_req_fini(). */
 				m0_cas_rep_mlock(creq, key_idx);
@@ -1620,29 +1649,61 @@ static void dix_cas_rop_rc_update(struct m0_dix_cas_rop *cas_rop, int rc)
 
 static void dix_rop_completed(struct m0_sm_group *grp, struct m0_sm_ast *ast)
 {
-	struct m0_dix_req     *req = ast->sa_datum;
-	struct m0_dix_rop_ctx *rop = req->dr_rop;
-	struct m0_dix_rop_ctx *rop_del_phase2 = NULL;
-	bool                   del_phase2 = false;
-	struct m0_dix_cas_rop *cas_rop;
+	struct m0_dix_req      *req = ast->sa_datum;
+	struct m0_dix_rop_ctx  *rop = req->dr_rop;
+	struct m0_dix_rop_ctx  *rop_del_phase2 = NULL;
+	bool                    del_phase2 = false;
+	struct m0_dix_cas_rop  *cas_rop;
+	int64_t                 min_success;
+	int64_t                 successful_ops = 0;
 
 	(void)grp;
 	if (req->dr_type == DIX_NEXT)
 		m0_dix_next_result_prepare(req);
 	else {
+		min_success = req->dr_min_success;
+		M0_ASSERT(min_success > 0);
+
+		successful_ops = m0_tl_reduce(cas_rop, scan, &rop->dg_cas_reqs, 0,
+				  + !!(scan->crp_creq.ccr_sm.sm_rc == 0));
+
 		/*
-		 * Consider DIX request to be successful if there is at least
-		 * one successful CAS request.
+		 * The idea here is that transient failures are likely to
+		 * occur and may not persist long enough that the node gets
+		 * marked as failed. These will still affect individual
+		 * operations, so we need to make sure that dix correctly
+		 * handles the issues (if possible) or returns a failure to
+		 * the client. We therefore let the user choose min_success,
+		 * which determines the minimum number of successful cas
+		 * operations to consider the parent dix operation successful.
+		 * This is necessary to ensure read-after-write consistency.
+		 * If min_success is set to (N+K)/2 + 1 for both reads and
+		 * writes, then even in the presence of transient failures at
+		 * least one copy of the most recent version of data will be
+		 * found. Other values can be set for reduced consistency or
+		 * balancing read vs. write.
+		 *
+		 * Here we compare the previously computed successful_ops
+		 * and min_success to decide if we can ignore failed cas
+		 * operations. If successful_ops >= min_success, we've met
+		 * the quorum requirement and can ignore failures. This is
+		 * done by skipping dix_cas_rop_rc_update for failed cas
+		 * operations. We're guaranteed to have at least one
+		 * successful cas op somewhere in the list, so this results
+		 * in the parent dix operation being considered a success,
+		 * and cas version is used to break ties between multiple
+		 * successful replies (see dix_item_version_cmp). In the
+		 * case that successful_ops < min_success, we call
+		 * dix_cas_rop_rc_update for every cas op, with the result
+		 * that the failed operations will cause the parent dix op
+		 * to fail. Since min_success must be greater than 0, this
+		 * covers the case that all cas requests fail.
 		 */
-		if (m0_tl_forall(cas_rop, cas_rop,
-				 &rop->dg_cas_reqs,
-				 cas_rop->crp_creq.ccr_sm.sm_rc != 0))
-			    dix_cas_rop_rc_update(cas_rop_tlist_tail(
-						  &rop->dg_cas_reqs), 0);
-
 		m0_tl_for (cas_rop, &rop->dg_cas_reqs, cas_rop) {
-			if (cas_rop->crp_creq.ccr_sm.sm_rc == 0)
+			if (successful_ops < min_success ||
+			    cas_rop->crp_creq.ccr_sm.sm_rc == 0) {
 				dix_cas_rop_rc_update(cas_rop, 0);
+			}
 			m0_cas_req_fini(&cas_rop->crp_creq);
 		} m0_tl_endfor;
 	}
@@ -2137,10 +2198,11 @@ static void dix_rop_units_set(struct m0_dix_req *req)
 	m0_rwlock_read_unlock(&pm->pm_lock);
 
 	/*
-	 * Only one CAS GET request should be sent for every record.
+	 * For meta requests,
+	 * only one CAS GET request should be sent for every record.
 	 * Choose the best destination for every record.
 	 */
-	if (req->dr_type == DIX_GET) {
+	if (req->dr_type == DIX_GET && req->dr_is_meta) {
 		for (i = 0; i < rop->dg_rec_ops_nr; i++)
 			dix_online_unit_choose(req, &rop->dg_rec_ops[i]);
 	}

dix/req.h

@@ -253,6 +253,11 @@ struct m0_dix_req {
 	 * starting key in DIX_NEXT request.
 	 */
 	uint32_t                     *dr_recs_nr;
+	/**
+	 * Minimum number of successful CAS operations to treat
+	 * parent DIX operation as successful.
+	 */
+	int64_t                       dr_min_success;
 	/** Request flags bitmask of m0_cas_op_flags values. */
 	uint32_t                      dr_flags;
 
@@ -283,7 +288,8 @@ struct m0_dix_next_reply {
 /** Initialises DIX request. */
 M0_INTERNAL void m0_dix_req_init(struct m0_dix_req  *req,
 				 struct m0_dix_cli  *cli,
-				 struct m0_sm_group *grp);
+				 struct m0_sm_group *grp,
+				 int64_t             min_success);
 
 /**
  * Initialises DIX request operating with meta-indices.

dix/ut/client_ut.c

@@ -1244,7 +1244,7 @@ static int dix_common_idx_flagged_op(const struct m0_dix *indices,
 	int               rc;
 	int               i;
 
-	m0_dix_req_init(&req, &dix_ut_cctx.cl_cli, dix_ut_cctx.cl_grp);
+	m0_dix_req_init(&req, &dix_ut_cctx.cl_cli, dix_ut_cctx.cl_grp, 1);
 	m0_dix_req_lock(&req);
 	switch (type) {
 	case REQ_CREATE:
@@ -1285,6 +1285,7 @@ static int dix_common_rec_op(const struct m0_dix    *index,
 			     const struct m0_bufvec *keys,
 			     struct m0_bufvec       *vals,
 			     const uint32_t         *recs_nr,
+			     int64_t                 min_success,
 			     uint32_t                flags,
 			     struct dix_rep_arr     *rep,
 			     enum ut_dix_req_type    type)
@@ -1294,7 +1295,7 @@ static int dix_common_rec_op(const struct m0_dix    *index,
 	int               i;
 	int               k = 0;
 
-	m0_dix_req_init(&req, &dix_ut_cctx.cl_cli, dix_ut_cctx.cl_grp);
+	m0_dix_req_init(&req, &dix_ut_cctx.cl_cli, dix_ut_cctx.cl_grp, min_success);
 	m0_dix_req_lock(&req);
 	switch (type) {
 	case REQ_PUT:
@@ -1397,21 +1398,31 @@ static int dix_ut_put(const struct m0_dix    *index,
 		      uint32_t                flags,
 		      struct dix_rep_arr     *rep)
 {
-	return dix_common_rec_op(index, keys, vals, NULL, flags, rep, REQ_PUT);
+	return dix_common_rec_op(index, keys, vals, NULL, 1, flags, rep, REQ_PUT);
+}
+
+static int dix_ut_put_min_success(const struct m0_dix    *index,
+		      const struct m0_bufvec *keys,
+		      struct m0_bufvec       *vals,
+		      int64_t                 min_success,
+		      uint32_t                flags,
+		      struct dix_rep_arr     *rep)
+{
+	return dix_common_rec_op(index, keys, vals, NULL, min_success, flags, rep, REQ_PUT);
 }
 
 static int dix_ut_get(const struct m0_dix    *index,
 		      const struct m0_bufvec *keys,
 		      struct dix_rep_arr     *rep)
 {
-	return dix_common_rec_op(index, keys, NULL, NULL, 0, rep, REQ_GET);
+	return dix_common_rec_op(index, keys, NULL, NULL, 1, 0, rep, REQ_GET);
 }
 
 static int dix_ut_del(const struct m0_dix    *index,
 		      const struct m0_bufvec *keys,
 		      struct dix_rep_arr     *rep)
 {
-	return dix_common_rec_op(index, keys, NULL, NULL, 0, rep, REQ_DEL);
+	return dix_common_rec_op(index, keys, NULL, NULL, 1, 0, rep, REQ_DEL);
 }
 
 static int dix_ut_next(const struct m0_dix    *index,
@@ -1420,7 +1431,7 @@ static int dix_ut_next(const struct m0_dix    *index,
 		       uint32_t                flags,
 		       struct dix_rep_arr     *rep)
 {
-	return dix_common_rec_op(index, start_keys, NULL, recs_nr, flags,
+	return dix_common_rec_op(index, start_keys, NULL, recs_nr, 1, flags,
 				 rep, REQ_NEXT);
 }
 
@@ -2655,17 +2666,35 @@ static void local_failures(void)
 	dix_kv_alloc_and_fill(&keys, &vals, COUNT);
 	rc = dix_common_idx_op(&index, 1, REQ_CREATE);
 	M0_UT_ASSERT(rc == 0);
+
 	/*
- 	 * Consider DIX request to be successful if there is at least
- 	 * one successful CAS request. Here two cas requests can be 
- 	 * sent successfully. 
+	 * Consider DIX request to be successful only if there are
+	 * enough successful CAS requests to satisfy min_success.
+	 * Here two cas requests can be sent successfully. First, try with
+	 * min_success = 3, which should result in all CAS requests failing.
 	 */
 	m0_fi_enable_off_n_on_m("cas_req_replied_cb", "send-failure", 2, 3);
-	rc = dix_ut_put(&index, &keys, &vals, 0, &rep);
+	rc = dix_ut_put_min_success(&index, &keys, &vals, 3, 0, &rep);
+	m0_fi_disable("cas_req_replied_cb", "send-failure");
+	M0_UT_ASSERT(rc == 0);
+	M0_UT_ASSERT(rep.dra_nr == COUNT);
+	M0_UT_ASSERT(m0_forall(i, COUNT, rep.dra_rep[i].dre_rc == -ENOTCONN));
+
+	dix_rep_free(&rep);
+	rc = dix_ut_del(&index, &keys, &rep);
+	M0_UT_ASSERT(rc == 0);
+	dix_rep_free(&rep);
+
+	/*
+	 * Now try again with min_success = 2, which should succeed.
+	 */
+	m0_fi_enable_off_n_on_m("cas_req_replied_cb", "send-failure", 2, 3);
+	rc = dix_ut_put_min_success(&index, &keys, &vals, 2, 0, &rep);
 	m0_fi_disable("cas_req_replied_cb", "send-failure");
 	M0_UT_ASSERT(rc == 0);
 	M0_UT_ASSERT(rep.dra_nr == COUNT);
 	M0_UT_ASSERT(m0_forall(i, COUNT, rep.dra_rep[i].dre_rc == 0));
+
 	dix_rep_free(&rep);
 	dix_kv_destroy(&keys, &vals);
 	dix_index_fini(&index);

motr/client.h

@@ -568,6 +568,24 @@ enum m0_idx_opcode {
 	M0_IC_NR                   /* 21 */
 } M0_XCA_ENUM;
 
+/**
+ * Option to set on an index operation.
+ *
+ * See @ref m0_idx_op_setoption.
+ */
+enum m0_op_idx_option {
+	/**
+	 * Index operations are distributed across one or more CAS services
+	 * to achieve the configured durability. This option specifies the
+	 * minimum number of services that must successfully complete an
+	 * operation for the operation as a whole to be considered
+	 * successful. The value for this option must be greater than zero
+	 * or a special value. This should normally be set to
+	 * @ref M0_DIX_MIN_REPLICA_QUORUM to ensure consistency.
+	 */
+	M0_OIO_MIN_SUCCESS = 1,
+};
+
 /**
  * Flags passed to m0_obj_op() to specify object IO operation behaviour.
  */
@@ -1628,6 +1646,18 @@ int m0_idx_op(struct m0_idx       *idx,
 	      uint32_t             flags,
 	      struct m0_op       **op);
 
+/**
+ * Set an option on an index operation.
+ *
+ * The index op must have been previously initialized with @ref m0_idx_op.
+ * It is undefined behavior to change an option after the index op has
+ * been launched. The meaning of each option is documented in
+ * @ref m0_op_idx_option.
+ */
+void m0_idx_op_setoption(struct m0_op *op,
+			 enum m0_op_idx_option option,
+			 int64_t value);
+
 void m0_realm_create(struct m0_realm    *realm,
 		     uint64_t wcount, uint64_t rcount,
 		     struct m0_op **op);

motr/client_internal.h

@@ -230,6 +230,12 @@ struct m0_op_idx {
 	struct m0_sm_group  *oi_sm_grp;
 	struct m0_ast_rc     oi_ar;
 
+	/**
+	 * Minimum number of successful CAS operations to treat
+	 * parent DIX operation as successful.
+	 */
+	int64_t              oi_min_success;
+
 	/* A bit-mask of m0_op_idx_flags. */
 	uint32_t             oi_flags;