klnet_core.c

Go to the documentation of this file.

/* -*- C -*- */
/*
 * Copyright (c) 2012-2020 Seagate Technology LLC and/or its Affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * For any questions about this software or licensing,
 * please email opensource@seagate.com or cortx-questions@seagate.com.
 *
 */


/*
 ******************************************************************************
 End of DLD
 ******************************************************************************
 */

#include "lib/mutex.h"
#include "lib/string.h"         /* m0_streq */
#include "net/lnet/linux_kernel/klnet_core.h"

/* LNet API, LNET_NIDSTR_SIZE */
#if M0_LUSTRE_VERSION < 2110
#include <lnet/lnet.h>
#else
#include <lnet/nidstr.h>
#endif

#if LUSTRE_VERSION_CODE >= OBD_OCD_VERSION(2, 7, 51, 0)
#include <lnet/api.h>
#include <lnet/lib-types.h>
#endif

/* include local files */
#include "net/lnet/linux_kernel/klnet_vec.c"
#include "net/lnet/linux_kernel/klnet_utils.c"

static struct m0_mutex nlx_kcore_mutex;

static struct m0_tl nlx_kcore_tms;

M0_TL_DESCR_DEFINE(tms, "nlx tms", static, struct nlx_kcore_transfer_mc,
                   ktm_tm_linkage, ktm_magic, M0_NET_LNET_KCORE_TM_MAGIC,
                   M0_NET_LNET_KCORE_TMS_MAGIC);
M0_TL_DEFINE(tms, static, struct nlx_kcore_transfer_mc);

M0_TL_DESCR_DEFINE(drv_tms, "drv tms", static, struct nlx_kcore_transfer_mc,
                   ktm_drv_linkage, ktm_magic, M0_NET_LNET_KCORE_TM_MAGIC,
                   M0_NET_LNET_DEV_TMS_MAGIC);
M0_TL_DEFINE(drv_tms, static, struct nlx_kcore_transfer_mc);

M0_TL_DESCR_DEFINE(drv_bufs, "drv bufs", static, struct nlx_kcore_buffer,
                   kb_drv_linkage, kb_magic, M0_NET_LNET_KCORE_BUF_MAGIC,
                   M0_NET_LNET_DEV_BUFS_MAGIC);
M0_TL_DEFINE(drv_bufs, static, struct nlx_kcore_buffer);

M0_TL_DESCR_DEFINE(drv_bevs, "drv bevs", static, struct nlx_kcore_buffer_event,
                   kbe_drv_linkage, kbe_magic, M0_NET_LNET_KCORE_BEV_MAGIC,
                   M0_NET_LNET_DEV_BEVS_MAGIC);
M0_TL_DEFINE(drv_bevs, static, struct nlx_kcore_buffer_event);

/* assert the equivalence of LNet and Motr data types */
M0_BASSERT(sizeof(__u64) == sizeof(uint64_t));

struct nlx_kcore_interceptable_subs {
        int (*_nlx_kcore_LNetMDAttach)(struct nlx_kcore_transfer_mc *kctm,
                                       struct nlx_core_buffer *lcbuf,
                                       struct nlx_kcore_buffer *kcb,
                                       lnet_md_t *umd);
        int (*_nlx_kcore_LNetPut)(struct nlx_kcore_transfer_mc *kctm,
                                  struct nlx_core_buffer *lcbuf,
                                  struct nlx_kcore_buffer *kcb,
                                  lnet_md_t *umd);
        int (*_nlx_kcore_LNetGet)(struct nlx_kcore_transfer_mc *kctm,
                                  struct nlx_core_buffer *lcbuf,
                                  struct nlx_kcore_buffer *kcb,
                                  lnet_md_t *umd);
};
static struct nlx_kcore_interceptable_subs nlx_kcore_iv = {
#define _NLXIS(s) ._##s = s

        _NLXIS(nlx_kcore_LNetMDAttach),
        _NLXIS(nlx_kcore_LNetPut),
        _NLXIS(nlx_kcore_LNetGet),

#undef _NLXI
};

#define NLX_kcore_LNetMDAttach(ktm, lcbuf, kb, umd)     \
        (*nlx_kcore_iv._nlx_kcore_LNetMDAttach)(ktm, lcbuf, kb, umd)
#define NLX_kcore_LNetPut(ktm, lcbuf, kb, umd)          \
        (*nlx_kcore_iv._nlx_kcore_LNetPut)(ktm, lcbuf, kb, umd)
#define NLX_kcore_LNetGet(ktm, lcbuf, kb, umd)          \
        (*nlx_kcore_iv._nlx_kcore_LNetGet)(ktm, lcbuf, kb, umd)

static bool nlx_kcore_domain_invariant(const struct nlx_kcore_domain *kd)
{
        return kd != NULL && kd->kd_magic == M0_NET_LNET_KCORE_DOM_MAGIC &&
               (nlx_core_kmem_loc_is_empty(&kd->kd_cd_loc) ||
                nlx_core_kmem_loc_invariant(&kd->kd_cd_loc));
}

static bool nlx_kcore_buffer_invariant(const struct nlx_kcore_buffer *kcb)
{
        return _0C(kcb != NULL) &&
               _0C(kcb->kb_magic == M0_NET_LNET_KCORE_BUF_MAGIC) &&
               nlx_core_kmem_loc_invariant(&kcb->kb_cb_loc);
}

