CORTX-27600 dtm0/log: initial implementation for dtx0 and pruner

be/btree.c

@@ -4113,6 +4113,7 @@ static bool fkvv_invariant(const struct nd *node)
 					    M0_BT_COB_OBJECT_INDEX,
 					    M0_BT_COB_FILEATTR_BASIC,
 					    M0_BT_CONFDB,
+					    M0_BT_DTM0_LOG,
 					    M0_BT_UT_KV_OPS))) &&
 	_0C(h->fkvv_ksize != 0);
 }

be/btree.h

@@ -61,6 +61,7 @@ enum m0_btree_types {
 	M0_BT_COB_FILEATTR_OMG,
 	M0_BT_COB_BYTECOUNT,
 	M0_BT_CONFDB,
+	M0_BT_DTM0_LOG,
 	M0_BT_UT_KV_OPS,
 	M0_BT_NR
 };

be/linux_kernel/stubs.c

@@ -111,6 +111,13 @@ M0_INTERNAL bool m0_be_op_is_done(struct m0_be_op *op)
 	return true;
 }
 
+M0_INTERNAL int m0_be_op_tick_ret(struct m0_be_op *op,
+				  struct m0_fom   *fom,
+				  int              next_state)
+{
+	return 0;
+}
+
 M0_INTERNAL struct m0_be_allocator *m0_be_seg_allocator(struct m0_be_seg *seg)
 {
 	return NULL;
@@ -187,6 +194,11 @@ M0_INTERNAL void m0_be_tx_open(struct m0_be_tx *tx)
 	tx->t_sm.sm_state = M0_BTS_ACTIVE;
 }
 
+M0_INTERNAL int m0_be_tx_exclusive_open_sync(struct m0_be_tx *tx)
+{
+	return -ENOSYS;
+}
+
 M0_INTERNAL void
 m0_be_tx_capture(struct m0_be_tx *tx, const struct m0_be_reg *reg)
 {
@@ -293,6 +305,12 @@ M0_INTERNAL struct m0_be_seg *m0_be_domain_seg0_get(struct m0_be_domain *dom)
 	return NULL;
 }
 
