rconfc.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_CONF
#include "lib/trace.h"

#include "lib/tlist.h"
#include "lib/mutex.h"
#include "lib/memory.h"           /* M0_ALLOC_PTR, m0_free */
#include "lib/errno.h"
#include "lib/string.h"
#include "lib/buf.h"              /* m0_buf_strdup */
#include "lib/finject.h"          /* M0_FI_ENABLED */
#include "motr/magic.h"
#include "rm/rm.h"
#include "rm/rm_service.h"        /* m0_rm_svc_rwlock_get */
#include "rpc/conn.h"             /* m0_rpc_conn_sessions_cancel */
#include "rpc/rpclib.h"           /* m0_rpc_client_connect */
#include "conf/cache.h"
#include "conf/obj_ops.h"         /* m0_conf_obj_find */
#include "conf/confc.h"
#include "conf/helpers.h"         /* m0_conf_ha_state_update */
#include "conf/rconfc.h"
#include "conf/rconfc_internal.h" /* rlock_ctx, rconfc_link,
                                   * ver_item, ver_accm */
#include "conf/rconfc_link_fom.h" /* rconfc_herd_link__on_death_cb */
#include "ha/entrypoint.h"        /* m0_ha_entrypoint_client */
#include "ha/ha.h"                /* m0_ha */
#include "ha/epoch.h"             /* m0_ha_client_add, m0_ha_client_del */
#include "module/instance.h"      /* m0_get */

static void rconfc_start(struct m0_rconfc *rconfc);
static void rconfc_stop_internal(struct m0_rconfc *rconfc);
static inline uint32_t rconfc_state(const struct m0_rconfc *rconfc);

static bool rconfc_gate_check(struct m0_confc *confc);
static int  rconfc_gate_skip(struct m0_confc *confc);
static bool rconfc_gate_drain(struct m0_clink *clink);
static bool ha_clink_cb(struct m0_clink *clink);

struct m0_confc_gate_ops m0_rconfc_gate_ops = {
        .go_check = rconfc_gate_check,
        .go_skip  = rconfc_gate_skip,
        .go_drain = rconfc_gate_drain,
};

static void rconfc_read_lock_get(struct m0_rconfc *rconfc);
static void rconfc_read_lock_complete(struct m0_rm_incoming *in, int32_t rc);
static void rconfc_read_lock_conflict(struct m0_rm_incoming *in);

struct m0_rm_incoming_ops m0_rconfc_ri_ops = {
        .rio_complete = rconfc_read_lock_complete,
        .rio_conflict = rconfc_read_lock_conflict,
};

static struct m0_sm_state_descr rconfc_states[] = {
        [M0_RCS_INIT] = {
                .sd_flags     = M0_SDF_INITIAL,
                .sd_name      = "M0_RCS_INIT",
                .sd_allowed   = M0_BITS(M0_RCS_FAILURE,
                                        M0_RCS_ENTRYPOINT_WAIT,
                                        M0_RCS_STOPPING)
        },
        [M0_RCS_ENTRYPOINT_WAIT] = {
                .sd_name      = "M0_RCS_ENTRYPOINT_WAIT",
                .sd_allowed   = M0_BITS(M0_RCS_ENTRYPOINT_WAIT,
                                        M0_RCS_ENTRYPOINT_CONSUME,
                                        M0_RCS_FAILURE),
        },
        [M0_RCS_ENTRYPOINT_CONSUME] = {
                .sd_name      = "M0_RCS_ENTRYPOINT_CONSUME",
                .sd_allowed   = M0_BITS(M0_RCS_ENTRYPOINT_WAIT,
                                        M0_RCS_CREDITOR_SETUP, M0_RCS_FAILURE),
        },
        [M0_RCS_CREDITOR_SETUP] = {
                .sd_name      = "M0_RCS_CREDITOR_SETUP",
                .sd_allowed   = M0_BITS(M0_RCS_GET_RLOCK,
                                        M0_RCS_ENTRYPOINT_WAIT)
        },
        [M0_RCS_GET_RLOCK] = {
                .sd_name      = "M0_RCS_GET_RLOCK",
                .sd_allowed   = M0_BITS(M0_RCS_VERSION_ELECT, M0_RCS_GET_RLOCK,
                                        M0_RCS_ENTRYPOINT_WAIT, M0_RCS_FAILURE),
        },
        [M0_RCS_VERSION_ELECT] = {
                .sd_name      = "M0_RCS_VERSION_ELECT",
                .sd_allowed   = M0_BITS(M0_RCS_IDLE, M0_RCS_ENTRYPOINT_WAIT,
                                        M0_RCS_STOPPING, M0_RCS_RLOCK_CONFLICT,
                                        M0_RCS_FAILURE),
        },
        [M0_RCS_IDLE] = {
                .sd_name      = "M0_RCS_IDLE",
                .sd_allowed   = M0_BITS(M0_RCS_FAILURE, M0_RCS_STOPPING,
                                        M0_RCS_RLOCK_CONFLICT),
        },
        [M0_RCS_RLOCK_CONFLICT] = {
                .sd_name      = "M0_RCS_RLOCK_CONFLICT",
                .sd_allowed   = M0_BITS(M0_RCS_CONDUCTOR_DRAIN),
        },
        [M0_RCS_CONDUCTOR_DRAIN] = {
                .sd_name      = "M0_RCS_CONDUCTOR_DRAIN",
                .sd_allowed   = M0_BITS(M0_RCS_CONDUCTOR_DISCONNECT,
                                        M0_RCS_FAILURE, M0_RCS_FINAL),
        },
        [M0_RCS_CONDUCTOR_DISCONNECT] = {
                .sd_name      = "M0_RCS_CONDUCTOR_DISCONNECT",
                .sd_allowed   = M0_BITS(M0_RCS_ENTRYPOINT_WAIT, M0_RCS_FAILURE),
        },
        [M0_RCS_STOPPING] = {
                .sd_name      = "M0_RCS_STOPPING",
                .sd_allowed   = M0_BITS(M0_RCS_CONDUCTOR_DRAIN),
        },
        [M0_RCS_FAILURE] = {
                .sd_flags     = M0_SDF_FAILURE,
                .sd_name      = "M0_RCS_FAILURE",
                .sd_allowed   = M0_BITS(M0_RCS_STOPPING)
        },
        [M0_RCS_FINAL] = {
                .sd_flags     = M0_SDF_TERMINAL,
                .sd_name      = "M0_RCS_FINAL"
        }
};

static const struct m0_sm_conf rconfc_sm_conf = {
        .scf_name      = "Rconfc",
        .scf_nr_states = ARRAY_SIZE(rconfc_states),
        .scf_state     = rconfc_states
};

/***************************************
 * List Definitions
 ***************************************/

M0_TL_DESCR_DEFINE(rcnf_herd, "rconfc's working  confc list", M0_INTERNAL,
                   struct rconfc_link, rl_herd, rl_magic,
                   M0_RCONFC_LINK_MAGIC, M0_RCONFC_HERD_HEAD_MAGIC
        );
M0_TL_DEFINE(rcnf_herd, M0_INTERNAL, struct rconfc_link);

M0_TL_DESCR_DEFINE(rcnf_active, "rconfc's active confc list", M0_INTERNAL,
                   struct rconfc_link, rl_active, rl_magic,
                   M0_RCONFC_LINK_MAGIC, M0_RCONFC_ACTIVE_HEAD_MAGIC
        );
M0_TL_DEFINE(rcnf_active, M0_INTERNAL, struct rconfc_link);

/* -------------- Thread for async full conf loading -------------- */

static void rconfc_load_ast_thread(struct rconfc_load_ctx *rx)
{
        while (rx->rx_ast.run) {
                m0_chan_wait(&rx->rx_grp.s_clink);
                m0_sm_group_lock(&rx->rx_grp);
                m0_sm_asts_run(&rx->rx_grp);
                m0_sm_group_unlock(&rx->rx_grp);
        }
}

static int rconfc_load_ast_thread_init(struct rconfc_load_ctx *rx)
{
        M0_SET0(&rx->rx_grp);
        M0_SET0(&rx->rx_ast);
        m0_sm_group_init(&rx->rx_grp);
        rx->rx_ast.run = true;
        return M0_THREAD_INIT(&rx->rx_ast.thread, struct rconfc_load_ctx *,
                              NULL, &rconfc_load_ast_thread, rx, "rx_ast_thr");
}

static void rconfc_load_ast_thread_fini(struct rconfc_load_ctx *rx)
{
        rx->rx_ast.run = false;
        m0_clink_signal(&rx->rx_grp.s_clink);
        m0_thread_join(&rx->rx_ast.thread);
        m0_sm_group_fini(&rx->rx_grp);
}

static void rconfc_idle(struct m0_rconfc *rconfc);
static void rconfc_fail(struct m0_rconfc *rconfc, int rc);

static void rconfc_conf_load_fini(struct m0_sm_group *grp,
                                  struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        M0_ENTRY("rconfc = %p", rconfc);
        m0_confc_gate_ops_set(&rconfc->rc_confc, &rconfc->rc_gops);
        if (rconfc->rc_rx.rx_rc == 0) {
                rconfc_idle(rconfc); /* unlock reading gate */
        } else {
                rconfc_fail(rconfc, rconfc->rc_rx.rx_rc);
        }
        rconfc_load_ast_thread_fini(&rconfc->rc_rx);
        M0_LEAVE();
}

static void rconfc_conf_full_load(struct m0_sm_group *grp,
                                  struct m0_sm_ast   *ast)
{
        int rc;
        struct m0_conf_root   *root = NULL;
        struct m0_rconfc      *rconfc = ast->sa_datum;
        struct m0_confc       *confc = &rconfc->rc_confc;
        struct m0_conf_cache  *cache = &confc->cc_cache;

        M0_ENTRY("rconfc = %p", rconfc);
        rc = m0_conf_obj_find_lock(cache, &M0_CONF_ROOT_FID, &confc->cc_root) ?:
             m0_confc_root_open(confc, &root) ?:
             m0_conf_full_load(root);
        if (root != NULL)
                m0_confc_close(&root->rt_obj);
        /*
         * The configuration might be loaded. Now we need to invoke
         * rconfc_ha_restore() to re-associate the kept clinks with their
         * objects in cache.
         */
        rconfc->rc_rx.rx_rc = rc;
        rconfc->rc_load_fini_ast.sa_datum = rconfc;
        rconfc->rc_load_fini_ast.sa_cb = rconfc_conf_load_fini;
        /* The call to execute in context of group rconfc initialised with */
        m0_sm_ast_post(rconfc->rc_sm.sm_grp, &rconfc->rc_load_fini_ast);
        M0_LEAVE();
}

/***************************************
 * Helpers
 ***************************************/

/* -------------- Confc Helpers -------------- */

static uint64_t _confc_ver_read(const struct m0_confc *confc)
{
        return confc->cc_cache.ca_ver;
}

static const char *_confc_remote_addr_read(const struct m0_confc *confc)
{
        return m0_confc_is_online(confc) ?
                m0_rpc_conn_addr(m0_confc2conn((struct m0_confc *)confc)) :
                NULL;
}

static int _confc_cache_clean(struct m0_confc *confc)
{
        struct m0_conf_cache *cache = &confc->cc_cache;

        M0_ENTRY();
        m0_conf_cache_clean(cache, NULL);
        /* Clear version to prevent version mismatch error after reelection */
        cache->ca_ver = M0_CONF_VER_UNKNOWN;
        return M0_RC(m0_conf_obj_find(cache, &M0_CONF_ROOT_FID,
                                      &confc->cc_root));
}

static int _confc_cache_clean_lock(struct m0_confc *confc)
{
        int rc;

        M0_ENTRY();
        m0_conf_cache_lock(&confc->cc_cache);
        rc = _confc_cache_clean(confc);
        m0_conf_cache_unlock(&confc->cc_cache);
        return M0_RC(rc);
}

/* ------------------- Phony confc ------------------ */

static int _confc_phony_init(struct m0_confc *confc)
{
        void *fake_ptr = (void *)1;
        int   rc;

        M0_ENTRY();
        /*
         * The confc never to do real conf reading, so no real group is
         * required, as well as no real RPC machine to be used because no RPC
         * connection to be established. Fake pointers are just to prevent
         * assertions from firing in m0_confc_init().
         */
        rc = m0_confc_init(confc, fake_ptr, NULL, fake_ptr, NULL);
        if (rc != 0)
                return M0_ERR(rc);
        rc = m0_ha_client_add(confc);
        if (rc != 0)
                m0_confc_fini(confc);
        return M0_RC(rc);
}

