cm.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2012-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_SNSCM
#include "lib/trace.h"

#include "lib/memory.h"
#include "lib/assert.h"
#include "lib/errno.h"
#include "lib/misc.h"
#include "lib/finject.h"
#include "lib/chan.h"
#include "lib/locality.h"
#include "lib/types.h"

#include "fop/fop.h"
#include "pool/pool_machine.h"

#include "motr/setup.h"
#include "net/net.h"
#include "ioservice/io_service.h"
#include "reqh/reqh.h"
#include "rpc/rpc.h"
#include "cob/ns_iter.h"
#include "cm/proxy.h"
#include "sns/cm/ha.h"
#include "ha/msg.h"

#include "sns/cm/cm_utils.h"
#include "sns/cm/iter.h"
#include "sns/cm/cm.h"
#include "sns/cm/cp.h"
#include "sns/cm/ag.h"
#include "sns/cm/file.h"
#include "lib/locality.h"
#include "rm/rm_service.h"

#include "conf/diter.h"

enum {
        SNS_SEG_NR = 10,
        SNS_SEG_SIZE = 4096,

        SNS_COB_FID_START = 4,
};

M0_EXTERN struct m0_cm_type sns_repair_cmt;
M0_EXTERN struct m0_cm_type sns_rebalance_cmt;

extern const struct m0_sns_cm_helpers repair_helpers;
extern const struct m0_sns_cm_helpers rebalance_helpers;

M0_INTERNAL struct m0_sns_cm *cm2sns(struct m0_cm *cm)
{
        return container_of(cm, struct m0_sns_cm, sc_base);
}

M0_INTERNAL int m0_sns_cm_type_register(void)
{
        int rc;

        rc = m0_cm_type_register(&sns_repair_cmt) ?:
             m0_cm_type_register(&sns_rebalance_cmt);

        return M0_RC(rc);
}

M0_INTERNAL void m0_sns_cm_type_deregister(void)
{
        m0_cm_type_deregister(&sns_repair_cmt);
        m0_cm_type_deregister(&sns_rebalance_cmt);
}

M0_INTERNAL struct m0_cm_cp *m0_sns_cm_cp_alloc(struct m0_cm *cm)
{
        struct m0_sns_cm_cp *scp;

        M0_ALLOC_PTR(scp);
        if (scp == NULL)
                return NULL;

        return &scp->sc_base;
}

static void bp_below_threshold(struct m0_net_buffer_pool *bp)
{
        /* Buffer pool is below threshold.  */
}

static void buf_available(struct m0_net_buffer_pool *pool)
{
        struct m0_sns_cm_buf_pool *bp = container_of(pool, struct m0_sns_cm_buf_pool,
                                                     sb_bp);
        m0_chan_signal(&bp->sb_wait);
}

const struct m0_net_buffer_pool_ops bp_ops = {
        .nbpo_not_empty       = buf_available,
        .nbpo_below_threshold = bp_below_threshold
};

static void sns_cm_bp_init(struct m0_sns_cm_buf_pool *sbp)
{
        m0_chan_init(&sbp->sb_wait, &sbp->sb_bp.nbp_mutex);
}

static void sns_cm_bp_fini(struct m0_sns_cm_buf_pool *sbp)
{
        m0_chan_fini_lock(&sbp->sb_wait);
}

M0_TL_DESCR_DECLARE(cs_eps, extern);
M0_TL_DECLARE(cs_eps, M0_INTERNAL, struct cs_endpoint_and_xprt);

