link.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/link.h"

#include "lib/memory.h"         /* M0_ALLOC_PTR */
#include "lib/errno.h"          /* ENOMEM */
#include "lib/tlist.h"          /* M0_TL_DESCR_DEFINE */
#include "lib/types.h"          /* m0_uint128 */
#include "lib/misc.h"           /* container_of */
#include "lib/time.h"           /* m0_time_from_now */

#include "sm/sm.h"              /* m0_sm_state_descr */
#include "rpc/rpc.h"            /* m0_rpc_reply_post */
#include "rpc/rpc_opcodes.h"    /* M0_HA_LINK_OUTGOING_OPCODE */

#include "fop/fom_generic.h"    /* M0_FOPH_FINISH */

#include "ha/link_fops.h"       /* m0_ha_link_msg_fopt */
#include "ha/link_service.h"    /* m0_ha_link_service_register */


enum {
        HA_LINK_SUPPRESS_START_NR = 3,
};

static struct m0_sm_state_descr ha_link_sm_states[] = {
        [M0_HA_LINK_STATE_INIT] = {
                .sd_flags   = M0_SDF_INITIAL,
                .sd_name    = "M0_HA_LINK_STATE_INIT",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_FINI,
                                      M0_HA_LINK_STATE_START),
        },
        [M0_HA_LINK_STATE_FINI] = {
                .sd_flags   = M0_SDF_TERMINAL,
                .sd_name    = "M0_HA_LINK_STATE_FINI",
                .sd_allowed = 0,
        },
        [M0_HA_LINK_STATE_START] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_START",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_STOP,
                                      M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_STOP] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_STOP",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_FINI),
        },
        [M0_HA_LINK_STATE_IDLE] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_IDLE",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_STOP,
                                      M0_HA_LINK_STATE_RECV,
                                      M0_HA_LINK_STATE_DELIVERY,
                                      M0_HA_LINK_STATE_RPC_FAILED,
                                      M0_HA_LINK_STATE_LINK_FAILED,
                                      M0_HA_LINK_STATE_LINK_REUSED,
                                      M0_HA_LINK_STATE_DISCONNECTING),
        },
        [M0_HA_LINK_STATE_RECV] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_RECV",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_DELIVERY] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_DELIVERY",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_RPC_FAILED] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_RPC_FAILED",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_LINK_FAILED] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_LINK_FAILED",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_LINK_REUSED] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_LINK_REUSED",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
        [M0_HA_LINK_STATE_DISCONNECTING] = {
                .sd_flags   = 0,
                .sd_name    = "M0_HA_LINK_STATE_DISCONNECTING",
                .sd_allowed = M0_BITS(M0_HA_LINK_STATE_IDLE),
        },
};

M0_BASSERT(ARRAY_SIZE(ha_link_sm_states) == M0_HA_LINK_STATE_NR);

static struct m0_sm_conf ha_link_sm_conf = {
        .scf_name      = "m0_ha_link::hln_sm",
        .scf_nr_states = ARRAY_SIZE(ha_link_sm_states),
        .scf_state     = ha_link_sm_states,
};

static struct m0_fom_type ha_link_outgoing_fom_type;
extern const struct m0_fom_ops ha_link_outgoing_fom_ops;

static void ha_link_outgoing_fom_wakeup(struct m0_ha_link *hl);

static bool ha_link_rpc_wait_cb(struct m0_clink *clink)
{
        struct m0_ha_link *hl;

        M0_ENTRY();
        hl = container_of(clink, struct m0_ha_link, hln_rpc_wait);
        m0_mutex_lock(&hl->hln_lock);
        hl->hln_rpc_event_occurred = true;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE();
        return true;
}

static bool ha_link_quiesce_wait_cb(struct m0_clink *clink)
{
        struct m0_ha_link *hl;

        M0_ENTRY();
        hl = container_of(clink, struct m0_ha_link, hln_quiesce_wait);
        m0_mutex_lock(&hl->hln_lock);
        hl->hln_quiesced = true;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE();
        return true;
}

M0_INTERNAL int m0_ha_link_init(struct m0_ha_link     *hl,
                                struct m0_ha_link_cfg *hl_cfg)
{
        int rc;

        M0_PRE(M0_IS0(hl));

        M0_ENTRY("hl=%p hlc_reqh=%p hlc_reqh_service=%p hlc_rpc_machine=%p",
                 hl, hl_cfg->hlc_reqh, hl_cfg->hlc_reqh_service,
                 hl_cfg->hlc_rpc_machine);
        hl->hln_cfg = *hl_cfg;
        m0_mutex_init(&hl->hln_lock);
        m0_ha_lq_init(&hl->hln_q_in, &hl->hln_cfg.hlq_q_cfg_in);
        m0_ha_lq_init(&hl->hln_q_out, &hl->hln_cfg.hlq_q_cfg_out);
        m0_fom_init(&hl->hln_fom, &ha_link_outgoing_fom_type,
                    &ha_link_outgoing_fom_ops, NULL, NULL,
                    hl->hln_cfg.hlc_reqh);
        rc = m0_semaphore_init(&hl->hln_stop_cond, 0);
        M0_ASSERT(rc == 0);
        m0_mutex_init(&hl->hln_stop_chan_lock);
        m0_chan_init(&hl->hln_stop_chan, &hl->hln_stop_chan_lock);
        m0_sm_group_init(&hl->hln_sm_group);
        m0_sm_init(&hl->hln_sm, &ha_link_sm_conf, M0_HA_LINK_STATE_INIT,
                   &hl->hln_sm_group);
        m0_clink_init(&hl->hln_rpc_wait, &ha_link_rpc_wait_cb);
        m0_clink_init(&hl->hln_quiesce_wait, &ha_link_quiesce_wait_cb);
        m0_mutex_init(&hl->hln_quiesce_chan_lock);
        m0_chan_init(&hl->hln_quiesce_chan, &hl->hln_quiesce_chan_lock);
        m0_clink_add_lock(&hl->hln_quiesce_chan, &hl->hln_quiesce_wait);
        m0_sm_timer_init(&hl->hln_reconnect_wait_timer);
        hl->hln_rpc_wait.cl_is_oneshot = true;
        hl->hln_reconnect = false;
        hl->hln_reconnect_wait = false;
        hl->hln_reconnect_cfg_is_set = false;
        hl->hln_waking_up = false;
        hl->hln_fom_is_stopping = false;
        hl->hln_fom_enable_wakeup = true;
        hl->hln_no_new_delivered = false;
        hl->hln_req_fop_seq = 0;
        hl->hln_backoff_rc = 0;
        hl->hln_backoff_nr = 0;
        return M0_RC(0);
}

M0_INTERNAL void m0_ha_link_fini(struct m0_ha_link *hl)
{
        M0_ENTRY("hl=%p", hl);
        if (hl->hln_backoff_rc != 0) {
                M0_LOG(M0_WARN, "hln_backoff_rc=%d hln_backoff_nr=%"PRIu64,
                       hl->hln_backoff_rc, hl->hln_backoff_nr);
        }
        m0_sm_timer_fini(&hl->hln_reconnect_wait_timer);
        m0_clink_del_lock(&hl->hln_quiesce_wait);
        m0_chan_fini_lock(&hl->hln_quiesce_chan);
        m0_mutex_fini(&hl->hln_quiesce_chan_lock);
        m0_clink_fini(&hl->hln_quiesce_wait);
        m0_clink_fini(&hl->hln_rpc_wait);
        m0_sm_group_lock(&hl->hln_sm_group);
        m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_FINI);
        m0_sm_fini(&hl->hln_sm);
        m0_sm_group_unlock(&hl->hln_sm_group);
        m0_sm_group_fini(&hl->hln_sm_group);
        m0_mutex_lock(&hl->hln_stop_chan_lock);
        m0_chan_fini(&hl->hln_stop_chan);
        m0_mutex_unlock(&hl->hln_stop_chan_lock);
        m0_mutex_fini(&hl->hln_stop_chan_lock);
        m0_semaphore_fini(&hl->hln_stop_cond);
        m0_ha_lq_fini(&hl->hln_q_out);
        m0_ha_lq_fini(&hl->hln_q_in);
        m0_mutex_fini(&hl->hln_lock);
        M0_LEAVE();
}

static int ha_link_conn_cfg_copy(struct m0_ha_link_conn_cfg       *dst,
                                 const struct m0_ha_link_conn_cfg *src)
{
        char *ep = m0_strdup(src->hlcc_rpc_endpoint);

        if (ep == NULL)
                return M0_ERR_INFO(-ENOMEM, "%s", src->hlcc_rpc_endpoint);
        *dst = *src;
        dst->hlcc_rpc_endpoint = ep;
        return M0_RC(0);
}

static void ha_link_conn_cfg_free(struct m0_ha_link_conn_cfg *hl_conn_cfg)
{
        /* This value is allocated in ha_link_conn_cfg_copy() */
        m0_free((char *)hl_conn_cfg->hlcc_rpc_endpoint);
}

