cp.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.
 *
 */


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

#include "lib/misc.h"   /* m0_forall */
#include "lib/memory.h"
#include "lib/errno.h"

#include "motr/magic.h"
#include "reqh/reqh.h"
#include "net/buffer_pool.h"
#include "rpc/rpc_opcodes.h" /* M0_CM_CP_OPCODE */
#include "rpc/bulk.h"

#include "cm/cp.h"
#include "cm/ag.h"
#include "cm/proxy.h"
#include "cm/cm.h"

#include "fop/fop.h"
#include "fop/fom.h"

M0_TL_DESCR_DEFINE(cp_data_buf, "copy packet data buffers", M0_INTERNAL,
                   struct m0_net_buffer, nb_extern_linkage, nb_magic,
                   M0_NET_BUFFER_LINK_MAGIC, CM_CP_DATA_BUF_HEAD_MAGIX);
M0_TL_DEFINE(cp_data_buf, M0_INTERNAL, struct m0_net_buffer);

static const struct m0_bob_type cp_bob = {
        .bt_name = "copy packet",
        .bt_magix_offset = M0_MAGIX_OFFSET(struct m0_cm_cp, c_magix),
        .bt_magix = CM_CP_MAGIX,
        .bt_check = NULL
};

M0_TL_DESCR_DEFINE(proxy_cps, "copy packets in proxy", M0_INTERNAL,
                   struct m0_cm_cp, c_cm_proxy_linkage, c_magix,
                   CM_CP_MAGIX, CM_PROXY_CP_HEAD_MAGIX);
M0_TL_DEFINE(proxy_cps, M0_INTERNAL, struct m0_cm_cp);


M0_BOB_DEFINE(static, &cp_bob, m0_cm_cp);

M0_INTERNAL void m0_cm_cp_fom_fini(struct m0_fom *fom)
{
        struct m0_cm_cp *cp = bob_of(fom, struct m0_cm_cp, c_fom,
                                     &cp_bob);
        M0_ENTRY();

        m0_cm_cp_fini(cp);
        cp->c_ops->co_free(cp);

        M0_LEAVE();
}

static uint64_t cp_fom_locality(const struct m0_fom *fom)
{
        struct m0_cm_cp *cp = bob_of(fom, struct m0_cm_cp, c_fom, &cp_bob);

        return cp->c_ops->co_home_loc_helper(cp);
}

static int cp_fom_tick(struct m0_fom *fom)
{
        struct m0_cm_cp *cp = bob_of(fom, struct m0_cm_cp, c_fom, &cp_bob);
        int              phase = m0_fom_phase(fom);
        int              rc;

        M0_PRE(phase < cp->c_ops->co_action_nr);
        M0_LOG(M0_DEBUG, "fom phase = %d", phase);

        rc = cp->c_ops->co_action[phase](cp);
        if (m0_fom_phase(fom) == M0_CCP_FINI)
                cp->c_ops->co_action[M0_CCP_FINI](cp);
        return M0_RC(rc);
}

static const struct m0_fom_ops cp_fom_ops = {
        .fo_fini          = m0_cm_cp_fom_fini,
        .fo_tick          = cp_fom_tick,
        .fo_home_locality = cp_fom_locality
};

M0_INTERNAL int m0_cm_cp_fom_create(struct m0_fop *fop, struct m0_fop *r_fop,
                                    struct m0_fom **m, struct m0_reqh *reqh)
{
        struct m0_cm_cp        *cp;
        struct m0_cm           *cm;
        struct m0_reqh_service *service;

        M0_PRE(fop != NULL);
        M0_PRE(m != NULL);

        service = m0_reqh_service_find(fop->f_type->ft_fom_type.ft_rstype,
                                       reqh);
        M0_PRE(service != NULL);
        cm = container_of(service, struct m0_cm, cm_service);
        M0_PRE(cm != NULL);
        cp = cm->cm_ops->cmo_cp_alloc(cm);
        if (cp == NULL)
                return M0_ERR(-ENOMEM);

        m0_cm_cp_fom_init(cm, cp, fop, r_fop);
        *m = &cp->c_fom;
        return 0;
}