static void _confc_phony_fini(struct m0_confc *phony)
{
        M0_ENTRY();
        /* Cache contains only root object */
        M0_PRE(m0_conf_cache_tlist_length(
                                &phony->cc_cache.ca_registry) == 1);
        m0_ha_client_del(phony);
        m0_confc_fini(phony);
        M0_SET0(phony);
        M0_LEAVE();
}

static int _confc_phony_cache_append(struct m0_confc      *confc,
                                     const struct m0_fid  *fid)
{
        struct m0_conf_cache *cache = &confc->cc_cache;
        struct m0_conf_obj   *obj;
        int                   rc = 0;

        M0_ENTRY();
        m0_conf_cache_lock(cache);
        obj = m0_conf_cache_lookup(cache, fid);
        if (obj != NULL)
                goto out;
        rc = -ENOMEM;
        obj = m0_conf_obj_create(fid, cache);
        if (obj == NULL)
                goto out;
        /*
         * fake it be an already normally read object to comply with HA
         * subscription mechanisms (see ha_state_accept())...
         */
        obj->co_status = M0_CS_READY;
        /*
         * ...and fake it be pinned, as further discharging will rely on
         * that pin while iterating the cache
         */
        obj->co_nrefs  = 1;
        rc = m0_conf_cache_add(cache, obj);
        if (rc != 0) {
                obj->co_nrefs  = 0;
                m0_conf_obj_delete(obj);
        }
out:
        m0_conf_cache_unlock(cache);
        return M0_RC(rc);
}

static void _confc_phony_cache_remove(struct m0_confc     *confc,
                                      const struct m0_fid *fid)
{
        struct m0_conf_obj   *obj;
        struct m0_conf_cache *cache = &confc->cc_cache;

        M0_ENTRY();
        m0_conf_cache_lock(cache);
        obj = m0_conf_cache_lookup(&confc->cc_cache, fid);
        M0_ASSERT(!m0_conf_obj_is_stub(obj));
        obj->co_status = M0_CS_MISSING;
        obj->co_nrefs  = 0;
        m0_conf_cache_del(cache, obj);
        m0_conf_cache_unlock(cache);
        M0_LEAVE();
}

/* -------------- Read lock context ----------------- */

static int rlock_ctx_read_domain_init(struct rlock_ctx *rlx)
{
        return m0_rwlockable_domain_type_init(&rlx->rlc_dom, &rlx->rlc_rt);
}

static void rlock_ctx_read_domain_fini(struct rlock_ctx *rlx)
{
        m0_rwlockable_domain_type_fini(&rlx->rlc_dom, &rlx->rlc_rt);
}

static struct m0_rconfc *rlock_ctx_incoming_to_rconfc(struct m0_rm_incoming *in)
{
        struct rlock_ctx *rlx;

        rlx = container_of(in, struct rlock_ctx, rlc_req);
        return rlx->rlc_parent;
}

static bool rlock_ctx_is_online(struct rlock_ctx *rlx)
{
        return rlx->rlc_rm_addr != NULL;
}

static void rlock_ctx_disconnect(struct rlock_ctx *rlx)
{
        int rc;

        M0_ENTRY("rconfc = %p, rlx = %p", rlx->rlc_parent, rlx);
        M0_PRE(rlock_ctx_is_online(rlx));
        rc = m0_rpc_session_destroy(&rlx->rlc_sess, m0_rpc__down_timeout());
        if (rc != 0)
                M0_LOG(M0_ERROR, "Failed to destroy rlock session");
        rc = m0_rpc_conn_destroy(&rlx->rlc_conn, m0_rpc__down_timeout());
        if (rc != 0)
                M0_LOG(M0_ERROR, "Failed to destroy rlock connection");
        m0_free(rlx->rlc_rm_addr);
        rlx->rlc_rm_addr = NULL;
        M0_LEAVE();
}

static int rlock_ctx_connect(struct rlock_ctx *rlx, const char *ep)
{
        enum {
                MAX_RPCS_IN_FLIGHT        = 15,
                RLOCK_CTX_TIMEOUT_DEFAULT = 3ULL * M0_TIME_ONE_SECOND,
        };

        int       rc;
        m0_time_t deadline = rlx->rlc_timeout == 0 ?
                m0_time_from_now(0, RLOCK_CTX_TIMEOUT_DEFAULT) :
                m0_time_from_now(0, rlx->rlc_timeout);

        M0_ENTRY("rconfc = %p, rlx = %p, ep = %s", rlx->rlc_parent, rlx, ep);
        M0_PRE(!rlock_ctx_is_online(rlx));
        M0_PRE(ep != NULL);
        rlx->rlc_rm_addr = m0_strdup(ep);
        if (rlx->rlc_rm_addr == NULL)
                return M0_ERR(-ENOMEM);
        if (M0_FI_ENABLED("rm_conn_failed"))
                rc = M0_ERR(-ECONNREFUSED);
        else
                rc = m0_rpc_client_connect(&rlx->rlc_conn, &rlx->rlc_sess,
                                           rlx->rlc_rmach, rlx->rlc_rm_addr,
                                           NULL, MAX_RPCS_IN_FLIGHT, deadline);
        if (rc != 0) {
                m0_free(rlx->rlc_rm_addr);
                rlx->rlc_rm_addr = NULL;
        }
        rlx->rlc_online = (rc == 0);
        /*
         * It might happen that rconfc was put to M0_RCS_FAILURE state before
         * connection was established, due to reasons unrelated to RM.
         */
        return rconfc_state(rlx->rlc_parent) == M0_RCS_FAILURE ?
                M0_ERR(-ESTALE) : M0_RC(rc);
}

static int rlock_ctx_create(struct m0_rconfc       *parent,
                            struct m0_rpc_machine  *rmach,
                            struct rlock_ctx      **out)
{
        struct rlock_ctx *rlx;
        int               rc;

        M0_ENTRY("rconfc = %p, out = %p", parent, out);
        M0_PRE(out != NULL);
        M0_PRE(rmach != NULL);

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

        rlx->rlc_parent = parent;
        rlx->rlc_rmach = rmach;
        rc = rlock_ctx_read_domain_init(rlx);
        if (rc != 0)
                return M0_ERR(rc);
        m0_rw_lockable_init(&rlx->rlc_rwlock, &M0_RWLOCK_FID, &rlx->rlc_dom);
        m0_fid_tgenerate(&rlx->rlc_owner_fid, M0_RM_OWNER_FT);
        m0_rm_rwlock_owner_init(&rlx->rlc_owner, &rlx->rlc_owner_fid,
                                &rlx->rlc_rwlock, NULL);
        rlx->rlc_rm_addr = NULL;
        *out = rlx;
        return M0_RC(0);
}

static void rlock_ctx_destroy(struct rlock_ctx *rlx)
{
        M0_ENTRY("rconfc = %p, rlx = %p", rlx->rlc_parent, rlx);
        M0_PRE(!rlock_ctx_is_online(rlx));
        m0_rw_lockable_fini(&rlx->rlc_rwlock);
        rlock_ctx_read_domain_fini(rlx);
        m0_free(rlx);
        M0_LEAVE();
}

static enum m0_rm_owner_state
rlock_ctx_creditor_state(struct rlock_ctx *rlx)
{
        return rlx->rlc_owner.ro_sm.sm_state;
}

static int rlock_ctx_creditor_setup(struct rlock_ctx *rlx,
                                    const char       *ep)
{
        struct m0_rm_owner  *owner = &rlx->rlc_owner;
        struct m0_rm_remote *creditor = &rlx->rlc_creditor;
        struct m0_fid       *fid = &rlx->rlc_rm_fid;
        struct m0_conf_obj  *obj;
        int                  rc;

        M0_ENTRY("rconfc = %p, rlx = %p, ep = %s", rlx->rlc_parent, rlx, ep);
        M0_PRE(M0_IN(rlock_ctx_creditor_state(rlx),
                     (ROS_ACTIVE, ROS_DEAD_CREDITOR)));
        rc = rlock_ctx_connect(rlx, ep);
        if (rc != 0) {
                _confc_phony_cache_remove(&rlx->rlc_parent->rc_phony, fid);
                return M0_ERR(rc);
        }
        m0_rm_remote_init(creditor, owner->ro_resource);
        creditor->rem_session = &rlx->rlc_sess;
        m0_rm_owner_creditor_reset(owner, creditor);
        rlx->rlc_rm_fid       = *fid;
        /*
         * Subscribe for HA state changes of creditor.
         * Unsubscription is automatically done in m0_rm_remote_fini().
         */
        obj = m0_conf_cache_lookup(&rlx->rlc_parent->rc_phony.cc_cache, fid);
        m0_clink_add_lock(&obj->co_ha_chan, &creditor->rem_tracker.rht_clink);

        return M0_RC(0);
}

static void rlock_ctx_creditor_unset(struct rlock_ctx *rlx)
{
        M0_ENTRY("rconfc = %p, rlx = %p", rlx->rlc_parent, rlx);
        rlock_ctx_disconnect(rlx);
        m0_rm_remote_fini(&rlx->rlc_creditor);
        M0_SET0(&rlx->rlc_creditor);
        rlx->rlc_owner.ro_creditor = NULL;
        _confc_phony_cache_remove(&rlx->rlc_parent->rc_phony, &rlx->rlc_rm_fid);
        M0_LEAVE();
}

static void rlock_ctx_owner_windup(struct rlock_ctx *rlx)
{
        int rc;

        M0_ENTRY("rconfc = %p, rlx = %p", rlx->rlc_parent, rlx);
        m0_rm_owner_windup(&rlx->rlc_owner);
        rc = m0_rm_owner_timedwait(&rlx->rlc_owner,
                                   M0_BITS(ROS_FINAL, ROS_INSOLVENT),
                                   M0_TIME_NEVER);
        M0_ASSERT(rc == 0);
        M0_LEAVE();
}

/* -------------- Quorum calculation context ----------------- */

static void ver_accm_init(struct ver_accm *va, int total)
{
        M0_ENTRY();
        M0_PRE(total <= VERSION_ITEMS_TOTAL_MAX);
        M0_SET0(va);
        va->va_total = total;
        M0_LEAVE();
}

/* -------------- Rconfc Helpers -------------- */

static inline uint32_t rconfc_state(const struct m0_rconfc *rconfc)
{
        return rconfc->rc_sm.sm_state;
}

static uint32_t rconfc_confd_count(const char **confd_addr)
{
        uint32_t count;

        M0_PRE(confd_addr != NULL);
        count = 0;
        while (*confd_addr++ != NULL)
                ++count;
        return count;
}

static bool rconfc_confd_addr_are_all_unique(const char **confd_addr)
{
        uint32_t count = rconfc_confd_count(confd_addr);
        return m0_forall(i, count,
                         m0_forall(j, count,
                                   i == j ? true :
                                   !m0_streq(confd_addr[i],
                                             confd_addr[j])));
}

/***************************************
 * Rconfc private part
 ***************************************/

static bool rconfc_is_locked(struct m0_rconfc *rconfc)
{
        return m0_sm_group_is_locked(rconfc->rc_sm.sm_grp);
}

static void rconfc_read_lock_put(struct m0_rconfc *rconfc)
{
        struct rlock_ctx *rlx;

        M0_ENTRY("rconfc = %p", rconfc);
        rlx = rconfc->rc_rlock_ctx;
        if (!M0_IS0(&rlx->rlc_req)) {
            if (!M0_IN(rlx->rlc_req.rin_sm.sm_state,
                       (RI_INITIALISED, RI_FAILURE, RI_RELEASED)))
                        m0_rm_credit_put(&rlx->rlc_req);
                m0_rm_rwlock_req_fini(&rlx->rlc_req);
        }
        M0_SET0(&rlx->rlc_req);
        M0_LEAVE();
}

static void rconfc__ast_post(struct m0_rconfc  *rconfc,
                             void              *datum,
                             void             (*cb)(struct m0_sm_group *,
                                                    struct m0_sm_ast *))
{
        struct m0_sm_ast *ast = &rconfc->rc_ast;

        M0_LOG(M0_DEBUG, "posting AST=%p old %p:%14p:%p new %p:%14p",
                          ast, ast->sa_cb, ast->sa_datum, ast->sa_next,
                          cb, datum);

        ast->sa_cb = cb;
        ast->sa_datum = datum;
        m0_sm_ast_post(rconfc->rc_sm.sm_grp, ast);
}

static void rconfc_link_ast_post(struct rconfc_link  *lnk,
                                 void                *datum,
                                 void                (*cb)(struct m0_sm_group *,
                                                           struct m0_sm_ast *))
{
        struct m0_sm_ast *ast = &lnk->rl_link_ast;
        struct m0_rconfc *rconfc = lnk->rl_rconfc;

        ast->sa_cb = cb;
        ast->sa_datum = datum;
        m0_sm_ast_post(rconfc->rc_sm.sm_grp, ast);
}