static void ha_link_tags_apply(struct m0_ha_link              *hl,
                               const struct m0_ha_link_params *lp)
{
        const struct m0_ha_link_tags *tags_in_new  = &lp->hlp_tags_remote;
        const struct m0_ha_link_tags *tags_out_new = &lp->hlp_tags_local;
        struct m0_ha_link_tags        tags_in;
        struct m0_ha_link_tags        tags_out;
        bool                          done;

        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));

        m0_ha_lq_tags_get(&hl->hln_q_out, &tags_out);
        m0_ha_lq_tags_get(&hl->hln_q_in,  &tags_in);
        M0_LOG(M0_DEBUG, "hl=%p     out="HLTAGS_F, hl, HLTAGS_P(&tags_out));
        M0_LOG(M0_DEBUG, "hl=%p new out="HLTAGS_F, hl, HLTAGS_P(tags_out_new));
        M0_LOG(M0_DEBUG, "hl=%p      in="HLTAGS_F, hl, HLTAGS_P(&tags_in));
        M0_LOG(M0_DEBUG, "hl=%p  new in="HLTAGS_F, hl, HLTAGS_P(tags_in_new));

        /* Move 'next' tag to 'delivered'. Leave all other tags intact */
        while (m0_ha_lq_tag_next(&hl->hln_q_out) >
               m0_ha_lq_tag_delivered(&hl->hln_q_out)) {
                done = m0_ha_lq_try_unnext(&hl->hln_q_out);
                M0_ASSERT(done);
        }
        M0_ASSERT_INFO(m0_ha_lq_tag_next(&hl->hln_q_out) ==
                       m0_ha_lq_tag_delivered(&hl->hln_q_out),
                       "m0_ha_lq_tag_next(&hl->hln_q_out)=%" PRIu64 " "
                       "m0_ha_lq_tag_delivered(&hl->hln_q_out)=%"PRIu64,
                       m0_ha_lq_tag_next(&hl->hln_q_out),
                       m0_ha_lq_tag_delivered(&hl->hln_q_out));

        m0_ha_lq_tags_get(&hl->hln_q_out, &tags_out);
        M0_ASSERT_INFO(m0_ha_lq_tag_delivered(&hl->hln_q_out) >=
                       tags_out_new->hlt_confirmed,
                       "m0_ha_lq_tag_delivered(&hl->hln_q_out)=%" PRIu64 " "
                       "tags_out_new->hlt_confirmed=%"PRIu64,
                       m0_ha_lq_tag_delivered(&hl->hln_q_out),
                       tags_out_new->hlt_confirmed);
        M0_ASSERT_INFO(m0_ha_lq_tag_delivered(&hl->hln_q_out) ==
                       tags_out_new->hlt_delivered,
                       "m0_ha_lq_tag_delivered(&hl->hln_q_out)=%" PRIu64 " "
                       "tags_out_new->hlt_delivered=%"PRIu64,
                       m0_ha_lq_tag_delivered(&hl->hln_q_out),
                       tags_out_new->hlt_delivered);
        M0_ASSERT_INFO(m0_ha_lq_tag_next(&hl->hln_q_out) ==
                       tags_out_new->hlt_next,
                       "m0_ha_lq_tag_next(&hl->hln_q_out)=%" PRIu64 " "
                       "tags_out_new->hlt_next=%"PRIu64,
                       m0_ha_lq_tag_next(&hl->hln_q_out),
                       tags_out_new->hlt_next);
        M0_ASSERT_INFO(m0_ha_lq_tag_assign(&hl->hln_q_out) >=
                       tags_out_new->hlt_assign,
                       "m0_ha_lq_tag_assign(&hl->hln_q_out)=%" PRIu64 " "
                       "tags_out_new->hlt_assign=%"PRIu64,
                       m0_ha_lq_tag_assign(&hl->hln_q_out),
                       tags_out_new->hlt_assign);
        M0_ASSERT_INFO(m0_ha_link_tags_eq(&tags_in, tags_in_new),
                       "tags_in="HLTAGS_F" tags_in_new="HLTAGS_F,
                       HLTAGS_P(&tags_in), HLTAGS_P(tags_in_new));

        m0_ha_lq_tags_get(&hl->hln_q_out, &tags_out);
        m0_ha_lq_tags_get(&hl->hln_q_in,  &tags_in);
        M0_LEAVE("hl=%p out="HLTAGS_F" in="HLTAGS_F,
                 hl, HLTAGS_P(&tags_out), HLTAGS_P(&tags_in));
}

M0_INTERNAL void m0_ha_link_start(struct m0_ha_link          *hl,
                                  struct m0_ha_link_conn_cfg *hl_conn_cfg)
{
        struct m0_ha_link_params *lp;
        int rc;

        M0_ENTRY("hl=%p hlp_id_local="U128X_F" hlp_id_remote="U128X_F" "
                 "hlp_id_connection="U128X_F, hl,
                 U128_P(&hl_conn_cfg->hlcc_params.hlp_id_local),
                 U128_P(&hl_conn_cfg->hlcc_params.hlp_id_remote),
                 U128_P(&hl_conn_cfg->hlcc_params.hlp_id_connection));
        M0_LOG(M0_DEBUG, "hlcc_rpc_service_fid="FID_F" "
               "hlcc_rpc_endpoint=%s hlcc_max_rpcs_in_flight=%"PRIu64,
               FID_P(&hl_conn_cfg->hlcc_rpc_service_fid),
               (const char *)hl_conn_cfg->hlcc_rpc_endpoint,
               hl_conn_cfg->hlcc_max_rpcs_in_flight);
        rc = ha_link_conn_cfg_copy(&hl->hln_conn_cfg, hl_conn_cfg);
        M0_ASSERT(rc == 0);     /* XXX */
        m0_mutex_lock(&hl->hln_lock);
        lp = &hl->hln_conn_cfg.hlcc_params;
        m0_ha_lq_tags_set(&hl->hln_q_out, &lp->hlp_tags_local);
        m0_ha_lq_tags_set(&hl->hln_q_in,  &lp->hlp_tags_remote);
        hl->hln_cb_disconnecting = false;
        hl->hln_cb_reused        = false;
        m0_mutex_unlock(&hl->hln_lock);
        m0_fom_queue(&hl->hln_fom);
        hl->hln_fom_locality = &hl->hln_fom.fo_loc->fl_locality;
        M0_LEAVE();
}

M0_INTERNAL void m0_ha_link_stop(struct m0_ha_link *hl, struct m0_clink *clink)
{
        M0_ENTRY("hl=%p", hl);
        M0_PRE(clink->cl_is_oneshot);
        m0_clink_add_lock(&hl->hln_stop_chan, clink);
        m0_semaphore_up(&hl->hln_stop_cond);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE();
}

M0_INTERNAL void m0_ha_link_reconnect_begin(struct m0_ha_link        *hl,
                                            struct m0_ha_link_params *lp)
{
        m0_mutex_lock(&hl->hln_lock);
        *lp = hl->hln_conn_cfg.hlcc_params;
        m0_ha_lq_tags_get(&hl->hln_q_out, &lp->hlp_tags_local);
        m0_ha_lq_tags_get(&hl->hln_q_in,  &lp->hlp_tags_remote);
        m0_mutex_unlock(&hl->hln_lock);
        /* TODO refactor it somehow */
        M0_LOG(M0_DEBUG, "hl=%p id_local="U128X_F" id_remote="U128X_F" "
               "id_connection="U128X_F, hl, U128_P(&lp->hlp_id_local),
               U128_P(&lp->hlp_id_remote), U128_P(&lp->hlp_id_connection));
        M0_LOG(M0_DEBUG, "hl=%p  hlp_tags_local="HLTAGS_F,
               hl, HLTAGS_P(&lp->hlp_tags_local));
        M0_LOG(M0_DEBUG, "hl=%p hlp_tags_remote="HLTAGS_F,
               hl, HLTAGS_P(&lp->hlp_tags_remote));
}

M0_INTERNAL void
m0_ha_link_reconnect_end(struct m0_ha_link                *hl,
                         const struct m0_ha_link_conn_cfg *hl_conn_cfg)
{
        const struct m0_ha_link_params *lp = &hl_conn_cfg->hlcc_params;
        int                             rc;

        /* TODO refactor the M0_LOG of hl_conn_cfg */
        M0_ENTRY("hl=%p hlcc_rpc_service_fid="FID_F" hlcc_rpc_endpoint=%s",
                 hl, FID_P(&hl_conn_cfg->hlcc_rpc_service_fid),
                 hl_conn_cfg->hlcc_rpc_endpoint);
        M0_ENTRY("hl=%p hlcc_max_rpcs_in_flight=%" PRIu64 " "
                 "hlcc_connect_timeout=%" PRIu64 " "
                 "hlcc_disconnect_timeout=%" PRIu64 " "
                 "hlcc_resend_interval=%" PRIu64 " "
                 "hlcc_nr_sent_max=%"PRIu64,
                 hl, hl_conn_cfg->hlcc_max_rpcs_in_flight,
                 hl_conn_cfg->hlcc_connect_timeout,
                 hl_conn_cfg->hlcc_disconnect_timeout,
                 hl_conn_cfg->hlcc_resend_interval,
                 hl_conn_cfg->hlcc_nr_sent_max);
        M0_LOG(M0_DEBUG, "hl=%p id_local="U128X_F" id_remote="U128X_F" "
               "id_connection="U128X_F, hl, U128_P(&lp->hlp_id_local),
               U128_P(&lp->hlp_id_remote), U128_P(&lp->hlp_id_connection));
        M0_LOG(M0_DEBUG, "hl=%p  hlp_tags_local="HLTAGS_F,
               hl, HLTAGS_P(&lp->hlp_tags_local));
        M0_LOG(M0_DEBUG, "hl=%p hlp_tags_remote="HLTAGS_F,
               hl, HLTAGS_P(&lp->hlp_tags_remote));
        m0_mutex_lock(&hl->hln_lock);
        if (hl->hln_reconnect_cfg_is_set)
                ha_link_conn_cfg_free(&hl->hln_conn_reconnect_cfg);
        rc = ha_link_conn_cfg_copy(&hl->hln_conn_reconnect_cfg, hl_conn_cfg);
        M0_ASSERT(rc == 0);     /* XXX */
        hl->hln_reconnect_cfg_is_set = true;
        hl->hln_reconnect = true;
        m0_mutex_unlock(&hl->hln_lock);
        M0_LEAVE("hl=%p", hl);
}

