bulkio_ut.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 <sys/stat.h>
#include <sys/types.h>

#include "lib/processor.h"
#include "lib/locality.h"
#include "lib/finject.h"
#include "ut/ut.h"
#include "bulkio_common.h"
#include "net/lnet/lnet.h"
#include "rpc/rpclib.h"
#include "ioservice/io_fops.c"  /* To access static APIs. */
#include "ioservice/io_foms.c"  /* To access static APIs. */
#include "motr/setup.h"
#include "motr/setup_internal.h" /* m0_motr_conf_setup */
#include "pool/pool.h"
#include "fop/fom_generic.c"

#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_IOSERVICE

static bool bulkio_stob_created = false;

static struct bulkio_params *bp;

static struct m0_buf payload_buf = M0_BUF_INIT0;
static bool fol_check_enabled = false;
extern void bulkioapi_test(void);
static int io_fop_server_write_fom_create(struct m0_fop  *fop,
                                          struct m0_fom **m,
                                          struct m0_reqh *reqh);
static int ut_io_fom_cob_rw_create(struct m0_fop *fop, struct m0_fom **m,
                                   struct m0_reqh *reqh);
static int io_fop_server_read_fom_create(struct m0_fop *fop, struct m0_fom **m,
                                         struct m0_reqh *reqh);
static int io_fop_stob_create_fom_create(struct m0_fop *fop, struct m0_fom **m,
                                         struct m0_reqh *reqh);
static int check_write_fom_tick(struct m0_fom *fom);
static int check_read_fom_tick(struct m0_fom *fom);

static const struct m0_fop_type_ops bulkio_stob_create_ops = {
        .fto_fop_replied = io_fop_replied,
        .fto_io_coalesce = io_fop_coalesce,
        .fto_io_desc_get = io_fop_desc_get,
};

static const struct m0_fop_type_ops bulkio_server_write_fop_ut_ops = {
        .fto_fop_replied = io_fop_replied,
        .fto_io_coalesce = io_fop_coalesce,
        .fto_io_desc_get = io_fop_desc_get,
};

static const struct m0_fop_type_ops bulkio_server_read_fop_ut_ops = {
        .fto_fop_replied = io_fop_replied,
        .fto_io_coalesce = io_fop_coalesce,
        .fto_io_desc_get = io_fop_desc_get,
};

static const struct m0_fom_type_ops bulkio_server_write_fomt_ops = {
        .fto_create = io_fop_server_write_fom_create,
};

static const struct m0_fom_type_ops bulkio_server_read_fomt_ops = {
        .fto_create = io_fop_server_read_fom_create,
};

static const struct m0_fom_type_ops bulkio_stob_create_fomt_ops = {
        .fto_create = io_fop_stob_create_fom_create,
};

static const struct m0_fom_type_ops ut_io_fom_cob_rw_type_ops = {
        .fto_create = ut_io_fom_cob_rw_create,
};

static void bulkio_stob_fom_fini(struct m0_fom *fom)
{
        struct m0_io_fom_cob_rw *fom_obj;

        fom_obj = container_of(fom, struct m0_io_fom_cob_rw, fcrw_gen);
        m0_stob_put(fom_obj->fcrw_stob);
        m0_fom_fini(fom);
        m0_free(fom);
}

struct m0_net_buffer_pool * ut_get_buffer_pool(struct m0_fom *fom)
{
        struct m0_reqh_io_service  *serv_obj;
        struct m0_rios_buffer_pool *bpdesc   = NULL;
        struct m0_net_domain       *fop_ndom = NULL;
        struct m0_fop              *fop      = NULL;

        fop = fom->fo_fop;
        serv_obj = container_of(fom->fo_service,
                                struct m0_reqh_io_service, rios_gen);

        /* Get network buffer pool for network domain */
        fop_ndom = m0_fop_domain_get(fop);

        bpdesc = m0_tl_find(bufferpools, bpdesc, &serv_obj->rios_buffer_pools,
                            bpdesc->rios_ndom == fop_ndom);

        return bpdesc == NULL ? NULL :  &bpdesc->rios_bp;
}


/*
 * - This is positive test case to test m0_io_fom_cob_rw_tick(fom).
 * - This function test next phase after every defined phase for Write FOM.
 * - Validation of next phase is done as per state transition in detail design.
 *   @see DLD-bulk-server-lspec-state
 */
static int bulkio_server_write_fom_tick(struct m0_fom *fom)
{
        int rc;
        int phase0;

        phase0 = m0_fom_phase(fom);
        M0_LOG(M0_DEBUG, "phase=%d", phase0);
        rc = m0_io_fom_cob_rw_tick(fom);
        if (fol_check_enabled) {
                if (phase0 == M0_FOPH_FOL_REC_ADD) {
                        int rc;

                        if (payload_buf.b_addr != NULL)
                                m0_buf_free(&payload_buf);
                        rc = m0_buf_copy(&payload_buf,
                                         &fom->fo_tx.tx_betx.t_payload);
                        M0_UT_ASSERT(rc == 0);
                }
        }
        if (m0_fom_rc(fom) != 0) {
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_FAILURE);
                return rc;
        }
        switch (phase0) {
        case M0_FOPH_IO_FOM_BUFFER_ACQUIRE :
                M0_UT_ASSERT(M0_IN(m0_fom_phase(fom),
                                   (M0_FOPH_IO_FOM_BUFFER_WAIT,
                                    M0_FOPH_IO_ZERO_COPY_INIT)));
                break;
        case M0_FOPH_IO_ZERO_COPY_INIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_WAIT);
                break;
        case M0_FOPH_IO_ZERO_COPY_WAIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_TXN_INIT);
                break;
        case M0_FOPH_IO_STOB_INIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_STOB_WAIT);
                break;
        case M0_FOPH_IO_STOB_WAIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_BUFFER_RELEASE);
                break;
        case M0_FOPH_IO_BUFFER_RELEASE:
                M0_UT_ASSERT(M0_IN(m0_fom_phase(fom),
                                   (M0_FOPH_IO_FOM_BUFFER_ACQUIRE,
                                    M0_FOPH_SUCCESS)));
                break;
        }
        return rc;
}

/*
 * - This is positive test case to test m0_io_fom_cob_rw_tick(fom).
 * - This function test next phase after every defined phase for Read FOM.
 * - Validation of next phase is done as per state transition in detail design.
 *   @see DLD-bulk-server-lspec-state
 */
static int bulkio_server_read_fom_tick(struct m0_fom *fom)
{
        int rc;
        int phase0;

        phase0 = m0_fom_phase(fom);
        rc = m0_io_fom_cob_rw_tick(fom);
        if (m0_fom_rc(fom) != 0) {
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_FAILURE);
                return rc;
        }
        switch (phase0) {
        case M0_FOPH_IO_FOM_BUFFER_ACQUIRE :
                M0_UT_ASSERT(M0_IN(m0_fom_phase(fom),
                                   (M0_FOPH_IO_FOM_BUFFER_WAIT,
                                    M0_FOPH_IO_STOB_INIT)));
                break;
        case M0_FOPH_IO_ZERO_COPY_INIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_WAIT);
                break;
        case M0_FOPH_IO_ZERO_COPY_WAIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_BUFFER_RELEASE);
                break;
        case M0_FOPH_IO_STOB_INIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_STOB_WAIT);
                break;
        case M0_FOPH_IO_STOB_WAIT:
                M0_UT_ASSERT(m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_INIT);
                break;
        case M0_FOPH_IO_BUFFER_RELEASE:
                M0_UT_ASSERT(M0_IN(m0_fom_phase(fom),
                                   (M0_FOPH_IO_FOM_BUFFER_ACQUIRE,
                                    M0_FOPH_SUCCESS)));
                break;
        }
        return rc;
}

/*
 * This function intercepts actual I/O FOM state,
 * for state transition testing.
 *
 * This ut FOM work with real fop send by bulk client.
 * - Client first send write fop
 * - Fops at server side are intercepted by this dummy state function and
     checks all possible state transitions.
 * - It simulates failure environment for particular state and restore
 *   it again after each test.
 * - After reply fop is received by client, client sends a read fop to read
 *   data written by previous write fop.
 * - Further it will checks remaining state transitions.
 * - After reply fop is received by client, at client side received data is
 *   compared with the original data used to send it.
 */
