interface.c

Go to the documentation of this file.

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

#include "ha/halon/interface.h"

#include <stdlib.h>             /* calloc */

#include "lib/types.h"          /* m0_bcount_t */
#include "lib/misc.h"           /* M0_IS0 */
#include "lib/memory.h"         /* M0_ALLOC_PTR */
#include "lib/bob.h"            /* M0_BOB_DEFINE */
#include "lib/errno.h"          /* ENOSYS */
#include "lib/string.h"         /* strcmp */
#include "lib/uuid.h"           /* m0_node_uuid_string_set */
#include "lib/thread.h"         /* m0_process */

#include "net/net.h"            /* M0_NET_TM_RECV_QUEUE_DEF_LEN */
#include "net/buffer_pool.h"    /* m0_net_buffer_pool */
#include "fid/fid.h"            /* m0_fid */
#include "module/instance.h"    /* m0 */
#include "reqh/reqh.h"          /* m0_reqh */
#include "reqh/reqh_service.h"  /* m0_reqh_service_setup */
#include "rpc/rpc.h"            /* m0_rpc_bufs_nr */
#include "rpc/rpc_machine.h"    /* m0_rpc_machine */
#include "sm/sm.h"              /* m0_sm */
#include "rm/rm_service.h"      /* m0_rms_type */
#include "spiel/spiel.h"        /* m0_spiel */
#include "cm/cm.h"              /* m0_sns_cm_repair_trigger_fop_init */
#include "conf/ha.h"            /* m0_conf_ha_service_event_post */
#include "conf/obj.h"           /* M0_CONF_PROCESS_TYPE */

#include "motr/init.h"          /* m0_init */
#include "motr/magic.h"         /* M0_HALON_INTERFACE_MAGIC */
#include "motr/version.h"       /* m0_build_info_get */

#include "ha/msg.h"             /* m0_ha_msg_debug_print */
#include "ha/ha.h"              /* m0_ha */
#include "ha/entrypoint_fops.h" /* m0_ha_entrypoint_req */
#include "ha/dispatcher.h"      /* m0_ha_dispatcher */
#include "ha/note.h"            /* M0_HA_NVEC_SET */


enum {
        HALON_INTERFACE_EP_BUF        = 0x40,
        HALON_INTERFACE_NVEC_SIZE_MAX = 0x1000,
};

struct m0_halon_interface_cfg {
        const char      *hic_build_git_rev_id;
        const char      *hic_build_configure_opts;
        bool             hic_disable_compat_check;
        char            *hic_local_rpc_endpoint;
        struct m0_fid    hic_process_fid;
        struct m0_fid    hic_ha_service_fid;
        struct m0_fid    hic_rm_service_fid;
        void           (*hic_entrypoint_request_cb)
                (struct m0_halon_interface         *hi,
                 const struct m0_uint128           *req_id,
                 const char                        *remote_rpc_endpoint,
                 const struct m0_fid               *process_fid,
                 const char                        *git_rev_id,
                 uint64_t                           pid,
                 bool                               first_request);
        void           (*hic_msg_received_cb)
                (struct m0_halon_interface *hi,
                 struct m0_ha_link         *hl,
                 const struct m0_ha_msg    *msg,
                 uint64_t                   tag);
        void           (*hic_msg_is_delivered_cb)
                (struct m0_halon_interface *hi,
                 struct m0_ha_link         *hl,
                 uint64_t                   tag);
        void           (*hic_msg_is_not_delivered_cb)
                (struct m0_halon_interface *hi,
                 struct m0_ha_link         *hl,
                 uint64_t                   tag);
        void           (*hic_link_connected_cb)
                (struct m0_halon_interface *hi,
                 const struct m0_uint128   *req_id,
                 struct m0_ha_link         *link);
        void           (*hic_link_reused_cb)
                (struct m0_halon_interface *hi,
                 const struct m0_uint128   *req_id,
                 struct m0_ha_link         *link);
        void           (*hic_link_absent_cb)
                (struct m0_halon_interface *hi,
                 const struct m0_uint128   *req_id);
        void           (*hic_link_is_disconnecting_cb)
                (struct m0_halon_interface *hi,
                 struct m0_ha_link         *link);
        void           (*hic_link_disconnected_cb)
                (struct m0_halon_interface *hi,
                 struct m0_ha_link         *link);

        uint32_t         hic_tm_nr;
        uint32_t         hic_bufs_nr;
        uint32_t         hic_colour;
        m0_bcount_t      hic_max_msg_size;
        uint32_t         hic_queue_len;
        struct m0_ha_cfg hic_ha_cfg;
        struct m0_ha_dispatcher_cfg hic_dispatcher_cfg;
        bool             hic_log_entrypoint;
        bool             hic_log_link;
        bool             hic_log_msg;
};

enum m0_halon_interface_level {
        M0_HALON_INTERFACE_LEVEL_ASSIGNS,
        M0_HALON_INTERFACE_LEVEL_NET_DOMAIN,
        M0_HALON_INTERFACE_LEVEL_NET_BUFFER_POOL,
        M0_HALON_INTERFACE_LEVEL_REQH_INIT,
        M0_HALON_INTERFACE_LEVEL_REQH_START,
        M0_HALON_INTERFACE_LEVEL_RPC_MACHINE,
        M0_HALON_INTERFACE_LEVEL_HA_INIT,
        M0_HALON_INTERFACE_LEVEL_DISPATCHER,
        M0_HALON_INTERFACE_LEVEL_HA_START,
        M0_HALON_INTERFACE_LEVEL_HA_CONNECT,
        M0_HALON_INTERFACE_LEVEL_INSTANCE_SET,
        M0_HALON_INTERFACE_LEVEL_EVENTS_STARTING,
        M0_HALON_INTERFACE_LEVEL_RM_SETUP,
        M0_HALON_INTERFACE_LEVEL_SPIEL_INIT,
        M0_HALON_INTERFACE_LEVEL_SNS_CM_TRIGGER_FOPS,
        M0_HALON_INTERFACE_LEVEL_EVENTS_STARTED,
        M0_HALON_INTERFACE_LEVEL_STARTED,
};

enum m0_halon_interface_state {
        M0_HALON_INTERFACE_STATE_UNINITIALISED,
        M0_HALON_INTERFACE_STATE_INITIALISED,
        M0_HALON_INTERFACE_STATE_WORKING,
        M0_HALON_INTERFACE_STATE_FINALISED,
};