static bool nlx_kcore_buffer_event_invariant(
                                      const struct nlx_kcore_buffer_event *kbe)
{
        return kbe != NULL && kbe->kbe_magic == M0_NET_LNET_KCORE_BEV_MAGIC &&
               nlx_core_kmem_loc_invariant(&kbe->kbe_bev_loc);
}

static bool nlx_kcore_tm_invariant(const struct nlx_kcore_transfer_mc *kctm)
{
        return kctm != NULL && kctm->ktm_magic == M0_NET_LNET_KCORE_TM_MAGIC &&
               nlx_core_kmem_loc_invariant(&kctm->ktm_ctm_loc);
}

static bool nlx_kcore_addr_in_use(struct nlx_core_ep_addr *cepa)
{
        M0_PRE(m0_mutex_is_locked(&nlx_kcore_mutex));

        return m0_tl_exists(tms, scan, &nlx_kcore_tms,
                            nlx_core_ep_eq(&scan->ktm_addr, cepa));
}

static int nlx_kcore_max_tmid_find(struct nlx_core_ep_addr *cepa)
{
        int tmid = M0_NET_LNET_TMID_MAX;
        struct nlx_kcore_transfer_mc *scan;
        struct nlx_core_ep_addr *scanaddr;
        M0_PRE(m0_mutex_is_locked(&nlx_kcore_mutex));

        /* list is in descending order by tmid */
        m0_tl_for(tms, &nlx_kcore_tms, scan) {
                scanaddr = &scan->ktm_addr;
                if (scanaddr->cepa_nid == cepa->cepa_nid &&
                    scanaddr->cepa_pid == cepa->cepa_pid &&
                    scanaddr->cepa_portal == cepa->cepa_portal) {
                        if (scanaddr->cepa_tmid == tmid)
                                --tmid;
                        else if (scanaddr->cepa_tmid < tmid)
                                break;
                }
        } m0_tl_endfor;
        return tmid >= 0 ? tmid : M0_ERR(-EADDRNOTAVAIL);
}

static void nlx_kcore_tms_list_add(struct nlx_kcore_transfer_mc *kctm)
{
        struct nlx_kcore_transfer_mc *scan;
        struct nlx_core_ep_addr *scanaddr;
        struct nlx_core_ep_addr *cepa = &kctm->ktm_addr;
        M0_PRE(m0_mutex_is_locked(&nlx_kcore_mutex));

        m0_tl_for(tms, &nlx_kcore_tms, scan) {
                scanaddr = &scan->ktm_addr;
                if (scanaddr->cepa_tmid <= cepa->cepa_tmid) {
                        tms_tlist_add_before(scan, kctm);
                        return;
                }
        } m0_tl_endfor;
        tms_tlist_add_tail(&nlx_kcore_tms, kctm);
}

static void nlx_kcore_eq_cb(lnet_event_t *event)
{
        struct nlx_kcore_buffer *kbp;
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_core_transfer_mc *ctm;
        struct nlx_core_bev_link *ql;
        struct nlx_core_buffer_event *bev;
        m0_time_t now = m0_time_now();
        bool is_unlinked = false;
        unsigned mlength;
        unsigned offset;
        int status;

        M0_PRE(event != NULL);
        if (event->type == LNET_EVENT_ACK) {
                /* we do not use ACK */
                NLXDBG(&nlx_debug, 1,
                       nlx_kprint_lnet_event("nlx_kcore_eq_cb: filtered ACK",
                                             event));
                return;
        }
        kbp = event->md.user_ptr;
        M0_ASSERT(nlx_kcore_buffer_invariant(kbp));
        ktm = kbp->kb_ktm;
        M0_ASSERT(nlx_kcore_tm_invariant(ktm));

        NLXDBGP(ktm, 1, "\t%p: eq_cb: %p %s U:%d S:%d T:%d buf:%lx\n",
                ktm, event, nlx_kcore_lnet_event_type_to_string(event->type),
                event->unlinked, event->status, event->md.threshold,
                (unsigned long) kbp->kb_buffer_id);
        NLXDBG(ktm, 2, nlx_kprint_lnet_event("eq_cb", event));
        NLXDBG(ktm, 3, nlx_kprint_kcore_tm("eq_cb", ktm));

        if (event->unlinked != 0) {
                LNetInvalidateMDHandle(&kbp->kb_mdh); /* Invalid use, but safe */
                /* kbp->kb_ktm = NULL set below */
                is_unlinked = true;
        }
        status  = event->status;
        mlength = event->mlength;
        offset  = event->offset;

        if (event->type == LNET_EVENT_SEND &&
            kbp->kb_qtype == M0_NET_QT_ACTIVE_BULK_RECV) {
                /* An LNetGet related event, normally ignored */
                if (!is_unlinked) {
                        NLXDBGP(ktm, 1, "\t%p: ignored LNetGet() SEND\n", ktm);
                        return;
                }
                /* An out-of-order SEND, or
                   cancellation notification piggy-backed onto an in-order SEND.
                   The only way to distinguish is from the value of the
                   event->kb_ooo_reply field.
                 */
                NLXDBGP(ktm, 1,
                        "\t%p: LNetGet() SEND with unlinked: thr:%d ooo:%d\n",
                        ktm, event->md.threshold, (int) kbp->kb_ooo_reply);
                if (status == 0) {
                        if (!kbp->kb_ooo_reply)
                                status = -ECANCELED;
                        else {  /* from earlier REPLY */
                                mlength = kbp->kb_ooo_mlength;
                                offset  = kbp->kb_ooo_offset;
                                status  = kbp->kb_ooo_status;
                        }
                }
        } else if (event->type == LNET_EVENT_UNLINK) {/* see nlx_core_buf_del */
                M0_ASSERT(is_unlinked);
                status = -ECANCELED;
        } else if (!is_unlinked) {
                NLXDBGP(ktm, 1, "\t%p: eq_cb: %p %s !unlinked Q=%d\n", ktm,
                        event, nlx_kcore_lnet_event_type_to_string(event->type),
                        kbp->kb_qtype);
                /* We may get REPLY before SEND, so ignore such events,
                   but save the significant values for when the SEND arrives.
                 */
                if (kbp->kb_qtype == M0_NET_QT_ACTIVE_BULK_RECV) {
                        kbp->kb_ooo_reply   = true;
                        kbp->kb_ooo_mlength = mlength;
                        kbp->kb_ooo_offset  = offset;
                        kbp->kb_ooo_status  = status;
                        return;
                }
                /* we don't expect anything other than receive messages */
                M0_ASSERT(kbp->kb_qtype == M0_NET_QT_MSG_RECV);
        }

        spin_lock(&ktm->ktm_bevq_lock);
        ctm = nlx_kcore_core_tm_map_atomic(ktm);
        ql = bev_cqueue_pnext(&ctm->ctm_bevq);
        bev = container_of(ql, struct nlx_core_buffer_event, cbe_tm_link);
        bev->cbe_buffer_id = kbp->kb_buffer_id;
        bev->cbe_time      = m0_time_sub(now, kbp->kb_add_time);
        bev->cbe_status    = status;
        bev->cbe_length    = mlength;
        bev->cbe_offset    = offset;
        bev->cbe_unlinked  = is_unlinked;
        if (event->hdr_data != 0) {
                bev->cbe_sender.cepa_nid = event->initiator.nid;
                bev->cbe_sender.cepa_pid = event->initiator.pid;
                nlx_kcore_hdr_data_decode(event->hdr_data,
                                          &bev->cbe_sender.cepa_portal,
                                          &bev->cbe_sender.cepa_tmid);
        } else
                M0_SET0(&bev->cbe_sender);

        /* Reset in spinlock to synchronize with driver nlx_dev_tm_cleanup() */
        if (is_unlinked)
                kbp->kb_ktm = NULL;
        bev_cqueue_put(&ctm->ctm_bevq, ql);
        nlx_kcore_core_tm_unmap_atomic(ctm);
        spin_unlock(&ktm->ktm_bevq_lock);

        wake_up(&ktm->ktm_wq);
}

