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_DIXCM
#include "lib/memory.h"
#include "lib/assert.h"
#include "lib/errno.h"
#include "lib/trace.h"
#include "lib/chan.h"

#include "conf/helpers.h"     /* m0_conf_service_get */
#include "fop/fom.h"
#include "reqh/reqh.h"
#include "rpc/rpc_machine.h"
#include "pool/pool.h"
#include "cm/proxy.h"

#include "dix/fid_convert.h"
#include "dix/cm/cp.h"
#include "dix/cm/cm.h"
#include "dix/cm/iter.h"

M0_INTERNAL struct m0_dix_cm *cm2dix(struct m0_cm *cm)
{
        return container_of(cm, struct m0_dix_cm, dcm_base);
}

M0_INTERNAL int m0_dix_cm_type_register(void)
{
        int rc;

        rc = m0_cm_type_register(&dix_repair_cmt) ?:
             m0_cm_type_register(&dix_rebalance_cmt);
        if (rc == 0) {
                m0_dix_cm_iter_type_register(&dix_repair_dcmt);
                m0_dix_cm_iter_type_register(&dix_rebalance_dcmt);
        }
        return M0_RC(rc);
}

M0_INTERNAL void m0_dix_cm_type_deregister(void)
{
        m0_cm_type_deregister(&dix_repair_cmt);
        m0_cm_type_deregister(&dix_rebalance_cmt);
}

M0_INTERNAL int m0_dix_cm_setup(struct m0_cm *cm)
{
        return M0_RC(0);
}

M0_INTERNAL int m0_dix_cm_prepare(struct m0_cm *cm)
{
        struct m0_dix_cm *dcm = cm2dix(cm);

        dcm->dcm_stats_key = -1;

        return M0_RC(0);
}

static bool dix_cm_proxies_completed_cb(struct m0_clink *cl)
{
        struct m0_dix_cm        *dcm = M0_AMB(dcm, cl, dcm_proxies_completed);
        struct m0_cm            *cm  = &dcm->dcm_base;
        struct m0_cm_aggr_group *end_mark;
        bool                     completed;
        M0_ENTRY();

        M0_PRE(m0_cm_is_locked(cm));
        completed = (cm->cm_proxy_active_nr == 0);
        if (completed || cm->cm_abort) {
                end_mark = m0_cm_aggr_group_locate(cm, &GRP_END_MARK_ID, true);
                M0_ASSERT(end_mark != NULL);
                m0_cm_aggr_group_fini(end_mark);
                m0_clink_del(cl);
                m0_clink_fini(cl);
        }

        return M0_RC(false);
}

M0_INTERNAL int m0_dix_cm_start(struct m0_cm *cm)
{
        struct m0_dix_cm        *dcm  = cm2dix(cm);
        struct m0_cm_aggr_group *ag   = NULL;
        struct m0_reqh          *reqh = cm->cm_service.rs_reqh;
        int                      rc;
        M0_ENTRY();

        M0_PRE(m0_cm_is_locked(cm));

        if (!proxy_tlist_is_empty(&cm->cm_proxies)) {
                /* Create end-marker ag and add it into ag incoming list. */
                rc = m0_cm_aggr_group_alloc(cm, &GRP_END_MARK_ID, true, &ag);
                if (rc != 0)
                        return M0_ERR(rc);
                m0_clink_init(&dcm->dcm_proxies_completed,
                              dix_cm_proxies_completed_cb);
                m0_clink_add_lock(&dcm->dcm_base.cm_wait,
                                  &dcm->dcm_proxies_completed);
        }

        dcm->dcm_stats_key = m0_locality_data_alloc(
                sizeof(struct m0_dix_cm_stats),
                NULL, NULL, NULL);
        if (dcm->dcm_stats_key < 0)
                return M0_ERR(dcm->dcm_stats_key);

        /*
         * Unlock CM since FOM locality lock is taken during waiting for
         * iterator to start. It prevents usual locality->CM locking order that
         * can lead to deadlock.
         */
        m0_cm_unlock(cm);
        rc = m0_dix_cm_iter_start(&dcm->dcm_it, dcm->dcm_type, reqh,
                                  m0_cm_rpc_machine_find(reqh)->rm_bulk_cutoff);
        m0_cm_lock(cm);
        if (rc != 0) {
                if (ag != NULL) {
                        m0_cm_aggr_group_fini(ag);
                        m0_free(ag);
                }
                m0_locality_data_free(dcm->dcm_stats_key);
        }
        dcm->dcm_start_time = m0_time_now();
        return M0_RC(rc);
}

