cob_foms.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2013-2020 Seagate Technology LLC and/or its Affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * For any questions about this software or licensing,
 * please email opensource@seagate.com or cortx-questions@seagate.com.
 *
 */


#include "ioservice/cob_foms.c" /* to access static APIs */

#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_COB
#include "lib/trace.h"

#include "ioservice/ut/bulkio_common.h" /* cob_attr_default_fill */
#include "mdservice/fsync_fops.h"       /* m0_fop_fsync */
#include "rpc/rpclib.h"                 /* m0_rpc_server_ctx */
#include "rpc/rpc_opcodes.h"            /* M0_UT_IOS_OPCODE */
#include "rpc/rpc_machine.h"            /* m0_rpc_machine_lock */
#include "stob/type.h"                  /* m0_stob_type */
#include "stob/ad.h"                    /* m0_stob_ad_type */
#include "stob/linux.h"                 /* m0_stob_linux_type */
#include "ut/cs_fop.h"                  /* m0_ut_fom_phase_set */
#include "ut/misc.h"                    /* M0_UT_PATH */
#include "ut/ut.h"

extern struct m0_fop_type m0_fop_cob_create_fopt;
extern struct m0_fop_type m0_fop_cob_delete_fopt;

extern struct m0_fop_type m0_fop_fsync_ios_fopt;
extern struct m0_reqh_service_type m0_ios_type;

/* Static instance of struct cobfoms_ut used by all test cases. */
static struct cobfoms_ut      *cut;
static struct m0_fom_locality  dummy_loc;

static struct m0_cob *test_cob = NULL;
static struct m0_fom_type ft;

static bool cc_cob_create_test_done = false;

static struct m0_fom *cd_fom_alloc();
static void cd_fom_dealloc(struct m0_fom *fom);
static void fom_dtx_init(struct m0_fom *fom, struct m0_sm_group *grp,
                         enum m0_cob_op opcode);
static void fom_stob_tx_credit(struct m0_fom *fom, enum m0_cob_op opcode);
static void fom_dtx_done(struct m0_fom *fom, struct m0_sm_group *grp);
static void cd_stob_delete_test();

enum cob_fom_type {
        COB_CREATE = 1,
        COB_DELETE = 2
};

enum {
        CLIENT_COB_DOM_ID         = 12,
        CLIENT_RPC_CONN_TIMEOUT   = 200,
        CLIENT_MAX_RPCS_IN_FLIGHT = 8,
        COB_NAME_STRLEN           = 34,
        GOB_FID_CONTAINER_ID      = 1000,
        GOB_FID_KEY_ID            = 5678,
        COB_FOP_SINGLE            = 1,
        COB_FOP_NR                = 5,
        POOL_WIDTH                = 10,
        COB_TEST_KEY              = 111,
};

static char COB_FOP_NR_STR[] = { '0' + COB_FOP_NR, '\0'};

#define SERVER_EP_ADDR              "0@lo:12345:34:1"
#define CLIENT_EP_ADDR              "0@lo:12345:34:*"
#define SERVER_ENDP                 M0_NET_XPRT_PREFIX_DEFAULT":"SERVER_EP_ADDR

static const char *SERVER_LOGFILE = "cobfoms_ut.log";

struct cobfoms_ut {
        struct m0_rpc_server_ctx      cu_sctx;
        struct m0_rpc_client_ctx      cu_cctx;
        uint64_t                      cu_cobfop_nr;
        struct m0_fop               **cu_createfops;
        struct m0_fop               **cu_deletefops;
        struct m0_fid                 cu_gfid;
        struct m0_fid                 cu_cfid;
        struct m0_net_xprt           *cu_xprt;
        struct m0_net_domain          cu_nd;
        struct m0_cob_domain          cu_cob_dom;
        uint64_t                      cu_thread_nr;
        struct m0_thread            **cu_threads;
        uint64_t                      cu_gobindex;
};

struct cobthread_arg {
        struct m0_fop_type *ca_ftype;
        int                 ca_index;
        int                 ca_rc;
        uint64_t            ca_flags;
};

static char *server_args[] = {
        "m0d", "-T", "AD", "-D", "cobfoms_ut.db", "-S",
        "cobfoms_ut_stob", "-A", "linuxstob:cobfoms_ut_addb_stob",
        "-w", "10", "-e", SERVER_ENDP, "-H", SERVER_EP_ADDR,
        "-f", M0_UT_CONF_PROCESS,
        "-q", COB_FOP_NR_STR,
        "-c", M0_UT_PATH("diter.xc")
};

static void cobfoms_utinit(void)
{
        int                       rc;
        struct m0_rpc_server_ctx *sctx;
        struct m0_rpc_client_ctx *cctx;

        M0_ALLOC_PTR(cut);
        M0_UT_ASSERT(cut != NULL);

        cut->cu_xprt = m0_net_xprt_default_get();

        rc = m0_net_domain_init(&cut->cu_nd, cut->cu_xprt);
        M0_UT_ASSERT(rc == 0);

        sctx = &cut->cu_sctx;
        sctx->rsx_xprts            = &cut->cu_xprt;
        sctx->rsx_xprts_nr         = 1;
        sctx->rsx_argv             = server_args;
        sctx->rsx_argc             = ARRAY_SIZE(server_args);
        sctx->rsx_log_file_name    = SERVER_LOGFILE;

        rc = m0_rpc_server_start(sctx);
        M0_UT_ASSERT(rc == 0);

        cctx = &cut->cu_cctx;
        cctx->rcx_net_dom            = &cut->cu_nd;
        cctx->rcx_local_addr         = CLIENT_EP_ADDR;
        cctx->rcx_remote_addr        = SERVER_EP_ADDR;
        cctx->rcx_max_rpcs_in_flight = CLIENT_MAX_RPCS_IN_FLIGHT;
        cctx->rcx_recv_queue_min_length = COB_FOP_NR;
        cctx->rcx_fid                = &g_process_fid;

        m0_fom_type_init(&ft, M0_UT_IOS_OPCODE,
                         NULL, &m0_ios_type, &cob_ops_conf);

        rc = m0_rpc_client_start(cctx);
        M0_UT_ASSERT(rc == 0);

        cut->cu_gobindex = 0;
}

static void cobfoms_utfini(void)
{
        int rc;

        M0_UT_ASSERT(cut != NULL);

        rc = m0_rpc_client_stop(&cut->cu_cctx);
        M0_UT_ASSERT(rc == 0);

        m0_rpc_server_stop(&cut->cu_sctx);
        m0_net_domain_fini(&cut->cu_nd);
        m0_free0(&cut);
}

static void cobfops_populate_internal(struct m0_fop *fop, uint64_t gob_fid_key)
{
        struct m0_fop_cob_common *common;
        struct m0_cob_attr        attr = { { 0, } };

        cob_attr_default_fill(&attr);
        attr.ca_nlink = fop->f_type == &m0_fop_cob_create_fopt ? 1 : 0;

        M0_UT_ASSERT(fop != NULL);
        M0_UT_ASSERT(fop->f_type != NULL);

        common = m0_cobfop_common_get(fop);
        m0_fid_gob_make(&common->c_gobfid, 1 + gob_fid_key % POOL_WIDTH,
                        GOB_FID_KEY_ID + gob_fid_key);
        m0_fid_convert_gob2cob(&common->c_gobfid, &common->c_cobfid,
                               M0_AD_STOB_DOM_KEY_DEFAULT);
        attr.ca_pver = common->c_pver = CONF_PVER_FID;
        m0_md_cob_mem2wire(&common->c_body, &attr);
}