static int ut_io_fom_cob_rw_state(struct m0_fom *fom)
{
        return m0_is_read_fop(fom->fo_fop) ?
                check_read_fom_tick(fom) : check_write_fom_tick(fom);
}

enum fom_state_transition_tests {
        TEST00 = M0_FOPH_IO_FOM_PREPARE,
        TEST01,
        TEST02,
        TEST03,
        TEST07,
        TEST10,
        TEST11,
        TEST12
};

static int                        nb_nr = 0;
static struct m0_net_buffer      *nb_list[64];
static struct m0_net_buffer_pool *buf_pool;
static int                        next_write_test = TEST00;
static int                        next_read_test  = TEST00;

static void empty_buffers_pool(uint32_t colour)
{
        nb_nr--;
        m0_net_buffer_pool_lock(buf_pool);
        do {
                nb_list[++nb_nr] = m0_net_buffer_pool_get(buf_pool, colour);
        } while (nb_list[nb_nr] != NULL);
        m0_net_buffer_pool_unlock(buf_pool);
}

static void release_one_buffer(uint32_t colour)
{
        m0_net_buffer_pool_lock(buf_pool);
        m0_net_buffer_pool_put(buf_pool, nb_list[--nb_nr], colour);
        m0_net_buffer_pool_unlock(buf_pool);
}

static void fill_buffers_pool(uint32_t colour)
{
        m0_net_buffer_pool_lock(buf_pool);
        while (nb_nr > 0)
                m0_net_buffer_pool_put(buf_pool, nb_list[--nb_nr], colour);
        m0_net_buffer_pool_unlock(buf_pool);
}

static void builkio_ut_stob_get(struct m0_io_fom_cob_rw *fom_obj)
{
        struct m0_storage_devs  *devs = m0_cs_storage_devs_get();
        struct m0_storage_dev   *dev;
        struct m0_stob_domain   *dom;

        M0_UT_ASSERT(devs != NULL);
        M0_UT_ASSERT(fom_obj->fcrw_stob != NULL);
        m0_stob_get(fom_obj->fcrw_stob);
        dom = m0_stob_dom_get(fom_obj->fcrw_stob);
        M0_UT_ASSERT(dom != NULL);
        m0_storage_devs_lock(devs);
        dev = m0_storage_devs_find_by_dom(devs, dom);
        M0_UT_ASSERT(dev != NULL);
        M0_LOG(M0_DEBUG, "get: dev=%p, ref=%" PRIi64
               "state=%d type=%d, %"PRIu64,
               dev,
               m0_ref_read(&dev->isd_ref),
               dev->isd_ha_state,
               dev->isd_srv_type,
               dev->isd_cid);
        m0_storage_dev_get(dev);
        m0_storage_devs_unlock(devs);
}

static void fom_phase_set(struct m0_fom *fom, int phase)
{
        if (m0_fom_phase(fom) == M0_FOPH_FAILURE) {
                const struct fom_phase_desc *fpd_phase;

                while (m0_fom_phase(fom) != M0_FOPH_FINISH) {
                        fpd_phase = &fpd_table[m0_fom_phase(fom)];
                        m0_fom_phase_set(fom, fpd_phase->fpd_nextphase);
                }

                m0_sm_fini(&fom->fo_sm_phase);
                M0_SET0(&fom->fo_sm_phase);
                m0_sm_init(&fom->fo_sm_phase, &fom->fo_type->ft_conf,
                           M0_FOM_PHASE_INIT, &fom->fo_loc->fl_group);

                while (m0_fom_phase(fom) != M0_FOPH_TYPE_SPECIFIC) {
                        fpd_phase = &fpd_table[m0_fom_phase(fom)];
                        m0_fom_phase_set(fom, fpd_phase->fpd_nextphase);
                }
        }

        while (phase != m0_fom_phase(fom)) {
                struct m0_io_fom_cob_rw_state_transition  st;

                st = m0_is_read_fop(fom->fo_fop) ?
                        io_fom_read_st[m0_fom_phase(fom)] :
                        io_fom_write_st[m0_fom_phase(fom)];

                if (M0_IN(phase, (st.fcrw_st_next_phase_again,
                                  st.fcrw_st_next_phase_wait))) {
                        m0_fom_phase_set(fom, phase);
                        break;
                }

                m0_fom_phase_set(fom, st.fcrw_st_next_phase_again != 0 ?
                                      st.fcrw_st_next_phase_again :
                                      st.fcrw_st_next_phase_wait);
        }
}

/*
 * - This function tests next phase after every defined phase for Write FOM.
 * - Validation of next phase is done as per state transition in detail design.
 *   @see DLD-bulk-server-lspec-state
 * - This test covers all positive as well as negative cases.
 * Note : For each test case it does following things,
 *      - simulates the environment,
 *      - run state function for respective I/O FOM,
 *      - check output state & return code,
 *      - restores the FOM to it's clean state by using the saved original data.
 */