M0_INTERNAL void m0_ha_link_reconnect_cancel(struct m0_ha_link *hl)
{
        M0_ENTRY("hl=%p", hl);
        M0_LEAVE();
}

M0_INTERNAL void
m0_ha_link_reconnect_params(const struct m0_ha_link_params *lp_alive,
                            struct m0_ha_link_params       *lp_alive_new,
                            struct m0_ha_link_params       *lp_dead_new,
                            const struct m0_uint128        *id_alive,
                            const struct m0_uint128        *id_dead,
                            const struct m0_uint128        *id_connection)
{
        const struct m0_ha_link_tags *tags_local  = &lp_alive->hlp_tags_local;
        const struct m0_ha_link_tags *tags_remote = &lp_alive->hlp_tags_remote;

        *lp_alive_new = (struct m0_ha_link_params){
                .hlp_id_local      = *id_alive,
                .hlp_id_remote     = *id_dead,
                .hlp_id_connection = *id_connection,
                .hlp_tags_local    = {
                        .hlt_confirmed   = tags_local->hlt_confirmed,
                        .hlt_delivered = tags_local->hlt_delivered,
                        .hlt_next        = tags_local->hlt_delivered,
                        .hlt_assign      = tags_local->hlt_assign,
                },
                .hlp_tags_remote   = {
                        .hlt_confirmed   = tags_remote->hlt_confirmed,
                        .hlt_delivered = tags_remote->hlt_delivered,
                        .hlt_next        = tags_remote->hlt_next,
                        .hlt_assign      = tags_remote->hlt_assign,
                },
        };
        *lp_dead_new = (struct m0_ha_link_params){
                .hlp_id_local      = *id_dead,
                .hlp_id_remote     = *id_alive,
                .hlp_id_connection = *id_connection,
                .hlp_tags_local    = {
                        .hlt_confirmed   = tags_remote->hlt_delivered,
                        .hlt_delivered = tags_remote->hlt_delivered,
                        .hlt_next        = tags_remote->hlt_next,
                        .hlt_assign      = tags_remote->hlt_next,
                },
                .hlp_tags_remote   = {
                        .hlt_confirmed   = tags_local->hlt_delivered,
                        .hlt_delivered = tags_local->hlt_delivered,
                        .hlt_next        = tags_local->hlt_delivered,
                        .hlt_assign      = tags_local->hlt_delivered,
                },
        };
}

M0_INTERNAL struct m0_chan *m0_ha_link_chan(struct m0_ha_link *hl)
{
        return &hl->hln_sm.sm_chan;
}

M0_INTERNAL enum m0_ha_link_state m0_ha_link_state_get(struct m0_ha_link *hl)
{
        M0_PRE(m0_sm_group_is_locked(&hl->hln_sm_group));
        return hl->hln_sm.sm_state;
}

M0_INTERNAL const char *m0_ha_link_state_name(enum m0_ha_link_state state)
{
        M0_PRE(_0C(state >= M0_HA_LINK_STATE_INIT) &&
               _0C(state < M0_HA_LINK_STATE_NR));
        return ha_link_sm_states[state].sd_name;
}

enum ha_link_send_type {
        HA_LINK_SEND_QUERY,
        HA_LINK_SEND_POST,
        HA_LINK_SEND_REPLY,
};

M0_INTERNAL void m0_ha_link_send(struct m0_ha_link      *hl,
                                 const struct m0_ha_msg *msg,
                                 uint64_t               *tag)
{
        M0_ENTRY("hl=%p msg=%p", hl, msg);
        m0_mutex_lock(&hl->hln_lock);
        *tag = m0_ha_lq_enqueue(&hl->hln_q_out, msg);
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE("hl=%p msg=%p tag=%"PRIu64, hl, msg, *tag);
}

M0_INTERNAL struct m0_ha_msg *m0_ha_link_recv(struct m0_ha_link *hl,
                                              uint64_t          *tag)
{
        struct m0_ha_msg *msg;

        m0_mutex_lock(&hl->hln_lock);
        msg = m0_ha_lq_next(&hl->hln_q_in);
        if (msg != NULL)
                *tag = m0_ha_msg_tag(msg);
        m0_mutex_unlock(&hl->hln_lock);

        M0_LOG(M0_DEBUG, "hl=%p msg=%p tag=%"PRIu64,
               hl, msg, msg == NULL ? M0_HA_MSG_TAG_UNKNOWN : *tag);
        return msg;
}

M0_INTERNAL void m0_ha_link_delivered(struct m0_ha_link *hl,
                                      struct m0_ha_msg  *msg)
{
        M0_ENTRY("hl=%p msg=%p tag=%"PRIu64, hl, msg, m0_ha_msg_tag(msg));
        m0_mutex_lock(&hl->hln_lock);
        m0_ha_lq_mark_delivered(&hl->hln_q_in, m0_ha_msg_tag(msg));
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE("hl=%p msg=%p tag=%"PRIu64, hl, msg, m0_ha_msg_tag(msg));
}

M0_INTERNAL bool m0_ha_link_msg_is_delivered(struct m0_ha_link *hl,
                                             uint64_t           tag)
{
        bool delivered;

        m0_mutex_lock(&hl->hln_lock);
        delivered = m0_ha_lq_is_delivered(&hl->hln_q_out, tag);
        m0_mutex_unlock(&hl->hln_lock);
        return delivered;
}

M0_INTERNAL uint64_t m0_ha_link_delivered_consume(struct m0_ha_link *hl)
{
        uint64_t tag;

        m0_mutex_lock(&hl->hln_lock);
        tag = hl->hln_no_new_delivered ? M0_HA_MSG_TAG_INVALID :
              m0_ha_lq_dequeue(&hl->hln_q_out);
        m0_mutex_unlock(&hl->hln_lock);
        M0_LOG(M0_DEBUG, "hl=%p tag=%"PRIu64, hl, tag);
        return tag;
}

M0_INTERNAL uint64_t m0_ha_link_not_delivered_consume(struct m0_ha_link *hl)
{
        uint64_t tag;

        m0_mutex_lock(&hl->hln_lock);
        tag = !hl->hln_no_new_delivered ? M0_HA_MSG_TAG_INVALID :
              m0_ha_lq_dequeue(&hl->hln_q_out);
        m0_mutex_unlock(&hl->hln_lock);
        M0_LOG(M0_DEBUG, "hl=%p tag=%"PRIu64, hl, tag);
        return tag;
}

struct ha_link_wait_ctx {
        struct m0_ha_link  *hwc_hl;
        uint64_t            hwc_tag;
        struct m0_clink     hwc_clink;
        struct m0_semaphore hwc_sem;
        bool                hwc_check_disable;
};

static void ha_link_wait(struct ha_link_wait_ctx *wait_ctx,
                                          bool (*check)(struct m0_clink *clink))
{
        int rc;

        m0_clink_init(&wait_ctx->hwc_clink, check);
        rc = m0_semaphore_init(&wait_ctx->hwc_sem, 0);
        M0_ASSERT(rc == 0);     /* XXX */
        m0_clink_add_lock(m0_ha_link_chan(wait_ctx->hwc_hl),
                          &wait_ctx->hwc_clink);
        check(&wait_ctx->hwc_clink);
        m0_semaphore_down(&wait_ctx->hwc_sem);
        m0_clink_del_lock(&wait_ctx->hwc_clink);
        m0_semaphore_fini(&wait_ctx->hwc_sem);
        m0_clink_fini(&wait_ctx->hwc_clink);
}

static bool ha_link_wait_delivery_check(struct m0_clink *clink)
{
        struct ha_link_wait_ctx *wait_ctx;

        /* XXX bob_of */
        wait_ctx = container_of(clink, struct ha_link_wait_ctx, hwc_clink);
        if (!wait_ctx->hwc_check_disable &&
            m0_ha_link_msg_is_delivered(wait_ctx->hwc_hl, wait_ctx->hwc_tag)) {
                wait_ctx->hwc_check_disable = true;
                m0_semaphore_up(&wait_ctx->hwc_sem);
        }
        return false;
}

M0_INTERNAL void m0_ha_link_wait_delivery(struct m0_ha_link *hl, uint64_t tag)
{
        bool                    delivered;
        struct ha_link_wait_ctx wait_ctx = {
                .hwc_hl            = hl,
                .hwc_tag           = tag,
                .hwc_check_disable = false,
        };

        M0_ENTRY("hl=%p tag=%"PRIu64, hl, tag);
        ha_link_wait(&wait_ctx, &ha_link_wait_delivery_check);
        delivered = m0_ha_link_msg_is_delivered(hl, tag);
        M0_ASSERT(delivered);
        M0_LEAVE("hl=%p tag=%"PRIu64, hl, tag);
}