static void cobfops_populate(uint64_t index)
{
        struct m0_fop            *fop;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_createfops != NULL);
        M0_UT_ASSERT(cut->cu_deletefops != NULL);

        fop = cut->cu_deletefops[index];
        cobfops_populate_internal(fop, cut->cu_gobindex);
        fop = cut->cu_createfops[index];
        cobfops_populate_internal(fop, cut->cu_gobindex);
        cut->cu_gobindex++;
}

static void cobfops_create(void)
{
        uint64_t               i;
        struct m0_rpc_machine *mach;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_createfops == NULL);
        M0_UT_ASSERT(cut->cu_deletefops == NULL);

        M0_ALLOC_ARR(cut->cu_createfops, cut->cu_cobfop_nr);
        M0_UT_ASSERT(cut->cu_createfops != NULL);

        M0_ALLOC_ARR(cut->cu_deletefops, cut->cu_cobfop_nr);
        M0_UT_ASSERT(cut->cu_deletefops != NULL);

        mach = &cut->cu_cctx.rcx_rpc_machine;
        for (i = 0; i < cut->cu_cobfop_nr; ++i) {
                cut->cu_createfops[i] = m0_fop_alloc(&m0_fop_cob_create_fopt,
                                                     NULL, mach);
                M0_UT_ASSERT(cut->cu_createfops[i] != NULL);

                cut->cu_deletefops[i] = m0_fop_alloc(&m0_fop_cob_delete_fopt,
                                                     NULL, mach);
                M0_UT_ASSERT(cut->cu_deletefops[i] != NULL);
                cobfops_populate(i);
        }
}

static void cobfops_destroy(struct m0_fop_type *ftype1,
                            struct m0_fop_type *ftype2)
{
        uint64_t i;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_createfops != NULL);
        M0_UT_ASSERT(cut->cu_deletefops != NULL);
        M0_UT_ASSERT(ftype1 == NULL || ftype1 == &m0_fop_cob_create_fopt);
        M0_UT_ASSERT(ftype2 == NULL || ftype2 == &m0_fop_cob_delete_fopt);

        m0_rpc_machine_lock(&cut->cu_cctx.rcx_rpc_machine);
        if (ftype1 != NULL)
                for (i = 0; i < cut->cu_cobfop_nr; ++i)
                        m0_fop_put(cut->cu_createfops[i]);

        if (ftype2 != NULL)
                for (i = 0; i < cut->cu_cobfop_nr; ++i)
                        m0_fop_put(cut->cu_deletefops[i]);
        m0_rpc_machine_unlock(&cut->cu_cctx.rcx_rpc_machine);

        m0_free0(&cut->cu_createfops);
        m0_free0(&cut->cu_deletefops);
}

static void cobfops_threads_init(void)
{
        int i;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_thread_nr > 0);

        M0_ALLOC_ARR(cut->cu_threads, cut->cu_thread_nr);
        M0_UT_ASSERT(cut->cu_threads != NULL);

        for (i = 0; i < cut->cu_thread_nr; ++i) {
                M0_ALLOC_PTR(cut->cu_threads[i]);
                M0_UT_ASSERT(cut->cu_threads[i] != NULL);
        }
}

static void cobfops_threads_fini(void)
{
        int i;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_threads != NULL);
        M0_UT_ASSERT(cut->cu_thread_nr > 0);

        for (i = 0; i < cut->cu_thread_nr; ++i)
                m0_free(cut->cu_threads[i]);
        m0_free(cut->cu_threads);
}

static void cobfops_send_wait(struct cobthread_arg *arg)
{
        int i;
        int rc;
        struct m0_fop *fop;
        struct m0_fop_cob_op_reply *rfop;
        struct m0_fop_cob_common *common;

        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(arg != NULL);
        M0_UT_ASSERT(arg->ca_ftype != NULL);

        i = arg->ca_index;
        M0_LOG(M0_DEBUG, "i=%d", i);
        fop = arg->ca_ftype == &m0_fop_cob_create_fopt ? cut->cu_createfops[i] :
                cut->cu_deletefops[i];

        common = m0_cobfop_common_get(fop);
        common->c_flags = arg->ca_flags;
        common->c_pver = CONF_PVER_FID;
        M0_LOG(M0_DEBUG, "gobfid="FID_F" cobfid="FID_F,
                FID_P(&common->c_gobfid), FID_P(&common->c_cobfid));

        rc = m0_rpc_post_sync(fop, &cut->cu_cctx.rcx_session, NULL,
                              0 /* deadline */);
        M0_UT_ASSERT(rc == 0);
        rc = m0_rpc_item_wait_for_reply(&fop->f_item, M0_TIME_NEVER);
        M0_UT_ASSERT(rc == 0);
        rfop = m0_fop_data(m0_rpc_item_to_fop(fop->f_item.ri_reply));
        M0_UT_ASSERT(rfop->cor_rc == arg->ca_rc);
}

static void cobfoms_fops_dispatch(struct m0_fop_type *ftype, uint64_t flags,
                                  int expected_rc)
{
        int                   rc;
        uint64_t              i;
        struct cobthread_arg *arg;

        M0_UT_ASSERT(ftype != NULL);
        M0_UT_ASSERT(cut != NULL);
        M0_UT_ASSERT(cut->cu_createfops != NULL);
        M0_UT_ASSERT(cut->cu_cobfop_nr > 0);
        M0_UT_ASSERT(cut->cu_deletefops != NULL);
        M0_UT_ASSERT(cut->cu_thread_nr > 0);

        M0_ALLOC_ARR(arg, cut->cu_cobfop_nr);
        M0_UT_ASSERT(arg != NULL);

        for (i = 0; i < cut->cu_cobfop_nr; ++i) {
                arg[i].ca_ftype = ftype;
                arg[i].ca_index = i;
                arg[i].ca_flags = flags;
                arg[i].ca_rc = expected_rc;
                M0_SET0(cut->cu_threads[i]);
                rc = M0_THREAD_INIT(cut->cu_threads[i], struct cobthread_arg *,
                                    NULL, &cobfops_send_wait, &arg[i],
                                    ftype == &m0_fop_cob_create_fopt ?
                                    "cob_create" : "cob_delete");
                M0_UT_ASSERT(rc == 0);
        }

        for (i = 0; i < cut->cu_cobfop_nr; ++i)
                m0_thread_join(cut->cu_threads[i]);

        m0_free(arg);
}

static void cobfoms_fop_thread_init(uint64_t fop_nr, uint64_t thread_nr)
{
        M0_UT_ASSERT(fop_nr > 0 && thread_nr > 0);
        M0_UT_ASSERT(cut != NULL);

        cut->cu_cobfop_nr = fop_nr;
        cobfops_create();
        cut->cu_thread_nr = thread_nr;
        cobfops_threads_init();
}

static void cobfoms_fop_thread_fini(struct m0_fop_type *ftype1,
                                    struct m0_fop_type *ftype2)
{
        cobfops_destroy(ftype1, ftype2);
        cobfops_threads_fini();
}