static int check_write_fom_tick(struct m0_fom *fom)
{
        int                        rc;
        uint32_t                   colour;
        int                        acquired_net_bufs;
        int                        saved_ndesc;
        struct m0_fop_cob_rw      *rwfop;
        struct m0_fop             *fop;
        struct m0_io_fom_cob_rw   *fom_obj;
        struct m0_fid              saved_fid;
        struct m0_fid              invalid_fid;
        struct m0_stob_io_desc    *saved_stobio_desc;
        struct m0_stob            *saved_stob;
        int                        saved_count;
        struct m0_net_domain      *netdom;
        struct m0_net_transfer_mc *tm;

        fom_obj = container_of(fom, struct m0_io_fom_cob_rw, fcrw_gen);
        fop = fom->fo_fop;
        rwfop = io_rw_get(fop);

        tm = m0_fop_tm_get(fop);
        colour = m0_net_tm_colour_get(tm);

        if (m0_fom_phase(fom) < M0_FOPH_NR) {
                /*
                 * No need to test generic phases.
                 */
                rc = m0_io_fom_cob_rw_tick(fom);
                if (next_write_test <= TEST00)
                        next_write_test = m0_fom_phase(fom);
        } else if (next_write_test == TEST00) {
                rc = m0_io_fom_cob_rw_tick(fom);
                next_write_test = TEST01;
        } else if (next_write_test == TEST01) {
                /* Acquire all buffer pool buffer test some of cases. */
                if (fom_obj->fcrw_bp == NULL)
                        buf_pool = ut_get_buffer_pool(fom);
                else
                        buf_pool = fom_obj->fcrw_bp;
                M0_UT_ASSERT(buf_pool != NULL);

                empty_buffers_pool(colour);

                /*
                 * Case 01: No network buffer is available with the buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_WAIT
                 */
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT &&
                             m0_fom_phase(fom) == M0_FOPH_IO_FOM_BUFFER_WAIT);

                /* Cleanup & make clean FOM for next test. */
                rc = M0_FSO_WAIT;
                fom->fo_sm_phase.sm_rc = 0;

                release_one_buffer(colour);
                next_write_test = TEST02;
        } else if (next_write_test == TEST02) {
                /*
                 * Case 02: No network buffer is available with the buffer pool.
                 *         Even after getting buffer pool not-empty event,
                 *         buffers are not available in pool (which could be
                 *         used by other FOMs in the server).
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_WAIT
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_WAIT
                 */

                empty_buffers_pool(colour);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT  &&
                             m0_fom_phase(fom) ==  M0_FOPH_IO_FOM_BUFFER_WAIT);

                /* Cleanup & rstore FOM for next test. */
                rc = M0_FSO_WAIT;

                release_one_buffer(colour);
                next_write_test = TEST03;
        } else if (next_write_test == TEST03) {
                int cdi = fom_obj->fcrw_curr_desc_index;

                /*
                 * Case 03 : Network buffer is available with the buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 *         Expected Output phase: M0_FOPH_IO_ZERO_COPY_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_FOM_BUFFER_ACQUIRE);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_INIT);

                /*
                 * Cleanup & restore FOM for next test.
                 * Since previous case successfully acquired network buffer
                 * and now buffer pool not having any network buffer, this
                 * buffer need to return back to the buffer pool.
                 */
                acquired_net_bufs =
                        netbufs_tlist_length(&fom_obj->fcrw_netbuf_list);
                m0_net_buffer_pool_lock(fom_obj->fcrw_bp);
                while (acquired_net_bufs > 0) {
                        struct m0_net_buffer *nb;

                        nb = netbufs_tlist_tail(&fom_obj->fcrw_netbuf_list);
                        m0_net_buffer_pool_put(fom_obj->fcrw_bp, nb, colour);
                        netbufs_tlink_del_fini(nb);
                        acquired_net_bufs--;
                }
                m0_net_buffer_pool_unlock(fom_obj->fcrw_bp);
                fom_obj->fcrw_batch_size = 0;

                /*
                 * Case 04 : Network buffer is available with the buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_WAIT
                 *         Expected Output phase: M0_FOPH_IO_ZERO_COPY_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_FOM_BUFFER_WAIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_INIT);
                /*
                 * No need to cleanup here, since FOM will be  transitioned to
                 * expected phase.
                 */

                /*
                 * Case 05 : Zero-copy failure
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_INIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /*
                 * Modify net buffer used count in fop (value greater than
                 * net domain max), so that zero-copy initialisation fails.
                 */
                saved_count = rwfop->crw_desc.id_descs[cdi].bdd_used;
                netdom = m0_fop_domain_get(fop);
                rwfop->crw_desc.id_descs[cdi].bdd_used =
                        m0_net_domain_get_max_buffer_size(netdom) + 4096;

                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_INIT);
                builkio_ut_stob_get(fom_obj);

                m0_fi_enable_once("zero_copy_initiate", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & restore FOM for next test. */
                rwfop->crw_desc.id_descs[cdi].bdd_used = saved_count;

                /*
                 * Case 06 : Zero-copy success
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_INIT
                 *         Expected Output phase: M0_FOPH_IO_ZERO_COPY_WAIT
                 */
                /*
                 * To bypass request handler need to change FOM callback
                 * function which wakeup FOM from wait.
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_INIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT &&
                             m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_WAIT);
                next_write_test = TEST07;
        } else if (next_write_test == TEST07) {
                /*
                 * Case 07 : Zero-copy failure
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_WAIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_WAIT);
                fom_obj->fcrw_bulk.rb_rc  = -1;
                builkio_ut_stob_get(fom_obj);

                m0_fi_enable_once("zero_copy_finish", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & make clean FOM for next test. */
                fom_obj->fcrw_bulk.rb_rc  = 0;

                /*
                 * Case 08 : Zero-copy success from wait state.
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_WAIT
                 *         Expected Output phase: M0_FOPH_TXN_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_WAIT);
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_TXN_INIT);
                next_write_test = TEST10;
        } else if (next_write_test == TEST10) {
                /*
                 * Case 09 : STOB I/O launch failure
                 *         Input phase          : M0_FOPH_IO_STOB_INIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /* Save original fid and pass invialid fid
                 * to make I/O launch fail. */
                saved_fid = rwfop->crw_fid;
                saved_stob = fom_obj->fcrw_stob;
                m0_fid_set(&invalid_fid, 111, 222);
                m0_fid_tassume(&invalid_fid, &m0_cob_fid_type);
                fom_obj->fcrw_stob = NULL;
                rwfop->crw_fid = invalid_fid;
                fom_phase_set(fom, M0_FOPH_IO_STOB_INIT);

                m0_fi_enable_once("io_launch", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 && rc == M0_FSO_AGAIN  &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & make clean FOM for next test. */
                rwfop->crw_fid = saved_fid;
                fom_obj->fcrw_stob = saved_stob;

                /*
                 * Case 10 : STOB I/O launch success
                 *         Input phase          : M0_FOPH_IO_STOB_INIT
                 *         Expected Output phase: M0_FOPH_IO_STOB_WAIT
                 */
                /*
                 * To bypass request handler need to change FOM callback
                 * function which wakeup FOM from wait.
                 */
                fom_phase_set(fom, M0_FOPH_IO_STOB_INIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 && rc == M0_FSO_WAIT  &&
                             m0_fom_phase(fom) == M0_FOPH_IO_STOB_WAIT);

                next_write_test = TEST11;
        } else if (next_write_test == TEST11) {
                /*
                 * Case 11 : STOB I/O failure from wait state.
                 *         Input phase          : M0_FOPH_IO_STOB_WAIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /*
                 * To test this case there is a need to invalidate stobio
                 * descriptor, since io_finish() removes the stobio descriptor
                 * from list.
                 * There is only one stobio descriptor.
                 * Before returning error this phase will do following phases :
                 * - free and remove stobio descriptors in list,
                 * - put stob object
                 * - leave FOM block
                 */
                saved_stobio_desc = stobio_tlist_pop(&fom_obj->fcrw_stio_list);
                M0_UT_ASSERT(saved_stobio_desc != NULL);

                builkio_ut_stob_get(fom_obj);
                fom_phase_set(fom, M0_FOPH_IO_STOB_WAIT);
                m0_fi_enable_once("io_finish", "fake_error");
                m0_fi_enable_once("io_finish", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 && rc == M0_FSO_AGAIN  &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /*
                 * Cleanup & make clean FOM for next test.
                 * Restore original fom.
                 */
                stobio_tlist_add(&fom_obj->fcrw_stio_list, saved_stobio_desc);

                /*
                 * Case 12 : STOB I/O success
                 *         Input phase          : M0_FOPH_IO_STOB_WAIT
                 *         Expected Output phase: M0_FOPH_IO_BUFFER_RELEASE
                 */
                fom_phase_set(fom, M0_FOPH_IO_STOB_WAIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN  &&
                             m0_fom_phase(fom) == M0_FOPH_IO_BUFFER_RELEASE);

                /*
                 * Case 13 : Processing of remaining buffer descriptors.
                 *         Input phase          : M0_FOPH_IO_BUFFER_RELEASE
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 */
                fom_phase_set(fom, M0_FOPH_IO_BUFFER_RELEASE);

                saved_ndesc = fom_obj->fcrw_ndesc;
                fom_obj->fcrw_ndesc = 2;
                rwfop->crw_desc.id_nr = 2;
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN  &&
                             m0_fom_phase(fom) ==
                             M0_FOPH_IO_FOM_BUFFER_ACQUIRE);

                /* Cleanup & make clean FOM for next test. */
                fom_obj->fcrw_ndesc = saved_ndesc;
                rwfop->crw_desc.id_nr = saved_ndesc;

                /*
                 * Case 14 : All buffer descriptors are processed.
                 *         Input phase          : M0_FOPH_IO_BUFFER_RELEASE
                 *         Expected Output phase: M0_FOPH_SUCCESS
                 */
                fom_phase_set(fom, M0_FOPH_IO_BUFFER_RELEASE);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_SUCCESS);

                fill_buffers_pool(colour);
                next_write_test = TEST12;
        } else if (next_write_test == TEST12) {
                /* @todo XXX Add tests for M0_FOPH_IO_SYNC, M0_IO_FLAG_SYNC. */
                fom_phase_set(fom, M0_FOPH_IO_SYNC);
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_QUEUE_REPLY);
        } else {
                M0_UT_ASSERT(0); /* this should not happen */
                rc = M0_FSO_WAIT; /* to avoid compiler warning */
        }

        return rc;
}

/*
 * - This function test next phase after every defined phase for Read FOM.
 * - Validation of next phase is done as per state transition in detail design.
 *   @see DLD-bulk-server-lspec-state
 * - This test cover positive as well as negative cases.
 * Note : For each test case it does following things
 *      - simulate environemnt,
 *      - run state function for respective I/O FOM,
 *      - check output state & return code,
 *      - restores the FOM to it's clean state by using the saved original data.
 */
