Fail dix operations that don't meet the specified min_success

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

Author: tshaffe1

dix/req.c

@@ -209,11 +209,14 @@ 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,
+			 uint32_t            min_success,
 			 bool                meta)
 {
+	M0_PRE(min_success > 0);
 	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 +225,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,
+				 uint32_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)
@@ -1402,6 +1406,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 bool dix_item_version_cmp(struct m0_dix_item *ditem,
+				 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 true;
+}
+
 static void dix_item_rc_update(struct m0_dix_req  *req,
 			       struct m0_cas_req  *creq,
 			       uint64_t            key_idx,
@@ -1418,7 +1436,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)) {
+				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);
@@ -1625,24 +1644,28 @@ static void dix_rop_completed(struct m0_sm_group *grp, struct m0_sm_ast *ast)
 	struct m0_dix_rop_ctx *rop_del_phase2 = NULL;
 	bool                   del_phase2 = false;
 	struct m0_dix_cas_rop *cas_rop;
+	uint32_t               successful_ops = 0;
 
 	(void)grp;
 	if (req->dr_type == DIX_NEXT)
 		m0_dix_next_result_prepare(req);
 	else {
+		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.
+		 * If enough operations succeeded to satisfy min_success,
+		 * ignore any operations that failed. If we don't have enough,
+		 * count all operations when deciding overall success (ensures the
+		 * overall operation is considered a failure).
 		 */
-		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_ASSERT(req->dr_min_success > 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 < req->dr_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;
 	}
@@ -1956,29 +1979,6 @@ static int dix_spare_target_with_data(struct m0_dix_rec_op         *rec_op,
 				 true);
 }
 
-static void dix_online_unit_choose(struct m0_dix_req    *req,
-				   struct m0_dix_rec_op *rec_op)
-{
-	struct m0_dix_pg_unit *pgu;
-	uint64_t               start_unit;
-	uint64_t               i;
-	uint64_t               j;
-
-	M0_ENTRY();
-	M0_PRE(req->dr_type == DIX_GET);
-	start_unit = req->dr_items[rec_op->dgp_item].dxi_pg_unit;
-	M0_ASSERT(start_unit < dix_rec_op_spare_offset(rec_op));
-	for (i = 0; i < start_unit; i++)
-		rec_op->dgp_units[i].dpu_unavail = true;
-	for (i = start_unit; i < rec_op->dgp_units_nr; i++) {
-		pgu = &rec_op->dgp_units[i];
-		if (!pgu->dpu_is_spare && !pgu->dpu_unavail)
-			break;
-	}
-	for (j = i + 1; j < rec_op->dgp_units_nr; j++)
-		rec_op->dgp_units[j].dpu_unavail = true;
-}
-
 static void dix_pg_unit_pd_assign(struct m0_dix_pg_unit *pgu,
 				  struct m0_pooldev     *pd)
 {
@@ -2135,15 +2135,6 @@ 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.
-	 * Choose the best destination for every record.
-	 */
-	if (req->dr_type == DIX_GET) {
-		for (i = 0; i < rop->dg_rec_ops_nr; i++)
-			dix_online_unit_choose(req, &rop->dg_rec_ops[i]);
-	}
 }
 
 static bool dix_pg_unit_skip(struct m0_dix_req     *req,

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.
+	 */
+	uint32_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,
+				 uint32_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:
@@ -1294,7 +1294,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, 1);
 	m0_dix_req_lock(&req);
 	switch (type) {
 	case REQ_PUT:

motr/client_internal.h

@@ -229,6 +229,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.
+	 */
+	uint32_t             oi_min_success;
+
 	/* A bit-mask of m0_op_idx_flags. */
 	uint32_t             oi_flags;
 

motr/idx_dix.c

@@ -663,7 +663,7 @@ static int dix_req_create(struct m0_op_idx  *oi,
 		m0_clink_init(&req->idr_dtx_clink, dixreq_clink_dtx_cb);
 		if (idx_is_distributed(oi)) {
 			m0_dix_req_init(&req->idr_dreq, op_dixc(oi),
-					oi->oi_sm_grp);
+					oi->oi_sm_grp, oi->oi_min_success);
 			to_dix_map(&oi->oi_oc.oc_op, &req->idr_dreq);
 			req->idr_dreq.dr_dtx = oi->oi_dtx;
 			m0_clink_init(&req->idr_clink, dixreq_clink_cb);