static void dix_cm_tstats_calc(int loc_idx, void *loc_stats, void *total_stats)
{
        M0_PRE(loc_stats != NULL);
        M0_PRE(total_stats != NULL);

        ((struct m0_dix_cm_stats *)total_stats)->dcs_read_size +=
                ((struct m0_dix_cm_stats *)loc_stats)->dcs_read_size;
        ((struct m0_dix_cm_stats *)total_stats)->dcs_write_size +=
                ((struct m0_dix_cm_stats *)loc_stats)->dcs_write_size;
}

M0_INTERNAL void m0_dix_cm_stop(struct m0_cm *cm)
{
        struct m0_dix_cm        *dcm = cm2dix(cm);
        struct m0_cm_aggr_group *end_mark;
        struct m0_dix_cm_stats   total_stats = {};
        M0_PRE(m0_cm_is_locked(cm));

        if (!cm->cm_done && !proxy_tlist_is_empty(&cm->cm_proxies)) {
                end_mark = m0_cm_aggr_group_locate(cm, &GRP_END_MARK_ID, true);
                M0_ASSERT(end_mark != NULL);
                m0_cm_aggr_group_fini(end_mark);
                m0_clink_del_lock(&dcm->dcm_proxies_completed);
                m0_clink_fini(&dcm->dcm_proxies_completed);
        }
        /* release the cm lock, because m0_dix_cm_iter_stop() may block. */
        m0_cm_unlock(cm);
        m0_dix_cm_iter_stop(&dcm->dcm_it);
        m0_cm_lock(cm);
        M0_SET0(&dcm->dcm_it);
        dcm->dcm_stop_time = m0_time_now();

        if (dcm->dcm_stats_key >= 0) {
                m0_locality_data_iterate(dcm->dcm_stats_key,
                                         &dix_cm_tstats_calc,
                                         &total_stats);

                M0_LOG(M0_DEBUG, "Time: %llu Read Size: %llu Write size: %llu",
                       (unsigned long long)m0_time_sub(dcm->dcm_stop_time,
                                                       dcm->dcm_start_time),
                       (unsigned long long)total_stats.dcs_read_size,
                       (unsigned long long)total_stats.dcs_write_size);

                m0_locality_data_free(dcm->dcm_stats_key);
        }
}

M0_INTERNAL void m0_dix_cm_fini(struct m0_cm *cm)
{
}

M0_INTERNAL int m0_dix_get_space_for(struct m0_cm             *cm,
                                     const struct m0_cm_ag_id *id,
                                     size_t                   *count)
{
        *count = 1;
        return M0_RC(0);
}

M0_INTERNAL int m0_dix_cm_ag_next(struct m0_cm *cm,
                                  const struct m0_cm_ag_id *id_curr,
                                  struct m0_cm_ag_id *id_next)
{
        M0_ENTRY();
        /*
         * Incoming aggregation group list is not populated in case of DIX
         * repair/re-balance, because local CM doesn't know which copy packets
         * it will receive.
         */
        return M0_RC(-ENODATA);
}

M0_INTERNAL struct m0_reqh *m0_dix_cm2reqh(const struct m0_dix_cm *dcm)
{
        return dcm->dcm_base.cm_service.rs_reqh;
}

static struct m0_cm_proxy *dix_cm_sdev2proxy(struct m0_dix_cm *dcm,
                                             uint32_t          sdev_id)
{
        struct m0_reqh             *reqh;
        struct m0_confc            *confc;
        struct m0_fid               svc_fid;
        struct m0_conf_service     *svc = NULL;
        int                         rc;

        reqh = m0_dix_cm2reqh(dcm);
        M0_ASSERT(reqh != NULL);
        confc = m0_reqh2confc(reqh);
        M0_ASSERT(confc != NULL);
        svc_fid = reqh->rh_pools->pc_dev2svc[sdev_id].pds_ctx->sc_fid;
        rc = m0_conf_service_get(confc, &svc_fid, &svc);
        M0_ASSERT(rc == 0);
        M0_ASSERT(svc->cs_type == M0_CST_CAS);
        M0_ASSERT(svc->cs_endpoints[0] != NULL);

        return m0_cm_proxy_locate(&dcm->dcm_base, svc->cs_endpoints[0]);
}