static int check_read_fom_tick(struct m0_fom *fom)
{
        int                        rc;
        uint32_t                   colour;
        int                        acquired_net_bufs;
        int                        saved_count;
        int                        saved_ndesc;
        struct m0_fop_cob_rw      *rwfop;
        struct m0_net_domain      *netdom;
        struct m0_fop             *fop;
        struct m0_io_fom_cob_rw   *fom_obj;
        struct m0_fid              saved_fid;
        struct m0_fid              invalid_fid;
        struct m0_stob_io_desc    *saved_stobio_desc;
        struct m0_stob            *saved_stob;
        struct m0_net_transfer_mc *tm;

        fom_obj = container_of(fom, struct m0_io_fom_cob_rw, fcrw_gen);
        fop = fom->fo_fop;
        rwfop = io_rw_get(fop);

        tm = m0_fop_tm_get(fop);
        colour = m0_net_tm_colour_get(tm);

        if (m0_fom_phase(fom) < M0_FOPH_NR) {
                /*
                 * No need to test generic phases.
                 */
                rc = m0_io_fom_cob_rw_tick(fom);
                next_read_test = m0_fom_phase(fom);
        } else if (next_read_test == TEST00) {
                rc = m0_io_fom_cob_rw_tick(fom);
                next_read_test = TEST01;
        } else if (next_read_test == TEST01) {
                /* Acquire all buffer pool buffer test some of cases. */
                if (fom_obj->fcrw_bp == NULL)
                        buf_pool = ut_get_buffer_pool(fom);
                else
                        buf_pool = fom_obj->fcrw_bp;
                M0_UT_ASSERT(buf_pool != NULL);

                /* Acquire all buffers from buffer pool to make it empty */
                empty_buffers_pool(colour);

                /*
                 * Case 01 : No network buffer is available with buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_WAIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_FOM_BUFFER_ACQUIRE);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 && rc == M0_FSO_WAIT &&
                             m0_fom_phase(fom) ==  M0_FOPH_IO_FOM_BUFFER_WAIT);

                /* Cleanup & make clean FOM for next test. */
                rc = M0_FSO_WAIT;

                release_one_buffer(colour);
                next_read_test = TEST02;
        } else if (next_read_test == TEST02) {
                /*
                 * Case 02 : No network buffer is available with buffer pool.
                 *         Even after getting buffer pool not-empty event,
                 *         buffers are not available in pool (which could be
                 *         used by other FOMs in the server).
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_WAIT
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_WAIT
                 */

                empty_buffers_pool(colour);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT  &&
                             m0_fom_phase(fom) ==  M0_FOPH_IO_FOM_BUFFER_WAIT);

                /* Cleanup & make clean FOM for next test. */
                rc = M0_FSO_WAIT;

                release_one_buffer(colour);
                next_read_test = TEST03;
        } else if (next_read_test == TEST03) {
                /*
                 * Case 03 : Network buffer is available with the buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 *         Expected Output phase: M0_FOPH_IO_STOB_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_FOM_BUFFER_ACQUIRE);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 && rc == M0_FSO_AGAIN  &&
                             m0_fom_phase(fom) == M0_FOPH_IO_STOB_INIT);

                /*
                 * Cleanup & make clean FOM for next test.
                 * Since previous this case successfully acquired network buffer
                 * and now buffer pool not having network buffer, this buffer
                 * need to return back to the buffer pool.
                 */
                acquired_net_bufs =
                        netbufs_tlist_length(&fom_obj->fcrw_netbuf_list);
                m0_net_buffer_pool_lock(fom_obj->fcrw_bp);
                while (acquired_net_bufs > 0) {
                        struct m0_net_buffer *nb;

                        nb = netbufs_tlist_tail(&fom_obj->fcrw_netbuf_list);
                        m0_net_buffer_pool_put(fom_obj->fcrw_bp, nb, colour);
                        netbufs_tlink_del_fini(nb);
                        acquired_net_bufs--;
                }
                m0_net_buffer_pool_unlock(fom_obj->fcrw_bp);
                fom_obj->fcrw_batch_size = 0;

                /*
                 * Case 04 : Network buffer available with buffer pool.
                 *         Input phase          : M0_FOPH_IO_FOM_BUFFER_WAIT
                 *         Expected Output phase: M0_FOPH_IO_STOB_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_FOM_BUFFER_WAIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_IO_STOB_INIT);

                /* No need to cleanup here, since FOM will transitioned
                 * to expected phase.
                 */

                /*
                 * Case 05 : STOB I/O launch failure
                 *         Input phase          : M0_FOPH_IO_STOB_INIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /* Save original fid and pass invalid fid to make I/O launch
                 * fail. */
                saved_fid = rwfop->crw_fid;
                saved_stob = fom_obj->fcrw_stob;
                m0_fid_set(&invalid_fid, 111, 222);
                m0_fid_tassume(&invalid_fid, &m0_cob_fid_type);
                fom_obj->fcrw_stob = NULL;
                rwfop->crw_fid = invalid_fid;

                fom_phase_set(fom, M0_FOPH_IO_STOB_INIT);

                m0_fi_enable_once("io_launch", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & make clean FOM for next test. */
                rwfop->crw_fid = saved_fid;
                fom_obj->fcrw_stob = saved_stob;

                /*
                 * Case 06 : STOB I/O launch success
                 *         Input phase          : M0_FOPH_IO_STOB_INIT
                 *         Expected Output phase: M0_FOPH_IO_STOB_WAIT
                 */
                /*
                 * To bypass request handler need to change FOM callback
                 * function which wakeup FOM from wait.
                 */
                fom_phase_set(fom, M0_FOPH_IO_STOB_INIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT &&
                             m0_fom_phase(fom) == M0_FOPH_IO_STOB_WAIT);
                next_read_test = TEST07;
        } else if (next_read_test == TEST07) {
                int cdi = fom_obj->fcrw_curr_desc_index;

                /*
                 * Case 07 : STOB I/O failure
                 *         Input phase          : M0_FOPH_IO_STOB_WAIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /*
                 * To test this case there is a need to invalidate stobio
                 * descriptor, since io_finish() remove stobio descriptor
                 * from list.
                 * There is only one stobio descriptor.
                 * Before returning error this phase will do following phases :
                 * - free and remove stobio descriptors in list,
                 * - put stob object
                 * - leave FOM block
                 */
                saved_stobio_desc = stobio_tlist_pop(&fom_obj->fcrw_stio_list);
                M0_UT_ASSERT(saved_stobio_desc != NULL);

                fom_phase_set(fom, M0_FOPH_IO_STOB_WAIT);
                builkio_ut_stob_get(fom_obj);

                m0_fi_enable_once("io_finish", "fake_error");
                m0_fi_enable_once("io_finish", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /*
                 * Cleanup & make clean FOM for next test.
                 * Restore original fom.
                 */
                stobio_tlist_add(&fom_obj->fcrw_stio_list, saved_stobio_desc);

                /*
                 * Case 08 : STOB I/O success
                 *         Input phase          : M0_FOPH_IO_STOB_WAIT
                 *         Expected Output phase: M0_FOPH_IO_ZERO_COPY_INIT
                 */
                fom_phase_set(fom, M0_FOPH_IO_STOB_WAIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_INIT);
                /*
                 * Case 09 : Zero-copy failure
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_INIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */

                /*
                 * Modify net buffer used count in fop (value greater than
                 * net domain max), so that zero-copy initialisation fails.
                 */
                saved_count = rwfop->crw_desc.id_descs[cdi].bdd_used;
                netdom = m0_fop_domain_get(fop);
                rwfop->crw_desc.id_descs[cdi].bdd_used =
                        m0_net_domain_get_max_buffer_size(netdom) + 4096;

                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_INIT);
                builkio_ut_stob_get(fom_obj);

                m0_fi_enable_once("zero_copy_initiate", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & make clean FOM for next test. */
                rwfop->crw_desc.id_descs[cdi].bdd_used = saved_count;
                fom_phase_set(fom, TEST10);

                /*
                 * Case 10 : Zero-copy success
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_INIT
                 *         Expected Output phase: M0_FOPH_IO_ZERO_COPY_WAIT
                 */
                /*
                 * To bypass request handler need to change FOM callback
                 * function which wakeup FOM from wait.
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_INIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_WAIT &&
                             m0_fom_phase(fom) == M0_FOPH_IO_ZERO_COPY_WAIT);
                next_read_test = TEST11;
        } else if (next_read_test == TEST11) {
                /*
                 * Case 11 : Zero-copy failure
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_WAIT
                 *         Expected Output phase: M0_FOPH_FAILURE
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_WAIT);
                fom_obj->fcrw_bulk.rb_rc  = -1;
                builkio_ut_stob_get(fom_obj);

                m0_fi_enable_once("zero_copy_finish", "keep-net-buffers");
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) != 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_FAILURE);

                /* Cleanup & make clean FOM for next test. */
                fom_obj->fcrw_bulk.rb_rc  = 0;

                /*
                 * Case 12 : Zero-copy success
                 *         Input phase          : M0_FOPH_IO_ZERO_COPY_WAIT
                 *         Expected Output phase: M0_FOPH_IO_BUFFER_RELEASE
                 */
                fom_phase_set(fom, M0_FOPH_IO_ZERO_COPY_WAIT);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_IO_BUFFER_RELEASE);


                /*
                 * Case 13 : Processing of remaining buffer descriptors.
                 *         Input phase          : M0_FOPH_IO_BUFFER_RELEASE
                 *         Expected Output phase: M0_FOPH_IO_FOM_BUFFER_ACQUIRE
                 */
                fom_phase_set(fom, M0_FOPH_IO_BUFFER_RELEASE);

                saved_ndesc = fom_obj->fcrw_ndesc;
                fom_obj->fcrw_ndesc = 2;
                rwfop->crw_desc.id_nr = 2;
                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) ==
                             M0_FOPH_IO_FOM_BUFFER_ACQUIRE);

                /* Cleanup & make clean FOM for next test. */
                fom_obj->fcrw_ndesc = saved_ndesc;
                rwfop->crw_desc.id_nr = saved_ndesc;

                /*
                 * Case 14 : All buffer descriptors are processed.
                 *         Input phase          : M0_FOPH_IO_BUFFER_RELEASE
                 *         Expected Output phase: M0_FOPH_SUCCESS
                 */
                fom_phase_set(fom, M0_FOPH_IO_BUFFER_RELEASE);

                rc = m0_io_fom_cob_rw_tick(fom);
                M0_UT_ASSERT(m0_fom_rc(fom) == 0 &&
                             rc == M0_FSO_AGAIN &&
                             m0_fom_phase(fom) == M0_FOPH_SUCCESS);

                fill_buffers_pool(colour);
        } else {
                M0_UT_ASSERT(0); /* this should not happen */
                rc = M0_FSO_WAIT; /* to avoid compiler warning */
        }

        return rc;
}