static void nlx_core_kmem_loc_set(struct nlx_core_kmem_loc *loc,
                                  struct page *pg, uint32_t off)
{
        M0_PRE(off < PAGE_SIZE && ergo(pg == NULL, off == 0));
        M0_PRE(loc != NULL);

        loc->kl_page = pg;
        loc->kl_offset = off;
        loc->kl_checksum = nlx_core_kmem_loc_checksum(loc);
}

M0_INTERNAL void *nlx_core_mem_alloc(size_t size, unsigned shift)
{
        return m0_alloc(size);
}

M0_INTERNAL void nlx_core_mem_free(void *data, size_t size, unsigned shift)
{
        m0_free(data);
}

static int nlx_kcore_core_dom_init(struct nlx_kcore_domain *kd,
                                   struct nlx_core_domain *cd)
{
        M0_PRE(kd != NULL && cd != NULL);
        M0_PRE(nlx_kcore_domain_invariant(kd));
        M0_PRE(!nlx_core_kmem_loc_is_empty(&kd->kd_cd_loc));
        cd->cd_kpvt = kd;
        return 0;
}

M0_INTERNAL int nlx_core_dom_init(struct m0_net_domain *dom,
                                  struct nlx_core_domain *cd)
{
        struct nlx_kcore_domain *kd;
        int rc;

        M0_PRE(dom != NULL && cd != NULL);
        NLX_ALLOC_PTR(kd);
        if (kd == NULL)
                return M0_ERR(-ENOMEM);
        rc = nlx_kcore_kcore_dom_init(kd);
        if (rc != 0)
                goto fail_free_kd;
        nlx_core_kmem_loc_set(&kd->kd_cd_loc, virt_to_page(cd),
                              NLX_PAGE_OFFSET((unsigned long) cd));
        rc = nlx_kcore_core_dom_init(kd, cd);
        if (rc == 0)
                return 0;

        /* failed */
        nlx_kcore_kcore_dom_fini(kd);
fail_free_kd:
        m0_free(kd);
        M0_ASSERT(rc != 0);
        return M0_RC(rc);
}

static void nlx_kcore_core_dom_fini(struct nlx_kcore_domain *kd,
                                    struct nlx_core_domain *cd)
{
        M0_PRE(cd != NULL && cd->cd_kpvt == kd);
        cd->cd_kpvt = NULL;
}

M0_INTERNAL void nlx_core_dom_fini(struct nlx_core_domain *cd)
{
        struct nlx_kcore_domain *kd;

        M0_PRE(cd != NULL);
        kd = cd->cd_kpvt;
        M0_PRE(nlx_kcore_domain_invariant(kd));
        nlx_kcore_core_dom_fini(kd, cd);
        nlx_core_kmem_loc_set(&kd->kd_cd_loc, NULL, 0);
        nlx_kcore_kcore_dom_fini(kd);
        m0_free(kd);
}

M0_INTERNAL m0_bcount_t nlx_core_get_max_buffer_size(struct nlx_core_domain
                                                     *lcdom)
{
        return LNET_MAX_PAYLOAD;
}

