Merge branch 'main' into issue-1606-p2

addb2/dump.c

@@ -861,6 +861,17 @@ static void dtx0_state_counter(struct m0_addb2__context *ctx, char *buf)
 	sm_trans(&m0_dtx_sm_conf, "dtx0", ctx, buf);
 }
 
+extern struct m0_sm_conf m0_drm_sm_conf;
+static void drm_state(struct m0_addb2__context *ctx, const uint64_t *v,
+		      char *buf)
+{
+	sm_state(&m0_drm_sm_conf, ctx, v, buf);
+}
+
+static void drm_state_counter(struct m0_addb2__context *ctx, char *buf)
+{
+	sm_trans(&m0_drm_sm_conf, "drm", ctx, buf);
+}
 
 extern struct m0_sm_conf op_states_conf;
 static void beop_state_counter(struct m0_addb2__context *ctx, char *buf)
@@ -1149,6 +1160,11 @@ struct m0_addb2__id_intrp ids[] = {
 	  .ii_repeat = M0_AVI_DTX0_SM_COUNTER_END - M0_AVI_DTX0_SM_COUNTER,
 	  .ii_spec   = &dtx0_state_counter },
 
+	{ M0_AVI_DRM_SM_STATE,     "drm-state",    { &drm_state, SKIP2  } },
+	{ M0_AVI_DRM_SM_COUNTER,   "",
+	  .ii_repeat = M0_AVI_DRM_SM_COUNTER_END - M0_AVI_DRM_SM_COUNTER,
+	  .ii_spec   = &drm_state_counter },
+
 	{ M0_AVI_BE_TX_STATE,     "tx-state",        { &tx_state, SKIP2  } },
 	{ M0_AVI_BE_TX_COUNTER,   "",
 	  .ii_repeat = M0_AVI_BE_TX_COUNTER_END - M0_AVI_BE_TX_COUNTER,

be/dtm0_log.c

@@ -116,15 +116,12 @@ M0_INTERNAL void m0_be_dtm0_log_fini(struct m0_be_dtm0_log *log)
 	log->dl_cs = NULL;
 }
 
-M0_INTERNAL void m0_be_dtm0_log_free(struct m0_be_dtm0_log **in_log)
+M0_INTERNAL void m0_be_dtm0_log_free(struct m0_be_dtm0_log *log)
 {
-	struct m0_be_dtm0_log *log = *in_log;
-
 	M0_PRE(!log->dl_is_persistent);
 
 	m0_free(log->u.dl_inmem);
 	m0_free(log);
-	*in_log = NULL;
 }
 
 /**
@@ -394,8 +391,8 @@ static int dtm0_log__insert(struct m0_be_dtm0_log  *log,
 			    struct m0_buf          *payload)
 {
 	int                      rc;
-	struct m0_dtm0_log_rec	*rec;
-	struct m0_be_seg	*seg = log->dl_seg;
+	struct m0_dtm0_log_rec  *rec;
+	struct m0_be_seg        *seg = log->dl_seg;
 
 	if (log->dl_is_persistent) {
 		rc = plog_rec_init(&rec, tx, seg, txd, payload);
@@ -454,7 +451,7 @@ M0_INTERNAL int m0_be_dtm0_log_update(struct m0_be_dtm0_log  *log,
 				      struct m0_dtm0_tx_desc *txd,
 				      struct m0_buf          *payload)
 {
-	struct m0_dtm0_log_rec	*rec;
+	struct m0_dtm0_log_rec *rec;
 
 	M0_PRE(payload != NULL);
 	M0_PRE(m0_be_dtm0_log__invariant(log));
@@ -653,8 +650,8 @@ static bool be_dtm0_log_iter_is_first(const struct m0_be_dtm0_log_iter *iter)
 static bool be_dtm0_log_iter_invariant(const struct m0_be_dtm0_log_iter *iter)
 {
 	return _0C(m0_be_dtm0_log__invariant(iter->dli_log)) &&
-	       _0C(be_dtm0_log_iter_is_first(iter) ||
-		   m0_dtm0_tid__invariant(&iter->dli_current_tid));
+	       _0C(ergo(!be_dtm0_log_iter_is_first(iter),
+			m0_dtm0_tid__invariant(&iter->dli_current_tid)));
 }
 
 M0_INTERNAL void m0_be_dtm0_log_iter_init(struct m0_be_dtm0_log_iter *iter,
@@ -671,10 +668,10 @@ M0_INTERNAL void m0_be_dtm0_log_iter_fini(struct m0_be_dtm0_log_iter *iter)
 }
 
 M0_INTERNAL int m0_be_dtm0_log_iter_next(struct m0_be_dtm0_log_iter *iter,
-					 struct m0_dtm0_log_rec	    *out)
+					 struct m0_dtm0_log_rec     *out)
 {
 	struct m0_dtm0_log_rec *rec;
-	int rc;
+	int    rc;
 
 	M0_PRE(m0_mutex_is_locked(&iter->dli_log->dl_lock));
 	M0_PRE(be_dtm0_log_iter_invariant(iter));
@@ -699,7 +696,7 @@ M0_INTERNAL int m0_be_dtm0_log_iter_next(struct m0_be_dtm0_log_iter *iter,
 			return M0_ERR(rc);
 	}
 
-	return M0_RC(rec == NULL ? 0 : +1);
+	return M0_RC(rec != NULL ? 0 : -ENOENT);
 }
 
 M0_INTERNAL int m0_dtm0_log_rec_copy(struct m0_dtm0_log_rec       *dst,

be/dtm0_log.h

@@ -179,7 +179,7 @@ struct m0_dtm0_log_rec {
 };
 
 /**
- * @b  m0_be_dtm0_log_rec structure
+ * @b  m0_be_dtm0_log structure
  *
  * A DTM0 log is represented by m0_be_dtm0_log structure. The important
  * fields in this structure are:
@@ -277,13 +277,21 @@ M0_INTERNAL void m0_be_dtm0_log_fini(struct m0_be_dtm0_log *log);
  * Free the memory allocated by m0_be_dtm0_log_alloc
  *
  * @pre m0_be_dtm0_log->dl_is_persistent needs to be false.
- * @post *log is set to NULL.
+ * @post None
  *
  * @param log Pointer to a log structure that has been previously allocated.
  *
  * @return None
  */