static bool ha_link_wait_arrival_check(struct m0_clink *clink)
{
        struct ha_link_wait_ctx *wait_ctx;
        bool                     arrived;
        struct m0_ha_link       *hl;

        /* XXX bob_of */
        wait_ctx = container_of(clink, struct ha_link_wait_ctx, hwc_clink);
        hl = wait_ctx->hwc_hl;
        M0_ENTRY("hl=%p", hl);
        if (!wait_ctx->hwc_check_disable) {
                m0_mutex_lock(&hl->hln_lock);
                arrived = m0_ha_lq_has_next(&hl->hln_q_in);
                m0_mutex_unlock(&hl->hln_lock);
                if (arrived) {
                        wait_ctx->hwc_check_disable = true;
                        m0_semaphore_up(&wait_ctx->hwc_sem);
                }
                M0_LOG(M0_DEBUG, "hl=%p arrived=%d", hl, !!arrived);
        }
        M0_LEAVE("hl=%p", hl);
        return false;
}

M0_INTERNAL void m0_ha_link_wait_arrival(struct m0_ha_link *hl)
{
        bool                    arrived;
        struct ha_link_wait_ctx wait_ctx = {
                .hwc_hl            = hl,
                .hwc_check_disable = false,
        };

        M0_ENTRY("hl=%p", hl);
        ha_link_wait(&wait_ctx, &ha_link_wait_arrival_check);
        m0_mutex_lock(&hl->hln_lock);
        arrived = m0_ha_lq_has_next(&hl->hln_q_in);
        m0_mutex_unlock(&hl->hln_lock);
        M0_ASSERT(arrived);
        M0_LEAVE("hl=%p", hl);
}


static bool ha_link_wait_confirmation_check(struct m0_clink *clink)
{
        struct ha_link_wait_ctx *wait_ctx;
        bool                     delivered_notified;
        struct m0_ha_link       *hl;

        /* XXX bob_of */
        wait_ctx = container_of(clink, struct ha_link_wait_ctx, hwc_clink);
        hl = wait_ctx->hwc_hl;
        M0_ENTRY("hl=%p", hl);
        if (!wait_ctx->hwc_check_disable) {
                m0_mutex_lock(&hl->hln_lock);
                delivered_notified = wait_ctx->hwc_tag <
                                     m0_ha_lq_tag_confirmed(&hl->hln_q_in);
                m0_mutex_unlock(&hl->hln_lock);
                if (delivered_notified) {
                        wait_ctx->hwc_check_disable = true;
                        m0_semaphore_up(&wait_ctx->hwc_sem);
                }
                M0_LOG(M0_DEBUG, "hl=%p delivered_notified=%d",
                       hl, !!delivered_notified);
        }
        M0_LEAVE("hl=%p", hl);
        return false;
}

M0_INTERNAL void m0_ha_link_wait_confirmation(struct m0_ha_link *hl,
                                              uint64_t           tag)
{
        struct ha_link_wait_ctx wait_ctx = {
                .hwc_hl            = hl,
                .hwc_tag           = tag,
                .hwc_check_disable = false,
        };

        M0_ENTRY("hl=%p tag=%"PRIu64, hl, tag);
        ha_link_wait(&wait_ctx, &ha_link_wait_confirmation_check);
        M0_LEAVE("hl=%p", hl);
}

M0_INTERNAL void m0_ha_link_flush(struct m0_ha_link *hl)
{
        uint64_t tag_out_assign;
        uint64_t tag_in_assign;

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

        m0_mutex_lock(&hl->hln_lock);
        tag_out_assign = m0_ha_lq_tag_assign(&hl->hln_q_out);
        tag_in_assign  = m0_ha_lq_tag_assign(&hl->hln_q_in);
        m0_mutex_unlock(&hl->hln_lock);

        if (!M0_IN(tag_out_assign, (1, 2)))
                m0_ha_link_wait_delivery(hl, tag_out_assign - 2);
        if (!M0_IN(tag_in_assign, (1, 2)))
                m0_ha_link_wait_confirmation(hl, tag_in_assign - 2);
        M0_LEAVE("hl=%p tag_out_assign=%" PRIu64 " tag_in_assign=%"PRIu64,
                 hl, tag_out_assign, tag_in_assign);
}

M0_INTERNAL void m0_ha_link_cb_disconnecting(struct m0_ha_link *hl)
{
        M0_ENTRY("hl=%p", hl);
        m0_mutex_lock(&hl->hln_lock);
        M0_ASSERT(!hl->hln_cb_disconnecting);
        hl->hln_cb_disconnecting = true;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE("hl=%p", hl);
}

M0_INTERNAL void m0_ha_link_cb_reused(struct m0_ha_link *hl)
{
        M0_ENTRY("hl=%p", hl);
        m0_mutex_lock(&hl->hln_lock);
        M0_ASSERT(!hl->hln_cb_reused);
        hl->hln_cb_reused = true;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE("hl=%p", hl);
}

static void ha_link_tags_update(struct m0_ha_link *hl,
                                uint64_t           out_next,
                                uint64_t           in_delivered)
{
        struct m0_ha_link_tags tags_out;
        struct m0_ha_link_tags tags_in;
        uint64_t               delivered;

        M0_ENTRY("hl=%p out_next=%" PRIu64 " in_delivered=%"PRIu64,
                 hl, out_next, in_delivered);
        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));
        m0_ha_lq_tags_get(&hl->hln_q_out, &tags_out);
        m0_ha_lq_tags_get(&hl->hln_q_in,  &tags_in);
        M0_LOG(M0_DEBUG, "hl=%p out="HLTAGS_F, hl, HLTAGS_P(&tags_out));
        M0_LOG(M0_DEBUG, "hl=%p  in="HLTAGS_F, hl, HLTAGS_P(&tags_in));

        while (m0_ha_lq_tag_next(&hl->hln_q_out) < in_delivered)
                (void)m0_ha_lq_next(&hl->hln_q_out);

        delivered = m0_ha_lq_tag_delivered(&hl->hln_q_out);
        while (delivered < in_delivered) {
                m0_ha_lq_mark_delivered(&hl->hln_q_out, delivered);
                delivered += 2;
        }

        m0_ha_lq_tags_get(&hl->hln_q_out, &tags_out);
        m0_ha_lq_tags_get(&hl->hln_q_in,  &tags_in);
        M0_LOG(M0_DEBUG, "hl=%p out="HLTAGS_F, hl, HLTAGS_P(&tags_out));
        M0_LOG(M0_DEBUG, "hl=%p  in="HLTAGS_F, hl, HLTAGS_P(&tags_in));
}

static void ha_link_tags_in_out(struct m0_ha_link *hl,
                                uint64_t          *out_next,
                                uint64_t          *in_delivered)
{
        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));

        *out_next     = m0_ha_lq_tag_next(&hl->hln_q_out);
        *in_delivered = m0_ha_lq_tag_delivered(&hl->hln_q_in);
        M0_LOG(M0_DEBUG, "out_next=%" PRIu64 " in_delivered=%"PRIu64,
               *out_next, *in_delivered);
}

static void ha_link_msg_received(struct m0_ha_link      *hl,
                                 const struct m0_ha_msg *msg)
{
        M0_ENTRY("hl=%p tag=%" PRIu64 " type=%d",
                 hl, m0_ha_msg_tag(msg), m0_ha_msg_type_get(msg));
        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));
        if (!M0_IN(m0_ha_msg_tag(msg), (M0_HA_MSG_TAG_UNKNOWN,
                                        m0_ha_lq_tag_assign(&hl->hln_q_in)))) {
                M0_LOG(M0_WARN, "dropping out-of-order message: hl=%p "
                       "tag=%" PRIu64 " hed_type=%d",
                       hl, m0_ha_msg_tag(msg), m0_ha_msg_type_get(msg));
        } else {
                m0_ha_lq_enqueue(&hl->hln_q_in, msg);
        }
}

static void ha_link_msg_recv_or_delivery_broadcast(struct m0_ha_link *hl)
{
        uint64_t in_next;
        uint64_t out_confirmed;
        uint64_t tag_recv;
        uint64_t tag_delivery;

        m0_mutex_lock(&hl->hln_lock);
        in_next = m0_ha_lq_tag_next(&hl->hln_q_in);
        tag_recv = in_next > 2 ? in_next :
                   in_next == m0_ha_lq_tag_assign(&hl->hln_q_in) ? 0 : in_next;
        out_confirmed = m0_ha_lq_tag_confirmed(&hl->hln_q_out);
        tag_delivery = out_confirmed > 2 ? out_confirmed :
                       out_confirmed == m0_ha_lq_tag_delivered(&hl->hln_q_out) ?
                       0 : out_confirmed;
        m0_mutex_unlock(&hl->hln_lock);

        m0_sm_group_lock(&hl->hln_sm_group);
        M0_LOG(M0_DEBUG, "hln_tag_broadcast_recv=%" PRIu64 " tag_recv=%"PRIu64,
               hl->hln_tag_broadcast_recv, tag_recv);
        M0_LOG(M0_DEBUG, "hln_tag_broadcast_delivery=%" PRIu64 " "
               "tag_delivery=%"PRIu64,
               hl->hln_tag_broadcast_delivery, tag_delivery);
        if (hl->hln_tag_broadcast_recv <= tag_recv) {
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_RECV);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                hl->hln_tag_broadcast_recv = tag_recv;
        }
        if (hl->hln_tag_broadcast_delivery <= tag_delivery) {
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_DELIVERY);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                hl->hln_tag_broadcast_delivery = tag_delivery;
        }
        m0_sm_group_unlock(&hl->hln_sm_group);
}

struct ha_link_incoming_fom {
        struct m0_ha_link *hli_hl;
        struct m0_fom      hli_fom;
};

static struct ha_link_incoming_fom *
ha_link_incoming_fom_container(struct m0_fom *fom)
{
        /* TODO bob_of */
        return container_of(fom, struct ha_link_incoming_fom, hli_fom);
}