struct m0_halon_interface_internal {
        struct m0_halon_interface     *hii_hi;
        struct m0                      hii_instance;
        struct m0_halon_interface_cfg  hii_cfg;
        struct m0_module               hii_module;
        struct m0_net_domain           hii_net_domain;
        struct m0_net_buffer_pool      hii_net_buffer_pool;
        struct m0_reqh                 hii_reqh;
        struct m0_rpc_machine          hii_rpc_machine;
        struct m0_ha                   hii_ha;
        struct m0_ha_dispatcher        hii_dispatcher;
        struct m0_sm_group             hii_sm_group;
        struct m0_sm                   hii_sm;
        uint64_t                       hii_magix;
        struct m0_ha_link             *hii_outgoing_link;
        struct m0_reqh_service        *hii_rm_service;
        struct m0_spiel                hii_spiel;
        uint64_t                       hii_nvec_size;
        struct m0_ha_note              hii_nvec[HALON_INTERFACE_NVEC_SIZE_MAX];
};

static const struct m0_bob_type halon_interface_bob_type = {
        .bt_name         = "halon interface",
        .bt_magix_offset = M0_MAGIX_OFFSET(struct m0_halon_interface_internal,
                                           hii_magix),
        .bt_magix        = M0_HALON_INTERFACE_MAGIC,
};
M0_BOB_DEFINE(static, &halon_interface_bob_type, m0_halon_interface_internal);

static bool
halon_interface_is_compatible(struct m0_halon_interface *hi,
                              const char                *build_git_rev_id,
                              const char                *build_configure_opts,
                              bool                       disable_compat_check)
{
        const struct m0_build_info *bi = m0_build_info_get();

        M0_ENTRY("build_git_rev_id=%s build_configure_opts=%s "
                 "disable_compat_check=%d", build_git_rev_id,
                 build_configure_opts, !!disable_compat_check);
        if (disable_compat_check)
                return true;
        if (!m0_streq(bi->bi_git_rev_id, build_git_rev_id)) {
                M0_LOG(M0_ERROR, "The loaded motr library (%s) "
                       "is not the expected one (%s)", bi->bi_git_rev_id,
                       build_git_rev_id);
                return false;
        }
        if (!m0_streq(bi->bi_configure_opts, build_configure_opts)) {
                M0_LOG(M0_ERROR, "The configuration options of the loaded "
                       "motr library (%s) do not match the expected ones (%s)",
                       bi->bi_configure_opts, build_configure_opts);
                return false;
        }
        return true;
}

static void
halon_interface_parse_debug_options(struct m0_halon_interface *hi,
                                    const char                *debug_options)
{
        struct m0_halon_interface_cfg *hi_cfg = &hi->hif_internal->hii_cfg;

        M0_ENTRY("hi=%p debug_options=%s", hi, debug_options);

        if (debug_options == NULL)
                return;
        hi_cfg->hic_log_entrypoint = strstr(debug_options, "log-entrypoint")
                                                                        != NULL;
        hi_cfg->hic_log_link       = strstr(debug_options, "log-link") != NULL;
        hi_cfg->hic_log_msg        = strstr(debug_options, "log-msg") != NULL;
}

static struct m0_halon_interface_internal *
halon_interface_ha2hii(struct m0_ha *ha)
{
        struct m0_halon_interface_internal *hii;

        hii = bob_of(ha, struct m0_halon_interface_internal, hii_ha,
                     &halon_interface_bob_type);
        M0_ASSERT(m0_get() == &hii->hii_instance);
        return hii;
}

static void
halon_interface_process_failure_check(struct m0_halon_interface_internal *hii,
                                      struct m0_ha_msg                   *msg)
{
        struct m0_ha_msg_nvec *nvec;
        struct m0_ha_note     *note;
        uint64_t               i;
        uint64_t               j;
        bool                   just_added;

        if (m0_ha_msg_type_get(msg) != M0_HA_MSG_NVEC)
                return;
        nvec = &msg->hm_data.u.hed_nvec;
        if (nvec->hmnv_type != M0_HA_NVEC_SET)
                return;
        for (i = 0; i < nvec->hmnv_nr; ++i) {
                note = &nvec->hmnv_arr.hmna_arr[i];
                if (m0_fid_tget(&note->no_id) ==
                    M0_CONF_PROCESS_TYPE.cot_ftype.ft_id) {
                        /*
                         * XXX Detect HA state duplications here and ignore the
                         * note if HA state for the process doesn't change.
                         * This allows Hare to be simple and always send as many
                         * HA states as it wants without worrying about sending
                         * each state only once.
                         *
                         * The duplicate detection code has to be removed after
                         * Hare can send states in a proper way.
                         */
                        for (j = 0; j < hii->hii_nvec_size; ++j) {
                                if (m0_fid_eq(&note->no_id,
                                              &hii->hii_nvec[j].no_id))
                                        break;
                        }
                        just_added = false;
                        if (j == hii->hii_nvec_size) {
                                M0_ASSERT_INFO(hii->hii_nvec_size <
                                               ARRAY_SIZE(hii->hii_nvec),
                                               "Currently we don't support "
                                               "more than %zu processes "
                                               "connected to a single "
                                               "m0_halon_interface.",
                                               ARRAY_SIZE(hii->hii_nvec));
                                hii->hii_nvec[hii->hii_nvec_size] =
                                        (struct m0_ha_note){
                                                .no_id    = note->no_id,
                                                .no_state = M0_NC_UNKNOWN,
                                };
                                M0_LOG(M0_DEBUG, "hii_nvec[%" PRIu64 "] = "
                                       "{.no_id = "FID_F", ...}",
                                       hii->hii_nvec_size,
                                       FID_P(&hii->hii_nvec[
                                             hii->hii_nvec_size].no_id));
                                ++hii->hii_nvec_size;
                                just_added = true;
                        }
                        /*
                         * Ignore process state if it hasn't changed and the
                         * hmnv_ignore_same_state flag is set.
                         */
                        if (note->no_state == hii->hii_nvec[j].no_state &&
                            nvec->hmnv_ignore_same_state) {
                                M0_LOG(M0_DEBUG, "ignoring the same state for "
                                       FID_F, FID_P(&note->no_id));
                                continue;
                        }
                        if (note->no_state == M0_NC_FAILED) {
                                if (just_added) {
                                        M0_LOG(M0_DEBUG,
                                               "Ignoring M0_NC_FAILED as "
                                               "it's the 1st notification "
                                               "for no_id="FID_F,
                                               FID_P(&note->no_id));
                                } else {
                                        M0_LOG(M0_DEBUG, "no_id="FID_F,
                                               FID_P(&note->no_id));
                                        if (hii->hii_cfg.hic_log_link) {
                                                M0_LOG(M0_WARN, "disconnecting "
                                                       "all links to "
                                                       "no_id="FID_F,
                                                       FID_P(&note->no_id));
                                        }
                                        m0_ha_process_failed(&hii->hii_ha,
                                                             &note->no_id);
                                }
                        }
                        hii->hii_nvec[j].no_state = note->no_state;
                }
        }
}