-M0_INTERNAL void m0_be_dtm0_log_free(struct m0_be_dtm0_log **log);
+M0_INTERNAL void m0_be_dtm0_log_free(struct m0_be_dtm0_log *log);
+
+/** Frees memory and zeroes the pointer. */
+#define m0_be_dtm0_log_free0(pptr)            \
+	do {                                  \
+		typeof(pptr) __pptr = (pptr); \
+		m0_be_dtm0_log_free(*__pptr); \
+		*__pptr = NULL;               \
+	} while (0)
 
 /**
  * For performing an operation on a persistent log, we need to take the
@@ -553,11 +561,11 @@ M0_INTERNAL void m0_be_dtm0_log_iter_fini(struct m0_be_dtm0_log_iter *iter);
  * @param out returned record which is copied and needs to be freed with
  *            m0_dtm0_log_iter_rec_fini()
  *
- * @return +1 when the iterator was successfully moved to the next record.
- * @return  0 when the end of the log has been reached.
+ * @return 0 when the iterator was successfully moved to the next record.
+ * @return  -ENOENT when the end of the log has been reached.
  */
 M0_INTERNAL int m0_be_dtm0_log_iter_next(struct m0_be_dtm0_log_iter *iter,
-					 struct m0_dtm0_log_rec	    *out);
+					 struct m0_dtm0_log_rec     *out);
 
 /** @} */ /* end DTM0Internals */
 

be/ut/dtm0_log_ut.c

@@ -281,7 +281,7 @@ void test_volatile_dtm0_log(void)
 
 	m0_mutex_unlock(&log->dl_lock);
 	m0_be_dtm0_log_fini(log);
-	m0_be_dtm0_log_free(&log);
+	m0_be_dtm0_log_free0(&log);
 	m0_dtm0_clk_src_fini(&cs);
 }
 
@@ -625,7 +625,7 @@ static void m0_be_ut_dtm0_log_init_fini(void)
 	m0_be_dtm0_log_iter_fini(&iter);
 
 	m0_be_dtm0_log_fini(log);
-	m0_be_dtm0_log_free(&log);
+	m0_be_dtm0_log_free0(&log);
 	m0_dtm0_clk_src_fini(&cs);
 }
 
@@ -635,10 +635,10 @@ static void m0_be_ut_dtm0_log_next(void)
 	struct m0_buf           buf = {};
 
 	struct m0_be_dtm0_log_iter iter;