static void rconfc_ast_post(struct m0_rconfc  *rconfc,
                            void             (*cb)(struct m0_sm_group *,
                                                   struct m0_sm_ast *))
{
        rconfc__ast_post(rconfc, rconfc, cb);
}

static void rconfc_state_set(struct m0_rconfc *rconfc, int state)
{
        M0_LOG(M0_DEBUG, "rconfc: %p, %s -> %s", rconfc,
               m0_sm_state_name(&rconfc->rc_sm, rconfc_state(rconfc)),
               m0_sm_state_name(&rconfc->rc_sm, state));

        m0_sm_state_set(&rconfc->rc_sm, state);
}

static void rconfc_fail(struct m0_rconfc *rconfc, int rc)
{
        M0_ENTRY("rconfc = %p, rc = %d", rconfc, rc);
        M0_PRE(rconfc_is_locked(rconfc));
        M0_LOG(M0_ERROR, "rconfc: %p, state %s failed with %d", rconfc,
               m0_sm_state_name(&rconfc->rc_sm, rconfc_state(rconfc)), rc);
        /*
         * Put read lock on failure, because this rconfc can prevent remote
         * write lock requests from completion.
         */
        rconfc_read_lock_put(rconfc);
        rconfc->rc_sm_state_on_abort = rconfc_state(rconfc);
        m0_sm_fail(&rconfc->rc_sm, M0_RCS_FAILURE, rc);
        if (rconfc->rc_stopping)
                rconfc_stop_internal(rconfc);
        if (rconfc->rc_fatal_cb != NULL)
                rconfc->rc_fatal_cb(rconfc);
        M0_LEAVE();
}

static void _failure_ast_cb(struct m0_sm_group *grp M0_UNUSED,
                            struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;
        struct rlock_ctx *rlx    = rconfc->rc_rlock_ctx;

        /*
         * RPC connection to RM may be lost, or there is no remote RM
         * service anymore to respond to read lock request, so need to
         * handle this to prevent read lock context from further
         * communication attempts
         */
        M0_ASSERT(rlock_ctx_is_online(rlx));
        if (!M0_FI_ENABLED("rlock_req_failed"))
                rlock_ctx_creditor_unset(rlx);
        rconfc_fail(rconfc, rconfc->rc_datum);
}

static void rconfc_fail_ast(struct m0_rconfc *rconfc, int rc)
{
        M0_ENTRY();
        rconfc->rc_datum = rc;
        rconfc_ast_post(rconfc, _failure_ast_cb);
        M0_LEAVE();
}

M0_INTERNAL bool m0_rconfc_reading_is_allowed(const struct m0_rconfc *rconfc)
{
        M0_PRE(rconfc != NULL);
        return rconfc_state(rconfc) == M0_RCS_IDLE;
}

static bool rconfc_quorum_is_reached(struct m0_rconfc *rconfc)
{
        M0_PRE(rconfc_is_locked(rconfc));
        return rconfc->rc_ver != M0_CONF_VER_UNKNOWN;
}

static void rconfc_active_all_unlink(struct m0_rconfc *rconfc)
{
        struct rconfc_link *lnk;

        /* unlink all active entries */
        m0_tl_teardown(rcnf_active, &rconfc->rc_active, lnk) {
                rcnf_active_tlink_fini(lnk);
        }
}

static void rconfc_herd_link_subscribe(struct rconfc_link *lnk)
{
        struct m0_conf_obj   *obj;
        struct m0_conf_cache *cache = &lnk->rl_rconfc->rc_phony.cc_cache;

        M0_PRE(lnk->rl_state == CONFC_IDLE);
        obj = m0_conf_cache_lookup(cache, &lnk->rl_confd_fid);
        M0_ASSERT(obj != NULL);
        m0_clink_add_lock(&obj->co_ha_chan, &lnk->rl_ha_clink);
}

static void rconfc_link_fini_ast(struct m0_sm_group *grp,
                                 struct m0_sm_ast   *ast);
static void rconfc_herd_link_unsubscribe(struct rconfc_link *lnk)
{
        if (lnk->rl_ha_clink.cl_chan != NULL)
                m0_clink_del_lock(&lnk->rl_ha_clink);
}

static inline struct m0_reqh *rconfc_link2reqh(struct rconfc_link *lnk)
{
        return lnk->rl_rconfc->rc_rmach->rm_reqh;
}

static bool rconfc_herd_link__on_death_cb(struct m0_clink *clink)
{
        struct rconfc_link *lnk = M0_AMB(lnk, clink, rl_ha_clink);
        struct m0_conf_obj *obj = M0_AMB(obj, clink->cl_chan, co_ha_chan);

        M0_ENTRY("lnk=%p ep:%s rc %d state:%d", lnk, lnk->rl_confd_addr,
                        lnk->rl_rc, lnk->rl_state);
        M0_ASSERT(m0_fid_eq(&lnk->rl_confd_fid, &obj->co_id));
        if (obj->co_ha_state != M0_NC_FAILED) {
                M0_LEAVE("co_ha_state = %d, return true", obj->co_ha_state);
                return true;
        }

        m0_rconfc_lock(lnk->rl_rconfc);
        if (!lnk->rl_finalised) {
                rconfc_link_ast_post(lnk, lnk, rconfc_link_fini_ast);
                lnk->rl_finalised = true;
        }
        m0_rconfc_unlock(lnk->rl_rconfc);
        M0_LEAVE("lnk=%p co_ha_state = M0_NC_FAILED, return false", lnk);
        return false;
}

static void rconfc_link_fini_ast(struct m0_sm_group *grp,
                                 struct m0_sm_ast   *ast)
{
        struct rconfc_link *lnk = ast->sa_datum;

        M0_ENTRY("lnk=%p ep:%s rc %d state:%d", lnk, lnk->rl_confd_addr,
                        lnk->rl_rc, lnk->rl_state);
        m0_rconfc_lock(lnk->rl_rconfc);
        m0_mutex_lock(&lnk->rl_rconfc->rc_herd_lock);
        if (lnk->rl_state != CONFC_DEAD && !lnk->rl_fom_queued) {
                if (lnk->rl_on_state_cb != NULL)  /* For UT only */
                        lnk->rl_on_state_cb(lnk);
                lnk->rl_fom_queued = true;
                if (m0_clink_is_armed(&lnk->rl_clink)) {
                        m0_clink_del(&lnk->rl_clink);
                        m0_clink_fini(&lnk->rl_clink);
                        m0_confc_ctx_fini_locked(&lnk->rl_cctx);
                }
                m0_fom_init(&lnk->rl_fom, &rconfc_link_fom_type,
                            &rconfc_link_fom_ops, NULL, NULL,
                            rconfc_link2reqh(lnk));
                m0_fom_queue(&lnk->rl_fom);
                M0_LOG(M0_DEBUG, "Created a fom %p to finalize this lnk %p",
                                 &lnk->rl_fom, lnk);
        } else {
                M0_LOG(M0_INFO, "Link to "FID_F" known to be CONFC_DEAD",
                       FID_P(&lnk->rl_confd_fid));
        }
        lnk->rl_finalised = true;
        m0_mutex_unlock(&lnk->rl_rconfc->rc_herd_lock);
        m0_rconfc_unlock(lnk->rl_rconfc);
}

static void rconfc_herd_link_init(struct rconfc_link *lnk)
{
        enum { HERD_LINK_TIMEOUT_DEFAULT = 1ULL * M0_TIME_ONE_SECOND };

        struct m0_rconfc *rconfc = lnk->rl_rconfc;

        M0_ENTRY("lnk %p", lnk);
        M0_PRE(rconfc != NULL);
        m0_clink_init(&lnk->rl_ha_clink, rconfc_herd_link__on_death_cb);
        if (M0_FI_ENABLED("confc_init"))
                lnk->rl_rc = -EIO;
        else
                lnk->rl_rc =
                        m0_confc_init_wait(&lnk->rl_confc, rconfc->rc_sm.sm_grp,
                                           lnk->rl_confd_addr, rconfc->rc_rmach,
                                           NULL, HERD_LINK_TIMEOUT_DEFAULT);
        if (lnk->rl_rc == 0)
                lnk->rl_state = CONFC_IDLE;
        else
                m0_clink_fini(&lnk->rl_ha_clink);
        lnk->rl_finalised = false;
        M0_LEAVE();
}

/*
 * M0_INTERNAL here is used for the sake of rconfc_herd_link_die() in
 * conf/rconfc_link_fom.c
 */
M0_INTERNAL void rconfc_herd_link_fini(struct rconfc_link *lnk)
{
        M0_ENTRY("lnk %p", lnk);
        M0_PRE(rconfc_is_locked(lnk->rl_rconfc));
        M0_PRE(lnk->rl_fom_clink.cl_chan == NULL);
        M0_PRE(lnk->rl_fom_queued || M0_IS0(&lnk->rl_fom));
        /* dead confc has no internals to fini */
        if (!M0_IN(lnk->rl_state, (CONFC_DEAD, CONFC_ARMED))) {
                rconfc_herd_link_unsubscribe(lnk);
                _confc_phony_cache_remove(&lnk->rl_rconfc->rc_phony,
                                          &lnk->rl_confd_fid);
                m0_clink_fini(&lnk->rl_ha_clink);
                if (m0_confc_is_inited(&lnk->rl_confc))
                        m0_confc_fini(&lnk->rl_confc);
        }
        M0_LEAVE();
}

static void rconfc_herd_ast(struct m0_sm_group *grp,
                            struct m0_sm_ast   *ast)
{
        M0_LOG(M0_DEBUG, "Re-trying to stop...");
        rconfc_stop_internal(ast->sa_datum);
}

static bool rconfc_herd_fini_cb(struct m0_clink *link)
{
        struct m0_rconfc *rconfc = M0_AMB(rconfc, link, rc_herd_cl);
        M0_ENTRY("rconfc %p", rconfc);
        m0_clink_del(link);
        m0_rconfc_lock(rconfc);
        rconfc_ast_post(rconfc, rconfc_herd_ast);
        m0_rconfc_unlock(rconfc);
        M0_LEAVE();
        return true;
}

static void rconfc_ha_update_ast(struct m0_sm_group *grp,
                                 struct m0_sm_ast   *ast)
{
        struct m0_rconfc            *rconfc = ast->sa_datum;
        struct rlock_ctx            *rlx = rconfc->rc_rlock_ctx;
        struct m0_ha_entrypoint_rep *hep = &rconfc->rc_ha_entrypoint_rep;
        char                        *rm_addr = hep->hae_active_rm_ep;
        int                          rc = 0;

        m0_free(rconfc->rc_nvec.nv_note);
        if (rconfc_state(rconfc) == M0_RCS_FAILURE) {
                M0_LOG(M0_WARN, "rconfc is in a failed state:%p", rconfc);
                return;
        }
        rconfc_state_set(rconfc, M0_RCS_CREDITOR_SETUP);

        if (rlx->rlc_rm_addr == NULL || !m0_streq(rlx->rlc_rm_addr, rm_addr)) {
                if (rlock_ctx_is_online(rlx))
                        rlock_ctx_creditor_unset(rlx);
                rc = rm_addr != NULL ? rlock_ctx_creditor_setup(rlx, rm_addr) :
                                       -EAGAIN;
        }
        if (rc == 0) {
                rconfc_read_lock_get(rconfc);
        } else {
                rconfc_start(rconfc);
                M0_CNT_INC(rconfc->rc_ha_entrypoint_retries);
        }

        m0_ha_entrypoint_rep_free(&rconfc->rc_ha_entrypoint_rep);
}

static bool rconfc_ha_update_cb(struct m0_clink *link)
{
        struct m0_rconfc *rconfc = M0_AMB(rconfc, link, rc_ha_update_cl);
        M0_ENTRY("rconfc %p", rconfc);
        m0_clink_del(link);
        rconfc_ast_post(rconfc, rconfc_ha_update_ast);
        M0_LEAVE();
        return true;
}