M0_INTERNAL m0_bcount_t nlx_core_get_max_buffer_segment_size(struct
                                                             nlx_core_domain
                                                             *lcdom)
{
        /* PAGE_SIZE limit applies only when LNET_MD_KIOV has been set in
         * lnet_md_t::options. There's no such limit in MD fragment size when
         * LNET_MD_IOVEC is set.  DLD calls for only LNET_MD_KIOV to be used.
         */
        return PAGE_SIZE;
}

M0_INTERNAL int32_t nlx_core_get_max_buffer_segments(struct nlx_core_domain
                                                     *lcdom)
{
        return LNET_MAX_IOV;
}

static int nlx_kcore_buf_register(struct nlx_kcore_domain *kd,
                                  nlx_core_opaque_ptr_t buffer_id,
                                  struct nlx_core_buffer *cb,
                                  struct nlx_kcore_buffer *kb)
{
        M0_PRE(nlx_kcore_domain_invariant(kd));
        drv_bufs_tlink_init(kb);
        kb->kb_ktm           = NULL;
        kb->kb_buffer_id     = buffer_id;
        kb->kb_kiov          = NULL;
        kb->kb_kiov_len      = 0;
        kb->kb_kiov_orig_len = 0;
        LNetInvalidateMDHandle(&kb->kb_mdh);
        kb->kb_ooo_reply     = false;
        kb->kb_ooo_mlength   = 0;
        kb->kb_ooo_status    = 0;
        kb->kb_ooo_offset    = 0;

        cb->cb_kpvt          = kb;
        cb->cb_buffer_id     = buffer_id;
        cb->cb_magic         = M0_NET_LNET_CORE_BUF_MAGIC;

        M0_POST(nlx_kcore_buffer_invariant(kb));
        return 0;
}

static void nlx_kcore_buf_deregister(struct nlx_core_buffer *cb,
                                     struct nlx_kcore_buffer *kb)
{
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(LNetMDHandleIsInvalid(kb->kb_mdh));
        drv_bufs_tlink_fini(kb);
        kb->kb_magic = 0;
        m0_free(kb->kb_kiov);
        cb->cb_buffer_id = 0;
        cb->cb_kpvt = NULL;
        cb->cb_magic = 0;
}

M0_INTERNAL int nlx_core_buf_register(struct nlx_core_domain *cd,
                                      nlx_core_opaque_ptr_t buffer_id,
                                      const struct m0_bufvec *bvec,
                                      struct nlx_core_buffer *cb)
{
        int rc;
        struct nlx_kcore_buffer *kb;
        struct nlx_kcore_domain *kd;

        M0_PRE(cb != NULL && cb->cb_kpvt == NULL);
        M0_PRE(cd != NULL);
        kd = cd->cd_kpvt;
        NLX_ALLOC_PTR(kb);
        if (kb == NULL)
                return M0_ERR(-ENOMEM);
        nlx_core_kmem_loc_set(&kb->kb_cb_loc, virt_to_page(cb),
                              NLX_PAGE_OFFSET((unsigned long) cb));
        rc = nlx_kcore_buf_register(kd, buffer_id, cb, kb);
        if (rc != 0)
                goto fail_free_kb;
        rc = nlx_kcore_buffer_kla_to_kiov(kb, bvec);
        if (rc != 0)
                goto fail_buf_registered;
        M0_ASSERT(kb->kb_kiov != NULL && kb->kb_kiov_len > 0);
        M0_POST(nlx_kcore_buffer_invariant(cb->cb_kpvt));
        return 0;

fail_buf_registered:
        nlx_kcore_buf_deregister(cb, kb);
fail_free_kb:
        m0_free(kb);
        M0_ASSERT(rc != 0);
        return M0_RC(rc);
}

M0_INTERNAL void nlx_core_buf_deregister(struct nlx_core_domain *cd,
                                         struct nlx_core_buffer *cb)
{
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        nlx_kcore_buf_deregister(cb, kb);
        m0_free(kb);
}

static int nlx_kcore_buf_msg_recv(struct nlx_kcore_transfer_mc *ktm,
                                  struct nlx_core_buffer *cb,
                                  struct nlx_kcore_buffer *kb)
{
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_MSG_RECV);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_min_receive_size <= cb->cb_length);
        M0_PRE(cb->cb_max_operations > 0);

        nlx_kcore_umd_init(ktm, cb, kb, cb->cb_max_operations,
                           cb->cb_min_receive_size, LNET_MD_OP_PUT,
                           false, &umd);
        cb->cb_match_bits =
                nlx_core_match_bits_encode(ktm->ktm_addr.cepa_tmid, 0);
        cb->cb_addr = ktm->ktm_addr;
        rc = NLX_kcore_LNetMDAttach(ktm, cb, kb, &umd);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_msg_recv(struct nlx_core_domain *cd,       /* not used */
                                      struct nlx_core_transfer_mc *ctm,
                                      struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_msg_recv(ktm, cb, kb);
}

static int nlx_kcore_buf_msg_send(struct nlx_kcore_transfer_mc *ktm,
                                  struct nlx_core_buffer *cb,
                                  struct nlx_kcore_buffer *kb)
{
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_MSG_SEND);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_max_operations == 1);

        nlx_kcore_umd_init(ktm, cb, kb, 1, 0, 0, false, &umd);
        nlx_kcore_kiov_adjust_length(ktm, kb, &umd, cb->cb_length);
        cb->cb_match_bits =
                nlx_core_match_bits_encode(cb->cb_addr.cepa_tmid, 0);
        rc = NLX_kcore_LNetPut(ktm, cb, kb, &umd);
        nlx_kcore_kiov_restore_length(kb);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_msg_send(struct nlx_core_domain *cd,       /* not used */
                                      struct nlx_core_transfer_mc *ctm,
                                      struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_msg_send(ktm, cb, kb);
}