static void
halon_interface_entrypoint_request_cb(struct m0_ha                      *ha,
                                      const struct m0_ha_entrypoint_req *req,
                                      const struct m0_uint128           *req_id)
{
        struct m0_halon_interface_internal *hii;

        hii = bob_of(ha, struct m0_halon_interface_internal, hii_ha,
                     &halon_interface_bob_type);
        M0_ENTRY("hi=%p req=%p req_id="U128X_F" remote_rpc_endpoint=%s "
                 "process_fid="FID_F,
                 hii->hii_hi, req, U128_P(req_id), req->heq_rpc_endpoint,
                 FID_P(&req->heq_process_fid));
        M0_LOG(M0_DEBUG, "git_rev_id=%s generation=%" PRIu64 " pid=%"PRIu64,
               req->heq_git_rev_id, req->heq_generation, req->heq_pid);
        if (hii->hii_cfg.hic_log_entrypoint) {
                M0_LOG(M0_WARN, "req_id="U128X_F" remote_rpc_endpoint=%s "
                 "process_fid="FID_F" git_rev_id=%s generation=%" PRIu64 " "
                 "pid=%"PRIu64, U128_P(req_id), req->heq_rpc_endpoint,
                 FID_P(&req->heq_process_fid), req->heq_git_rev_id,
                 req->heq_generation, req->heq_pid);
        }
        hii->hii_cfg.hic_entrypoint_request_cb(hii->hii_hi, req_id,
                                               req->heq_rpc_endpoint,
                                               &req->heq_process_fid,
                                               req->heq_git_rev_id,
                                               req->heq_pid,
                                               req->heq_first_request);
        M0_LEAVE();
}

static void
halon_interface_entrypoint_replied_cb(struct m0_ha                *ha,
                                      struct m0_ha_entrypoint_rep *rep)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);

        M0_ENTRY("hii=%p ha=%p rep=%p", hii, ha, rep);
        /*
         * Nothing to do here for now.
         * In the future some default handler should be called.
         */
        M0_LEAVE();
}

static void halon_interface_msg_received_cb(struct m0_ha      *ha,
                                            struct m0_ha_link *hl,
                                            struct m0_ha_msg  *msg,
                                            uint64_t           tag)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%"PRIu64,
                 hii->hii_hi, &hii->hii_ha, hl, ep, msg, msg->hm_epoch, tag);
        m0_ha_msg_debug_print(msg, __func__);
        if (hl != hii->hii_outgoing_link) {
                if (hii->hii_cfg.hic_log_msg) {
                        M0_LOG(M0_WARN, "hl=%p ep=%s epoch=%" PRIu64 " "
                               "tag=%" PRIu64 " type=%"PRIu64,
                               hl, ep, msg->hm_epoch, tag,
                               msg->hm_data.hed_type);
                }
                hii->hii_cfg.hic_msg_received_cb(hii->hii_hi, hl, msg, tag);
        } else {
                m0_ha_dispatcher_handle(&hii->hii_dispatcher, ha, hl, msg, tag);
                halon_interface_process_failure_check(hii, msg);
                m0_ha_delivered(ha, hl, msg);
        }
        M0_LEAVE("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%"PRIu64,
                 hii->hii_hi, &hii->hii_ha, hl, ep, msg, msg->hm_epoch, tag);
}

static void halon_interface_msg_is_delivered_cb(struct m0_ha      *ha,
                                                struct m0_ha_link *hl,
                                                uint64_t           tag)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p hl=%p ep=%s tag=%"PRIu64, hii, ha, hl, ep, tag);
        if (hl != hii->hii_outgoing_link) {
                if (hii->hii_cfg.hic_log_msg)
                        M0_LOG(M0_WARN, "hl=%p ep=%s tag=%"PRIu64, hl, ep, tag);
                hii->hii_cfg.hic_msg_is_delivered_cb(hii->hii_hi, hl, tag);
        }
        M0_LEAVE("hii=%p ha=%p hl=%p ep=%s tag=%"PRIu64, hii, ha, hl, ep, tag);
}

static void halon_interface_msg_is_not_delivered_cb(struct m0_ha      *ha,
                                                    struct m0_ha_link *hl,
                                                    uint64_t           tag)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p hl=%p ep=%s tag=%"PRIu64, hii, ha, hl, ep, tag);
        if (hl != hii->hii_outgoing_link) {
                if (hii->hii_cfg.hic_log_msg)
                        M0_LOG(M0_WARN, "hl=%p ep=%s tag=%"PRIu64, hl, ep, tag);
                hii->hii_cfg.hic_msg_is_not_delivered_cb(hii->hii_hi, hl, tag);
        }
        M0_LEAVE("hii=%p ha=%p hl=%p ep=%s tag=%"PRIu64, hii, ha, hl, ep, tag);
}

static void halon_interface_link_connected_cb(struct m0_ha            *ha,
                                              const struct m0_uint128 *req_id,
                                              struct m0_ha_link       *hl)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p req_id="U128X_F" hl=%p ep=%s", hii, ha,
                 U128_P(req_id), hl, ep);
        if (hii->hii_cfg.hic_log_link) {
                M0_LOG(M0_WARN, "req_id="U128X_F" hl=%p ep=%s",
                       U128_P(req_id), hl, ep);
        }
        hii->hii_cfg.hic_link_connected_cb(hii->hii_hi, req_id, hl);
        M0_LEAVE("hii=%p ha=%p req_id="U128X_F" hl=%p ep=%s", hii, ha,
                 U128_P(req_id), hl, ep);
}

static void halon_interface_link_reused_cb(struct m0_ha            *ha,
                                           const struct m0_uint128 *req_id,
                                           struct m0_ha_link       *hl)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p req_id="U128X_F" hl=%p ep=%s", hii, ha,
                 U128_P(req_id), hl, ep);
        if (hii->hii_cfg.hic_log_link) {
                M0_LOG(M0_WARN, "req_id="U128X_F" hl=%p ep=%s",
                       U128_P(req_id), hl, ep);
        }
        hii->hii_cfg.hic_link_reused_cb(hii->hii_hi, req_id, hl);
        M0_LEAVE("hii=%p ha=%p req_id="U128X_F" hl=%p ep=%s", hii, ha,
                 U128_P(req_id), hl, ep);
}