static int rconfc_herd_fini(struct m0_rconfc *rconfc)
{
        struct rconfc_link *lnk;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc_is_locked(rconfc));
        m0_mutex_lock(&rconfc->rc_herd_lock);
        m0_tl_for(rcnf_herd, &rconfc->rc_herd, lnk) {
                if (lnk->rl_cctx.fc_confc == NULL) {
                        M0_LOG(M0_DEBUG, "lnk %p ctx is finalised", lnk);
                } else if (m0_confc_ctx_is_completed(&lnk->rl_cctx)) {
                        if (m0_clink_is_armed(&lnk->rl_clink)) {
                                m0_clink_del(&lnk->rl_clink);
                                m0_clink_fini(&lnk->rl_clink);
                                m0_confc_ctx_fini_locked(&lnk->rl_cctx);
                        }
                } else {
                        /*
                         * Found link which conf reading context is still
                         * active. We need to cancel context operations and wait
                         * until the context is completed.
                         */
                        M0_LOG(M0_DEBUG, "Stop re-trying required (cctx)");
                        M0_LOG(M0_DEBUG, "adding clink %p to chan %p",
                                         &rconfc->rc_herd_cl,
                                         &lnk->rl_cctx.fc_mach.sm_chan);
                        m0_clink_add(&lnk->rl_cctx.fc_mach.sm_chan,
                                     &rconfc->rc_herd_cl);
                        m0_rpc_conn_sessions_cancel(
                                m0_confc2conn(&lnk->rl_confc));
                        m0_mutex_unlock(&rconfc->rc_herd_lock);
                        return M0_RC(1);
                }
                if (lnk->rl_fom_queued) {
                        /*
                         * Found link already queued for finalisation. We need
                         * to wait until FOM finalisation is announced on the
                         * channel m0_rconfc::rc_herd_chan.
                         */
                        M0_LOG(M0_DEBUG, "Stop re-trying required (link FOM)");
                        m0_clink_add(&rconfc->rc_herd_chan,
                                     &rconfc->rc_herd_cl);
                        m0_mutex_unlock(&rconfc->rc_herd_lock);
                        return M0_RC(1);
                }
        } m0_tl_endfor;
        m0_mutex_unlock(&rconfc->rc_herd_lock);
        m0_tl_for(rcnf_herd, &rconfc->rc_herd, lnk) {
                rconfc_herd_link_fini(lnk);
        } m0_tl_endfor;
        return M0_RC(0);
}

static void rconfc_herd_link_destroy(struct rconfc_link *lnk)
{
        rcnf_herd_tlink_fini(lnk);
        m0_free(lnk->rl_confd_addr);
        m0_free(lnk);
}

M0_INTERNAL void rconfc_herd_link_cleanup(struct rconfc_link *lnk)
{
        M0_ENTRY("lnk %p", lnk);
        rcnf_herd_tlist_del(lnk);
        M0_LEAVE();
}

static void rconfc_herd_prune(struct m0_rconfc *rconfc)
{
        struct rconfc_link *lnk;

        M0_ENTRY("rconfc = %p", rconfc);
        m0_tl_teardown(rcnf_herd, &rconfc->rc_herd, lnk)
                rconfc_herd_link_destroy(lnk);
        M0_LEAVE();
}

static int rconfc_herd_destroy(struct m0_rconfc *rconfc)
{
        int rc;

        M0_ENTRY();

        rconfc_active_all_unlink(rconfc);
        rc = rconfc_herd_fini(rconfc);
        if (rc == 0)
                rconfc_herd_prune(rconfc);

        return M0_RC(rc);
}

M0_INTERNAL struct rconfc_link *rconfc_herd_find(struct m0_rconfc *rconfc,
                                                 const char       *addr)
{
        M0_PRE(addr != NULL);
        return m0_tl_find(rcnf_herd, lnk, &rconfc->rc_herd,
                                m0_streq(lnk->rl_confd_addr, addr));
}

static int rconfc_herd_update(struct m0_rconfc   *rconfc,
                              const char        **confd_addr,
                              struct m0_fid_arr  *confd_fids)
{
        struct rconfc_link *lnk;
        uint32_t            count = rconfc_confd_count(confd_addr);
        uint32_t            idx;
        uint32_t            link_quorum = 0;

        M0_ENTRY("rconfc = %p, confd_addr[] = %p, [confd_addr] = %u",
                 rconfc, confd_addr, count);
        M0_PRE(confd_addr != NULL);
        M0_PRE(count > 0);
        M0_PRE(count == confd_fids->af_count);
        M0_PRE(rconfc_confd_addr_are_all_unique(confd_addr));
        M0_PRE(m0_fid_arr_all_unique(confd_fids));

        for (idx = 0; *confd_addr != NULL; confd_addr++, idx++) {
                lnk = rconfc_herd_find(rconfc, *confd_addr);
                if (lnk == NULL) {
                        M0_ALLOC_PTR(lnk);
                        if (lnk == NULL)
                                goto no_mem;
                        /* add the allocated element to herd */
                        lnk->rl_rconfc     = rconfc;
                        lnk->rl_confd_fid  = confd_fids->af_elems[idx];
                        lnk->rl_confd_addr = m0_strdup(*confd_addr);
                        if (lnk->rl_confd_addr == NULL) {
                                m0_free(lnk);
                                goto no_mem;
                        }
                        lnk->rl_state      = CONFC_DEAD;
                        rcnf_herd_tlink_init_at_tail(lnk, &rconfc->rc_herd);
                        /* Dead links are destroyed during rm revoke and will
                         * be created again.
                         * XXX:  rconfc_herd_link_init() can block on waiting
                         *       and this function is called from an AST:
                         *       rconfc_conductor_disconnected_ast() ->
                         *         rconfc_conductor_disconnected() ->
                         *           rconfc_start() ->
                         *               or:
                         *       rlock_owner_clink_cb() ->
                         *         rconfc_owner_creditor_reset() ->
                         *           rconfc_start() ->
                         *               or:
                         *         rconfc_start_ast_cb() ->
                         *           rconfc_start() ->
                         *             rconfc_start_internal() ->
                         *               rconfc_entrypoint_consume() ->
                         *                 rconfc_herd_update()
                         */
                        rconfc_herd_link_init(lnk);
                        /*
                         * only successfully connected @ref rconfc_link gets
                         * subscribed to HA notifications, and therefore, its
                         * confd fid is to be added to phony cache
                         */
                        if (lnk->rl_state == CONFC_IDLE) {
                                _confc_phony_cache_append(
                                        &rconfc->rc_phony,
                                        &lnk->rl_confd_fid);
                                rconfc_herd_link_subscribe(lnk);
                        } else {
                                M0_ASSERT(lnk->rl_state == CONFC_DEAD);
                        }
                }
                if (lnk->rl_rc == 0)
                        link_quorum++;
                lnk->rl_preserve = true;
        }

        M0_LOG(M0_DEBUG, "rconfc: link_quorum = %d rc_quorum = %d",
                          link_quorum, rconfc->rc_quorum);
        ver_accm_init(rconfc->rc_qctx, count);
        m0_tl_for (rcnf_herd, &rconfc->rc_herd, lnk) {
                if (!lnk->rl_preserve ||
                     link_quorum < rconfc->rc_quorum) {
                        rconfc_herd_link_fini(lnk);
                        rcnf_herd_tlist_del(lnk);
                        rconfc_herd_link_destroy(lnk);
                } else {
                        /*
                         * Clean confc cache, so version will be actually
                         * queried from confd, not extracted from cache.
                         */
                        if (!M0_IN(lnk->rl_state, (CONFC_DEAD, CONFC_ARMED)))
                                _confc_cache_clean_lock(&lnk->rl_confc);
                        lnk->rl_preserve = false;
                }
        } m0_tl_endfor;
        if (link_quorum < rconfc->rc_quorum) {
                M0_LOG(M0_DEBUG, "rc_quorum = %d, retrying entrypoint.. ",
                                  rconfc->rc_quorum);
                return M0_RC(-EAGAIN);
        }
        return m0_tl_exists(rcnf_herd, lnk, &rconfc->rc_herd,
                            lnk->rl_state != CONFC_DEAD) ? M0_RC(0) :
                                                           M0_ERR(-ENOENT);
no_mem:
        rconfc_herd_prune(rconfc);
        return M0_ERR(-ENOMEM);
}

static void rconfc_active_add(struct m0_rconfc *rconfc, struct rconfc_link *lnk)
{
        if (lnk->rl_state == CONFC_OPEN &&
            _confc_ver_read(&lnk->rl_confc) == rconfc->rc_ver)
                rcnf_active_tlink_init_at_tail(lnk, &rconfc->rc_active);
}

static void rconfc_active_populate(struct m0_rconfc *rconfc)
{
        struct rconfc_link *lnk;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc->rc_ver != M0_CONF_VER_UNKNOWN);
        rconfc_active_all_unlink(rconfc);
        /* re-populate active list */
        m0_tl_for(rcnf_herd, &rconfc->rc_herd, lnk) {
                rconfc_active_add(rconfc, lnk);
        } m0_tl_endfor;
        M0_LEAVE();
}

static int rconfc_conductor_connect(struct m0_rconfc   *rconfc,
                                    struct rconfc_link *lnk)
{
        /*
         * As the conductor is expected to operate on rcnf_active list entries,
         * i.e. previously known to be connectable alright, there is no need in
         * any lengthy timeout here.
         *
         * In case active list iteration brings no success, rconfc is just to
         * repeat version election starting with ENTRYPOINT request.
         */
        int rc;
        enum { CONDUCTOR_TIMEOUT_DEFAULT = 200ULL * M0_TIME_ONE_MSEC };

        M0_ENTRY("rconfc = %p, lnk = %p, confd_addr = %s", rconfc, lnk,
                 lnk != NULL ? lnk->rl_confd_addr : "(null)");
        M0_PRE(rconfc != NULL);
        M0_PRE(lnk    != NULL);
        if (!m0_confc_is_inited(&rconfc->rc_confc)) {
                /* first use, initialization required */
                M0_PRE(_confc_ver_read(&rconfc->rc_confc) ==
                       M0_CONF_VER_UNKNOWN);
                rc = m0_confc_init_wait(&rconfc->rc_confc, rconfc->rc_sm.sm_grp,
                                        lnk->rl_confd_addr, rconfc->rc_rmach,
                                        rconfc->rc_local_conf,
                                        CONDUCTOR_TIMEOUT_DEFAULT) ?:
                     m0_ha_client_add(&rconfc->rc_confc);
                if (rc != 0)
                        return M0_ERR(rc);

                m0_confc_gate_ops_set(&rconfc->rc_confc, &rconfc->rc_gops);
        }

        return M0_RC(m0_confc_reconnect(&rconfc->rc_confc, rconfc->rc_rmach,
                                        lnk->rl_confd_addr));
}

static int rconfc_conductor_iterate(struct m0_rconfc *rconfc)
{
        struct rconfc_link *next;
        struct rconfc_link *prev;
        const char         *confd_addr;
        int                 rc;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc != NULL);
        M0_PRE(rconfc->rc_ver != M0_CONF_VER_UNKNOWN);
        confd_addr = _confc_remote_addr_read(&rconfc->rc_confc) ?: "";
        M0_PRE((prev = m0_tl_find(rcnf_active, item, &rconfc->rc_active,
                                  m0_streq(confd_addr, item->rl_confd_addr)))
               == NULL || prev->rl_state == CONFC_OPEN);
        /* mark items idle except the failed one */
        m0_tl_for(rcnf_active, &rconfc->rc_active, next) {
                M0_PRE(next->rl_state != CONFC_DEAD);
                next->rl_state = m0_streq(confd_addr, next->rl_confd_addr) ?
                        CONFC_FAILED : CONFC_IDLE;
        } m0_tl_endfor;
        /* start from the last connected item, or from list tail otherwise */
        prev = m0_tl_find(rcnf_active, item, &rconfc->rc_active,
                          m0_streq(confd_addr, item->rl_confd_addr)) ?:
                rcnf_active_tlist_tail(&rconfc->rc_active);
        while (1) {
                /*
                 * loop through the list until successful connect or no more
                 * idle items to try
                 */
                next = rcnf_active_tlist_next(&rconfc->rc_active, prev) ?:
                        rcnf_active_tlist_head(&rconfc->rc_active);
                if (next->rl_state == CONFC_FAILED)
                        /* this is to start version reelection */
                        return M0_ERR(-ENOENT);
                rc = rconfc_conductor_connect(rconfc, next);
                if (rc == 0 && !M0_FI_ENABLED("conductor_conn_fail")) {
                        next->rl_state = CONFC_OPEN;
                        return M0_RC(rc);
                }
                M0_LOG(M0_ERROR, "Failed to connect to confd_addr = %s rc = %d",
                       next->rl_confd_addr, rc);
                next->rl_state = CONFC_FAILED;
                prev = next;
        }
}

static void rconfc_read_lock_get(struct m0_rconfc *rconfc)
{
        struct rlock_ctx      *rlx;
        struct m0_rm_incoming *req;

        M0_ENTRY("rconfc = %p", rconfc);
        rconfc_state_set(rconfc, M0_RCS_GET_RLOCK);
        rlx = rconfc->rc_rlock_ctx;
        req = &rlx->rlc_req;
        m0_rm_rwlock_req_init(req, &rlx->rlc_owner, &m0_rconfc_ri_ops,
                              RIF_MAY_BORROW | RIF_MAY_REVOKE, RM_RWLOCK_READ);
        m0_rm_credit_get(req);
        M0_LEAVE();
}