-	struct m0_dtm0_log_rec	   out;
+	struct m0_dtm0_log_rec     out;
 	struct m0_dtm0_clk_src     cs;
 	struct m0_be_dtm0_log     *log;
-	int rc;
+	int                        rc;
 
 	m0_dtm0_clk_src_init(&cs, M0_DTM0_CS_PHYS);
 
@@ -660,19 +660,20 @@ static void m0_be_ut_dtm0_log_next(void)
 
 	m0_be_dtm0_log_iter_init(&iter, log);
 	rc = m0_be_dtm0_log_iter_next(&iter, &out);
-	M0_UT_ASSERT(rc == 1);
+	M0_UT_ASSERT(rc == 0);
+	M0_UT_ASSERT(ut_dl_verify_log_rec(&out, 42));
 	m0_dtm0_log_iter_rec_fini(&out);
 
 	rc = m0_be_dtm0_log_iter_next(&iter, &out);
-	M0_UT_ASSERT(rc == 0);
+	M0_UT_ASSERT(rc != 0);
 	m0_be_dtm0_log_iter_fini(&iter);
 
 	/* make log finalisation happy */
 	rc = m0_be_dtm0_log_prune(log, NULL, &txd.dtd_id);
 	M0_UT_ASSERT(rc == 0);
 	m0_mutex_unlock(&log->dl_lock);
 	m0_be_dtm0_log_fini(log);
-	m0_be_dtm0_log_free(&log);
+	m0_be_dtm0_log_free0(&log);
 	m0_dtm0_clk_src_fini(&cs);
 }
 

cas/cas.h

@@ -471,6 +471,9 @@ M0_INTERNAL int m0_cas_fom_spawn(
 	struct m0_fop           *cas_fop,
 	void                   (*on_fom_complete)(struct m0_fom_thralldom *,
 						  struct m0_fom           *));
+M0_INTERNAL uint32_t m0_cas_svc_device_id_get(
+	const struct m0_reqh_service_type *stype,
+	const struct m0_reqh              *reqh);
 #else
 #define m0_cas_svc_init()
 #define m0_cas_svc_fini()

cas/service.c

@@ -308,9 +308,21 @@ enum {
 	STATS_NR
 };
 
+enum {
+	INVALID_CAS_SDEV_ID = UINT32_MAX
+};
+
 struct cas_service {
 	struct m0_reqh_service  c_service;
 	struct m0_be_domain    *c_be_domain;
+
+	/**
+	 * sdev used by an instance of CAS service. This should be just one
+	 * for the standard configuration, but current unit tests use more
+	 * than one storage device attached to emulate several CAS services.
+	 * In this case a special invalid value is used.
+	 */
+	uint32_t                c_sdev_id;
 };
 
 struct cas_kv {
@@ -556,6 +568,60 @@ m0_cas__ut_svc_be_get(struct m0_reqh_service *svc)
 	return service->c_be_domain;
 }
 
+static int cas_service_sdev_id_set(struct cas_service *cas_svc)
+{
+	struct m0_reqh         *reqh;
+	struct m0_conf_cache   *cache;
+	struct m0_conf_obj     *obj;
+	struct m0_fid          *cas_svc_fid;
+	struct m0_conf_service *service;
+	struct m0_conf_obj     *sdevs_dir;
+	struct m0_conf_sdev    *sdev = NULL;
+	bool                    found = false;
+	int                     rc = 0;
+
+	M0_ASSERT(cas_svc->c_sdev_id == INVALID_CAS_SDEV_ID);
+
+	if (cas_in_ut())
+		return M0_RC(0);
+
+	reqh = cas_svc->c_service.rs_reqh;
+	cache = &m0_reqh2confc(reqh)->cc_cache;
+
+	cas_svc_fid = &cas_svc->c_service.rs_service_fid;
+	obj = m0_conf_cache_lookup(cache, cas_svc_fid);
+	M0_ASSERT(obj != NULL);
+
+	service = M0_CONF_CAST(obj, m0_conf_service);
+	M0_ASSERT(service != NULL);
+
+	sdevs_dir = &service->cs_sdevs->cd_obj;
+	rc = m0_confc_open_sync(&sdevs_dir, sdevs_dir, M0_FID0);
+	if (rc != 0)
+		return M0_RC(rc);
+
+	obj = NULL;
+	while ((rc = m0_confc_readdir_sync(sdevs_dir, &obj)) > 0) {
+		sdev = M0_CONF_CAST(obj, m0_conf_sdev);
+		if (!found) {
+			cas_svc->c_sdev_id = sdev->sd_dev_idx;
+			found = true;
+		} else {
+			/*
+			 * Several devices attached to a single CAS service are
+			 * possible if we are run in ut. In this case we use
+			 * an invalid value for attached storage device.
+			 * Also we can not break the loop until whole directory
+			 * is iterated since it leads to crash during request
+			 * handler finalisation.
+			 */
+			cas_svc->c_sdev_id = INVALID_CAS_SDEV_ID;
+		}
+	}
+
+	m0_confc_close(sdevs_dir);
+	return M0_RC(rc);
+}
 
 static int cas_service_start(struct m0_reqh_service *svc)
 {
@@ -568,6 +634,7 @@ static int cas_service_start(struct m0_reqh_service *svc)
 	/* XXX It's a workaround. It's needed until we have a better way. */
 	service->c_be_domain = ut_dom != NULL ?
 			       ut_dom : svc->rs_reqh_ctx->rc_beseg->bs_domain;
+	service->c_sdev_id = INVALID_CAS_SDEV_ID;
 	rc = m0_ctg_store_init(service->c_be_domain);
 	if (rc == 0) {
 		/*
@@ -576,6 +643,7 @@ static int cas_service_start(struct m0_reqh_service *svc)
 		 * If no pending index drop, it finishes soon.
 		 */
 		m0_cas_gc_start(svc);
+		rc = cas_service_sdev_id_set(service);
 	}
 	return rc;
 }