+M0_INTERNAL struct m0_be_seg *
+m0_be_domain_seg_first(const struct m0_be_domain *dom)
+{
+	return NULL;
+}
+
 int m0_be_tx_fol_add(struct m0_be_tx *tx, struct m0_fol_rec *rec)
 {
 	return M0_ERR(-EINVAL);
@@ -327,5 +345,22 @@ M0_INTERNAL void m0_be_queue_unlock(struct m0_be_queue *bq)
 {
 }
 
+void m0_be_0type_add_credit(struct m0_be_domain *dom,
+			    const struct m0_be_0type  *zt,
+			    const char                *suffix,
+			    const struct m0_buf       *data,
+			    struct m0_be_tx_credit    *credit)
+{
+}
+
+int m0_be_0type_add(struct m0_be_0type  *zt,
+		    struct m0_be_domain *dom,
+		    struct m0_be_tx     *tx,
+		    const char          *suffix,
+		    const struct m0_buf *data)
+{
+	return -ENOSYS;
+}
+
 
 #undef M0_TRACE_SUBSYSTEM

be/seg0.c

@@ -115,7 +115,7 @@ int m0_be_0type_del(struct m0_be_0type  *zt,
 		    const char          *suffix)
 {
 	struct m0_be_seg *seg;
-	struct m0_buf    *opt;
+	struct m0_buf    *opt = NULL;
 	char              keyname[256] = {};
 	int               rc;
 

be/seg0.h

@@ -114,6 +114,15 @@ int m0_be_0type_del(struct m0_be_0type  *zt,
 		    struct m0_be_tx     *tx,
 		    const char          *suffix);
 
+
+/**
+ * Returns a record with a given m0_be_0type and a given suffix.
+ * @return -ENOENT no such record exists.
+ */
+int m0_be_0type_get(struct m0_be_0type  *zt,
+                    struct m0_be_domain *dom,
+                    const char          *suffix,
+                    struct m0_buf       *data);
 /**
  * <hr> <!------------------------------------------------------------>
  * @section seg0-metadata Meta-segment (seg0), systematic BE storage startup.

be/ut/helper.c

@@ -262,13 +262,15 @@ void m0_be_ut_backend_cfg_default(struct m0_be_domain_cfg *cfg)
 	extern struct m0_be_0type m0_be_cob0;
 	extern struct m0_be_0type m0_be_active_record0;
 	extern struct m0_be_0type m0_be_dtm0;
+	extern struct m0_be_0type m0_dtm0_log0;
 
 	static struct m0_atomic64        dom_key = { .a_value = 0xbef11e };
 	static const struct m0_be_0type *zts[] = {
 		&m0_stob_ad_0type,
 		&m0_be_cob0,
 		&m0_be_active_record0,
 		&m0_be_dtm0,
+		&m0_dtm0_log0,
 	};
 	struct m0_reqh *reqh = cfg->bc_engine.bec_reqh;
 

cas/cas.c

@@ -34,6 +34,7 @@
 #include "mdservice/fsync_fops.h"       /* m0_fsync_fom_ops */
 #include "mdservice/fsync_fops_xc.h"    /* m0_fop_fsync_xc */
 #include "cas/client.h"                 /* m0_cas_sm_conf_init */
+#include "lib/memory.h"                 /* M0_ALLOC_PTR */
 
 struct m0_fom_type_ops;
 struct m0_sm_conf;
@@ -285,6 +286,63 @@ M0_INTERNAL bool m0_crv_is_none(const struct m0_crv *crv)
 	return memcmp(crv, &M0_CRV_INIT_NONE, sizeof(*crv)) == 0;
 }
 
+M0_INTERNAL bool m0_cas_fop_is_redoable(struct m0_fop *fop)
+{
+	struct m0_fop_type *cas_fopt = &cas_put_fopt;
+	return fop->f_type == cas_fopt;
+}
+
+M0_INTERNAL int m0_cas_fop2redo(struct m0_fop *fop,
+				struct m0_dtm0_redo *redo)
+{
+	struct m0_cas_op *op = m0_fop_data(fop);
+	struct m0_buf     payload = {};
+	int               rc;
+
+	/* TODO: encode fop opcode and fop reply as well. */
+
+	rc = m0_xcode_obj_enc_to_buf(&M0_XCODE_OBJ(m0_cas_op_xc, op),
+				     &payload.b_addr, &payload.b_nob);
+	if (rc != 0)
+		return M0_ERR(rc);
+
+	rc = m0_dtm0_redo_init(redo, &op->cg_descriptor,
+			       &payload, M0_DTX0_PAYLOAD_CAS);
+	if (rc != 0)
+		m0_buf_free(&payload);
+
+	return M0_RC(rc);
+}
+
+M0_INTERNAL int m0_cas_redo2fop(struct m0_fop *fop,
+				struct m0_dtm0_redo *redo)
+{
+	struct m0_buf      *payload = redo->dtr_payload.dtp_data.ab_elems;
+	struct m0_cas_op   *op;
+	struct m0_fop_type *cas_fopt;
+	int                 rc;
+
+	M0_PRE(redo->dtr_payload.dtp_type == M0_DTX0_PAYLOAD_CAS);
+	M0_PRE(redo->dtr_payload.dtp_data.ab_count == 1);
+
+	/* TODO: Select the right fop type based on encoded req type. */
+	cas_fopt = &cas_put_fopt;
+
+	M0_ALLOC_PTR(op);
+	if (op == NULL)
+		return M0_ERR(-ENOMEM);
+
+	rc = m0_xcode_obj_dec_from_buf(&M0_XCODE_OBJ(m0_cas_op_xc, op),
+				       payload->b_addr, payload->b_nob);
+	if (rc == 0)
+		m0_fop_init(fop, cas_fopt, op, &m0_fop_release);
+	else
+		m0_free(op);
+
+	return M0_RC(rc);
+}
+
+
 #undef M0_TRACE_SUBSYSTEM
 
 /** @} end of cas group */

cas/cas.h

@@ -42,6 +42,8 @@
 #include "dix/layout_xc.h"
 #include "dtm0/tx_desc.h"	/* tx_desc */
 #include "dtm0/tx_desc_xc.h"	/* xc for tx_desc */
+#include "dtm0/dtm0.h"          /* m0_dtx0_descriptor */
+#include "dtm0/dtm0_xc.h"       /* xc for m0_dtx0_descriptor */
 
 /**
  * @page cas-fspec The catalogue service (CAS)
@@ -398,6 +400,12 @@ struct m0_cas_op {
 	 * Transaction descriptor associated with CAS operation.
 	 */
 	struct m0_dtm0_tx_desc cg_txd;
+
+	/**
+	 * Transaction descriptor (new DTM0) associated with this CAS
+	 * operation.
+	 */
+	struct m0_dtx0_descriptor cg_descriptor;
 } M0_XCA_RECORD M0_XCA_DOMAIN(rpc);
 
 /**
@@ -475,6 +483,11 @@ M0_INTERNAL void m0_cas__ut_svc_be_set(struct m0_reqh_service *svc,
 				       struct m0_be_domain *dom);
 M0_INTERNAL struct m0_be_domain *
 m0_cas__ut_svc_be_get(struct m0_reqh_service *svc);
+struct m0_dtm0_domain;
+M0_INTERNAL void m0_cas__ut_svc_dtm0_domain_set(struct m0_reqh_service *svc,
+						struct m0_dtm0_domain  *dod);
+M0_INTERNAL struct m0_be_domain *
+m0_cas__ut_svc_be_get(struct m0_reqh_service *svc);
 M0_INTERNAL int m0_cas_fom_spawn(
 	struct m0_fom           *lead,
 	struct m0_fom_thralldom *thrall,
@@ -542,6 +555,15 @@ M0_INTERNAL struct m0_dtm0_ts m0_crv_ts(const struct m0_crv *crv);
 M0_INTERNAL void m0_crv_ts_set(struct m0_crv           *crv,
 			       const struct m0_dtm0_ts *ts);
 
+struct m0_fop;
+struct m0_dtm0_redo;
+
+M0_INTERNAL bool m0_cas_fop_is_redoable(struct m0_fop *fop);
+M0_INTERNAL int m0_cas_fop2redo(struct m0_fop *fop,
+				struct m0_dtm0_redo *redo);
+M0_INTERNAL int m0_cas_redo2fop(struct m0_fop *fop,
+				struct m0_dtm0_redo *redo);
+
 /** @} end of cas_dfspec */
 #endif /* __MOTR_CAS_CAS_H__ */
 