M0_INTERNAL int m0_sns_cm_rm_init(struct m0_sns_cm *scm)
{
        struct m0_reqh         *reqh = m0_sns_cm2reqh(scm);
        struct m0_motr         *motr = m0_cs_ctx_get(reqh);
        struct m0_pools_common *pc = &motr->cc_pools_common;
        int                     rc;

        M0_ENTRY("scm: %p", scm);
        M0_ASSERT(pc->pc_nr_devices > 0);

        if (!reqh->rh_oostore) {
                scm->sc_rm_ctx.rc_pc = pc;
                m0_rm_domain_init(&scm->sc_rm_ctx.rc_dom);
        /*
         * XXX Init and register new resource type for file locks.
         * m0_rm_type_register() starts local credit processing thread for
         * processing incoming replies. This is necessary when rm service
         * runs on the same motr instance on which sns cm service is running.
         * Currently, it is not clear if the RM service would be running in a
         * separate motr instance on which no other service (except MDS) is
         * running.
         **/
                m0_sns_cm_flock_resource_set(scm);
                m0_rm_type_register(&scm->sc_rm_ctx.rc_dom,
                                    &scm->sc_rm_ctx.rc_rt);
        }

        rc = m0_scmfctx_htable_init(&scm->sc_file_ctx, pc->pc_nr_devices);
        if (rc != 0)
                goto end;

        m0_mutex_init(&scm->sc_file_ctx_mutex);
end:
        return M0_RC(rc);
}

M0_INTERNAL int m0_sns_cm_setup(struct m0_cm *cm)
{
        struct m0_reqh        *reqh;
        struct m0_net_domain  *ndom;
        struct m0_sns_cm      *scm;
        uint64_t               colours;
        m0_bcount_t            segment_size;
        uint32_t               segments_nr;
        int                    rc;

        M0_ENTRY("cm: %p", cm);

        scm = cm2sns(cm);
        reqh = cm->cm_service.rs_reqh;
        /*
         * Total number of colours in incoming and outgoing buffer pools is
         * same as the total number of localities in the reqh fom domain.
         */
        colours = m0_reqh_nr_localities(reqh);
        ndom = m0_cs_net_domain_locate(m0_cs_ctx_get(reqh),
                                       m0_net_xprt_default_get()->nx_name);
        /*
         * XXX This should be fixed, buffer pool ops should be a parameter to
         * m0_net_buffer_pool_init() as it is NULL checked in
         * m0_net_buffer_pool_invariant().
         */
        scm->sc_ibp.sb_bp.nbp_ops = &bp_ops;
        scm->sc_obp.sb_bp.nbp_ops = &bp_ops;
        segment_size = m0_net_domain_get_max_buffer_segment_size(ndom);
        segments_nr  = m0_net_domain_get_max_buffer_segments(ndom);
        M0_LOG(M0_DEBUG, "sns segments_nr=%d", segments_nr);
        rc = m0_net_buffer_pool_init(&scm->sc_ibp.sb_bp, ndom,
                                     0, segments_nr, segment_size,
                                     colours, M0_0VEC_SHIFT,
                                     /* dont_dump */true);
        if (rc == 0) {
                rc = m0_net_buffer_pool_init(&scm->sc_obp.sb_bp, ndom,
                                             0, segments_nr, segment_size,
                                             colours, M0_0VEC_SHIFT,
                                             /* dont_dump */true);
                if (rc != 0)
                        m0_net_buffer_pool_fini(&scm->sc_ibp.sb_bp);
        }

        if (rc == 0) {
                rc = m0_sns_cm_iter_init(&scm->sc_it);
                if (rc != 0)
                        return M0_RC(rc);
                sns_cm_bp_init(&scm->sc_obp);
                sns_cm_bp_init(&scm->sc_ibp);
        }
        M0_LEAVE();
        return M0_RC(rc);
}

M0_INTERNAL bool m0_sns_is_peer(struct m0_cm               *cm,
                                struct m0_reqh_service_ctx *ctx)
{
        return ctx->sc_type == M0_CST_IOS;
}

M0_INTERNAL size_t m0_sns_cm_buffer_pool_provision(struct m0_net_buffer_pool *bp,
                                                   size_t bufs_nr)
{
        size_t bnr;

        m0_net_buffer_pool_lock(bp);
        M0_ASSERT(m0_net_buffer_pool_invariant(bp));
        bnr = m0_net_buffer_pool_provision(bp, bufs_nr);
        m0_net_buffer_pool_unlock(bp);

        return bnr;
}