static int nlx_kcore_buf_active_recv(struct nlx_kcore_transfer_mc *ktm,
                                     struct nlx_core_buffer *cb,
                                     struct nlx_kcore_buffer *kb)
{
        uint32_t tmid;
        uint64_t counter;
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_ACTIVE_BULK_RECV);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_max_operations == 1);

        M0_PRE(cb->cb_match_bits > 0);
        nlx_core_match_bits_decode(cb->cb_match_bits, &tmid, &counter);
        M0_PRE(tmid == cb->cb_addr.cepa_tmid);
        M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
        M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);

        nlx_kcore_umd_init(ktm, cb, kb, 1, 0, 0, true, &umd);
        nlx_kcore_kiov_adjust_length(ktm, kb, &umd, cb->cb_length);
        rc = NLX_kcore_LNetGet(ktm, cb, kb, &umd);
        nlx_kcore_kiov_restore_length(kb);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_active_recv(struct nlx_core_domain *cd,    /* not used */
                                         struct nlx_core_transfer_mc *ctm,
                                         struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_active_recv(ktm, cb, kb);
}

static int nlx_kcore_buf_active_send(struct nlx_kcore_transfer_mc *ktm,
                                     struct nlx_core_buffer *cb,
                                     struct nlx_kcore_buffer *kb)
{
        uint32_t tmid;
        uint64_t counter;
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_ACTIVE_BULK_SEND);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_max_operations == 1);

        M0_PRE(cb->cb_match_bits > 0);
        nlx_core_match_bits_decode(cb->cb_match_bits, &tmid, &counter);
        M0_PRE(tmid == cb->cb_addr.cepa_tmid);
        M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
        M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);

        nlx_kcore_umd_init(ktm, cb, kb, 1, 0, 0, false, &umd);
        nlx_kcore_kiov_adjust_length(ktm, kb, &umd, cb->cb_length);
        rc = NLX_kcore_LNetPut(ktm, cb, kb, &umd);
        nlx_kcore_kiov_restore_length(kb);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_active_send(struct nlx_core_domain *cd,    /* not used */
                                         struct nlx_core_transfer_mc *ctm,
                                         struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_active_send(ktm, cb, kb);
}

static int nlx_kcore_buf_passive_recv(struct nlx_kcore_transfer_mc *ktm,
                                      struct nlx_core_buffer *cb,
                                      struct nlx_kcore_buffer *kb)
{
        uint32_t tmid;
        uint64_t counter;
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_PASSIVE_BULK_RECV);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_max_operations == 1);
        M0_PRE(cb->cb_match_bits > 0);

        nlx_core_match_bits_decode(cb->cb_match_bits, &tmid, &counter);
        M0_PRE(tmid == ktm->ktm_addr.cepa_tmid);
        M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
        M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);

        nlx_kcore_umd_init(ktm, cb, kb, 1, 0, LNET_MD_OP_PUT, false, &umd);
        cb->cb_addr = ktm->ktm_addr;
        rc = NLX_kcore_LNetMDAttach(ktm, cb, kb, &umd);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_passive_recv(struct nlx_core_domain *cd,   /* not used */
                                          struct nlx_core_transfer_mc *ctm,
                                          struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_passive_recv(ktm, cb, kb);
}

static int nlx_kcore_buf_passive_send(struct nlx_kcore_transfer_mc *ktm,
                                      struct nlx_core_buffer *cb,
                                      struct nlx_kcore_buffer *kb)
{
        uint32_t tmid;
        uint64_t counter;
        lnet_md_t umd;
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(nlx_core_buffer_invariant(cb));
        M0_PRE(nlx_kcore_buffer_invariant(kb));
        M0_PRE(cb->cb_qtype == M0_NET_QT_PASSIVE_BULK_SEND);
        M0_PRE(cb->cb_length > 0);
        M0_PRE(cb->cb_max_operations == 1);
        M0_PRE(cb->cb_match_bits > 0);

        nlx_core_match_bits_decode(cb->cb_match_bits, &tmid, &counter);
        M0_PRE(tmid == ktm->ktm_addr.cepa_tmid);
        M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
        M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);

        nlx_kcore_umd_init(ktm, cb, kb, 1, 0, LNET_MD_OP_GET, false, &umd);
        nlx_kcore_kiov_adjust_length(ktm, kb, &umd, cb->cb_length);
        cb->cb_addr = ktm->ktm_addr;
        rc = NLX_kcore_LNetMDAttach(ktm, cb, kb, &umd);
        nlx_kcore_kiov_restore_length(kb);
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_passive_send(struct nlx_core_domain *cd,   /* not used */
                                          struct nlx_core_transfer_mc *ctm,
                                          struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;
        return nlx_kcore_buf_passive_send(ktm, cb, kb);
}

M0_INTERNAL int nlx_core_buf_del(struct nlx_core_domain *cd,    /* not used */
                                 struct nlx_core_transfer_mc *ctm,
                                 struct nlx_core_buffer *cb)
{
        struct nlx_kcore_transfer_mc *ktm;
        struct nlx_kcore_buffer *kb;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        M0_PRE(nlx_core_buffer_invariant(cb));
        kb = cb->cb_kpvt;