static int ha_link_incoming_fom_tick(struct m0_fom *fom)
{
        struct m0_ha_link_msg_fop     *req_fop;
        struct m0_ha_link_msg_rep_fop *rep_fop;
        struct ha_link_incoming_fom   *hli;
        struct m0_ha_link             *hl;
        struct m0_uint128              id_connection;
        struct m0_ha_msg              *msg;
        const char                    *ep;

        req_fop = m0_fop_data(fom->fo_fop);
        rep_fop = m0_fop_data(fom->fo_rep_fop);
        hli = ha_link_incoming_fom_container(fom);
        ep = m0_rpc_conn_addr(
                         m0_fop_to_rpc_item(fom->fo_fop)->ri_session->s_conn);
        M0_ENTRY("fom=%p req_fop=%p rep_fop=%p ep=%s",
                 fom, req_fop, rep_fop, ep);
        M0_LOG(M0_DEBUG, "ep=%p lmf_msg_nr=%" PRIu64 " lmf_id_remote="U128X_F" "
               "lmf_id_local="U128X_F" lmf_id_connection="U128X_F,
               ep, req_fop->lmf_msg_nr, U128_P(&req_fop->lmf_id_remote),
               U128_P(&req_fop->lmf_id_local),
               U128_P(&req_fop->lmf_id_connection));

        hl = m0_ha_link_service_find_get(fom->fo_service,
                                         &req_fop->lmf_id_remote,
                                         &id_connection);
        hli->hli_hl = hl;
        M0_LOG(M0_DEBUG, "fom=%p hl=%p", fom, hl);
        if (req_fop->lmf_msg_nr != 0) {
                msg = &req_fop->lmf_msg;
                M0_LOG(M0_DEBUG, "ep=%s lmf_id_remote="U128X_F" "
                       "hm_fid="FID_F" hed_type=%d tag=%"PRIu64,
                       ep, U128_P(&req_fop->lmf_id_remote), FID_P(&msg->hm_fid),
                       m0_ha_msg_type_get(msg), m0_ha_msg_tag(msg));
        }
        if (hl == NULL) {
                /*
                 * The first M0_LOG() parameter must be a compile-time
                 * constant.
                 */
                if (req_fop->lmf_seq <= HA_LINK_SUPPRESS_START_NR) {
                        M0_LOG(M0_DEBUG, "no such link: ep=%s "
                               "lmf_id_remote="U128X_F" lmf_seq=%"PRIu64,
                               ep, U128_P(&req_fop->lmf_id_remote),
                               req_fop->lmf_seq);
                } else {
                        M0_LOG(M0_WARN, "no such link: ep=%s "
                               "lmf_id_remote="U128X_F" lmf_seq=%"PRIu64,
                               ep, U128_P(&req_fop->lmf_id_remote),
                               req_fop->lmf_seq);
                }
                rep_fop->lmr_rc = -ENOLINK;
        } else if (!m0_uint128_eq(&id_connection,
                                  &req_fop->lmf_id_connection)) {
                /* link exists but connection ID doesn't match */
                M0_LOG(M0_WARN, "connection ID doesn't match: "
                       "id_connection="U128X_F" lmf_id_connection="U128X_F,
                       U128_P(&id_connection),
                       U128_P(&req_fop->lmf_id_connection));
                rep_fop->lmr_rc = -EBADSLT;
        } else {
                m0_mutex_lock(&hl->hln_lock);
                if (req_fop->lmf_msg_nr != 0)
                        ha_link_msg_received(hl, &req_fop->lmf_msg);
                if (!hl->hln_no_new_delivered) {
                        ha_link_tags_update(hl, req_fop->lmf_out_next,
                                            req_fop->lmf_in_delivered);
                }
                ha_link_tags_in_out(hl, &rep_fop->lmr_out_next,
                                    &rep_fop->lmr_in_delivered);
                m0_mutex_unlock(&hl->hln_lock);
                ha_link_msg_recv_or_delivery_broadcast(hl);
                rep_fop->lmr_rc = 0;
        }
        M0_LOG(M0_DEBUG, "hl=%p lmr_rc=%"PRIu32, hl, rep_fop->lmr_rc);

        m0_rpc_reply_post(m0_fop_to_rpc_item(fom->fo_fop),
                          m0_fop_to_rpc_item(fom->fo_rep_fop));
        m0_fom_phase_set(fom, M0_FOPH_FINISH);
        return M0_FSO_WAIT;
}

static void ha_link_incoming_fom_fini(struct m0_fom *fom)
{
        struct ha_link_incoming_fom *hli = ha_link_incoming_fom_container(fom);

        m0_fom_fini(fom);
        if (hli->hli_hl != NULL)
                m0_ha_link_service_put(hli->hli_fom.fo_service, hli->hli_hl);
        m0_free(hli);
}

static size_t ha_link_incoming_fom_locality(const struct m0_fom *fom)
{
        return 0;
}

const struct m0_fom_ops ha_link_incoming_fom_ops = {
        .fo_tick          = &ha_link_incoming_fom_tick,
        .fo_fini          = &ha_link_incoming_fom_fini,
        .fo_home_locality = &ha_link_incoming_fom_locality,
};

static int ha_link_incoming_fom_create(struct m0_fop   *fop,
                                       struct m0_fom  **m,
                                       struct m0_reqh  *reqh)
{
        struct ha_link_incoming_fom   *hli;
        struct m0_fom                 *fom;
        struct m0_ha_link_msg_rep_fop *reply;

        M0_PRE(fop != NULL);
        M0_PRE(m != NULL);

        M0_ALLOC_PTR(hli);
        if (hli == NULL)
                return M0_ERR(-ENOMEM);
        fom = &hli->hli_fom;

        M0_ALLOC_PTR(reply);
        if (reply == NULL) {
                m0_free(hli);
                return M0_ERR(-ENOMEM);
        }

        fom->fo_rep_fop = m0_fop_alloc(&m0_ha_link_msg_rep_fopt,
                                       reply, m0_fop_rpc_machine(fop));
        if (fom->fo_rep_fop == NULL) {
                m0_free(reply);
                m0_free(hli);
                return M0_ERR(-ENOMEM);
        }

        m0_fom_init(fom, &fop->f_type->ft_fom_type, &ha_link_incoming_fom_ops,
                    fop, fom->fo_rep_fop, reqh);

        *m = fom;
        return M0_RC(0);
}

const struct m0_fom_type_ops m0_ha_link_incoming_fom_type_ops = {
        .fto_create = &ha_link_incoming_fom_create,
};

enum ha_link_outgoing_fom_state {
        HA_LINK_OUTGOING_STATE_INIT   = M0_FOM_PHASE_INIT,
        HA_LINK_OUTGOING_STATE_FINISH = M0_FOM_PHASE_FINISH,
        HA_LINK_OUTGOING_STATE_INCOMING_REGISTER,
        HA_LINK_OUTGOING_STATE_INCOMING_DEREGISTER,
        HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE,
        HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE_WAIT,
        HA_LINK_OUTGOING_STATE_RPC_LINK_INIT,
        HA_LINK_OUTGOING_STATE_RPC_LINK_FINI,
        HA_LINK_OUTGOING_STATE_NOT_CONNECTED,
        HA_LINK_OUTGOING_STATE_CONNECT,
        HA_LINK_OUTGOING_STATE_CONNECTING,
        HA_LINK_OUTGOING_STATE_DISCONNECT,
        HA_LINK_OUTGOING_STATE_DISCONNECTING,
        HA_LINK_OUTGOING_STATE_IDLE,
        HA_LINK_OUTGOING_STATE_SEND,
        HA_LINK_OUTGOING_STATE_WAIT_REPLY,
        HA_LINK_OUTGOING_STATE_WAIT_RELEASE,
        HA_LINK_OUTGOING_STATE_NR,
};

static struct m0_sm_state_descr
ha_link_outgoing_fom_states[HA_LINK_OUTGOING_STATE_NR] = {
#define _ST(name, flags, allowed)      \
        [name] = {                    \
                .sd_flags   = flags,  \
                .sd_name    = #name,  \
                .sd_allowed = allowed \
        }
        _ST(HA_LINK_OUTGOING_STATE_INIT, M0_SDF_INITIAL,
           M0_BITS(HA_LINK_OUTGOING_STATE_RPC_LINK_INIT)),
        _ST(HA_LINK_OUTGOING_STATE_RPC_LINK_INIT, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_NOT_CONNECTED)),
        _ST(HA_LINK_OUTGOING_STATE_RPC_LINK_FINI, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_FINISH)),
        _ST(HA_LINK_OUTGOING_STATE_NOT_CONNECTED, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_CONNECT,
                   HA_LINK_OUTGOING_STATE_RPC_LINK_FINI)),
        _ST(HA_LINK_OUTGOING_STATE_CONNECT, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_INCOMING_REGISTER)),
        _ST(HA_LINK_OUTGOING_STATE_INCOMING_REGISTER, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_CONNECTING)),
        _ST(HA_LINK_OUTGOING_STATE_CONNECTING, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_IDLE)),
        _ST(HA_LINK_OUTGOING_STATE_IDLE, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_SEND,
                   HA_LINK_OUTGOING_STATE_DISCONNECT)),
        _ST(HA_LINK_OUTGOING_STATE_SEND, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_WAIT_REPLY)),
        _ST(HA_LINK_OUTGOING_STATE_WAIT_REPLY, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_WAIT_RELEASE)),
        _ST(HA_LINK_OUTGOING_STATE_WAIT_RELEASE, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_IDLE)),
        _ST(HA_LINK_OUTGOING_STATE_DISCONNECT, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE)),
        _ST(HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE_WAIT)),
        _ST(HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE_WAIT, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_INCOMING_DEREGISTER)),
        _ST(HA_LINK_OUTGOING_STATE_INCOMING_DEREGISTER, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_DISCONNECTING)),
        _ST(HA_LINK_OUTGOING_STATE_DISCONNECTING, 0,
           M0_BITS(HA_LINK_OUTGOING_STATE_NOT_CONNECTED)),
        _ST(HA_LINK_OUTGOING_STATE_FINISH, M0_SDF_TERMINAL, 0),
