ulnet_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.
 *
 */


#include <sys/types.h>
#include <sys/stat.h>
#include <sys/ioctl.h>
#include <fcntl.h>
#include <unistd.h> /* close */
#include "lib/errno.h"  /* EADDRINUSE */

static const char *nlx_ucore_dev_name = "/dev/" M0_LNET_DEV;

static bool nlx_ucore_domain_invariant(const struct nlx_ucore_domain *ud)
{
        return ud != NULL && ud->ud_magic == M0_NET_LNET_UCORE_DOM_MAGIC &&
            ud->ud_fd >= 0 &&
            ud->ud_max_buffer_size > 0 &&
            ud->ud_max_buffer_segment_size > 0 &&
            ud->ud_max_buffer_segments > 0;
}

static bool nlx_ucore_buffer_invariant(const struct nlx_ucore_buffer *ub)
{
        return ub != NULL && ub->ub_magic == M0_NET_LNET_UCORE_BUF_MAGIC;
}

static bool nlx_ucore_tm_invariant(const struct nlx_ucore_transfer_mc *utm)
{
        return utm != NULL && utm->utm_magic == M0_NET_LNET_UCORE_TM_MAGIC;
}

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

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

static int nlx_ucore_ioctl(int fd, unsigned long cmd, void *arg)
{
        int rc;

        M0_PRE(fd >= 0);
        M0_PRE(_IOC_TYPE(cmd) == M0_LNET_IOC_MAGIC);
        M0_PRE(_IOC_NR(cmd)   >= M0_LNET_IOC_MIN_NR);
        M0_PRE(_IOC_NR(cmd)   <= M0_LNET_IOC_MAX_NR);

        rc = ioctl(fd, cmd, arg);
        if (rc >= 0)
                return M0_RC(rc);
        M0_ASSERT_INFO(errno > 0, "errno=%d", errno);
        M0_ASSERT_INFO(_0C(errno != EFAULT) && _0C(errno != EBADR),
                       "errno=%d", errno);
        return M0_RC(-errno);
}

static unsigned nlx_ucore_nidstrs_thunk = 128;

static int nlx_ucore_nidstrs_get(struct nlx_ucore_domain *ud, char ***nidary)
{
        struct m0_lnet_dev_nidstrs_get_params dngp;
        unsigned i;
        char *p;
        char **nidstrs;
        unsigned nidstrs_nr;
        int rc;

        M0_PRE(ud != NULL && ud->ud_fd >= 0);

        /* Repeat until the buffer is large enough to hold all the strings. */
        for (i = 0, dngp.dng_buf = NULL; dngp.dng_buf == NULL; ++i) {
                dngp.dng_size = ++i * nlx_ucore_nidstrs_thunk;
                NLX_ALLOC_ARR(dngp.dng_buf, dngp.dng_size);
                if (dngp.dng_buf == NULL)
                        return M0_ERR(-ENOMEM);
                rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_NIDSTRS_GET, &dngp);
                if (rc < 0) {
                        m0_free0(&dngp.dng_buf);
                        if (rc != -EFBIG)
                                return M0_RC(rc);
                }
        }
        nidstrs_nr = rc;

        /* Create a string array. */
        NLX_ALLOC_ARR(nidstrs, nidstrs_nr + 1);
        if (nidstrs == NULL) {
                m0_free(dngp.dng_buf);
                return M0_ERR(-ENOMEM);
        }
        for (i = 0, p = dngp.dng_buf; i < nidstrs_nr; ++i, ++p) {
                nidstrs[i] = p;
                while (*p != '\0')
                        ++p;
        }
        nidstrs[nidstrs_nr] = NULL;

        *nidary = nidstrs;
        return 0;
}

static void nlx_ucore_nidstrs_put(struct nlx_ucore_domain *ud, char ***nidary)
{
        M0_PRE(nidary != NULL);
        m0_free((*nidary)[0]); /* string buffer */
        m0_free0(nidary);      /* string array */
}


M0_INTERNAL int nlx_core_dom_init(struct m0_net_domain *dom,
                                  struct nlx_core_domain *cd)
{
        static const struct {
                int         err;
                const char *err_str;
                const char *err_help;
        } possible_open_errors[] = {
                { EACCES, "EACCES", "Please check permissions." },
                { ENOENT, "ENOENT", "Please ensure that m0tr.ko is loaded." },
        };
        struct nlx_ucore_domain *ud;
        struct m0_lnet_dev_dom_init_params ip = {
                .ddi_cd = cd,
        };
        int rc;
        int i;