static void halon_interface_link_absent_cb(struct m0_ha            *ha,
                                           const struct m0_uint128 *req_id)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        M0_ENTRY("hii=%p ha=%p req_id="U128X_F, hii, ha, U128_P(req_id));
        if (hii->hii_cfg.hic_log_link) {
                M0_LOG(M0_WARN, "req_id="U128X_F" ep=%s", U128_P(req_id), ep);
        }
        hii->hii_cfg.hic_link_absent_cb(hii->hii_hi, req_id);
        M0_LEAVE("hii=%p ha=%p req_id="U128X_F, hii, ha, U128_P(req_id));
}

static void halon_interface_link_is_disconnecting_cb(struct m0_ha      *ha,
                                                     struct m0_ha_link *hl)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p hl=%p ep=%s", hii, ha, hl, ep);
        if (hii->hii_cfg.hic_log_link)
                M0_LOG(M0_WARN, "hl=%p ep=%s", hl, ep);
        hii->hii_cfg.hic_link_is_disconnecting_cb(hii->hii_hi, hl);
        M0_LEAVE("hii=%p ha=%p hl=%p ep=%s", hii, ha, hl, ep);
}

static void halon_interface_link_disconnected_cb(struct m0_ha      *ha,
                                                 struct m0_ha_link *hl)
{
        struct m0_halon_interface_internal *hii = halon_interface_ha2hii(ha);
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hii=%p ha=%p hl=%p ep=%s", hii, ha, hl, ep);
        if (hii->hii_cfg.hic_log_link)
                M0_LOG(M0_WARN, "hl=%p ep=%s", hl, ep);
        hii->hii_cfg.hic_link_disconnected_cb(hii->hii_hi, hl);
        M0_LEAVE("hii=%p ha=%p hl=%p ep=%s", hii, ha, hl, ep);
}

static const struct m0_ha_ops halon_interface_ha_ops = {
        .hao_entrypoint_request    = &halon_interface_entrypoint_request_cb,
        .hao_entrypoint_replied    = &halon_interface_entrypoint_replied_cb,
        .hao_msg_received          = &halon_interface_msg_received_cb,
        .hao_msg_is_delivered      = &halon_interface_msg_is_delivered_cb,
        .hao_msg_is_not_delivered  = &halon_interface_msg_is_not_delivered_cb,
        .hao_link_connected        = &halon_interface_link_connected_cb,
        .hao_link_reused           = &halon_interface_link_reused_cb,
        .hao_link_absent           = &halon_interface_link_absent_cb,
        .hao_link_is_disconnecting = &halon_interface_link_is_disconnecting_cb,
        .hao_link_disconnected     = &halon_interface_link_disconnected_cb,
};

static struct m0_sm_state_descr halon_interface_states[] = {
        [M0_HALON_INTERFACE_STATE_UNINITIALISED] = {
                .sd_flags = M0_SDF_INITIAL,
                .sd_name = "M0_HALON_INTERFACE_STATE_UNINITIALISED",
                .sd_allowed = M0_BITS(M0_HALON_INTERFACE_STATE_INITIALISED),
        },
        [M0_HALON_INTERFACE_STATE_INITIALISED] = {
                .sd_name = "M0_HALON_INTERFACE_STATE_INITIALISED",
                .sd_allowed = M0_BITS(M0_HALON_INTERFACE_STATE_WORKING,
                                      M0_HALON_INTERFACE_STATE_FINALISED),
        },
        [M0_HALON_INTERFACE_STATE_WORKING] = {
                .sd_name = "M0_HALON_INTERFACE_STATE_WORKING",
                .sd_allowed = M0_BITS(M0_HALON_INTERFACE_STATE_INITIALISED),
        },
        [M0_HALON_INTERFACE_STATE_FINALISED] = {
                .sd_flags = M0_SDF_TERMINAL,
                .sd_name = "M0_HALON_INTERFACE_STATE_FINALISED",
                .sd_allowed = 0,
        },
};

static struct m0_sm_conf halon_interface_sm_conf = {
        .scf_name      = "m0_halon_interface_internal::hii_sm",
        .scf_nr_states = ARRAY_SIZE(halon_interface_states),
        .scf_state     = halon_interface_states,
};

int m0_halon_interface_init(struct m0_halon_interface **hi_out,
                            const char                 *build_git_rev_id,
                            const char                 *build_configure_opts,
                            const char                 *debug_options,
                            const char                 *node_uuid)
{
        struct m0_halon_interface_internal *hii;
        bool                                disable_compat_check;
        int                                 rc;

        disable_compat_check = debug_options != NULL &&
                strstr(debug_options, "disable-compatibility-check") != NULL;
        if (!halon_interface_is_compatible(NULL, build_git_rev_id,
                                           build_configure_opts,
                                           disable_compat_check))
                return M0_ERR(-EINVAL);

        /* M0_ALLOC_PTR() can't be used before m0_init() */
        *hi_out = calloc(1, sizeof **hi_out);
        hii = calloc(1, sizeof *hii);
        if (*hi_out == NULL || hii == NULL) {
                free(*hi_out);
                free(hii);
                return M0_ERR(-ENOMEM);
        }
        (*hi_out)->hif_internal = hii;
        (*hi_out)->hif_internal->hii_hi = *hi_out;
        m0_halon_interface_internal_bob_init(hii);
        m0_node_uuid_string_set(node_uuid);
        rc = m0_init(&hii->hii_instance);
        if (rc != 0) {
                free(hii);
                return M0_ERR(rc);
        }
        halon_interface_parse_debug_options(*hi_out, debug_options);
        m0_sm_group_init(&hii->hii_sm_group);
        m0_sm_init(&hii->hii_sm, &halon_interface_sm_conf,
                   M0_HALON_INTERFACE_STATE_UNINITIALISED, &hii->hii_sm_group);
        m0_sm_group_lock(&hii->hii_sm_group);
        m0_sm_state_set(&hii->hii_sm, M0_HALON_INTERFACE_STATE_INITIALISED);
        m0_sm_group_unlock(&hii->hii_sm_group);
        return M0_RC(0);
}