static void
rconfc_entrypoint_debug_print(struct m0_ha_entrypoint_rep *entrypoint)
{
        int i;

        M0_LOG(M0_DEBUG, "hbp_quorum=%"PRIu32" confd_nr=%"PRIu32,
               entrypoint->hae_quorum, entrypoint->hae_confd_fids.af_count);
        M0_LOG(M0_DEBUG, "hbp_active_rm_fid="FID_F" hbp_active_rm_ep=%s",
               FID_P(&entrypoint->hae_active_rm_fid),
               entrypoint->hae_active_rm_ep);
        for (i = 0; entrypoint->hae_confd_eps[i] != NULL; ++i) {
                M0_LOG(M0_DEBUG, "hbp_confd_fids[%d]="FID_F
                       " hbp_confd_eps[%d]=%s",
                       i, FID_P(&entrypoint->hae_confd_fids.af_elems[i]),
                       i, entrypoint->hae_confd_eps[i]);
        }
}

static int rconfc_entrypoint_consume(struct m0_rconfc *rconfc)
{
        struct rlock_ctx               *rlx = rconfc->rc_rlock_ctx;
        struct m0_confc                *phony = &rlx->rlc_parent->rc_phony;
        struct m0_ha_entrypoint_rep    *hep = &rconfc->rc_ha_entrypoint_rep;
        int                             rc;

        M0_ENTRY();
        rconfc_state_set(rconfc, M0_RCS_ENTRYPOINT_CONSUME);
        if (hep->hae_control == M0_HA_ENTRYPOINT_QUIT)
                /* HA commanded stop querying. No operation permitted. */
                return M0_ERR(-EPERM);
        if (!m0_fid_is_set(&hep->hae_active_rm_fid))
                /*
                 * The situation when active RM fid is unset cannot happen
                 * during regular cluster operation. Whichever reason actually
                 * caused the fid to be zero, it must be considered an
                 * unrecoverable issue preventing rconfc from further running.
                 */
                return M0_ERR(-ENOKEY);
        rconfc->rc_quorum = hep->hae_quorum;
        rlx->rlc_rm_fid = hep->hae_active_rm_fid;
        M0_LOG(M0_DEBUG, "rm_fid="FID_F, FID_P(&rlx->rlc_rm_fid));
        if (hep->hae_active_rm_ep == NULL || hep->hae_active_rm_ep[0] == '\0')
                return M0_ERR(-ENOENT);
        rconfc_entrypoint_debug_print(hep);
        _confc_phony_cache_append(phony, &rlx->rlc_rm_fid);

        /*
         * rconfc SM channel is used for the convenience here:
         * 1) No additional channel and mutex structures are required.
         * 2) No additional locks are required here since the code is always
         *    run under the same SM group lock.
         * The drawback of this approach is that the threads waiting
         * for the SM state change (like at m0_rconfc_start_wait()) will be
         * awaken needlessly. But this seems to be harmless.
         */
        m0_clink_add(&rconfc->rc_sm.sm_chan, &rconfc->rc_ha_update_cl);

        rc = rconfc_herd_update(rconfc, hep->hae_confd_eps,
                                &hep->hae_confd_fids) ?:
             m0_conf_confc_ha_update_async(&rconfc->rc_phony,
                                           &rconfc->rc_nvec,
                                           &rconfc->rc_sm.sm_chan);

        if (rc != 0) {
                m0_clink_del(&rconfc->rc_ha_update_cl);
                return M0_ERR(rc);
        }

        return M0_RC(rc);
}

static int rconfc_start_internal(struct m0_rconfc *rconfc)
{
        int               rc;

        M0_ENTRY();

        rc = rconfc_entrypoint_consume(rconfc);
        if (rc != 0) {
                if (M0_IN(rc, (-ENOKEY, -EPERM)))
                        /* HA requested rconfc to stop querying entrypoint */
                        rconfc_fail(rconfc, rc);
                else
                        /* rconfc is to keep trying with next entrypoint */
                        rconfc_herd_destroy(rconfc);
                return M0_ERR(rc);
        }
        rconfc->rc_quorum_decision_is_made = false;
        return M0_RC(rc);
}

static void rconfc_start(struct m0_rconfc *rconfc)
{
        struct m0_ha                   *ha  = m0_get()->i_ha;
        struct m0_ha_entrypoint_client *ecl = &ha->h_entrypoint_client;
        int                             rc  = 0;

        M0_PRE(rconfc->rc_local_conf == NULL);

        M0_ENTRY();

        while (rconfc_state(rconfc) != M0_RCS_FAILURE) {
                rconfc_state_set(rconfc, M0_RCS_ENTRYPOINT_WAIT);

                if (rconfc->rc_ha_entrypoint_rep.hae_control == M0_HA_ENTRYPOINT_QUERY) {
                        M0_LOG(M0_DEBUG, "Querying ENTRYPOINT...");
                        m0_ha_entrypoint_client_request(ecl);
                        break;
                }

                M0_LOG(M0_DEBUG, "ENTRYPOINT ready...");
                rc = rconfc_start_internal(rconfc);
                rconfc->rc_ha_entrypoint_rep.hae_control = M0_HA_ENTRYPOINT_QUERY;
                if (rc == 0)
                        break;
                m0_ha_entrypoint_rep_free(&rconfc->rc_ha_entrypoint_rep);
                M0_LOG(M0_DEBUG, "Try reconnecting after rc=%d", rc);
                M0_CNT_INC(rconfc->rc_ha_entrypoint_retries);
        }

        M0_LEAVE("rc=%d hae_control=%d is_failed=%s entrypoint_retries=%"PRIu32,
                 rc, rconfc->rc_ha_entrypoint_rep.hae_control,
                 m0_bool_to_str(rconfc_state(rconfc) == M0_RCS_FAILURE),
                 rconfc->rc_ha_entrypoint_retries);
}

static void rconfc_start_ast_cb(struct m0_sm_group *grp M0_UNUSED,
                                struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        M0_ENTRY();
        rconfc_start(rconfc);
        M0_LEAVE();
}

static void rconfc_read_lock_retry(struct m0_sm_group *grp M0_UNUSED,
                                   struct m0_sm_ast *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        M0_ENTRY("rconfc = %p", rconfc);
        rconfc_state_set(rconfc, M0_RCS_GET_RLOCK);
        rconfc_read_lock_get(rconfc);
        M0_LEAVE();
}

static void rconfc_owner_creditor_reset(struct m0_sm_group *grp M0_UNUSED,
                                        struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;
        struct rlock_ctx *rlx    = rconfc->rc_rlock_ctx;

        M0_ENTRY("rconfc = %p", rconfc);
        rlock_ctx_creditor_unset(rlx);
        /*
         * Start conf reelection.
         * RM creditor is likely to be changed by HA.
         */
        rconfc_start(rconfc);

        M0_LEAVE();
}

static bool rlock_owner_clink_cb(struct m0_clink *cl)
{
        struct rlock_ctx *rlx = container_of(cl, struct rlock_ctx, rlc_clink);
        uint32_t          state = rlx->rlc_owner.ro_sm.sm_state;
        struct m0_rconfc *rconfc = rlx->rlc_parent;
        M0_ENTRY("cl=%p rconfc=%p", cl, rconfc);

        M0_ASSERT(state != ROS_INSOLVENT);
        if (state == ROS_DEAD_CREDITOR) {
                rconfc_ast_post(rconfc, rconfc_owner_creditor_reset);
                m0_clink_del(cl);
                m0_clink_fini(cl);
        }
        M0_LEAVE();
        return true;
}

static void rconfc_creditor_death_handle(struct m0_rconfc *rconfc)
{
        struct rlock_ctx   *rlx = rconfc->rc_rlock_ctx;
        struct m0_rm_owner *owner = &rlx->rlc_owner;

        M0_ENTRY("rconfc= %p ro_sm.sm_state=%d", rconfc, owner->ro_sm.sm_state);
        M0_PRE(rlock_ctx_creditor_state(rlx) != ROS_ACTIVE);
        if (owner->ro_sm.sm_state == ROS_DEAD_CREDITOR) {
                rconfc_ast_post(rconfc, rconfc_owner_creditor_reset);
        } else {
                /* Wait until owner is in ROS_DEAD_CREDITOR state. */
                m0_clink_init(&rlx->rlc_clink, rlock_owner_clink_cb);
                m0_clink_add(&owner->ro_sm.sm_chan, &rlx->rlc_clink);
        }
        M0_LEAVE();
}

static void rconfc_conductor_disconnected(struct m0_rconfc *rconfc)
{
        struct rlock_ctx   *rlx   = rconfc->rc_rlock_ctx;
        struct m0_rm_owner *owner = &rlx->rlc_owner;

        M0_ENTRY("rconfc = %p", rconfc);
        /* return read lock back to RM */
        rconfc_read_lock_put(rconfc);
        /* prepare for version election */
        rconfc_active_all_unlink(rconfc);
        rconfc->rc_ver = M0_CONF_VER_UNKNOWN;
        if (rlock_ctx_creditor_state(rlx) != ROS_ACTIVE) {
                m0_rm_owner_lock(owner);
                /* Creditor is considered dead by HA */
                rconfc_creditor_death_handle(rconfc);
                m0_rm_owner_unlock(owner);
        } else {
                /*
                 * Start process of conf reelection.
                 * List of confd or active RM could possibly have changed.
                 */
                rconfc_start(rconfc);
        }
        M0_LEAVE();
}

static void rconfc_conductor_disconnected_ast(struct m0_sm_group *grp,
                                              struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        m0_clink_cleanup(&rconfc->rc_conductor_clink);
        m0_rpc_link_fini(&rconfc->rc_confc.cc_rlink);
        rconfc_conductor_disconnected(rconfc);
}

static bool rconfc_conductor_disconnect_cb(struct m0_clink *clink)
{
        struct m0_rconfc *rconfc = M0_AMB(rconfc, clink, rc_conductor_clink);

        M0_ENTRY("rconfc = %p", rconfc);
        m0_rconfc_lock(rconfc);
        rconfc_ast_post(rconfc, rconfc_conductor_disconnected_ast);
        m0_rconfc_unlock(rconfc);
        M0_LEAVE();
        return true;
}

static void rconfc_conductor_drained(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc = %p", rconfc);
        /* disconnect confc until read lock being granted */
        rconfc_state_set(rconfc, M0_RCS_CONDUCTOR_DISCONNECT);
        m0_clink_init(&rconfc->rc_conductor_clink,
                      rconfc_conductor_disconnect_cb);
        rconfc->rc_conductor_clink.cl_is_oneshot = true;
        m0_rpc_link_disconnect_async(&rconfc->rc_confc.cc_rlink,
                                     m0_rpc__down_timeout(),
                                     &rconfc->rc_conductor_clink);
        M0_LEAVE();
}

static void rconfc_conductor_drain(struct m0_sm_group *grp,
                                   struct m0_sm_ast   *ast)
{
        struct m0_rconfc     *rconfc = ast->sa_datum;
        struct m0_conf_cache *cache = &rconfc->rc_confc.cc_cache;
        struct m0_conf_obj   *obj;
        int                   rc = 1;

        M0_ENTRY("rconfc = %p", rconfc);
        m0_conf_cache_lock(cache);
        if ((obj = m0_conf_cache_pinned(cache)) != NULL) {
                M0_LOG(M0_DEBUG, "* pinned (%" PRIu64 ") obj "FID_F", "
                       "waiters %d, ha %d", obj->co_nrefs, FID_P(&obj->co_id),
                       obj->co_chan.ch_waiters, obj->co_ha_chan.ch_waiters);
                m0_clink_add(&obj->co_chan, &rconfc->rc_unpinned_cl);
        } else {
                rc = _confc_cache_clean(&rconfc->rc_confc);
                if (rc != 0)
                        rconfc_fail(rconfc, rc);
        }
        m0_conf_cache_unlock(cache);

        if (rc == 0)
                rconfc->rc_stopping ? rconfc_state_set(rconfc, M0_RCS_FINAL) :
                                      rconfc_conductor_drained(rconfc);
        M0_LEAVE("rc=%d, stopping=%d", rc, rconfc->rc_stopping);
}

static bool rconfc_unpinned_cb(struct m0_clink *link)
{
        struct m0_rconfc *rconfc = container_of(
                                link, struct m0_rconfc, rc_unpinned_cl);

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc_state(rconfc) == M0_RCS_CONDUCTOR_DRAIN);
        m0_clink_del(link);
        m0_rconfc_lock(rconfc);
        rconfc_ast_post(rconfc, rconfc_conductor_drain);
        m0_rconfc_unlock(rconfc);
        M0_LEAVE();
        return false;
}

