io_nw_xfer.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 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 "motr/client.h"
#include "motr/client_internal.h"
#include "motr/addb.h"
#include "motr/pg.h"
#include "motr/io.h"

#include "lib/memory.h"          /* m0_alloc, m0_free */
#include "lib/errno.h"           /* ENOMEM */
#include "lib/finject.h"         /* M0_FI_ */
#include "lib/cksum_utils.h"
#include "fid/fid.h"             /* m0_fid */
#include "rpc/rpclib.h"          /* m0_rpc_ */
#include "lib/ext.h"             /* struct m0_ext */
#include "lib/misc.h"            /* m0_extent_vec_get_checksum_addr */
#include "fop/fom_generic.h"     /* m0_rpc_item_generic_reply_rc */
#include "sns/parity_repair.h"   /* m0_sns_repair_spare_map*/
#include "fd/fd.h"               /* m0_fd_fwd_map m0_fd_bwd_map */
#include "motr/addb.h"
#include "rpc/item.h"
#include "rpc/rpc_internal.h"

#define M0_TRACE_SUBSYSTEM M0_TRACE_SUBSYS_CLIENT
#include "lib/trace.h"           /* M0_LOG */

const struct m0_bob_type nwxfer_bobtype;
const struct m0_bob_type tioreq_bobtype;

M0_BOB_DEFINE(M0_INTERNAL, &nwxfer_bobtype,  nw_xfer_request);
M0_BOB_DEFINE(M0_INTERNAL, &tioreq_bobtype,  target_ioreq);

const struct m0_bob_type nwxfer_bobtype = {
        .bt_name         = "nw_xfer_request_bobtype",
        .bt_magix_offset = offsetof(struct nw_xfer_request, nxr_magic),
        .bt_magix        = M0_NWREQ_MAGIC,
        .bt_check        = NULL,
};

const struct m0_bob_type tioreq_bobtype = {
        .bt_name         = "target_ioreq",
        .bt_magix_offset = offsetof(struct target_ioreq, ti_magic),
        .bt_magix        = M0_TIOREQ_MAGIC,
        .bt_check        = NULL,
};

static void to_op_io_map(const struct m0_op *op,
                         struct m0_op_io *ioo)
{
        uint64_t oid  = m0_sm_id_get(&op->op_sm);
        uint64_t ioid = m0_sm_id_get(&ioo->ioo_sm);

        if (ioo->ioo_addb2_mapped++ == 0)
                M0_ADDB2_ADD(M0_AVI_CLIENT_TO_IOO, oid, ioid);
}

static void m0_op_io_to_rpc_map(const struct m0_op_io    *ioo,
                                const struct m0_rpc_item *item)
{
        uint64_t rid  = m0_sm_id_get(&item->ri_sm);
        uint64_t ioid = m0_sm_id_get(&ioo->ioo_sm);
        M0_ADDB2_ADD(M0_AVI_IOO_TO_RPC, ioid, rid);
}

static uint32_t io_di_size(struct m0_op_io *ioo)
{
        uint32_t                rc = 0;
        const struct m0_fid    *fid;
        const struct m0_di_ops *di_ops;
        struct m0_file         *file;

        M0_PRE(ioo != NULL);

        #ifndef ENABLE_DATA_INTEGRITY
                return M0_RC(rc);
        #endif
        /* Get di details (workaround!) by setting the dom be NULL*/
        file = &ioo->ioo_flock;
        fid = &ioo->ioo_oo.oo_fid;
        m0_file_init(file, fid, NULL, M0_DI_DEFAULT_TYPE);
        di_ops = file->fi_di_ops;

        if (di_ops->do_out_shift(file) == 0)
                return M0_RC(0);

        rc = di_ops->do_out_shift(file) * M0_DI_ELEMENT_SIZE;

        return M0_RC(rc);
}

static void parity_page_pos_get(struct pargrp_iomap *map,
                                m0_bindex_t          index,
                                uint32_t            *row,
                                uint32_t            *col)
{
        uint64_t                  pg_id;
        struct m0_pdclust_layout *play;

        M0_PRE(map != NULL);
        M0_PRE(row != NULL);
        M0_PRE(col != NULL);

        play = pdlayout_get(map->pi_ioo);

        pg_id = page_id(index, map->pi_ioo->ioo_obj);
        *row  = pg_id % rows_nr(play, map->pi_ioo->ioo_obj);
        *col  = pg_id / rows_nr(play, map->pi_ioo->ioo_obj);
}