/* It is used to create the stob specified in the fid of each fop. */
static int bulkio_stob_create_fom_tick(struct m0_fom *fom)
{
        struct m0_fop_cob_rw         *rwfop;
        struct m0_stob_domain        *fom_stdom;
        struct m0_stob_id             stob_id;
        int                           rc;
        struct m0_fop_cob_writev_rep *wrep;
        struct m0_io_fom_cob_rw      *fom_obj;
        struct m0_fom_cob_op          cc;
        struct m0_reqh_io_service    *ios;
        struct m0_cob_attr            attr = { {0, } };
        struct m0_cob_oikey           oikey;
        struct m0_cob                *cob;

        cob_attr_default_fill(&attr);
        fom_obj = container_of(fom, struct m0_io_fom_cob_rw, fcrw_gen);
        ios = container_of(fom->fo_service, struct m0_reqh_io_service,
                           rios_gen);

        rwfop = io_rw_get(fom->fo_fop);
        m0_fid_convert_cob2stob(&rwfop->crw_fid, &stob_id);
        fom_stdom = m0_stob_domain_find_by_stob_id(&stob_id);
        M0_UT_ASSERT(fom_stdom != NULL);

        if (m0_fom_phase(fom) < M0_FOPH_NR) {
                if (m0_fom_phase(fom) == M0_FOPH_TXN_OPEN) {
                        m0_cob_tx_credit(ios->rios_cdom, M0_COB_OP_CREATE,
                                         m0_fom_tx_credit(fom));
                        m0_stob_create_credit(fom_stdom, m0_fom_tx_credit(fom));
                }
                return m0_fom_tick_generic(fom);
        }

        cc.fco_stob_id  = stob_id;
        cc.fco_gfid     = rwfop->crw_gfid;
        cc.fco_cfid     = rwfop->crw_fid;
        cc.fco_cob_idx  = (uint32_t) rwfop->crw_gfid.f_key;
        cc.fco_cob_type = M0_COB_IO;

        rc = m0_cc_cob_setup(&cc, ios->rios_cdom, &attr, m0_fom_tx(fom));
        if (!bulkio_stob_created)
                M0_UT_ASSERT(rc == 0);
        else {
                M0_UT_ASSERT(rc == -EEXIST);
        }

        rc = m0_stob_find(&stob_id, &fom_obj->fcrw_stob);
        M0_UT_ASSERT(rc == 0);
        rc = m0_stob_locate(fom_obj->fcrw_stob);
        M0_UT_ASSERT(rc == 0);
        rc = m0_stob_create(fom_obj->fcrw_stob, &fom->fo_tx, NULL);
        if (!bulkio_stob_created)
                M0_UT_ASSERT(rc == 0);
        else {
                M0_UT_ASSERT(rc == -EEXIST);
        }

        m0_cob_oikey_make(&oikey, &cc.fco_cfid, 0);
        rc = m0_cob_locate(ios->rios_cdom, &oikey, 0, &cob);
        M0_UT_ASSERT(rc == 0);

        wrep = m0_fop_data(fom->fo_rep_fop);
        wrep->c_rep.rwr_rc = 0;
        wrep->c_rep.rwr_count = rwfop->crw_ivec.ci_nr;
        m0_fom_phase_set(fom, M0_FOPH_SUCCESS);
        return M0_FSO_AGAIN;
}

static const struct m0_fom_ops bulkio_stob_create_fom_ops = {
        .fo_fini          = bulkio_stob_fom_fini,
        .fo_tick          = bulkio_stob_create_fom_tick,
        .fo_home_locality = m0_io_fom_cob_rw_locality_get
};

static const struct m0_fom_ops bulkio_server_write_fom_ops = {
        .fo_fini          = m0_io_fom_cob_rw_fini,
        .fo_tick          = bulkio_server_write_fom_tick,
        .fo_home_locality = m0_io_fom_cob_rw_locality_get
};

static const struct m0_fom_ops ut_io_fom_cob_rw_ops = {
        .fo_fini          = m0_io_fom_cob_rw_fini,
        .fo_tick          = ut_io_fom_cob_rw_state,
        .fo_home_locality = m0_io_fom_cob_rw_locality_get
};

static const struct m0_fom_ops bulkio_server_read_fom_ops = {
        .fo_fini          = m0_io_fom_cob_rw_fini,
        .fo_tick          = bulkio_server_read_fom_tick,
        .fo_home_locality = m0_io_fom_cob_rw_locality_get
};

static int io_fop_stob_create_fom_create(struct m0_fop  *fop,
                                         struct m0_fom **m,
                                         struct m0_reqh *reqh)
{
        int            rc;
        struct m0_fom *fom;

        rc = m0_io_fom_cob_rw_create(fop, &fom, reqh);
        M0_UT_ASSERT(rc == 0);
        fom->fo_ops = &bulkio_stob_create_fom_ops;
        *m = fom;

        return 0;
}

static int io_fop_server_write_fom_create(struct m0_fop  *fop,
                                          struct m0_fom **m,
                                          struct m0_reqh *reqh)
{
        int            rc;
        struct m0_fom *fom;

        rc = m0_io_fom_cob_rw_create(fop, &fom, reqh);
        M0_UT_ASSERT(rc == 0);
        fom->fo_ops = &bulkio_server_write_fom_ops;
        *m = fom;
        M0_UT_ASSERT(fom->fo_fop != 0);
        return rc;
}

/*
 * This creates FOM for ut.
 */
static int ut_io_fom_cob_rw_create(struct m0_fop *fop, struct m0_fom **m,
                                   struct m0_reqh *reqh)
{
        int            rc;
        struct m0_fom *fom;

        /*
         * Case : This tests the I/O FOM create api.
         *        It use real I/O FOP
         */
        rc = m0_io_fom_cob_rw_create(fop, &fom, reqh);
        M0_UT_ASSERT(rc == 0 &&
                     fom != NULL &&
                     fom->fo_rep_fop != NULL &&
                     fom->fo_fop != NULL &&
                     fom->fo_type != NULL &&
                     fom->fo_ops != NULL);

        fom->fo_ops = &ut_io_fom_cob_rw_ops;
        *m = fom;
        M0_UT_ASSERT(fom->fo_fop != 0);
        return rc;
}