static struct m0_sm_state_descr m0_cm_cp_state_descr[] = {
        [M0_CCP_INIT] = {
                .sd_flags       = M0_SDF_INITIAL,
                .sd_name        = "Init",
                .sd_allowed     = M0_BITS(M0_CCP_READ, M0_CCP_WRITE,
                                          M0_CCP_XFORM, M0_CCP_SEND,
                                          M0_CCP_SW_CHECK, M0_CCP_FAIL,
                                          M0_CCP_FINI)
        },
        [M0_CCP_READ] = {
                .sd_flags       = 0,
                .sd_name        = "Read",
                .sd_allowed     = M0_BITS(M0_CCP_IO_WAIT, M0_CCP_FAIL,
                                          M0_CCP_FINI)
        },
        [M0_CCP_WRITE_PRE] = {
                .sd_flags       = 0,
                .sd_name        = "Write-pre",
                .sd_allowed     = M0_BITS(M0_CCP_TX_OPEN, M0_CCP_WRITE,
                                          M0_CCP_FAIL)
        },
        [M0_CCP_TX_OPEN] = {
                .sd_flags       = 0,
                .sd_name        = "TX Open",
                .sd_allowed     = M0_BITS(M0_CCP_WRITE, M0_CCP_FAIL)
        },
        [M0_CCP_WRITE] = {
                .sd_flags       = 0,
                .sd_name        = "Write",
                .sd_allowed     = M0_BITS(M0_CCP_TX_DONE, M0_CCP_IO_WAIT,
                                          M0_CCP_FAIL, M0_CCP_FINI)
        },
        [M0_CCP_TX_DONE] = {
                .sd_flags       = 0,
                .sd_name        = "TX Done",
                .sd_allowed     = M0_BITS(M0_CCP_IO_WAIT, M0_CCP_FAIL)
        },
        [M0_CCP_IO_WAIT] = {
                .sd_flags       = 0,
                .sd_name        = "IO Wait",
                .sd_allowed     = M0_BITS(M0_CCP_XFORM, M0_CCP_SEND,
                                          M0_CCP_FAIL, M0_CCP_FINI)
        },
        [M0_CCP_XFORM] = {
                .sd_flags       = 0,
                .sd_name        = "Xform",
                .sd_allowed     = M0_BITS(M0_CCP_FAIL, M0_CCP_FINI,
                                          M0_CCP_WRITE_PRE, M0_CCP_WRITE,
                                          M0_CCP_SEND, M0_CCP_SW_CHECK)
        },
        [M0_CCP_SW_CHECK] = {
                .sd_flags       = 0,
                .sd_name        = "Sliding window check",
                .sd_allowed     = M0_BITS(M0_CCP_SEND, M0_CCP_FINI, M0_CCP_FAIL)
        },
        [M0_CCP_SEND] = {
                .sd_flags       = 0,
                .sd_name        = "Send",
                .sd_allowed     = M0_BITS(M0_CCP_FINI, M0_CCP_RECV_INIT,
                                          M0_CCP_SEND_WAIT, M0_CCP_FAIL)
        },
        [M0_CCP_SEND_WAIT] = {
                .sd_flags       = 0,
                .sd_name        = "Send Wait",
                .sd_allowed     = M0_BITS(M0_CCP_FINI, M0_CCP_FAIL)
        },
        [M0_CCP_RECV_INIT] = {
                .sd_flags       = 0,
                .sd_name        = "Recv Init",
                .sd_allowed     = M0_BITS(M0_CCP_RECV_WAIT, M0_CCP_FINI)
        },
        [M0_CCP_RECV_WAIT] = {
                .sd_flags       = 0,
                .sd_name        = "Recv Wait",
                .sd_allowed     = M0_BITS(M0_CCP_XFORM, M0_CCP_RECV_INIT,
                                          M0_CCP_FAIL)
        },
        [M0_CCP_FAIL] = {
                .sd_flags       = M0_SDF_FAILURE,
                .sd_name        = "Failure",
                .sd_allowed     = M0_BITS(M0_CCP_FINI)
        },
        [M0_CCP_FINI] = {
                .sd_flags       = M0_SDF_TERMINAL,
                .sd_name        = "Fini",
                .sd_allowed     = 0
        },
};

static const struct m0_sm_conf m0_cm_cp_sm_conf = {
        .scf_name = "sm:cp conf",
        .scf_nr_states = ARRAY_SIZE(m0_cm_cp_state_descr),
        .scf_state = m0_cm_cp_state_descr
};