        M0_PRE(dom != NULL && cd != NULL);
        M0_PRE(cd->cd_kpvt == NULL && cd->cd_upvt == NULL);
        NLX_ALLOC_PTR(ud);
        if (ud == NULL)
                return M0_ERR(-ENOMEM);
        ud->ud_fd = open(nlx_ucore_dev_name, O_RDWR|O_CLOEXEC);
        if (ud->ud_fd == -1) {
                M0_ASSERT(errno != 0);
                rc = -errno;
                M0_ASSERT(rc < 0);
                for (i = 0; i < ARRAY_SIZE(possible_open_errors); ++i) {
                        if (possible_open_errors[i].err == -rc) {
                                M0_LOG(M0_ERROR,
                                       "open(\"%s\", O_RDWR|O_CLOEXEC) failed: "
                                       "errno=%d (%s). %s" ,
                                       nlx_ucore_dev_name, -rc,
                                       possible_open_errors[i].err_str,
                                       possible_open_errors[i].err_help);
                                goto fail_open;
                        }
                }
                M0_LOG(M0_ERROR, "open(\"%s\", O_RDWR|O_CLOEXEC) failed: "
                       "errno=%d", nlx_ucore_dev_name, -rc);
                goto fail_open;
        }

        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_DOM_INIT, &ip);
        if (rc < 0)
                goto fail_dom_init;
        M0_ASSERT(cd->cd_kpvt != NULL);

        /* cache buffer size constants */
#define NLX_IP_SET(f) ud->ud_##f = ip.ddi_##f
        NLX_IP_SET(max_buffer_size);
        NLX_IP_SET(max_buffer_segment_size);
        NLX_IP_SET(max_buffer_segments);
#undef NLX_IP_SET

        ud->ud_magic = M0_NET_LNET_UCORE_DOM_MAGIC;
        cd->cd_upvt = ud;

        M0_POST(nlx_ucore_domain_invariant(ud));
        return 0;

 fail_dom_init:
        close(ud->ud_fd);
 fail_open:
        m0_free(ud);
        M0_POST(cd->cd_kpvt == NULL && cd->cd_upvt == NULL);
        return M0_RC(rc);
}

M0_INTERNAL void nlx_core_dom_fini(struct nlx_core_domain *cd)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        close(ud->ud_fd);
        ud->ud_magic = 0;
        m0_free0(&cd->cd_upvt);
        cd->cd_kpvt = NULL;
}

M0_INTERNAL m0_bcount_t nlx_core_get_max_buffer_size(struct nlx_core_domain *cd)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        return ud->ud_max_buffer_size;
}

M0_INTERNAL m0_bcount_t nlx_core_get_max_buffer_segment_size(struct
                                                             nlx_core_domain
                                                             *cd)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        return ud->ud_max_buffer_segment_size;
}

M0_INTERNAL int32_t nlx_core_get_max_buffer_segments(struct nlx_core_domain *cd)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        return ud->ud_max_buffer_segments;
}

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_ucore_buffer *ub;
        struct nlx_ucore_domain *ud;
        struct m0_lnet_dev_buf_register_params rp = {
                .dbr_lcbuf     = cb,
                .dbr_buffer_id = buffer_id,
        };

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        M0_PRE(buffer_id != 0);
        M0_PRE(bvec != NULL);
        M0_PRE(cb != NULL);
        M0_PRE(cb->cb_kpvt == NULL && cb->cb_upvt == NULL);

        NLX_ALLOC_PTR(ub);
        if (ub == NULL)
                return M0_ERR(-ENOMEM);
        ub->ub_magic = M0_NET_LNET_UCORE_BUF_MAGIC;
        M0_POST(nlx_ucore_buffer_invariant(ub));
        cb->cb_upvt = ub;

        rp.dbr_bvec = *bvec;
        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_BUF_REGISTER, &rp);
        if (rc < 0) {
                cb->cb_upvt = NULL;
                ub->ub_magic = 0;
                m0_free(ub);
                return M0_RC(rc);
        }
        M0_ASSERT(cb->cb_kpvt != NULL);
        M0_ASSERT(cb->cb_upvt == ub);

        M0_POST(nlx_core_buffer_invariant(cb));
        return 0;
}