@@ -1832,6 +1900,7 @@ static int cas_device_check(const struct cas_fom   *fom,
 			pm = &pver->pv_mach;
 			rc = cas_sdev_state(pm, device_id, &state);
 			if (rc == 0 && !M0_IN(state, (M0_PNDS_ONLINE,
+						      M0_PNDS_OFFLINE,
 						      M0_PNDS_SNS_REBALANCING)))
 				rc = M0_ERR(-EBADFD);
 		} else
@@ -2610,6 +2679,26 @@ M0_INTERNAL int m0_cas_fom_spawn(
 	return M0_RC(rc);
 }
 
+M0_INTERNAL uint32_t m0_cas_svc_device_id_get(
+	const struct m0_reqh_service_type *stype,
+	const struct m0_reqh              *reqh)
+{
+	struct m0_reqh_service *svc;
+	struct cas_service     *cas_svc;
+
+	M0_PRE(stype != NULL);
+	M0_PRE(reqh != NULL);
+
+	svc = m0_reqh_service_find(stype, reqh);
+	M0_ASSERT(svc != NULL);
+	M0_ASSERT(m0_reqh_service_state_get(svc) == M0_RST_STARTED);
+
+	cas_svc = M0_AMB(cas_svc, svc, c_service);
+
+	M0_ASSERT(cas_svc->c_sdev_id != INVALID_CAS_SDEV_ID);
+	return cas_svc->c_sdev_id;
+}
+
 static int cas_ctg_crow_handle(struct cas_fom *fom, const struct m0_cas_id *cid)
 {
 	struct m0_fop         *fop;

dix/fid_convert.c

@@ -35,6 +35,15 @@
  * @{
  */
 
+/* Set device id in a DIX fid. */
+M0_INTERNAL void m0_dix_fid__device_id_set(struct m0_fid *fid,
+					   uint32_t       dev_id)
+{
+	M0_PRE(fid != NULL && (dev_id <= M0_DIX_FID_DEVICE_ID_MAX));
+	fid->f_container = (fid->f_container & ~M0_DIX_FID_DEVICE_ID_MASK) |
+			   (((uint64_t)dev_id) << M0_DIX_FID_DEVICE_ID_OFFSET);
+}
+
 /* extract bits [32, 56) from fid->f_container */
 M0_INTERNAL uint32_t m0_dix_fid__device_id_extract(const struct m0_fid *fid)
 {

dix/fid_convert.h

@@ -78,6 +78,8 @@ M0_INTERNAL bool m0_dix_fid_validate_cctg(const struct m0_fid *cctg_fid);
 
 M0_INTERNAL uint32_t m0_dix_fid__device_id_extract(const struct m0_fid *fid);
 
+M0_INTERNAL void m0_dix_fid__device_id_set(struct m0_fid *fid,
+					   uint32_t       dev_id);
 /** @} end of dix group */
 #endif /* __MOTR_DIX_FID_CONVERT_H__ */