M0_INTERNAL int m0_sns_cm_prepare(struct m0_cm *cm)
{
        struct m0_sns_cm *scm = cm2sns(cm);
        struct m0_reqh   *reqh = m0_sns_cm2reqh(scm);
        struct m0_motr   *motr = m0_cs_ctx_get(reqh);
        int               bufs_nr;
        int               rc;

        M0_ENTRY("cm: %p", cm);
        M0_PRE(M0_IN(scm->sc_op, (CM_OP_REPAIR, CM_OP_REBALANCE)));

        rc = m0_sns_cm_rm_init(scm);
        if (rc != 0)
                return M0_ERR_INFO(rc, "SNS RM init failed");

        if (scm->sc_ibp.sb_bp.nbp_buf_nr == 0 &&
            scm->sc_obp.sb_bp.nbp_buf_nr == 0) {
                bufs_nr = m0_sns_cm_buffer_pool_provision(&scm->sc_ibp.sb_bp,
                                                          motr->cc_sns_buf_nr);
                M0_LOG(M0_DEBUG, "Got buffers in: [%d]", bufs_nr);
                if (bufs_nr == 0)
                        return M0_ERR(-ENOMEM);
                bufs_nr = m0_sns_cm_buffer_pool_provision(&scm->sc_obp.sb_bp,
                                                          motr->cc_sns_buf_nr);
                M0_LOG(M0_DEBUG, "Got buffers out: [%d]", bufs_nr);
                /*
                 * If bufs_nr is 0, then just return -ENOMEM, as cm_setup() was
                 * successful, both the buffer pools (incoming and outgoing)
                 * will be finalised in cm_fini().
                 */
                if (bufs_nr == 0)
                        return M0_ERR(-ENOMEM);
        }
        scm->sc_ibp_reserved_nr = 0;

        rc = m0_sns_cm_ag_iter_init(&scm->sc_ag_it);
        scm->sc_total_read_size = NULL;
        scm->sc_total_write_size = NULL;
        m0_ios_cdom_get(scm->sc_base.cm_service.rs_reqh, &scm->sc_cob_dom);

        M0_LEAVE();
        return M0_RC(rc);
}

M0_INTERNAL int m0_sns_cm_fail_dev_log(struct m0_cm *cm,
                                        enum m0_pool_nd_state state)
{
        struct m0_pools_common *pc = cm->cm_service.rs_reqh->rh_pools;
        struct m0_pool         *pool;
        struct m0_pooldev      *pd;
        uint32_t                nr_devs = 0;
        char                   *pd_state[M0_PNDS_NR] = {
                                        [M0_PNDS_SNS_REPAIRING] = "Repairing",
                                        [M0_PNDS_SNS_REBALANCING] = "Rebalancing"
                                };

        M0_PRE(cm != NULL);
        M0_PRE(M0_IN(state, (M0_PNDS_SNS_REPAIRING, M0_PNDS_SNS_REBALANCING)));

        m0_tl_for(pools, &pc->pc_pools, pool) {
                m0_tl_for(pool_failed_devs, &pool->po_failed_devices, pd) {
                        if (pd->pd_state == state) {
                                M0_LOG(M0_NOTICE, " %s disk "FID_F
                                                " from pool "FID_F,
                                                pd_state[pd->pd_state],
                                                FID_P(&pd->pd_id),
                                                FID_P(&pool->po_id));
                                M0_CNT_INC(nr_devs);
                        }
                } m0_tl_endfor;
        } m0_tl_endfor;

        if (nr_devs == 0) {
                M0_LOG(M0_ERROR, "No disk found for  %s", pd_state[state]);
                return -ENOENT;
        }

        return 0;
}