static bool rconfc_gate_check(struct m0_confc *confc)
{
        struct m0_rconfc *rconfc;
        bool              result;

        M0_ENTRY("confc = %p", confc);
        M0_PRE(confc != NULL);
        M0_PRE(m0_mutex_is_locked(&confc->cc_lock));

        rconfc = container_of(confc, struct m0_rconfc, rc_confc);
        if (!m0_rconfc_reading_is_allowed(rconfc)) {
                m0_mutex_unlock(&confc->cc_lock);
                m0_rconfc_lock(rconfc);
                m0_sm_timedwait(&rconfc->rc_sm,
                                M0_BITS(M0_RCS_IDLE, M0_RCS_FAILURE),
                                M0_TIME_NEVER);
                m0_rconfc_unlock(rconfc);
                m0_mutex_lock(&confc->cc_lock);
        }
        result = m0_rconfc_reading_is_allowed(rconfc);
        M0_LEAVE("result=%s", result ? "true" : "false");
        return result;
}

static int rconfc_gate_skip(struct m0_confc *confc)
{
        struct m0_rconfc *rconfc;

        M0_ENTRY("confc = %p", confc);
        M0_PRE(confc != NULL);
        rconfc = container_of(confc, struct m0_rconfc, rc_confc);
        return M0_RC(rconfc_conductor_iterate(rconfc));
}

static bool rconfc_gate_drain(struct m0_clink *clink)
{
        struct m0_rconfc *rconfc;
        struct m0_confc  *confc;

        M0_ENTRY("clink = %p", clink);
        confc = container_of(clink, struct m0_confc, cc_drain);
        rconfc = container_of(confc, struct m0_rconfc, rc_confc);
        rconfc->rc_gops.go_drain = NULL;
        m0_confc_gate_ops_set(&rconfc->rc_confc, &rconfc->rc_gops);
        /*
         * Conductor confc has no active confc contexts at this moment,
         * so it's safe to call rconfc_ast_drain() in this sm group.
         *
         * If several rconfc operate in one sm group, then processing
         * 'foreign' confc contexts is stopped for some time, but let's live
         * with that for now, because no dead-locks are possible.
         */
        rconfc_state_set(rconfc, M0_RCS_CONDUCTOR_DRAIN);
        rconfc_ast_post(rconfc, rconfc_conductor_drain);
        M0_LEAVE();
        return false; /* leave this event "unconsumed" */
}

static void rlock_conflict_handle(struct m0_sm_group *grp,
                                  struct m0_sm_ast   *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        M0_ENTRY("rconfc = %p, expired_cb = %p, ready_cb = %p", rconfc,
                 rconfc->rc_expired_cb, rconfc->rc_ready_cb);
        M0_PRE(rconfc_is_locked(rconfc));
        /* prepare for emptying conductor's cache */
        if (rconfc->rc_expired_cb != NULL)
                rconfc->rc_expired_cb(rconfc);
        /*
         * if no context attached, call it directly, otherwise it is
         * going to be called during the very last context finalisation
         */
        m0_mutex_lock(&rconfc->rc_confc.cc_lock);
        M0_LOG(M0_DEBUG, "rc_confc.cc_nr_ctx = %d", rconfc->rc_confc.cc_nr_ctx);
        if (rconfc->rc_confc.cc_nr_ctx == 0) {
                rconfc_state_set(rconfc, M0_RCS_CONDUCTOR_DRAIN);
                rconfc_ast_post(rconfc, rconfc_conductor_drain);
        } else {
                rconfc->rc_gops.go_drain = m0_rconfc_gate_ops.go_drain;
                m0_confc_gate_ops_set(&rconfc->rc_confc,
                                      &rconfc->rc_gops);
        }
        m0_mutex_unlock(&rconfc->rc_confc.cc_lock);
        M0_LEAVE();
}

static void rconfc_rlock_windup(struct m0_rconfc *rconfc)
{
        struct rlock_ctx *rlx = rconfc->rc_rlock_ctx;

        M0_ENTRY("rconfc = %p", rconfc);
        m0_rconfc_unlock(rconfc);
        rconfc_read_lock_put(rconfc);
        m0_rconfc_lock(rconfc);
        if (rlock_ctx_creditor_state(rlx) == ROS_ACTIVE)
                rlock_ctx_owner_windup(rlx);
        m0_rm_rwlock_owner_fini(&rlx->rlc_owner);
        M0_LEAVE();
}

static void rconfc_stop_internal(struct m0_rconfc *rconfc)
{
        struct rlock_ctx *rlx = rconfc->rc_rlock_ctx;
        int               rc;

        M0_ENTRY("rconfc = %p", rconfc);
        if (rconfc->rc_sm.sm_state == M0_RCS_FINAL)
                goto exit;
        /*
         * This function may be called several times from rconfc_herd_fini_cb().
         * It is possible that rconfc already in M0_RCS_STOPPING state.
         */
        if (rconfc->rc_sm.sm_state != M0_RCS_STOPPING)
                rconfc_state_set(rconfc, M0_RCS_STOPPING);
        rc = rconfc_herd_fini(rconfc);
        if (rc != 0)
                goto exit;
        rconfc_rlock_windup(rconfc);
        if (rlock_ctx_is_online(rlx))
                rlock_ctx_creditor_unset(rlx);
        rconfc_state_set(rconfc, M0_RCS_CONDUCTOR_DRAIN);
        if (!m0_confc_is_inited(&rconfc->rc_confc)) {
                rconfc_state_set(rconfc, M0_RCS_FINAL);
                goto exit;
        }
        rconfc_ast_post(rconfc, rconfc_conductor_drain);
exit:
        M0_LEAVE();
}

static void rconfc_stop_ast_cb(struct m0_sm_group *grp, struct m0_sm_ast *ast)
{
        struct m0_rconfc *rconfc = ast->sa_datum;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc_is_locked(rconfc));
        rconfc->rc_stopping = true;
        if (rconfc->rc_expired_cb != NULL)
                rconfc->rc_expired_cb(rconfc);
        rconfc_stop_internal(rconfc);
        M0_LEAVE();
}

static void rconfc_read_lock_conflict(struct m0_rm_incoming *in)
{
        struct m0_rconfc *rconfc;

        M0_ENTRY("in = %p", in);
        rconfc = rlock_ctx_incoming_to_rconfc(in);
        m0_rconfc_lock(rconfc);
        M0_LOG(M0_DEBUG, "rconfc_state(rconfc) = %d", rconfc_state(rconfc));
        if (rconfc_state(rconfc) == M0_RCS_IDLE) {
                rconfc_state_set(rconfc, M0_RCS_RLOCK_CONFLICT);
                rconfc_ast_post(rconfc, rlock_conflict_handle);
        } else {
                rconfc->rc_rlock_conflict = true;
        }
        m0_rconfc_unlock(rconfc);
        M0_LEAVE();
}

static void rconfc_idle(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc_is_locked(rconfc));
        if (rconfc->rc_stopping) {
                rconfc_stop_internal(rconfc);
                M0_LEAVE("Stopped internally");
                return;
        }
        if (rconfc->rc_rlock_conflict) {
                rconfc->rc_rlock_conflict = false;
                rconfc_state_set(rconfc, M0_RCS_RLOCK_CONFLICT);
                rconfc_ast_post(rconfc, rlock_conflict_handle);
                M0_LEAVE("Conflict to be handled...");
                return;
        }
        /*
         * If both rc_stopping and rc_rlock_conflict flags aren't set, then
         * SM will be idle until read lock conflict is observed provoking
         * reelection or user requests stopping rconfc.
         */
        rconfc_state_set(rconfc, M0_RCS_IDLE);
        if (rconfc->rc_ready_cb != NULL)
                rconfc->rc_ready_cb(rconfc);
        M0_LEAVE("Idle");
}

static bool rconfc_quorum_is_possible(struct m0_rconfc *rconfc)
{
        struct ver_accm *va = rconfc->rc_qctx;
        int              ver_count_max;
        int              armed_count;

        M0_PRE(rconfc_is_locked(rconfc));
        armed_count = m0_tl_reduce(rcnf_herd, lnk, &rconfc->rc_herd, 0,
                                   + (lnk->rl_state == CONFC_ARMED));
        ver_count_max = m0_fold(idx, acc, va->va_count, 0,
                                max_type(int, acc, va->va_items[idx].vi_count));
        if (ver_count_max + armed_count < rconfc->rc_quorum) {
                M0_LOG(M0_WARN, "No chance left to reach the quorum");
                rconfc->rc_ver = M0_CONF_VER_UNKNOWN;
                /* Notify consumer about conf expired */
                if (rconfc->rc_expired_cb != NULL)
                        rconfc->rc_expired_cb(rconfc);
                return false;
        }
        return true;
}

static bool rconfc_quorum_test(struct m0_rconfc *rconfc,
                               struct m0_confc *confc)
{
        struct ver_accm *va = rconfc->rc_qctx;
        struct ver_item *vi = NULL;
        uint64_t         ver;
        int              idx;
        bool             quorum_reached = false;

        M0_ENTRY("rconfc = %p, confc = %p", rconfc, confc);
        M0_PRE(va != NULL);
        ver = _confc_ver_read(confc);
        M0_ASSERT(ver != M0_CONF_VER_UNKNOWN);
        for (idx = 0; idx < va->va_count; idx++) {
                if (va->va_items[idx].vi_ver == ver) {
                        vi = va->va_items + idx;
                        break;
                }
        }
        if (vi == NULL) {
                /* new version appeared */
                M0_ASSERT(va->va_count < va->va_total);
                vi = va->va_items + va->va_count;
                M0_PRE(vi->vi_ver == 0);
                M0_PRE(vi->vi_count == 0);
                vi->vi_ver = ver;
                ++va->va_count;
        }
        ++vi->vi_count;

        /* Walk along the herd and see if quorum of any version is reached. */
        for (idx = 0; idx < va->va_count; idx++) {
                if (va->va_items[idx].vi_count >= rconfc->rc_quorum) {
                        /* remember the winner */
                        rconfc->rc_ver = va->va_items[idx].vi_ver;
                        quorum_reached = true;
                        break;
                }
        }
        M0_LEAVE("quorum %sreached", quorum_reached ? "" : "not ");
        return quorum_reached;
}

static int rconfc_conductor_engage(struct m0_rconfc *rconfc)
{
        int rc = 0;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc != NULL);
        M0_PRE(rconfc->rc_ver != M0_CONF_VER_UNKNOWN);
        /*
         * See if the confc not initialized yet, or having different
         * version compared to the newly elected one
         */
        if (!m0_confc_is_inited(&rconfc->rc_confc) ||
            _confc_ver_read(&rconfc->rc_confc) != rconfc->rc_ver) {
                /* need to connect conductor to a new version confd */
                rc = rconfc_conductor_iterate(rconfc);
        }
        return M0_RC(rc);
}

static void rconfc_cctx_fini(struct m0_sm_group *grp,
                             struct m0_sm_ast   *ast)
{
        struct rconfc_link *lnk = ast->sa_datum;

        M0_ASSERT(lnk->rl_state != CONFC_IDLE);
        /*
         * The context might become complete just at the time
         * of rconfc_version_elected() execution and be
         * finalised from rconfc_herd_cctxs_fini() already.
         */
        if (m0_clink_is_armed(&lnk->rl_clink)) {
                m0_clink_del(&lnk->rl_clink);
                m0_clink_fini(&lnk->rl_clink);
                m0_confc_ctx_fini_locked(&lnk->rl_cctx);
        }
}

static void rconfc_herd_cctxs_fini(struct m0_rconfc *rconfc)
{
        struct rconfc_link *lnk;

        m0_tl_for (rcnf_herd, &rconfc->rc_herd, lnk) {
                /*
                 * When lnk->fl_fom_queued is true then it means
                 * that the fom fini is in progress.
                 */
                if (lnk->rl_state == CONFC_DEAD ||
                    (lnk->rl_fom_queued  && lnk->rl_state == CONFC_IDLE ))
                        /*
                         * Even with version elected some links may remain dead
                         * in the herd and require no finalisation. Dead link is
                         * a link that failed to connect to the respective confd
                         * during version election.
                         */
                        continue;
                M0_ASSERT(lnk->rl_state != CONFC_IDLE);
                /*
                 * rconfc_cctx_fini() might be called already for some
                 * of the contexts (because ASTs are not always executed
                 * in the same order they were scheduled).
                 */
                M0_LOG(M0_DEBUG, "lnk:%p, rc = %d, ep = %s state:%d", lnk,
                                lnk->rl_rc, lnk->rl_confd_addr, lnk->rl_state);
                if (m0_clink_is_armed(&lnk->rl_clink) &&
                    m0_confc_ctx_is_completed(&lnk->rl_cctx)) {
                        m0_clink_del(&lnk->rl_clink);
                        m0_clink_fini(&lnk->rl_clink);
                        m0_confc_ctx_fini_locked(&lnk->rl_cctx);
                }
        } m0_tl_endfor;
        M0_LEAVE();
}