static void cobfoms_send_internal(struct m0_fop_type *ftype1,
                                  struct m0_fop_type *ftype2,
                                  uint64_t flags,
                                  int rc1, int rc2,
                                  uint64_t nr)
{
        cobfoms_fop_thread_init(nr, nr);

        if (ftype1 != NULL)
                cobfoms_fops_dispatch(ftype1, flags, rc1);
        if (ftype2 != NULL)
                cobfoms_fops_dispatch(ftype2, flags, rc2);

        cobfoms_fop_thread_fini(ftype1, ftype2);
}

static void cobfoms_single(void)
{
        cobfoms_send_internal(&m0_fop_cob_create_fopt, &m0_fop_cob_delete_fopt,
                              0, 0, 0, COB_FOP_SINGLE);
}

/*
 * Sends multiple cob_create fops to same ioservice instance
 * so as to stress the fom code with multiple simultaneous requests.
 */
static void cobfoms_multiple(void)
{
        cobfoms_send_internal(&m0_fop_cob_create_fopt, &m0_fop_cob_delete_fopt,
                              0, 0, 0, COB_FOP_NR);
}

static void cobfoms_preexisting_cob(void)
{
        cobfoms_send_internal(&m0_fop_cob_create_fopt, NULL, 0, 0, 0,
                              COB_FOP_SINGLE);

        /*
         * Get the value of cobfoms_ut::cu_gobindex to send cob_create
         * fop and subsequence cob_delete fop with same fid.
         */
        --cut->cu_gobindex;
        cobfoms_send_internal(&m0_fop_cob_create_fopt, NULL, 0, -EEXIST, 0,
                              COB_FOP_SINGLE);

        --cut->cu_gobindex;

        /* Cleanup. */
        cobfoms_send_internal(NULL, &m0_fop_cob_delete_fopt, 0, 0, 0,
                              COB_FOP_SINGLE);
        cut->cu_gobindex++;
        cut->cu_gobindex++;
}

static void cobfoms_del_nonexist_cob(void)
{
        cobfoms_send_internal(NULL, &m0_fop_cob_delete_fopt, 0, 0, -ENOENT,
                              COB_FOP_SINGLE);
        cobfoms_send_internal(NULL, &m0_fop_cob_delete_fopt, M0_IO_FLAG_CROW,
                              0, 0, COB_FOP_SINGLE);
}

static void cobfoms_fsync_send_fop(struct m0_be_tx_remid *remid,
                                   int expected_rc);

static void cobfoms_fsync_nonexist_tx(void)
{
        struct m0_be_tx_remid   remid;

        remid.tri_txid = 666;
        remid.tri_locality = 0;

        cobfoms_fsync_send_fop(&remid, 0);
}

static void cobfoms_fsync_send_fop(struct m0_be_tx_remid       *remid,
                                   int                          expected_rc)
{
        int                             rc;
        struct m0_fop                  *fop;
        struct m0_fop_fsync            *ffop;
        struct m0_fop_fsync_rep        *rfop;
        struct m0_rpc_machine          *machine;

        machine = session_machine(&cut->cu_cctx.rcx_session);

        /* create a fop from fsync_fopt */
        fop = m0_fop_alloc(&m0_fop_fsync_ios_fopt, NULL, machine);

        /* populate fop */
        ffop = m0_fop_data(fop);
        ffop->ff_be_remid.tri_txid =  remid->tri_txid;
        ffop->ff_be_remid.tri_locality = remid->tri_locality;
        ffop->ff_fsync_mode = M0_FSYNC_MODE_ACTIVE;

        /* send fop */
        rc = m0_rpc_post_sync(fop, &cut->cu_cctx.rcx_session,
                              NULL, 0 /* deadline */);
        M0_UT_ASSERT(rc == 0);
        rc = m0_rpc_item_wait_for_reply(&fop->f_item, M0_TIME_NEVER);
        M0_UT_ASSERT(rc == 0);
        rfop = m0_fop_data(m0_rpc_item_to_fop(fop->f_item.ri_reply));

        /* such tx shouldn't exist */
        M0_UT_ASSERT(rfop->ffr_rc == expected_rc);
        M0_UT_ASSERT(rfop->ffr_be_remid.tri_txid == remid->tri_txid);

        /* release structs */
        m0_fop_put_lock(fop);
}

static void cobfoms_fsync_create_delete(void)
{
        int                             rc;
        int                             i;
        struct m0_fop                 **fops;
        struct m0_fop_cob_op_reply     *rfop;
        /* how many create-delete pairs are we issuing? */
        uint32_t                        cobfop_nr = 5;
        struct m0_be_tx_remid           remid;
        struct m0_rpc_machine          *machine;

        machine = session_machine(&cut->cu_cctx.rcx_session);

        /* allocate the required fops */
        M0_ALLOC_ARR(fops, cobfop_nr*2);
        M0_UT_ASSERT(fops != NULL);

        /* fill the aux fop arrays with create-delete pairs */
        for (i = 0; i < cobfop_nr; ++i ) {
                /* allocate and fill with the right fid */
                fops[i] = m0_fop_alloc(&m0_fop_cob_create_fopt, NULL, machine);
                M0_UT_ASSERT(fops[i] != NULL);
                cobfops_populate_internal(fops[i], cut->cu_gobindex);

                fops[i+cobfop_nr] = m0_fop_alloc(&m0_fop_cob_delete_fopt,
                                                 NULL, machine);
                M0_UT_ASSERT(fops[i+cobfop_nr] != NULL);
                cobfops_populate_internal(fops[i+cobfop_nr], cut->cu_gobindex);

                /* cu_goibindex is shared across all the suite's tests */
                ++cut->cu_gobindex;
        }

        /* send the fops and keep the txid returned by the last one */
        for (i = 0; i < cobfop_nr*2; ++i ) {
                /* sequentially, no need to use threads here */
                rc = m0_rpc_post_sync(fops[i], &cut->cu_cctx.rcx_session,
                                      NULL, 0 /* deadline */);
                M0_UT_ASSERT(rc == 0);
                rc = m0_rpc_item_wait_for_reply(
                        &fops[i]->f_item, M0_TIME_NEVER);
                M0_UT_ASSERT(rc == 0);
                rfop = (struct m0_fop_cob_op_reply *)
                        m0_fop_data(
                                m0_rpc_item_to_fop(fops[i]->f_item.ri_reply));
                M0_UT_ASSERT(rfop->cor_rc == 0);
                remid.tri_txid =
                        rfop->cor_common.cor_mod_rep.fmr_remid.tri_txid;
                remid.tri_locality =
                        rfop->cor_common.cor_mod_rep.fmr_remid.tri_locality;
        }

        /* now try to fsync the last transaction and check the op. succeeds */
        cobfoms_fsync_send_fop(&remid, 0);

        /* release all the fops */
        for (i = 0; i < cobfop_nr*2; ++i) {
                m0_fop_put_lock(fops[i]);
        }
        m0_free(fops);
}

extern const struct m0_sm_conf cob_ops_conf;

/*
 * Create COB FOMs - create or delete
 */