M0_INTERNAL void m0_cm_cp_init(struct m0_cm_type *cmtype,
                               const struct m0_fom_type_ops *ft_ops)
{
        m0_fom_type_init(&cmtype->ct_fomt, cmtype->ct_fom_id, ft_ops,
                         &cmtype->ct_stype, &m0_cm_cp_sm_conf);
}

M0_INTERNAL bool m0_cm_cp_invariant(const struct m0_cm_cp *cp)
{
        const struct m0_cm_cp_ops *ops = cp->c_ops;

        return m0_cm_cp_bob_check(cp) && ops != NULL &&
               cp->c_ag != NULL &&
               m0_fom_phase(&cp->c_fom) < ops->co_action_nr &&
               cp->c_ops->co_invariant(cp);
}

M0_INTERNAL void m0_cm_cp_only_init(struct m0_cm *cm, struct m0_cm_cp *cp)
{
        m0_cm_cp_bob_init(cp);
        cp_data_buf_tlist_init(&cp->c_buffers);
        m0_rpc_bulk_init(&cp->c_bulk);
        m0_mutex_init(&cp->c_reply_wait_mutex);
        m0_chan_init(&cp->c_reply_wait, &cp->c_reply_wait_mutex);
        proxy_cp_tlink_init(cp);

        /* copy packet epoch is derived from its cm */
        cp->c_epoch = cm->cm_epoch;
}

M0_INTERNAL void m0_cm_cp_fom_init(struct m0_cm *cm, struct m0_cm_cp *cp,
                                   struct m0_fop *fop, struct m0_fop *r_fop)
{
        struct m0_reqh_service *service;

        M0_PRE(cm != NULL && cp != NULL && cp->c_ops != NULL);

        m0_cm_cp_only_init(cm, cp);
        service = &cm->cm_service;
        m0_fom_init(&cp->c_fom, &cm->cm_type->ct_fomt, &cp_fom_ops, fop, r_fop,
                    service->rs_reqh);
}

M0_TL_DECLARE(rpcbulk, M0_INTERNAL, struct m0_rpc_bulk_buf);
M0_TL_DESCR_DECLARE(rpcbulk, M0_EXTERN);

M0_INTERNAL void m0_cm_cp_only_fini(struct m0_cm_cp *cp)
{
        struct m0_rpc_bulk *rbulk;
        /*
         * If the copy packet is not an accumulator copy packet, it will get
         * finalised after transformation. For such copy packets, finalise the
         * bitmap.
         */
        if (cp->c_xform_cp_indices.b_nr > 0)
                m0_bitmap_fini(&cp->c_xform_cp_indices);
        m0_chan_fini_lock(&cp->c_reply_wait);
        m0_mutex_fini(&cp->c_reply_wait_mutex);

        /*
         * Release the net buffers if rpc bulk object is still dirty.
         * And wait on channel till all net buffers are deleted from
         * transfer machine.
         */
        rbulk = &cp->c_bulk;
        m0_mutex_lock(&rbulk->rb_mutex);
        if (!rpcbulk_tlist_is_empty(&rbulk->rb_buflist)) {
                struct m0_clink clink;
                size_t          buf_nr;
                size_t          non_queued_buf_nr;

                m0_clink_init(&clink, NULL);
                m0_clink_add(&rbulk->rb_chan, &clink);
                buf_nr = rpcbulk_tlist_length(&rbulk->rb_buflist);
                non_queued_buf_nr = m0_rpc_bulk_store_del_unqueued(rbulk);
                m0_mutex_unlock(&rbulk->rb_mutex);

                m0_rpc_bulk_store_del(rbulk);
                M0_LOG(M0_DEBUG, "bulk %p, buf_nr %llu, non_queued_buf_nr %llu",
                                           rbulk,
                       (unsigned long long)buf_nr,
                       (unsigned long long)non_queued_buf_nr);
                /*
                 * If there were some queued net bufs which had to be deleted,
                 * then it is required to wait for their callbacks.
                 */
                if (buf_nr > non_queued_buf_nr) {
                        m0_chan_wait(&clink);
                }
                m0_clink_del_lock(&clink);
                m0_clink_fini(&clink);
        } else {
                m0_mutex_unlock(&rbulk->rb_mutex);
        }
        M0_ASSERT(m0_rpc_bulk_is_empty(rbulk));

        m0_rpc_bulk_buflist_empty(&cp->c_bulk);
        m0_rpc_bulk_fini(&cp->c_bulk);
        proxy_cp_tlink_fini(cp);
        m0_cm_cp_bob_fini(cp);
}