void m0_halon_interface_fini(struct m0_halon_interface *hi)
{
        struct m0_halon_interface_internal *hii;

        M0_ENTRY("hi=%p", hi);

        hii = hi->hif_internal;
        m0_sm_group_lock(&hii->hii_sm_group);
        m0_sm_state_set(&hii->hii_sm, M0_HALON_INTERFACE_STATE_FINALISED);
        m0_sm_fini(&hii->hii_sm);
        m0_sm_group_unlock(&hii->hii_sm_group);
        m0_sm_group_fini(&hii->hii_sm_group);
        m0_fini();
        m0_halon_interface_internal_bob_fini(hii);
        free(hii);
        M0_LEAVE();
}

static void
halon_interface_process_event(struct m0_halon_interface_internal *hii,
                              enum m0_conf_ha_process_event       event)
{
        m0_conf_ha_process_event_post(&hii->hii_ha, hii->hii_outgoing_link,
                                      &hii->hii_cfg.hic_process_fid,
                                      m0_process(), event,
                                      M0_CONF_HA_PROCESS_M0D);
}

static void
halon_interface_service_event(struct m0_halon_interface_internal *hii,
                              enum m0_conf_ha_service_event       event)
{
        m0_conf_ha_service_event_post(&hii->hii_ha, hii->hii_outgoing_link,
                                      &hii->hii_cfg.hic_process_fid,
                                      &hii->hii_cfg.hic_ha_service_fid,
                                      &hii->hii_cfg.hic_ha_service_fid,
                                      m0_process(),
                                      event, M0_CST_HA);
}

static const struct m0_modlev halon_interface_levels[];

static int halon_interface_level_enter(struct m0_module *module)
{
        struct m0_halon_interface_internal *hii;
        enum m0_halon_interface_level       level = module->m_cur + 1;

        hii = bob_of(module, struct m0_halon_interface_internal, hii_module,
                     &halon_interface_bob_type);
        M0_ENTRY("hii=%p level=%d %s", hii, level,
                 halon_interface_levels[level].ml_name);
        switch (level) {
        case M0_HALON_INTERFACE_LEVEL_ASSIGNS:
                /*
                 * Zero all data structures initialised later to allow
                 * m0_halon_interface_start() after m0_halon_interface_stop().
                 */
                M0_SET0(&hii->hii_net_domain);
                M0_SET0(&hii->hii_net_buffer_pool);
                M0_SET0(&hii->hii_reqh);
                M0_SET0(&hii->hii_rpc_machine);
                M0_SET0(&hii->hii_ha);
                M0_SET0(&hii->hii_dispatcher);
                hii->hii_outgoing_link = NULL;
                hii->hii_rm_service    = NULL;
                M0_SET0(&hii->hii_spiel);
                hii->hii_nvec_size     = 0;

                hii->hii_cfg.hic_tm_nr        = 1;
                hii->hii_cfg.hic_bufs_nr      = 100;
                hii->hii_cfg.hic_colour       = M0_BUFFER_ANY_COLOUR;
                hii->hii_cfg.hic_max_msg_size = 1UL << 17;
                hii->hii_cfg.hic_queue_len    = 100;
                hii->hii_cfg.hic_ha_cfg = (struct m0_ha_cfg){
                        .hcf_ops         = halon_interface_ha_ops,
                        .hcf_rpc_machine = &hii->hii_rpc_machine,
                        .hcf_addr        = hii->hii_cfg.hic_local_rpc_endpoint,
                        .hcf_reqh        = &hii->hii_reqh,
                        .hcf_process_fid = hii->hii_cfg.hic_process_fid,
                };
                hii->hii_cfg.hic_dispatcher_cfg = (struct m0_ha_dispatcher_cfg){
                        .hdc_enable_note      = true,
                        .hdc_enable_keepalive = true,
                        .hdc_enable_fvec      = true,
                };
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_NET_DOMAIN:
                return M0_RC(m0_net_domain_init(&hii->hii_net_domain,
                                                m0_net_xprt_default_get()));
        case M0_HALON_INTERFACE_LEVEL_NET_BUFFER_POOL:
                return M0_RC(m0_rpc_net_buffer_pool_setup(
                                &hii->hii_net_domain, &hii->hii_net_buffer_pool,
                                hii->hii_cfg.hic_bufs_nr,
                                hii->hii_cfg.hic_tm_nr));
        case M0_HALON_INTERFACE_LEVEL_REQH_INIT:
                return M0_RC(M0_REQH_INIT(&hii->hii_reqh,
                                          .rhia_dtm          = (void*)1,
                                          .rhia_mdstore      = (void*)1,
                                          .rhia_fid          =
                                                &hii->hii_cfg.hic_process_fid));
        case M0_HALON_INTERFACE_LEVEL_REQH_START:
                m0_reqh_start(&hii->hii_reqh);
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_RPC_MACHINE:
                return M0_RC(m0_rpc_machine_init(
                                &hii->hii_rpc_machine, &hii->hii_net_domain,
                                 hii->hii_cfg.hic_local_rpc_endpoint,
                                &hii->hii_reqh, &hii->hii_net_buffer_pool,
                                 hii->hii_cfg.hic_colour,
                                 hii->hii_cfg.hic_max_msg_size,
                                 hii->hii_cfg.hic_queue_len));
        case M0_HALON_INTERFACE_LEVEL_HA_INIT:
                return M0_RC(m0_ha_init(&hii->hii_ha,
                                        &hii->hii_cfg.hic_ha_cfg));
        case M0_HALON_INTERFACE_LEVEL_DISPATCHER:
                return M0_RC(m0_ha_dispatcher_init(&hii->hii_dispatcher,
                                           &hii->hii_cfg.hic_dispatcher_cfg));
        case M0_HALON_INTERFACE_LEVEL_HA_START:
                return M0_RC(m0_ha_start(&hii->hii_ha));
        case M0_HALON_INTERFACE_LEVEL_HA_CONNECT:
                hii->hii_outgoing_link = m0_ha_connect(&hii->hii_ha);
                M0_LOG(M0_DEBUG, "hii_outgoing_link=%p",
                       hii->hii_outgoing_link);
                return hii->hii_outgoing_link == NULL ? M0_ERR(-EINVAL) :
                                                        M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_INSTANCE_SET:
                M0_ASSERT(m0_get()->i_ha      == NULL);
                M0_ASSERT(m0_get()->i_ha_link == NULL);
                m0_get()->i_ha      = &hii->hii_ha;
                m0_get()->i_ha_link =  hii->hii_outgoing_link;
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_EVENTS_STARTING:
                halon_interface_process_event(hii, M0_CONF_HA_PROCESS_STARTING);
                halon_interface_service_event(hii, M0_CONF_HA_SERVICE_STARTING);
                halon_interface_service_event(hii, M0_CONF_HA_SERVICE_STARTED);
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_RM_SETUP:
                return M0_RC(m0_reqh_service_setup(&hii->hii_rm_service,
                                                   &m0_rms_type,
                                                   &hii->hii_reqh, NULL,
                                           &hii->hii_cfg.hic_rm_service_fid));
        case M0_HALON_INTERFACE_LEVEL_SPIEL_INIT:
                return M0_RC(m0_spiel_init(&hii->hii_spiel, &hii->hii_reqh));
        case M0_HALON_INTERFACE_LEVEL_SNS_CM_TRIGGER_FOPS:
                m0_sns_cm_repair_trigger_fop_init();
                m0_sns_cm_rebalance_trigger_fop_init();
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_EVENTS_STARTED:
                halon_interface_process_event(hii, M0_CONF_HA_PROCESS_STARTED);
                /*
                   For HAX daemon, M0_NC_DTM_RECOVERING state is transient,
                   sending M0_CONF_HA_PROCESS_DTM_RECOVERED just after
                   M0_CONF_HA_PROCESS_STARTED.

                halon_interface_process_event(hii,
                                              M0_CONF_HA_PROCESS_DTM_RECOVERED);
                */
                return M0_RC(0);
        case M0_HALON_INTERFACE_LEVEL_STARTED:
                return M0_ERR(-ENOSYS);
        }
        return M0_ERR(-ENOSYS);
}