M0_INTERNAL void nlx_core_buf_deregister(struct nlx_core_domain *cd,
                                         struct nlx_core_buffer *cb)
{
        struct nlx_ucore_buffer *ub;
        struct nlx_ucore_domain *ud;
        struct m0_lnet_dev_buf_deregister_params dp;
        int rc;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        M0_PRE(nlx_core_buffer_invariant(cb));
        ub = cb->cb_upvt;
        M0_PRE(nlx_ucore_buffer_invariant(ub));
        M0_PRE(cb->cb_kpvt != NULL);
        dp.dbd_kb = cb->cb_kpvt;
        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_BUF_DEREGISTER, &dp);
        M0_ASSERT(rc == 0);

        ub->ub_magic = 0;
        m0_free(ub);

        cb->cb_kpvt = NULL;
        cb->cb_upvt = NULL;
        return;
}

#define NLX_UCORE_BUF_OP(op, loc, ...)                          \
        struct nlx_ucore_domain *ud;                            \
        struct nlx_ucore_transfer_mc *utm;                      \
        struct nlx_ucore_buffer *ub;                            \
        struct m0_lnet_dev_buf_queue_params dbqp;               \
        int rc = 0;                                             \
                                                                \
        M0_PRE(cd != NULL);                                     \
        ud = cd->cd_upvt;                                       \
        M0_PRE(nlx_ucore_domain_invariant(ud));                 \
                                                                \
        M0_PRE(nlx_core_tm_invariant(ctm));                     \
        utm = ctm->ctm_upvt;                                    \
        M0_PRE(nlx_ucore_tm_invariant(utm));                    \
                                                                \
        M0_PRE(nlx_core_buffer_invariant(cb));                  \
        ub = cb->cb_upvt;                                       \
        M0_PRE(nlx_ucore_buffer_invariant(ub));                 \
                                                                \
        __VA_ARGS__;                                            \
                                                                \
        dbqp.dbq_ktm = ctm->ctm_kpvt;                           \
        dbqp.dbq_kb  = cb->cb_kpvt;                             \
        rc = nlx_ucore_ioctl(ud->ud_fd, op, &dbqp)

