Merge branch 'main' into fis-release

CONTRIBUTING.md

@@ -175,5 +175,5 @@ Refer to the [Motr Coding Style Guide](../dev/doc/coding-style.md) and the [CORT
 
 You can reach out to us with your questions, feedback, and comments through our CORTX Communication Channels:
 
-- Join our CORTX-Open Source Slack Channel to interact with your fellow community members and gets your questions answered. [![Slack Channel](https://img.shields.io/badge/chat-on%20Slack-blue)](https://join.slack.com/t/cortxcommunity/shared_invite/zt-femhm3zm-yiCs5V9NBxh89a_709FFXQ?)
+- Join our CORTX-Open Source Slack Channel to interact with community members and gets your questions answered. [![Slack Channel](https://img.shields.io/badge/chat-on%20Slack-blue)](https://join.slack.com/t/cortxcommunity/shared_invite/zt-femhm3zm-yiCs5V9NBxh89a_709FFXQ?)
 - If you'd like to contact us directly, drop us a mail at cortx-questions@seagate.com.

SUPPORT.md

@@ -3,6 +3,6 @@ Looking for support for CORTX parent or a repository ?
 Consider some of these resources:
 
 -   Join our CORTX-Open Source Slack channel [![Slack](https://img.shields.io/badge/chat-on%20Slack-blue")](https://join.slack.com/t/cortxcommunity/shared_invite/zt-femhm3zm-yiCs5V9NBxh89a_709FFXQ?) to interact with community members and gets your questions answered.
--   Join [GitHub Discussions](https://github.com/Seagate/cortx-motr/discussions) to ask, answer, and discuss topics with your fellow CORTX contributors! 
+-   Join [GitHub Discussions](https://github.com/Seagate/cortx-motr/discussions) to ask, answer, and discuss topics with CORTX contributors! 
 -   If you'd like to contact us directly, drop us a mail at [cortx-questions@seagate.com](mailto:cortx-questions@seagate.com) .
 -   We like to highlight the work and contributions of our community members—if you have solved an interesting challenge, or you are interested in sharing your experience or use cases, we want to talk to you! Please email our Community Manager [rachel.novak@seagate.com](mailto:rachel.novak@seagate.com) or [schedule a meeting with us](https://outlook.office365.com/owa/calendar/CORTXCommunity@seagate.com/bookings/s/x8yMn2ODxUCOdhxvXkH4FA2) to share.

be/btree.c

@@ -1893,10 +1893,10 @@ M0_INTERNAL void m0_btree_lrulist_set_lru_config(int64_t slow_lru_mem_release,
 	M0_ASSERT(lru_space_wm_high >= lru_space_wm_target &&
 		  lru_space_wm_target >= lru_space_wm_low);
 
-	M0_LOG(M0_INFO, "Btree LRU List Watermarks: Low - %"PRIi64" Mid - "
+	M0_LOG(M0_NOTICE, "Btree LRU List Watermarks: Low - %"PRIi64" Mid - "
 	       "%"PRIi64" High - %"PRIi64" \n", lru_space_wm_low,
 	       lru_space_wm_target, lru_space_wm_high);
-	M0_LOG(M0_INFO, "Btree LRU List trickle release: %s \n",
+	M0_LOG(M0_NOTICE, "Btree LRU List trickle release: %s \n",
 	       lru_trickle_release_en ? "true" : "false");
 }
 
@@ -2025,7 +2025,6 @@ static int64_t tree_get(struct node_op *op, struct segaddr *addr, int nxt)
 	return nxt;
 }
 
-
 /**
  * Returns the tree to the free tree pool if the reference count for this tree
  * reaches zero.
@@ -8780,7 +8779,7 @@ M0_INTERNAL int64_t m0_btree_lrulist_purge_check(enum m0_btree_purge_user user,
 	if (lru_space_used < lru_space_wm_low) {
 		/** Do nothing. */
 		if (user == M0_PU_EXTERNAL)
-			M0_LOG(M0_INFO, "Skipping memory release since used "
+			M0_LOG(M0_ALWAYS, "Skipping memory release since used "
 			       "space is below threshold requested size=%"PRId64
 			       " used space=%"PRId64, size, lru_space_used);
 		lru_trickle_release_mode = false;
@@ -8806,7 +8805,7 @@ M0_INTERNAL int64_t m0_btree_lrulist_purge_check(enum m0_btree_purge_user user,
 			purged_size = m0_btree_lrulist_purge(size_to_purge,
 						size_to_purge != 0 ? 0 :
 						M0_BTREE_TRICKLE_NUM_NODES);
-			M0_LOG(M0_INFO, " Below critical External user Purge,"
+			M0_LOG(M0_ALWAYS, " Below critical External user Purge,"
 			       " requested size=%"PRId64" used space=%"PRId64
 			       " purged size=%"PRId64, size, lru_space_used,
 			       purged_size);
@@ -8827,7 +8826,7 @@ M0_INTERNAL int64_t m0_btree_lrulist_purge_check(enum m0_btree_purge_user user,
 	purged_size = m0_btree_lrulist_purge(size_to_purge,
 			      (lru_trickle_release_mode && size_to_purge == 0) ?
 			      M0_BTREE_TRICKLE_NUM_NODES : 0);
-	M0_LOG(M0_INFO, " Above critical purge, User=%s requested size="
+	M0_LOG(M0_ALWAYS, " Above critical purge, User=%s requested size="
 	       "%"PRId64" used space=%"PRIu64" purged size="
 	       "%"PRIu64, user == M0_PU_BTREE ? "btree" : "external", size,
 	       lru_space_used, purged_size);

cas/ctg_store.c

@@ -169,6 +169,10 @@ static int  ctg_op_exec      (struct m0_ctg_op *ctg_op, int next_phase);
 static int  ctg_meta_exec    (struct m0_ctg_op    *ctg_op,
 			      const struct m0_fid *fid,
 			      int                  next_phase);
+static int  ctg_dead_exec    (struct m0_ctg_op    *ctg_op,
+			      struct m0_cas_ctg   *ctg,
+			      const struct m0_buf *key,
+			      int                  next_phase);
 static int  ctg_exec         (struct m0_ctg_op    *ctg_op,
 			      struct m0_cas_ctg   *ctg,
 			      const struct m0_buf *key,
@@ -479,7 +483,7 @@ int m0_ctg_create(struct m0_be_seg *seg, struct m0_be_tx *tx,
 			bt.vsize = sizeof(struct meta_value);
 			break;
 		case CTT_DEADIDX:
-			bt.ksize = sizeof(struct meta_value);
+			bt.ksize = sizeof(struct generic_key *) + sizeof(void *);
 			bt.vsize = sizeof(void *);
 			break;
 		case CTT_CTIDX:
@@ -966,6 +970,11 @@ static void ctg_store_release(struct m0_ref *ref)
 
 	M0_ENTRY();
 	m0_mutex_fini(&ctg_store->cs_state_mutex);
+	/* TODO: Clean up every index in memory tree allocation upon any CAS
+	   operation on the index */
+	ctg_fini(ctg_store->cs_state->cs_meta);
+	ctg_fini(ctg_store->cs_ctidx);
+	ctg_fini(ctg_store->cs_dead_index);
 	ctg_store->cs_state = NULL;
 	ctg_store->cs_ctidx = NULL;
 	m0_long_lock_fini(&ctg_store->cs_del_lock);
@@ -1382,7 +1391,7 @@ static int ctg_op_exec_normal(struct m0_ctg_op *ctg_op, int next_phase)
 		 * m0_be_btree_insert_inplace() have 0 there.
 		 */
 
-		vsize = sizeof(struct generic_value);
+		vsize = sizeof(void *);
 		rec.r_key.k_data = M0_BUFVEC_INIT_BUF(&k_ptr, &ksize);
 		rec.r_val        = M0_BUFVEC_INIT_BUF(&v_ptr, &vsize);
 		rec.r_crc_type   = M0_BCT_NO_CRC;
@@ -1425,6 +1434,8 @@ static int ctg_op_exec_normal(struct m0_ctg_op *ctg_op, int next_phase)
 								&kv_op, tx));
 		m0_be_op_done(beop);
 		break;
+	case CTG_OP_COMBINE(CO_DEL, CT_DEAD_INDEX):
+		ksize = sizeof(struct generic_key *) + sizeof(void *);
 	case CTG_OP_COMBINE(CO_DEL, CT_BTREE):
 	case CTG_OP_COMBINE(CO_DEL, CT_META):
 		m0_be_op_active(beop);
@@ -1649,13 +1660,49 @@ M0_INTERNAL int m0_ctg_dead_index_insert(struct m0_ctg_op  *ctg_op,
 					 struct m0_cas_ctg *ctg,
 					 int                next_phase)
 {
-	ctg_op->co_ctg = m0_ctg_dead_index();
-	ctg_op->co_ct = CT_DEAD_INDEX;
 	ctg_op->co_opcode = CO_PUT;
 	/* Dead index value is empty */
 	ctg_op->co_val = M0_BUF_INIT0;
 	/* Dead index key is a pointer to a catalogue */
-	return ctg_exec(ctg_op, ctg, &M0_BUF_INIT_PTR(&ctg), next_phase);
+	return ctg_dead_exec(ctg_op, ctg, &M0_BUF_INIT_PTR(&ctg), next_phase);
+}
+
+M0_INTERNAL int m0_ctg_dead_delete(struct m0_ctg_op    *ctg_op,
+				   struct m0_cas_ctg   *ctg,
+				   const struct m0_buf *key,
+				   int                  next_phase)
+{
+	M0_PRE(ctg_op != NULL);
+	M0_PRE(ctg != NULL);
+	M0_PRE(key != NULL);
+	M0_PRE(ctg_op->co_beop.bo_sm.sm_state == M0_BOS_INIT);
+
+	ctg_op->co_opcode = CO_DEL;
+
+	return ctg_dead_exec(ctg_op, ctg, key, next_phase);
+}
+
+static int ctg_dead_exec(struct m0_ctg_op    *ctg_op,
+			 struct m0_cas_ctg   *ctg,
+			 const struct m0_buf *key,
+			 int                  next_phase)
+{
+	int ret = M0_FSO_AGAIN;
+
+	ctg_op->co_ctg = m0_ctg_dead_index();
+	ctg_op->co_ct  = CT_DEAD_INDEX;
+
+	if (!M0_IN(ctg_op->co_opcode, (CO_MIN, CO_TRUNC, CO_DROP)) &&
+	    (ctg_op->co_opcode != CO_CUR ||
+	     ctg_op->co_cur_phase != CPH_NEXT))
+		ctg_op->co_rc = ctg_kbuf_get(&ctg_op->co_key, key, true);
+
+	if (ctg_op->co_rc != 0)
+		m0_fom_phase_set(ctg_op->co_fom, next_phase);
+	else
+		ret = ctg_op_exec(ctg_op, next_phase);
+
+	return ret;
 }
 
 static int ctg_exec(struct m0_ctg_op    *ctg_op,
@@ -1666,7 +1713,7 @@ static int ctg_exec(struct m0_ctg_op    *ctg_op,
 	int ret = M0_FSO_AGAIN;
 
 	ctg_op->co_ctg = ctg;
-	ctg_op->co_ct = CT_BTREE;
+	ctg_op->co_ct  = CT_BTREE;
 
 	if (!M0_IN(ctg_op->co_opcode, (CO_MIN, CO_TRUNC, CO_DROP)) &&
 	    (ctg_op->co_opcode != CO_CUR ||

cas/ctg_store.h

@@ -390,6 +390,18 @@ M0_INTERNAL int m0_ctg_dead_index_insert(struct m0_ctg_op  *ctg_op,
 					 struct m0_cas_ctg *ctg,
 					 int                next_phase);
 
+/**
+ * Deletes 'ctg' from "dead index" catalogue.
+ *
+ * @param ctg_op     Catalogue operation context.
+ * @param ctg        Catalogue to be deleted from "dead index" catalogue.
+ * @param next_phase Next phase of caller FOM.
+ */
+M0_INTERNAL int m0_ctg_dead_delete(struct m0_ctg_op    *ctg_op,
+				   struct m0_cas_ctg   *ctg,
+				   const struct m0_buf *key,
+				   int                  next_phase);
+
 /**
  * Looks up a catalogue in meta catalogue.
  *

cas/index_gc.c

@@ -361,6 +361,12 @@ static int cgc_fom_tick(struct m0_fom *fom0)
 			fom->cg_ctg_op_initialized = true;
 			result = m0_ctg_drop(ctg_op, fom->cg_ctg,
 					     CGC_LOCK_DEAD_INDEX);
+			/*
+			 * Free the memory allocated for the root node after
+			 * destroying the tree.
+			 */
+			if (result == M0_FSO_AGAIN)
+				m0_free0(&fom->cg_ctg->cc_tree);
 		} else {
 			M0_LOG(M0_DEBUG, "out of credits, commit & restart");
 			m0_long_unlock(m0_ctg_lock(m0_ctg_dead_index()),
@@ -387,14 +393,13 @@ static int cgc_fom_tick(struct m0_fom *fom0)
 		m0_ctg_op_fini(ctg_op);
 		m0_ctg_op_init(ctg_op, fom0, 0);
 		fom->cg_ctg_op_initialized = true;
-		fom->cg_ctg_key = M0_BUF_INIT(M0_CAS_CTG_KEY_HDR_SIZE,
-					      &fom->cg_ctg);
+		fom->cg_ctg_key = M0_BUF_INIT_PTR(&fom->cg_ctg);
 		/*
 		 * Now completely forget this ctg by deleting its descriptor
 		 * from "dead index" catalogue.
 		 */
-		result = m0_ctg_delete(ctg_op, m0_ctg_dead_index(),
-				       &fom->cg_ctg_key, CGC_SUCCESS);
+		result = m0_ctg_dead_delete(ctg_op, m0_ctg_dead_index(),
+					    &fom->cg_ctg_key, CGC_SUCCESS);
 		break;
 	case CGC_SUCCESS:
 		m0_long_unlock(m0_ctg_lock(m0_ctg_dead_index()),

cas/service.c

@@ -1151,6 +1151,22 @@ static int cas_dtm0_logrec_add(struct m0_fom *fom0,
 	int                            i;
 	int                            rc;
 
+	/*
+	 * It is impossible to commit a transaction without DTM0 service up and
+	 * running.
+	 */
+	if (dtms == NULL) {
+		static uint32_t count = 0;
+		if (count == 0) {
+			M0_LOG(M0_FATAL, "DTM is enabled but is not "
+					 "configured in conf. Skip "
+					 "DTM now. Please Check!");
+			count++; /* Only print the message at the first time. */
+		}
+		return 0; /* FIXME but now let's skip it if no DTM service. */
+	}
+	M0_ASSERT(dtms != NULL);
+
 	for (i = 0; i < msg->dtd_ps.dtp_nr; ++i) {
 		if (m0_fid_eq(&msg->dtd_ps.dtp_pa[i].p_fid,
 			      &dtms->dos_generic.rs_service_fid)) {

cas/ut/client_ut.c

@@ -89,6 +89,8 @@ static char *cas_startup_cmd[] = {
 	"-w", "10", "-F",
 	"-f", M0_UT_CONF_PROCESS,
 	"-c", M0_SRC_PATH("cas/ut/conf.xc")
+	/* FIXME If DTM is enabled, the above conf.xc must be updated to include
+	 * DTM0 services. */
 };
 
 static const char         *cdbnames[] = { "cas1" };