static void fom_create(struct m0_fom **fom, enum cob_fom_type fomtype)
{
        struct m0_fom          *base_fom;
        struct m0_reqh         *reqh;
        int                     rc;

        rc = cob_op_fom_create(fom);
        M0_UT_ASSERT(rc == 0);

        base_fom = *fom;
        reqh = m0_cs_reqh_get(&cut->cu_sctx.rsx_motr_ctx);
        m0_fom_init(base_fom, &ft,
                    fomtype == COB_CREATE ? &cc_fom_ops : &cd_fom_ops,
                    NULL, NULL, reqh);

        base_fom->fo_service = m0_reqh_service_find(ft.ft_rstype, reqh);
        M0_UT_ASSERT(base_fom->fo_service != NULL);

        base_fom->fo_loc = &dummy_loc;
        m0_fom_locality_inc(base_fom);
        base_fom->fo_type = &ft;

        m0_fom_sm_init(base_fom);
        m0_fol_rec_init(&base_fom->fo_tx.tx_fol_rec, &reqh->rh_fol);
}

/*
 * Delete COB FOMs - create or delete
 */
static void fom_fini(struct m0_fom *fom, enum cob_fom_type fomtype)
{
        m0_ut_fom_phase_set(fom, M0_FOPH_FINISH);

        m0_fol_rec_fini(&fom->fo_tx.tx_fol_rec);
        switch (fomtype) {
        case COB_CREATE:
                cc_fom_fini(fom);
                break;
        case COB_DELETE:
                cd_fom_fini(fom);
                break;
        default:
                M0_IMPOSSIBLE("Invalid COB-FOM type");
        }
}

/*
 * Allocate desired FOP and populate test-data in it.
 */
static void fop_alloc(struct m0_fom *fom, enum cob_fom_type fomtype)
{
        struct m0_fop_cob_common *c;
        struct m0_fop            *base_fop;
        struct m0_rpc_machine    *mach = &cut->cu_cctx.rcx_rpc_machine;
        struct m0_cob_attr        attr = { { 0, } };

        switch (fomtype) {
        case COB_CREATE:
                base_fop = m0_fop_alloc(&m0_fop_cob_create_fopt, NULL, mach);
                M0_UT_ASSERT(base_fop != NULL);
                break;
        case COB_DELETE:
                base_fop = m0_fop_alloc(&m0_fop_cob_delete_fopt, NULL, mach);
                M0_UT_ASSERT(base_fop != NULL);
                break;
        default:
                M0_IMPOSSIBLE("Invalid COB-FOM type");
                base_fop = NULL;
                break;
        }
        c = m0_cobfop_common_get(base_fop);
        c->c_pver = CONF_PVER_FID;
        m0_fid_gob_make(&c->c_gobfid, 0, COB_TEST_KEY);
        m0_fid_convert_gob2cob(&c->c_gobfid, &c->c_cobfid,
                               M0_AD_STOB_DOM_KEY_DEFAULT);

        cob_attr_default_fill(&attr);
        attr.ca_nlink = base_fop->f_type == &m0_fop_cob_create_fopt ? 1 : 0;

        m0_md_cob_mem2wire(&c->c_body, &attr);

        c->c_cob_idx = M0_AD_STOB_DOM_KEY_DEFAULT;
        fom->fo_fop = base_fop;
        fom->fo_type = &base_fop->f_type->ft_fom_type;

        fom->fo_rep_fop = m0_fop_alloc(&m0_fop_cob_op_reply_fopt, NULL, mach);
        M0_UT_ASSERT(fom->fo_rep_fop != NULL);
}

/*
 * A generic COB-FOM-delete verification function.
 */
static void fom_fini_test(enum cob_fom_type fomtype)
{
        struct m0_fom  *fom;
        struct m0_reqh *reqh;

        /*
         * 1. Allocate FOM object of interest
         * 2. Calculate memory usage before and after object allocation
         *    and de-allocation.
         * 3. Before taking memory record, make sure there are no
         *    stray foms around.
         */
        reqh = m0_cs_reqh_get(&cut->cu_sctx.rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);
        fom_create(&fom, fomtype);
        fom_fini(fom, fomtype);
}

/*
 * A generic COB-FOM test function that validates the sub-class FOM object.
 */
static void fom_get_test(enum cob_fom_type fomtype)
{
        struct m0_fom        *fom;
        struct m0_fom_cob_op *cc;

        fom_create(&fom, fomtype);
        M0_UT_ASSERT(fom != NULL);

        cc = cob_fom_get(fom);
        M0_UT_ASSERT(cc != NULL);
        M0_UT_ASSERT(&cc->fco_fom == fom);
        fom_fini(fom, fomtype);
}

/*
 * A generic test to verify COM-FOM create functions.
 */
static void fom_create_test(enum cob_fom_type fomtype)
{
        struct m0_fom *fom;

        fom_create(&fom, fomtype);
        M0_UT_ASSERT(fom != NULL);
        fom_fini(fom, fomtype);
}

/*
 * Delete COB-create FOM.
 */
static void cc_fom_dealloc(struct m0_fom *fom)
{
        m0_ut_fom_phase_set(fom, M0_FOPH_FINISH);

        cc_fom_fini(fom);
}

/*
 * Create COB-create FOM and populate it with testdata.
 */
static struct m0_fom *cc_fom_alloc()
{
        struct m0_fom *fom = NULL;

        fom_create(&fom, COB_CREATE);
        M0_UT_ASSERT(fom != NULL);

        fop_alloc(fom, COB_CREATE);
        M0_UT_ASSERT(fom->fo_fop != NULL);
        cob_fom_populate(fom);
        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_PREPARE);

        return fom;
}

/*
 * Test function for cc_fom_create().
 */
static void cc_fom_create_test()
{
        fom_create_test(COB_CREATE);
}

/*
 * Test function for cc_fom_fini().
 */
static void cc_fom_fini_test()
{
        fom_fini_test(COB_CREATE);
}

/*
 * Test function for cc_fom_get().
 */
static void cc_fom_get_test()
{
        fom_get_test(COB_CREATE);
}

/*
 * Test function for m0_cc_stob_create().
 */
static void cc_stob_create_test()
{
        int                   rc;
        struct m0_fom        *fom;
        struct m0_fom_cob_op *cc;
        struct m0_sm_group   *grp = m0_locality0_get()->lo_grp;

        fom = cc_fom_alloc();
        M0_UT_ASSERT(fom != NULL);

        cc = cob_fom_get(fom);

        fom_dtx_init(fom, grp, M0_COB_OP_CREATE);
        fom_stob_tx_credit(fom, M0_COB_OP_CREATE);
        rc = m0_cc_stob_create(fom, &cc->fco_stob_id);
        M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_COB_OPS_PREPARE);
        fom_dtx_done(fom, grp);

        M0_UT_ASSERT(rc == 0);

        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(fom, M0_FOPH_SUCCESS);

        cc_fom_dealloc(fom);
}

/*
 * Test function to check COB record in the database.
 */
static void cob_verify(struct m0_fom *fom, const bool exists)
{
        int                   rc;
        struct m0_cob_domain *cobdom;
        struct m0_cob_nskey  *nskey;
        struct m0_fid         gfid;
        struct m0_fid         cfid;
        char                  nskey_bs[UINT32_STR_LEN];
        uint32_t              nskey_bs_len;
        uint32_t              cob_idx = M0_AD_STOB_DOM_KEY_DEFAULT;

        m0_fid_gob_make(&gfid, 0, COB_TEST_KEY);
        m0_fid_convert_gob2cob(&gfid, &cfid, M0_AD_STOB_DOM_KEY_DEFAULT);
        m0_ios_cdom_get(m0_fom_reqh(fom), &cobdom);
        M0_UT_ASSERT(cobdom != NULL);

        snprintf(nskey_bs, UINT32_STR_LEN, "%u", cob_idx);
        nskey_bs_len = strlen(nskey_bs);

        rc = m0_cob_nskey_make(&nskey, &gfid, nskey_bs, nskey_bs_len);
        M0_UT_ASSERT(rc == 0);

        test_cob = NULL;
        rc = m0_cob_lookup(cobdom, nskey, 0, &test_cob);

        if (exists) {
                M0_UT_ASSERT(rc == 0);
                M0_UT_ASSERT(test_cob != NULL);
                M0_UT_ASSERT(test_cob->co_flags & M0_CA_NSREC);
        } else
                M0_UT_ASSERT(rc == -ENOENT);
        if (rc != 0)
                m0_free(nskey);
}