M0_INTERNAL int nlx_core_buf_msg_recv(struct nlx_core_domain *cd,
                                      struct nlx_core_transfer_mc *ctm,
                                      struct nlx_core_buffer *cb)
{
        NLX_UCORE_BUF_OP(M0_LNET_BUF_MSG_RECV, U_BUF_MRECV,
                         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);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_msg_send(struct nlx_core_domain *cd,
                                      struct nlx_core_transfer_mc *ctm,
                                      struct nlx_core_buffer *cb)
{
        NLX_UCORE_BUF_OP(M0_LNET_BUF_MSG_SEND, U_BUF_MSEND,
                         M0_PRE(cb->cb_qtype == M0_NET_QT_MSG_SEND);
                         M0_PRE(cb->cb_length > 0);
                         M0_PRE(cb->cb_max_operations == 1);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_active_recv(struct nlx_core_domain *cd,
                                         struct nlx_core_transfer_mc *ctm,
                                         struct nlx_core_buffer *cb)
{
        uint32_t tmid;
        uint64_t counter;
        NLX_UCORE_BUF_OP(M0_LNET_BUF_ACTIVE_RECV, U_BUF_ARECV,
                         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);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_active_send(struct nlx_core_domain *cd,
                                         struct nlx_core_transfer_mc *ctm,
                                         struct nlx_core_buffer *cb)
{
        uint32_t tmid;
        uint64_t counter;
        NLX_UCORE_BUF_OP(M0_LNET_BUF_ACTIVE_SEND, U_BUF_ASEND,
                         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);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_passive_recv(struct nlx_core_domain *cd,
                                          struct nlx_core_transfer_mc *ctm,
                                          struct nlx_core_buffer *cb)
{
        uint32_t tmid;
        uint64_t counter;
        NLX_UCORE_BUF_OP(M0_LNET_BUF_PASSIVE_RECV, U_BUF_PRECV,
                         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 == ctm->ctm_addr.cepa_tmid);
                         M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
                         M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_passive_send(struct nlx_core_domain *cd,
                                          struct nlx_core_transfer_mc *ctm,
                                          struct nlx_core_buffer *cb)
{
        uint32_t tmid;
        uint64_t counter;
        NLX_UCORE_BUF_OP(M0_LNET_BUF_PASSIVE_SEND, U_BUF_PSEND,
                         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 == ctm->ctm_addr.cepa_tmid);
                         M0_PRE(counter >= M0_NET_LNET_BUFFER_ID_MIN);
                         M0_PRE(counter <= M0_NET_LNET_BUFFER_ID_MAX);
                         );
        return M0_RC(rc);
}

M0_INTERNAL int nlx_core_buf_del(struct nlx_core_domain *cd,
                                 struct nlx_core_transfer_mc *ctm,
                                 struct nlx_core_buffer *cb)
{
        NLX_UCORE_BUF_OP(M0_LNET_BUF_DEL, U_BUF_DEL, );
        return M0_RC(rc);
}

#undef NLX_UCORE_BUF_OP

int nlx_core_buf_event_wait(struct nlx_core_domain *cd,
                            struct nlx_core_transfer_mc *ctm,
                            m0_time_t timeout)
{
        struct nlx_ucore_domain *ud;
        struct nlx_ucore_transfer_mc *utm;
        struct m0_lnet_dev_buf_event_wait_params bewp;
        int rc;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        M0_PRE(nlx_core_tm_invariant(ctm));
        M0_PRE(ctm->ctm_kpvt != NULL);
        utm = ctm->ctm_upvt;
        M0_PRE(nlx_ucore_tm_invariant(utm));

        bewp.dbw_ktm = ctm->ctm_kpvt;
        bewp.dbw_timeout = m0_time_to_realtime(timeout);
        do {
                rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_BUF_EVENT_WAIT, &bewp);
        } while (rc == -EINTR);
        return rc < 0 ? rc : 0;
}

M0_INTERNAL int nlx_core_nidstr_decode(struct nlx_core_domain *cd,
                                       const char *nidstr, uint64_t * nid)
{
        struct m0_lnet_dev_nid_encdec_params dnep;
        struct nlx_ucore_domain *ud;
        int rc;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        strncpy(dnep.dn_buf, nidstr, ARRAY_SIZE(dnep.dn_buf) - 1);
        dnep.dn_buf[ARRAY_SIZE(dnep.dn_buf) - 1] = '\0';
        dnep.dn_nid = 0;

        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_NIDSTR_DECODE, &dnep);
        if (rc < 0)
                return M0_RC(rc);

        *nid = dnep.dn_nid;
        return 0;
}

M0_INTERNAL int nlx_core_nidstr_encode(struct nlx_core_domain *cd,
                                       uint64_t nid,
                                       char nidstr[M0_NET_LNET_NIDSTR_SIZE])
{
        struct m0_lnet_dev_nid_encdec_params dnep;
        struct nlx_ucore_domain *ud;
        int rc;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        dnep.dn_nid = nid;
        M0_SET_ARR0(dnep.dn_buf);

        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_NIDSTR_ENCODE, &dnep);
        if (rc < 0)
                return M0_ERR(rc);
        M0_POST(dnep.dn_buf[0] != '\0');

        strncpy(nidstr, dnep.dn_buf, M0_NET_LNET_NIDSTR_SIZE);
        nidstr[M0_NET_LNET_NIDSTR_SIZE - 1] = '\0';
        return 0;
}

/*
   This call is not protected by any mutex. It relies on the
   application to not invoke it during domain finalization.
 */
M0_INTERNAL int nlx_core_nidstrs_get(struct nlx_core_domain   *cd,
                                     char                   ***nidary)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        return nlx_ucore_nidstrs_get(ud, nidary);
}

M0_INTERNAL void nlx_core_nidstrs_put(struct nlx_core_domain   *cd,
                                      char                   ***nidary)
{
        struct nlx_ucore_domain *ud;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        nlx_ucore_nidstrs_put(ud, nidary);
}

static void nlx_ucore_tm_stop(struct nlx_core_domain *cd,
                              struct nlx_core_transfer_mc *ctm)
{
        struct nlx_ucore_domain *ud;
        struct m0_lnet_dev_tm_stop_params tpp;
        int rc;

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        M0_PRE(nlx_core_tm_invariant(ctm));
        M0_PRE(ctm->ctm_kpvt != NULL);
        tpp.dts_ktm = ctm->ctm_kpvt;

        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_TM_STOP, &tpp);
        M0_ASSERT(rc == 0);
        return;
}

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_ucore_domain *ud;
        struct nlx_ucore_transfer_mc *utm;
        struct nlx_core_buffer_event *e1 = NULL;
        struct nlx_core_buffer_event *e2 = NULL;
        struct m0_lnet_dev_tm_start_params tsp = {
                .dts_ctm = ctm,
        };
        int rc;

        M0_PRE(tm != NULL);
        M0_PRE(m0_mutex_is_locked(&tm->ntm_mutex));
        M0_PRE(nlx_tm_invariant(tm));

        M0_PRE(cd != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));

        M0_PRE(ctm != NULL);
        M0_PRE(ctm->ctm_kpvt == NULL);
        M0_PRE(ctm->ctm_magic == 0);

        NLX_ALLOC_PTR(utm);
        if (utm == NULL) {
                rc = M0_ERR(-ENOMEM);
                goto fail_utm;
        }
        utm->utm_magic = M0_NET_LNET_UCORE_TM_MAGIC;
        M0_POST(nlx_ucore_tm_invariant(utm));
        ctm->ctm_upvt = utm;

        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_TM_START, &tsp);
        if (rc < 0)
                goto fail_start;
        M0_ASSERT(ctm->ctm_kpvt != NULL);
        M0_ASSERT(ctm->ctm_upvt == utm);

        rc = nlx_core_new_blessed_bev(cd, ctm, &e1) ?:
             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));

        M0_POST(nlx_ucore_tm_invariant(utm));
        M0_POST(ctm->ctm_kpvt != NULL);
        M0_POST(ctm->ctm_upvt == utm);
        M0_POST(nlx_core_tm_invariant(ctm));
        return 0;

 fail_blessed_bev:
        M0_ASSERT(e2 == NULL);
        if (e1 != NULL)
                nlx_core_bev_free_cb(&e1->cbe_tm_link);
        nlx_ucore_tm_stop(cd, ctm);
 fail_start:
        ctm->ctm_upvt = NULL;
        utm->utm_magic = 0;
        m0_free(utm);
 fail_utm:
        if (rc == -EADDRINUSE)
                return M0_ERR_INFO(rc, "The address 0x%" PRIx64 ":%"PRIu32
                                   ":%"PRIu32":%"PRIu32" is already in use "
                                   "by another process.",
                                   ctm->ctm_addr.cepa_nid,
                                   ctm->ctm_addr.cepa_pid,
                                   ctm->ctm_addr.cepa_portal,
                                   ctm->ctm_addr.cepa_tmid);
        return M0_ERR(rc);
}