M0_INTERNAL void m0_cm_cp_fini(struct m0_cm_cp *cp)
{
        m0_cm_cp_only_fini(cp);
        m0_fom_fini(&cp->c_fom);
}

M0_INTERNAL int m0_cm_cp_enqueue(struct m0_cm *cm, struct m0_cm_cp *cp)
{
        struct m0_fom  *fom = &cp->c_fom;
        struct m0_reqh *reqh = cm->cm_service.rs_reqh;

        M0_PRE(reqh != NULL);
        M0_PRE(m0_cm_cp_invariant(cp));

        if (m0_reqh_state_get(reqh) >= M0_REQH_ST_DRAIN)
                return M0_ERR(-ESHUTDOWN);

        m0_fom_queue(fom);
        return M0_RC(0);
}

M0_INTERNAL void m0_cm_cp_buf_add(struct m0_cm_cp *cp, struct m0_net_buffer *nb)
{
        M0_PRE(cp != NULL);
        M0_PRE(nb != NULL);

        cp_data_buf_tlink_init(nb);
        cp_data_buf_tlist_add_tail(&cp->c_buffers, nb);
        M0_CNT_INC(cp->c_buf_nr);
}

M0_INTERNAL void m0_cm_cp_buf_release(struct m0_cm_cp *cp)
{
        struct m0_net_buffer_pool *nbp;
        struct m0_net_buffer      *nbuf;
        struct m0_net_buffer      *nbuf_head;
        uint64_t                   colour;

        nbuf_head = cp_data_buf_tlist_head(&cp->c_buffers);
        if (nbuf_head != NULL) {
                nbp = nbuf_head->nb_pool;
                M0_ASSERT(nbp != NULL);
                m0_net_buffer_pool_lock(nbp);
                m0_tl_for(cp_data_buf, &cp->c_buffers, nbuf) {
                        colour = cp->c_ops->co_home_loc_helper(cp) %
                                                nbp->nbp_colours_nr;
                        cp_data_buf_tlink_del_fini(nbuf);
                        m0_cm_buffer_put(nbp, nbuf, colour);
                        M0_CNT_DEC(cp->c_buf_nr);
                } m0_tl_endfor;
                nbp->nbp_ops->nbpo_not_empty(nbp);
                m0_net_buffer_pool_unlock(nbp);
        }
        cp_data_buf_tlist_fini(&cp->c_buffers);
}

M0_INTERNAL uint64_t m0_cm_cp_nr(struct m0_cm_cp *cp)
{
        struct m0_bitmap *bm = &cp->c_xform_cp_indices;
        int               i;
        uint64_t          cnt = 0;

        for (i = 0; i < bm->b_nr; ++i) {
                if (m0_bitmap_get(bm, i))
                        M0_CNT_INC(cnt);
        }

        return cnt;
}

M0_INTERNAL int m0_cm_cp_bufvec_merge(struct m0_cm_cp *cp)
{
        struct m0_net_buffer    *nbuf;
        struct m0_net_buffer    *nbuf_head;
        int                      rc;

        nbuf_head = cp_data_buf_tlist_head(&cp->c_buffers);
        nbuf = nbuf_head;
        while ((nbuf = cp_data_buf_tlist_next(&cp->c_buffers, nbuf)) != NULL) {
                rc = m0_bufvec_merge(&nbuf_head->nb_buffer, &nbuf->nb_buffer);
                if (rc != 0)
                        return M0_RC(rc);
        }
        return 0;
}