static void halon_interface_level_leave(struct m0_module *module)
{
        struct m0_halon_interface_internal *hii;
        enum m0_halon_interface_level       level = module->m_cur;

        hii = bob_of(module, struct m0_halon_interface_internal, hii_module,
                     &halon_interface_bob_type);
        M0_ENTRY("hii=%p level=%d", hii, level);
        switch (level) {
        case M0_HALON_INTERFACE_LEVEL_ASSIGNS:
                break;
        case M0_HALON_INTERFACE_LEVEL_NET_DOMAIN:
                m0_net_domain_fini(&hii->hii_net_domain);
                break;
        case M0_HALON_INTERFACE_LEVEL_NET_BUFFER_POOL:
                m0_rpc_net_buffer_pool_cleanup(&hii->hii_net_buffer_pool);
                break;
        case M0_HALON_INTERFACE_LEVEL_REQH_INIT:
                m0_reqh_fini(&hii->hii_reqh);
                break;
        case M0_HALON_INTERFACE_LEVEL_REQH_START:
                m0_reqh_services_terminate(&hii->hii_reqh);
                break;
        case M0_HALON_INTERFACE_LEVEL_RPC_MACHINE:
                m0_reqh_shutdown_wait(&hii->hii_reqh);
                m0_rpc_machine_fini(&hii->hii_rpc_machine);
                break;
        case M0_HALON_INTERFACE_LEVEL_HA_INIT:
                m0_ha_fini(&hii->hii_ha);
                break;
        case M0_HALON_INTERFACE_LEVEL_DISPATCHER:
                m0_ha_dispatcher_fini(&hii->hii_dispatcher);
                break;
        case M0_HALON_INTERFACE_LEVEL_HA_START:
                m0_ha_stop(&hii->hii_ha);
                break;
        case M0_HALON_INTERFACE_LEVEL_HA_CONNECT:
                m0_ha_flush(&hii->hii_ha, m0_ha_outgoing_link(&hii->hii_ha));
                m0_ha_disconnect(&hii->hii_ha);
                break;
        case M0_HALON_INTERFACE_LEVEL_INSTANCE_SET:
                M0_ASSERT(m0_get()->i_ha      == &hii->hii_ha);
                M0_ASSERT(m0_get()->i_ha_link ==  hii->hii_outgoing_link);
                m0_get()->i_ha      = NULL;
                m0_get()->i_ha_link = NULL;
                break;
        case M0_HALON_INTERFACE_LEVEL_EVENTS_STARTING:
                halon_interface_service_event(hii, M0_CONF_HA_SERVICE_STOPPING);
                halon_interface_service_event(hii, M0_CONF_HA_SERVICE_STOPPED);
                halon_interface_process_event(hii, M0_CONF_HA_PROCESS_STOPPED);
                break;
        case M0_HALON_INTERFACE_LEVEL_RM_SETUP:
                m0_reqh_service_quit(hii->hii_rm_service);
                break;
        case M0_HALON_INTERFACE_LEVEL_SPIEL_INIT:
                m0_spiel_fini(&hii->hii_spiel);
                break;
        case M0_HALON_INTERFACE_LEVEL_SNS_CM_TRIGGER_FOPS:
                m0_sns_cm_rebalance_trigger_fop_fini();
                m0_sns_cm_repair_trigger_fop_fini();
                break;
        case M0_HALON_INTERFACE_LEVEL_EVENTS_STARTED:
                halon_interface_process_event(hii, M0_CONF_HA_PROCESS_STOPPING);
                break;
        case M0_HALON_INTERFACE_LEVEL_STARTED:
                M0_IMPOSSIBLE("can't be here");
                break;
        }
        M0_LEAVE();
}

static const struct m0_modlev halon_interface_levels[] = {
        [M0_HALON_INTERFACE_LEVEL_ASSIGNS] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_ASSIGNS",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_NET_DOMAIN] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_NET_DOMAIN",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_NET_BUFFER_POOL] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_NET_BUFFER_POOL",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_REQH_INIT] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_REQH_INIT",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_REQH_START] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_REQH_START",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_RPC_MACHINE] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_RPC_MACHINE",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_HA_INIT] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_HA_INIT",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_DISPATCHER] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_DISPATCHER",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_HA_START] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_HA_START",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_HA_CONNECT] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_HA_CONNECT",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_INSTANCE_SET] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_INSTANCE_SET",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_EVENTS_STARTING] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_EVENTS_STARTING",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_RM_SETUP] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_RM_SETUP",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_SPIEL_INIT] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_SPIEL_INIT",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_SNS_CM_TRIGGER_FOPS] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_SNS_CM_TRIGGER_FOPS",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_EVENTS_STARTED] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_EVENTS_STARTED",
                .ml_enter = halon_interface_level_enter,
                .ml_leave = halon_interface_level_leave,
        },
        [M0_HALON_INTERFACE_LEVEL_STARTED] = {
                .ml_name  = "M0_HALON_INTERFACE_LEVEL_STARTED",
        },
};