M0_INTERNAL void nlx_core_tm_stop(struct nlx_core_domain *cd,
                                  struct nlx_core_transfer_mc *ctm)
{
        struct nlx_ucore_transfer_mc *utm;

        utm = ctm->ctm_upvt;
        M0_PRE(nlx_ucore_tm_invariant(utm));

        nlx_ucore_tm_stop(cd, ctm); /* other invariants checked */
        bev_cqueue_fini(&ctm->ctm_bevq, nlx_core_bev_free_cb);

        ctm->ctm_upvt = NULL;
        utm->utm_magic = 0;
        m0_free(utm);
        return;
}

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 m0_lnet_dev_bev_bless_params bp;
        struct nlx_ucore_domain *ud;
        struct nlx_ucore_transfer_mc *utm;
        int rc;

        M0_PRE(cd != NULL);
        M0_PRE(nlx_core_tm_invariant(ctm));
        M0_PRE(ctm->ctm_kpvt != NULL);
        ud = cd->cd_upvt;
        M0_PRE(nlx_ucore_domain_invariant(ud));
        utm = ctm->ctm_upvt;
        M0_PRE(nlx_ucore_tm_invariant(utm));

        NLX_ALLOC_ALIGNED_PTR(bev);
        if (bev == NULL) {
                *bevp = NULL;
                return M0_ERR(-ENOMEM);
        }

        bp.dbb_ktm = ctm->ctm_kpvt;
        bp.dbb_bev = bev;
        rc = nlx_ucore_ioctl(ud->ud_fd, M0_LNET_BEV_BLESS, &bp);
        if (rc < 0) {
                NLX_FREE_ALIGNED_PTR(bev);
                return M0_ERR(rc);
        }
        M0_ASSERT(bev->cbe_kpvt != NULL);

        *bevp = bev;
        return 0;
}

static void nlx_core_fini(void)
{
        return;
}

static int nlx_core_init(void)
{
        return 0;
}
 /* ULNetCore */

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