static int dgmode_rwvec_alloc_init(struct target_ioreq *ti)
{
        int                       rc;
        uint64_t                  cnt;
        struct dgmode_rwvec      *dg;
        struct m0_pdclust_layout *play;
        struct m0_op_io   *ioo;

        M0_ENTRY();
        M0_PRE(ti != NULL);
        M0_PRE(ti->ti_dgvec == NULL);

        ioo = bob_of(ti->ti_nwxfer, struct m0_op_io, ioo_nwxfer,
                     &ioo_bobtype);

        M0_ALLOC_PTR(dg);
        if (dg == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        play = pdlayout_get(ioo);
        dg->dr_tioreq = ti;

        cnt = page_nr(ioo->ioo_iomap_nr
                      * layout_unit_size(play)
                      * (layout_n(play) + layout_k(play)),
                      ioo->ioo_obj);
        rc  = m0_indexvec_alloc(&dg->dr_ivec, cnt);
        if (rc != 0)
                goto failed;

        M0_ALLOC_ARR(dg->dr_bufvec.ov_buf, cnt);
        if (dg->dr_bufvec.ov_buf == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        M0_ALLOC_ARR(dg->dr_bufvec.ov_vec.v_count, cnt);
        if (dg->dr_bufvec.ov_vec.v_count == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        M0_ALLOC_ARR(dg->dr_auxbufvec.ov_buf, cnt);
        if (dg->dr_auxbufvec.ov_buf == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        M0_ALLOC_ARR(dg->dr_auxbufvec.ov_vec.v_count, cnt);
        if (dg->dr_auxbufvec.ov_vec.v_count == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        M0_ALLOC_ARR(dg->dr_pageattrs, cnt);
        if (dg->dr_pageattrs == NULL) {
                rc = -ENOMEM;
                goto failed;
        }

        /*
         * This value is incremented every time a new segment is added
         * to this index vector.
         */
        dg->dr_ivec.iv_vec.v_nr = 0;

        ti->ti_dgvec = dg;
        return M0_RC(0);
failed:
        ti->ti_dgvec = NULL;
        if (dg->dr_bufvec.ov_buf != NULL)
                m0_free(dg->dr_bufvec.ov_buf);
        if (dg->dr_bufvec.ov_vec.v_count != NULL)
                m0_free(dg->dr_bufvec.ov_vec.v_count);
        if (dg->dr_auxbufvec.ov_buf != NULL)
                m0_free(dg->dr_auxbufvec.ov_buf);
        if (dg->dr_auxbufvec.ov_vec.v_count != NULL)
                m0_free(dg->dr_auxbufvec.ov_vec.v_count);
        m0_free(dg);
        return M0_ERR(rc);
}

static void dgmode_rwvec_dealloc_fini(struct dgmode_rwvec *dg)
{
        M0_ENTRY();

        M0_PRE(dg != NULL);

        dg->dr_tioreq = NULL;
        /*
         * Will need to go through array of parity groups to find out
         * exact number of segments allocated for the index vector.
         * Instead, a fixed number of segments is enough to avoid
         * triggering the assert from m0_indexvec_free().
         * The memory allocator knows the size of memory area held by
         * dg->dr_ivec.iv_index and dg->dr_ivec.iv_vec.v_count.
         */
        if (dg->dr_ivec.iv_vec.v_nr == 0)
                ++dg->dr_ivec.iv_vec.v_nr;

        m0_indexvec_free(&dg->dr_ivec);
        m0_free(dg->dr_bufvec.ov_buf);
        m0_free(dg->dr_bufvec.ov_vec.v_count);
        m0_free(dg->dr_auxbufvec.ov_buf);
        m0_free(dg->dr_auxbufvec.ov_vec.v_count);
        m0_free(dg->dr_pageattrs);
        m0_free(dg);
}

static uint64_t tioreqs_hash_func(const struct m0_htable *htable, const void *k)
{
        const uint64_t *key;
        M0_PRE(htable != NULL);
        M0_PRE(htable->h_bucket_nr > 0);
        M0_PRE(k != NULL);

        key = (uint64_t *)k;

        return *key % htable->h_bucket_nr;
}

static bool tioreq_key_eq(const void *key1, const void *key2)
{
        const uint64_t *k1 = (uint64_t *)key1;
        const uint64_t *k2 = (uint64_t *)key2;

        M0_PRE(k1 != NULL);
        M0_PRE(k2 != NULL);

        return *k1 == *k2;
}

M0_HT_DESCR_DEFINE(tioreqht, "Hash of target_ioreq objects", M0_INTERNAL,
                   struct target_ioreq, ti_link, ti_magic,
                   M0_TIOREQ_MAGIC, M0_TLIST_HEAD_MAGIC,
                   ti_fid.f_container, tioreqs_hash_func, tioreq_key_eq);

M0_HT_DEFINE(tioreqht, M0_INTERNAL, struct target_ioreq, uint64_t);

static bool target_ioreq_invariant(const struct target_ioreq *ti)
{
        return M0_RC(ti != NULL &&
                     _0C(target_ioreq_bob_check(ti)) &&
                     _0C(ti->ti_session       != NULL) &&
                     _0C(ti->ti_nwxfer        != NULL) &&
                     _0C(ti->ti_bufvec.ov_buf != NULL) &&
                     _0C(ti->ti_auxbufvec.ov_buf != NULL) &&
                     _0C(m0_fid_is_valid(&ti->ti_fid)) &&
                     m0_tl_forall(iofops, iofop, &ti->ti_iofops,
                                  ioreq_fop_invariant(iofop)));
}

M0_INTERNAL bool nw_xfer_request_invariant(const struct nw_xfer_request *xfer)
{
        return xfer != NULL &&
               _0C(nw_xfer_request_bob_check(xfer)) &&
               _0C(xfer->nxr_state < NXS_STATE_NR) &&

               _0C(ergo(xfer->nxr_state == NXS_INITIALIZED,
                        xfer->nxr_rc == 0 && xfer->nxr_bytes == 0 &&
                        m0_atomic64_get(&xfer->nxr_iofop_nr) == 0)) &&

               _0C(ergo(xfer->nxr_state == NXS_INFLIGHT,
                        !tioreqht_htable_is_empty(&xfer->nxr_tioreqs_hash))) &&

               _0C(ergo(xfer->nxr_state == NXS_COMPLETE,
                        m0_atomic64_get(&xfer->nxr_iofop_nr) == 0 &&
                        m0_atomic64_get(&xfer->nxr_rdbulk_nr) == 0)) &&

               m0_htable_forall(tioreqht, tioreq, &xfer->nxr_tioreqs_hash,
                                target_ioreq_invariant(tioreq));
}

void target_ioreq_fini(struct target_ioreq *ti)
{
        struct m0_op_io *ioo;
        unsigned int     opcode;

        M0_ENTRY("target_ioreq %p", ti);

        M0_PRE(target_ioreq_invariant(ti));
        M0_PRE(iofops_tlist_is_empty(&ti->ti_iofops));

        ioo = bob_of(ti->ti_nwxfer, struct m0_op_io,
                     ioo_nwxfer, &ioo_bobtype);
        opcode = ioo->ioo_oo.oo_oc.oc_op.op_code;
        target_ioreq_bob_fini(ti);
        tioreqht_tlink_fini(ti);
        iofops_tlist_fini(&ti->ti_iofops);
        ti->ti_ops     = NULL;
        ti->ti_session = NULL;
        ti->ti_nwxfer  = NULL;

        /* Resets the number of segments in vector. */
        if (ti->ti_ivec.iv_vec.v_nr == 0)
                ti->ti_ivec.iv_vec.v_nr = ti->ti_bufvec.ov_vec.v_nr;

        m0_indexvec_free(&ti->ti_ivec);
        if (opcode == M0_OC_FREE)
                m0_indexvec_free(&ti->ti_trunc_ivec);
        m0_free0(&ti->ti_bufvec.ov_buf);
        m0_free0(&ti->ti_bufvec.ov_vec.v_count);
        m0_free0(&ti->ti_auxbufvec.ov_buf);
        m0_free0(&ti->ti_auxbufvec.ov_vec.v_count);

        /* For the write path the ti_attrbuf which is m0_buf will be freed by
         * RPC layer, so no need to explicitly free it m0_buf_free(&ti->ti_attrbuf);
         * TODO: Further validate this by checking if memory is actually freed.
         */

        m0_free0(&ti->ti_pageattrs);

        if (ti->ti_dgvec != NULL)
                dgmode_rwvec_dealloc_fini(ti->ti_dgvec);
        if (ti->ti_cc_fop_inited) {
                struct m0_rpc_item *item = &ti->ti_cc_fop.crf_fop.f_item;
                M0_LOG(M0_DEBUG, "item="ITEM_FMT" osr_xid=%"PRIu64,
                                  ITEM_ARG(item), item->ri_header.osr_xid);
                ti->ti_cc_fop_inited = false;
                m0_fop_put_lock(&ti->ti_cc_fop.crf_fop);
        }

        if ( opcode == M0_OC_WRITE ) {
                m0_buf_free( &ti->ti_attrbuf );
                m0_free( (void *)ti->ti_cksum_seg_b_nob );
        } else if ( opcode == M0_OC_READ )
                m0_indexvec_free(&ti->ti_goff_ivec);

        m0_free(ti);
        M0_LEAVE();
}

void target_ioreq_cancel(struct target_ioreq *ti)
{
        struct ioreq_fop *irfop;

        m0_tl_for (iofops, &ti->ti_iofops, irfop) {
                m0_rpc_item_cancel(&irfop->irf_iofop.if_fop.f_item);
        } m0_tl_endfor;
}

static struct target_ioreq *target_ioreq_locate(struct nw_xfer_request *xfer,
                                                struct m0_fid          *fid)
{
        struct target_ioreq *ti;

        M0_ENTRY("nw_xfer_request %p, fid %p", xfer, fid);

        M0_PRE(nw_xfer_request_invariant(xfer));
        M0_PRE(fid != NULL);

        ti = tioreqht_htable_lookup(&xfer->nxr_tioreqs_hash, &fid->f_container);
        /* WARN: Searches only with the container but compares the whole fid. */
        M0_ASSERT(ergo(ti != NULL, m0_fid_cmp(fid, &ti->ti_fid) == 0));

        M0_LEAVE();
        return ti;
}

/*
 * For partially parity groups only data units present in the truncate range
 * will be truncated. For fully spanned parity group both data and parity
 * units will be truncated.
 */
static bool should_unit_be_truncated(bool                      partial,
                                     enum m0_pdclust_unit_type unit_type,
                                     enum page_attr            flags)
{
        return (!partial || unit_type == M0_PUT_DATA) &&
               (flags & PA_WRITE);
}

static void target_ioreq_seg_add(struct target_ioreq              *ti,
                                 const struct m0_pdclust_src_addr *src,
                                 const struct m0_pdclust_tgt_addr *tgt,
                                 m0_bindex_t                       gob_offset,
                                 m0_bcount_t                       count,
                                 struct pargrp_iomap              *map)
{
        uint32_t                   seg;
        uint32_t                   tseg;
        m0_bindex_t                toff;
        m0_bindex_t                goff;
        m0_bindex_t                pgstart;
        m0_bindex_t                pgend;
        m0_bindex_t                unit_sz;
        struct data_buf           *buf;
        struct m0_op_io           *ioo;
        struct m0_pdclust_layout  *play;
        uint64_t                   frame;
        uint64_t                   unit;
        struct m0_indexvec        *ivec;
        struct m0_indexvec        *trunc_ivec = NULL;
        struct m0_indexvec        *goff_ivec = NULL;
        struct m0_bufvec          *bvec;
        struct m0_bufvec          *auxbvec;
        enum m0_pdclust_unit_type  unit_type;
        enum page_attr            *pattr;
        uint64_t                   cnt;
        unsigned int               opcode;
        m0_bcount_t                grp_size;
        uint64_t                   page_size;
        struct m0_ext              goff_span_ext;
        bool                       is_goff_in_range;
        void                      *dst_attr = NULL;
        uint32_t                   b_nob;

        M0_PRE(tgt != NULL);
        frame = tgt->ta_frame;
        M0_PRE(src != NULL);
        unit  = src->sa_unit;
        M0_ENTRY("tio req %p, gob_offset %" PRIu64 ", count %"PRIu64
                 " frame %" PRIu64 " unit %"PRIu64,
                 ti, gob_offset, count, frame, unit);

        M0_PRE(ti != NULL);
        M0_PRE(map != NULL);
        M0_PRE(target_ioreq_invariant(ti));

        ti->ti_goff = gob_offset;

        ioo = bob_of(ti->ti_nwxfer, struct m0_op_io,
                     ioo_nwxfer, &ioo_bobtype);
        opcode = ioo->ioo_oo.oo_oc.oc_op.op_code;
        play = pdlayout_get(ioo);

        page_size = m0__page_size(ioo);
        grp_size = data_size(play) * map->pi_grpid;
        unit_type = m0_pdclust_unit_classify(play, unit);
        M0_ASSERT(M0_IN(unit_type, (M0_PUT_DATA, M0_PUT_PARITY)));

        unit_sz = layout_unit_size(play);
        toff    = target_offset(frame, play, gob_offset);
        pgstart = toff;
        goff    = unit_type == M0_PUT_DATA ? gob_offset : 0;

        M0_LOG(M0_DEBUG,
               "[gpos %" PRIu64 ", count %" PRIu64 "] [%" PRIu64 ", %" PRIu64 "]"
               "->[%" PRIu64 ",%" PRIu64 "] %c", gob_offset, count, src->sa_group,
               src->sa_unit, tgt->ta_frame, tgt->ta_obj,
               unit_type == M0_PUT_DATA ? 'D' : 'P');

        /* Use ti_dgvec as long as it is dgmode-read/write. */
        if (ioreq_sm_state(ioo) == IRS_DEGRADED_READING ||
            ioreq_sm_state(ioo) == IRS_DEGRADED_WRITING)  {
                M0_ASSERT(ti->ti_dgvec != NULL);
                ivec  = &ti->ti_dgvec->dr_ivec;
                bvec  = &ti->ti_dgvec->dr_bufvec;
                auxbvec  = &ti->ti_dgvec->dr_auxbufvec;
                pattr = ti->ti_dgvec->dr_pageattrs;
                cnt = page_nr(ioo->ioo_iomap_nr * layout_unit_size(play) *
                      (layout_n(play) + layout_k(play)), ioo->ioo_obj);
                M0_LOG(M0_DEBUG, "map_nr=%" PRIu64 " req state=%u cnt=%"PRIu64,
                                 ioo->ioo_iomap_nr, ioreq_sm_state(ioo), cnt);
        } else {
                ivec  = &ti->ti_ivec;
                trunc_ivec  = &ti->ti_trunc_ivec;
                bvec  = &ti->ti_bufvec;
                auxbvec = &ti->ti_auxbufvec;
                dst_attr = ti->ti_attrbuf.b_addr;
                goff_ivec = &ti->ti_goff_ivec;
                pattr = ti->ti_pageattrs;
                cnt = page_nr(ioo->ioo_iomap_nr * layout_unit_size(play) *
                              layout_n(play), ioo->ioo_obj);
                M0_LOG(M0_DEBUG, "map_nr=%" PRIu64 " req state=%u cnt=%"PRIu64,
                                 ioo->ioo_iomap_nr, ioreq_sm_state(ioo), cnt);
        }

        while (pgstart < toff + count) {
                pgend = min64u(pgstart + page_size,
                               toff + count);
                seg   = SEG_NR(ivec);

                /* Save COB offsets in ti_ivec */
                INDEX(ivec, seg) = pgstart;
                COUNT(ivec, seg) = pgend - pgstart;

                if (unit_type == M0_PUT_DATA) {
                        uint32_t row = map->pi_max_row;
                        uint32_t col = map->pi_max_col;

                        page_pos_get(map, goff, grp_size, &row, &col);
                        M0_ASSERT(row <= map->pi_max_row);
                        M0_ASSERT(col <= map->pi_max_col);
                        buf = map->pi_databufs[row][col];

                        pattr[seg] |= PA_DATA;
                        M0_LOG(M0_DEBUG, "Data seg %u added", seg);
                } else {
                        buf = map->pi_paritybufs[page_id(goff, ioo->ioo_obj)]
                        [unit - layout_n(play)];
                        pattr[seg] |= PA_PARITY;
                        M0_LOG(M0_DEBUG, "Parity seg %u added", seg);
                }
                buf->db_tioreq = ti;
                if (buf->db_flags & PA_WRITE)
                        ti->ti_req_type = TI_READ_WRITE;

                if (opcode == M0_OC_FREE &&
                    should_unit_be_truncated(map->pi_trunc_partial,
                                             unit_type, buf->db_flags)) {
                        tseg   = SEG_NR(trunc_ivec);
                        INDEX(trunc_ivec, tseg) = pgstart;
                        COUNT(trunc_ivec, tseg) = pgend - pgstart;
                        ++trunc_ivec->iv_vec.v_nr;
                        M0_LOG(M0_DEBUG, "Seg id %d [%" PRIu64 ", %" PRIu64 "]"
                                         "added to target ioreq with "FID_F,
                                         tseg, INDEX(trunc_ivec, tseg),
                                         COUNT(trunc_ivec, tseg),
                                         FID_P(&ti->ti_fid));
                }

                if (opcode == M0_OC_FREE && !map->pi_trunc_partial)
                        pattr[seg] |= PA_TRUNC;

                M0_ASSERT(addr_is_network_aligned(buf->db_buf.b_addr));
                bvec->ov_buf[seg] = buf->db_buf.b_addr;
                bvec->ov_vec.v_count[seg] = COUNT(ivec, seg);
                if (map->pi_rtype == PIR_READOLD &&
                    unit_type == M0_PUT_DATA) {
                        M0_ASSERT(buf->db_auxbuf.b_addr != NULL);
                        auxbvec->ov_buf[seg] = buf->db_auxbuf.b_addr;
                        auxbvec->ov_vec.v_count[seg] = page_size;
                }
                pattr[seg] |= buf->db_flags;
                M0_LOG(M0_DEBUG, "pageaddr=%p, auxpage=%p,"
                                 " index=%6" PRIu64 ", size=%4"PRIu64
                                 " grpid=%3" PRIu64 " flags=%4x for "FID_F,
                                 bvec->ov_buf[seg], auxbvec->ov_buf[seg],
                                 INDEX(ivec, seg), COUNT(ivec, seg),
                                 map->pi_grpid, pattr[seg],
                                 FID_P(&ti->ti_fid));
                M0_LOG(M0_DEBUG, "Seg id %d [%" PRIu64 ", %"PRIu64
                                 "] added to target_ioreq with "FID_F
                                 " with flags 0x%x: ", seg,
                                 INDEX(ivec, seg), COUNT(ivec, seg),
                                 FID_P(&ti->ti_fid), pattr[seg]);
                goff_span_ext.e_start = ioo->ioo_ext.iv_index[0];
                goff_span_ext.e_end = ioo->ioo_ext.iv_index[ioo->ioo_ext.iv_vec.v_nr - 1]
                                      + ioo->ioo_ext.iv_vec.v_count[ioo->ioo_ext.iv_vec.v_nr - 1];
                /* If ioo_attr struct is not allocated then skip checksum computation */
                is_goff_in_range = m0_ext_is_in(&goff_span_ext, goff) &&
                                        m0__obj_is_di_enabled(ioo);
                if (dst_attr != NULL && unit_type == M0_PUT_DATA &&
                    opcode == M0_OC_WRITE && is_goff_in_range) {
                        void         *src_attr;
                        m0_bcount_t   cs_sz;

                        cs_sz = ioo->ioo_attr.ov_vec.v_count[0];
                        /* This we can do as page_size <= unit_sz */
                        b_nob = m0_extent_get_checksum_nob(goff,
                                                           COUNT(ivec, seg),
                                                           unit_sz, cs_sz );
                        if (b_nob) {
                                /* This function will get checksum address from application provided
                                 * buffer. Checksum is corresponding to on gob offset and ioo_ext and
                                 * this function helps to locate exact address for the above.
                                 * Note: ioo_ext is span of offset for which ioo_attr is provided and
                                 * goff should lie within that span
                                 */
                                src_attr = m0_extent_vec_get_checksum_addr( &ioo->ioo_attr, goff,
                                                &ioo->ioo_ext, unit_sz, cs_sz);
                                M0_ASSERT(b_nob == cs_sz);
                                memcpy((char *)dst_attr + ti->ti_cksum_copied, src_attr, b_nob);

                                /* Track checksum copied as we need to do overallocation for
                                 * ti_attrbuf for traget and while sending FOP we use this
                                 * counter to send the actual checksum size.
                                 */
                                ti->ti_cksum_copied += b_nob;

                                /* Make sure we are not exceeding the allocated buffer size */
                                M0_ASSERT(ti->ti_cksum_copied <= ti->ti_attrbuf.b_nob);
                        }

                        ti->ti_cksum_seg_b_nob[seg] = b_nob;
                } else if (goff_ivec != NULL && unit_type == M0_PUT_DATA &&
                                opcode == M0_OC_READ && is_goff_in_range) {
                        INDEX(goff_ivec, seg) = goff;
                        COUNT(goff_ivec, seg) = COUNT(ivec, seg);
                        goff_ivec->iv_vec.v_nr++;
                }

                goff += COUNT(ivec, seg);
                ++ivec->iv_vec.v_nr;
                pgstart = pgend;
        }
        M0_LEAVE();
}

M0_INTERNAL struct m0_fid target_fid(struct m0_op_io *ioo,
                                     struct m0_pdclust_tgt_addr *tgt)
{
        struct m0_fid fid;

        m0_poolmach_gob2cob(ioo_to_poolmach(ioo),
                            &ioo->ioo_oo.oo_fid, tgt->ta_obj,
                            &fid);
        return fid;
}

static inline struct m0_rpc_session *
target_session(struct m0_op_io *ioo, struct m0_fid tfid)
{
        struct m0_op           *op;
        struct m0_pool_version *pv;
        struct m0_client       *instance;

        M0_PRE(ioo != NULL);
        op = &ioo->ioo_oo.oo_oc.oc_op;
        instance = m0__op_instance(op);
        pv = m0_pool_version_find(&instance->m0c_pools_common, &ioo->ioo_pver);
        M0_ASSERT(pv != NULL);

        return m0_obj_container_id_to_session(
                        pv, m0_fid_cob_device_id(&tfid));
}

static int bulk_buffer_add(struct ioreq_fop        *irfop,
                           struct m0_net_domain    *dom,
                           struct m0_rpc_bulk_buf **rbuf,
                           uint32_t                *delta,
                           uint32_t                 maxsize)
{
        int                      rc;
        int                      seg_nr;
        struct m0_op_io  *ioo;
        struct m0_indexvec      *ivec;

        M0_PRE(irfop  != NULL);
        M0_PRE(dom    != NULL);
        M0_PRE(rbuf   != NULL);
        M0_PRE(delta  != NULL);
        M0_PRE(maxsize > 0);
        M0_ENTRY("ioreq_fop %p net_domain %p delta_size %d",
                 irfop, dom, *delta);

        ioo     = bob_of(irfop->irf_tioreq->ti_nwxfer, struct m0_op_io,
                         ioo_nwxfer, &ioo_bobtype);
        ivec    = M0_IN(ioreq_sm_state(ioo), (IRS_READING, IRS_WRITING)) ?
                        &irfop->irf_tioreq->ti_ivec :
                        &irfop->irf_tioreq->ti_dgvec->dr_ivec;
        seg_nr  = min32(m0_net_domain_get_max_buffer_segments(dom),
                        SEG_NR(ivec));
        *delta += io_desc_size(dom);

        if (m0_io_fop_size_get(&irfop->irf_iofop.if_fop) + *delta < maxsize) {
                rc = m0_rpc_bulk_buf_add(&irfop->irf_iofop.if_rbulk, seg_nr,
                                         0, dom, NULL, rbuf);
                if (rc != 0) {
                        *delta -= io_desc_size(dom);
                        return M0_ERR(rc);
                }
        } else {
                rc      = -ENOSPC;
                *delta -= io_desc_size(dom);
        }

        M0_POST(ergo(rc == 0, *rbuf != NULL));
        return M0_RC(rc);
}

static void irfop_fini(struct ioreq_fop *irfop)
{
        M0_ENTRY("ioreq_fop %p", irfop);

        M0_PRE(irfop != NULL);

        m0_rpc_bulk_buflist_empty(&irfop->irf_iofop.if_rbulk);
        ioreq_fop_fini(irfop);
        m0_free(irfop);

        M0_LEAVE();
}

static void *buf_aux_chk_get(struct m0_bufvec *aux, enum page_attr p_attr,
                             uint32_t seg_idx, bool rd_in_wr)
{
        return (p_attr == PA_DATA && rd_in_wr && aux != NULL &&
                aux->ov_buf[seg_idx] != NULL) ? aux->ov_buf[seg_idx] : NULL;
}

static int target_ioreq_iofops_prepare(struct target_ioreq *ti,
                                       enum page_attr       filter)
{
        int                          rc = 0;
        uint32_t                     seg = 0;
        /* Number of segments in one m0_rpc_bulk_buf structure. */
        uint32_t                     bbsegs;
        uint32_t                     maxsize;
        uint32_t                     delta;
        uint32_t                     fop_cksm_nob;
        uint32_t                     dispatched_cksm_nob = 0;
        enum page_attr               rw;
        enum page_attr              *pattr;
        struct m0_bufvec            *bvec;
        struct m0_bufvec            *auxbvec;
        struct m0_op_io             *ioo;
        struct m0_obj_attr          *io_attr;
        struct m0_indexvec          *ivec;
        struct ioreq_fop            *irfop;
        struct m0_net_domain        *ndom;
        struct m0_rpc_bulk_buf      *rbuf;
        struct m0_io_fop            *iofop;
        struct m0_fop_cob_rw        *rw_fop;
        struct nw_xfer_request      *xfer;
        /* Is it in the READ phase of WRITE request. */
        bool                         read_in_write = false;
        void                        *buf;
        void                        *bufnext;
        m0_bcount_t                  max_seg_size;
        m0_bcount_t                  xfer_len;
        m0_bindex_t                  offset;
        uint32_t                     segnext;
        uint32_t                     ndom_max_segs;
        struct m0_client            *instance;

        M0_ENTRY("prepare io fops for target ioreq %p filter 0x%x, tfid "FID_F,
                 ti, filter, FID_P(&ti->ti_fid));

        M0_PRE(target_ioreq_invariant(ti));
        M0_PRE(M0_IN(filter, (PA_DATA, PA_PARITY)));

        rc = m0_rpc_session_validate(ti->ti_session);
        if (rc != 0 && (!M0_IN(rc, (-ECANCELED, -EINVAL))))
                return M0_ERR(rc);

        xfer = ti->ti_nwxfer;
        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        M0_ASSERT(M0_IN(ioreq_sm_state(ioo),
                        (IRS_READING, IRS_DEGRADED_READING,
                         IRS_WRITING, IRS_DEGRADED_WRITING)));

        if (M0_IN(ioo->ioo_oo.oo_oc.oc_op.op_code, (M0_OC_WRITE,
                                                    M0_OC_FREE)) &&
            M0_IN(ioreq_sm_state(ioo), (IRS_READING, IRS_DEGRADED_READING)))
                read_in_write = true;

        if (M0_IN(ioreq_sm_state(ioo), (IRS_READING, IRS_WRITING))) {
                ivec    = &ti->ti_ivec;
                bvec    = &ti->ti_bufvec;
                auxbvec = &ti->ti_auxbufvec;
                pattr   = ti->ti_pageattrs;
        } else {
                if (ti->ti_dgvec == NULL) {
                        return M0_RC(0);
                }
                ivec    = &ti->ti_dgvec->dr_ivec;
                bvec    = &ti->ti_dgvec->dr_bufvec;
                auxbvec = &ti->ti_dgvec->dr_auxbufvec;
                pattr   = ti->ti_dgvec->dr_pageattrs;
        }

        ndom = ti->ti_session->s_conn->c_rpc_machine->rm_tm.ntm_dom;
        rw = ioreq_sm_state(ioo) == IRS_DEGRADED_WRITING ? PA_DGMODE_WRITE :
             ioreq_sm_state(ioo) == IRS_WRITING ? PA_WRITE :
             ioreq_sm_state(ioo) == IRS_DEGRADED_READING ? PA_DGMODE_READ :
             PA_READ;
        maxsize = m0_rpc_session_get_max_item_payload_size(ti->ti_session);

        max_seg_size = m0_net_domain_get_max_buffer_segment_size(ndom);

        ndom_max_segs = m0_net_domain_get_max_buffer_segments(ndom);

        while (seg < SEG_NR(ivec)) {
                delta  = 0;
                bbsegs = 0;

                M0_LOG(M0_DEBUG, "pageattr = %u, filter = %u, rw = %u",
                       pattr[seg], filter, rw);

                if (!(pattr[seg] & filter) || !(pattr[seg] & rw) ||
                     (pattr[seg] & PA_TRUNC)) {
                        ++seg;
                        continue;
                }

                M0_ALLOC_PTR(irfop);
                if (irfop == NULL) {
                        rc = M0_ERR(-ENOMEM);
                        goto err;
                }
                rc = ioreq_fop_init(irfop, ti, filter);
                if (rc != 0) {
                        m0_free(irfop);
                        goto err;
                }
                fop_cksm_nob = 0;

                iofop = &irfop->irf_iofop;
                rw_fop = io_rw_get(&iofop->if_fop);

                rc = bulk_buffer_add(irfop, ndom, &rbuf, &delta, maxsize);
                if (rc != 0) {
                        ioreq_fop_fini(irfop);
                        m0_free(irfop);
                        goto err;
                }
                delta += io_seg_size();


                /*
                * Adds io segments and io descriptor only if it fits within
                * permitted size.
                */
                /* TODO: can this loop become a function call?
                 * -- too many levels of indentation */
                while (seg < SEG_NR(ivec) &&
                       m0_io_fop_size_get(&iofop->if_fop) + delta < maxsize &&
                       bbsegs < ndom_max_segs) {

                        /*
                        * Adds a page to rpc bulk buffer only if it passes
                        * through the filter.
                        */
                        if (pattr[seg] & rw && pattr[seg] & filter &&
                            !(pattr[seg] & PA_TRUNC)) {
                                delta += io_seg_size() + io_di_size(ioo);

                                buf = buf_aux_chk_get(auxbvec, filter, seg,
                                                      read_in_write);

                                if (buf == NULL) {
                                        buf = bvec->ov_buf[seg];
                                        /* Add the size for checksum generated for every segment, skip parity */
                                        if ((filter == PA_DATA) && m0__obj_is_di_enabled(ioo) &&
                                            (ioo->ioo_oo.oo_oc.oc_op.op_code == M0_OC_WRITE)) {
                                                delta += ti->ti_cksum_seg_b_nob[seg];
                                                fop_cksm_nob += ti->ti_cksum_seg_b_nob[seg];
                                        }
                                }

                                xfer_len = COUNT(ivec, seg);
                                offset = INDEX(ivec, seg);

                                /*
                                 * Accommodate multiple pages in a single
                                 * net buffer segment, if they are consecutive
                                 * pages.
                                 */
                                segnext = seg + 1;
                                while (segnext < SEG_NR(ivec) &&
                                       xfer_len < max_seg_size) {
                                        bufnext = buf_aux_chk_get(auxbvec,
                                                                  filter,
                                                                  segnext,
                                                                  read_in_write);
                                        if (bufnext == NULL)
                                                bufnext = bvec->ov_buf[segnext];

                                        if (buf + xfer_len == bufnext) {
                                                xfer_len += COUNT(ivec, ++seg);
                                                segnext = seg + 1;
                                        } else
                                                break;
                                }

                                rc = m0_rpc_bulk_buf_databuf_add(rbuf, buf,
                                                                 xfer_len,
                                                                 offset, ndom);

                                if (rc == -EMSGSIZE) {
                                        /*
                                         * Fix the number of segments in
                                         * current m0_rpc_bulk_buf structure.
                                         */
                                        rbuf->bb_nbuf->nb_buffer.ov_vec.v_nr =
                                                bbsegs;
                                        rbuf->bb_zerovec.z_bvec.ov_vec.v_nr =
                                                bbsegs;
                                        bbsegs = 0;

                                        delta -= io_seg_size() - io_di_size(ioo);

                                        if ((filter == PA_DATA) && m0__obj_is_di_enabled(ioo) &&
                                            (ioo->ioo_oo.oo_oc.oc_op.op_code == M0_OC_WRITE)) {
                                                delta -= ti->ti_cksum_seg_b_nob[seg];
                                                fop_cksm_nob -= ti->ti_cksum_seg_b_nob[seg];
                                        }

                                        /*
                                         * Buffer must be 4k aligned to be
                                         * used by network hw
                                         */
                                        M0_ASSERT(addr_is_network_aligned(buf));
                                        rc     = bulk_buffer_add(irfop, ndom,
                                                        &rbuf, &delta, maxsize);
                                        if (rc == -ENOSPC)
                                                break;
                                        else if (rc != 0)
                                                goto fini_fop;

                                        /*
                                         * Since current bulk buffer is full,
                                         * new bulk buffer is added and
                                         * existing segment is attempted to
                                         * be added to new bulk buffer.
                                         */
                                        continue;
                                } else if (rc == 0)
                                        ++bbsegs;
                        }

                        ++seg;
                }

                if (m0_io_fop_byte_count(iofop) == 0) {
                        irfop_fini(irfop);
                        continue;
                }

                rbuf->bb_nbuf->nb_buffer.ov_vec.v_nr = bbsegs;
                rbuf->bb_zerovec.z_bvec.ov_vec.v_nr = bbsegs;

                rw_fop->crw_fid = ti->ti_fid;
                rw_fop->crw_pver = ioo->ioo_pver;
                rw_fop->crw_index = ti->ti_obj;

                /*
                 * Use NOHOLE by default (i.e. return error for missing
                 * units instead of zeros), unless we are in read-verify
                 * mode or in the degraded-read mode. Otherwise, in case of
                 * partially spanned parity groups (last groups, usually),
                 * we will get a lot of bogus errors when all data units
                 * of the group are read.
                 *
                 * Note: parity units are always present in the groups, even
                 * in the partially spanned ones. So we always use NOHOLE
                 * for them. Otherwise, the user may get corrupted data.
                 */
                instance = m0__op_instance(&ioo->ioo_oo.oo_oc.oc_op);
                if (ioreq_sm_state(ioo) == IRS_READING && !read_in_write &&
                    (filter == PA_PARITY ||
                     (!instance->m0c_config->mc_is_read_verify &&
                      !(ioo->ioo_flags & M0_OOF_HOLE))))
                        rw_fop->crw_flags |= M0_IO_FLAG_NOHOLE;

                if (ioreq_sm_state(ioo) == IRS_DEGRADED_READING &&
                    !read_in_write && filter == PA_PARITY)
                        rw_fop->crw_flags |= M0_IO_FLAG_NOHOLE;

                /* Assign the checksum buffer for traget */
                if (filter == PA_DATA && m0__obj_is_di_enabled(ioo)) {
                        if (m0_is_write_fop(&iofop->if_fop))    {
                                M0_ASSERT(fop_cksm_nob != 0);
                                /* RPC layer to free crw_di_data_cksum */
                                if ( m0_buf_alloc(&rw_fop->crw_di_data_cksum, fop_cksm_nob) != 0 )
                                        goto fini_fop;

                                memcpy( rw_fop->crw_di_data_cksum.b_addr,
                                                ti->ti_attrbuf.b_addr + dispatched_cksm_nob,
                                                fop_cksm_nob );
                                dispatched_cksm_nob += fop_cksm_nob;
                                M0_ASSERT(dispatched_cksm_nob <= ti->ti_cksum_copied);
                        }
                        else {
                                rw_fop->crw_di_data_cksum.b_addr = NULL;
                                rw_fop->crw_di_data_cksum.b_nob = 0;
                        }

                        rw_fop->crw_cksum_size = (read_in_write ||
                                                 !m0__obj_is_di_enabled(ioo)) ?
                                                 0 : ioo->ioo_attr.ov_vec.v_count[0];
                }
                else {
                        rw_fop->crw_di_data_cksum.b_addr = NULL;
                        rw_fop->crw_di_data_cksum.b_nob  = 0;
                        rw_fop->crw_cksum_size = 0;
                }


                if (ioo->ioo_flags & M0_OOF_SYNC)
                        rw_fop->crw_flags |= M0_IO_FLAG_SYNC;
                io_attr = m0_io_attr(ioo);
                rw_fop->crw_lid = io_attr->oa_layout_id;

                /*
                 * XXX(Sining): This is a bit tricky: m0_io_fop_prepare in
                 * ioservice/io_fops.c calls io_fop_di_prepare which has only
                 * file system in mind and uses super block and file related
                 * information to do something (it returns 0 directly for user
                 * space). This is not the case for Client kernel mode!!
                 *
                 * Simply return 0 just like it does for user space at this
                 * moment.
                 */
                rc = m0_io_fop_prepare(&iofop->if_fop);
                if (rc != 0)
                        goto fini_fop;

                if (m0_is_read_fop(&iofop->if_fop))
                        m0_atomic64_add(&xfer->nxr_rdbulk_nr,
                                        m0_rpc_bulk_buf_length(
                                        &iofop->if_rbulk));

                m0_atomic64_inc(&ti->ti_nwxfer->nxr_iofop_nr);
                iofops_tlist_add(&ti->ti_iofops, irfop);

                M0_LOG(M0_DEBUG,
                       "fop=%p bulk=%p (%s) @"FID_F" io fops = %"PRIu64
                       " read bulks = %" PRIu64 ", list_len=%d",
                       &iofop->if_fop, &iofop->if_rbulk,
                       m0_is_read_fop(&iofop->if_fop) ? "r" : "w",
                       FID_P(&ti->ti_fid),
                       m0_atomic64_get(&xfer->nxr_iofop_nr),
                       m0_atomic64_get(&xfer->nxr_rdbulk_nr),
                       (int)iofops_tlist_length(&ti->ti_iofops));
        }

        return M0_RC(0);

fini_fop:
        irfop_fini(irfop);
err:
        m0_tl_teardown(iofops, &ti->ti_iofops, irfop) {
                irfop_fini(irfop);
        }

        return M0_ERR(rc);
}
static int target_cob_fop_prepare(struct target_ioreq *ti);
static const struct target_ioreq_ops tioreq_ops = {
        .tio_seg_add         = target_ioreq_seg_add,
        .tio_iofops_prepare  = target_ioreq_iofops_prepare,
        .tio_cc_fops_prepare = target_cob_fop_prepare,
};

static int target_cob_fop_prepare(struct target_ioreq *ti)
{
        int rc;
        M0_ENTRY("ti = %p type = %d", ti, ti->ti_req_type);
        M0_PRE(M0_IN(ti->ti_req_type, (TI_COB_CREATE, TI_COB_TRUNCATE)));

        rc = ioreq_cc_fop_init(ti);
        return M0_RC(rc);
}

static int target_ioreq_init(struct target_ioreq    *ti,
                             struct nw_xfer_request *xfer,
                             const struct m0_fid    *cobfid,
                             uint64_t                ta_obj,
                             struct m0_rpc_session  *session,
                             uint64_t                size)
{
        int                     rc;
        struct m0_op_io        *ioo;
        struct m0_op           *op;
        struct m0_client       *instance;
        uint32_t                nr;

        M0_PRE(cobfid  != NULL);
        M0_ENTRY("target_ioreq %p, nw_xfer_request %p, "FID_F,
                 ti, xfer, FID_P(cobfid));

        M0_PRE(ti      != NULL);
        M0_PRE(xfer    != NULL);
        M0_PRE(session != NULL);
        M0_PRE(size    >  0);

        ti->ti_rc        = 0;
        ti->ti_ops       = &tioreq_ops;
        ti->ti_fid       = *cobfid;
        ti->ti_nwxfer    = xfer;
        ti->ti_dgvec     = NULL;
        ti->ti_req_type  = TI_NONE;
        M0_SET0(&ti->ti_cc_fop);
        ti->ti_cc_fop_inited = false;

        /*
         * Target object is usually in ONLINE state unless explicitly
         * told otherwise.
         */
        ti->ti_state     = M0_PNDS_ONLINE;
        ti->ti_session   = session;
        ti->ti_parbytes  = 0;
        ti->ti_databytes = 0;

        ioo      = bob_of(xfer, struct m0_op_io, ioo_nwxfer,
                          &ioo_bobtype);
        op       = &ioo->ioo_oo.oo_oc.oc_op;
        instance = m0__op_instance(op);
        M0_PRE(instance != NULL);

        ti->ti_obj = ta_obj;

        iofops_tlist_init(&ti->ti_iofops);
        tioreqht_tlink_init(ti);
        target_ioreq_bob_init(ti);

        nr = page_nr(size, ioo->ioo_obj);
        rc = m0_indexvec_alloc(&ti->ti_ivec, nr);
        if (rc != 0)
                goto out;

        if (op->op_code == M0_OC_READ) {
                rc = m0_indexvec_alloc(&ti->ti_goff_ivec, nr);
                ti->ti_goff_ivec.iv_vec.v_nr = 0;
                if (rc != 0)
                        goto fail;
        }

        if (op->op_code == M0_OC_FREE) {
                rc = m0_indexvec_alloc(&ti->ti_trunc_ivec, nr);
                if (rc != 0)
                        goto fail;
        }

        ti->ti_bufvec.ov_vec.v_nr = nr;
        M0_ALLOC_ARR(ti->ti_bufvec.ov_vec.v_count, nr);
        if (ti->ti_bufvec.ov_vec.v_count == NULL)
                goto fail;

        M0_ALLOC_ARR(ti->ti_bufvec.ov_buf, nr);
        if (ti->ti_bufvec.ov_buf == NULL)
                goto fail;

        /* Memory allocation for checksum computation */
        if (op->op_code == M0_OC_WRITE && m0__obj_is_di_enabled(ioo)) {
                uint32_t b_nob;

                ti->ti_attrbuf.b_addr = NULL;
                b_nob = (size * ioo->ioo_attr.ov_vec.v_count[0]) /
                         m0_obj_layout_id_to_unit_size(m0__obj_lid(ioo->ioo_obj));
                rc = m0_buf_alloc(&ti->ti_attrbuf, b_nob);
                if (rc != 0)
                        goto fail;
                ti->ti_cksum_copied = 0;
                M0_ALLOC_ARR(ti->ti_cksum_seg_b_nob, nr);
        }
        else {
                ti->ti_attrbuf.b_addr = NULL;
                ti->ti_attrbuf.b_nob = 0;
                ti->ti_cksum_copied = 0;
                ti->ti_cksum_seg_b_nob = NULL;
        }
        /*
         * For READOLD method, an extra bufvec is needed to remember
         * the addresses of auxillary buffers so those auxillary
         * buffers can be used in rpc bulk transfer to avoid polluting
         * real data buffers which are the application's memory for IO
         * in case zero copy method is in use.
         */
        ti->ti_auxbufvec.ov_vec.v_nr = nr;
        M0_ALLOC_ARR(ti->ti_auxbufvec.ov_vec.v_count, nr);
        if (ti->ti_auxbufvec.ov_vec.v_count == NULL)
                goto fail;

        M0_ALLOC_ARR(ti->ti_auxbufvec.ov_buf, nr);
        if (ti->ti_auxbufvec.ov_buf == NULL)
                goto fail;

        if (M0_FI_ENABLED("no-mem-err"))
                goto fail;
        M0_ALLOC_ARR(ti->ti_pageattrs, nr);
        if (ti->ti_pageattrs == NULL)
                goto fail;

        /*
         * This value is incremented when new segments are added to the
         * index vector in target_ioreq_seg_add().
         */
        ti->ti_ivec.iv_vec.v_nr = 0;
        ti->ti_trunc_ivec.iv_vec.v_nr = 0;

        M0_POST_EX(target_ioreq_invariant(ti));
        return M0_RC(0);
fail:
        m0_indexvec_free(&ti->ti_ivec);
        if (op->op_code == M0_OC_READ)
                m0_indexvec_free(&ti->ti_goff_ivec);
        if (op->op_code == M0_OC_FREE)
                m0_indexvec_free(&ti->ti_trunc_ivec);
        m0_free(ti->ti_bufvec.ov_vec.v_count);
        m0_free(ti->ti_bufvec.ov_buf);
        m0_free(ti->ti_auxbufvec.ov_vec.v_count);
        m0_free(ti->ti_auxbufvec.ov_buf);

out:
        return M0_ERR(-ENOMEM);
}

static int nw_xfer_tioreq_get(struct nw_xfer_request *xfer,
                              struct m0_fid          *fid,
                              uint64_t                ta_obj,
                              struct m0_rpc_session  *session,
                              uint64_t                size,
                              struct target_ioreq   **out)
{
        int                     rc = 0;
        struct target_ioreq    *ti;
        struct m0_op_io        *ioo;
        struct m0_op           *op;
        struct m0_client       *instance;

        M0_PRE(fid != NULL);
        M0_ENTRY("nw_xfer_request %p, "FID_F, xfer, FID_P(fid));

        M0_PRE(session != NULL);
        M0_PRE(out != NULL);
        M0_PRE(nw_xfer_request_invariant(xfer));

        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        op = &ioo->ioo_oo.oo_oc.oc_op;
        instance = m0__op_instance(op);
        M0_PRE(instance != NULL);

        ti = target_ioreq_locate(xfer, fid);
        if (ti == NULL) {
                M0_ALLOC_PTR(ti);
                if (ti == NULL)
                        return M0_ERR(-ENOMEM);

                rc = target_ioreq_init(ti, xfer, fid, ta_obj, session, size);
                if (rc == 0) {
                        tioreqht_htable_add(&xfer->nxr_tioreqs_hash, ti);
                        M0_LOG(M0_INFO, "New target_ioreq %p added for "FID_F,
                                        ti, FID_P(fid));
                } else {
                        m0_free(ti);
                        return M0_ERR_INFO(rc, "target_ioreq_init() failed");
                }
        }

        if (ti->ti_dgvec == NULL && M0_IN(ioreq_sm_state(ioo),
                (IRS_DEGRADED_READING, IRS_DEGRADED_WRITING)))
                rc = dgmode_rwvec_alloc_init(ti);

        *out = ti;

        return M0_RC(rc);
}

static void databufs_set_dgw_mode(struct pargrp_iomap *iomap,
                                  struct m0_pdclust_layout *play,
                                  struct m0_ext *ext)
{
        uint32_t         row_start;
        uint32_t         row_end;
        uint32_t         row;
        uint32_t         col;
        m0_bcount_t      grp_off;
        struct data_buf *dbuf;

        grp_off = data_size(play) * iomap->pi_grpid;
        page_pos_get(iomap, ext->e_start, grp_off, &row_start, &col);
        page_pos_get(iomap, ext->e_end - 1, grp_off, &row_end, &col);

        for (row = row_start; row <= row_end; ++row) {
                dbuf = iomap->pi_databufs[row][col];
                if (dbuf->db_flags & PA_WRITE)
                        dbuf->db_flags |= PA_DGMODE_WRITE;
        }
}

static void paritybufs_set_dgw_mode(struct pargrp_iomap *iomap,
                                    struct m0_op_io *ioo,
                                    uint64_t unit)
{
        uint32_t                  row;
        uint32_t                  col;
        struct data_buf          *dbuf;
        struct m0_pdclust_layout *play = pdlayout_get(ioo);
        uint64_t                  unit_size = layout_unit_size(play);

        parity_page_pos_get(iomap, unit * unit_size, &row, &col);
        for (; row < rows_nr(play, ioo->ioo_obj); ++row) {
                dbuf = iomap->pi_paritybufs[row][col];
                if (m0_pdclust_is_replicated(play) &&
                    iomap->pi_databufs[row][0] == NULL)
                        continue;
                if (dbuf->db_flags & PA_WRITE)
                        dbuf->db_flags |= PA_DGMODE_WRITE;
        }
}

static int nw_xfer_io_distribute(struct nw_xfer_request *xfer)
{
        bool                        do_cobs = true;
        int                         rc = 0;
        unsigned int                op_code;
        uint64_t                    i;
        uint64_t                    unit;
        uint64_t                    unit_size;
        uint64_t                    count;
        uint64_t                    pgstart;
        struct m0_op               *op;
        /* Extent representing a data unit. */
        struct m0_ext               u_ext;
        /* Extent representing resultant extent. */
        struct m0_ext               r_ext;
        /* Extent representing a segment from index vector. */
        struct m0_ext               v_ext;
        struct m0_op_io            *ioo;
        struct target_ioreq        *ti;
        struct m0_ivec_cursor       cursor;
        struct m0_pdclust_layout   *play;
        enum m0_pdclust_unit_type   unit_type;
        struct m0_pdclust_src_addr  src;
        struct m0_pdclust_tgt_addr  tgt;
        struct m0_bitmap            units_spanned;
        struct pargrp_iomap        *iomap;
        struct m0_client           *instance;

        M0_ENTRY("nw_xfer_request %p", xfer);

        M0_PRE(nw_xfer_request_invariant(xfer));

        ioo       = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        op        = &ioo->ioo_oo.oo_oc.oc_op;
        op_code   = op->op_code,
        play      = pdlayout_get(ioo);
        unit_size = layout_unit_size(play);
        instance  = m0__op_instance(op);

        /*
         * In non-oostore mode, all cobs are created on object creation.
         * In oostore mode, CROW is enabled and cobs are created automatically
         * at the server side on the 1st write request. But, because of SNS,
         * we need to create cobs for the spare units, and to make sure all cobs
         * are created for all units in the parity group touched by the update
         * request. See more below.
         */
        if (!m0__is_oostore(instance) || op_code == M0_OC_READ)
                do_cobs = false;
        /*
         * In replicated layout (N == 1), all units in the parity group are
         * always spanned. And there are no spare units, so...
         */
        if (ioo->ioo_pbuf_type == M0_PBUF_IND)
                do_cobs = false;

        if (do_cobs) {
                rc = m0_bitmap_init(&units_spanned, m0_pdclust_size(play));
                if (rc != 0)
                        return M0_ERR(rc);
        }

        for (i = 0; i < ioo->ioo_iomap_nr; ++i) {
                count        = 0;
                iomap        = ioo->ioo_iomaps[i];
                pgstart      = data_size(play) * iomap->pi_grpid;
                src.sa_group = iomap->pi_grpid;

                M0_LOG(M0_DEBUG, "xfer=%p map=%p [grpid=%" PRIu64 " state=%u]",
                                 xfer, iomap, iomap->pi_grpid, iomap->pi_state);

                if (do_cobs)
                        m0_bitmap_reset(&units_spanned);

                /* traverse parity group ivec by units */
                m0_ivec_cursor_init(&cursor, &iomap->pi_ivec);
                while (!m0_ivec_cursor_move(&cursor, count)) {
                        unit = (m0_ivec_cursor_index(&cursor) - pgstart) /
                                unit_size;

                        u_ext.e_start = pgstart + unit * unit_size;
                        u_ext.e_end   = u_ext.e_start + unit_size;

                        v_ext.e_start  = m0_ivec_cursor_index(&cursor);
                        v_ext.e_end    = v_ext.e_start +
                                m0_ivec_cursor_step(&cursor);

                        m0_ext_intersection(&u_ext, &v_ext, &r_ext);
                        M0_ASSERT(m0_ext_is_valid(&r_ext));
                        count = m0_ext_length(&r_ext);

                        unit_type = m0_pdclust_unit_classify(play, unit);
                        M0_ASSERT(unit_type == M0_PUT_DATA);

                        if (ioreq_sm_state(ioo) == IRS_DEGRADED_WRITING)
                                databufs_set_dgw_mode(iomap, play, &r_ext);

                        src.sa_unit = unit;
                        rc = xfer->nxr_ops->nxo_tioreq_map(xfer, &src, &tgt,
                                                           &ti);
                        if (rc != 0)
                                goto err;

                        ti->ti_ops->tio_seg_add(ti, &src, &tgt, r_ext.e_start,
                                                m0_ext_length(&r_ext), iomap);
                        if (op_code == M0_OC_WRITE && do_cobs &&
                            ti->ti_req_type == TI_READ_WRITE)
                                m0_bitmap_set(&units_spanned, unit, true);

                }

                M0_ASSERT(ergo(M0_IN(op_code, (M0_OC_READ, M0_OC_WRITE)),
                               m0_vec_count(&ioo->ioo_ext.iv_vec) ==
                               m0_vec_count(&ioo->ioo_data.ov_vec)));

                /* process parity units */
                if (M0_IN(ioo->ioo_pbuf_type, (M0_PBUF_DIR,
                                               M0_PBUF_IND)) ||
                    (ioreq_sm_state(ioo) == IRS_DEGRADED_READING &&
                     iomap->pi_state == PI_DEGRADED)) {

                        for (unit = 0; unit < layout_k(play); ++unit) {
                                src.sa_unit = layout_n(play) + unit;
                                M0_ASSERT(m0_pdclust_unit_classify(play,
                                          src.sa_unit) == M0_PUT_PARITY);

                                rc = xfer->nxr_ops->nxo_tioreq_map(xfer, &src,
                                                                   &tgt, &ti);
                                if (rc != 0)
                                        goto err;

                                if (ioreq_sm_state(ioo) == IRS_DEGRADED_WRITING)
                                        paritybufs_set_dgw_mode(iomap, ioo,
                                                                unit);

                                if (op_code == M0_OC_WRITE && do_cobs)
                                        m0_bitmap_set(&units_spanned,
                                                      src.sa_unit, true);

                                ti->ti_ops->tio_seg_add(ti, &src, &tgt, pgstart,
                                                        layout_unit_size(play),
                                                        iomap);
                        }

                        if (!do_cobs)
                                continue; /* to next iomap */

                        /*
                         * Create cobs for all units not spanned by the
                         * IO request (data or spare units).
                         *
                         * If some data unit is not present in the group (hole
                         * or not complete last group), we still need to create
                         * cob for it. Otherwise, during SNS-repair the receiver
                         * will wait forever for this unit without knowing that
                         * its size is actually zero.
                         */
                        for (unit = 0; unit < m0_pdclust_size(play); ++unit) {
                                if (m0_bitmap_get(&units_spanned, unit))
                                        continue;

                                src.sa_unit = unit;
                                rc = xfer->nxr_ops->nxo_tioreq_map(xfer, &src,
                                                                   &tgt, &ti);
                                if (rc != 0)
                                        M0_LOG(M0_ERROR, "[%p] map=%p "
                                               "nxo_tioreq_map() failed: rc=%d",
                                               ioo, iomap, rc);
                                /*
                                 * Skip the case when some other parity group
                                 * has spanned the particular target already.
                                 */
                                if (ti->ti_req_type != TI_NONE)
                                        continue;

                                ti->ti_req_type = TI_COB_CREATE;
                        }
                }
        }

        if (do_cobs)
                m0_bitmap_fini(&units_spanned);

        M0_ASSERT(ergo(M0_IN(op_code, (M0_OC_READ, M0_OC_WRITE)),
                       m0_vec_count(&ioo->ioo_ext.iv_vec) ==
                       m0_vec_count(&ioo->ioo_data.ov_vec)));

        return M0_RC(0);
err:
        m0_htable_for(tioreqht, ti, &xfer->nxr_tioreqs_hash) {
                tioreqht_htable_del(&xfer->nxr_tioreqs_hash, ti);
                target_ioreq_fini(ti);
                m0_free0(&ti);
        } m0_htable_endfor;

        return M0_ERR(rc);
}

static void nw_xfer_req_complete(struct nw_xfer_request *xfer, bool rmw)
{
        struct m0_client       *instance;
        struct m0_op_io        *ioo;
        struct target_ioreq    *ti;
        struct ioreq_fop       *irfop;
        struct m0_fop          *fop;
        struct m0_rpc_item     *item;

        M0_ENTRY("nw_xfer_request %p, rmw %s", xfer,
                 rmw ? (char *)"true" : (char *)"false");

        M0_PRE(xfer != NULL);
        xfer->nxr_state = NXS_COMPLETE;
        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        M0_PRE(m0_sm_group_is_locked(ioo->ioo_sm.sm_grp));

        instance = m0__op_instance(m0__ioo_to_op(ioo));
        /*
         * Ignore the following invariant check as there exists cases in which
         * io fops are created sucessfully for some target services but fail
         * for some services in nxo_dispatch (for example, session/connection
         * to a service is invalid, resulting a 'dirty' op in which
         * nr_iofops != 0 and nxr_state == NXS_COMPLETE.
         *
         * M0_PRE_EX(m0_op_io_invariant(ioo));
         */

        m0_htable_for(tioreqht, ti, &xfer->nxr_tioreqs_hash) {

                /* Maintains only the first error encountered. */
                if (xfer->nxr_rc == 0)
                        xfer->nxr_rc = ti->ti_rc;

                xfer->nxr_bytes += ti->ti_databytes;
                ti->ti_databytes = 0;

                if (m0__is_oostore(instance) &&
                    ti->ti_req_type == TI_COB_CREATE &&
                    ioreq_sm_state(ioo) == IRS_WRITE_COMPLETE) {
                        ti->ti_req_type = TI_NONE;
                        continue;
                }

                if (m0__is_oostore(instance) &&
                    ti->ti_req_type == TI_COB_TRUNCATE &&
                    ioreq_sm_state(ioo) == IRS_TRUNCATE_COMPLETE) {
                        ti->ti_req_type = TI_NONE;
                }

                m0_tl_teardown(iofops, &ti->ti_iofops, irfop) {
                        fop = &irfop->irf_iofop.if_fop;
                        item = m0_fop_to_rpc_item(fop);
                        M0_LOG(M0_DEBUG, "[%p] fop %p, ref %llu, "
                               "item %p[%u], ri_error %d, ri_state %d",
                               ioo, fop,
                               (unsigned long long)m0_ref_read(&fop->f_ref),
                               item, item->ri_type->rit_opcode, item->ri_error,
                               item->ri_sm.sm_state);

                        M0_ASSERT(ergo(item->ri_sm.sm_state !=
                                       M0_RPC_ITEM_UNINITIALISED,
                                       item->ri_rmachine != NULL));
                        if (item->ri_rmachine == NULL) {
                                M0_ASSERT(ti->ti_session != NULL);
                                m0_fop_rpc_machine_set(fop,
                                        ti->ti_session->s_conn->c_rpc_machine);
                        }

                        M0_LOG(M0_DEBUG,
                               "[%p] item %p, target fid "FID_F"fop %p, "
                               "ref %llu", ioo, item, FID_P(&ti->ti_fid), fop,
                               (unsigned long long)m0_ref_read(&fop->f_ref));
                        m0_fop_put_lock(fop);
                }

        } m0_htable_endfor;

        M0_LOG(M0_INFO, "Number of bytes %s = %"PRIu64,
               ioreq_sm_state(ioo) == IRS_READ_COMPLETE ? "read" : "written",
               xfer->nxr_bytes);

        /*
         * This function is invoked from 4 states - IRS_READ_COMPLETE,
         * IRS_WRITE_COMPLETE, IRS_DEGRADED_READING, IRS_DEGRADED_WRITING.
         * And the state change is applicable only for healthy state IO,
         * meaning for states IRS_READ_COMPLETE and IRS_WRITE_COMPLETE.
         */
        if (M0_IN(ioreq_sm_state(ioo),
                  (IRS_READ_COMPLETE, IRS_WRITE_COMPLETE,
                   IRS_TRUNCATE_COMPLETE))) {
                if (!rmw)
                        ioreq_sm_state_set_locked(ioo, IRS_REQ_COMPLETE);
                else if (ioreq_sm_state(ioo) == IRS_READ_COMPLETE)
                        xfer->nxr_bytes = 0;
        }

        /*
         * nxo_dispatch may fail if connections to services have not been
         * established yet. In this case, ioo_rc contains error code and
         * xfer->nxr_rc == 0, don't overwrite ioo_rc.
         *
         * TODO: merge this with op->op_sm.sm_rc ?
         */
        if (xfer->nxr_rc != 0)
                ioo->ioo_rc = xfer->nxr_rc;

        M0_LEAVE();
}

static int nw_xfer_req_dispatch(struct nw_xfer_request *xfer)
{
        int                     rc = 0;
        int                     post_error = 0;
        int                     ri_error;
        uint64_t                nr_dispatched = 0;
        struct ioreq_fop       *irfop;
        struct m0_op_io        *ioo;
        struct m0_op           *op;
        struct target_ioreq    *ti;
        struct m0_client       *instance;

        M0_ENTRY();

        M0_PRE(xfer != NULL);
        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        op = &ioo->ioo_oo.oo_oc.oc_op;
        instance = m0__op_instance(op);
        M0_PRE(instance != NULL);

        to_op_io_map(op, ioo);

        /* FOPs' preparation */
        m0_htable_for(tioreqht, ti, &xfer->nxr_tioreqs_hash) {
                if (ti->ti_state != M0_PNDS_ONLINE) {
                        M0_LOG(M0_INFO, "Skipped iofops prepare for "FID_F,
                               FID_P(&ti->ti_fid));
                        continue;
                }
                ti->ti_start_time = m0_time_now();
                if (ti->ti_req_type == TI_COB_CREATE &&
                    ioreq_sm_state(ioo) == IRS_WRITING) {
                        rc = ti->ti_ops->tio_cc_fops_prepare(ti);
                        if (rc != 0)
                                return M0_ERR_INFO(rc, "[%p] cob create fop"
                                                   "failed", ioo);
                        continue;
                }

                if (ioreq_sm_state(ioo) == IRS_TRUNCATE) {
                    if (ti->ti_req_type == TI_READ_WRITE) {
                        ti->ti_req_type = TI_COB_TRUNCATE;
                        rc = ti->ti_ops->tio_cc_fops_prepare(ti);
                        if (rc != 0)
                                return M0_ERR(rc);
                        }
                        continue;
                }
                rc = ti->ti_ops->tio_iofops_prepare(ti, PA_DATA) ?:
                     ti->ti_ops->tio_iofops_prepare(ti, PA_PARITY);
                if (rc != 0)
                        return M0_ERR(rc);
        } m0_htable_endfor;

        /* Submit io FOPs */
        m0_htable_for(tioreqht, ti, &xfer->nxr_tioreqs_hash) {
                struct m0_rpc_item *item = &ti->ti_cc_fop.crf_fop.f_item;

                /* Skips the target device if it is not online. */
                if (ti->ti_state != M0_PNDS_ONLINE) {
                        M0_LOG(M0_INFO, "Skipped device "FID_F,
                               FID_P(&ti->ti_fid));
                        continue;
                }
                if (ti->ti_req_type == TI_COB_CREATE &&
                    ioreq_sm_state(ioo) == IRS_WRITING) {
                        /*
                         * An error returned by rpc post has been ignored.
                         * It will be handled in the respective bottom half.
                         */
                        M0_LOG(M0_DEBUG, "item="ITEM_FMT" osr_xid=%"PRIu64,
                                ITEM_ARG(item), item->ri_header.osr_xid);
                        rc = m0_rpc_post(item);
                        M0_CNT_INC(nr_dispatched);
                        m0_op_io_to_rpc_map(ioo, item);
                        continue;
                }
                if (op->op_code == M0_OC_FREE &&
                    ioreq_sm_state(ioo) == IRS_TRUNCATE &&
                    ti->ti_req_type == TI_COB_TRUNCATE) {
                        if (ti->ti_trunc_ivec.iv_vec.v_nr > 0) {
                                /*
                                 * An error returned by rpc post has been
                                 * ignored. It will be handled in the
                                 * io_bottom_half().
                                 */
                                M0_LOG(M0_DEBUG, "item="ITEM_FMT
                                                 " osr_xid=%"PRIu64,
                                                 ITEM_ARG(item),
                                                 item->ri_header.osr_xid);
                                rc = m0_rpc_post(item);
                                M0_CNT_INC(nr_dispatched);
                                m0_op_io_to_rpc_map(ioo, item);
                        }
                        continue;
                }
                m0_tl_for (iofops, &ti->ti_iofops, irfop) {
                        rc = ioreq_fop_async_submit(&irfop->irf_iofop,
                                                    ti->ti_session);
                        ri_error = irfop->irf_iofop.if_fop.f_item.ri_error;
                        M0_LOG(M0_DEBUG, "[%p] Submitted fop for device "
                               FID_F"@%p, item %p, fop_nr=%llu, rc=%d, "
                               "ri_error=%d", ioo, FID_P(&ti->ti_fid), irfop,
                               &irfop->irf_iofop.if_fop.f_item,
                               (unsigned long long)
                               m0_atomic64_get(&xfer->nxr_iofop_nr),
                               rc, ri_error);

                        /* XXX: noisy */
                        m0_op_io_to_rpc_map(ioo,
                                        &irfop->irf_iofop.if_fop.f_item);

                        if (rc != 0)
                                goto out;
                        m0_atomic64_inc(&instance->m0c_pending_io_nr);
                        if (ri_error == 0)
                                M0_CNT_INC(nr_dispatched);
                        else if (post_error == 0)
                                post_error = ri_error;
                } m0_tl_endfor;
        } m0_htable_endfor;

out:
        if (rc == 0 && nr_dispatched == 0 && post_error == 0) {
                /* No fop has been dispatched.
                 *
                 * This might happen in dgmode reading:
                 *    In 'parity verify' mode, a whole parity group, including
                 *    data and parity units are all read from ioservices.
                 *    If some units failed to read, no need to read extra unit.
                 *    The units needed for recvoery are ready.
                 */
                M0_ASSERT(ioreq_sm_state(ioo) == IRS_DEGRADED_READING);
                M0_ASSERT(op->op_code == M0_OC_READ &&
                          instance->m0c_config->mc_is_read_verify);
                ioreq_sm_state_set_locked(ioo, IRS_READ_COMPLETE);
        } else if (rc == 0)
                xfer->nxr_state = NXS_INFLIGHT;

        M0_LOG(M0_DEBUG, "[%p] nxr_iofop_nr %llu, nxr_rdbulk_nr %llu, "
               "nr_dispatched %llu", ioo,
               (unsigned long long)m0_atomic64_get(&xfer->nxr_iofop_nr),
               (unsigned long long)m0_atomic64_get(&xfer->nxr_rdbulk_nr),
               (unsigned long long)nr_dispatched);

        return M0_RC(rc);
}

static bool should_spare_be_mapped(struct m0_op_io *ioo,
                                   enum m0_pool_nd_state dev_state)
{
        return (M0_IN(ioreq_sm_state(ioo),
                      (IRS_READING, IRS_DEGRADED_READING)) &&
                dev_state == M0_PNDS_SNS_REPAIRED)
                                  ||
               (ioreq_sm_state(ioo) == IRS_DEGRADED_WRITING &&
                (dev_state == M0_PNDS_SNS_REPAIRED ||
                 (dev_state == M0_PNDS_SNS_REPAIRING &&
                  ioo->ioo_sns_state == SRS_REPAIR_DONE)));

}

static int nw_xfer_tioreq_map(struct nw_xfer_request           *xfer,
                              const struct m0_pdclust_src_addr *src,
                              struct m0_pdclust_tgt_addr       *tgt,
                              struct target_ioreq             **tio)
{
        int                         rc;
        struct m0_fid               tfid;
        const struct m0_fid        *gfid;
        struct m0_op_io            *ioo;
        struct m0_rpc_session      *session;
        struct m0_pdclust_layout   *play;
        struct m0_pdclust_instance *play_instance;
        enum m0_pool_nd_state       dev_state;
        enum m0_pool_nd_state       dev_state_prev;
        uint32_t                    spare_slot;
        uint32_t                    spare_slot_prev;
        struct m0_pdclust_src_addr  spare;
        struct m0_poolmach         *pm;

        M0_ENTRY("nw_xfer_request=%p", xfer);

        M0_PRE(nw_xfer_request_invariant(xfer));
        M0_PRE(src != NULL);
        M0_PRE(tgt != NULL);
        M0_PRE(tio != NULL);

        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);

        play = pdlayout_get(ioo);
        M0_PRE(play != NULL);
        play_instance = pdlayout_instance(layout_instance(ioo));
        M0_PRE(play_instance != NULL);

        spare = *src;
        m0_fd_fwd_map(play_instance, src, tgt);
        tfid = target_fid(ioo, tgt);
        M0_LOG(M0_DEBUG, "src_id[%" PRIu64 ":%" PRIu64 "] -> "
                         "dest_id[%" PRIu64 ":%" PRIu64 "] @ tfid="FID_F,
               src->sa_group, src->sa_unit, tgt->ta_frame, tgt->ta_obj,
               FID_P(&tfid));

        pm = ioo_to_poolmach(ioo);
        M0_ASSERT(pm != NULL);
        rc = m0_poolmach_device_state(pm, tgt->ta_obj, &dev_state);
        if (rc != 0)
                return M0_RC(rc);

        if (M0_FI_ENABLED("poolmach_client_repaired_device1")) {
                if (tfid.f_container == 1)
                        dev_state = M0_PNDS_SNS_REPAIRED;
        }

        M0_LOG(M0_INFO, "[%p] tfid="FID_F" dev_state=%d\n",
                         ioo, FID_P(&tfid), dev_state);

        if (should_spare_be_mapped(ioo, dev_state)) {
                gfid = &ioo->ioo_oo.oo_fid;
                rc = m0_sns_repair_spare_map(pm, gfid, play, play_instance,
                                             src->sa_group, src->sa_unit,
                                             &spare_slot, &spare_slot_prev);
                if (rc != 0)
                        return M0_RC(rc);

                /* Check if there is an effective-failure. */
                if (spare_slot_prev != src->sa_unit) {
                        spare.sa_unit = spare_slot_prev;
                        m0_fd_fwd_map(play_instance, &spare, tgt);
                        tfid = target_fid(ioo, tgt);
                        rc = m0_poolmach_device_state(pm, tgt->ta_obj,
                                                      &dev_state_prev);
                        if (rc != 0)
                                return M0_RC(rc);
                } else
                        dev_state_prev = M0_PNDS_SNS_REPAIRED;

                if (dev_state_prev == M0_PNDS_SNS_REPAIRED) {
                        spare.sa_unit = spare_slot;
                        m0_fd_fwd_map(play_instance, &spare, tgt);
                        tfid = target_fid(ioo, tgt);
                }
                dev_state = dev_state_prev;
        }

        session = target_session(ioo, tfid);

        rc = nw_xfer_tioreq_get(xfer, &tfid, tgt->ta_obj, session,
                        layout_unit_size(play) * ioo->ioo_iomap_nr, tio);

        if (M0_IN(ioreq_sm_state(ioo), (IRS_DEGRADED_READING,
                                        IRS_DEGRADED_WRITING)) &&
            dev_state != M0_PNDS_SNS_REPAIRED)
                (*tio)->ti_state = dev_state;

        return M0_RC(rc);
}

static const struct nw_xfer_ops xfer_ops = {
        .nxo_distribute = nw_xfer_io_distribute,
        .nxo_complete   = nw_xfer_req_complete,
        .nxo_dispatch   = nw_xfer_req_dispatch,
        .nxo_tioreq_map = nw_xfer_tioreq_map,
};

M0_INTERNAL void nw_xfer_request_init(struct nw_xfer_request *xfer)
{
        uint64_t                  bucket_nr;
        struct m0_op_io          *ioo;
        struct m0_pdclust_layout *play;

        M0_ENTRY("nw_xfer_request : %p", xfer);

        M0_PRE(xfer != NULL);

        ioo = bob_of(xfer, struct m0_op_io, ioo_nwxfer, &ioo_bobtype);
        nw_xfer_request_bob_init(xfer);
        xfer->nxr_rc        = 0;
        xfer->nxr_bytes     = 0;
        m0_atomic64_set(&xfer->nxr_iofop_nr, 0);
        m0_atomic64_set(&xfer->nxr_rdbulk_nr, 0);
        xfer->nxr_state     = NXS_INITIALIZED;
        xfer->nxr_ops       = &xfer_ops;
        m0_mutex_init(&xfer->nxr_lock);

        play = pdlayout_get(ioo);
        bucket_nr = layout_n(play) + 2 * layout_k(play);
        xfer->nxr_rc = tioreqht_htable_init(&xfer->nxr_tioreqs_hash,
                                            bucket_nr);

        M0_POST_EX(nw_xfer_request_invariant(xfer));
        M0_LEAVE();
}

M0_INTERNAL void nw_xfer_request_fini(struct nw_xfer_request *xfer)
{
        M0_ENTRY("nw_xfer_request : %p", xfer);

        M0_PRE(xfer != NULL);
        M0_PRE(M0_IN(xfer->nxr_state, (NXS_COMPLETE, NXS_INITIALIZED)));
        M0_PRE(nw_xfer_request_invariant(xfer));
        M0_LOG(M0_DEBUG, "nw_xfer_request : %p, nxr_rc = %d",
               xfer, xfer->nxr_rc);

        xfer->nxr_ops = NULL;
        m0_mutex_fini(&xfer->nxr_lock);
        nw_xfer_request_bob_fini(xfer);
        tioreqht_htable_fini(&xfer->nxr_tioreqs_hash);

        M0_LEAVE();
}

#undef M0_TRACE_SUBSYSTEM

/*
 *  Local variables:
 *  c-indentation-style: "K&R"

 *  c-basic-offset: 8
 *  tab-width: 8
 *  fill-column: 80
 *  scroll-step: 1
 *  End:
 */
/*
 * vim: tabstop=8 shiftwidth=8 noexpandtab textwidth=80 nowrap
 */