DAOS-19429 rebuild: debug ec agg and rebuild

src/container/srv_target.c

@@ -63,10 +63,11 @@ agg_rate_ctl(void *arg)
 		return -1;
 
 	/*
-	 * XXX temporary workaround: EC aggregation needs to be paused during rebuilding
-	 * to avoid the race between EC rebuild and EC aggregation.
+	 * EC aggregation needs to be paused by the rebuild EC agg barrier to avoid
+	 * the race between EC rebuild and EC aggregation.
 	 **/
-	if (ds_pool_is_rebuilding(pool) && cont->sc_ec_agg_active && !param->ap_vos_agg)
+	if (!param->ap_vos_agg && cont->sc_ec_agg_active &&
+	    ds_pool_child_ec_agg_paused(cont->sc_pool))
 		return -1;
 
 	/* When system is idle or under space pressure, let aggregation run in tight mode */
@@ -207,10 +208,10 @@ cont_aggregate_runnable(struct ds_cont_child *cont, struct sched_request *req,
 		return false;
 	}
 
-	if (ds_pool_is_rebuilding(pool) && !vos_agg) {
-		D_DEBUG(DB_EPC, DF_CONT ": skip EC aggregation during rebuild %d, %d.\n",
+	if (ds_pool_child_ec_agg_paused(cont->sc_pool) && !vos_agg) {
+		D_DEBUG(DB_EPC, DF_CONT ": skip EC aggregation during pause gate %u.\n",
 			DP_CONT(cont->sc_pool->spc_uuid, cont->sc_uuid),
-			atomic_load(&pool->sp_rebuilding), atomic_load(&pool->sp_rebuild_enum));
+			atomic_load(&cont->sc_pool->spc_ec_agg_pause_gate));
 		return false;
 	}
 
@@ -511,10 +512,8 @@ cont_aggregate_interval(struct ds_cont_child *cont, cont_aggregate_cb_t cb,
 		if (dss_ult_exiting(req))
 			break;
 
-		/* sleep 18 seconds for EC aggregation ULT if the pool is in rebuilding,
-		 * if no space pressure.
-		 */
-		if (ds_pool_is_rebuilding(cont->sc_pool->spc_pool) && !param->ap_vos_agg &&
+		/* Sleep longer while the rebuild EC agg barrier is active if no space pressure. */
+		if (ds_pool_child_ec_agg_paused(cont->sc_pool) && !param->ap_vos_agg &&
 		    msecs != 200)
 			msecs = 18000;
 
@@ -964,11 +963,11 @@ ds_cont_child_reset_ec_agg_eph_all(struct ds_pool_child *pool_child)
 
 #define WAIT_EC_PAUSE_MAX 600
 
-void
+int
 ds_cont_child_wait_ec_agg_pause(struct ds_pool_child *pool_child, int wait_timeout)
 {
 	uint64_t start_time = daos_wallclock_secs();
-	int      wait_intv  = 10;
+	int      next_warn  = 60;
 	int      waited     = 0;
 
 	D_DEBUG(DB_MD, DF_UUID "[%d]: wait for pausing EC aggregation\n",
@@ -981,35 +980,31 @@ ds_cont_child_wait_ec_agg_pause(struct ds_pool_child *pool_child, int wait_timeo
 		struct ds_cont_child *coc;
 		bool                  paused = true;
 
-		/* Wait for pausing aggregation
-		 * XXX: There is no global barrier so we always wait for at least 10 seconds to
-		 * lower the chance that remote targets are still running EC aggregation.
-		 */
-		if (wait_intv > wait_timeout - waited)
-			wait_intv = wait_timeout - waited;
+		if (!ds_pool_child_ec_agg_paused(pool_child))
+			return -DER_OP_CANCELED;
 
-		dss_sleep(wait_intv * 1000);
 		d_list_for_each_entry(coc, &pool_child->spc_cont_list, sc_link) {
 			if (ds_cont_child_ec_aggregating(coc)) {
-				/* Aggregation is active on this container */
 				paused = false;
 				break;
 			}
 		}
 		if (paused)
-			return;
+			return 0;
 
 		waited = daos_wallclock_secs() - start_time;
 		if (waited >= wait_timeout) {
 			D_WARN("can't pause EC aggregation after %d seconds\n", waited);
-			return; /* XXX what can I do? */
+			return -DER_TIMEDOUT;
 		}
 
-		if (waited % 60 == 0) {
+		if (waited >= next_warn) {
 			D_WARN(DF_UUID "[%d]: waited %d secs for EC aggregation to pause\n",
 			       DP_UUID(pool_child->spc_uuid), dss_get_module_info()->dmi_tgt_id,
 			       waited);
+			next_warn += 60;
 		}
+		dss_sleep(1000);
 	}
 }
 

src/include/daos/rpc.h

@@ -71,7 +71,7 @@ enum daos_module_id {
 #define DAOS_POOL_VERSION     7
 #define DAOS_CONT_VERSION     9
 #define DAOS_OBJ_VERSION      10
-#define DAOS_REBUILD_VERSION  5
+#define DAOS_REBUILD_VERSION  6
 #define DAOS_RSVC_VERSION     5
 #define DAOS_RDB_VERSION      5
 #define DAOS_RDBT_VERSION     3

src/include/daos_srv/container.h

@@ -228,7 +228,7 @@ ds_cont_child_destroy(uuid_t pool_uuid, uuid_t cont_uuid);
 
 void
 ds_cont_child_reset_ec_agg_eph_all(struct ds_pool_child *pool_child);
-void
+int
      ds_cont_child_wait_ec_agg_pause(struct ds_pool_child *pool_child, int wait_timeout);
 
 /** initialize a csummer based on container properties. Will retrieve the

src/include/daos_srv/pool.h

@@ -183,6 +183,10 @@ struct ds_pool_child {
 	ABT_eventual	spc_ref_eventual;
 
 	uint64_t                 spc_no_storage : 1; /* The pool shard has no storage. */
+	ATOMIC uint32_t          spc_ec_agg_pause_gate;
+	ATOMIC uint64_t          spc_ec_agg_pause_term;
+	/* Debug state set by RB_OP_REBUILD START after the global EC agg pause. */
+	ATOMIC uint64_t          spc_rebuild_ec_agg_paused_hlc;
 
 	uint32_t	spc_reint_mode;
 	uint32_t	*spc_state;	/* Pointer to ds_pool->sp_states[i] */
@@ -212,6 +216,26 @@ ds_pool_is_rebuilding(struct ds_pool *pool)
 	return (atomic_load(&pool->sp_rebuilding) > 0 || atomic_load(&pool->sp_rebuild_enum) > 0);
 }
 
+static inline bool
+ds_pool_child_ec_agg_paused(struct ds_pool_child *child)
+{
+	return atomic_load(&child->spc_ec_agg_pause_gate) != 0;
+}
+
+static inline bool
+ds_pool_child_rebuild_started(struct ds_pool_child *child)
+{
+	return atomic_load(&child->spc_rebuild_ec_agg_paused_hlc) != 0;
+}
+
+static inline bool
+ds_pool_child_ec_agg_token_match(struct ds_pool_child *child, uint64_t leader_term,
+				 uint32_t rebuild_gen)
+{
+	return atomic_load(&child->spc_ec_agg_pause_gate) == rebuild_gen &&
+	       atomic_load(&child->spc_ec_agg_pause_term) <= leader_term;
+}
+
 /* encode metadata RPC operation key: HLC time first, in network order, for keys sorted by time.
  * allocates the byte-stream, caller must free with D_FREE().
  */

src/object/srv_ec_aggregate.c

@@ -1340,7 +1340,7 @@ agg_peer_check_avail(struct ec_agg_param *agg_param, struct ec_agg_entry *entry)
 	int                    i;
 	int                    rc;
 
-	if (ds_pool_is_rebuilding(agg_param->ap_cont->sc_pool->spc_pool)) {
+	if (ds_pool_child_ec_agg_paused(agg_param->ap_cont->sc_pool)) {
 		rc = -DER_OP_CANCELED;
 		/* We currently pause EC aggregation for rebuild, so just cancel the
 		 * aggregation for the current stripe. It means the following peer status
@@ -1426,8 +1426,15 @@ agg_peer_update_ult(void *arg)
 
 	if (unlikely(DAOS_FAIL_CHECK(DAOS_FORCE_EC_AGG_PEER_FAIL)))
 		D_GOTO(out, rc = -DER_TIMEDOUT);
-
 	agg_param = container_of(entry, struct ec_agg_param, ap_agg_entry);
+	D_ASSERTF(!ds_pool_child_rebuild_started(agg_param->ap_cont->sc_pool),
+		  DF_UUID " EC aggregation peer update after rebuild START: "
+			  "rank %u tgt %d " DF_UOID " stripe " DF_U64 " global_pause_done " DF_X64
+			  "\n",
+		  DP_UUID(agg_param->ap_pool_info.api_pool_uuid), dss_self_rank(),
+		  dss_get_module_info()->dmi_tgt_id, DP_UOID(entry->ae_oid),
+		  entry->ae_cur_stripe.as_stripenum,
+		  atomic_load(&agg_param->ap_cont->sc_pool->spc_rebuild_ec_agg_paused_hlc));
 	rc        = agg_peer_check_avail(agg_param, entry);
 	if (rc != 0)
 		D_GOTO(out, rc);
@@ -1939,9 +1946,10 @@ agg_process_stripe(struct ec_agg_param *agg_param, struct ec_agg_entry *entry)
 
 	/* avoid race between EC aggregation and rebuild scanner */
 	agg_param->ap_cont->sc_ec_agg_updates++;
-	if (ds_pool_is_rebuilding(agg_param->ap_cont->sc_pool->spc_pool)) {
+	if (ds_pool_child_ec_agg_paused(agg_param->ap_cont->sc_pool)) {
 		rc = -DER_OP_CANCELED;
-		DL_INFO(rc, DF_UOID " abort as rebuild started", DP_UOID(entry->ae_oid));
+		DL_INFO(rc, DF_UOID " abort as EC aggregation pause gate is set",
+			DP_UOID(entry->ae_oid));
 		update_vos = false;
 		goto out;
 	}
@@ -2366,10 +2374,10 @@ ec_aggregate_yield(struct ec_agg_param *agg_param)
 {
 	int	rc;
 
-	if (ds_pool_is_rebuilding(agg_param->ap_pool_info.api_pool)) {
-		D_INFO(DF_UUID ": abort ec aggregation, sp_rebuilding %d\n",
+	if (ds_pool_child_ec_agg_paused(agg_param->ap_cont->sc_pool)) {
+		D_INFO(DF_UUID ": abort EC aggregation, pause gate %u\n",
 		       DP_UUID(agg_param->ap_pool_info.api_pool->sp_uuid),
-		       atomic_load(&agg_param->ap_pool_info.api_pool->sp_rebuilding));
+		       atomic_load(&agg_param->ap_cont->sc_pool->spc_ec_agg_pause_gate));
 		return true;
 	}
 
@@ -2578,15 +2586,15 @@ agg_iterate_pre_cb(daos_handle_t ih, vos_iter_entry_t *entry,
 
 	D_ASSERT(agg_param->ap_initialized);
 
-	/* If rebuild started, abort it to save conflict window with rebuild
+	/* If the rebuild EC agg barrier is active, abort to save conflict window with rebuild
 	 * (see obj_inflight_io_check()).
 	 */
-	if (ds_pool_is_rebuilding(agg_param->ap_pool_info.api_pool)) {
+	if (ds_pool_child_ec_agg_paused(agg_param->ap_cont->sc_pool)) {
 		rc = -DER_OP_CANCELED;
-		DL_INFO(rc, DF_CONT " abort as rebuild started, sp_rebuilding %d.",
+		DL_INFO(rc, DF_CONT " abort as EC aggregation pause gate %u is set.",
 			DP_CONT(agg_param->ap_pool_info.api_pool_uuid,
 				agg_param->ap_pool_info.api_cont_uuid),
-			atomic_load(&agg_param->ap_pool_info.api_pool->sp_rebuilding));
+			atomic_load(&agg_param->ap_cont->sc_pool->spc_ec_agg_pause_gate));
 		return rc;
 	}
 
@@ -2887,7 +2895,7 @@ cont_ec_aggregate_cb(struct ds_cont_child *cont, daos_epoch_range_t *epr,
 	/* clear the flag before next turn's cont_aggregate_runnable(), to save conflict
 	 * window with rebuild (see obj_inflight_io_check()).
 	 */
-	if (ds_pool_is_rebuilding(cont->sc_pool->spc_pool))
+	if (ds_pool_child_ec_agg_paused(cont->sc_pool))
 		cont->sc_ec_agg_active = 0;
 
 	if (rc == 0) {

src/object/srv_obj.c

@@ -2664,6 +2664,20 @@ ds_obj_ec_agg_handler(crt_rpc_t *rpc)
 	}
 
 	D_ASSERT(ioc.ioc_coc != NULL);
+	if (ds_pool_child_ec_agg_paused(ioc.ioc_coc->sc_pool)) {
+		rc = -DER_OP_CANCELED;
+		DL_INFO(rc, DF_CONT " reject EC aggregation update while rebuild pause gate is set",
+			DP_CONT(ioc.ioc_coc->sc_pool_uuid, ioc.ioc_coc->sc_uuid));
+		goto out;
+	}
+	D_ASSERTF(!ds_pool_child_rebuild_started(ioc.ioc_coc->sc_pool),
+		  DF_CONT " EC aggregation handler after rebuild START: "
+			  "rank %u tgt %d " DF_UOID " stripe " DF_U64 " epoch " DF_X64 "-" DF_X64
+			  " global_pause_done " DF_X64 "\n",
+		  DP_CONT(ioc.ioc_coc->sc_pool_uuid, ioc.ioc_coc->sc_uuid), dss_self_rank(),
+		  dss_get_module_info()->dmi_tgt_id, DP_UOID(oea->ea_oid), oea->ea_stripenum,
+		  oea->ea_epoch_range.epr_lo, oea->ea_epoch_range.epr_hi,
+		  atomic_load(&ioc.ioc_coc->sc_pool->spc_rebuild_ec_agg_paused_hlc));
 	ioc.ioc_coc->sc_ec_agg_updates++;
 
 	dkey = (daos_key_t *)&oea->ea_dkey;

src/rebuild/rebuild_internal.h

@@ -87,22 +87,17 @@ struct rebuild_tgt_pool_tracker {
 	/* new layout version for upgrade rebuild */
 	uint32_t		rt_new_layout_ver;
 
-	unsigned int		rt_lead_puller_running:1,
-				rt_abort:1,
-				/* re-report #rebuilt cnt per master change */
-				rt_re_report:1,
-				rt_finishing:1,
-				rt_scan_done:1,
-				rt_global_scan_done:1,
-				rt_global_done:1;
+	unsigned int             rt_lead_puller_running : 1, rt_abort : 1,
+	    /* re-report #rebuilt cnt per master change */
+	    rt_re_report : 1, rt_finishing : 1, rt_ec_agg_paused : 1, rt_scan_done : 1,
+	    rt_global_scan_done : 1, rt_global_done : 1;
 };
 
 struct rebuild_server_status {
 	d_rank_t	rank;
 	double          last_update;
 	uint32_t	dtx_resync_version;
-	uint32_t	scan_done:1,
-			pull_done:1;
+	uint32_t        ec_agg_paused : 1, scan_done : 1, pull_done : 1;
 };
 
 /* Track the rebuild status globally */
@@ -325,11 +320,8 @@ struct rebuild_iv {
 	uint32_t        riv_master_rank;
 	uint32_t        riv_ver;
 	uint32_t        riv_rebuild_gen;
-	uint32_t	riv_global_done:1,
-			riv_global_scan_done:1,
-			riv_scan_done:1,
-			riv_pull_done:1,
-			riv_sync:1;
+	uint32_t        riv_global_done : 1, riv_global_scan_done : 1, riv_scan_done : 1,
+	    riv_pull_done : 1, riv_sync : 1, riv_ec_agg_paused : 1;
 	int32_t  riv_status;
 	uint32_t riv_bukid; /* bucket ID */
 };