        /* Subtle: Cancelling the MD associated with the buffer
           could result in a LNet UNLINK event if the buffer operation is
           terminated by LNet.
           The unlink bit is also set in other LNet events but does not
           signify cancel in those cases.
         */
        return nlx_kcore_LNetMDUnlink(ktm, kb);
}

static int nlx_kcore_buf_event_wait(struct nlx_core_transfer_mc *ctm,
                                    struct nlx_kcore_transfer_mc *ktm,
                                    m0_time_t timeout)
{
        int             rc;
        struct timespec ts;
        m0_time_t       now;

        M0_PRE(nlx_core_tm_invariant(ctm));
        M0_PRE(nlx_kcore_tm_invariant(ktm));

        if (!bev_cqueue_is_empty(&ctm->ctm_bevq))
                return M0_RC(0);
        else if (timeout == 0)
                return M0_RC(-ETIMEDOUT);

        now = m0_time_now();
        timeout = timeout > now ? m0_time_sub(timeout, now) : 0;
        ts.tv_sec  = m0_time_seconds(timeout);
        ts.tv_nsec = m0_time_nanoseconds(timeout);

        rc = wait_event_interruptible_timeout(ktm->ktm_wq,
                !bev_cqueue_is_empty(&ctm->ctm_bevq), timespec_to_jiffies(&ts));

        return M0_RC(rc == 0 ? -ETIMEDOUT : rc < 0 ? rc : 0);
}

M0_INTERNAL int nlx_core_buf_event_wait(struct nlx_core_domain *cd,
                                        struct nlx_core_transfer_mc *ctm,
                                        m0_time_t timeout)
{
        struct nlx_kcore_transfer_mc *ktm;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        return nlx_kcore_buf_event_wait(ctm, ktm, timeout);
}

static int nlx_kcore_nidstr_decode(const char *nidstr, uint64_t *nid)
{
        *nid = libcfs_str2nid(nidstr);
        if (*nid == LNET_NID_ANY)
                return M0_ERR_INFO(-EINVAL, "nidstr=%s", nidstr);
        return 0;
}

M0_INTERNAL int nlx_core_nidstr_decode(struct nlx_core_domain *lcdom,   /* not used */
                                       const char *nidstr, uint64_t * nid)
{
        return nlx_kcore_nidstr_decode(nidstr, nid);
}

static int nlx_kcore_nidstr_encode(uint64_t nid,
                                   char nidstr[M0_NET_LNET_NIDSTR_SIZE])
{
        const char *cp = libcfs_nid2str(nid);

        M0_ASSERT(cp != NULL && *cp != 0);
        strncpy(nidstr, cp, M0_NET_LNET_NIDSTR_SIZE - 1);
        nidstr[M0_NET_LNET_NIDSTR_SIZE - 1] = 0;
        return 0;
}

M0_INTERNAL int nlx_core_nidstr_encode(struct nlx_core_domain *lcdom,   /* not used */
                                       uint64_t nid,
                                       char nidstr[M0_NET_LNET_NIDSTR_SIZE])
{
        return nlx_kcore_nidstr_encode(nid, nidstr);
}

M0_INTERNAL int nlx_core_nidstrs_get(struct nlx_core_domain *lcdom,
                                     char ***nidary)
{
        lnet_process_id_t  id;
        char              *nidstr;
        int                i;
        int                j;
        int                nr;
        int                rc;

        for (nr = 0, rc = 0; rc != -ENOENT; ++nr)
                rc = LNetGetId(nr, &id);
        M0_ALLOC_ARR(*nidary, nr);
        if (*nidary == NULL)
                return M0_ERR(-ENOMEM);
        for (i = 0; i < nr - 1; ++i) {
                rc = LNetGetId(i, &id);
                M0_ASSERT(rc == 0);
                nidstr = libcfs_nid2str(id.nid);
                M0_ASSERT(nidstr != NULL);
                (*nidary)[i] = m0_strdup(nidstr);
                if ((*nidary)[i] == NULL) {
                        for (j = 0; j < i; ++j)
                                m0_free((*nidary)[j]);
                        return M0_ERR(-ENOMEM);
                }
        }
        return M0_RC(0);
}

M0_INTERNAL void nlx_core_nidstrs_put(struct nlx_core_domain *lcdom,
                                      char ***nidary)
{
        int i;

        for (i = 0; (*nidary)[i] != NULL; ++i)
                m0_free((*nidary)[i]);
        m0_free(*nidary);
        *nidary = NULL;
}

M0_INTERNAL int nlx_core_new_blessed_bev(struct nlx_core_domain *cd,
                                         struct nlx_core_transfer_mc *ctm,
                                         struct nlx_core_buffer_event **bevp)
{
        struct nlx_core_buffer_event *bev;
        struct nlx_kcore_transfer_mc *ktm;

        ktm = ctm->ctm_kpvt;
        M0_ASSERT(nlx_kcore_tm_invariant(ktm));

        NLX_ALLOC_ALIGNED_PTR(bev);
        if (bev == NULL) {
                *bevp = NULL;
                return M0_ERR(-ENOMEM);
        }
        bev_link_bless(&bev->cbe_tm_link, virt_to_page(&bev->cbe_tm_link));
        *bevp = bev;
        return 0;
}

static void nlx_kcore_tm_stop(struct nlx_core_transfer_mc *ctm,
                              struct nlx_kcore_transfer_mc *ktm)
{
        int rc;