static int dix_cm_ag_setup(struct m0_cm    *cm,
                           struct m0_cm_cp *cp,
                           struct m0_fid   *cctg_fid,
                           uint64_t         recs_nr)
{
        struct m0_cm_ag_id       ag_id;
        struct m0_cm_aggr_group *ag;
        int                      rc;

        M0_ENTRY("cm %p", cp);
        /* Build aggregation group id. */
        ag_id.ai_hi.u_hi = cctg_fid->f_container;
        ag_id.ai_hi.u_lo = cctg_fid->f_key;
        ag_id.ai_lo.u_hi = 0;
        ag_id.ai_lo.u_hi = recs_nr;
        rc = m0_cm_aggr_group_alloc(cm, &ag_id, false, &ag);
        if (rc == 0)
                m0_cm_ag_cp_add(ag, cp);

        return M0_RC(rc);
}

M0_INTERNAL int m0_dix_cm_data_next(struct m0_cm *cm, struct m0_cm_cp *cp)
{
        struct m0_dix_cm      *dcm  = cm2dix(cm);
        struct m0_dix_cm_iter *iter = &dcm->dcm_it;
        struct m0_fom         *pfom = &cm->cm_cp_pump.p_fom;
        struct m0_dix_cm_cp   *dix_cp;
        struct m0_buf          key = {};
        struct m0_buf          val = {};
        struct m0_cm_proxy    *proxy;
        struct m0_fid          local_cctg_fid = {};
        struct m0_fid          remote_cctg_fid = {};
        struct m0_fid          dix_fid = {};
        uint64_t               processed_recs_nr;
        uint32_t               sdev_id = (uint32_t)-1;
        int                    rc;

        /* Inc progress counter. */
        dcm->dcm_processed_nr++;

        if (cm->cm_quiesce || cm->cm_abort) {
                M0_LOG(M0_WARN, "%" PRId64 ": Got %s cmd: returning -ENODATA",
                                 cm->cm_id,
                                 cm->cm_quiesce ? "QUIESCE" : "ABORT");
                return M0_RC(-ENODATA);
        }

        if (!dcm->dcm_iter_inprogress) {
                if (!dcm->dcm_cp_in_progress) {
                        m0_chan_lock(&iter->di_completed);
                        m0_fom_wait_on(pfom, &iter->di_completed, &pfom->fo_cb);
                        m0_chan_unlock(&iter->di_completed);
                        m0_dix_cm_iter_next(iter);
                        dcm->dcm_iter_inprogress = true;
                        M0_LOG(M0_DEBUG, "pump fom %p going to wait for "
                                         "iter fom %p",
                                         pfom, &iter->di_fom);
                }
                return M0_FSO_WAIT;
        } else {
                dcm->dcm_iter_inprogress = false;
                /* Set proxy for copy packet and fill key/value AT buffers. */
                rc = m0_dix_cm_iter_get(iter, &key, &val, &sdev_id);
                if (rc != 0) {
                        if (rc == -ENODATA)
                                cm->cm_last_out_hi = GRP_END_MARK_ID;
                        return M0_ERR(rc);
                }
                /* Setup aggregation group. */
                m0_dix_cm_iter_cur_pos(iter, &local_cctg_fid,
                                       &processed_recs_nr);
                rc = dix_cm_ag_setup(cm, cp, &local_cctg_fid,
                                     processed_recs_nr);
                if (rc == 0) {
                        dix_cp = M0_AMB(dix_cp, cp, dc_base);
                        M0_ASSERT(dix_cp != NULL);
                        proxy = dix_cm_sdev2proxy(dcm, sdev_id);
                        M0_ASSERT(proxy != NULL);
                        cp->c_cm_proxy = proxy;
                        dix_cp->dc_key = key;
                        dix_cp->dc_val = val;

                        /*
                         * Convert FID of local component catalogue to FID of
                         * remote component catalogue,
                         */
                        m0_dix_fid_convert_cctg2dix(&local_cctg_fid, &dix_fid);
                        m0_dix_fid_convert_dix2cctg(&dix_fid, &remote_cctg_fid,
                                                    sdev_id);

                        dix_cp->dc_ctg_fid       = remote_cctg_fid;
                        dix_cp->dc_ctg_op_flags |= COF_CREATE;
                        dix_cp->dc_is_local      = true;
                        dcm->dcm_cp_in_progress  = true;

                        rc = M0_FSO_AGAIN;
                } else {
                        /*
                         * Key/value buffers are not needed anymore in error
                         * case. In case rc == 0 we will destroy these key and
                         * val in dix_cp_fop_release().
                         */
                        m0_buf_free(&key);
                        m0_buf_free(&val);
                }
                return rc;
        }
}

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

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