static void md_cob_fop_create_delete_test(bool create_p,
                                          const struct m0_fid *pver,
                                          int expected_reply_rc)
{
        struct m0_fop              *fop;
        struct m0_fop_cob_common   *common;
        struct m0_fop_cob_op_reply *reply;
        int                         rc;

        cut->cu_gobindex = 0;
        cobfops_create();
        fop = *(create_p ? cut->cu_createfops : cut->cu_deletefops);
        common = m0_cobfop_common_get(fop);
        common->c_body.b_valid = M0_COB_PVER | M0_COB_NLINK;
        common->c_body.b_pver  = *pver;
        common->c_body.b_nlink = !!create_p;
        common->c_cob_type = M0_COB_MD;
        rc = m0_rpc_post_sync(fop, &cut->cu_cctx.rcx_session, NULL,
                              0 /* deadline */);
        M0_UT_ASSERT(rc == 0);
        rc = m0_rpc_item_wait_for_reply(&fop->f_item, M0_TIME_NEVER);
        M0_UT_ASSERT(rc == 0);
        reply = m0_fop_data(m0_rpc_item_to_fop(fop->f_item.ri_reply));
        M0_UT_ASSERT(reply->cor_rc == expected_reply_rc);
        cobfops_destroy(&m0_fop_cob_create_fopt, &m0_fop_cob_delete_fopt);
}

static void md_cob_create_delete()
{
        M0_UT_ASSERT(cut != NULL);
        cut->cu_cobfop_nr = 2;
        md_cob_fop_create_delete_test(true, &CONF_PVER_FID, 0);
        /* Create the same mdcob again. */
        md_cob_fop_create_delete_test(true, &CONF_PVER_FID, -EEXIST);
        /* Create the same mdcob with different pool version. */
        md_cob_fop_create_delete_test(true, &CONF_PVER_FID1, 0);
        /* Delete the mdcob. */
        md_cob_fop_create_delete_test(false, &CONF_PVER_FID1, 0);
}

/*
 * Test function for cc_cob_create().
 */
static void cc_cob_create_test()
{
        int                   rc;
        struct m0_fom        *fom;
        struct m0_sm_group   *grp = m0_locality0_get()->lo_grp;
        struct m0_fom_cob_op *cc;
        struct m0_cob_attr    attr = { { 0, } };

        cob_attr_default_fill(&attr);

        fom = cc_fom_alloc();
        M0_UT_ASSERT(fom != NULL);
        cc = cob_fom_get(fom);

        /* Create md cob */
        fom_dtx_init(fom, grp, M0_COB_OP_CREATE);
        rc = m0_dtx_open_sync(&fom->fo_tx);
        M0_UT_ASSERT(rc == 0);
        cc->fco_cob_type = M0_COB_MD;
        rc = cc_cob_create(fom, cc, &attr);
        if (!cc_cob_create_test_done)
                M0_UT_ASSERT(rc == 0);
        else
                M0_UT_ASSERT(rc == -EEXIST);
        fom_dtx_done(fom, grp);
        cc->fco_cob_type = M0_COB_IO;
        /*
         * Set the FOM phase and set transaction context
         * Test-case 1: Test successful creation of COB
         */

        fom_dtx_init(fom, grp, M0_COB_OP_CREATE);
        fom_stob_tx_credit(fom, M0_COB_OP_CREATE);
        /*
         * Create STOB first.
         */
        rc = m0_cc_stob_create(fom, &cc->fco_stob_id);
        /* stob may be already created by another test */
        M0_UT_ASSERT(M0_IN(rc, (0, -EEXIST)));

        rc = cc_cob_create(fom, cc, &attr);
        fom_dtx_done(fom, grp);

        M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_COB_OPS_PREPARE);
        M0_UT_ASSERT(rc == 0);

        /*
         * Test-case 1 - Verify COB creation
         */
        cob_verify(fom, true);

        /*
         * Test-case 2 - Test failure case. Try to create the
         * same COB.
         */
        fom_dtx_init(fom, grp, M0_COB_OP_CREATE);
        rc = m0_dtx_open_sync(&fom->fo_tx);
        M0_UT_ASSERT(rc == 0);
        rc = cc_cob_create(fom, cc, &attr);
        M0_UT_ASSERT(rc != 0);
        fom_dtx_done(fom, grp);

        /*
         * Start cleanup by deleting the COB
         */
        fom_dtx_init(fom, grp, M0_COB_OP_DELETE_PUT);
        rc = m0_dtx_open_sync(&fom->fo_tx);
        M0_UT_ASSERT(rc == 0);
        rc = m0_cob_delete_put(test_cob, m0_fom_tx(fom));
        M0_UT_ASSERT(rc == 0);
        fom_dtx_done(fom, grp);
        test_cob = NULL;

        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(fom, M0_FOPH_SUCCESS);
        cc_fom_dealloc(fom);
}

/*
 * Test function for create.
 */
static void cc_fom_state_test(void)
{
        int                   rc;
        struct m0_fom        *cfom;
        struct m0_fom        *dfom;
        struct m0_sm_group   *grp = m0_locality0_get()->lo_grp;

        /* delete existing stob */
        cd_stob_delete_test();

        cfom = cc_fom_alloc();
        M0_UT_ASSERT(cfom != NULL);

        fom_dtx_init(cfom, grp, M0_COB_OP_CREATE);
        fom_stob_tx_credit(cfom, M0_COB_OP_CREATE);
        rc = cob_ops_fom_tick(cfom);
        M0_UT_ASSERT(m0_fom_rc(cfom) == 0);
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        rc = cob_ops_fom_tick(cfom);
        fom_dtx_done(cfom, grp);

        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        M0_UT_ASSERT(m0_fom_phase(cfom) == M0_FOPH_SUCCESS);

        cob_verify(cfom, true);

        /*
         * Now create delete fom. Use FOM functions to delete cob-data.
         */
        dfom = cd_fom_alloc();
        M0_UT_ASSERT(dfom != NULL);

        fom_dtx_init(dfom, grp, M0_COB_OP_DELETE);
        fom_stob_tx_credit(dfom, M0_COB_OP_DELETE);
        m0_stob_delete_mark(cob_fom_get(dfom)->fco_stob);
        rc = cob_ops_fom_tick(dfom); /* for M0_FOPH_COB_OPS_PREPARE */
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        rc = cob_ops_fom_tick(dfom); /* for M0_FOPH_COB_OPS_EXECUTE */
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        M0_UT_ASSERT(m0_fom_phase(dfom) == M0_FOPH_SUCCESS);

        fom_dtx_done(dfom, grp);

        cc_fom_dealloc(cfom);
        cd_fom_dealloc(dfom);
}

/*
 * Test function for cc_fom_populate().
 */