M0_INTERNAL int m0_sns_cm_start(struct m0_cm *cm)
{
        struct m0_sns_cm *scm = cm2sns(cm);
        size_t            loc_nr = m0_fom_dom()->fd_localities_nr;
        int               rc;

        M0_ENTRY("cm: %p", cm);
        M0_PRE(M0_IN(scm->sc_op, (CM_OP_REPAIR, CM_OP_REBALANCE)));

        M0_ALLOC_ARR(scm->sc_total_read_size, loc_nr);
        if (scm->sc_total_read_size == NULL)
                return M0_ERR(-ENOMEM);
        M0_ALLOC_ARR(scm->sc_total_write_size, loc_nr);
        if (scm->sc_total_write_size == NULL) {
                m0_free(scm->sc_total_read_size);
                return M0_ERR(-ENOMEM);
        }
        rc = m0_sns_cm_iter_start(&scm->sc_it);
        if (rc != 0) {
                m0_free(scm->sc_total_read_size);
                m0_free(scm->sc_total_write_size);
                return M0_ERR(rc);
        }
        scm->sc_start_time = m0_time_now();
        scm->sc_last_status_print_time = m0_time_now();

        return M0_RC(rc);
}

M0_INTERNAL void m0_sns_cm_rm_fini(struct m0_sns_cm *scm)
{
        M0_ENTRY("scm: %p", scm);

        m0_sns_cm_fctx_cleanup(scm);
        m0_scmfctx_htable_fini(&scm->sc_file_ctx);

        if (!m0_sns_cm2reqh(scm)->rh_oostore)
                m0_rm_type_deregister(&scm->sc_rm_ctx.rc_rt);
}

static void buffer_pool_prune(struct m0_net_buffer_pool *bp)
{
        m0_net_buffer_pool_lock(bp);
        while (m0_net_buffer_pool_prune(bp))
        {;}
        m0_net_buffer_pool_unlock(bp);
}

static void sns_cm_buffer_pools_prune(struct m0_sns_cm *scm)
{
        buffer_pool_prune(&scm->sc_obp.sb_bp);
        buffer_pool_prune(&scm->sc_ibp.sb_bp);
        M0_ASSERT(scm->sc_obp.sb_bp.nbp_buf_nr == 0 &&
                  scm->sc_ibp.sb_bp.nbp_buf_nr == 0);
}

static bool sns_cm_status_get(struct m0_sns_cm *scm,
                              size_t *tread, size_t *twrite)
{
        size_t loc_nr = m0_fom_dom()->fd_localities_nr;
        int    i;

        if (scm->sc_total_read_size != NULL &&
            scm->sc_total_write_size != NULL) {
                *tread = 0;
                *twrite = 0;
                for (i = 0; i < loc_nr; ++i) {
                        *tread += scm->sc_total_read_size[i];
                        *twrite += scm->sc_total_write_size[i];
                }
                return true;
        }
        return false;
}

M0_INTERNAL void m0_sns_cm_stop(struct m0_cm *cm)
{
        struct m0_sns_cm *scm = cm2sns(cm);
        size_t            twrite;
        size_t            tread;

        M0_ENTRY();

        m0_sns_cm_iter_stop(&scm->sc_it);
        scm->sc_stop_time = m0_time_now();
        if (sns_cm_status_get(scm, &tread, &twrite)) {
                M0_LOG(M0_WARN, "Time: %llu Read Size: %llu Write size: %llu",
                       (unsigned long long)m0_time_sub(scm->sc_stop_time,
                                                       scm->sc_start_time),
                       (unsigned long long)tread, (unsigned long long)twrite);
                M0_CNT_INC(scm->sc_repair_done);
                m0_free(scm->sc_total_read_size);
                m0_free(scm->sc_total_write_size);
        }
        m0_sns_cm_rm_fini(scm);
        m0_sns_cm_ag_iter_fini(&scm->sc_ag_it);
        sns_cm_buffer_pools_prune(scm);

        M0_LEAVE();
}

M0_INTERNAL void m0_sns_cm_print_status(struct m0_sns_cm *scm)
{
        size_t           twrite;
        size_t           tread;
        m0_time_t        now = m0_time_now();

        if (scm->sc_last_status_print_time + M0_MKTIME(60,0) < now &&
            sns_cm_status_get(scm, &tread, &twrite)) {
                scm->sc_last_status_print_time = now;
                m0_console_printf("SNS-%02"PRIu32": read=%zu written=%zu\n",
                                  scm->sc_repair_done, tread, twrite);
        }
}