int m0_halon_interface_start(struct m0_halon_interface *hi,
                             const char                *local_rpc_endpoint,
                             const struct m0_fid       *process_fid,
                             const struct m0_fid       *ha_service_fid,
                             const struct m0_fid       *rm_service_fid,
                             void                     (*entrypoint_request_cb)
                                (struct m0_halon_interface         *hi,
                                 const struct m0_uint128           *req_id,
                                 const char             *remote_rpc_endpoint,
                                 const struct m0_fid    *process_fid,
                                 const char             *git_rev_id,
                                 uint64_t                pid,
                                 bool                    first_request),
                             void                     (*msg_received_cb)
                                (struct m0_halon_interface *hi,
                                 struct m0_ha_link         *hl,
                                 const struct m0_ha_msg    *msg,
                                 uint64_t                   tag),
                             void                     (*msg_is_delivered_cb)
                                (struct m0_halon_interface *hi,
                                 struct m0_ha_link         *hl,
                                 uint64_t                   tag),
                             void                     (*msg_is_not_delivered_cb)
                                (struct m0_halon_interface *hi,
                                 struct m0_ha_link         *hl,
                                 uint64_t                   tag),
                             void                    (*link_connected_cb)
                                (struct m0_halon_interface *hi,
                                 const struct m0_uint128   *req_id,
                                 struct m0_ha_link         *link),
                             void                    (*link_reused_cb)
                                (struct m0_halon_interface *hi,
                                 const struct m0_uint128   *req_id,
                                 struct m0_ha_link         *link),
                             void                    (*link_absent_cb)
                                (struct m0_halon_interface *hi,
                                 const struct m0_uint128   *req_id),
                             void                    (*link_is_disconnecting_cb)
                                (struct m0_halon_interface *hi,
                                 struct m0_ha_link         *link),
                             void                     (*link_disconnected_cb)
                                (struct m0_halon_interface *hi,
                                 struct m0_ha_link         *link))
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        char                               *ep;
        int                                 rc;

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);
        M0_PRE(process_fid    != NULL);
        M0_PRE(rm_service_fid != NULL);

        M0_ENTRY("hi=%p local_rpc_endpoint=%s process_fid="FID_F,
                 hi, local_rpc_endpoint, FID_P(process_fid));
        M0_LOG(M0_DEBUG, "hi=%p ha_service_fid="FID_F" rm_service_fid="FID_F,
                 hi, FID_P(ha_service_fid), FID_P(rm_service_fid));

        ep = m0_strdup(local_rpc_endpoint);
        if (ep == NULL)
                return M0_ERR(-ENOMEM);

        hii->hii_cfg.hic_local_rpc_endpoint       =  ep;
        hii->hii_cfg.hic_process_fid              = *process_fid;
        hii->hii_cfg.hic_ha_service_fid           = *ha_service_fid;
        hii->hii_cfg.hic_rm_service_fid           = *rm_service_fid;
        hii->hii_cfg.hic_entrypoint_request_cb    =  entrypoint_request_cb;
        hii->hii_cfg.hic_msg_received_cb          =  msg_received_cb;
        hii->hii_cfg.hic_msg_is_delivered_cb      =  msg_is_delivered_cb;
        hii->hii_cfg.hic_msg_is_not_delivered_cb  =  msg_is_not_delivered_cb;
        hii->hii_cfg.hic_link_connected_cb        =  link_connected_cb;
        hii->hii_cfg.hic_link_reused_cb           =  link_reused_cb;
        hii->hii_cfg.hic_link_absent_cb           =  link_absent_cb;
        hii->hii_cfg.hic_link_is_disconnecting_cb =  link_is_disconnecting_cb;
        hii->hii_cfg.hic_link_disconnected_cb     =  link_disconnected_cb;

        m0_module_setup(&hii->hii_module, "m0_halon_interface",
                        halon_interface_levels,
                        ARRAY_SIZE(halon_interface_levels),
                        &hii->hii_instance);
        rc = m0_module_init(&hii->hii_module, M0_HALON_INTERFACE_LEVEL_STARTED);
        if (rc != 0) {
                m0_module_fini(&hii->hii_module, M0_MODLEV_NONE);
                m0_free(ep);
                return M0_ERR(rc);
        }
        m0_sm_group_lock(&hii->hii_sm_group);
        m0_sm_state_set(&hii->hii_sm, M0_HALON_INTERFACE_STATE_WORKING);
        m0_sm_group_unlock(&hii->hii_sm_group);
        return M0_RC(rc);
}

void m0_halon_interface_stop(struct m0_halon_interface *hi)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;

        M0_ASSERT(m0_halon_interface_internal_bob_check(hii));
        M0_ASSERT(m0_get() == &hii->hii_instance);

        M0_ENTRY("hi=%p", hi);

        m0_sm_group_lock(&hii->hii_sm_group);
        m0_sm_state_set(&hii->hii_sm, M0_HALON_INTERFACE_STATE_INITIALISED);
        m0_sm_group_unlock(&hii->hii_sm_group);

        m0_free(hii->hii_cfg.hic_local_rpc_endpoint);
        m0_module_fini(&hii->hii_module, M0_MODLEV_NONE);

        M0_LEAVE();
}

void m0_halon_interface_entrypoint_reply(
                struct m0_halon_interface  *hi,
                const struct m0_uint128    *req_id,
                int                         rc,
                uint32_t                    confd_nr,
                const struct m0_fid        *confd_fid_data,
                const char                **confd_eps_data,
                uint32_t                    confd_quorum,
                const struct m0_fid        *rm_fid,
                const char                 *rm_eps)
{
        struct m0_ha_entrypoint_rep rep;

        M0_ENTRY("hi=%p req_id="U128X_F" rc=%d confd_nr=%"PRIu32" "
                 "confd_quorum=%"PRIu32" rm_fid="FID_F" rm_eps=%s",
                 hi, U128_P(req_id), rc, confd_nr, confd_quorum,
                 FID_P(rm_fid == NULL ? &M0_FID0 : rm_fid), rm_eps);

        if (hi->hif_internal->hii_cfg.hic_log_entrypoint) {
                M0_LOG(M0_WARN, "req_id="U128X_F" rc=%d confd_nr=%"PRIu32" "
                 "confd_quorum=%"PRIu32" rm_fid="FID_F" rm_eps=%s",
                 U128_P(req_id), rc, confd_nr, confd_quorum,
                 FID_P(rm_fid == NULL ? &M0_FID0 : rm_fid), rm_eps);
        }
        rep = (struct m0_ha_entrypoint_rep){
                .hae_quorum        = confd_quorum,
                .hae_confd_fids    = {
                        .af_count = confd_nr,
                        .af_elems = (struct m0_fid *)confd_fid_data,
                },
                .hae_confd_eps     = confd_eps_data,
                .hae_active_rm_fid = rm_fid == NULL ? M0_FID0 : *rm_fid,
                .hae_active_rm_ep  = (char *)rm_eps,
                .hae_control       = rc == 0 ? M0_HA_ENTRYPOINT_CONSUME :
                                               M0_HA_ENTRYPOINT_QUERY,
        };
        m0_ha_entrypoint_reply(&hi->hif_internal->hii_ha, req_id, &rep, NULL);
        M0_LEAVE();
}