        M0_PRE(nlx_kcore_tm_invariant(ktm));
        M0_PRE(drv_bevs_tlist_is_empty(&ktm->ktm_drv_bevs));

        rc = LNetEQFree(ktm->ktm_eqh);
        M0_ASSERT(rc == 0);

        m0_mutex_lock(&nlx_kcore_mutex);
        tms_tlist_del(ktm);
        m0_mutex_unlock(&nlx_kcore_mutex);
        drv_bevs_tlist_fini(&ktm->ktm_drv_bevs);
        drv_tms_tlink_fini(ktm);
        tms_tlink_fini(ktm);
        ktm->ktm_magic = 0;
        /* allow kernel cleanup even if core is invalid */
        if (nlx_core_tm_invariant(ctm))
                ctm->ctm_kpvt = NULL;
}

M0_INTERNAL void nlx_core_tm_stop(struct nlx_core_domain *cd,
                                  struct nlx_core_transfer_mc *ctm)
{
        struct nlx_kcore_transfer_mc *ktm;

        M0_PRE(nlx_core_tm_invariant(ctm));
        ktm = ctm->ctm_kpvt;
        nlx_kcore_tm_stop(ctm, ktm);
        bev_cqueue_fini(&ctm->ctm_bevq, nlx_core_bev_free_cb);
        m0_free(ktm);
}

static int nlx_kcore_tm_start(struct nlx_kcore_domain      *kd,
                              struct nlx_core_transfer_mc  *ctm,
                              struct nlx_kcore_transfer_mc *ktm)
{
        struct nlx_core_ep_addr *cepa;
        lnet_process_id_t id;
        int rc;
        int i;

        M0_ENTRY();
        M0_PRE(kd != NULL && ctm != NULL && ktm != NULL);
        cepa = &ctm->ctm_addr;

        /*
         * cepa_nid/cepa_pid must match a local NID/PID.
         * cepa_portal must be in range.  cepa_tmid is checked below.
         */
        if (cepa->cepa_portal == LNET_RESERVED_PORTAL ||
            cepa->cepa_portal >= M0_NET_LNET_MAX_PORTALS) {
                M0_LOG(M0_ERROR, "cepa_portal=%"PRIu32" "
                       "LNET_RESERVED_PORTAL=%d M0_NET_LNET_MAX_PORTALS=%d",
                       cepa->cepa_portal, LNET_RESERVED_PORTAL,
                       M0_NET_LNET_MAX_PORTALS);
                rc = -EINVAL;
                goto fail;
        }
        for (i = 0; ; ++i) {
                rc = LNetGetId(i, &id);
                if (rc == -ENOENT)
                        break;
                M0_ASSERT_INFO(rc == 0, "rc=%d", rc);
                if (id.nid == cepa->cepa_nid && id.pid == cepa->cepa_pid)
                        break;
        }
        if (rc == -ENOENT)
                goto fail;

        rc = LNetEQAlloc(M0_NET_LNET_EQ_SIZE, nlx_kcore_eq_cb, &ktm->ktm_eqh);
        if (rc < 0) {
                M0_LOG(M0_ERROR, "LNetEQAlloc() failed: rc=%d", rc);
                goto fail;
        }

        m0_mutex_lock(&nlx_kcore_mutex);
        if (cepa->cepa_tmid == M0_NET_LNET_TMID_INVALID) {
                rc = nlx_kcore_max_tmid_find(cepa);
                if (rc < 0) {
                        m0_mutex_unlock(&nlx_kcore_mutex);
                        goto fail_with_eq;
                }
                cepa->cepa_tmid = rc;
        } else if (cepa->cepa_tmid > M0_NET_LNET_TMID_MAX) {
                m0_mutex_unlock(&nlx_kcore_mutex);
                rc = M0_ERR(-EINVAL);
                goto fail_with_eq;
        } else if (nlx_kcore_addr_in_use(cepa)) {
                m0_mutex_unlock(&nlx_kcore_mutex);
                rc = M0_ERR(-EADDRINUSE);
                goto fail_with_eq;
        }
        ktm->ktm_addr = *cepa;
        tms_tlink_init(ktm);
        nlx_kcore_tms_list_add(ktm);
        m0_mutex_unlock(&nlx_kcore_mutex);

        drv_tms_tlink_init(ktm);
        drv_bevs_tlist_init(&ktm->ktm_drv_bevs);
        ctm->ctm_mb_counter = M0_NET_LNET_BUFFER_ID_MIN;
        spin_lock_init(&ktm->ktm_bevq_lock);
        init_waitqueue_head(&ktm->ktm_wq);
        ktm->_debug_ = ctm->_debug_;
        ctm->ctm_kpvt = ktm;
        ctm->ctm_magic = M0_NET_LNET_CORE_TM_MAGIC;
        M0_POST(nlx_kcore_tm_invariant(ktm));
        return 0;

fail_with_eq:
        i = LNetEQFree(ktm->ktm_eqh);
        M0_ASSERT(i == 0);
fail:
        M0_ASSERT(rc != 0);
        return M0_ERR(rc);
}

M0_INTERNAL int nlx_core_tm_start(struct nlx_core_domain *cd,
                                  struct m0_net_transfer_mc *tm,
                                  struct nlx_core_transfer_mc *ctm)
{
        struct nlx_kcore_domain *kd;
        struct nlx_core_buffer_event *e1;
        struct nlx_core_buffer_event *e2;
        struct nlx_kcore_transfer_mc *ktm;
        int rc;