#undef _ST
};

const static struct m0_sm_conf ha_link_outgoing_fom_conf = {
        .scf_name      = "ha_link_outgoing_fom",
        .scf_nr_states = ARRAY_SIZE(ha_link_outgoing_fom_states),
        .scf_state     = ha_link_outgoing_fom_states,
};

static void ha_link_outgoing_item_sent(struct m0_rpc_item *item)
{
        struct m0_ha_link *hl;

        /* XXX bob_of */
        hl = container_of(container_of(item, struct m0_fop, f_item),
                          struct m0_ha_link, hln_outgoing_fop);
        M0_ENTRY("hl=%p item=%p", hl, item);
        M0_LEAVE();
}

static void ha_link_outgoing_item_replied(struct m0_rpc_item *item)
{
        struct m0_ha_link *hl;
        int                rc;

        /* XXX bob_of */
        hl = container_of(container_of(item, struct m0_fop, f_item),
                          struct m0_ha_link, hln_outgoing_fop);
        M0_ENTRY("hl=%p item=%p ri_error=%"PRIi32, hl, item, item->ri_error);
        m0_mutex_lock(&hl->hln_lock);
        M0_ASSERT(!hl->hln_replied);
        hl->hln_replied = true;
        hl->hln_rpc_rc  = item->ri_error ?:
                          m0_rpc_item_generic_reply_rc(item->ri_reply);
        rc = hl->hln_rpc_rc;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE("hl=%p item=%p rc=%d", hl, item, rc);
}

const static struct m0_rpc_item_ops ha_link_outgoing_item_ops = {
        .rio_sent    = &ha_link_outgoing_item_sent,
        .rio_replied = &ha_link_outgoing_item_replied,
};

static void ha_link_outgoing_fop_release(struct m0_ref *ref)
{
        struct m0_ha_link *hl;
        struct m0_fop     *fop = container_of(ref, struct m0_fop, f_ref);

        /* XXX bob_of */
        hl = container_of(fop, struct m0_ha_link, hln_outgoing_fop);
        M0_ENTRY("hl=%p fop=%p", hl, fop);
        fop->f_data.fd_data = NULL;
        m0_fop_fini(fop);
        m0_mutex_lock(&hl->hln_lock);
        M0_ASSERT(!hl->hln_released);
        hl->hln_released = true;
        m0_mutex_unlock(&hl->hln_lock);
        ha_link_outgoing_fom_wakeup(hl);
        M0_LEAVE();
}

static int ha_link_outgoing_fop_send(struct m0_ha_link *hl)
{
        struct m0_ha_link_msg_fop *req_fop = &hl->hln_req_fop_data;
        struct m0_ha_link_params  *params;
        struct m0_rpc_item        *item;

        M0_ENTRY("hl=%p", hl);
        M0_SET0(&hl->hln_outgoing_fop);
        M0_SET0(req_fop);
        m0_fop_init(&hl->hln_outgoing_fop, &m0_ha_link_msg_fopt,
                    req_fop, &ha_link_outgoing_fop_release);

        if (hl->hln_msg_to_send == NULL) {
                req_fop->lmf_msg_nr = 0;
        } else {
                req_fop->lmf_msg_nr = 1;
                req_fop->lmf_msg    = *hl->hln_msg_to_send;
        }
        m0_mutex_lock(&hl->hln_lock);
        /* TODO use designated initialiser after m0_ha_msg become small */
        params = &hl->hln_conn_cfg.hlcc_params;
        req_fop->lmf_id_local       = params->hlp_id_local;
        req_fop->lmf_id_remote      = params->hlp_id_remote;
        req_fop->lmf_id_connection  = params->hlp_id_connection;
        req_fop->lmf_seq            = ++hl->hln_req_fop_seq;
        ha_link_tags_in_out(hl, &req_fop->lmf_out_next,
                            &req_fop->lmf_in_delivered);
        M0_LOG(M0_DEBUG, "lmf_id_remote="U128X_F" lmf_id_local="U128X_F" "
               "lmf_id_connection="U128X_F" lmf_msg_nr=%" PRIu64 " tag=%" PRIu64 " "
               "lmf_seq=%"PRIu64,
               U128_P(&req_fop->lmf_id_remote),
               U128_P(&req_fop->lmf_id_local),
               U128_P(&req_fop->lmf_id_connection),
               req_fop->lmf_msg_nr,
               hl->hln_msg_to_send == NULL ?
               M0_HA_MSG_TAG_UNKNOWN : m0_ha_msg_tag(hl->hln_msg_to_send),
               req_fop->lmf_seq);
        m0_mutex_unlock(&hl->hln_lock);
        item = m0_fop_to_rpc_item(&hl->hln_outgoing_fop);
        item->ri_prio            = M0_RPC_ITEM_PRIO_MID;
        item->ri_deadline        = m0_time_from_now(0, 0);
        item->ri_resend_interval = hl->hln_conn_cfg.hlcc_resend_interval;
        item->ri_nr_sent_max     = hl->hln_conn_cfg.hlcc_nr_sent_max;
        item->ri_ops             = &ha_link_outgoing_item_ops;
        item->ri_session         = &hl->hln_rpc_link.rlk_sess;
        m0_rpc_post(item);
        M0_LEAVE("hl=%p", hl);
        return 0;
}

static bool ha_link_backoff_check(struct m0_ha_link *hl,
                                  int                rc,
                                  uint64_t          *nr,
                                  int               *old_rc,
                                  uint64_t          *old_nr)
{
        *old_nr = hl->hln_backoff_nr;
        *old_rc = hl->hln_backoff_rc;
        if (hl->hln_backoff_rc == rc || hl->hln_backoff_nr == 0) {
                *nr = ++hl->hln_backoff_nr;
                return rc != 0 && m0_is_po2(hl->hln_backoff_nr);
        }
        hl->hln_backoff_nr = 1;
        hl->hln_backoff_rc = rc;
        *nr = 1;
        return true;
}

static int ha_link_outgoing_fop_replied(struct m0_ha_link *hl)
{
        struct m0_ha_link_msg_rep_fop *rep_fop;
        struct m0_ha_link_tags         tags;
        struct m0_rpc_item            *req_item;
        uint64_t                       old_nr;
        uint64_t                       nr;
        int                            old_rc;
        int                            rc;

        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));

        rc = hl->hln_rpc_rc;
        if (rc == 0) {
                req_item = m0_fop_to_rpc_item(&hl->hln_outgoing_fop);
                rep_fop  = m0_fop_data(m0_rpc_item_to_fop(req_item->ri_reply));
                rc = rep_fop->lmr_rc;
                M0_LOG(M0_DEBUG, "lmr_rc=%"PRIi32, rep_fop->lmr_rc);
                if (rc == 0)
                        ha_link_tags_update(hl, rep_fop->lmr_out_next,
                                            rep_fop->lmr_in_delivered);
        }

        if (rc != 0)
                m0_ha_lq_try_unnext(&hl->hln_q_out);

        if (ha_link_backoff_check(hl, rc, &nr, &old_rc, &old_nr)) {
                m0_ha_lq_tags_get(&hl->hln_q_out, &tags);
                M0_LOG(M0_WARN, "rc=%d nr=%" PRIu64 " hl=%p ep=%s "
                       "lq_tags="HLTAGS_F, rc, nr, hl,
                       hl->hln_conn_cfg.hlcc_rpc_endpoint, HLTAGS_P(&tags));
                if (rc != old_rc) {
                        M0_LOG(M0_WARN, "old_rc=%d old_nr=%" PRIu64 " "
                               "hl=%p ep=%s", old_rc, old_nr, hl,
                               hl->hln_conn_cfg.hlcc_rpc_endpoint);
                }
        }

        return M0_RC(rc);
}

static bool ha_link_q_in_confirm_all(struct m0_ha_link *hl)
{
        bool confirmed_updated = false;

        M0_PRE(m0_mutex_is_locked(&hl->hln_lock));
        while (m0_ha_lq_dequeue(&hl->hln_q_in) != M0_HA_MSG_TAG_INVALID)
                confirmed_updated = true;
        return confirmed_updated;
}

static void ha_link_cb_disconnecting_reused(struct m0_ha_link *hl)
{
        bool cb_disconnecting;
        bool cb_reused;

        M0_ENTRY("hl=%p", hl);
        m0_mutex_lock(&hl->hln_lock);
        cb_disconnecting = hl->hln_cb_disconnecting;
        cb_reused        = hl->hln_cb_reused;
        hl->hln_cb_disconnecting = false;
        hl->hln_cb_reused        = false;
        m0_mutex_unlock(&hl->hln_lock);
        if (cb_disconnecting) {
                m0_sm_group_lock(&hl->hln_sm_group);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_DISCONNECTING);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                m0_sm_group_unlock(&hl->hln_sm_group);
        }
        if (cb_reused) {
                m0_sm_group_lock(&hl->hln_sm_group);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_LINK_REUSED);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                m0_sm_group_unlock(&hl->hln_sm_group);
        }
        M0_LEAVE("hl=%p cb_disconnecting=%d cb_reused=%d",
                 hl, !!cb_disconnecting, !!cb_reused);
}