M0_INTERNAL void m0_sns_cm_ha_msg(struct m0_cm *cm, struct m0_ha_msg *msg,
                                    int rc)
{
        struct m0_sns_cm *scm = cm2sns(cm);

        *msg = (struct m0_ha_msg) {
                .hm_time = m0_time_now(),
                .hm_data = {
                        .hed_type = M0_HA_MSG_SNS_ERR,
                        .u.hed_ha_sns_err = (struct m0_ha_sns_err) {
                                .hse_errcode = (uint32_t)rc,
                                .hse_opcode  = (uint8_t)scm->sc_op
                        }
                }
        };
}

M0_INTERNAL void m0_sns_cm_fini(struct m0_cm *cm)
{
        struct m0_sns_cm *scm;

        M0_ENTRY("cm: %p", cm);

        scm = cm2sns(cm);
        m0_sns_cm_iter_fini(&scm->sc_it);

        /*
         * Finalise parents first to avoid usage of finalised mutexes.
         * m0_sns_cm_setup() makes initialisation in reverse order too.
         */
        sns_cm_bp_fini(&scm->sc_obp);
        sns_cm_bp_fini(&scm->sc_ibp);

        m0_net_buffer_pool_fini(&scm->sc_ibp.sb_bp);
        m0_net_buffer_pool_fini(&scm->sc_obp.sb_bp);

        M0_LEAVE();
}

M0_INTERNAL uint64_t m0_sns_cm_cp_buf_nr(struct m0_net_buffer_pool *bp,
                                         uint64_t data_seg_nr)
{
        return data_seg_nr % bp->nbp_seg_nr ?
               data_seg_nr / bp->nbp_seg_nr + 1 :
               data_seg_nr / bp->nbp_seg_nr;
}

M0_INTERNAL void m0_sns_cm_buf_wait(struct m0_sns_cm_buf_pool *sbp, struct m0_fom *fom)
{
        m0_net_buffer_pool_lock(&sbp->sb_bp);
        m0_fom_wait_on(fom, &sbp->sb_wait, &fom->fo_cb);
        m0_net_buffer_pool_unlock(&sbp->sb_bp);
}

M0_INTERNAL int m0_sns_cm_buf_attach(struct m0_net_buffer_pool *bp,
                                     struct m0_cm_cp *cp)
{
        struct m0_net_buffer *buf;
        size_t                colour;
        uint32_t              seg_nr = 0;
        uint32_t              rem_bufs;
        int                   rc = 0;

        colour =  cp_home_loc_helper(cp) % bp->nbp_colours_nr;
        rem_bufs = m0_sns_cm_cp_buf_nr(bp, cp->c_data_seg_nr);
        rem_bufs -= cp->c_buf_nr;
        m0_net_buffer_pool_lock(bp);
        while (rem_bufs > 0) {
                buf = m0_cm_buffer_get(bp, colour);
                if (buf == NULL) {
                        rc = -ENOBUFS;
                        goto out;
                }
                m0_cm_cp_buf_add(cp, buf);
                if (cp->c_data_seg_nr > (cp->c_buf_nr * bp->nbp_seg_nr))
                        seg_nr = bp->nbp_seg_nr;
                else
                        seg_nr = cp->c_data_seg_nr -
                                 ((cp->c_buf_nr - 1) * bp->nbp_seg_nr);
                buf->nb_buffer.ov_vec.v_nr = seg_nr;
                M0_CNT_DEC(rem_bufs);
        }
        if (bp->nbp_free > 0)
                bp->nbp_ops->nbpo_not_empty(bp);
out:
        m0_net_buffer_pool_unlock(bp);
        return M0_RC(rc);
}

M0_INTERNAL uint64_t m0_sns_cm_data_seg_nr(struct m0_sns_cm *scm,
                                           struct m0_pdclust_layout *pl)
{
        M0_PRE(scm != NULL && pl != NULL);

        return m0_pdclust_unit_size(pl) %
               scm->sc_obp.sb_bp.nbp_seg_size ?
               m0_pdclust_unit_size(pl) /
               scm->sc_obp.sb_bp.nbp_seg_size + 1 :
               m0_pdclust_unit_size(pl) /
               scm->sc_obp.sb_bp.nbp_seg_size;
}