static void rconfc_version_elected(struct m0_sm_group *grp,
                                   struct m0_sm_ast   *ast)
{
        struct m0_rconfc   *rconfc = ast->sa_datum;
        int                 rc;

        M0_PRE(rconfc_is_locked(rconfc));
        M0_ENTRY("rconfc = %p", rconfc);

        rconfc_herd_cctxs_fini(rconfc);

        rc = rconfc_quorum_is_reached(rconfc) ?
                rconfc_conductor_engage(rconfc) : -EPROTO;
        if (rc != 0) {
                M0_ERR_INFO(rc, "re-election started");
                rc = rconfc_herd_destroy(rconfc);
                if (rc != 0) {
                        M0_ERR_INFO(rc, "herd_destroy() failed");
                        goto out;
                }
                /*
                 * Start re-election. As version was not elected, the conductor
                 * never was engaged, and therefore needs no draining...
                 */
                if (m0_confc_is_inited(&rconfc->rc_confc)) {
                        m0_ha_client_del(&rconfc->rc_confc);
                        m0_confc_fini(&rconfc->rc_confc);
                }
                /* ... and no disconnection. */
                rconfc_conductor_disconnected(rconfc);
                M0_CNT_INC(rconfc->rc_ha_entrypoint_retries);
        } else {
                M0_SET0(&rconfc->rc_rx);
                M0_SET0(&rconfc->rc_load_ast);
                M0_SET0(&rconfc->rc_load_fini_ast);
                /*
                 * disable gating operations to let rc_confc read configuration
                 * before rconfc gets to M0_RCS_IDLE state
                 */
                m0_confc_gate_ops_set(&rconfc->rc_confc, NULL);
                /* conf load kick-off */
                rconfc->rc_load_ast.sa_datum = rconfc;
                rconfc->rc_load_ast.sa_cb    = rconfc_conf_full_load;
                /*
                 * make conf reading occur in a thread other than standard group
                 * context rconfc was initialised with
                 *
                 * Note: this trick is absolutely temporary, and needs to be
                 * eliminated later when conf clients learn to handle conf
                 * updates on their own
                 */
                rconfc_load_ast_thread_init(&rconfc->rc_rx);
                m0_sm_ast_post(&rconfc->rc_rx.rx_grp, &rconfc->rc_load_ast);
        }
out:
        M0_LEAVE();
}

static bool rconfc__cb_quorum_test(struct m0_clink *clink)
{
        struct rconfc_link *lnk;
        struct m0_rconfc   *rconfc;
        bool                quorum_was;
        bool                quorum_is = false;

        M0_ENTRY("clink = %p", clink);
        M0_PRE(clink != NULL);
        lnk = container_of(clink, struct rconfc_link, rl_clink);
        M0_ASSERT(lnk->rl_state == CONFC_ARMED);
        rconfc = lnk->rl_rconfc;
        M0_ASSERT(rconfc_is_locked(rconfc));

        M0_LOG(M0_DEBUG, "lnk:%p, rc = %d, ep = %s state:%d", lnk, lnk->rl_rc,
                        lnk->rl_confd_addr, lnk->rl_state);
        if (m0_confc_ctx_is_completed(&lnk->rl_cctx)) {
                lnk->rl_rc = m0_confc_ctx_error(&lnk->rl_cctx);
                if (M0_FI_ENABLED("read_ver_failed")) {
                        lnk->rl_confc.cc_cache.ca_ver = M0_CONF_VER_UNKNOWN;
                        lnk->rl_rc = -ENODATA;
                }
                lnk->rl_state = lnk->rl_rc == 0 ? CONFC_OPEN : CONFC_FAILED;

                /*
                 * The code may be called after quorum was already
                 * reached, so we need to see if it was
                 */
                quorum_was = rconfc_quorum_is_reached(rconfc);

                if (lnk->rl_state == CONFC_FAILED)
                        M0_LOG(M0_DEBUG, "Lnk failed, rc = %d, ep = %s",
                               lnk->rl_rc, lnk->rl_confd_addr);
                else if (!quorum_was)
                        quorum_is = rconfc_quorum_test(rconfc, &lnk->rl_confc);

                if (quorum_was)
                        rconfc_active_add(rconfc, lnk);
                else if (quorum_is)
                        rconfc_active_populate(rconfc);

                if (quorum_was) {
                        struct m0_sm_ast *ast = &rconfc->rc_cctx_fini_ast;
                        /* A different AST rc_cctx_fini_ast is used here,
                         * other than the rc_ast, to avoid current access to
                         * the same AST when triggered concurrently from
                         * different sources.
                         */

                        M0_LOG(M0_DEBUG, "cctx_fini_ast=%p old cb %p:d %p:n %p "
                                         "new cb %p:d %p",
                                          ast, ast->sa_cb, ast->sa_datum,
                                          ast->sa_next,
                                          &rconfc_cctx_fini, lnk);
                        ast->sa_cb = rconfc_cctx_fini;
                        ast->sa_datum = lnk;
                        m0_sm_ast_post(rconfc->rc_sm.sm_grp, ast);
                } else if (quorum_is || !rconfc_quorum_is_possible(rconfc))
                        if (!rconfc->rc_quorum_decision_is_made) {
                                rconfc->rc_quorum_decision_is_made = true;
                                rconfc_ast_post(rconfc, rconfc_version_elected);
                        }
        }
        M0_LEAVE();
        return true;
}

static void rconfc_version_elect(struct m0_sm_group *grp, struct m0_sm_ast *ast)
{
        struct m0_rconfc   *rconfc = ast->sa_datum;
        struct ver_accm    *va;
        struct rconfc_link *lnk;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc != NULL && rconfc->rc_qctx != NULL);

        rconfc_state_set(rconfc, M0_RCS_VERSION_ELECT);
        va = rconfc->rc_qctx;
        M0_PRE(va->va_total != 0);
        va->va_count = 0;
        m0_forall(idx, va->va_total, M0_SET0(&va->va_items[idx]));
        /* query confd instances */
        m0_tl_for (rcnf_herd, &rconfc->rc_herd, lnk) {
                if (!M0_IN(lnk->rl_state, (CONFC_DEAD, CONFC_ARMED)) &&
                    lnk->rl_rc == 0 && !lnk->rl_fom_queued) {
                        m0_confc_ctx_init(&lnk->rl_cctx, &lnk->rl_confc);

                        m0_clink_init(&lnk->rl_clink, rconfc__cb_quorum_test);
                        m0_clink_add(&lnk->rl_cctx.fc_mach.sm_chan,
                                     &lnk->rl_clink);
                        lnk->rl_state = CONFC_ARMED;
                        m0_confc_open(&lnk->rl_cctx, NULL,
                                      M0_CONF_ROOT_PROFILES_FID);
                }
        } m0_tl_endfor;
        M0_LEAVE();
}

static void rconfc_read_lock_complete(struct m0_rm_incoming *in, int32_t rc)
{
        enum {
                MAX_RETRY_COUNT     = 100,
        };
        struct m0_rconfc *rconfc;
        struct rlock_ctx *rlx;
        static uint32_t retry_count = 0;

        M0_ENTRY("in = %p, rc = %d", in, rc);
        rconfc = rlock_ctx_incoming_to_rconfc(in);
        M0_ASSERT(rconfc_state(rconfc) == M0_RCS_GET_RLOCK);
        rlx = rconfc->rc_rlock_ctx;
        if (rc != 0) {
                M0_LOG(M0_ERROR, "Read lock request failed with rc = %d", rc);
        }

        if (M0_FI_ENABLED("rlock_req_failed"))
                rc = M0_ERR(-ESRCH);
        m0_rconfc_lock(rconfc);
        if (rc == 0)
                rconfc_ast_post(rconfc, rconfc_version_elect);
        else if (rlock_ctx_creditor_state(rlx) == ROS_ACTIVE) {
                if (rc == -ECONNREFUSED && retry_count++ < MAX_RETRY_COUNT) {
                        M0_LOG(M0_DEBUG, "retrying read lock request because of \
                                          connection refused error");
                        rconfc_ast_post(rconfc, rconfc_read_lock_retry);
                        M0_CNT_INC(rconfc->rc_ha_entrypoint_retries);
                } else {
                        rconfc_fail_ast(rconfc, rc);
                }
        } else
                /* Creditor is considered dead by HA */
                rconfc_creditor_death_handle(rconfc);
        m0_rconfc_unlock(rconfc);
        M0_LEAVE();
}

static int rconfc_local_load(struct m0_rconfc *rconfc)
{
        struct m0_conf_root *root = NULL;
        int                  rc;

        M0_ENTRY("rconfc %p, local_conf = '%s'", rconfc, rconfc->rc_local_conf);
        M0_PRE(rconfc->rc_local_conf != NULL && *rconfc->rc_local_conf != '\0');
        M0_PRE(!rconfc_is_locked(rconfc));
        rc = m0_confc_init(&rconfc->rc_confc, rconfc->rc_sm.sm_grp,
                           NULL, rconfc->rc_rmach, rconfc->rc_local_conf) ?:
                m0_confc_root_open(&rconfc->rc_confc, &root);
        m0_rconfc_lock(rconfc);
        if (rc != 0) {
                rconfc->rc_sm_state_on_abort = rconfc_state(rconfc);
                rconfc_fail(rconfc, rc);
        } else {
                M0_ASSERT(root != NULL);
                rconfc->rc_ver = root->rt_verno;
                m0_confc_close(&root->rt_obj);
                rc = m0_ha_client_add(&rconfc->rc_confc);
                if (rc != 0) {
                        m0_rconfc_unlock(rconfc);
                        M0_LEAVE("rc=%d", rc);
                        return M0_RC(rc);
                }
                if (rconfc->rc_ready_cb != NULL)
                        rconfc->rc_ready_cb(rconfc);
        }
        m0_rconfc_unlock(rconfc);
        M0_LEAVE("rc=%d", rc);
        return M0_RC(rc);
}

/**************************************
 * Rconfc public interface
 **************************************/

M0_INTERNAL void m0_rconfc_lock(struct m0_rconfc *rconfc)
{
        /*
         * Don't use m0_sm_group_lock() to ensure that ASTs posted via
         * rconfc_ast_post() are executed in well-known context.
         * m0_sm_group_lock() has side effect of running ASTs in current context
         * after acquiring the mutex.
         *
         * We use the recursive version of sm group lock, because it might
         * be called from ios start sm group AST, which uses the same sm group.
         */
        m0_sm_group_lock_rec(rconfc->rc_sm.sm_grp, false);
}

M0_INTERNAL void m0_rconfc_unlock(struct m0_rconfc *rconfc)
{
        m0_sm_group_unlock_rec(rconfc->rc_sm.sm_grp, false);
}

M0_INTERNAL int m0_rconfc_init(struct m0_rconfc      *rconfc,
                               const struct m0_fid   *profile,
                               struct m0_sm_group    *sm_group,
                               struct m0_rpc_machine *rmach,
                               m0_rconfc_cb_t         expired_cb,
                               m0_rconfc_cb_t         ready_cb)
{
        int               rc;
        struct rlock_ctx *rlock_ctx;
        struct ver_accm  *va;
        struct m0_ha     *ha;


        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc != NULL);
        M0_PRE(rmach != NULL);
        M0_PRE(sm_group != NULL);

        M0_SET0(rconfc);

        M0_ALLOC_PTR(va);
        if (va == NULL)
                return M0_ERR(-ENOMEM);
        rc = rlock_ctx_create(rconfc, rmach, &rlock_ctx);
        if (rc != 0)
                goto rlock_err;
        rc = _confc_phony_init(&rconfc->rc_phony);
        if (rc != 0)
                goto confc_err;
        rconfc->rc_profile = *profile;
        rconfc->rc_rmach   = rmach;
        rconfc->rc_qctx    = va;
        rconfc->rc_ver     = M0_CONF_VER_UNKNOWN;
        rconfc->rc_gops = (struct m0_confc_gate_ops) {
                .go_check = m0_rconfc_gate_ops.go_check,
                .go_skip  = m0_rconfc_gate_ops.go_skip,
                .go_drain = NULL,
        };
        rconfc->rc_expired_cb = expired_cb;
        rconfc->rc_ready_cb   = ready_cb;
        rconfc->rc_rlock_ctx  = rlock_ctx;
        rconfc->rc_quorum_decision_is_made = false;

        rcnf_herd_tlist_init(&rconfc->rc_herd);
        rcnf_active_tlist_init(&rconfc->rc_active);
        m0_clink_init(&rconfc->rc_unpinned_cl, rconfc_unpinned_cb);
        m0_clink_init(&rconfc->rc_herd_cl, rconfc_herd_fini_cb);
        m0_mutex_init(&rconfc->rc_herd_lock);
        m0_chan_init(&rconfc->rc_herd_chan, &rconfc->rc_herd_lock);
        m0_sm_init(&rconfc->rc_sm, &rconfc_sm_conf, M0_RCS_INIT, sm_group);

        /* Subscribe on ha entrypoint callbacks */
        ha = m0_get()->i_ha;
        m0_clink_init(&rconfc->rc_ha_entrypoint_cl, ha_clink_cb);
        m0_clink_add_lock(m0_ha_entrypoint_client_chan(&ha->h_entrypoint_client),
                          &rconfc->rc_ha_entrypoint_cl);
        rconfc->rc_ha_entrypoint_rep.hae_control = M0_HA_ENTRYPOINT_QUERY;

        m0_clink_init(&rconfc->rc_ha_update_cl, rconfc_ha_update_cb);

        return M0_RC(0);