static void ha_link_outgoing_reconnect_timeout(struct m0_sm_timer *timer)
{
        struct m0_ha_link *hl = container_of(timer, struct m0_ha_link,
                                             hln_reconnect_wait_timer);

        m0_mutex_lock(&hl->hln_lock);
        hl->hln_reconnect_wait = false;
        m0_mutex_unlock(&hl->hln_lock);

        ha_link_outgoing_fom_wakeup(hl);
}

static int ha_link_outgoing_fom_tick(struct m0_fom *fom)
{
        enum ha_link_outgoing_fom_state  phase;
        struct m0_ha_link               *hl;
        m0_time_t                        abs_timeout;
        bool                             replied;
        bool                             released;
        bool                             stopping;
        bool                             rpc_event_occurred;
        bool                             reconnect;
        bool                             reconnect_wait;
        bool                             quiesced;
        int                              reply_rc;
        int                              rc;

        hl = container_of(fom, struct m0_ha_link, hln_fom); /* XXX bob_of */
        phase = m0_fom_phase(&hl->hln_fom);
        M0_ENTRY("hl=%p phase=%s", hl, m0_fom_phase_name(&hl->hln_fom, phase));

        switch (phase) {
        case HA_LINK_OUTGOING_STATE_INIT:
                hl->hln_msg_to_send      = NULL;
                hl->hln_confirmed_update = false;
                hl->hln_rpc_rc           = 0;
                hl->hln_reply_rc         = 0;
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_RPC_LINK_INIT);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_RPC_LINK_INIT:
                m0_sm_group_lock(&hl->hln_sm_group);
                hl->hln_tag_broadcast_recv     = 0;
                hl->hln_tag_broadcast_delivery = 0;
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_START);
                m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                m0_sm_group_unlock(&hl->hln_sm_group);
                rc = m0_rpc_link_init(&hl->hln_rpc_link,
                                      hl->hln_cfg.hlc_rpc_machine,
                                      &hl->hln_conn_cfg.hlcc_rpc_service_fid,
                                      hl->hln_conn_cfg.hlcc_rpc_endpoint,
                                      hl->hln_conn_cfg.hlcc_max_rpcs_in_flight);
                M0_ASSERT(rc == 0);     /* XXX handle it */
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_NOT_CONNECTED);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_RPC_LINK_FINI:
                if (m0_sm_timer_is_armed(&hl->hln_reconnect_wait_timer))
                        m0_sm_timer_cancel(&hl->hln_reconnect_wait_timer);
                m0_rpc_link_fini(&hl->hln_rpc_link);
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_FINISH);
                return M0_RC(M0_FSO_WAIT);
        case HA_LINK_OUTGOING_STATE_NOT_CONNECTED:
                ha_link_cb_disconnecting_reused(hl);
                m0_mutex_lock(&hl->hln_lock);
                stopping = hl->hln_fom_is_stopping;
                m0_mutex_unlock(&hl->hln_lock);
                if (!stopping) {
                        m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_CONNECT);
                        return M0_RC(M0_FSO_AGAIN);
                } else {
                        m0_mutex_lock(&hl->hln_lock);
                        hl->hln_fom_enable_wakeup = false;
                        hl->hln_no_new_delivered = true;
                        while (m0_ha_lq_next(&hl->hln_q_out) != NULL)
                                ;
                        while (m0_ha_lq_tag_delivered(&hl->hln_q_out) <
                               m0_ha_lq_tag_next(&hl->hln_q_out)) {
                                m0_ha_lq_mark_delivered(&hl->hln_q_out,
                                        m0_ha_lq_tag_delivered(&hl->hln_q_out));
                        }
                        (void)ha_link_q_in_confirm_all(hl);
                        m0_mutex_unlock(&hl->hln_lock);
                        ha_link_msg_recv_or_delivery_broadcast(hl);
                        m0_sm_group_lock(&hl->hln_sm_group);
                        m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_STOP);
                        m0_sm_group_unlock(&hl->hln_sm_group);
                        m0_mutex_lock(&hl->hln_lock);
                        if (hl->hln_reconnect_cfg_is_set) {
                                ha_link_conn_cfg_free(
                                                &hl->hln_conn_reconnect_cfg);
                        }
                        ha_link_conn_cfg_free(&hl->hln_conn_cfg);
                        m0_mutex_unlock(&hl->hln_lock);
                        m0_sm_ast_cancel(hl->hln_fom_locality->lo_grp,
                                         &hl->hln_waking_ast);
                        m0_fom_phase_set(fom,
                                         HA_LINK_OUTGOING_STATE_RPC_LINK_FINI);
                        return M0_RC(M0_FSO_AGAIN);
                }
        case HA_LINK_OUTGOING_STATE_CONNECT:
                m0_mutex_lock(&hl->hln_lock);
                hl->hln_rpc_event_occurred = false;
                m0_mutex_unlock(&hl->hln_lock);
                m0_rpc_link_reset(&hl->hln_rpc_link);
                abs_timeout = m0_time_add(m0_time_now(),
                                          hl->hln_conn_cfg.hlcc_connect_timeout);
                m0_rpc_link_connect_async(&hl->hln_rpc_link, abs_timeout,
                                          &hl->hln_rpc_wait);
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_INCOMING_REGISTER);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_INCOMING_REGISTER:
                m0_mutex_lock(&hl->hln_lock);
                if (hl->hln_reconnect) {
                        hl->hln_reconnect = false;
                        M0_ASSERT(hl->hln_reconnect_cfg_is_set);
                        ha_link_conn_cfg_free(&hl->hln_conn_cfg);
                        rc = ha_link_conn_cfg_copy(&hl->hln_conn_cfg,
                                                   &hl->hln_conn_reconnect_cfg);
                        M0_ASSERT(rc == 0);     /* XXX */
                        ha_link_conn_cfg_free(&hl->hln_conn_reconnect_cfg);
                        hl->hln_reconnect_cfg_is_set = false;
                        ha_link_tags_apply(hl, &hl->hln_conn_cfg.hlcc_params);
                }
                m0_ha_link_service_register(hl->hln_cfg.hlc_reqh_service, hl,
                            &hl->hln_conn_cfg.hlcc_params.hlp_id_local,
                            &hl->hln_conn_cfg.hlcc_params.hlp_id_connection);
                m0_mutex_unlock(&hl->hln_lock);
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_CONNECTING);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_CONNECTING:
                m0_mutex_lock(&hl->hln_lock);
                rpc_event_occurred = hl->hln_rpc_event_occurred;
                m0_mutex_unlock(&hl->hln_lock);
                if (rpc_event_occurred) {
                        M0_ASSERT_INFO(hl->hln_rpc_link.rlk_rc == 0,
                                       "rlk_rc=%d", hl->hln_rpc_link.rlk_rc);
                        m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_IDLE);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_RC(M0_FSO_WAIT);
        case HA_LINK_OUTGOING_STATE_DISCONNECT:
                m0_mutex_lock(&hl->hln_lock);
                hl->hln_rpc_event_occurred = false;
                m0_mutex_unlock(&hl->hln_lock);
                abs_timeout = m0_time_add(m0_time_now(),
                                  hl->hln_conn_cfg.hlcc_disconnect_timeout);
                m0_rpc_link_disconnect_async(&hl->hln_rpc_link, abs_timeout,
                                             &hl->hln_rpc_wait);
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE:
                m0_mutex_lock(&hl->hln_lock);
                hl->hln_quiesced = false;
                m0_mutex_unlock(&hl->hln_lock);
                m0_ha_link_service_quiesce(hl->hln_cfg.hlc_reqh_service, hl,
                                           &hl->hln_quiesce_chan);
                m0_fom_phase_set(fom,
                                 HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE_WAIT);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_INCOMING_QUIESCE_WAIT:
                m0_mutex_lock(&hl->hln_lock);
                quiesced = hl->hln_quiesced;
                m0_mutex_unlock(&hl->hln_lock);
                if (quiesced) {
                        m0_fom_phase_set(fom,
                                 HA_LINK_OUTGOING_STATE_INCOMING_DEREGISTER);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_RC(M0_FSO_WAIT);
        case HA_LINK_OUTGOING_STATE_INCOMING_DEREGISTER:
                m0_ha_link_service_deregister(hl->hln_cfg.hlc_reqh_service, hl);
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_DISCONNECTING);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_DISCONNECTING:
                m0_mutex_lock(&hl->hln_lock);
                rpc_event_occurred = hl->hln_rpc_event_occurred;
                m0_mutex_unlock(&hl->hln_lock);
                if (rpc_event_occurred) {
                        if (hl->hln_rpc_link.rlk_rc != 0) {
                                M0_LOG(M0_WARN, "rlk_rc=%d endpoint=%s",
                                       hl->hln_rpc_link.rlk_rc,
                                      m0_rpc_link_end_point(&hl->hln_rpc_link));
                        }
                        m0_fom_phase_set(fom,
                                         HA_LINK_OUTGOING_STATE_NOT_CONNECTED);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_RC(M0_FSO_WAIT);
        case HA_LINK_OUTGOING_STATE_IDLE:
                M0_ASSERT(hl->hln_msg_to_send == NULL);
                hl->hln_replied  = false;
                hl->hln_released = false;
                ha_link_cb_disconnecting_reused(hl);
                if (m0_semaphore_trydown(&hl->hln_stop_cond)) {
                        M0_LOG(M0_DEBUG, "stop case");
                        m0_mutex_lock(&hl->hln_lock);
                        hl->hln_fom_is_stopping = true;
                        m0_mutex_unlock(&hl->hln_lock);
                        ha_link_msg_recv_or_delivery_broadcast(hl);
                        m0_fom_phase_set(fom,
                                         HA_LINK_OUTGOING_STATE_DISCONNECT);
                        return M0_RC(M0_FSO_AGAIN);
                }
                reply_rc = hl->hln_reply_rc;
                if (reply_rc != 0) {
                        M0_LOG(M0_DEBUG, "link failed, ha_link reconnect case. "
                               "reply_rc=%d", reply_rc);
                        hl->hln_reply_rc = 0;
                        m0_sm_group_lock(&hl->hln_sm_group);
                        m0_sm_state_set(&hl->hln_sm,
                                        M0_HA_LINK_STATE_LINK_FAILED);
                        m0_sm_state_set(&hl->hln_sm, M0_HA_LINK_STATE_IDLE);
                        m0_sm_group_unlock(&hl->hln_sm_group);

                        m0_mutex_lock(&hl->hln_lock);
                        hl->hln_reconnect_wait = true;
                        m0_mutex_unlock(&hl->hln_lock);

                        return M0_RC(M0_FSO_AGAIN);
                }
                m0_mutex_lock(&hl->hln_lock);
                reconnect_wait = hl->hln_reconnect_wait;
                m0_mutex_unlock(&hl->hln_lock);
                if (reconnect_wait) {
                        time_t rtime = hl->hln_conn_cfg.hlcc_reconnect_interval;
                        M0_LOG(M0_DEBUG, "link failed, reconnect wait case");
                        if (m0_sm_timer_is_armed(&hl->hln_reconnect_wait_timer))
                                return M0_RC(M0_FSO_WAIT);

                        M0_LOG(M0_DEBUG, "link failed, reconnect wait case "
                               "armed");
                        m0_sm_timer_fini(&hl->hln_reconnect_wait_timer);
                        m0_sm_timer_init(&hl->hln_reconnect_wait_timer);
                        rc = m0_sm_timer_start(&hl->hln_reconnect_wait_timer,
                                       hl->hln_fom_locality->lo_grp,
                                       ha_link_outgoing_reconnect_timeout,
                                       m0_time_add(m0_time_now(), rtime));
                        M0_ASSERT_INFO(rc == 0, "hl->hln_reconnect_wait_timer "
                                       "failed to start, rc=%d", rc);
                        return M0_RC(M0_FSO_WAIT);
                }
                m0_mutex_lock(&hl->hln_lock);
                reconnect = hl->hln_reconnect;
                m0_mutex_unlock(&hl->hln_lock);
                if (reconnect) {
                        M0_LOG(M0_DEBUG, "link reconnect case");
                        m0_fom_phase_set(fom,
                                         HA_LINK_OUTGOING_STATE_DISCONNECT);
                        return M0_RC(M0_FSO_AGAIN);
                }
                m0_mutex_lock(&hl->hln_lock);
                hl->hln_msg_to_send = m0_ha_lq_next(&hl->hln_q_out);
                hl->hln_confirmed_update = ha_link_q_in_confirm_all(hl);
                m0_mutex_unlock(&hl->hln_lock);
                if (hl->hln_msg_to_send != NULL || hl->hln_confirmed_update) {
                        m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_SEND);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_RC(M0_FSO_WAIT);
        case HA_LINK_OUTGOING_STATE_SEND:
                rc = ha_link_outgoing_fop_send(hl);
                M0_ASSERT(rc == 0);     /* XXX handle it */
                m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_WAIT_REPLY);
                return M0_RC(M0_FSO_AGAIN);
        case HA_LINK_OUTGOING_STATE_WAIT_REPLY:
                m0_mutex_lock(&hl->hln_lock);
                replied = hl->hln_replied;
                m0_mutex_unlock(&hl->hln_lock);
                if (replied) {
                        m0_mutex_lock(&hl->hln_lock);
                        hl->hln_reply_rc = ha_link_outgoing_fop_replied(hl);
                        hl->hln_msg_to_send = NULL;
                        m0_mutex_unlock(&hl->hln_lock);
                        if (hl->hln_reply_rc == 0)
                                hl->hln_confirmed_update = false;
                        ha_link_msg_recv_or_delivery_broadcast(hl);
                        m0_fop_put_lock(&hl->hln_outgoing_fop);
                        m0_fom_phase_set(fom,
                                         HA_LINK_OUTGOING_STATE_WAIT_RELEASE);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_FSO_WAIT;
        case HA_LINK_OUTGOING_STATE_WAIT_RELEASE:
                m0_mutex_lock(&hl->hln_lock);
                released = hl->hln_released;
                m0_mutex_unlock(&hl->hln_lock);
                if (released) {
                        m0_fom_phase_set(fom, HA_LINK_OUTGOING_STATE_IDLE);
                        return M0_RC(M0_FSO_AGAIN);
                }
                return M0_FSO_WAIT;
        case HA_LINK_OUTGOING_STATE_FINISH:
        case HA_LINK_OUTGOING_STATE_NR:
                M0_IMPOSSIBLE("");
        }
        return M0_RC(M0_FSO_WAIT);
}