M0_INTERNAL int64_t
m0_sns_cm_incoming_reserve_bufs(struct m0_sns_cm *scm,
                                const struct m0_cm_ag_id *id)
{
        uint64_t                   nr_cp_bufs;
        uint64_t                   cp_data_seg_nr;
        uint32_t                   nr_cps_in;
        uint32_t                   nr_units_in;
        struct m0_pdclust_layout  *pl;
        struct m0_sns_cm_file_ctx *fctx;
        int                        rc;

        fctx = m0_sns_cm_fctx_get(scm, id);
        M0_ASSERT(fctx != NULL);

        pl = m0_layout_to_pdl(fctx->sf_layout);
        rc = m0_sns_cm_ag_in_cp_units(scm, id, fctx, &nr_cps_in,
                                      &nr_units_in, NULL);
        if (rc != 0)
                return M0_ERR(rc);
        cp_data_seg_nr = m0_sns_cm_data_seg_nr(scm, pl);
        nr_cp_bufs = m0_sns_cm_cp_buf_nr(&scm->sc_ibp.sb_bp, cp_data_seg_nr);
        m0_sns_cm_fctx_put(scm, id);

        return nr_cp_bufs * nr_cps_in;
}

M0_INTERNAL int m0_sns_cm_has_space_for(struct m0_sns_cm *scm,
                                        struct m0_pdclust_layout *pl,
                                        uint64_t nr_bufs)
{
        struct m0_net_buffer_pool *ibp;
        int                        rc = 0;

        M0_PRE(scm != NULL && pl != NULL);
        M0_PRE(m0_cm_is_locked(&scm->sc_base));

        ibp = &scm->sc_ibp.sb_bp;
        m0_net_buffer_pool_lock(ibp);
        if (nr_bufs + scm->sc_ibp_reserved_nr > ibp->nbp_free)
                rc = -ENOSPC;
        m0_net_buffer_pool_unlock(ibp);
        M0_LOG(M0_DEBUG, "nr_bufs: [%" PRIu64 "] free buffers in: [%u] out: [%u] "
               "sc_ibp_reserved_nr: [%" PRIu64 "]", nr_bufs, ibp->nbp_free,
               scm->sc_obp.sb_bp.nbp_free, scm->sc_ibp_reserved_nr);

        return M0_RC(rc);
}

M0_INTERNAL void m0_sns_cm_reserve_space(struct m0_sns_cm *scm, size_t nr_bufs)
{
        struct m0_net_buffer_pool *ibp;

        ibp = &scm->sc_ibp.sb_bp;
        m0_net_buffer_pool_lock(ibp);
        scm->sc_ibp_reserved_nr += nr_bufs;
        m0_net_buffer_pool_unlock(ibp);
}

M0_INTERNAL void m0_sns_cm_cancel_reservation(struct m0_sns_cm *scm, size_t nr_bufs)
{
        struct m0_net_buffer_pool *ibp;

        ibp = &scm->sc_ibp.sb_bp;
        m0_net_buffer_pool_lock(ibp);
        scm->sc_ibp_reserved_nr -= min32u(scm->sc_ibp_reserved_nr, nr_bufs);
        M0_LOG(M0_DEBUG, "reserved: %u nr_bufs: %u", (unsigned)scm->sc_ibp_reserved_nr, (unsigned)nr_bufs);
        ibp->nbp_ops->nbpo_not_empty(ibp);
        m0_net_buffer_pool_unlock(ibp);
}

M0_INTERNAL int m0_sns_cm_ag_next(struct m0_cm *cm,
                                  const struct m0_cm_ag_id *id_curr,
                                  struct m0_cm_ag_id *id_next)
{
        struct m0_sns_cm *scm = cm2sns(cm);

        M0_PRE(cm != NULL);
        M0_PRE(m0_cm_is_locked(cm));

        return m0_sns_cm_ag__next(scm, id_curr, id_next);
}

#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:
 */