static void cc_fom_populate_test()
{
        struct m0_fom        *fom;
        struct m0_fom_cob_op *cc;

        fom = cc_fom_alloc();
        M0_UT_ASSERT(fom != NULL);

        cc = cob_fom_get(fom);
        M0_UT_ASSERT(cc->fco_cfid.f_key == COB_TEST_KEY);
        M0_UT_ASSERT(m0_fid_cob_device_id(&cc->fco_cfid) ==
                        M0_AD_STOB_DOM_KEY_DEFAULT);
        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(fom, M0_FOPH_SUCCESS);
        cc_fom_dealloc(fom);
}

/*
 *****************
 * COB delete-FOM test functions
 ******************
 */

/*
 * Delete COB-delete FOM object.
 */
static void cd_fom_dealloc(struct m0_fom *fom)
{
        m0_ut_fom_phase_set(fom, M0_FOPH_FINISH);
        cd_fom_fini(fom);
}

/*
 * Create COB-delete FOM and populate it with testdata.
 */
static struct m0_fom *cd_fom_alloc()
{
        struct m0_fom *fom = NULL;

        fom_create(&fom, COB_DELETE);
        M0_UT_ASSERT(fom != NULL);

        fop_alloc(fom, COB_DELETE);
        M0_UT_ASSERT(fom->fo_fop != NULL);
        cob_fom_populate(fom);
        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_PREPARE);

        return fom;
}

/*
 * Test function for cd_fom_create().
 */
static void cd_fom_create_test()
{
        fom_create_test(COB_DELETE);
}

/*
 * Test function for cd_fom_fini().
 */
static void cd_fom_fini_test()
{
        fom_fini_test(COB_DELETE);
}

/*
 * Test function for cd_fom_get().
 */
static void cd_fom_get_test()
{
        fom_get_test(COB_DELETE);
}

/*
 * Test function for cd_fom_populate().
 */
static void cd_fom_populate_test()
{
        struct m0_fom        *fom;
        struct m0_fom_cob_op *cd;

        fom = cd_fom_alloc();
        M0_UT_ASSERT(fom != NULL);

        cd = cob_fom_get(fom);
        M0_UT_ASSERT(cd->fco_cfid.f_key == COB_TEST_KEY);
        M0_UT_ASSERT(m0_fid_cob_device_id(&cd->fco_cfid) ==
                        M0_AD_STOB_DOM_KEY_DEFAULT);
        m0_fom_phase_set(fom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(fom, M0_FOPH_SUCCESS);
        cd_fom_dealloc(fom);
}

/*
 * Before testing COB-delete FOM functions, create COB testdata.
 */
static struct m0_fom *cob_testdata_create()
{
        struct m0_sm_group   *grp = m0_locality0_get()->lo_grp;
        struct m0_fom        *fom;
        int                   rc;

        /*
         * Create cob-create FOM.
         * Crate COB and related meta-data.
         */
        fom = cc_fom_alloc();
        M0_UT_ASSERT(fom != NULL);

        fom_dtx_init(fom, grp, M0_COB_OP_CREATE);
        fom_stob_tx_credit(fom, M0_COB_OP_CREATE);
        rc = cob_ops_fom_tick(fom);
        M0_UT_ASSERT(m0_fom_rc(fom) == 0);
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        rc = cob_ops_fom_tick(fom);
        fom_dtx_done(fom, grp);

        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_SUCCESS);

        return fom;
}

/*
 * Delete COB testdata. In this case we delete COB-create FOM.
 */
static void cob_testdata_cleanup(struct m0_fom *fom)
{
        cc_fom_dealloc(fom);
}

/*
 * Test function for cd_stob_delete()
 */