static int io_fop_server_read_fom_create(struct m0_fop  *fop,
                                         struct m0_fom **m,
                                         struct m0_reqh *reqh)
{
        int            rc;
        struct m0_fom *fom;

        rc = m0_io_fom_cob_rw_create(fop, &fom, reqh);
        M0_UT_ASSERT(rc == 0);
        fom->fo_ops = &bulkio_server_read_fom_ops;
        *m = fom;
        M0_UT_ASSERT(fom->fo_fop != 0);
        return rc;
}

static void bulkio_stob_create(void)
{
        struct m0_fop_cob_rw *rw;
        enum M0_RPC_OPCODES   op;
        struct thrd_arg       targ[IO_FIDS_NR];
        int                   i;
        int                   rc;

        op = M0_IOSERVICE_WRITEV_OPCODE;
        M0_ALLOC_ARR(bp->bp_wfops, IO_FIDS_NR);
        bp->bp_wfops_nr = IO_FIDS_NR;
        for (i = 0; i < IO_FIDS_NR; ++i) {
                M0_ALLOC_PTR(bp->bp_wfops[i]);
                rc = m0_io_fop_init(bp->bp_wfops[i], &bp->bp_fids[i],
                                    &m0_fop_cob_writev_fopt,
                                    NULL);
                M0_UT_ASSERT(rc == 0);
                /*
                 * We replace the original ->ft_ops and ->ft_fom_type for
                 * regular io_fops. This is reset later.
                 */
                bp->bp_wfops[i]->if_fop.f_type->ft_fom_type.ft_ops =
                        &bulkio_stob_create_fomt_ops;
                bp->bp_wfops[i]->if_fop.f_type->ft_fom_type.ft_conf =
                        m0_generic_conf;
                rw = io_rw_get(&bp->bp_wfops[i]->if_fop);
                bp->bp_wfops[i]->if_fop.f_type->ft_ops =
                        &bulkio_stob_create_ops;
                m0_file_init(&bp->bp_file[i], &bp->bp_fids[i], &bp->bp_rdom,
                             M0_DI_CRC32_4K);
                rw->crw_fid = bp->bp_fids[i];
                targ[i].ta_index = i;
                targ[i].ta_op = op;
                targ[i].ta_bp = bp;
                io_fops_rpc_submit(&targ[i]);
        }
        io_fops_destroy(bp);
        bulkio_stob_created = true;
}

static void io_fops_submit(uint32_t index, enum M0_RPC_OPCODES op)
{
        struct thrd_arg targ = {};

        targ.ta_index = index;
        targ.ta_op = op;
        targ.ta_bp = bp;
        io_fops_rpc_submit(&targ);
}

static void io_single_fop_submit(enum M0_RPC_OPCODES op)
{
        struct m0_fop *fop;

        io_fops_create(bp, op, 1, 1, bp->bp_seg_nr);
        if (op == M0_IOSERVICE_WRITEV_OPCODE) {
                fop = &bp->bp_wfops[0]->if_fop;
                fop->f_type->ft_fom_type.ft_ops = &bulkio_server_write_fomt_ops;
        } else {
                fop = &bp->bp_rfops[0]->if_fop;
                fop->f_type->ft_fom_type.ft_ops = &io_fom_type_ops;
        }
        /*
         * Here we replace the original ->ft_ops and ->ft_fom_type as they were
         * changed during bulkio_stob_create test.
         */
        fop->f_type->ft_ops = &io_fop_rwv_ops;
        fop->f_type->ft_fom_type.ft_conf = io_conf;
        io_fops_submit(0, op);
}

static void bulkio_server_single_read_write(void)
{
        int               j;
        struct m0_bufvec *buf;

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
        }
        fol_check_enabled = true;
        io_single_fop_submit(M0_IOSERVICE_WRITEV_OPCODE);

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
        }
        io_single_fop_submit(M0_IOSERVICE_READV_OPCODE);
        fol_check_enabled = false;
}

#define WRITE_FOP_DATA(fop) M0_XCODE_OBJ(m0_fop_cob_writev_xc, fop)

static void bulkio_server_write_fol_rec_verify(void)
{
        struct m0_reqh           *reqh;
        struct m0_fol_rec         dec_rec;
        int                       result;
        struct m0_fol_frag       *dec_frag;
        struct m0_fop            *fop;
        struct m0_fop_cob_writev *wfop;

        fop = &bp->bp_wfops[0]->if_fop;
        wfop = (struct m0_fop_cob_writev *)m0_fop_data(fop);

        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        M0_LOG(M0_DEBUG, "payload_buf=" BUF_F, BUF_P(&payload_buf));
        m0_fol_rec_init(&dec_rec, &reqh->rh_fol);
        result = m0_fol_rec_decode(&dec_rec, &payload_buf);
        M0_UT_ASSERT(result == 0);

        /* FOL record frags are 2 for AD stob type and 1 for LINUX stob type. */
        M0_UT_ASSERT(dec_rec.fr_header.rh_frags_nr == 1 ||
                     dec_rec.fr_header.rh_frags_nr == 2);
        m0_tl_for(m0_rec_frag, &dec_rec.fr_frags, dec_frag) {
                struct m0_fop_fol_frag *fp_frag = dec_frag->rp_data;

                if (dec_frag->rp_ops->rpo_type->rpt_index ==
                    m0_fop_fol_frag_type.rpt_index &&
                    fp_frag->ffrp_fop_code == M0_IOSERVICE_WRITEV_OPCODE) {
                        struct m0_fop_cob_writev_rep *wfop_rep;

                        M0_UT_ASSERT(m0_xcode_cmp(
                             &WRITE_FOP_DATA(fp_frag->ffrp_fop),
                             &WRITE_FOP_DATA(wfop)) == 0);
                        wfop_rep = fp_frag->ffrp_rep;
                        M0_UT_ASSERT(wfop_rep->c_rep.rwr_rc == 0);
                        M0_UT_ASSERT(wfop_rep->c_rep.rwr_count > 0);
                }
        } m0_tl_endfor;

        m0_fol_rec_fini(&dec_rec);
        io_fops_destroy(bp);
}

static void bulkio_server_write_fol_rec_undo_verify(void)
{
        int                     j;
        struct m0_bufvec       *buf;
        struct m0_reqh         *reqh;
        struct m0_fol_rec       dec_rec;
        struct m0_dtx           dtx;
        struct m0_sm_group     *grp = m0_locality0_get()->lo_grp;
        int                     result;
        struct m0_fol_frag     *dec_frag;
        struct m0_buf           save_buf = M0_BUF_INIT0;
        /* m0_get()->i_reqh_uses_ad_stob */
        bool                    stob_ad = m0_cs_storage_devs_get() != NULL;

        if (!stob_ad)
                return;

        buf = &bp->bp_iobuf[0]->nb_buffer;
        /* Write data "b" in the file. */
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
        }
        io_single_fop_submit(M0_IOSERVICE_WRITEV_OPCODE);
        io_fops_destroy(bp);
        result = m0_buf_copy(&save_buf, &payload_buf);
        M0_UT_ASSERT(result == 0);
        io_single_fop_submit(M0_IOSERVICE_READV_OPCODE);
        io_fops_destroy(bp);

        /* Write data "a" in the file. */
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
        }
        io_single_fop_submit(M0_IOSERVICE_WRITEV_OPCODE);
        io_fops_destroy(bp);

        /* Undo the last write, so that file contains data "b". */
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        M0_UT_ASSERT(reqh != NULL);
        m0_reqh_idle_wait(reqh);

        m0_fol_rec_init(&dec_rec, &reqh->rh_fol);
        result = m0_fol_rec_decode(&dec_rec, &save_buf);
        M0_UT_ASSERT(result == 0);
        m0_buf_free(&save_buf);

        M0_UT_ASSERT(dec_rec.fr_header.rh_frags_nr == 1 ||
                     dec_rec.fr_header.rh_frags_nr == 2);
        m0_tl_for(m0_rec_frag, &dec_rec.fr_frags, dec_frag) {
                if (dec_frag->rp_ops->rpo_type->rpt_index ==
                    m0_fop_fol_frag_type.rpt_index) {
                        struct m0_fop_fol_frag *fp_frag;
                        struct m0_fop_type     *ftype;

                        fp_frag = dec_frag->rp_data;
                        M0_UT_ASSERT(fp_frag->ffrp_fop_code ==
                                     M0_IOSERVICE_WRITEV_OPCODE);

                        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 = ftype->ft_ops->fto_undo(fp_frag,
                                                         &reqh->rh_fol);
                } else {
                        m0_sm_group_lock(grp);
                        M0_SET0(&dtx);
                        m0_dtx_init(&dtx, reqh->rh_beseg->bs_domain, grp);
                        dec_frag->rp_ops->rpo_undo_credit(dec_frag,
                                &dtx.tx_betx_cred);
                        m0_dtx_open_sync(&dtx);
                        result = dec_frag->rp_ops->rpo_undo(dec_frag,
                                                            &dtx.tx_betx);
                        m0_dtx_done_sync(&dtx);
                        m0_dtx_fini(&dtx);
                        m0_sm_group_unlock(grp);
                }
                M0_UT_ASSERT(result == 0);
        } m0_tl_endfor;
        m0_fol_rec_fini(&dec_rec);

        /* Read that data from file and compare it with data "b". */
        io_single_fop_submit(M0_IOSERVICE_READV_OPCODE);
        io_fops_destroy(bp);
        m0_reqh_idle_wait(reqh);
}