M0_INTERNAL int m0_cm_cp_bufvec_split(struct m0_cm_cp *cp)
{
        struct m0_net_buffer       *nbuf_head;
        struct m0_bufvec           *bufvec;
        uint32_t                    new_v_nr;
        m0_bcount_t                *new_v_count;
        uint32_t                    i;

        nbuf_head = cp_data_buf_tlist_head(&cp->c_buffers);
        new_v_nr = nbuf_head->nb_pool->nbp_seg_nr;
        M0_ALLOC_ARR(new_v_count, new_v_nr);
        if (new_v_count == NULL)
                return M0_ERR(-ENOMEM);

        bufvec = &nbuf_head->nb_buffer;
        for (i = 0; i < new_v_nr; ++i)
                new_v_count[i] = bufvec->ov_vec.v_count[i];

        m0_free(bufvec->ov_vec.v_count);
        bufvec->ov_vec.v_nr = new_v_nr;
        bufvec->ov_vec.v_count = new_v_count;

        return 0;
}

M0_INTERNAL void m0_cm_cp_buf_move(struct m0_cm_cp *src, struct m0_cm_cp *dest)
{
        struct m0_net_buffer *nbuf;
        M0_PRE(src->c_data_seg_nr == dest->c_data_seg_nr);

        m0_tl_for(cp_data_buf, &src->c_buffers, nbuf) {
                cp_data_buf_tlink_del_fini(nbuf);
                M0_CNT_DEC(src->c_buf_nr);
                m0_cm_cp_buf_add(dest, nbuf);
        } m0_tl_endfor;
}

M0_INTERNAL int m0_cm_cp_dup(struct m0_cm_cp *src, struct m0_cm_cp **dest)
{
        struct m0_cm         *cm;
        struct m0_cm_cp      *cp;

        cm = src->c_ag->cag_cm;
        cp = cm->cm_ops->cmo_cp_alloc(cm);
        if (cp == NULL)
                return -ENOMEM;
        m0_cm_ag_cp_add_locked(src->c_ag, cp);
        cp->c_ag_cp_idx = src->c_ag_cp_idx;
        cp->c_data_seg_nr = src->c_data_seg_nr;
        m0_cm_cp_fom_init(cm, cp, NULL, NULL);
        m0_bitmap_init(&cp->c_xform_cp_indices,
                       src->c_xform_cp_indices.b_nr);
        m0_cm_cp_buf_move(src, cp);
        if (src->c_xform_cp_indices.b_nr > 0)
                m0_bitmap_copy(&cp->c_xform_cp_indices, &src->c_xform_cp_indices);
        *dest = cp;

        return M0_RC(0);
}

M0_INTERNAL void m0_cm_cp_data_copy(struct m0_cm_cp *src, struct m0_cm_cp *dst)
{
        struct m0_net_buffer      *src_nbuf;
        struct m0_net_buffer      *dst_nbuf;
        struct m0_net_buffer_pool *nbp;
        m0_bcount_t                bytes_copied = 0;
        uint64_t                   buf_size = 0;
        uint64_t                   total_data_seg_nr;

        M0_PRE(!cp_data_buf_tlist_is_empty(&src->c_buffers));
        M0_PRE(!cp_data_buf_tlist_is_empty(&dst->c_buffers));
        M0_PRE(src->c_buf_nr == dst->c_buf_nr);

        total_data_seg_nr = src->c_data_seg_nr;
        for (src_nbuf = cp_data_buf_tlist_head(&src->c_buffers),
             dst_nbuf = cp_data_buf_tlist_head(&dst->c_buffers);
             src_nbuf != NULL && dst_nbuf != NULL;
             src_nbuf = cp_data_buf_tlist_next(&src->c_buffers, src_nbuf),
             dst_nbuf = cp_data_buf_tlist_next(&dst->c_buffers, dst_nbuf)) {
                nbp = src_nbuf->nb_pool;
                if (total_data_seg_nr < nbp->nbp_seg_nr)
                        buf_size = total_data_seg_nr * nbp->nbp_seg_size;
                else {
                        total_data_seg_nr -= nbp->nbp_seg_nr;
                        buf_size = nbp->nbp_seg_nr * nbp->nbp_seg_size;
                }
                bytes_copied = m0_bufvec_copy(&dst_nbuf->nb_buffer,
                                              &src_nbuf->nb_buffer, buf_size);
                M0_ASSERT(bytes_copied == buf_size);
        }
}

/*
 *  Local variables:
 *  c-indentation-style: "K&R"
 *  c-basic-offset: 8
 *  tab-width: 8
 *  fill-column: 80
 *  scroll-step: 1
 *  End:
 */