static void cd_stob_delete_test()
{
        struct m0_fom_cob_op     *cd;
        struct m0_fom_cob_op     *cc;
        struct m0_fom            *cfom;
        struct m0_fom            *dfom;
        int                       rc;
        struct m0_sm_group      *grp = m0_locality0_get()->lo_grp;

        cfom = cc_fom_alloc();
        M0_UT_ASSERT(cfom != NULL);
        cc = cob_fom_get(cfom);
        fom_dtx_init(cfom, grp, M0_COB_OP_CREATE);
        fom_stob_tx_credit(cfom, M0_COB_OP_CREATE);
        rc = m0_cc_stob_create(cfom, &cc->fco_stob_id);
        M0_UT_ASSERT(M0_IN(rc, (0, -EEXIST)));
        fom_dtx_done(cfom, grp);

        /* Test stob delete after it has been created */
        dfom = cd_fom_alloc();
        M0_UT_ASSERT(dfom != NULL);

        cd = cob_fom_get(dfom);
        fom_dtx_init(dfom, grp, M0_COB_OP_DELETE);
        fom_stob_tx_credit(dfom, M0_COB_OP_DELETE);
        m0_stob_delete_mark(cd->fco_stob);
        rc = ce_stob_edit(dfom, cd, M0_COB_OP_DELETE);
        M0_UT_ASSERT(m0_fom_phase(dfom) == M0_FOPH_COB_OPS_PREPARE);
        M0_ASSERT(rc == 0);
        fom_dtx_done(dfom, grp);

        m0_fom_phase_set(dfom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(dfom, M0_FOPH_SUCCESS);
        cd_fom_dealloc(dfom);
        m0_fom_phase_set(cfom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(cfom, M0_FOPH_SUCCESS);
        cc_fom_dealloc(cfom);
}

/*
 * Test function for cd_cob_delete()
 */
static void cd_cob_delete_test()
{
        int                   rc;
        struct m0_fom        *cfom;
        struct m0_fom        *dfom;
        struct m0_fom_cob_op *cd;
        struct m0_sm_group   *grp = m0_locality0_get()->lo_grp;
        struct m0_cob_attr    attr = { { 0, } };

        attr.ca_ctime = 1416970585;
        attr.ca_nlink = 0;
        attr.ca_valid = M0_COB_CTIME | M0_COB_NLINK;


        cfom = cob_testdata_create();

        /* Test COB delete after COB has been created */
        dfom = cd_fom_alloc();
        M0_UT_ASSERT(dfom != NULL);

        cd = cob_fom_get(dfom);
        /*
         * Test-case 1: Delete cob. The test should succeed.
         */
        fom_dtx_init(dfom, grp, M0_COB_OP_DELETE);
        rc = m0_dtx_open_sync(&dfom->fo_tx);
        M0_UT_ASSERT(rc == 0);
        rc = cd_cob_delete(dfom, cd, &attr);
        M0_UT_ASSERT(m0_fom_phase(dfom) == M0_FOPH_COB_OPS_PREPARE);
        M0_UT_ASSERT(rc == 0);

        fom_dtx_done(dfom, grp);

        /*
         * Make sure that there no entry in the database.
         */
        cob_verify(cfom, false);

        /*
         * Test-case 2: Delete cob again. The test should fail.
         */
        fom_dtx_init(dfom, grp, M0_COB_OP_DELETE);
        fom_stob_tx_credit(dfom, M0_COB_OP_DELETE);
        m0_stob_delete_mark(cd->fco_stob);
        rc = cd_cob_delete(dfom, cd, &attr);
        M0_UT_ASSERT(rc != 0);

        /*
         * Now do the cleanup.
         */
        rc = ce_stob_edit(dfom, cd, M0_COB_OP_DELETE);
        M0_UT_ASSERT(rc == 0);
        fom_dtx_done(dfom, grp);

        m0_fom_phase_set(dfom, M0_FOPH_COB_OPS_EXECUTE);
        m0_fom_phase_set(dfom, M0_FOPH_SUCCESS);
        cd_fom_dealloc(dfom);
        cob_testdata_cleanup(cfom);
}

/*
 * Test function for cob_ops_fom_tick()
 */
static void cd_fom_state_test(void)
{
        struct m0_fom            *cfom;
        struct m0_fom            *dfom;
        struct m0_sm_group       *grp = m0_locality0_get()->lo_grp;
        int                       rc;

        cfom = cob_testdata_create();

        /* Test if COB-map got deleted */
        dfom = cd_fom_alloc();
        M0_UT_ASSERT(dfom != NULL);

        fom_dtx_init(dfom, grp, M0_COB_OP_DELETE);
        fom_stob_tx_credit(dfom, M0_COB_OP_DELETE);
        m0_stob_delete_mark(cob_fom_get(dfom)->fco_stob);
        rc = cob_ops_fom_tick(dfom);
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        rc = cob_ops_fom_tick(dfom);
        M0_UT_ASSERT(m0_fom_phase(dfom) == M0_FOPH_SUCCESS);
        M0_UT_ASSERT(rc == M0_FSO_AGAIN);
        fom_dtx_done(dfom, grp);


        cd_fom_dealloc(dfom);
        cob_testdata_cleanup(cfom);
}

static void dummy_locality_setup()
{
        dummy_loc.fl_dom = m0_fom_dom();
        m0_sm_group_init(&dummy_loc.fl_group);
        m0_locality_lockers_init(&dummy_loc.fl_locality);
}

static void cob_create_api_test(void)
{
        /* Dummy locality setup */
        dummy_locality_setup();
        m0_sm_group_lock(&dummy_loc.fl_group);

        /* Test for cc_fom_create() */
        cc_fom_create_test();

        /* Test for cc_fom_fini() */
        cc_fom_fini_test();

        /* Test for cc_fom_get() */
        cc_fom_get_test();

        /* Test cc_fom_populate() */
        cc_fom_populate_test();

        /* Test m0_cc_stob_create() */
        cc_stob_create_test();

        /* Test cc_cob_create() */
        cc_cob_create_test();
        cc_cob_create_test_done = true;

        /* Test for cob_ops_fom_tick() */
        cc_fom_state_test();

        m0_sm_group_unlock(&dummy_loc.fl_group);
}

static void cob_delete_api_test(void)
{
        m0_sm_group_lock(&dummy_loc.fl_group);

        /* Test for cd_fom_create() */
        cd_fom_create_test();

        /* Test for cd_fom_fini() */
        cd_fom_fini_test();

        /* Test for cd_fom_fini() */
        cd_fom_get_test();

        /* Test cd_fom_populate() */
        cd_fom_populate_test();

        /* Test cd_stob_delete() */
        cd_stob_delete_test();

        /* Test cd_cob_delete() */
        cd_cob_delete_test();

        /* Test for cob_ops_fom_tick() */
        cd_fom_state_test();

        m0_sm_group_unlock(&dummy_loc.fl_group);
}

# if 0
#define COB_DATA(data) M0_XCODE_OBJ(m0_fop_cob_common_xc, data)

static int cob_cd_op(struct m0_fol_rec *rec, struct m0_fop *fop, bool undo) {
        struct m0_fol_frag       *dec_frag;
        struct m0_fop_cob_common *cob_cmn;
        int                       result = 0;

        cob_cmn =  m0_cobfop_common_get(fop);
        m0_tl_for(m0_rec_frag, &rec->fr_frags, dec_frag) {
                if (dec_frag->rp_ops->rpo_type->rpt_index ==
                    m0_fop_fol_frag_type.rpt_index) {
                        struct m0_fop_cob_common   *cob_data;
                        struct m0_fop_fol_frag     *fp_frag;
                        struct m0_fop_type         *ftype;
                        struct m0_fop_cob_op_reply *cob_rep;

                        fp_frag = dec_frag->rp_data;
                        cob_rep = fp_frag->ffrp_rep;

                        cob_data = m0_is_cob_create_fop(fop) ?
                                &((struct m0_fop_cob_create *)
                                fp_frag->ffrp_fop)->cc_common :
                                &((struct m0_fop_cob_delete *)
                                fp_frag->ffrp_fop)->cd_common;

                        M0_UT_ASSERT(m0_xcode_cmp(&COB_DATA(cob_data),
                                                  &COB_DATA(cob_cmn)) == 0);
                        M0_UT_ASSERT(cob_rep->cor_rc == 0);

                        ftype = m0_fop_type_find(fp_frag->ffrp_fop_code);
                        M0_UT_ASSERT(ftype != NULL);
                        M0_UT_ASSERT(ftype->ft_ops->fto_undo != NULL &&
                                     ftype->ft_ops->fto_redo != NULL);
                        result = undo ?
                                 ftype->ft_ops->fto_undo(fp_frag, rec->fr_fol) :
                                 ftype->ft_ops->fto_redo(fp_frag, rec->fr_fol);
                        M0_UT_ASSERT(result == 0);
                }
        } m0_tl_endfor;

        return result;
}

static void cobfoms_fol_verify(void)
{
        struct m0_reqh    *reqh;
        struct m0_fol_rec  dec_cc_rec;
        struct m0_fol_rec  dec_cd_rec;
        int                result;
        struct m0_fop     *c_fop;
        struct m0_fop     *d_fop;
        struct m0_be_tx   *cctx;
        struct m0_be_tx   *cdtx;

        cobfoms_fop_thread_init(1, 1);
        cobfoms_fops_dispatch(&m0_fop_cob_create_fopt, 0);
        cobfoms_fops_dispatch(&m0_fop_cob_delete_fopt, 0);

        c_fop = cut->cu_createfops[0];
        d_fop = cut->cu_deletefops[0];

        reqh = m0_cs_reqh_get(&cut->cu_sctx.rsx_motr_ctx);
        result = m0_fol_rec_lookup(reqh->rh_fol,
                                   reqh->rh_fol->f_lsn - 2, &dec_cc_rec);
        M0_UT_ASSERT(result == 0);
        M0_UT_ASSERT(dec_cc_rec.fr_header.rh_frags_nr == 1);

        result = m0_fol_rec_lookup(reqh->rh_fol,
                                   reqh->rh_fol->f_lsn - 1, &dec_cd_rec);
        M0_UT_ASSERT(result == 0);
        M0_UT_ASSERT(dec_cd_rec.fr_header.rh_frags_nr == 1);

        /* Perform undo operations */
        result = cob_cd_op(&dec_cd_rec, d_fop, true);
        M0_UT_ASSERT(result == 0);
        result = cob_cd_op(&dec_cc_rec, c_fop, true);
        M0_UT_ASSERT(result == 0);

        /* Perform redo operations */
        result = cob_cd_op(&dec_cd_rec, d_fop, false);
        M0_UT_ASSERT(result == 0);
        result = cob_cd_op(&dec_cc_rec, c_fop, false);
        M0_UT_ASSERT(result == 0);

        m0_fol_rec_fini(&dec_cc_rec);
        m0_fol_rec_fini(&dec_cd_rec);

        cobfoms_fop_thread_fini(&m0_fop_cob_create_fopt, &m0_fop_cob_delete_fopt);
}
#endif

static void fom_stob_tx_credit(struct m0_fom *fom, enum m0_cob_op opcode)
{
        struct m0_fom_cob_op *co;
        int                   rc;

        co = cob_fom_get(fom);
        if (opcode == M0_COB_OP_DELETE) {
                rc = cob_ops_stob_find(co);
                M0_ASSERT(rc == 0);
                rc = ce_stob_edit_credit(fom, co, m0_fom_tx_credit(fom),
                                         opcode);
        } else
                rc = m0_cc_stob_cr_credit(&co->fco_stob_id,
                                          m0_fom_tx_credit(fom));
        M0_UT_ASSERT(rc == 0);
        rc = m0_dtx_open_sync(&fom->fo_tx);
        M0_UT_ASSERT(rc == 0);
}

static void fom_dtx_init(struct m0_fom *fom, struct m0_sm_group *grp,
                         enum m0_cob_op opcode)
{
        struct m0_be_seg     *beseg;
        struct m0_fom_cob_op *co;

        beseg = m0_fom_reqh(fom)->rh_beseg;
        m0_sm_group_lock(grp);
        M0_SET0(&fom->fo_tx);
        m0_dtx_init(&fom->fo_tx, beseg->bs_domain, grp);
        cob_op_credit(fom, opcode, m0_fom_tx_credit(fom));
        co = cob_fom_get(fom);
        co->fco_is_done = true;
}

static void fom_dtx_done(struct m0_fom *fom, struct m0_sm_group *grp)
{
        m0_dtx_done_sync(&fom->fo_tx);
        m0_dtx_fini(&fom->fo_tx);
        m0_sm_group_unlock(grp);
}

enum {
        FID_UT_CONTAINER_BITS_MAX = 32,
        FID_UT_KEY_BITS_MAX       = 64,
        FID_UT_DEVICE_ID_BITS_MAX = 24,
};

static void fid_convert_ut_check(uint32_t container,
                                 uint64_t key,
                                 uint32_t device_id)
{
        struct m0_fid     gob_fid;
        struct m0_fid     gob_fid2;
        struct m0_fid     cob_fid;
        struct m0_fid     cob_fid2;
        struct m0_stob_id stob_id;
        struct m0_stob_id stob_id2;
        struct m0_fid     bstore_fid;
        uint8_t           type_id_gob       = m0_file_fid_type.ft_id;
        uint8_t           type_id_cob       = m0_cob_fid_type.ft_id;
        uint8_t           type_id_adstob    = m0_stob_ad_type.st_fidt.ft_id;
        uint8_t           type_id_linuxstob = m0_stob_linux_type.st_fidt.ft_id;

        m0_fid_gob_make(&gob_fid, container, key);
        M0_UT_ASSERT(m0_fid_validate_gob(&gob_fid));
        M0_UT_ASSERT(m0_fid_tget(&gob_fid) == type_id_gob);

        m0_fid_convert_gob2cob(&gob_fid, &cob_fid, device_id);
        M0_UT_ASSERT(m0_fid_tget(&cob_fid) == type_id_cob);
        M0_UT_ASSERT(m0_fid_cob_device_id(&cob_fid) == device_id);
        M0_UT_ASSERT(m0_fid_validate_cob(&cob_fid));
        m0_fid_convert_cob2gob(&cob_fid, &gob_fid2);
        M0_UT_ASSERT(m0_fid_validate_gob(&gob_fid2));
        M0_UT_ASSERT(m0_fid_eq(&gob_fid, &gob_fid2));

        m0_fid_convert_cob2adstob(&cob_fid, &stob_id);
        M0_UT_ASSERT(m0_fid_tget(&stob_id.si_fid)        == type_id_adstob);
        M0_UT_ASSERT(m0_fid_tget(&stob_id.si_domain_fid) == type_id_adstob);
        M0_UT_ASSERT(m0_fid_validate_adstob(&stob_id));
        m0_fid_convert_adstob2cob(&stob_id, &cob_fid2);
        M0_UT_ASSERT(m0_fid_validate_cob(&cob_fid2));
        M0_UT_ASSERT(m0_fid_eq(&cob_fid, &cob_fid2));

        m0_fid_convert_adstob2bstore(&stob_id.si_domain_fid, &bstore_fid);
        M0_UT_ASSERT(m0_fid_tget(&bstore_fid) == type_id_linuxstob);
        M0_UT_ASSERT(m0_fid_validate_bstore(&bstore_fid));
        m0_fid_convert_bstore2adstob(&bstore_fid, &stob_id2.si_domain_fid);
        stob_id2.si_fid = stob_id.si_fid;
        M0_UT_ASSERT(m0_fid_validate_adstob(&stob_id2));
        M0_UT_ASSERT(m0_fid_eq(&stob_id.si_domain_fid,
                               &stob_id2.si_domain_fid));

        m0_fid_convert_adstob2cob(&stob_id2, &cob_fid2);
        M0_UT_ASSERT(m0_fid_eq(&cob_fid, &cob_fid2));
        M0_UT_ASSERT(m0_fid_cob_device_id(&cob_fid2) == device_id);
        m0_fid_convert_cob2gob(&cob_fid2, &gob_fid2);
        M0_UT_ASSERT(m0_fid_eq(&gob_fid, &gob_fid2));
}

static void fid_convert_ut(void)
{
        uint32_t container;
        uint64_t key;
        uint32_t device_id;
        int      i;
        int      j;
        int      k;

        key = 0;
        for (i = 0; i <= FID_UT_KEY_BITS_MAX; ++i) {
                container = 0;
                for (j = 0; j <= FID_UT_CONTAINER_BITS_MAX; ++j) {
                        device_id = 0;
                        for (k = 0; k <= FID_UT_DEVICE_ID_BITS_MAX; ++k) {
                                fid_convert_ut_check(container, key, device_id);
                                device_id = device_id * 2 + 1;
                        }
                        container = container * 2 + 1;
                }
                key = key * 2 + 1;
        }
}

struct m0_ut_suite cobfoms_ut = {
        .ts_name  = "cob-foms-ut",
        .ts_init  = NULL,
        .ts_fini  = NULL,
        .ts_tests = {
                { "cobfoms_utinit",                 cobfoms_utinit},
                { "fid_convert",                    fid_convert_ut},
                { "cobfoms_fsync_nonexistent_tx",   cobfoms_fsync_nonexist_tx},
                { "cobfoms_fsync_create_delete",
                   cobfoms_fsync_create_delete},
                { "cobfoms_single_fop",             cobfoms_single},
                { "cobfoms_multiple_fops",          cobfoms_multiple},
                { "cobfoms_preexisting_cob_create", cobfoms_preexisting_cob},
                { "cobfoms_delete_nonexistent_cob", cobfoms_del_nonexist_cob},
                { "cobfoms_md_cob_fop",             md_cob_create_delete},
                { "cobfoms_create_cob_apitest",     cob_create_api_test},
                { "cobfoms_delete_cob_apitest",     cob_delete_api_test},
                /* XXX: enable after m0_be_domain_tx_find() will be able
                 * to read back from stob finalized transactions payloads.
                 * This might be available along with recovery.
                { "cob_create_delete_fol_verify",   cobfoms_fol_verify}, */
                { "cobfoms_utfini",                 cobfoms_utfini},
                { NULL, NULL }
        }
};

#undef M0_TRACE_SUBSYSTEM