confc_err:
        rlock_ctx_destroy(rlock_ctx);
rlock_err:
        m0_free(va);
        return M0_ERR(rc);
}

M0_INTERNAL int m0_rconfc_start(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc = %p, profile = "FID_F, rconfc,
                 FID_P(&rconfc->rc_profile));
        M0_PRE(rconfc->rc_fatal_cb == NULL);
        if (rconfc->rc_local_conf != NULL)
                return M0_RC(rconfc_local_load(rconfc));

        m0_rconfc_lock(rconfc);
        rconfc_ast_post(rconfc, rconfc_start_ast_cb);
        m0_rconfc_unlock(rconfc);
        return M0_RC(0);
}

M0_INTERNAL int m0_rconfc_start_wait(struct m0_rconfc *rconfc,
                                     uint64_t          timeout_ns)
{
        struct rlock_ctx *rlx = M0_MEMBER(rconfc, rc_rlock_ctx);
        int               rc;

        M0_ENTRY("rconfc = %p, profile = "FID_F, rconfc,
                 FID_P(&rconfc->rc_profile));
        if (timeout_ns != M0_TIME_NEVER)
                rlx->rlc_timeout = timeout_ns;
        rc = m0_rconfc_start(rconfc);
        if (rc != 0)
                return M0_ERR(rc);

        m0_rconfc_lock(rconfc);
        rc = rconfc->rc_sm.sm_rc;
        if (rc == 0 && !m0_rconfc_is_preloaded(rconfc)) {
                m0_time_t deadline = timeout_ns == M0_TIME_NEVER ?
                        M0_TIME_NEVER : m0_time_from_now(0, timeout_ns);

                if (m0_sm_timedwait(&rconfc->rc_sm,
                                    M0_BITS(M0_RCS_IDLE, M0_RCS_FAILURE),
                                    deadline) == -ETIMEDOUT) {
                        rconfc_fail(rconfc, M0_ERR(-ETIMEDOUT));
                }
                /*
                 * Wait mat result in some error (failed to take lock,
                 * etc). Let's have this under control.
                 */
                rc = rconfc->rc_sm.sm_rc;
        }

        m0_rconfc_unlock(rconfc);
        return M0_RC(rc);
}

M0_INTERNAL void m0_rconfc_stop(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc %p", rconfc);
        m0_rconfc_lock(rconfc);
        m0_sm_timedwait(&rconfc->rc_sm, M0_BITS(M0_RCS_INIT, M0_RCS_IDLE,
                                                M0_RCS_FAILURE), M0_TIME_NEVER);
        m0_rconfc_unlock(rconfc);
        /*
         * Can't use rconfc_ast_post() here, because this AST can be already
         * posted (after observing read lock conflict, for example).
         */
        rconfc->rc_stop_ast.sa_cb    = rconfc_stop_ast_cb;
        rconfc->rc_stop_ast.sa_datum = rconfc;
        m0_sm_ast_post(rconfc->rc_sm.sm_grp, &rconfc->rc_stop_ast);
        M0_LEAVE();
}

M0_INTERNAL void m0_rconfc_stop_sync(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc = %p", rconfc);
        if (rconfc_state(rconfc) == M0_RCS_INIT &&
            !m0_confc_is_inited(&rconfc->rc_confc))
                goto leave;
        m0_rconfc_stop(rconfc);
        m0_rconfc_lock(rconfc);
        m0_sm_timedwait(&rconfc->rc_sm, M0_BITS(M0_RCS_FINAL), M0_TIME_NEVER);
        m0_rconfc_unlock(rconfc);
leave:
        M0_LEAVE();
}

M0_INTERNAL void m0_rconfc_fini(struct m0_rconfc *rconfc)
{
        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc != NULL);
        M0_PRE(rconfc->rc_rlock_ctx != NULL);
        M0_PRE(m0_confc_is_inited(&rconfc->rc_phony));

        m0_clink_del_lock(&rconfc->rc_ha_entrypoint_cl);
        m0_clink_fini(&rconfc->rc_ha_entrypoint_cl);

        m0_rconfc_lock(rconfc);
        if (rconfc_state(rconfc) == M0_RCS_INIT)
                /*
                 * looks like this rconfc instance never was started, so do
                 * internal cleanup prior to read lock context destruction and
                 * finalising state machine
                 */
                rconfc_stop_internal(rconfc);
        m0_rconfc_unlock(rconfc);
        m0_free(rconfc->rc_local_conf);
        m0_free(rconfc->rc_qctx);
        rlock_ctx_destroy(rconfc->rc_rlock_ctx);
        if (m0_confc_is_inited(&rconfc->rc_confc)) {
                m0_ha_client_del(&rconfc->rc_confc);
                m0_confc_fini(&rconfc->rc_confc);
        }
        rconfc_active_all_unlink(rconfc);
        rcnf_active_tlist_fini(&rconfc->rc_active);
        rconfc_herd_prune(rconfc);
        rcnf_herd_tlist_fini(&rconfc->rc_herd);
        m0_mutex_lock(&rconfc->rc_herd_lock);
        m0_chan_fini(&rconfc->rc_herd_chan);
        m0_mutex_unlock(&rconfc->rc_herd_lock);
        m0_mutex_fini(&rconfc->rc_herd_lock);
        m0_clink_fini(&rconfc->rc_herd_cl);
        _confc_phony_fini(&rconfc->rc_phony);
        m0_rconfc_lock(rconfc);
        m0_sm_fini(&rconfc->rc_sm);
        m0_rconfc_unlock(rconfc);

        M0_LEAVE();
}

M0_INTERNAL uint64_t m0_rconfc_ver_max_read(struct m0_rconfc *rconfc)
{
        uint64_t ver_max;

        M0_ENTRY("rconfc = %p", rconfc);
        M0_PRE(rconfc_state(rconfc) != M0_RCS_INIT);
        m0_rconfc_lock(rconfc);
        ver_max = m0_tl_fold(rcnf_herd, lnk, acc, &rconfc->rc_herd,
                             M0_CONF_VER_UNKNOWN,
                             max64(acc, _confc_ver_read(&lnk->rl_confc)));
        m0_rconfc_unlock(rconfc);
        M0_LEAVE("ver_max = %"PRIu64, ver_max);
        return ver_max;
}

M0_INTERNAL void m0_rconfc_fatal_cb_set(struct m0_rconfc *rconfc,
                                        m0_rconfc_cb_t    cb)
{
        M0_PRE(rconfc_is_locked(rconfc));
        M0_PRE(rconfc->rc_fatal_cb == NULL);
        rconfc->rc_fatal_cb = cb;
        if (rconfc_state(rconfc) == M0_RCS_FAILURE)
                /*
                 * Failure might occur between successful rconfc start and this
                 * callback setup. If occurred, callback is fired right away.
                 */
                cb(rconfc);
}

M0_INTERNAL int m0_rconfc_confd_endpoints(struct m0_rconfc   *rconfc,
                                          const char       ***eps)
{

        struct rconfc_link  *lnk;
        size_t               confd_eps_length;
        int                  i = 0;

        M0_PRE(rconfc_state(rconfc) != M0_RCS_INIT);
        M0_PRE(*eps == NULL);
        confd_eps_length = m0_tlist_length(&rcnf_herd_tl, &rconfc->rc_herd);
        M0_ALLOC_ARR(*eps, confd_eps_length + 1);
        m0_tl_for(rcnf_herd, &rconfc->rc_herd, lnk) {
                (*eps)[i] = m0_strdup(lnk->rl_confd_addr);
                if ((*eps)[i] == NULL)
                        goto fail;
                M0_CNT_INC(i);
        } m0_tl_endfor;
        (*eps)[i] = NULL;
        return i;
fail:
        m0_strings_free(*eps);
        return M0_ERR(-ENOMEM);
}

M0_INTERNAL int m0_rconfc_rm_endpoint(struct m0_rconfc *rconfc, char **ep)
{
        struct rlock_ctx *rlx;

        M0_PRE(rconfc_state(rconfc) != M0_RCS_INIT);
        M0_PRE(_0C(rconfc != NULL) && _0C(rconfc->rc_rlock_ctx != NULL) &&
               rlock_ctx_is_online(rconfc->rc_rlock_ctx));
        M0_PRE(*ep == NULL);

        rlx = rconfc->rc_rlock_ctx;
        *ep = m0_strdup(rlx->rlc_rm_addr);
        if (*ep == NULL)
                return M0_ERR(-ENOMEM);
        return M0_RC(0);
}

M0_INTERNAL void m0_rconfc_rm_fid(struct m0_rconfc *rconfc, struct m0_fid *out)
{
        struct rlock_ctx *rlx;

        M0_PRE(rconfc_state(rconfc) != M0_RCS_INIT);
        M0_PRE(_0C(rconfc != NULL) && _0C(rconfc->rc_rlock_ctx != NULL) &&
               rlock_ctx_is_online(rconfc->rc_rlock_ctx));
        M0_PRE(m0_fid_eq(out, &M0_FID0));

        rlx = rconfc->rc_rlock_ctx;
        *out = rlx->rlc_rm_fid;
}

M0_INTERNAL bool m0_rconfc_is_preloaded(struct m0_rconfc *rconfc)
{
        return rconfc_state(rconfc) == M0_RCS_INIT &&
                m0_confc_is_inited(&rconfc->rc_confc) &&
                rconfc->rc_local_conf != NULL;
}

static bool ha_clink_cb(struct m0_clink *clink)
{
        struct m0_rconfc                   *rconfc = container_of(clink,
                                                          struct m0_rconfc,
                                                          rc_ha_entrypoint_cl);
        struct m0_ha                       *ha  = m0_get()->i_ha;
        struct m0_ha_entrypoint_client     *ecl = &ha->h_entrypoint_client;
        enum m0_ha_entrypoint_client_state  state;
        int                                 rc;

        state = m0_ha_entrypoint_client_state_get(ecl);
        M0_ENTRY("ha_entrypoint_client_state=%s",
                 m0_sm_state_name(&ecl->ecl_sm, state));

        if (rconfc->rc_local_conf != NULL)
                return true;

        /*
         * As there is no external stop event coming to stop rconfc election,
         * rconfc re-election can go into a loop with only two confds like in
         * LDR-R1 and after a failure of one of the confd, quorum is not possible.
         * This is blocking the finalisation of motr processes. To fix this
         * issue, during shutdown when HA module in motr process is stopped,
         * rconfc is moved to failed state which then can be finalised. Ideally
         * rconfc should be stopped by sending shutdown command by external
         * entity like HA component.
         */
        if (state == M0_HEC_STOPPED) {
                m0_rconfc_lock(rconfc);
                /* rconfc is not started. */
                if (!M0_IN(rconfc_state(rconfc), (M0_RCS_INIT, M0_RCS_FAILURE)))
                        rconfc_fail(rconfc, -EINVAL);
                m0_rconfc_unlock(rconfc);
                M0_LEAVE();
                return true;
        }

        if (state == M0_HEC_AVAILABLE &&
            ecl->ecl_rep.hae_control != M0_HA_ENTRYPOINT_QUERY) {
                m0_rconfc_lock(rconfc);
                rc = m0_ha_entrypoint_rep_copy(&rconfc->rc_ha_entrypoint_rep,
                                               &ecl->ecl_rep);
                M0_ASSERT(rc == 0 || rc == -EINVAL);

                if (rc == -EINVAL)
                        rconfc_fail(rconfc, rc);
                else if (rconfc_state(rconfc) == M0_RCS_ENTRYPOINT_WAIT)
                        rconfc_ast_post(rconfc, rconfc_start_ast_cb);

                m0_rconfc_unlock(rconfc);
        }

        M0_LEAVE();
        return true;
}

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