/*
 * Sends regular write and read io fops, although replaces the original FOM
 * types in each io fop type with UT specific FOM types.
 */
static void bulkio_server_read_write_state_test(void)
{
        int                 j;
        enum M0_RPC_OPCODES op;
        struct m0_bufvec   *buf;
        struct m0_reqh     *reqh;

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
        }
        op = M0_IOSERVICE_WRITEV_OPCODE;
        io_fops_create(bp, op, 1, 1, bp->bp_seg_nr);
        bp->bp_wfops[0]->if_fop.f_type->ft_fom_type.ft_ops =
                &bulkio_server_write_fomt_ops;
        bp->bp_wfops[0]->if_fop.f_type->ft_ops =
                &bulkio_server_write_fop_ut_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
        }
        op = M0_IOSERVICE_READV_OPCODE;
        io_fops_create(bp, op, 1, 1, bp->bp_seg_nr);
        bp->bp_rfops[0]->if_fop.f_type->ft_fom_type.ft_ops =
                &bulkio_server_read_fomt_ops;
        bp->bp_rfops[0]->if_fop.f_type->ft_ops =
                &bulkio_server_read_fop_ut_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        m0_reqh_idle_wait(reqh);
}

/*
 * Sends regular write and read fops although replaces the original FOM types
 * in each io fop type with UT specific FOM types to check state transition for
 * I/O FOM.
 */
static void bulkio_server_rw_state_transition_test(void)
{
        int                 j;
        enum M0_RPC_OPCODES op;
        struct m0_bufvec   *buf;
        struct m0_reqh     *reqh;

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
        }
        op = M0_IOSERVICE_WRITEV_OPCODE;
        io_fops_create(bp, op, 1, 1, bp->bp_seg_nr);
        bp->bp_wfops[0]->if_fop.f_type->ft_fom_type.ft_ops =
                &ut_io_fom_cob_rw_type_ops;
        bp->bp_wfops[0]->if_fop.f_type->ft_ops =
                &bulkio_server_write_fop_ut_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);

        buf = &bp->bp_iobuf[0]->nb_buffer;
        for (j = 0; j < bp->bp_seg_nr; ++j) {
                memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
        }
        op = M0_IOSERVICE_READV_OPCODE;
        io_fops_create(bp, op, 1, 1, bp->bp_seg_nr);
        bp->bp_rfops[0]->if_fop.f_type->ft_fom_type.ft_ops =
                &ut_io_fom_cob_rw_type_ops;
        bp->bp_rfops[0]->if_fop.f_type->ft_ops =
                &bulkio_server_read_fop_ut_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        m0_reqh_idle_wait(reqh);
}

static void bulkio_server_fsync_multiple_read_write(void)
{
        int                     i;
        int                     j;
        enum M0_RPC_OPCODES     op;
        struct thrd_arg         targ[IO_FOPS_NR];
        struct m0_bufvec       *buf;
        struct m0_be_tx_remid   remid;
        int                     rc;
        struct m0_reqh         *reqh;

        for (i = 0; i < IO_FOPS_NR; ++i) {
                buf = &bp->bp_iobuf[i]->nb_buffer;
                for (j = 0; j < bp->bp_seg_nr; ++j) {
                        memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
                }
        }
        for (op = M0_IOSERVICE_WRITEV_OPCODE; op >= M0_IOSERVICE_READV_OPCODE;
             --op) {
                /*
                 * IO fops are deallocated by an rpc item type op on receiving
                 * the reply fop. See io_item_free().
                 */
                io_fops_create(bp, op, IO_FIDS_NR, IO_FOPS_NR, bp->bp_seg_nr);
                for (i = 0; i < IO_FOPS_NR; ++i) {
                        targ[i].ta_index = i;
                        targ[i].ta_op = op;
                        targ[i].ta_bp = bp;
                        io_fops_rpc_submit(&targ[i]);
                        remid = targ[i].ta_bp->bp_remid;
                }
                io_fops_destroy(bp);
        }

        /*
         * Send the fsync fop request.
         * targ[0] suffices, since it contains the session.
         */
        rc = io_fsync_send_fop(&remid, &targ[0]);
        M0_UT_ASSERT(rc == 0);
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);
}

static void bulkio_server_multiple_read_write(void)
{
        int                  rc;
        int                  i;
        int                  j;
        enum M0_RPC_OPCODES  op;
        struct thrd_arg      targ[IO_FOPS_NR];
        struct m0_bufvec    *buf;
        struct m0_reqh      *reqh;

        for (i = 0; i < IO_FOPS_NR; ++i) {
                buf = &bp->bp_iobuf[i]->nb_buffer;
                for (j = 0; j < bp->bp_seg_nr; ++j) {
                        memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
                }
        }
        for (op = M0_IOSERVICE_WRITEV_OPCODE; op >= M0_IOSERVICE_READV_OPCODE;
          --op) {
                /*
                 * IO fops are deallocated by an rpc item type op on receiving
                 * the reply fop. See io_item_free().
                 */
                io_fops_create(bp, op, IO_FIDS_NR, IO_FOPS_NR, bp->bp_seg_nr);
                for (i = 0; i < IO_FOPS_NR; ++i) {
                        targ[i].ta_index = i;
                        targ[i].ta_op = op;
                        targ[i].ta_bp = bp;
                        M0_SET0(bp->bp_threads[i]);
                        rc = M0_THREAD_INIT(bp->bp_threads[i],
                                            struct thrd_arg *,
                                            NULL, &io_fops_rpc_submit,
                                            &targ[i], "io_thrd");
                        M0_UT_ASSERT(rc == 0);
                }
                /* Waits till all threads finish their job. */
                for (i = 0; i < IO_FOPS_NR; ++i) {
                        m0_thread_join(bp->bp_threads[i]);
                        buf = &bp->bp_iobuf[i]->nb_buffer;
                        for (j = 0; j < bp->bp_seg_nr; ++j) {
                                memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
                        }
                }
                io_fops_destroy(bp);
        }
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);
}

static void add_buffer_bulk(struct m0_rpc_bulk *rbulk,
                            enum M0_RPC_OPCODES op,
                            int                 index)
{
        struct m0_rpc_bulk_buf *rbuf;
        int                     rc;
        int                     i;

        /*
         * Adds a m0_rpc_bulk_buf structure to list of such structures
         * in m0_rpc_bulk.
         */
        rc = m0_rpc_bulk_buf_add(rbulk, bp->bp_seg_nr, 0, &bp->bp_cnetdom,
                                 NULL, &rbuf);
        M0_UT_ASSERT(rc == 0);
        M0_UT_ASSERT(rbuf != NULL);

        /* Adds io buffers to m0_rpc_bulk_buf structure. */
        for (i = 0; i < bp->bp_seg_nr; ++i) {
                rc = m0_rpc_bulk_buf_databuf_add(rbuf,
                         bp->bp_iobuf[index]->nb_buffer.ov_buf[i],
                         bp->bp_iobuf[index]->nb_buffer.ov_vec.v_count[i],
                         bp->bp_offsets[0], &bp->bp_cnetdom);
                M0_UT_ASSERT(rc == 0);
                bp->bp_offsets[0] +=
                        bp->bp_iobuf[index]->nb_buffer.ov_vec.v_count[i];
        }
        bp->bp_offsets[0] += IO_SEG_SIZE;

        rbuf->bb_nbuf->nb_qtype = (op == M0_IOSERVICE_WRITEV_OPCODE) ?
                M0_NET_QT_PASSIVE_BULK_SEND :
                M0_NET_QT_PASSIVE_BULK_RECV;
}