static void ha_link_outgoing_fom_wakeup_ast(struct m0_sm_group *gr,
                                            struct m0_sm_ast   *ast)
{
        struct m0_ha_link *hl = ast->sa_datum;

        M0_ENTRY("hl=%p", hl);
        m0_mutex_lock(&hl->hln_lock);
        hl->hln_waking_up = false;
        m0_mutex_unlock(&hl->hln_lock);
        if (m0_fom_is_waiting(&hl->hln_fom)) {
                M0_LOG(M0_DEBUG, "waking up");
                m0_fom_ready(&hl->hln_fom);
        }
        M0_LEAVE();
}

static void ha_link_outgoing_fom_wakeup(struct m0_ha_link *hl)
{
        M0_ENTRY("hl=%p", hl);
        m0_mutex_lock(&hl->hln_lock);
        if (!hl->hln_waking_up && hl->hln_fom_enable_wakeup) {
                M0_LOG(M0_DEBUG, "posting ast");
                hl->hln_waking_up = true;
                hl->hln_waking_ast = (struct m0_sm_ast){
                        .sa_cb    = &ha_link_outgoing_fom_wakeup_ast,
                        .sa_datum = hl,
                };
                m0_sm_ast_post(hl->hln_fom_locality->lo_grp,
                               &hl->hln_waking_ast);
        }
        m0_mutex_unlock(&hl->hln_lock);
        M0_LEAVE();
}

static void ha_link_outgoing_fom_fini(struct m0_fom *fom)
{
        struct m0_ha_link *hl;

        hl = container_of(fom, struct m0_ha_link, hln_fom); /* XXX bob_of */
        M0_ENTRY("fom=%p hl=%p", fom, hl);
        m0_fom_fini(fom);
        m0_chan_broadcast_lock(&hl->hln_stop_chan);
        M0_LEAVE();
}

static size_t ha_link_outgoing_fom_locality(const struct m0_fom *fom)
{
        return 0;
}

const struct m0_fom_ops ha_link_outgoing_fom_ops = {
        .fo_tick          = &ha_link_outgoing_fom_tick,
        .fo_fini          = &ha_link_outgoing_fom_fini,
        .fo_home_locality = &ha_link_outgoing_fom_locality,
};

static int ha_link_outgoing_fom_create(struct m0_fop   *fop,
                                       struct m0_fom  **m,
                                       struct m0_reqh  *reqh)
{
        struct m0_fom *fom;

        M0_PRE(fop != NULL);
        M0_PRE(m != NULL);

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

        m0_fom_init(fom, &fop->f_type->ft_fom_type, &ha_link_outgoing_fom_ops,
                    fop, fom->fo_rep_fop, reqh);

        *m = fom;
        return M0_RC(0);
}

const struct m0_fom_type_ops m0_ha_link_outgoing_fom_type_ops = {
        .fto_create = &ha_link_outgoing_fom_create,
};

M0_INTERNAL struct m0_rpc_session *m0_ha_link_rpc_session(struct m0_ha_link *hl)
{
        return &hl->hln_rpc_link.rlk_sess;
}

M0_INTERNAL void m0_ha_link_rpc_endpoint(struct m0_ha_link *hl,
                                         char              *buf,
                                         m0_bcount_t        buf_len)
{
        m0_mutex_lock(&hl->hln_lock);
        strncpy(buf, hl->hln_conn_cfg.hlcc_rpc_endpoint, buf_len - 1);
        buf[buf_len - 1] = 0;
        m0_mutex_unlock(&hl->hln_lock);
}

M0_INTERNAL int m0_ha_link_mod_init(void)
{
        int rc;

        rc = m0_ha_link_fops_init();
        M0_ASSERT(rc == 0);
        m0_fom_type_init(&ha_link_outgoing_fom_type, M0_HA_LINK_OUTGOING_OPCODE,
                         &m0_ha_link_outgoing_fom_type_ops,
                         &m0_ha_link_service_type, &ha_link_outgoing_fom_conf);
        return 0;
}

M0_INTERNAL void m0_ha_link_mod_fini(void)
{
        m0_ha_link_fops_fini();
}

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