cm.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"          /* M0_LOG */
#include "lib/bob.h"            /* M0_BOB_DEFINE */
#include "lib/misc.h"           /* M0_SET0 */
#include "lib/assert.h"         /* M0_PRE, M0_POST */
#include "lib/errno.h"
#include "lib/finject.h"
#include "lib/memory.h"

#include "module/instance.h"    /* m0_get */
#include "motr/magic.h"
#include "motr/setup.h"
#include "net/buffer_pool.h"
#include "rpc/rpclib.h"
#include "rpc/rpc_machine.h"
#include "reqh/reqh.h"
#include "fop/fop.h"
#include "pool/pool.h"          /* pools_common_svc */
#include "conf/obj_ops.h"       /* m0_conf_obj_find_lock */

#include "ha/ha.h"              /* m0_ha_send */

#include "cm/cm.h"
#include "cm/ag.h"
#include "cm/cp.h"
#include "cm/proxy.h"
#include "cm/sw.h"
#include "cm/sw_xc.h"
#include "cm/cp_onwire_xc.h"
#include "cm/ag_xc.h"

static struct m0_tl cmtypes;

static struct m0_mutex cmtypes_mutex;

M0_TL_DESCR_DEFINE(cmtypes, "copy machine types", static,
                   struct m0_cm_type, ct_linkage, ct_magix,
                   CM_TYPE_LINK_MAGIX, CM_TYPE_HEAD_MAGIX);

M0_TL_DEFINE(cmtypes, static, struct m0_cm_type);

static struct m0_bob_type cmtypes_bob;
M0_BOB_DEFINE(static, &cmtypes_bob, m0_cm_type);

static struct m0_sm_state_descr cm_state_descr[M0_CMS_NR] = {
        [M0_CMS_INIT] = {
                .sd_flags       = M0_SDF_INITIAL,
                .sd_name        = "cm_init",
                .sd_allowed     = M0_BITS(M0_CMS_IDLE, M0_CMS_FINI)
        },
        [M0_CMS_IDLE] = {
                .sd_flags       = 0,
                .sd_name        = "cm_idle",
                .sd_allowed     = M0_BITS(M0_CMS_FAIL, M0_CMS_PREPARE,
                                          M0_CMS_FINI)
        },
        [M0_CMS_PREPARE] = {
                .sd_flags       = 0,
                .sd_name        = "cm_prepare",
                .sd_allowed     = M0_BITS(M0_CMS_READY, M0_CMS_FAIL)
        },
        [M0_CMS_READY] = {
                .sd_flags       = 0,
                .sd_name        = "cm_ready",
                .sd_allowed     = M0_BITS(M0_CMS_ACTIVE, M0_CMS_FAIL)
        },
        [M0_CMS_ACTIVE] = {
                .sd_flags       = 0,
                .sd_name        = "cm_active",
                .sd_allowed     = M0_BITS(M0_CMS_STOP, M0_CMS_FAIL)
        },
        [M0_CMS_FAIL] = {
                .sd_flags       = M0_SDF_FAILURE,
                .sd_name        = "cm_fail",
                .sd_allowed     = M0_BITS(M0_CMS_PREPARE, M0_CMS_FINI)
        },
        [M0_CMS_STOP] = {
                .sd_flags       = 0,
                .sd_name        = "cm_stop",
                .sd_allowed     = M0_BITS(M0_CMS_IDLE, M0_CMS_FINI,
                                          M0_CMS_PREPARE)
        },
        [M0_CMS_FINI] = {
                .sd_flags       = M0_SDF_TERMINAL,
                .sd_name        = "cm_fini",
                .sd_allowed     = 0
        },
};

static const struct m0_sm_conf cm_sm_conf = {
        .scf_name       = "sm:cm conf",
        .scf_nr_states  = M0_CMS_NR,
        .scf_state      = cm_state_descr
};

M0_INTERNAL struct m0_cm *m0_cmsvc2cm(struct m0_reqh_service *cmsvc)
{
        return container_of(cmsvc, struct m0_cm, cm_service);
}