cas/client.c

@@ -1649,6 +1649,47 @@ static int cas_req_prep(struct m0_cas_req       *req,
 	return M0_RC(rc);
 }
 
+static int
+m0_dtm0_tx_desc2dtx0_descriptor_prep(const struct m0_dtm0_tx_desc *txd,
+				     struct m0_dtx0_descriptor    *descriptor)
+{
+	static int i = 0;
+
+	/*
+	 * TODO: Check if txd is valid or not. If txd is not valid then
+	 * just fill the descriptor with some random data.
+	 * (txd is not valid == it was zeroed).
+	 */
+	descriptor->dtd_id.dti_originator_sdev_fid.f_key = i++;
+	descriptor->dtd_id.dti_originator_sdev_fid.f_container = 1;
+	descriptor->dtd_id.dti_timestamp = m0_time_now();
+	M0_LOG(M0_DEBUG, "DTX id: " DTID1_F, DTID1_P(&descriptor->dtd_id));
+
+	/**
+	 * @todo update all participants of this request in the descriptor,
+	 * until then originator only is added as a participant.
+	 */ 
+	if (txd->dtd_ps.dtp_nr == 0) {
+		descriptor->dtd_participants.dtpa_participants_nr = 1;
+		M0_ALLOC_ARR(descriptor->dtd_participants.dtpa_participants, 1);
+		if (descriptor->dtd_participants.dtpa_participants == NULL)
+			return -ENOMEM;
+		descriptor->dtd_participants.dtpa_participants[0] =
+			descriptor->dtd_id.dti_originator_sdev_fid;
+		return 0;
+	}
+
+	descriptor->dtd_participants.dtpa_participants_nr = txd->dtd_ps.dtp_nr;
+	M0_ALLOC_ARR(descriptor->dtd_participants.dtpa_participants,
+		     txd->dtd_ps.dtp_nr);
+	if (descriptor->dtd_participants.dtpa_participants == NULL)
+		return -ENOMEM;
+	for (i = 0; i < txd->dtd_ps.dtp_nr; i++)
+		descriptor->dtd_participants.dtpa_participants[i] =
+				txd->dtd_ps.dtp_pa->p_fid;
+	return 0;
+}
+
 M0_INTERNAL int m0_cas_put(struct m0_cas_req      *req,
 			   struct m0_cas_id       *index,
 			   const struct m0_bufvec *keys,
@@ -1683,6 +1724,12 @@ M0_INTERNAL int m0_cas_put(struct m0_cas_req      *req,
 	rc = m0_dtx0_txd_copy(dtx, &op->cg_txd);
 	if (rc != 0)
 		return M0_ERR(rc);
+
+	rc = m0_dtm0_tx_desc2dtx0_descriptor_prep(&op->cg_txd,
+						  &op->cg_descriptor);
+	if (rc != 0)
+		return M0_ERR(rc);
+
 	rc = creq_fop_create_and_prepare(req, &cas_put_fopt, op, &next_state);
 	if (rc == 0) {
 		cas_fop_send(req);

cas/index_gc.c

@@ -490,6 +490,7 @@ static void cgc_start_fom(struct m0_fom *fom0, struct m0_fop *fop)
 	m0_fom_init(fom0, &fop->f_type->ft_fom_type,
 		    &cgc_fom_ops, fop, NULL, fom->cg_reqh);
 	fom0->fo_local = true;
+	fom0->fo_local_update = true;
 	fom->cg_ctg_op_initialized = false;
 	m0_long_lock_link_init(&fom->cg_dead_index, fom0,
 			       &fom->cg_dead_index_addb2);