        M0_PRE(m0_mutex_is_locked(&tm->ntm_mutex));
        M0_PRE(nlx_tm_invariant(tm));
        M0_PRE(cd != NULL);
        kd = cd->cd_kpvt;
        M0_PRE(nlx_kcore_domain_invariant(kd));

        NLX_ALLOC_PTR(ktm);
        if (ktm == NULL) {
                rc = M0_ERR(-ENOMEM);
                goto fail_ktm;
        }

        nlx_core_kmem_loc_set(&ktm->ktm_ctm_loc, virt_to_page(ctm),
                              NLX_PAGE_OFFSET((unsigned long) ctm));
        rc = nlx_kcore_tm_start(kd, ctm, ktm);
        if (rc != 0)
                goto fail_ktm;

        ctm->ctm_upvt = NULL;
        rc = nlx_core_new_blessed_bev(cd, ctm, &e1);
        if (rc == 0)
                rc = nlx_core_new_blessed_bev(cd, ctm, &e2);
        if (rc != 0)
                goto fail_blessed_bev;
        M0_ASSERT(e1 != NULL && e2 != NULL);
        bev_cqueue_init(&ctm->ctm_bevq, &e1->cbe_tm_link, &e2->cbe_tm_link);
        M0_ASSERT(bev_cqueue_is_empty(&ctm->ctm_bevq));
        return 0;

 fail_blessed_bev:
        if (e1 != NULL)
                nlx_core_bev_free_cb(&e1->cbe_tm_link);
        nlx_kcore_tm_stop(ctm, ktm);
 fail_ktm:
        m0_free(ktm);
        M0_ASSERT(rc != 0);
        return M0_RC(rc);
}

static void nlx_core_fini(void)
{
        int rc;

        nlx_dev_fini();
        tms_tlist_fini(&nlx_kcore_tms);
        m0_mutex_fini(&nlx_kcore_mutex);
        rc = LNetNIFini();
        M0_ASSERT(rc == 0);
}

static int nlx_core_init(void)
{
        int rc;
        /*
         * Temporarily reset current->journal_info, because LNetNIInit assumes
         * it is NULL.
         */
        struct m0_thread_tls *tls = m0_thread_tls_pop();

        /*
         * Init LNet with same PID as Lustre would use in case we are first.
         * Depending on the lustre version, the PID symbol may be called
         * LUSTRE_SRV_LNET_PID or LNET_PID_LUSTRE.
         */
#ifdef LNET_PID_LUSTRE
        rc = LNetNIInit(LNET_PID_LUSTRE);
#else
        rc = LNetNIInit(LUSTRE_SRV_LNET_PID);
#endif
        m0_thread_tls_back(tls);
        if (rc < 0)
                return M0_RC(rc);

        m0_mutex_init(&nlx_kcore_mutex);
        tms_tlist_init(&nlx_kcore_tms);

        rc = nlx_dev_init();
        if (rc != 0)
                nlx_core_fini();

        return M0_RC(rc);
}

static struct nlx_kcore_ops nlx_kcore_def_ops = {
        .ko_dom_init = nlx_kcore_core_dom_init,
        .ko_dom_fini = nlx_kcore_core_dom_fini,
        .ko_buf_register = nlx_kcore_buf_register,
        .ko_buf_deregister = nlx_kcore_buf_deregister,
        .ko_tm_start = nlx_kcore_tm_start,
        .ko_tm_stop = nlx_kcore_tm_stop,
        .ko_buf_msg_recv = nlx_kcore_buf_msg_recv,
        .ko_buf_msg_send = nlx_kcore_buf_msg_send,
        .ko_buf_active_recv = nlx_kcore_buf_active_recv,
        .ko_buf_active_send = nlx_kcore_buf_active_send,
        .ko_buf_passive_recv = nlx_kcore_buf_passive_recv,
        .ko_buf_passive_send = nlx_kcore_buf_passive_send,
        .ko_buf_del = nlx_kcore_LNetMDUnlink,
        .ko_buf_event_wait = nlx_kcore_buf_event_wait,
};

static int nlx_kcore_kcore_dom_init(struct nlx_kcore_domain *kd)
{
        M0_PRE(kd != NULL);
        kd->kd_magic = M0_NET_LNET_KCORE_DOM_MAGIC;
        nlx_core_kmem_loc_set(&kd->kd_cd_loc, NULL, 0);
        m0_mutex_init(&kd->kd_drv_mutex);
        kd->kd_drv_ops = &nlx_kcore_def_ops;
        drv_tms_tlist_init(&kd->kd_drv_tms);
        drv_bufs_tlist_init(&kd->kd_drv_bufs);
        M0_POST(nlx_kcore_domain_invariant(kd));
        return 0;
}

static void nlx_kcore_kcore_dom_fini(struct nlx_kcore_domain *kd)
{
        M0_PRE(nlx_kcore_domain_invariant(kd));
        M0_PRE(nlx_core_kmem_loc_is_empty(&kd->kd_cd_loc));

        drv_bufs_tlist_fini(&kd->kd_drv_bufs);
        drv_tms_tlist_fini(&kd->kd_drv_tms);
        kd->kd_drv_ops = NULL;
        m0_mutex_fini(&kd->kd_drv_mutex);
}
 /* KLNetCore */

/*
 *  Local variables:
 *  c-indentation-style: "K&R"
 *  c-basic-offset: 8
 *  tab-width: 8
 *  fill-column: 79
 *  scroll-step: 1
 *  End:
 */