M0_INTERNAL int m0_ha_cm_err_send(struct m0_cm *cm, int rc)
{
        struct m0_ha_msg *msg;
        uint64_t          tag;

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

        M0_PRE(cm != NULL);
        M0_PRE(rc < 0);

        M0_LOG(M0_DEBUG, "Notify HA about cm failure, state=%d rc=%d",
                m0_cm_state_get(cm), rc);

        M0_ALLOC_PTR(msg);
        if (msg == NULL)
                return M0_ERR(-ENOMEM);

        cm->cm_ops->cmo_ha_msg(cm, msg, rc);
        m0_ha_send(m0_get()->i_ha, m0_get()->i_ha_link, msg, &tag);
        m0_free(msg);

        M0_LEAVE();
        return M0_RC(0);
}

M0_INTERNAL void m0_cm_fail(struct m0_cm *cm, int rc)
{
        M0_ENTRY("cm: %p rc: %d", cm, rc);

        M0_PRE(cm != NULL);
        M0_PRE(rc < 0);

        /* If copy machine is already marked failed, do nothing. */
        if (m0_cm_state_get(cm) == M0_CMS_FAIL)
                return;

        M0_LOG(M0_ERROR, "copy machine failure, state=%d rc=%d",
               m0_cm_state_get(cm), rc);
        m0_sm_fail(&cm->cm_mach, M0_CMS_FAIL, rc);
        m0_ha_cm_err_send(cm, rc);
        M0_LEAVE();
}

M0_INTERNAL void m0_cm_lock(struct m0_cm *cm)
{
        m0_sm_group_lock(&cm->cm_sm_group);
}

M0_INTERNAL void m0_cm_unlock(struct m0_cm *cm)
{
        m0_sm_group_unlock(&cm->cm_sm_group);
}

M0_INTERNAL int m0_cm_trylock(struct m0_cm *cm)
{
        return m0_mutex_trylock(&cm->cm_sm_group.s_lock);
}

M0_INTERNAL bool m0_cm_is_locked(const struct m0_cm *cm)
{
        return m0_mutex_is_locked(&cm->cm_sm_group.s_lock);
}

M0_INTERNAL enum m0_cm_state m0_cm_state_get(const struct m0_cm *cm)
{
        return (enum m0_cm_state)cm->cm_mach.sm_state;
}

M0_INTERNAL void m0_cm_state_set(struct m0_cm *cm, enum m0_cm_state state)
{
        M0_PRE(m0_cm_is_locked(cm));

        m0_sm_state_set(&cm->cm_mach, state);
        M0_LOG(M0_INFO, "CM:%s%" PRId64 ": %i",
               (char *)cm->cm_type->ct_stype.rst_name,
               cm->cm_id, m0_cm_state_get(cm));
}

static int cm_rc(struct m0_cm *cm)
{
        return cm->cm_mach.sm_rc;
}

M0_INTERNAL bool m0_cm_invariant(const struct m0_cm *cm)
{
        return cm != NULL && cm->cm_ops != NULL && cm->cm_type != NULL &&
               m0_sm_invariant(&cm->cm_mach) &&
               ergo(M0_IN(m0_cm_state_get(cm), (M0_CMS_IDLE, M0_CMS_PREPARE,
                                                M0_CMS_READY, M0_CMS_ACTIVE,
                                                M0_CMS_STOP)),
                    m0_reqh_service_invariant(&cm->cm_service));
}