void m0_halon_interface_send(struct m0_halon_interface *hi,
                             struct m0_ha_link         *hl,
                             const struct m0_ha_msg    *msg,
                             uint64_t                  *tag)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%p",
                 hi, &hii->hii_ha, hl, ep, msg, msg->hm_epoch, tag);
        m0_ha_msg_debug_print(msg, __func__);
        m0_ha_send(&hii->hii_ha, hl, msg, tag);
        if (hii->hii_cfg.hic_log_msg) {
                M0_LOG(M0_WARN, "hl=%p ep=%s epoch=%" PRIu64 " tag=%" PRIu64 " "
                       "type=%"PRIu64,
                       hl, ep, msg->hm_epoch, *tag, msg->hm_data.hed_type);
        }
        M0_LEAVE("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%"PRIu64,
                 hi, &hii->hii_ha, hl, ep, msg, msg->hm_epoch, *tag);
}

void m0_halon_interface_delivered(struct m0_halon_interface *hi,
                                  struct m0_ha_link         *hl,
                                  const struct m0_ha_msg    *msg)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%"PRIu64,
                 hi, &hii->hii_ha, hl, ep,
                 msg, msg->hm_epoch, m0_ha_msg_tag(msg));
        if (hii->hii_cfg.hic_log_msg) {
                M0_LOG(M0_WARN, "hl=%p ep=%s epoch=%" PRIu64 " tag=%" PRIu64 " "
                       "type=%"PRIu64, hl, ep, msg->hm_epoch, m0_ha_msg_tag(msg),
                       msg->hm_data.hed_type);
        }
        /*
         * Remove 'const' here.
         *
         * 'msg' should be the message passed to hic_msg_received_cb().
         * halon_interface_msg_received_cb() has non-const msg, the function
         * makes the message const. If the 'const' is removed here then no
         * harm should be done.
         */
        m0_ha_delivered(&hii->hii_ha, hl, (struct m0_ha_msg *)msg);
        M0_LEAVE("hi=%p ha=%p hl=%p ep=%s msg=%p epoch=%" PRIu64 " tag=%"PRIu64,
                 hi, &hii->hii_ha, hl, ep,
                 msg, msg->hm_epoch, m0_ha_msg_tag(msg));
}

void m0_halon_interface_disconnect(struct m0_halon_interface *hi,
                                   struct m0_ha_link         *hl)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        char                                buf[HALON_INTERFACE_EP_BUF];
        const char                         *ep = &buf[0];

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        m0_ha_rpc_endpoint(&hii->hii_ha, hl, buf, ARRAY_SIZE(buf));
        M0_ENTRY("hi=%p ha=%p hl=%p ep=%s", hi, &hii->hii_ha, hl, ep);
        if (hii->hii_cfg.hic_log_link)
                M0_LOG(M0_WARN, "hl=%p ep=%s", hl, ep);
        m0_ha_disconnect_incoming(&hii->hii_ha, hl);
        M0_LEAVE("hi=%p ha=%p hl=%p ep=%s", hi, &hii->hii_ha, hl, ep);
}

static bool halon_interface_is_working(struct m0_halon_interface_internal *hii)
{
        bool working;

        m0_sm_group_lock(&hii->hii_sm_group);
        working = hii->hii_sm.sm_state == M0_HALON_INTERFACE_STATE_WORKING;
        m0_sm_group_unlock(&hii->hii_sm_group);
        return working;
}

struct m0_rpc_machine *
m0_halon_interface_rpc_machine(struct m0_halon_interface *hi)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        struct m0_rpc_machine              *rpc_machine;
        bool                                working;

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        rpc_machine = &hii->hii_rpc_machine;
        working = halon_interface_is_working(hii);
        M0_LOG(M0_DEBUG, "hi=%p hii=%p rpc_machine=%p working=%d",
               hi, hii, rpc_machine, !!working);
        return working ? rpc_machine : NULL;
}

struct m0_reqh *m0_halon_interface_reqh(struct m0_halon_interface *hi)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        struct m0_reqh                     *reqh;
        bool                                working;

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        reqh = &hii->hii_reqh;
        working = halon_interface_is_working(hii);
        M0_LOG(M0_DEBUG, "hi=%p hii=%p reqh=%p working=%d",
               hi, hii, reqh, !!working);
        return working ? reqh : NULL;
}

struct m0_spiel *m0_halon_interface_spiel(struct m0_halon_interface *hi)
{
        struct m0_halon_interface_internal *hii = hi->hif_internal;
        struct m0_spiel                    *spiel;
        bool                                working;

        M0_PRE(m0_halon_interface_internal_bob_check(hii));
        M0_PRE(m0_get() == &hii->hii_instance);

        spiel = &hii->hii_spiel;
        working = halon_interface_is_working(hii);
        M0_LOG(M0_DEBUG, "hi=%p hii=%p spiel=%p working=%d",
               hi, hii, spiel, !!working);
        return working ? spiel : NULL;
}

M0_INTERNAL int m0_halon_interface_thread_adopt(struct m0_halon_interface *hi,
                                                struct m0_thread *thread)
{
        struct m0 *motr;
        int        rc;

        M0_PRE(m0_halon_interface_internal_bob_check(hi->hif_internal));

        motr = &hi->hif_internal->hii_instance;
        rc = m0_thread_adopt(thread, motr);

        return M0_RC(rc);
}

M0_INTERNAL void m0_halon_interface_thread_shun(void)
{
        m0_thread_shun();
}

#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:
 */
/*
 * vim: tabstop=8 shiftwidth=8 noexpandtab textwidth=80 nowrap
 */