static void fop_create_populate(int index, enum M0_RPC_OPCODES op, int buf_nr)
{
        struct m0_io_fop       **io_fops;
        struct m0_rpc_bulk      *rbulk;
        struct m0_io_fop        *iofop;
        struct m0_fop_cob_rw    *rw;
        int                      i;
        int                      rc;

        if (op == M0_IOSERVICE_WRITEV_OPCODE) {
                M0_ALLOC_ARR(bp->bp_wfops, IO_FOPS_NR);
                bp->bp_wfops_nr = IO_FOPS_NR;
        } else {
                M0_ALLOC_ARR(bp->bp_rfops, IO_FOPS_NR);
                bp->bp_rfops_nr = IO_FOPS_NR;
        }

        io_fops = (op == M0_IOSERVICE_WRITEV_OPCODE) ? bp->bp_wfops :
                                                       bp->bp_rfops;
        M0_ALLOC_PTR(io_fops[index]);

        if (op == M0_IOSERVICE_WRITEV_OPCODE)
                rc = m0_io_fop_init(io_fops[index], &bp->bp_fids[0],
                                    &m0_fop_cob_writev_fopt, NULL);
        else
                rc = m0_io_fop_init(io_fops[index], &bp->bp_fids[0],
                                    &m0_fop_cob_readv_fopt, NULL);
        M0_UT_ASSERT(rc == 0);
        iofop = io_fops[index];
        rbulk = &iofop->if_rbulk;
        rw = io_rw_get(&io_fops[index]->if_fop);

        rw->crw_fid = bp->bp_fids[0];
        rw->crw_index = m0_fid_cob_device_id(&bp->bp_fids[0]);
        rw->crw_pver = CONF_PVER_FID;
        bp->bp_offsets[0] = IO_SEG_START_OFFSET;


        for (i = 0; i < buf_nr; ++i)
                add_buffer_bulk(rbulk, op, i);

        /*
         * Allocates memory for array of net buf descriptors and array of
         * index vectors from io fop.
         */
        rc = m0_io_fop_prepare(&iofop->if_fop);
        M0_UT_ASSERT(rc == 0);

        /*
         * Stores the net buf desc/s after adding the corresponding
         * net buffers to transfer machine to io fop wire format.
         */
        rc = m0_rpc_bulk_store(rbulk, &bp->bp_cctx->rcx_connection,
                               rw->crw_desc.id_descs, &m0_rpc__buf_bulk_cb);
        M0_UT_ASSERT(rc == 0);

        for (i = 0; i < IO_FIDS_NR; ++i)
                bp->bp_offsets[i] = IO_SEG_START_OFFSET;
}

static void bulkio_server_read_write_multiple_nb(void)
{
        int                 i;
        int                 j;
        int                 buf_nr;
        enum M0_RPC_OPCODES op;
        struct m0_bufvec   *buf;
        struct m0_reqh     *reqh;

        buf_nr = IO_FOPS_NR / 4;
        for (i = 0; i < buf_nr; ++i) {
                buf = &bp->bp_iobuf[i]->nb_buffer;
                for (j = 0; j < bp->bp_seg_nr; ++j) {
                        memset(buf->ov_buf[j], 'b', IO_SEG_SIZE);
                }
        }
        op = M0_IOSERVICE_WRITEV_OPCODE;
        fop_create_populate(0, op, buf_nr);
        bp->bp_wfops[0]->if_fop.f_type->ft_ops = &io_fop_rwv_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);

        for (i = 0; i < buf_nr; ++i) {
                buf = &bp->bp_iobuf[i]->nb_buffer;
                for (j = 0; j < bp->bp_seg_nr; ++j) {
                        memset(buf->ov_buf[j], 'a', IO_SEG_SIZE);
                }
        }
        op = M0_IOSERVICE_READV_OPCODE;
        fop_create_populate(0, op, buf_nr);
        bp->bp_rfops[0]->if_fop.f_type->ft_ops = &io_fop_rwv_ops;
        io_fops_submit(0, op);
        io_fops_destroy(bp);
        m0_reqh_idle_wait(reqh);
}

static void bulkio_init(void)
{
        int         rc;
        const char *caddr = "0@lo:12345:34:*";
        const char *saddr = "0@lo:12345:34:1";

        nb_nr = 0;
        M0_SET0(&nb_list);
        buf_pool = NULL;
        next_write_test = TEST00;
        next_read_test  = TEST00;
        /*
         * Current set of tests work with standalone io_fops, but
         * io_fop_di_prepare() relies on the fact that an io_fop is embedded
         * into the io_req_fop structure. Therefore, we have to skip DI prepare
         * for the tests in order to avoid crashes.
         */
        m0_fi_enable("io_fop_di_prepare", "skip_di_for_ut");
        m0_fi_enable("m0_file_init", "skip_di_for_ut");
        m0_fi_enable("stob_ad_domain_create", "write_undo");

        M0_ALLOC_PTR(bp);
        M0_ASSERT(bp != NULL);
        bulkio_params_init(bp);

        rc = bulkio_server_start(bp, saddr);
        M0_UT_ASSERT(rc == 0);
        M0_UT_ASSERT(bp->bp_sctx != NULL);
        rc = bulkio_client_start(bp, caddr, saddr);
        M0_UT_ASSERT(rc == 0);
        M0_UT_ASSERT(bp->bp_cctx != NULL);

        bulkio_stob_create();
}

static void bulkio_fini(void)
{
        struct m0_reqh *reqh;
        int             i;

        for (i = 0; i < IO_FIDS_NR; ++i)
                m0_file_fini(&bp->bp_file[i]);
        reqh = m0_cs_reqh_get(&bp->bp_sctx->rsx_motr_ctx);
        m0_reqh_idle_wait(reqh);
        bulkio_client_stop(bp->bp_cctx);
        bulkio_server_stop(bp->bp_sctx);
        bulkio_params_fini(bp);
        m0_free(bp);

        m0_fi_disable("io_fop_di_prepare", "skip_di_for_ut");
        m0_fi_disable("m0_file_init", "skip_di_for_ut");
        m0_fi_disable("stob_ad_domain_create", "write_undo");
}

/*
 * Only used for user-space UT.
 */
struct m0_ut_suite bulkio_server_ut = {
        .ts_name = "bulk-server-ut",
        .ts_tests = {
                /*
                 * Intentionally kept as first test case. It initializes
                 * all necessary data for sending IO fops. Keeping
                 * bulkio_init() as .ts_init requires changing all
                 * M0_UT_ASSERTS to M0_ASSERTS.
                 */
                { "bulkio_init", bulkio_init},
                { "bulkio_server_single_read_write",
                   bulkio_server_single_read_write},
                { "bulkio_server_write_fol_rec_verify",
                   bulkio_server_write_fol_rec_verify},
                { "bulkio_server_write_fol_rec_undo_verify",
                   bulkio_server_write_fol_rec_undo_verify},
                { "bulkio_server_read_write_state_test",
                   bulkio_server_read_write_state_test},
                { "bulkio_server_vectored_read_write",
                   bulkio_server_multiple_read_write},
                /*
                 * Keep this test close to bulkio_server_multiple_read_write.
                 * Otherwise something breaks :)
                 */
                { "bulkio_server_fsync_multiple_read_write",
                   bulkio_server_fsync_multiple_read_write},
                { "bulkio_server_rw_multiple_nb_server",
                   bulkio_server_read_write_multiple_nb},
                { "bulkio_server_rw_state_transition_test",
                   bulkio_server_rw_state_transition_test},
#if 0

                { "bulkio_server_read_write_fv_mismatch",
                   bulkio_server_read_write_fv_mismatch},
#endif
                { "bulkio_fini", bulkio_fini},
                { NULL, NULL }
        }
};
M0_EXPORTED(bulkio_server_ut);
#undef M0_TRACE_SUBSYSTEM