M0_INTERNAL int m0_cm_setup(struct m0_cm *cm)
{
        int rc;

        M0_ENTRY("cm: %p", cm);
        M0_PRE(cm != NULL);
        M0_PRE(cm->cm_type != NULL);

        if (M0_FI_ENABLED("setup_failure"))
                return M0_ERR(-EINVAL);

        m0_cm_lock(cm);
        M0_PRE(m0_cm_state_get(cm) == M0_CMS_INIT);
        M0_PRE(m0_cm_invariant(cm));

        rc = cm->cm_ops->cmo_setup(cm);
        if (M0_FI_ENABLED("setup_failure_2"))
                rc = -EINVAL;
        if (rc == 0)
                m0_cm_state_set(cm, M0_CMS_IDLE);

        M0_POST(m0_cm_invariant(cm));
        m0_cm_unlock(cm);

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

/*
 * Establishes rpc connection with other remote replicas and allocate local
 * struct m0_cm_proxy for each of them.
 */
static int cm_replicas_connect(struct m0_cm *cm, struct m0_rpc_machine *rmach,
                               struct m0_reqh *reqh)
{
        struct m0_cm_ag_id          ag_id0;
        const char                 *lep;
        int                         rc = -ENOENT;
        struct m0_pools_common     *pc;
        struct m0_reqh_service_ctx *ctx;
        struct m0_conf_obj         *svc_obj;
        uint32_t                    proxy_cnt = 0;

        M0_PRE(cm != NULL && rmach != NULL && reqh != NULL);
        M0_PRE(m0_cm_is_locked(cm));

        pc = reqh->rh_pools;
        M0_SET0(&ag_id0);
        lep = m0_rpc_machine_ep(rmach);
        m0_tl_for(pools_common_svc_ctx, &pc->pc_svc_ctxs, ctx) {
                struct m0_cm_proxy *pxy;
                const char *dep = m0_rpc_link_end_point(&ctx->sc_rlink);

                M0_LOG(M0_DEBUG, "Connect %s dep %s type %d", lep, dep,
                                ctx->sc_type);
                if ((strcmp(lep, dep) == 0 ||
                    !cm->cm_ops->cmo_is_peer(cm, ctx)))
                        continue;
                rc = m0_conf_obj_find_lock(&pc->pc_confc->cc_cache,
                                           &ctx->sc_fid, &svc_obj);
                if (rc != 0)
                        return M0_ERR(rc);
                M0_ALLOC_PTR(pxy);
                if (pxy == NULL)
                        return M0_ERR(-ENOMEM);
                rc = m0_cm_proxy_init(pxy, proxy_cnt, &ag_id0, &ag_id0, dep);
                if (rc == 0) {
                        pxy->px_conn = &ctx->sc_rlink.rlk_conn;
                        pxy->px_session = &ctx->sc_rlink.rlk_sess;
                        m0_cm_proxy_add(cm, pxy);
                        m0_cm_proxy_event_handle_register(pxy, svc_obj);
                        M0_CNT_INC(proxy_cnt);
                        if (M0_IN(svc_obj->co_ha_state, (M0_NC_FAILED,
                                                         M0_NC_TRANSIENT))) {
                                m0_cm_proxy_lock(pxy);
                                pxy->px_status = M0_PX_FAILED;
                                pxy->px_is_done = true;
                                proxy_fail_tlist_add_tail(
                                           &pxy->px_cm->cm_failed_proxies, pxy);
                                m0_cm_proxy_unlock(pxy);
                        }
                        M0_LOG(M0_DEBUG, "Connected to %s", dep);
                }
        } m0_tl_endfor;

        rc = m0_bitmap_init(&cm->cm_proxy_update_map, cm->cm_proxy_nr);

        M0_LOG(M0_DEBUG, "Connected to its proxies from local ep %s", lep);
        return M0_RC(rc);
}

static void cm_replicas_destroy(struct m0_cm *cm)
{
        struct m0_cm_proxy *pxy;

        M0_PRE(m0_cm_is_locked(cm));

        m0_tl_for(proxy, &cm->cm_proxies, pxy) {
                m0_cm_proxy_del(cm, pxy);
                m0_cm_proxy_fini(pxy);
                m0_free(pxy);
        } m0_tl_endfor;
}

static int cm_pre_start_cleanup(struct m0_cm *cm)
{
        int rc;

        rc = m0_cm_complete(cm);
        if (rc == -EAGAIN)
                return rc;
        m0_cm_unlock(cm);
        m0_cm_stop(cm);
        m0_cm_lock(cm);
        return 0;
}

M0_INTERNAL struct m0_rpc_machine *m0_cm_rpc_machine_find(struct m0_reqh *reqh)
{
        return m0_reqh_rpc_mach_tlist_head(&reqh->rh_rpc_machines);
}

M0_INTERNAL int m0_cm_prepare(struct m0_cm *cm)
{
        struct m0_rpc_machine *rmach;
        struct m0_reqh        *reqh = cm->cm_service.rs_reqh;
        int                    rc;

        m0_cm_lock(cm);
        M0_PRE(M0_IN(m0_cm_state_get(cm), (M0_CMS_IDLE, M0_CMS_STOP,
                                           M0_CMS_FAIL)));

        m0_cm_state_set(cm, M0_CMS_PREPARE);
        cm->cm_done = false;
        cm->cm_nr_proxy_updated = 0;
        cm->cm_quiesce = false;
        cm->cm_abort   = false;
        cm->cm_epoch = m0_time_now();
        rmach = m0_cm_rpc_machine_find(reqh);
        rc = cm_replicas_connect(cm, rmach, reqh);
        if (rc == -ENOENT)
                rc = 0;
        if (rc == 0) {
                if (M0_FI_ENABLED("prepare_failure")) {
                        //m0_cm_unlock(cm);
                        rc = -EINVAL;
                } else
                        rc = cm->cm_ops->cmo_prepare(cm);
        }
        if (rc == 0) {
                m0_cm_ag_store_fom_start(cm);
                m0_cm_cp_pump_prepare(cm);
        }
        if (rc != 0) {
                m0_cm_fail(cm, rc);
                cm_replicas_destroy(cm);
                m0_cm_sw_update_complete(cm);
                cm->cm_done = true;
        }
        m0_cm_unlock(cm);

        return M0_RC(rc);
}

M0_INTERNAL int m0_cm_ready(struct m0_cm *cm)
{
        int rc;

        M0_ENTRY("cm: %p", cm);
        M0_PRE(cm != NULL);
        M0_PRE(cm->cm_type != NULL);

        m0_cm_lock(cm);
        M0_PRE(M0_IN(m0_cm_state_get(cm), (M0_CMS_PREPARE, M0_CMS_FAIL)));
        M0_PRE(m0_cm_invariant(cm));

        rc = cm_rc(cm);
        if (M0_FI_ENABLED("ready_failure"))
                rc = -EINVAL;

        if (rc == 0)
                m0_cm_ast_run_thread_init(cm);

        if (rc == 0) {
                m0_cm_state_set(cm, M0_CMS_READY);
                rc = m0_cm_sw_remote_update(cm);
        }

        if (rc != 0) {
                m0_cm_abort(cm, rc);
                rc = cm_pre_start_cleanup(cm);
        }
        m0_cm_unlock(cm);

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

M0_INTERNAL bool m0_cm_is_ready(struct m0_cm *cm)
{
        M0_PRE(m0_cm_is_locked(cm));
        return m0_cm_state_get(cm) == M0_CMS_READY;
}

M0_INTERNAL bool m0_cm_is_active(struct m0_cm *cm)
{
        M0_PRE(m0_cm_is_locked(cm));
        return m0_cm_state_get(cm) == M0_CMS_ACTIVE;
}

M0_INTERNAL int m0_cm_start(struct m0_cm *cm)
{
        int rc = cm_rc(cm);

        M0_ENTRY("cm: %p", cm);
        M0_PRE(cm != NULL);
        M0_PRE(cm->cm_type != NULL);

        m0_cm_lock(cm);
        M0_PRE(M0_IN(m0_cm_state_get(cm), (M0_CMS_READY, M0_CMS_FAIL)));
        M0_PRE(m0_cm_invariant(cm));

        m0_cm_proxies_sent_reset(cm);
        if (rc == 0)
                rc = cm->cm_ops->cmo_start(cm);
        if (M0_FI_ENABLED("start_failure"))
                rc = -EINVAL;
        /* Start pump FOM to create copy packets. */
        if (rc == 0) {
                m0_cm_cp_pump_start(cm);
                m0_cm_state_set(cm, M0_CMS_ACTIVE);
        }

        if (rc != 0) {
                m0_cm_abort(cm, rc);
                rc = cm_pre_start_cleanup(cm);
        }

        M0_POST(m0_cm_invariant(cm));
        m0_cm_unlock(cm);

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

M0_INTERNAL int m0_cm_proxies_fini(struct m0_cm *cm)
{
        struct m0_cm_proxy *pxy;
        int                 rc = 0;

        M0_ENTRY();
        M0_PRE(m0_cm_is_locked(cm));

        m0_tl_for(proxy, &cm->cm_proxies, pxy) {
                /* Check if proxy has completed. */
                M0_LOG(M0_DEBUG, "pxy %p (to %s), is_done %d",
                                  pxy, pxy->px_endpoint, !!pxy->px_is_done);
                if (!m0_cm_proxy_is_done(pxy)) {
                        rc = -EAGAIN;
                        M0_LOG(M0_DEBUG, "pxy %p is still active", pxy);
                        continue;
                }
                M0_LOG(M0_DEBUG, "Stop proxy. cm %p, pxy %p",cm,  pxy);
                m0_cm_proxy_del(cm, pxy);
                m0_cm_proxy_fini(pxy);
                m0_free(pxy);
        } m0_tl_endfor;

        return M0_RC(rc);
}

M0_INTERNAL int m0_cm_stop(struct m0_cm *cm)
{
        int rc = cm->cm_mach.sm_rc;

        M0_ENTRY("cm: %p", cm);
        M0_PRE(cm != NULL);

        m0_cm_lock(cm);
        M0_PRE(M0_IN(m0_cm_state_get(cm), (M0_CMS_PREPARE, M0_CMS_READY,
                                           M0_CMS_ACTIVE, M0_CMS_FAIL)));
        M0_PRE(m0_cm_invariant(cm));

        cm->cm_ops->cmo_stop(cm);
        /* In-case of failure (rc != 0) keep copy machine in failed state. */
        if (rc == 0)
                m0_cm_state_set(cm, M0_CMS_STOP);
        m0_bitmap_fini(&cm->cm_proxy_update_map);
        M0_POST(m0_cm_invariant(cm));
        m0_cm_unlock(cm);

        if (cm->cm_asts_run.car_run)
                m0_cm_ast_run_thread_fini(cm);

        cm->cm_reset = false;

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

M0_INTERNAL int m0_cm_module_init(void)
{

        M0_ENTRY();
        cmtypes_tlist_init(&cmtypes);
        m0_bob_type_tlist_init(&cmtypes_bob, &cmtypes_tl);
        m0_mutex_init(&cmtypes_mutex);

        M0_LEAVE();
        return 0;
}

M0_INTERNAL void m0_cm_module_fini(void)
{
        M0_ENTRY();
        cmtypes_tlist_fini(&cmtypes);
        m0_mutex_fini(&cmtypes_mutex);
        M0_LEAVE();
}

static uint64_t cm_id_generate(void)
{
        static uint64_t id = 0;
        return ++id;
}

M0_INTERNAL int m0_cm_init(struct m0_cm *cm, struct m0_cm_type *cm_type,
                           const struct m0_cm_ops *cm_ops)
{
        M0_ENTRY("cm_type: %p cm: %p", cm_type, cm);
        M0_PRE(cm != NULL && cm_type != NULL && cm_ops != NULL &&
               cmtypes_tlist_contains(&cmtypes, cm_type));

        if (M0_FI_ENABLED("init_failure"))
                return M0_ERR(-EINVAL);

        cm->cm_type = cm_type;
        cm->cm_ops = cm_ops;
        cm->cm_id = cm_id_generate();
        m0_sm_group_init(&cm->cm_sm_group);
        m0_sm_init(&cm->cm_mach, &cm_sm_conf, M0_CMS_INIT,
                   &cm->cm_sm_group);
        /*
         * We lock the copy machine here just to satisfy the
         * pre-condition of m0_cm_state_get and not to control
         * concurrency to m0_cm_init.
         */
        m0_cm_lock(cm);
        M0_ASSERT(m0_cm_state_get(cm) == M0_CMS_INIT);
        aggr_grps_in_tlist_init(&cm->cm_aggr_grps_in);
        aggr_grps_out_tlist_init(&cm->cm_aggr_grps_out);
        proxy_tlist_init(&cm->cm_proxies);
        proxy_fail_tlist_init(&cm->cm_failed_proxies);
        m0_mutex_init(&cm->cm_wait_mutex);
        m0_chan_init(&cm->cm_wait, &cm->cm_wait_mutex);
        m0_chan_init(&cm->cm_complete, &cm->cm_sm_group.s_lock);
        m0_chan_init(&cm->cm_proxy_init_wait, &cm->cm_sm_group.s_lock);

        M0_POST(m0_cm_invariant(cm));
        m0_cm_unlock(cm);

        M0_LEAVE();
        return 0;
}

M0_INTERNAL void m0_cm_fini(struct m0_cm *cm)
{
        M0_ENTRY("cm: %p", cm);
        M0_PRE(cm != NULL);

        m0_cm_lock(cm);
        M0_PRE(M0_IN(m0_cm_state_get(cm), (M0_CMS_INIT, M0_CMS_IDLE,
                                           M0_CMS_STOP, M0_CMS_FAIL)));
        M0_PRE(m0_cm_invariant(cm));

        cm->cm_ops->cmo_fini(cm);
        M0_LOG(M0_INFO, "CM: %s:%" PRId64 ": %i",
              (char *)cm->cm_type->ct_stype.rst_name,
              cm->cm_id, cm->cm_mach.sm_state);
        m0_cm_state_set(cm, M0_CMS_FINI);
        aggr_grps_in_tlist_fini(&cm->cm_aggr_grps_in);
        aggr_grps_out_tlist_fini(&cm->cm_aggr_grps_out);
        proxy_tlist_fini(&cm->cm_proxies);
        proxy_fail_tlist_fini(&cm->cm_failed_proxies);
        m0_sm_fini(&cm->cm_mach);
        m0_chan_fini_lock(&cm->cm_wait);
        m0_chan_fini(&cm->cm_complete);
        m0_chan_fini(&cm->cm_proxy_init_wait);
        m0_mutex_fini(&cm->cm_wait_mutex);
        m0_cm_unlock(cm);

        m0_sm_group_fini(&cm->cm_sm_group);

        M0_LEAVE();
}

M0_INTERNAL int m0_cm_type_register(struct m0_cm_type *cmtype)
{
        int     rc;

        M0_ENTRY("cmtype: %p", cmtype);
        M0_PRE(cmtype != NULL);
        M0_PRE(m0_reqh_service_type_find(cmtype->ct_stype.rst_name) == NULL);

        rc = m0_reqh_service_type_register(&cmtype->ct_stype);
        if (rc == 0) {
                m0_cm_type_bob_init(cmtype);
                m0_cm_sw_update_init(cmtype);
                m0_cm_ag_store_init(cmtype);
                m0_cm_cp_pump_init(cmtype);
                m0_mutex_lock(&cmtypes_mutex);
                cmtypes_tlink_init_at_tail(cmtype, &cmtypes);
                m0_mutex_unlock(&cmtypes_mutex);
                M0_ASSERT(cmtypes_tlink_is_in(cmtype));
        }

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

M0_INTERNAL void m0_cm_type_deregister(struct m0_cm_type *cmtype)
{
        M0_ENTRY("cmtype: %p", cmtype);
        M0_PRE(cmtype != NULL && m0_cm_type_bob_check(cmtype));
        M0_PRE(cmtypes_tlist_contains(&cmtypes, cmtype));

        m0_mutex_lock(&cmtypes_mutex);
        cmtypes_tlink_del_fini(cmtype);
        m0_mutex_unlock(&cmtypes_mutex);
        m0_cm_type_bob_fini(cmtype);
        m0_reqh_service_type_unregister(&cmtype->ct_stype);

        M0_LEAVE();
}

M0_INTERNAL int m0_cm_data_next(struct m0_cm *cm, struct m0_cm_cp *cp)
{
        int rc;

        M0_ENTRY("cm: %p cp: %p", cm, cp);
        M0_PRE(m0_cm_invariant(cm));
        M0_PRE(m0_cm_is_locked(cm));
        M0_PRE(cp != NULL);

        rc = cm->cm_ops->cmo_data_next(cm, cp);

        M0_POST(ergo(rc == 0,
                     cp_data_buf_tlist_length(&cp->c_buffers) == cp->c_buf_nr));

        M0_LEAVE("rc: %d", rc);
        return M0_RC(rc);
}

M0_INTERNAL bool m0_cm_has_more_data(const struct m0_cm *cm)
{
        return !m0_cm_cp_pump_is_complete(&cm->cm_cp_pump);
}

M0_INTERNAL struct m0_net_buffer *m0_cm_buffer_get(struct m0_net_buffer_pool
                                                   *bp, uint64_t colour)
{
        struct m0_net_buffer *buf;
        int                   i;

        M0_ASSERT(m0_net_buffer_pool_invariant(bp));
        buf = m0_net_buffer_pool_get(bp, colour);

        if (buf != NULL) {
                for (i = 0; i < bp->nbp_seg_nr; ++i)
                        memset(buf->nb_buffer.ov_buf[i], 0, bp->nbp_seg_size);
        }

        return buf;
}

M0_INTERNAL void m0_cm_buffer_put(struct m0_net_buffer_pool *bp,
                                  struct m0_net_buffer *buf,
                                  uint64_t colour)
{
        m0_net_buffer_pool_put(bp, buf, colour);
}

M0_INTERNAL void m0_cm_notify(struct m0_cm *cm)
{
        m0_chan_signal_lock(&cm->cm_wait);
}

M0_INTERNAL void m0_cm_wait(struct m0_cm *cm, struct m0_fom *fom)
{
        m0_mutex_lock(&cm->cm_wait_mutex);
        m0_fom_wait_on(fom, &cm->cm_wait, &fom->fo_cb);
        m0_mutex_unlock(&cm->cm_wait_mutex);
}

M0_INTERNAL void m0_cm_wait_cancel(struct m0_cm *cm, struct m0_fom *fom)
{
        m0_mutex_lock(&cm->cm_wait_mutex);
        m0_fom_callback_cancel(&fom->fo_cb);
        m0_mutex_unlock(&cm->cm_wait_mutex);
}

M0_INTERNAL int m0_cm_complete(struct m0_cm *cm)
{
        int rc;

        M0_ENTRY("cm %p, done %d", cm, (int)cm->cm_done);
        M0_PRE(m0_cm_is_locked(cm));

        if (!cm->cm_done) {
                if (cm->cm_quiesce || cm->cm_abort)
                        m0_cm_ag_store_fini(&cm->cm_ag_store);
                else
                        m0_cm_ag_store_complete(&cm->cm_ag_store);

                cm->cm_done = true;
                m0_cm_sw_remote_update(cm);
        }
        /*
         * Finalising proxies is a blocking operation because we wait until
         * the remote replica, corresponding to the copy machine proxy is
         * complete. Thus we check for -EAGAIN if proxy is not yet ready to
         * be finalised.
         */
        rc = m0_cm_proxies_fini(cm);
        if (rc == -EAGAIN)
                return M0_RC(rc);
        if (!m0_cm_ag_store_is_complete(&cm->cm_ag_store))
                return M0_RC(-EAGAIN);

        m0_cm_notify(cm);

        return M0_RC(rc);
}

M0_INTERNAL void m0_cm_complete_notify(struct m0_cm *cm)
{
        M0_ASSERT(m0_cm_is_locked(cm));
        M0_ENTRY("Notifying cm %p id=%"PRIu64, cm, cm->cm_id);

        m0_chan_signal(&cm->cm_complete);
        M0_LEAVE();
}

M0_INTERNAL void m0_cm_proxies_init_wait(struct m0_cm *cm, struct m0_fom *fom)
{
        M0_PRE(m0_cm_is_locked(cm));
        m0_fom_wait_on(fom, &cm->cm_proxy_init_wait, &fom->fo_cb);
}

M0_INTERNAL void m0_cm_frozen_ag_cleanup(struct m0_cm *cm,
                                         struct m0_cm_proxy *proxy)
{
        struct m0_cm_aggr_group *ag = NULL;
        bool                     cleanup;
        M0_ENTRY();

        M0_PRE(m0_cm_is_locked(cm));

        m0_tlist_for(&aggr_grps_in_tl, &cm->cm_aggr_grps_in, ag) {
                m0_cm_ag_lock(ag);
                cleanup = ag->cag_ops->cago_is_frozen_on(ag, proxy) &&
                          m0_cm_ag_can_fini(ag);
                m0_cm_ag_unlock(ag);
                ID_LOG("cm_aggr_grps_in", &ag->cag_id);
                if (cleanup) {
                        M0_ASSERT(ag->cag_fini_ast.sa_next == NULL);
                        M0_LOG(M0_DEBUG, "finalizing frozen aggregation group ["M0_AG_F"]",
                                         M0_AG_P(&ag->cag_id));
                        ag->cag_ops->cago_fini(ag);
                }
        } m0_tlist_endfor;
}

M0_INTERNAL void m0_cm_proxy_failed_cleanup(struct m0_cm *cm)
{
        struct m0_cm_proxy *pxy;

        m0_tl_for(proxy_fail, &cm->cm_failed_proxies, pxy) {
                m0_cm_frozen_ag_cleanup(cm, pxy);
        } m0_tl_endfor;
}

M0_INTERNAL void m0_cm_abort(struct m0_cm *cm, int rc)
{
        M0_PRE(m0_cm_is_locked(cm));
        if (M0_IN(m0_cm_state_get(cm), (M0_CMS_PREPARE, M0_CMS_READY,
                                        M0_CMS_ACTIVE))) {
                cm->cm_abort = true;
                if (rc < 0)
                        m0_cm_fail(cm, rc);
        }
        m0_cm_notify(cm);
}

M0_INTERNAL bool m0_cm_is_dirty(struct m0_cm *cm)
{
        return cm->cm_abort || cm->cm_quiesce || m0_cm_state_get(cm) == M0_CMS_FAIL;
}

M0_INTERNAL bool m0_cm_proxies_updated(struct m0_cm *cm)
{
        M0_PRE(m0_cm_is_locked(cm));

        return cm->cm_nr_proxy_updated == cm->cm_proxy_nr;
}

static void cm_ast_run_thread(struct m0_cm *cm)
{
        while (cm->cm_asts_run.car_run) {
                m0_chan_wait(&cm->cm_sm_group.s_clink);
                m0_sm_group_lock(&cm->cm_sm_group);
                m0_sm_asts_run(&cm->cm_sm_group);
                m0_sm_group_unlock(&cm->cm_sm_group);
        }
}

M0_INTERNAL int m0_cm_ast_run_thread_init(struct m0_cm *cm)
{
        M0_SET0(&cm->cm_asts_run);
        cm->cm_asts_run.car_run = true;
        return M0_THREAD_INIT(&cm->cm_asts_run.car_th, struct m0_cm *,
                              NULL, &cm_ast_run_thread, cm, "cm_ast_run_thr");
}

M0_INTERNAL void m0_cm_ast_run_thread_fini(struct m0_cm *cm)
{
        cm->cm_asts_run.car_run = false;
        m0_clink_signal(&cm->cm_sm_group.s_clink);
        m0_thread_join(&cm->cm_asts_run.car_th);
}


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