/*
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
 *
 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 * CA 95054 USA or visit www.sun.com if you need additional information or
 * have any questions.
 *
 * GPL HEADER END
 */
/*
 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
 * Use is subject to license terms.
 *
 * Copyright (c) 2012, Intel Corporation.
 *
 * Copyright (C) 2006 Myricom, Inc.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lnet/klnds/mxlnd/mxlnd.c
 *
 * Author: Eric Barton <eric@bartonsoftware.com>
 * Author: Scott Atchley <atchley at myri.com>
 */

#include "mxlnd.h"

mx_endpoint_addr_t MX_EPA_NULL; /* use to determine if an endpoint is NULL */

inline int
mxlnd_endpoint_addr_null(mx_endpoint_addr_t epa)
{
        /* if memcmp() == 0, it is NULL */
        return !(memcmp(&epa, &MX_EPA_NULL, sizeof(epa)));
}

char *
mxlnd_ctxstate_to_str(int mxc_state)
{
        switch (mxc_state) {
        case MXLND_CTX_INIT:
                return "MXLND_CTX_INIT";
        case MXLND_CTX_IDLE:
                return "MXLND_CTX_IDLE";
        case MXLND_CTX_PREP:
                return "MXLND_CTX_PREP";
        case MXLND_CTX_PENDING:
                return "MXLND_CTX_PENDING";
        case MXLND_CTX_COMPLETED:
                return "MXLND_CTX_COMPLETED";
        case MXLND_CTX_CANCELED:
                return "MXLND_CTX_CANCELED";
        default:
                return "*unknown*";
        }
}

char *
mxlnd_connstatus_to_str(int mxk_status)
{
        switch (mxk_status) {
        case MXLND_CONN_READY:
                return "MXLND_CONN_READY";
        case MXLND_CONN_INIT:
                return "MXLND_CONN_INIT";
        case MXLND_CONN_WAIT:
                return "MXLND_CONN_WAIT";
        case MXLND_CONN_DISCONNECT:
                return "MXLND_CONN_DISCONNECT";
        case MXLND_CONN_FAIL:
                return "MXLND_CONN_FAIL";
        default:
                return "unknown";
        }
}

char *
mxlnd_msgtype_to_str(int type) {
        switch (type) {
        case MXLND_MSG_EAGER:
                return "MXLND_MSG_EAGER";
        case MXLND_MSG_CONN_REQ:
                return "MXLND_MSG_CONN_REQ";
        case MXLND_MSG_CONN_ACK:
                return "MXLND_MSG_CONN_ACK";
        case MXLND_MSG_BYE:
                return "MXLND_MSG_BYE";
        case MXLND_MSG_NOOP:
                return "MXLND_MSG_NOOP";
        case MXLND_MSG_PUT_REQ:
                return "MXLND_MSG_PUT_REQ";
        case MXLND_MSG_PUT_ACK:
                return "MXLND_MSG_PUT_ACK";
        case MXLND_MSG_PUT_DATA:
                return "MXLND_MSG_PUT_DATA";
        case MXLND_MSG_GET_REQ:
                return "MXLND_MSG_GET_REQ";
        case MXLND_MSG_GET_DATA:
                return "MXLND_MSG_GET_DATA";
        default:
                return "unknown";
        }
}

char *
mxlnd_lnetmsg_to_str(int type)
{
        switch (type) {
        case LNET_MSG_ACK:
                return "LNET_MSG_ACK";
        case LNET_MSG_PUT:
                return "LNET_MSG_PUT";
        case LNET_MSG_GET:
                return "LNET_MSG_GET";
        case LNET_MSG_REPLY:
                return "LNET_MSG_REPLY";
        case LNET_MSG_HELLO:
                return "LNET_MSG_HELLO";
        default:
                LBUG();
                return "*unknown*";
        }
}

static inline u64
mxlnd_create_match(kmx_ctx_t *ctx, u8 error)
{
        u64 type        = (u64) ctx->mxc_msg_type;
        u64 err         = (u64) error;
        u64 match       = 0ULL;

        mxlnd_valid_msg_type(ctx->mxc_msg_type);
        LASSERT(ctx->mxc_cookie >> MXLND_ERROR_OFFSET == 0);
        match = (type << MXLND_MSG_OFFSET) | (err << MXLND_ERROR_OFFSET) | ctx->mxc_cookie;
        return match;
}

static inline void
mxlnd_parse_match(u64 match, u8 *msg_type, u8 *error, u64 *cookie)
{
        *msg_type = (u8) MXLND_MSG_TYPE(match);
        *error    = (u8) MXLND_ERROR_VAL(match);
        *cookie   = match & MXLND_MAX_COOKIE;
        mxlnd_valid_msg_type(*msg_type);
        return;
}

kmx_ctx_t *
mxlnd_get_idle_rx(kmx_conn_t *conn)
{
        cfs_list_t              *rxs    = NULL;
        kmx_ctx_t               *rx     = NULL;

        LASSERT(conn != NULL);

        rxs = &conn->mxk_rx_idle;

	spin_lock(&conn->mxk_lock);

	if (cfs_list_empty (rxs)) {
		spin_unlock(&conn->mxk_lock);
		return NULL;
	}

	rx = cfs_list_entry (rxs->next, kmx_ctx_t, mxc_list);
	cfs_list_del_init(&rx->mxc_list);
	spin_unlock(&conn->mxk_lock);

#if MXLND_DEBUG
        if (rx->mxc_get != rx->mxc_put) {
                CNETERR("*** RX get (%llu) != put (%llu) ***\n", rx->mxc_get, rx->mxc_put);
                CNETERR("*** incarnation= %lld ***\n", rx->mxc_incarnation);
                CNETERR("*** deadline= %ld ***\n", rx->mxc_deadline);
                CNETERR("*** state= %s ***\n", mxlnd_ctxstate_to_str(rx->mxc_state));
                CNETERR("*** listed?= %d ***\n", !cfs_list_empty(&rx->mxc_list));
                CNETERR("*** nid= 0x%llx ***\n", rx->mxc_nid);
                CNETERR("*** peer= 0x%p ***\n", rx->mxc_peer);
                CNETERR("*** msg_type= %s ***\n", mxlnd_msgtype_to_str(rx->mxc_msg_type));
                CNETERR("*** cookie= 0x%llx ***\n", rx->mxc_cookie);
                CNETERR("*** nob= %d ***\n", rx->mxc_nob);
        }
#endif
        LASSERT (rx->mxc_get == rx->mxc_put);

        rx->mxc_get++;

        LASSERT (rx->mxc_state == MXLND_CTX_IDLE);
        rx->mxc_state = MXLND_CTX_PREP;
        rx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;

        return rx;
}

int
mxlnd_put_idle_rx(kmx_ctx_t *rx)
{
        kmx_conn_t              *conn   = rx->mxc_conn;
        cfs_list_t              *rxs    = &conn->mxk_rx_idle;

        LASSERT(rx->mxc_type == MXLND_REQ_RX);

        mxlnd_ctx_init(rx);

        rx->mxc_put++;
        LASSERT(rx->mxc_get == rx->mxc_put);

	spin_lock(&conn->mxk_lock);
	cfs_list_add(&rx->mxc_list, rxs);
	spin_unlock(&conn->mxk_lock);
	return 0;
}

kmx_ctx_t *
mxlnd_get_idle_tx(void)
{
	cfs_list_t		*tmp    = &kmxlnd_data.kmx_tx_idle;
	kmx_ctx_t		*tx     = NULL;

	spin_lock(&kmxlnd_data.kmx_tx_idle_lock);

	if (cfs_list_empty (&kmxlnd_data.kmx_tx_idle)) {
		CNETERR("%d txs in use\n", kmxlnd_data.kmx_tx_used);
		spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);
		return NULL;
	}

        tmp = &kmxlnd_data.kmx_tx_idle;
        tx = cfs_list_entry (tmp->next, kmx_ctx_t, mxc_list);
        cfs_list_del_init(&tx->mxc_list);

        /* Allocate a new completion cookie.  It might not be needed,
         * but we've got a lock right now and we're unlikely to
         * wrap... */
        tx->mxc_cookie = kmxlnd_data.kmx_tx_next_cookie++;
        if (kmxlnd_data.kmx_tx_next_cookie > MXLND_MAX_COOKIE) {
                kmxlnd_data.kmx_tx_next_cookie = 1;
        }
        kmxlnd_data.kmx_tx_used++;
	spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);

        LASSERT (tx->mxc_get == tx->mxc_put);

        tx->mxc_get++;

        LASSERT (tx->mxc_state == MXLND_CTX_IDLE);
        LASSERT (tx->mxc_lntmsg[0] == NULL);
        LASSERT (tx->mxc_lntmsg[1] == NULL);

        tx->mxc_state = MXLND_CTX_PREP;
        tx->mxc_deadline = jiffies + MXLND_COMM_TIMEOUT;

        return tx;
}

void
mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye);

int
mxlnd_put_idle_tx(kmx_ctx_t *tx)
{
        int             result  = 0;
        lnet_msg_t      *lntmsg[2];

        LASSERT(tx->mxc_type == MXLND_REQ_TX);

        if (tx->mxc_status.code != MX_STATUS_SUCCESS || tx->mxc_errno != 0) {
                kmx_conn_t      *conn   = tx->mxc_conn;

                result = -EIO;
                if (tx->mxc_errno != 0) result = tx->mxc_errno;
                /* FIXME should we set mx_dis? */
                mxlnd_conn_disconnect(conn, 0, 1);
        }

        lntmsg[0] = tx->mxc_lntmsg[0];
        lntmsg[1] = tx->mxc_lntmsg[1];

        mxlnd_ctx_init(tx);

        tx->mxc_put++;
        LASSERT(tx->mxc_get == tx->mxc_put);

	spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
	cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_idle);
	kmxlnd_data.kmx_tx_used--;
	spin_unlock(&kmxlnd_data.kmx_tx_idle_lock);

	if (lntmsg[0] != NULL)
		lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[0], result);
	if (lntmsg[1] != NULL)
		lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[1], result);
	return 0;
}


void
mxlnd_connparams_free(kmx_connparams_t *cp)
{
        LASSERT(cfs_list_empty(&cp->mxr_list));
        MXLND_FREE(cp, sizeof(*cp));
        return;
}

int
mxlnd_connparams_alloc(kmx_connparams_t **cp, void *context,
                            mx_endpoint_addr_t epa, u64 match, u32 length,
                            kmx_conn_t *conn, kmx_peer_t *peer, void *data)
{
        kmx_connparams_t *c = NULL;

        MXLND_ALLOC(c, sizeof(*c));
        if (!c) return -ENOMEM;

        CFS_INIT_LIST_HEAD(&c->mxr_list);
        c->mxr_context = context;
        c->mxr_epa = epa;
        c->mxr_match = match;
        c->mxr_nob = length;
        c->mxr_conn = conn;
        c->mxr_peer = peer;
        c->mxr_msg = *((kmx_msg_t *) data);

        *cp = c;
        return 0;
}

static inline void
mxlnd_set_conn_status(kmx_conn_t *conn, int status)
{
	conn->mxk_status = status;
	smp_mb();
}

/**
 * mxlnd_conn_free_locked - free the conn
 * @conn - a kmx_conn pointer
 *
 * The calling function should remove the conn from the conns list first
 * then destroy it. Caller should have write-locked kmx_global_lock.
 */
void
mxlnd_conn_free_locked(kmx_conn_t *conn)
{
        int             valid   = !mxlnd_endpoint_addr_null(conn->mxk_epa);
        kmx_peer_t      *peer   = conn->mxk_peer;

        CDEBUG(D_NET, "freeing conn 0x%p *****\n", conn);
        LASSERT (cfs_list_empty (&conn->mxk_tx_credit_queue) &&
                 cfs_list_empty (&conn->mxk_tx_free_queue) &&
                 cfs_list_empty (&conn->mxk_pending));
        if (!cfs_list_empty(&conn->mxk_list)) {
                cfs_list_del_init(&conn->mxk_list);
                if (peer->mxp_conn == conn) {
                        peer->mxp_conn = NULL;
                        if (valid) {
                                kmx_conn_t      *temp   = NULL;

                                mx_get_endpoint_addr_context(conn->mxk_epa,
                                                             (void **) &temp);
                                if (conn == temp) {
                                        mx_set_endpoint_addr_context(conn->mxk_epa,
                                                                     (void *) NULL);
                                }
                        }
                        /* unlink from global list and drop its ref */
                        cfs_list_del_init(&peer->mxp_list);
                        mxlnd_peer_decref(peer);
                }
        }
        mxlnd_peer_decref(peer); /* drop conn's ref to peer */
        if (conn->mxk_rx_pages) {
                LASSERT (conn->mxk_rxs != NULL);
                mxlnd_free_pages(conn->mxk_rx_pages);
        }
        if (conn->mxk_rxs) {
                int             i       = 0;
                kmx_ctx_t       *rx     = NULL;

                for (i = 0; i < MXLND_RX_MSGS(); i++) {
                        rx = &conn->mxk_rxs[i];
                        if (rx->mxc_seg_list != NULL) {
                                LASSERT(rx->mxc_nseg > 0);
                                MXLND_FREE(rx->mxc_seg_list,
                                           rx->mxc_nseg *
                                           sizeof(*rx->mxc_seg_list));
                        }
                }
                MXLND_FREE(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
        }

        MXLND_FREE(conn, sizeof (*conn));
        return;
}


int
mxlnd_conn_cancel_pending_rxs(kmx_conn_t *conn)
{
        int             found   = 0;
        int             count   = 0;
        kmx_ctx_t       *ctx    = NULL;
        kmx_ctx_t       *next   = NULL;
        mx_return_t     mxret   = MX_SUCCESS;
        u32             result  = 0;

        do {
                found = 0;
		spin_lock(&conn->mxk_lock);
                cfs_list_for_each_entry_safe(ctx, next, &conn->mxk_pending,
                                             mxc_list) {
                        cfs_list_del_init(&ctx->mxc_list);
                        if (ctx->mxc_type == MXLND_REQ_RX) {
                                found = 1;
                                mxret = mx_cancel(kmxlnd_data.kmx_endpt,
                                                  &ctx->mxc_mxreq,
                                                  &result);
                                if (mxret != MX_SUCCESS) {
                                        CNETERR("mx_cancel() returned %s (%d)\n", mx_strerror(mxret), mxret);
                                }
                                if (result == 1) {
                                        ctx->mxc_errno = -ECONNABORTED;
                                        ctx->mxc_state = MXLND_CTX_CANCELED;
					spin_unlock(&conn->mxk_lock);
					spin_lock(&kmxlnd_data.kmx_conn_lock);
                                        /* we may be holding the global lock,
                                         * move to orphan list so that it can free it */
                                        cfs_list_add_tail(&ctx->mxc_list,
                                                          &kmxlnd_data.kmx_orphan_msgs);
                                        count++;
					spin_unlock(&kmxlnd_data.kmx_conn_lock);
					spin_lock(&conn->mxk_lock);
				}
				break;
			}
		}
		spin_unlock(&conn->mxk_lock);
	} while (found);

	return count;
}

int
mxlnd_cancel_queued_txs(kmx_conn_t *conn)
{
	int		count   = 0;
	cfs_list_t	*tmp    = NULL;

	spin_lock(&conn->mxk_lock);
        while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
               !cfs_list_empty(&conn->mxk_tx_credit_queue)) {

                kmx_ctx_t       *tx     = NULL;

                if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
                        tmp = &conn->mxk_tx_free_queue;
                } else {
                        tmp = &conn->mxk_tx_credit_queue;
                }

                tx = cfs_list_entry(tmp->next, kmx_ctx_t, mxc_list);
                cfs_list_del_init(&tx->mxc_list);
		spin_unlock(&conn->mxk_lock);
		tx->mxc_errno = -ECONNABORTED;
		tx->mxc_state = MXLND_CTX_CANCELED;
		/* move to orphan list and then abort */
		spin_lock(&kmxlnd_data.kmx_conn_lock);
		cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_orphan_msgs);
		spin_unlock(&kmxlnd_data.kmx_conn_lock);
		count++;
		spin_lock(&conn->mxk_lock);
	}
	spin_unlock(&conn->mxk_lock);

	return count;
}

void
mxlnd_send_message(mx_endpoint_addr_t epa, u8 msg_type, int error, u64 cookie)
{
        u64 match = (((u64) msg_type) << MXLND_MSG_OFFSET) |
                    (((u64) error) << MXLND_ERROR_OFFSET) | cookie;

        mx_kisend(kmxlnd_data.kmx_endpt, NULL, 0, MX_PIN_PHYSICAL,
                  epa, match, NULL, NULL);
        return;
}

/**
 * mxlnd_conn_disconnect - shutdown a connection
 * @conn - a kmx_conn pointer
 * @mx_dis - call mx_disconnect()
 * @send_bye - send peer a BYE msg
 *
 * This function sets the status to DISCONNECT, completes queued
 * txs with failure, calls mx_disconnect, which will complete
 * pending txs and matched rxs with failure.
 */
void
mxlnd_conn_disconnect(kmx_conn_t *conn, int mx_dis, int send_bye)
{
	mx_endpoint_addr_t	epa	= conn->mxk_epa;
	int			valid	= !mxlnd_endpoint_addr_null(epa);
	int			count	= 0;

	spin_lock(&conn->mxk_lock);
	if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
		spin_unlock(&conn->mxk_lock);
		return;
	}
	mxlnd_set_conn_status(conn, MXLND_CONN_DISCONNECT);
	conn->mxk_timeout = 0;
	spin_unlock(&conn->mxk_lock);

	count = mxlnd_cancel_queued_txs(conn);
	count += mxlnd_conn_cancel_pending_rxs(conn);

	if (count) /* let connd call kmxlnd_abort_msgs() */
		up(&kmxlnd_data.kmx_conn_sem);

        if (send_bye && valid &&
            conn->mxk_peer->mxp_nid != kmxlnd_data.kmx_ni->ni_nid) {
                /* send a BYE to the peer */
                CDEBUG(D_NET, "%s: sending a BYE msg to %s\n", __func__,
                                libcfs_nid2str(conn->mxk_peer->mxp_nid));
                mxlnd_send_message(epa, MXLND_MSG_BYE, 0, 0);
                /* wait to allow the peer to ack our message */
                mxlnd_sleep(msecs_to_jiffies(20));
        }

	if (atomic_read(&kmxlnd_data.kmx_shutdown) != 1) {
		unsigned long   last_msg        = 0;

                /* notify LNET that we are giving up on this peer */
                if (cfs_time_after(conn->mxk_last_rx, conn->mxk_last_tx))
                        last_msg = conn->mxk_last_rx;
                else
                        last_msg = conn->mxk_last_tx;

                lnet_notify(kmxlnd_data.kmx_ni, conn->mxk_peer->mxp_nid, 0, last_msg);

                if (mx_dis && valid &&
                    (memcmp(&epa, &kmxlnd_data.kmx_epa, sizeof(epa) != 0)))
                        mx_disconnect(kmxlnd_data.kmx_endpt, epa);
        }
        mxlnd_conn_decref(conn); /* drop the owning peer's reference */

        return;
}

/**
 * mxlnd_conn_alloc - allocate and initialize a new conn struct
 * @connp - address of a kmx_conn pointer
 * @peer - owning kmx_peer
 *
 * Returns 0 on success and -ENOMEM on failure
 */
int
mxlnd_conn_alloc_locked(kmx_conn_t **connp, kmx_peer_t *peer)
{
        int             i       = 0;
        int             ret     = 0;
        int             ipage   = 0;
        int             offset  = 0;
        void           *addr    = NULL;
        kmx_conn_t     *conn    = NULL;
        kmx_pages_t    *pages   = NULL;
        struct page    *page    = NULL;
        kmx_ctx_t      *rx      = NULL;

        LASSERT(peer != NULL);

        MXLND_ALLOC(conn, sizeof (*conn));
        if (conn == NULL) {
                CNETERR("Cannot allocate conn\n");
                return -ENOMEM;
        }
        CDEBUG(D_NET, "allocated conn 0x%p for peer 0x%p\n", conn, peer);

        memset(conn, 0, sizeof(*conn));

        ret = mxlnd_alloc_pages(&pages, MXLND_RX_MSG_PAGES());
        if (ret != 0) {
                CERROR("Can't allocate rx pages\n");
                MXLND_FREE(conn, sizeof(*conn));
                return -ENOMEM;
        }
        conn->mxk_rx_pages = pages;

        MXLND_ALLOC(conn->mxk_rxs, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));
        if (conn->mxk_rxs == NULL) {
                CERROR("Can't allocate %d rx descriptors\n", MXLND_RX_MSGS());
                mxlnd_free_pages(pages);
                MXLND_FREE(conn, sizeof(*conn));
                return -ENOMEM;
        }

        memset(conn->mxk_rxs, 0, MXLND_RX_MSGS() * sizeof(kmx_ctx_t));

	conn->mxk_peer = peer;
	CFS_INIT_LIST_HEAD(&conn->mxk_list);
	CFS_INIT_LIST_HEAD(&conn->mxk_zombie);
	atomic_set(&conn->mxk_refcount, 2); /* ref for owning peer
						   and one for the caller */
	if (peer->mxp_nid == kmxlnd_data.kmx_ni->ni_nid) {
		u64     nic_id  = 0ULL;
                u32     ep_id   = 0;

                /* this is localhost, set the epa and status as up */
                mxlnd_set_conn_status(conn, MXLND_CONN_READY);
                conn->mxk_epa = kmxlnd_data.kmx_epa;
                mx_set_endpoint_addr_context(conn->mxk_epa, (void *) conn);
                peer->mxp_reconnect_time = 0;
                mx_decompose_endpoint_addr(kmxlnd_data.kmx_epa, &nic_id, &ep_id);
                peer->mxp_nic_id = nic_id;
                peer->mxp_ep_id = ep_id;
                conn->mxk_incarnation = kmxlnd_data.kmx_incarnation;
                conn->mxk_timeout = 0;
        } else {
                /* conn->mxk_incarnation = 0 - will be set by peer */
                /* conn->mxk_sid = 0 - will be set by peer */
                mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
                /* mxk_epa - to be set after mx_iconnect() */
        }
	spin_lock_init(&conn->mxk_lock);
        /* conn->mxk_timeout = 0 */
        /* conn->mxk_last_tx = 0 */
        /* conn->mxk_last_rx = 0 */
        CFS_INIT_LIST_HEAD(&conn->mxk_rx_idle);

        conn->mxk_credits = *kmxlnd_tunables.kmx_peercredits;
        /* mxk_outstanding = 0 */

        CFS_INIT_LIST_HEAD(&conn->mxk_tx_credit_queue);
        CFS_INIT_LIST_HEAD(&conn->mxk_tx_free_queue);
        /* conn->mxk_ntx_msgs = 0 */
        /* conn->mxk_ntx_data = 0 */
        /* conn->mxk_ntx_posted = 0 */
        /* conn->mxk_data_posted = 0 */
        CFS_INIT_LIST_HEAD(&conn->mxk_pending);

        for (i = 0; i < MXLND_RX_MSGS(); i++) {

                rx = &conn->mxk_rxs[i];
                rx->mxc_type = MXLND_REQ_RX;
                CFS_INIT_LIST_HEAD(&rx->mxc_list);

                /* map mxc_msg to page */
                page = pages->mxg_pages[ipage];
                addr = page_address(page);
                LASSERT(addr != NULL);
                rx->mxc_msg = (kmx_msg_t *)(addr + offset);
                rx->mxc_seg.segment_ptr = MX_PA_TO_U64(virt_to_phys(rx->mxc_msg));

                rx->mxc_conn = conn;
                rx->mxc_peer = peer;
                rx->mxc_nid = peer->mxp_nid;

                mxlnd_ctx_init(rx);

                offset += MXLND_MSG_SIZE;
                LASSERT (offset <= PAGE_SIZE);

                if (offset == PAGE_SIZE) {
                        offset = 0;
                        ipage++;
                        LASSERT (ipage <= MXLND_TX_MSG_PAGES());
                }

                cfs_list_add_tail(&rx->mxc_list, &conn->mxk_rx_idle);
        }

        *connp = conn;

        mxlnd_peer_addref(peer);        /* add a ref for this conn */

        /* add to front of peer's conns list */
        cfs_list_add(&conn->mxk_list, &peer->mxp_conns);
        peer->mxp_conn = conn;
        return 0;
}

int
mxlnd_conn_alloc(kmx_conn_t **connp, kmx_peer_t *peer)
{
        int             ret     = 0;
	rwlock_t   *g_lock  = &kmxlnd_data.kmx_global_lock;

	write_lock(g_lock);
        ret = mxlnd_conn_alloc_locked(connp, peer);
	write_unlock(g_lock);
        return ret;
}

int
mxlnd_q_pending_ctx(kmx_ctx_t *ctx)
{
	int		ret	= 0;
	kmx_conn_t	*conn	= ctx->mxc_conn;

	ctx->mxc_state = MXLND_CTX_PENDING;
	if (conn != NULL) {
		spin_lock(&conn->mxk_lock);
                if (conn->mxk_status >= MXLND_CONN_INIT) {
                        cfs_list_add_tail(&ctx->mxc_list, &conn->mxk_pending);
                        if (conn->mxk_timeout == 0 || ctx->mxc_deadline < conn->mxk_timeout) {
                                conn->mxk_timeout = ctx->mxc_deadline;
                        }
                } else {
                        ctx->mxc_state = MXLND_CTX_COMPLETED;
                        ret = -1;
                }
		spin_unlock(&conn->mxk_lock);
	}
	return ret;
}

int
mxlnd_deq_pending_ctx(kmx_ctx_t *ctx)
{
        LASSERT(ctx->mxc_state == MXLND_CTX_PENDING ||
                ctx->mxc_state == MXLND_CTX_COMPLETED);
        if (ctx->mxc_state != MXLND_CTX_PENDING &&
            ctx->mxc_state != MXLND_CTX_COMPLETED) {
                CNETERR("deq ctx->mxc_state = %s\n",
                        mxlnd_ctxstate_to_str(ctx->mxc_state));
        }
        ctx->mxc_state = MXLND_CTX_COMPLETED;
        if (!cfs_list_empty(&ctx->mxc_list)) {
                kmx_conn_t      *conn = ctx->mxc_conn;
                kmx_ctx_t       *next = NULL;

                LASSERT(conn != NULL);
		spin_lock(&conn->mxk_lock);
                cfs_list_del_init(&ctx->mxc_list);
                conn->mxk_timeout = 0;
                if (!cfs_list_empty(&conn->mxk_pending)) {
                        next = cfs_list_entry(conn->mxk_pending.next,
                                              kmx_ctx_t, mxc_list);
                        conn->mxk_timeout = next->mxc_deadline;
                }
		spin_unlock(&conn->mxk_lock);
	}
	return 0;
}

/**
 * mxlnd_peer_free - free the peer
 * @peer - a kmx_peer pointer
 *
 * The calling function should decrement the rxs, drain the tx queues and
 * remove the peer from the peers list first then destroy it.
 */
void
mxlnd_peer_free(kmx_peer_t *peer)
{
	CDEBUG(D_NET, "freeing peer 0x%p %s\n", peer, libcfs_nid2str(peer->mxp_nid));

	LASSERT (atomic_read(&peer->mxp_refcount) == 0);

	if (!cfs_list_empty(&peer->mxp_list)) {
		/* assume we are locked */
		cfs_list_del_init(&peer->mxp_list);
	}

	MXLND_FREE(peer, sizeof (*peer));
	atomic_dec(&kmxlnd_data.kmx_npeers);
	return;
}

static int
mxlnd_lookup_mac(u32 ip, u64 *tmp_id)
{
        int                     ret     = -EHOSTUNREACH;
        unsigned char           *haddr  = NULL;
        struct net_device       *dev    = NULL;
        struct neighbour        *n      = NULL;
        __be32                  dst_ip  = htonl(ip);

        dev = dev_get_by_name(*kmxlnd_tunables.kmx_default_ipif);
        if (dev == NULL)
                return -ENODEV;

        haddr = (unsigned char *) tmp_id + 2; /* MAC is only 6 bytes */

        n = neigh_lookup(&arp_tbl, &dst_ip, dev);
        if (n) {
                n->used = jiffies;
                if (n->nud_state & NUD_VALID) {
                        memcpy(haddr, n->ha, dev->addr_len);
                        neigh_release(n);
                        ret = 0;
                }
        }

        dev_put(dev);

        return ret;
}


/* We only want the MAC address of the peer's Myricom NIC. We
 * require that each node has the IPoMX interface (myriN) up.
 * We will not pass any traffic over IPoMX, but it allows us
 * to get the MAC address. */
static int
mxlnd_ip2nic_id(u32 ip, u64 *nic_id, int tries)
{
        int                     ret     = 0;
        int                     try     = 1;
        int                     fatal   = 0;
        u64                     tmp_id  = 0ULL;
        cfs_socket_t            *sock   = NULL;

        do {
                CDEBUG(D_NET, "try %d of %d tries\n", try, tries);
                ret = mxlnd_lookup_mac(ip, &tmp_id);
                if (ret == 0) {
                        break;
                } else {
                        /* not found, try to connect (force an arp) */
                        ret = libcfs_sock_connect(&sock, &fatal, 0, 0, ip, 987);
                        if (ret == -ECONNREFUSED) {
                                /* peer is there, get the MAC address */
                                mxlnd_lookup_mac(ip, &tmp_id);
                                if (tmp_id != 0ULL)
                                        ret = 0;
                                break;
			} else if (ret == -EHOSTUNREACH && try < tries) {
				/* add a little backoff */
				CDEBUG(D_NET, "sleeping for %lu jiffies\n",
				       msecs_to_jiffies(MSEC_PER_SEC / 4));
				mxlnd_sleep(msecs_to_jiffies(MSEC_PER_SEC / 4));
			}
                }
        } while (try++ < tries);
        CDEBUG(D_NET, "done trying. ret = %d\n", ret);

        if (tmp_id == 0ULL)
                ret = -EHOSTUNREACH;
#if __BYTE_ORDER == __LITTLE_ENDIAN
        *nic_id = ___arch__swab64(tmp_id);
#else
        *nic_id = tmp_id;
#endif
        return ret;
}

/**
 * mxlnd_peer_alloc - allocate and initialize a new peer struct
 * @peerp - address of a kmx_peer pointer
 * @nid - LNET node id
 *
 * Returns 0 on success and -ENOMEM on failure
 */
int
mxlnd_peer_alloc(kmx_peer_t **peerp, lnet_nid_t nid, u32 board, u32 ep_id, u64 nic_id)
{
        int             ret     = 0;
        u32             ip      = LNET_NIDADDR(nid);
        kmx_peer_t     *peer    = NULL;

        LASSERT (nid != LNET_NID_ANY && nid != 0LL);

        MXLND_ALLOC(peer, sizeof (*peer));
        if (peer == NULL) {
                CNETERR("Cannot allocate peer for NID 0x%llx\n",
                       nid);
                return -ENOMEM;
        }
        CDEBUG(D_NET, "allocated peer 0x%p for NID 0x%llx\n", peer, nid);

        memset(peer, 0, sizeof(*peer));

	CFS_INIT_LIST_HEAD(&peer->mxp_list);
	peer->mxp_nid = nid;
	/* peer->mxp_ni unused - may be used for multi-rail */
	atomic_set(&peer->mxp_refcount, 1);     /* ref for kmx_peers list */

        peer->mxp_board = board;
        peer->mxp_ep_id = ep_id;
        peer->mxp_nic_id = nic_id;

        CFS_INIT_LIST_HEAD(&peer->mxp_conns);
        ret = mxlnd_conn_alloc(&peer->mxp_conn, peer); /* adds 2nd conn ref here... */
        if (ret != 0) {
                mxlnd_peer_decref(peer);
                return ret;
        }
        CFS_INIT_LIST_HEAD(&peer->mxp_tx_queue);

        if (peer->mxp_nic_id != 0ULL)
                nic_id = peer->mxp_nic_id;

        if (nic_id == 0ULL) {
                ret = mxlnd_ip2nic_id(ip, &nic_id, 1);
                if (ret == 0) {
                        peer->mxp_nic_id = nic_id;
                        mx_nic_id_to_board_number(nic_id, &peer->mxp_board);
                }
        }

        peer->mxp_nic_id = nic_id; /* may be 0ULL if ip2nic_id() failed */

        /* peer->mxp_reconnect_time = 0 */
        /* peer->mxp_incompatible = 0 */

        *peerp = peer;
        return 0;
}

static inline kmx_peer_t *
mxlnd_find_peer_by_nid_locked(lnet_nid_t nid)
{
        int             found   = 0;
        int             hash    = 0;
        kmx_peer_t      *peer   = NULL;

        hash = mxlnd_nid_to_hash(nid);

        cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[hash], mxp_list) {
                if (peer->mxp_nid == nid) {
                        found = 1;
                        mxlnd_peer_addref(peer);
                        break;
                }
        }
        return (found ? peer : NULL);
}

static kmx_peer_t *
mxlnd_find_peer_by_nid(lnet_nid_t nid, int create)
{
        int             ret     = 0;
        int             hash    = 0;
        kmx_peer_t      *peer   = NULL;
        kmx_peer_t      *old    = NULL;
	rwlock_t    *g_lock = &kmxlnd_data.kmx_global_lock;

	read_lock(g_lock);
        peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */

        if ((peer && peer->mxp_conn) || /* found peer with conn or */
            (!peer && !create)) {       /* did not find peer and do not create one */
		read_unlock(g_lock);
                return peer;
        }

	read_unlock(g_lock);

        /* if peer but _not_ conn */
        if (peer && !peer->mxp_conn) {
                if (create) {
			write_lock(g_lock);
                        if (!peer->mxp_conn) { /* check again */
                                /* create the conn */
                                ret = mxlnd_conn_alloc_locked(&peer->mxp_conn, peer);
                                if (ret != 0) {
                                        /* we tried, return the peer only.
                                         * the caller needs to see if the conn exists */
                                        CNETERR("%s: %s could not alloc conn\n",
                                        __func__, libcfs_nid2str(peer->mxp_nid));
                                } else {
                                        /* drop extra conn ref */
                                        mxlnd_conn_decref(peer->mxp_conn);
                                }
                        }
			write_unlock(g_lock);
                }
                return peer;
        }

        /* peer not found and we need to create one */
        hash = mxlnd_nid_to_hash(nid);

        /* create peer (and conn) */
        /* adds conn ref for peer and one for this function */
        ret = mxlnd_peer_alloc(&peer, nid, *kmxlnd_tunables.kmx_board,
                               *kmxlnd_tunables.kmx_ep_id, 0ULL);
        if (ret != 0) /* no memory, peer is NULL */
                return NULL;

	write_lock(g_lock);

        /* look again */
        old = mxlnd_find_peer_by_nid_locked(nid);
        if (old) {
                /* someone already created one */
                mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
                mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
                mxlnd_peer_decref(peer);
                peer = old;
        } else {
		/* no other peer, use this one */
		cfs_list_add_tail(&peer->mxp_list,
				  &kmxlnd_data.kmx_peers[hash]);
		atomic_inc(&kmxlnd_data.kmx_npeers);
		mxlnd_peer_addref(peer);
		mxlnd_conn_decref(peer->mxp_conn); /* drop ref from peer_alloc */
        }

	write_unlock(g_lock);

        return peer;
}

static inline int
mxlnd_tx_requires_credit(kmx_ctx_t *tx)
{
        return (tx->mxc_msg_type == MXLND_MSG_EAGER ||
                tx->mxc_msg_type == MXLND_MSG_GET_REQ ||
                tx->mxc_msg_type == MXLND_MSG_PUT_REQ ||
                tx->mxc_msg_type == MXLND_MSG_NOOP);
}

/**
 * mxlnd_init_msg - set type and number of bytes
 * @msg - msg pointer
 * @type - of message
 * @body_nob - bytes in msg body
 */
static inline void
mxlnd_init_msg(kmx_msg_t *msg, u8 type, int body_nob)
{
        msg->mxm_type = type;
        msg->mxm_nob  = offsetof(kmx_msg_t, mxm_u) + body_nob;
}

static inline void
mxlnd_init_tx_msg (kmx_ctx_t *tx, u8 type, int body_nob, lnet_nid_t nid)
{
        int             nob     = offsetof (kmx_msg_t, mxm_u) + body_nob;
        kmx_msg_t       *msg    = NULL;

        LASSERT (tx != NULL);
        LASSERT (nob <= MXLND_MSG_SIZE);

        tx->mxc_nid = nid;
        /* tx->mxc_peer should have already been set if we know it */
        tx->mxc_msg_type = type;
        tx->mxc_nseg = 1;
        /* tx->mxc_seg.segment_ptr is already pointing to mxc_page */
        tx->mxc_seg.segment_length = nob;
        tx->mxc_pin_type = MX_PIN_PHYSICAL;

        msg = tx->mxc_msg;
        msg->mxm_type = type;
        msg->mxm_nob  = nob;

        return;
}

static inline __u32
mxlnd_cksum (void *ptr, int nob)
{
        char  *c  = ptr;
        __u32  sum = 0;

        while (nob-- > 0)
                sum = ((sum << 1) | (sum >> 31)) + *c++;

        /* ensure I don't return 0 (== no checksum) */
        return (sum == 0) ? 1 : sum;
}

/**
 * mxlnd_pack_msg_locked - complete msg info
 * @tx - msg to send
 */
static inline void
mxlnd_pack_msg_locked(kmx_ctx_t *tx)
{
        kmx_msg_t       *msg    = tx->mxc_msg;

        /* type and nob should already be set in init_msg() */
        msg->mxm_magic    = MXLND_MSG_MAGIC;
        msg->mxm_version  = MXLND_MSG_VERSION;
        /*   mxm_type */
        /* don't use mxlnd_tx_requires_credit() since we want PUT_ACK to
         * return credits as well */
        if (tx->mxc_msg_type != MXLND_MSG_CONN_REQ &&
            tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
                msg->mxm_credits  = tx->mxc_conn->mxk_outstanding;
                tx->mxc_conn->mxk_outstanding = 0;
        } else {
                msg->mxm_credits  = 0;
        }
        /*   mxm_nob */
        msg->mxm_cksum    = 0;
        msg->mxm_srcnid   = kmxlnd_data.kmx_ni->ni_nid;
        msg->mxm_srcstamp = kmxlnd_data.kmx_incarnation;
        msg->mxm_dstnid   = tx->mxc_nid;
        /* if it is a new peer, the dststamp will be 0 */
        msg->mxm_dststamp = tx->mxc_conn->mxk_incarnation;

        if (*kmxlnd_tunables.kmx_cksum) {
                msg->mxm_cksum = mxlnd_cksum(msg, msg->mxm_nob);
        }
}

int
mxlnd_unpack_msg(kmx_msg_t *msg, int nob)
{
        const int hdr_size      = offsetof(kmx_msg_t, mxm_u);
        __u32     msg_cksum     = 0;
        int       flip          = 0;
        int       msg_nob       = 0;

        /* 6 bytes are enough to have received magic + version */
        if (nob < 6) {
                CNETERR("not enough bytes for magic + hdr: %d\n", nob);
                return -EPROTO;
        }

        if (msg->mxm_magic == MXLND_MSG_MAGIC) {
                flip = 0;
        } else if (msg->mxm_magic == __swab32(MXLND_MSG_MAGIC)) {
                flip = 1;
        } else {
                CNETERR("Bad magic: %08x\n", msg->mxm_magic);
                return -EPROTO;
        }

        if (msg->mxm_version !=
            (flip ? __swab16(MXLND_MSG_VERSION) : MXLND_MSG_VERSION)) {
                CNETERR("Bad version: %d\n", msg->mxm_version);
                return -EPROTO;
        }

        if (nob < hdr_size) {
                CNETERR("not enough for a header: %d\n", nob);
                return -EPROTO;
        }

        msg_nob = flip ? __swab32(msg->mxm_nob) : msg->mxm_nob;
        if (msg_nob > nob) {
                CNETERR("Short message: got %d, wanted %d\n", nob, msg_nob);
                return -EPROTO;
        }

        /* checksum must be computed with mxm_cksum zero and BEFORE anything
         * gets flipped */
        msg_cksum = flip ? __swab32(msg->mxm_cksum) : msg->mxm_cksum;
        msg->mxm_cksum = 0;
        if (msg_cksum != 0 && msg_cksum != mxlnd_cksum(msg, msg_nob)) {
                CNETERR("Bad checksum\n");
                return -EPROTO;
        }
        msg->mxm_cksum = msg_cksum;

        if (flip) {
                /* leave magic unflipped as a clue to peer endianness */
                __swab16s(&msg->mxm_version);
                CLASSERT (sizeof(msg->mxm_type) == 1);
                CLASSERT (sizeof(msg->mxm_credits) == 1);
                msg->mxm_nob = msg_nob;
                __swab64s(&msg->mxm_srcnid);
                __swab64s(&msg->mxm_srcstamp);
                __swab64s(&msg->mxm_dstnid);
                __swab64s(&msg->mxm_dststamp);
        }

        if (msg->mxm_srcnid == LNET_NID_ANY) {
                CNETERR("Bad src nid: %s\n", libcfs_nid2str(msg->mxm_srcnid));
                return -EPROTO;
        }

        switch (msg->mxm_type) {
        default:
                CNETERR("Unknown message type %x\n", msg->mxm_type);
                return -EPROTO;

        case MXLND_MSG_NOOP:
                break;

        case MXLND_MSG_EAGER:
                if (msg_nob < offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0])) {
                        CNETERR("Short EAGER: %d(%d)\n", msg_nob,
                               (int)offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[0]));
                        return -EPROTO;
                }
                break;

        case MXLND_MSG_PUT_REQ:
                if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_req)) {
                        CNETERR("Short PUT_REQ: %d(%d)\n", msg_nob,
                               (int)(hdr_size + sizeof(msg->mxm_u.put_req)));
                        return -EPROTO;
                }
                if (flip)
                        __swab64s(&msg->mxm_u.put_req.mxprm_cookie);
                break;

        case MXLND_MSG_PUT_ACK:
                if (msg_nob < hdr_size + sizeof(msg->mxm_u.put_ack)) {
                        CNETERR("Short PUT_ACK: %d(%d)\n", msg_nob,
                               (int)(hdr_size + sizeof(msg->mxm_u.put_ack)));
                        return -EPROTO;
                }
                if (flip) {
                        __swab64s(&msg->mxm_u.put_ack.mxpam_src_cookie);
                        __swab64s(&msg->mxm_u.put_ack.mxpam_dst_cookie);
                }
                break;

        case MXLND_MSG_GET_REQ:
                if (msg_nob < hdr_size + sizeof(msg->mxm_u.get_req)) {
                        CNETERR("Short GET_REQ: %d(%d)\n", msg_nob,
                               (int)(hdr_size + sizeof(msg->mxm_u.get_req)));
                        return -EPROTO;
                }
                if (flip) {
                        __swab64s(&msg->mxm_u.get_req.mxgrm_cookie);
                }
                break;

        case MXLND_MSG_CONN_REQ:
        case MXLND_MSG_CONN_ACK:
                if (msg_nob < hdr_size + sizeof(msg->mxm_u.conn_req)) {
                        CNETERR("Short connreq/ack: %d(%d)\n", msg_nob,
                               (int)(hdr_size + sizeof(msg->mxm_u.conn_req)));
                        return -EPROTO;
                }
                if (flip) {
                        __swab32s(&msg->mxm_u.conn_req.mxcrm_queue_depth);
                        __swab32s(&msg->mxm_u.conn_req.mxcrm_eager_size);
                }
                break;
        }
        return 0;
}


/**
 * mxlnd_recv_msg
 * @lntmsg - the LNET msg that this is continuing. If EAGER, then NULL.
 * @rx
 * @msg_type
 * @cookie
 * @length - length of incoming message
 * @pending - add to kmx_pending (0 is NO and 1 is YES)
 *
 * The caller gets the rx and sets nid, peer and conn if known.
 *
 * Returns 0 on success and -1 on failure
 */
int
mxlnd_recv_msg(lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie, u32 length)
{
        int             ret     = 0;
        mx_return_t     mxret   = MX_SUCCESS;
        uint64_t        mask    = ~(MXLND_ERROR_MASK);

        rx->mxc_msg_type = msg_type;
        rx->mxc_lntmsg[0] = lntmsg; /* may be NULL if EAGER */
        rx->mxc_cookie = cookie;
        /* rx->mxc_match may already be set */
        /* rx->mxc_seg.segment_ptr is already set */
        rx->mxc_seg.segment_length = length;
        ret = mxlnd_q_pending_ctx(rx);
        if (ret == -1) {
                /* the caller is responsible for calling conn_decref() if needed */
                return -1;
        }
        mxret = mx_kirecv(kmxlnd_data.kmx_endpt, &rx->mxc_seg, 1, MX_PIN_PHYSICAL,
                          cookie, mask, (void *) rx, &rx->mxc_mxreq);
        if (mxret != MX_SUCCESS) {
                mxlnd_deq_pending_ctx(rx);
                CNETERR("mx_kirecv() failed with %s (%d)\n",
                        mx_strerror(mxret), (int) mxret);
                return -1;
        }
        return 0;
}


/**
 * mxlnd_unexpected_recv - this is the callback function that will handle
 *                         unexpected receives
 * @context - NULL, ignore
 * @source - the peer's mx_endpoint_addr_t
 * @match_value - the msg's bits, should be MXLND_MSG_EAGER
 * @length - length of incoming message
 * @data_if_available - used for CONN_[REQ|ACK]
 *
 * If it is an eager-sized msg, we will call recv_msg() with the actual
 * length. If it is a large message, we will call recv_msg() with a
 * length of 0 bytes to drop it because we should never have a large,
 * unexpected message.
 *
 * NOTE - The MX library blocks until this function completes. Make it as fast as
 * possible. DO NOT allocate memory which can block!
 *
 * If we cannot get a rx or the conn is closed, drop the message on the floor
 * (i.e. recv 0 bytes and ignore).
 */
mx_unexp_handler_action_t
mxlnd_unexpected_recv(void *context, mx_endpoint_addr_t source,
                 uint64_t match_value, uint32_t length, void *data_if_available)
{
        int             ret             = 0;
        kmx_ctx_t       *rx             = NULL;
        mx_ksegment_t   seg;
        u8              msg_type        = 0;
        u8              error           = 0;
        u64             cookie          = 0ULL;
        kmx_conn_t      *conn           = NULL;
        kmx_peer_t      *peer           = NULL;
        u64             nic_id          = 0ULL;
        u32             ep_id           = 0;
        u32             sid             = 0;

        /* TODO this will change to the net struct */
        if (context != NULL) {
                CNETERR("non-NULL context\n");
        }

#if MXLND_DEBUG
        CDEBUG(D_NET, "bits=0x%llx length=%d\n", match_value, length);
#endif

        mx_decompose_endpoint_addr2(source, &nic_id, &ep_id, &sid);
        mxlnd_parse_match(match_value, &msg_type, &error, &cookie);
	read_lock(&kmxlnd_data.kmx_global_lock);
        mx_get_endpoint_addr_context(source, (void **) &conn);
        if (conn) {
                mxlnd_conn_addref(conn); /* add ref for this function */
                peer = conn->mxk_peer;
        }
	read_unlock(&kmxlnd_data.kmx_global_lock);

        if (msg_type == MXLND_MSG_BYE) {
                if (conn) {
                        CDEBUG(D_NET, "peer %s sent BYE msg\n",
                                        libcfs_nid2str(peer->mxp_nid));
                        mxlnd_conn_disconnect(conn, 1, 0);
                        mxlnd_conn_decref(conn); /* drop ref taken above */
                }
                return MX_RECV_FINISHED;
        }

        if (msg_type == MXLND_MSG_CONN_REQ) {
                kmx_connparams_t       *cp      = NULL;
                const int       expected        = offsetof(kmx_msg_t, mxm_u) +
                                                  sizeof(kmx_connreq_msg_t);

                if (conn) mxlnd_conn_decref(conn); /* drop ref taken above */
                if (unlikely(length != expected || !data_if_available)) {
                        CNETERR("received invalid CONN_REQ from %llx "
                                "length=%d (expected %d)\n", nic_id, length, expected);
                        mxlnd_send_message(source, MXLND_MSG_CONN_ACK, EPROTO, 0);
                        return MX_RECV_FINISHED;
                }

                ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
                                         conn, peer, data_if_available);
                if (unlikely(ret != 0)) {
                        CNETERR("unable to alloc CONN_REQ from %llx:%d\n",
                                nic_id, ep_id);
                        mxlnd_send_message(source, MXLND_MSG_CONN_ACK, ENOMEM, 0);
                        return MX_RECV_FINISHED;
                }
		spin_lock(&kmxlnd_data.kmx_conn_lock);
		cfs_list_add_tail(&cp->mxr_list, &kmxlnd_data.kmx_conn_reqs);
		spin_unlock(&kmxlnd_data.kmx_conn_lock);
		up(&kmxlnd_data.kmx_conn_sem);
		return MX_RECV_FINISHED;
	}
        if (msg_type == MXLND_MSG_CONN_ACK) {
                kmx_connparams_t  *cp           = NULL;
                const int       expected        = offsetof(kmx_msg_t, mxm_u) +
                                                  sizeof(kmx_connreq_msg_t);

                LASSERT(conn);
                if (unlikely(error != 0)) {
                        CNETERR("received CONN_ACK from %s with error -%d\n",
                               libcfs_nid2str(peer->mxp_nid), (int) error);
                        mxlnd_conn_disconnect(conn, 1, 0);
                } else if (unlikely(length != expected || !data_if_available)) {
                        CNETERR("received %s CONN_ACK from %s "
                               "length=%d (expected %d)\n",
                               data_if_available ? "short" : "missing",
                               libcfs_nid2str(peer->mxp_nid), length, expected);
                        mxlnd_conn_disconnect(conn, 1, 1);
                } else {
                        /* peer is ready for messages */
                        ret = mxlnd_connparams_alloc(&cp, context, source, match_value, length,
                                         conn, peer, data_if_available);
                        if (unlikely(ret != 0)) {
                                CNETERR("unable to alloc kmx_connparams_t"
                                               " from %llx:%d\n", nic_id, ep_id);
                                mxlnd_conn_disconnect(conn, 1, 1);
			} else {
				spin_lock(&kmxlnd_data.kmx_conn_lock);
				cfs_list_add_tail(&cp->mxr_list,
						  &kmxlnd_data.kmx_conn_reqs);
				spin_unlock(&kmxlnd_data.kmx_conn_lock);
				up(&kmxlnd_data.kmx_conn_sem);
			}
                }
                mxlnd_conn_decref(conn); /* drop ref taken above */

                return MX_RECV_FINISHED;
        }

        /* Handle unexpected messages (PUT_REQ and GET_REQ) */

        LASSERT(peer != NULL && conn != NULL);

        rx = mxlnd_get_idle_rx(conn);
        if (rx != NULL) {
                if (length <= MXLND_MSG_SIZE) {
                        ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, length);
                } else {
                        CNETERR("unexpected large receive with "
                                "match_value=0x%llx length=%d\n",
                                match_value, length);
                        ret = mxlnd_recv_msg(NULL, rx, msg_type, match_value, 0);
                }

                if (ret == 0) {
                        /* hold conn ref until rx completes */
                        rx->mxc_conn = conn;
                        rx->mxc_peer = peer;
                        rx->mxc_nid = peer->mxp_nid;
                } else {
                        CNETERR("could not post receive\n");
                        mxlnd_put_idle_rx(rx);
                }
        }

        /* Encountered error, drop incoming message on the floor */
        /* We could use MX_RECV_FINISHED but posting the receive of 0 bytes
         * uses the standard code path and acks the sender normally */

        if (rx == NULL || ret != 0) {
                mxlnd_conn_decref(conn); /* drop ref taken above */
                if (rx == NULL) {
                        CNETERR("no idle rxs available - dropping rx"
                                " 0x%llx from %s\n", match_value,
                                libcfs_nid2str(peer->mxp_nid));
                } else {
                        /* ret != 0 */
                        CNETERR("disconnected peer - dropping rx\n");
                }
                seg.segment_ptr = 0ULL;
                seg.segment_length = 0;
                mx_kirecv(kmxlnd_data.kmx_endpt, &seg, 1, MX_PIN_PHYSICAL,
                          match_value, ~0ULL, NULL, NULL);
        }

        return MX_RECV_CONTINUE;
}


int
mxlnd_get_peer_info(int index, lnet_nid_t *nidp, int *count)
{
        int              i      = 0;
        int              ret    = -ENOENT;
        kmx_peer_t      *peer   = NULL;

	read_lock(&kmxlnd_data.kmx_global_lock);
        for (i = 0; i < MXLND_HASH_SIZE; i++) {
                cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
                                        mxp_list) {
			if (index-- == 0) {
				*nidp = peer->mxp_nid;
				*count = atomic_read(&peer->mxp_refcount);
				ret = 0;
				break;
			}
                }
        }
	read_unlock(&kmxlnd_data.kmx_global_lock);

        return ret;
}

void
mxlnd_del_peer_locked(kmx_peer_t *peer)
{
        if (peer->mxp_conn) {
                mxlnd_conn_disconnect(peer->mxp_conn, 1, 1);
        } else {
                cfs_list_del_init(&peer->mxp_list); /* remove from the global list */
                mxlnd_peer_decref(peer); /* drop global list ref */
        }
        return;
}

int
mxlnd_del_peer(lnet_nid_t nid)
{
        int             i       = 0;
        int             ret     = 0;
        kmx_peer_t      *peer   = NULL;
        kmx_peer_t      *next   = NULL;

        if (nid != LNET_NID_ANY) {
                peer = mxlnd_find_peer_by_nid(nid, 0); /* adds peer ref */
        }
	write_lock(&kmxlnd_data.kmx_global_lock);
        if (nid != LNET_NID_ANY) {
                if (peer == NULL) {
                        ret = -ENOENT;
                } else {
                        mxlnd_peer_decref(peer); /* and drops it */
                        mxlnd_del_peer_locked(peer);
                }
        } else { /* LNET_NID_ANY */
                for (i = 0; i < MXLND_HASH_SIZE; i++) {
                        cfs_list_for_each_entry_safe(peer, next,
                                                     &kmxlnd_data.kmx_peers[i],
                                                     mxp_list) {
                                mxlnd_del_peer_locked(peer);
                        }
                }
        }
	write_unlock(&kmxlnd_data.kmx_global_lock);

        return ret;
}

kmx_conn_t *
mxlnd_get_conn_by_idx(int index)
{
        int              i      = 0;
        kmx_peer_t      *peer   = NULL;
        kmx_conn_t      *conn   = NULL;

	read_lock(&kmxlnd_data.kmx_global_lock);
        for (i = 0; i < MXLND_HASH_SIZE; i++) {
                cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
                                        mxp_list) {
                        cfs_list_for_each_entry(conn, &peer->mxp_conns,
                                                mxk_list) {
                                if (index-- > 0) {
                                        continue;
                                }

                                mxlnd_conn_addref(conn); /* add ref here, dec in ctl() */
				read_unlock(&kmxlnd_data.kmx_global_lock);
                                return conn;
                        }
                }
        }
	read_unlock(&kmxlnd_data.kmx_global_lock);

        return NULL;
}

void
mxlnd_close_matching_conns_locked(kmx_peer_t *peer)
{
        kmx_conn_t      *conn   = NULL;
        kmx_conn_t      *next   = NULL;

        cfs_list_for_each_entry_safe(conn, next, &peer->mxp_conns, mxk_list)
                mxlnd_conn_disconnect(conn, 0, 1);

        return;
}

int
mxlnd_close_matching_conns(lnet_nid_t nid)
{
        int             i       = 0;
        int             ret     = 0;
        kmx_peer_t      *peer   = NULL;

	write_lock(&kmxlnd_data.kmx_global_lock);
        if (nid != LNET_NID_ANY) {
                peer = mxlnd_find_peer_by_nid_locked(nid); /* adds peer ref */
                if (peer == NULL) {
                        ret = -ENOENT;
                } else {
                        mxlnd_close_matching_conns_locked(peer);
                        mxlnd_peer_decref(peer); /* and drops it here */
                }
        } else { /* LNET_NID_ANY */
                for (i = 0; i < MXLND_HASH_SIZE; i++) {
                        cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],                                                mxp_list)
                                mxlnd_close_matching_conns_locked(peer);
                }
        }
	write_unlock(&kmxlnd_data.kmx_global_lock);

        return ret;
}

/**
 * mxlnd_ctl - modify MXLND parameters
 * @ni - LNET interface handle
 * @cmd - command to change
 * @arg - the ioctl data
 */
int
mxlnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
{
        struct libcfs_ioctl_data *data  = arg;
        int                       ret   = -EINVAL;

        LASSERT (ni == kmxlnd_data.kmx_ni);

        switch (cmd) {
        case IOC_LIBCFS_GET_PEER: {
                lnet_nid_t      nid     = 0;
                int             count   = 0;

                ret = mxlnd_get_peer_info(data->ioc_count, &nid, &count);
                data->ioc_nid    = nid;
                data->ioc_count  = count;
                break;
        }
        case IOC_LIBCFS_DEL_PEER: {
                ret = mxlnd_del_peer(data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_GET_CONN: {
                kmx_conn_t      *conn = NULL;

                conn = mxlnd_get_conn_by_idx(data->ioc_count);
                if (conn == NULL) {
                        ret = -ENOENT;
                } else {
                        ret = 0;
                        data->ioc_nid = conn->mxk_peer->mxp_nid;
                        mxlnd_conn_decref(conn); /* dec ref taken in get_conn_by_idx() */
                }
                break;
        }
        case IOC_LIBCFS_CLOSE_CONNECTION: {
                ret = mxlnd_close_matching_conns(data->ioc_nid);
                break;
        }
        default:
                CNETERR("unknown ctl(%d)\n", cmd);
                break;
        }

        return ret;
}

/**
 * mxlnd_peer_queue_tx_locked - add the tx to the peer's tx queue
 * @tx
 *
 * Add the tx to the peer's msg or data queue. The caller has locked the peer.
 */
void
mxlnd_peer_queue_tx_locked(kmx_ctx_t *tx)
{
        u8              msg_type        = tx->mxc_msg_type;
        kmx_conn_t      *conn           = tx->mxc_conn;

        LASSERT (msg_type != 0);
        LASSERT (tx->mxc_nid != 0);
        LASSERT (tx->mxc_peer != NULL);
        LASSERT (tx->mxc_conn != NULL);

        tx->mxc_incarnation = conn->mxk_incarnation;

        if (msg_type != MXLND_MSG_PUT_DATA &&
            msg_type != MXLND_MSG_GET_DATA) {
                /* msg style tx */
                if (mxlnd_tx_requires_credit(tx)) {
                        cfs_list_add_tail(&tx->mxc_list,
                                          &conn->mxk_tx_credit_queue);
                        conn->mxk_ntx_msgs++;
                } else if (msg_type == MXLND_MSG_CONN_REQ ||
                           msg_type == MXLND_MSG_CONN_ACK) {
                        /* put conn msgs at the front of the queue */
                        cfs_list_add(&tx->mxc_list, &conn->mxk_tx_free_queue);
                } else {
                        /* PUT_ACK, PUT_NAK */
                        cfs_list_add_tail(&tx->mxc_list,
                                          &conn->mxk_tx_free_queue);
                        conn->mxk_ntx_msgs++;
                }
        } else {
                /* data style tx */
                cfs_list_add_tail(&tx->mxc_list, &conn->mxk_tx_free_queue);
                conn->mxk_ntx_data++;
        }

        return;
}

/**
 * mxlnd_peer_queue_tx - add the tx to the global tx queue
 * @tx
 *
 * Add the tx to the peer's msg or data queue
 */
static inline void
mxlnd_peer_queue_tx(kmx_ctx_t *tx)
{
	LASSERT(tx->mxc_peer != NULL);
	LASSERT(tx->mxc_conn != NULL);
	spin_lock(&tx->mxc_conn->mxk_lock);
	mxlnd_peer_queue_tx_locked(tx);
	spin_unlock(&tx->mxc_conn->mxk_lock);

	return;
}

/**
 * mxlnd_queue_tx - add the tx to the global tx queue
 * @tx
 *
 * Add the tx to the global queue and up the tx_queue_sem
 */
void
mxlnd_queue_tx(kmx_ctx_t *tx)
{
        kmx_peer_t *peer   = tx->mxc_peer;
        LASSERT (tx->mxc_nid != 0);

        if (peer != NULL) {
                if (peer->mxp_incompatible &&
                    tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
                        /* let this fail now */
                        tx->mxc_errno = -ECONNABORTED;
                        mxlnd_conn_decref(peer->mxp_conn);
                        mxlnd_put_idle_tx(tx);
                        return;
                }
                if (tx->mxc_conn == NULL) {
                        int             ret     = 0;
                        kmx_conn_t      *conn   = NULL;

                        ret = mxlnd_conn_alloc(&conn, peer); /* adds 2nd ref for tx... */
                        if (ret != 0) {
                                tx->mxc_errno = ret;
                                mxlnd_put_idle_tx(tx);
                                goto done;
                        }
                        tx->mxc_conn = conn;
                        mxlnd_peer_decref(peer); /* and takes it from peer */
                }
                LASSERT(tx->mxc_conn != NULL);
                mxlnd_peer_queue_tx(tx);
                mxlnd_check_sends(peer);
        } else {
		spin_lock(&kmxlnd_data.kmx_tx_queue_lock);
		cfs_list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_queue);
		spin_unlock(&kmxlnd_data.kmx_tx_queue_lock);
		up(&kmxlnd_data.kmx_tx_queue_sem);
	}
done:
	return;
}

int
mxlnd_setup_iov(kmx_ctx_t *ctx, u32 niov, struct iovec *iov, u32 offset, u32 nob)
{
        int             i                       = 0;
        int             sum                     = 0;
        int             old_sum                 = 0;
        int             nseg                    = 0;
        int             first_iov               = -1;
        int             first_iov_offset        = 0;
        int             first_found             = 0;
        int             last_iov                = -1;
        int             last_iov_length         = 0;
        mx_ksegment_t  *seg                     = NULL;

        if (niov == 0) return 0;
        LASSERT(iov != NULL);

        for (i = 0; i < niov; i++) {
                sum = old_sum + (u32) iov[i].iov_len;
                if (!first_found && (sum > offset)) {
                        first_iov = i;
                        first_iov_offset = offset - old_sum;
                        first_found = 1;
                        sum = (u32) iov[i].iov_len - first_iov_offset;
                        old_sum = 0;
                }
                if (sum >= nob) {
                        last_iov = i;
                        last_iov_length = (u32) iov[i].iov_len - (sum - nob);
                        if (first_iov == last_iov) last_iov_length -= first_iov_offset;
                        break;
                }
                old_sum = sum;
        }
        LASSERT(first_iov >= 0 && last_iov >= first_iov);
        nseg = last_iov - first_iov + 1;
        LASSERT(nseg > 0);

        MXLND_ALLOC(seg, nseg * sizeof(*seg));
        if (seg == NULL) {
                CNETERR("MXLND_ALLOC() failed\n");
                return -1;
        }
        memset(seg, 0, nseg * sizeof(*seg));
        ctx->mxc_nseg = nseg;
        sum = 0;
        for (i = 0; i < nseg; i++) {
                seg[i].segment_ptr = MX_PA_TO_U64(virt_to_phys(iov[first_iov + i].iov_base));
                seg[i].segment_length = (u32) iov[first_iov + i].iov_len;
                if (i == 0) {
                        seg[i].segment_ptr += (u64) first_iov_offset;
                        seg[i].segment_length -= (u32) first_iov_offset;
                }
                if (i == (nseg - 1)) {
                        seg[i].segment_length = (u32) last_iov_length;
                }
                sum += seg[i].segment_length;
        }
        ctx->mxc_seg_list = seg;
        ctx->mxc_pin_type = MX_PIN_PHYSICAL;
#ifdef MX_PIN_FULLPAGES
        ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
#endif
        LASSERT(nob == sum);
        return 0;
}

int
mxlnd_setup_kiov(kmx_ctx_t *ctx, u32 niov, lnet_kiov_t *kiov, u32 offset, u32 nob)
{
        int             i                       = 0;
        int             sum                     = 0;
        int             old_sum                 = 0;
        int             nseg                    = 0;
        int             first_kiov              = -1;
        int             first_kiov_offset       = 0;
        int             first_found             = 0;
        int             last_kiov               = -1;
        int             last_kiov_length        = 0;
        mx_ksegment_t  *seg                     = NULL;

        if (niov == 0) return 0;
        LASSERT(kiov != NULL);

        for (i = 0; i < niov; i++) {
                sum = old_sum + kiov[i].kiov_len;
                if (i == 0) sum -= kiov[i].kiov_offset;
                if (!first_found && (sum > offset)) {
                        first_kiov = i;
                        first_kiov_offset = offset - old_sum;
                        if (i == 0) first_kiov_offset = kiov[i].kiov_offset;
                        first_found = 1;
                        sum = kiov[i].kiov_len - first_kiov_offset;
                        old_sum = 0;
                }
                if (sum >= nob) {
                        last_kiov = i;
                        last_kiov_length = kiov[i].kiov_len - (sum - nob);
                        if (first_kiov == last_kiov) last_kiov_length -= first_kiov_offset;
                        break;
                }
                old_sum = sum;
        }
        LASSERT(first_kiov >= 0 && last_kiov >= first_kiov);
        nseg = last_kiov - first_kiov + 1;
        LASSERT(nseg > 0);

        MXLND_ALLOC(seg, nseg * sizeof(*seg));
        if (seg == NULL) {
                CNETERR("MXLND_ALLOC() failed\n");
                return -1;
        }
        memset(seg, 0, niov * sizeof(*seg));
        ctx->mxc_nseg = niov;
        sum = 0;
        for (i = 0; i < niov; i++) {
                seg[i].segment_ptr = lnet_page2phys(kiov[first_kiov + i].kiov_page);
                seg[i].segment_length = kiov[first_kiov + i].kiov_len;
                if (i == 0) {
                        seg[i].segment_ptr += (u64) first_kiov_offset;
                        /* we have to add back the original kiov_offset */
                        seg[i].segment_length -= first_kiov_offset +
                                                 kiov[first_kiov].kiov_offset;
                }
                if (i == (nseg - 1)) {
                        seg[i].segment_length = last_kiov_length;
                }
                sum += seg[i].segment_length;
        }
        ctx->mxc_seg_list = seg;
        ctx->mxc_pin_type = MX_PIN_PHYSICAL;
#ifdef MX_PIN_FULLPAGES
        ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
#endif
        LASSERT(nob == sum);
        return 0;
}

void
mxlnd_send_nak(kmx_ctx_t *tx, lnet_nid_t nid, int type, int status, __u64 cookie)
{
        LASSERT(type == MXLND_MSG_PUT_ACK);
        mxlnd_init_tx_msg(tx, type, sizeof(kmx_putack_msg_t), tx->mxc_nid);
        tx->mxc_cookie = cookie;
        tx->mxc_msg->mxm_u.put_ack.mxpam_src_cookie = cookie;
        tx->mxc_msg->mxm_u.put_ack.mxpam_dst_cookie = ((u64) status << MXLND_ERROR_OFFSET); /* error code */
        tx->mxc_match = mxlnd_create_match(tx, status);

        mxlnd_queue_tx(tx);
}


/**
 * mxlnd_send_data - get tx, map [k]iov, queue tx
 * @ni
 * @lntmsg
 * @peer
 * @msg_type
 * @cookie
 *
 * This setups the DATA send for PUT or GET.
 *
 * On success, it queues the tx, on failure it calls lnet_finalize()
 */
void
mxlnd_send_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_peer_t *peer, u8 msg_type, u64 cookie)
{
        int                     ret     = 0;
        lnet_process_id_t       target  = lntmsg->msg_target;
        unsigned int            niov    = lntmsg->msg_niov;
        struct iovec           *iov     = lntmsg->msg_iov;
        lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
        unsigned int            offset  = lntmsg->msg_offset;
        unsigned int            nob     = lntmsg->msg_len;
        kmx_ctx_t              *tx      = NULL;

        LASSERT(lntmsg != NULL);
        LASSERT(peer != NULL);
        LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
        LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0);

        tx = mxlnd_get_idle_tx();
        if (tx == NULL) {
                CNETERR("Can't allocate %s tx for %s\n",
                        msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
                        libcfs_nid2str(target.nid));
                goto failed_0;
        }
        tx->mxc_nid = target.nid;
        /* NOTE called when we have a ref on the conn, get one for this tx */
        mxlnd_conn_addref(peer->mxp_conn);
        tx->mxc_peer = peer;
        tx->mxc_conn = peer->mxp_conn;
        tx->mxc_msg_type = msg_type;
        tx->mxc_lntmsg[0] = lntmsg;
        tx->mxc_cookie = cookie;
        tx->mxc_match = mxlnd_create_match(tx, 0);

        /* This setups up the mx_ksegment_t to send the DATA payload  */
        if (nob == 0) {
                /* do not setup the segments */
                CNETERR("nob = 0; why didn't we use an EAGER reply "
                        "to %s?\n", libcfs_nid2str(target.nid));
                ret = 0;
        } else if (kiov == NULL) {
                ret = mxlnd_setup_iov(tx, niov, iov, offset, nob);
        } else {
                ret = mxlnd_setup_kiov(tx, niov, kiov, offset, nob);
        }
        if (ret != 0) {
                CNETERR("Can't setup send DATA for %s\n",
                        libcfs_nid2str(target.nid));
                tx->mxc_errno = -EIO;
                goto failed_1;
        }
        mxlnd_queue_tx(tx);
        return;

failed_1:
        mxlnd_conn_decref(peer->mxp_conn);
        mxlnd_put_idle_tx(tx);
        return;

failed_0:
        CNETERR("no tx avail\n");
        lnet_finalize(ni, lntmsg, -EIO);
        return;
}

/**
 * mxlnd_recv_data - map [k]iov, post rx
 * @ni
 * @lntmsg
 * @rx
 * @msg_type
 * @cookie
 *
 * This setups the DATA receive for PUT or GET.
 *
 * On success, it returns 0, on failure it returns -1
 */
int
mxlnd_recv_data(lnet_ni_t *ni, lnet_msg_t *lntmsg, kmx_ctx_t *rx, u8 msg_type, u64 cookie)
{
        int                     ret     = 0;
        lnet_process_id_t       target  = lntmsg->msg_target;
        unsigned int            niov    = lntmsg->msg_niov;
        struct iovec           *iov     = lntmsg->msg_iov;
        lnet_kiov_t            *kiov    = lntmsg->msg_kiov;
        unsigned int            offset  = lntmsg->msg_offset;
        unsigned int            nob     = lntmsg->msg_len;
        mx_return_t             mxret   = MX_SUCCESS;
        u64                     mask    = ~(MXLND_ERROR_MASK);

        /* above assumes MXLND_MSG_PUT_DATA */
        if (msg_type == MXLND_MSG_GET_DATA) {
                niov = lntmsg->msg_md->md_niov;
                iov = lntmsg->msg_md->md_iov.iov;
                kiov = lntmsg->msg_md->md_iov.kiov;
                offset = 0;
                nob = lntmsg->msg_md->md_length;
        }

        LASSERT(lntmsg != NULL);
        LASSERT(rx != NULL);
        LASSERT(msg_type == MXLND_MSG_PUT_DATA || msg_type == MXLND_MSG_GET_DATA);
        LASSERT((cookie>>MXLND_ERROR_OFFSET) == 0); /* ensure top 12 bits are 0 */

        rx->mxc_msg_type = msg_type;
        rx->mxc_state = MXLND_CTX_PENDING;
        rx->mxc_nid = target.nid;
        /* if posting a GET_DATA, we may not yet know the peer */
        if (rx->mxc_peer != NULL) {
                rx->mxc_conn = rx->mxc_peer->mxp_conn;
        }
        rx->mxc_lntmsg[0] = lntmsg;
        rx->mxc_cookie = cookie;
        rx->mxc_match = mxlnd_create_match(rx, 0);
        /* This setups up the mx_ksegment_t to receive the DATA payload  */
        if (kiov == NULL) {
                ret = mxlnd_setup_iov(rx, niov, iov, offset, nob);
        } else {
                ret = mxlnd_setup_kiov(rx, niov, kiov, offset, nob);
        }
        if (msg_type == MXLND_MSG_GET_DATA) {
                rx->mxc_lntmsg[1] = lnet_create_reply_msg(kmxlnd_data.kmx_ni, lntmsg);
                if (rx->mxc_lntmsg[1] == NULL) {
                        CNETERR("Can't create reply for GET -> %s\n",
                                libcfs_nid2str(target.nid));
                        ret = -1;
                }
        }
        if (ret != 0) {
                CNETERR("Can't setup %s rx for %s\n",
                       msg_type == MXLND_MSG_PUT_DATA ? "PUT_DATA" : "GET_DATA",
                       libcfs_nid2str(target.nid));
                return -1;
        }
        ret = mxlnd_q_pending_ctx(rx);
        if (ret == -1) {
                return -1;
        }
        CDEBUG(D_NET, "receiving %s 0x%llx\n", mxlnd_msgtype_to_str(msg_type), rx->mxc_cookie);
        mxret = mx_kirecv(kmxlnd_data.kmx_endpt,
                          rx->mxc_seg_list, rx->mxc_nseg,
                          rx->mxc_pin_type, rx->mxc_match,
                          mask, (void *) rx,
                          &rx->mxc_mxreq);
        if (mxret != MX_SUCCESS) {
                if (rx->mxc_conn != NULL) {
                        mxlnd_deq_pending_ctx(rx);
                }
                CNETERR("mx_kirecv() failed with %d for %s\n",
                        (int) mxret, libcfs_nid2str(target.nid));
                return -1;
        }

        return 0;
}

/**
 * mxlnd_send - the LND required send function
 * @ni
 * @private
 * @lntmsg
 *
 * This must not block. Since we may not have a peer struct for the receiver,
 * it will append send messages on a global tx list. We will then up the
 * tx_queued's semaphore to notify it of the new send.
 */
int
mxlnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
{
        int                     ret             = 0;
        int                     type            = lntmsg->msg_type;
        lnet_hdr_t             *hdr             = &lntmsg->msg_hdr;
        lnet_process_id_t       target          = lntmsg->msg_target;
        lnet_nid_t              nid             = target.nid;
        int                     target_is_router = lntmsg->msg_target_is_router;
        int                     routing         = lntmsg->msg_routing;
        unsigned int            payload_niov    = lntmsg->msg_niov;
        struct iovec           *payload_iov     = lntmsg->msg_iov;
        lnet_kiov_t            *payload_kiov    = lntmsg->msg_kiov;
        unsigned int            payload_offset  = lntmsg->msg_offset;
        unsigned int            payload_nob     = lntmsg->msg_len;
        kmx_ctx_t              *tx              = NULL;
        kmx_msg_t              *txmsg           = NULL;
        kmx_ctx_t              *rx              = (kmx_ctx_t *) private; /* for REPLY */
        kmx_ctx_t              *rx_data         = NULL;
        kmx_conn_t             *conn            = NULL;
        int                     nob             = 0;
        uint32_t                length          = 0;
        kmx_peer_t             *peer            = NULL;
	rwlock_t		*g_lock		=&kmxlnd_data.kmx_global_lock;

        CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
                       payload_nob, payload_niov, libcfs_id2str(target));

        LASSERT (payload_nob == 0 || payload_niov > 0);
        LASSERT (payload_niov <= LNET_MAX_IOV);
        /* payload is either all vaddrs or all pages */
        LASSERT (!(payload_kiov != NULL && payload_iov != NULL));

        /* private is used on LNET_GET_REPLY only, NULL for all other cases */

        /* NOTE we may not know the peer if it is the very first PUT_REQ or GET_REQ
         * to a new peer, so create one if not found */
        peer = mxlnd_find_peer_by_nid(nid, 1); /* adds peer ref */
        if (peer == NULL || peer->mxp_conn == NULL) {
                /* we could not find it nor could we create one or
                 * one exists but we cannot create a conn,
                 * fail this message */
                if (peer) {
                        /* found peer without conn, drop ref taken above */
                        LASSERT(peer->mxp_conn == NULL);
                        mxlnd_peer_decref(peer);
                }
                return -ENOMEM;
        }

        /* we have a peer with a conn */

        if (unlikely(peer->mxp_incompatible)) {
                mxlnd_peer_decref(peer); /* drop ref taken above */
        } else {
		read_lock(g_lock);
		conn = peer->mxp_conn;
		if (conn && conn->mxk_status != MXLND_CONN_DISCONNECT)
			mxlnd_conn_addref(conn);
		else
			conn = NULL;
		read_unlock(g_lock);
                mxlnd_peer_decref(peer); /* drop peer ref taken above */
                if (!conn)
                        return -ENOTCONN;
        }

        LASSERT(peer && conn);

        CDEBUG(D_NET, "%s: peer 0x%llx is 0x%p\n", __func__, nid, peer);

        switch (type) {
        case LNET_MSG_ACK:
                LASSERT (payload_nob == 0);
                break;

        case LNET_MSG_REPLY:
        case LNET_MSG_PUT:
                /* Is the payload small enough not to need DATA? */
                nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob]);
                if (nob <= MXLND_MSG_SIZE)
                        break;                  /* send EAGER */

                tx = mxlnd_get_idle_tx();
                if (unlikely(tx == NULL)) {
                        CNETERR("Can't allocate %s tx for %s\n",
                                type == LNET_MSG_PUT ? "PUT" : "REPLY",
                                libcfs_nid2str(nid));
                        if (conn) mxlnd_conn_decref(conn);
                        return -ENOMEM;
                }

                tx->mxc_peer = peer;
                tx->mxc_conn = conn;
                /* we added a conn ref above */
                mxlnd_init_tx_msg (tx, MXLND_MSG_PUT_REQ, sizeof(kmx_putreq_msg_t), nid);
                txmsg = tx->mxc_msg;
                txmsg->mxm_u.put_req.mxprm_hdr = *hdr;
                txmsg->mxm_u.put_req.mxprm_cookie = tx->mxc_cookie;
                tx->mxc_match = mxlnd_create_match(tx, 0);

                /* we must post a receive _before_ sending the request.
                 * we need to determine how much to receive, it will be either
                 * a put_ack or a put_nak. The put_ack is larger, so use it. */

                rx = mxlnd_get_idle_rx(conn);
                if (unlikely(rx == NULL)) {
                        CNETERR("Can't allocate rx for PUT_ACK for %s\n",
                                libcfs_nid2str(nid));
                        mxlnd_put_idle_tx(tx);
                        if (conn) mxlnd_conn_decref(conn); /* for the ref taken above */
                        return -ENOMEM;
                }
                rx->mxc_nid = nid;
                rx->mxc_peer = peer;
                mxlnd_conn_addref(conn); /* for this rx */
                rx->mxc_conn = conn;
                rx->mxc_msg_type = MXLND_MSG_PUT_ACK;
                rx->mxc_cookie = tx->mxc_cookie;
                rx->mxc_match = mxlnd_create_match(rx, 0);

                length = offsetof(kmx_msg_t, mxm_u) + sizeof(kmx_putack_msg_t);
                ret = mxlnd_recv_msg(lntmsg, rx, MXLND_MSG_PUT_ACK, rx->mxc_match, length);
                if (unlikely(ret != 0)) {
                        CNETERR("recv_msg() failed for PUT_ACK for %s\n",
                                           libcfs_nid2str(nid));
                        rx->mxc_lntmsg[0] = NULL;
                        mxlnd_put_idle_rx(rx);
                        mxlnd_put_idle_tx(tx);
                        mxlnd_conn_decref(conn); /* for the rx... */
                        mxlnd_conn_decref(conn); /* and for the tx */
                        return -EHOSTUNREACH;
                }

                mxlnd_queue_tx(tx);
                return 0;

        case LNET_MSG_GET:
                if (routing || target_is_router)
                        break;                  /* send EAGER */

                /* is the REPLY message too small for DATA? */
                nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[lntmsg->msg_md->md_length]);
                if (nob <= MXLND_MSG_SIZE)
                        break;                  /* send EAGER */

                /* get tx (we need the cookie) , post rx for incoming DATA,
                 * then post GET_REQ tx */
                tx = mxlnd_get_idle_tx();
                if (unlikely(tx == NULL)) {
                        CNETERR("Can't allocate GET tx for %s\n",
                                libcfs_nid2str(nid));
                        mxlnd_conn_decref(conn); /* for the ref taken above */
                        return -ENOMEM;
                }
                rx_data = mxlnd_get_idle_rx(conn);
                if (unlikely(rx_data == NULL)) {
                        CNETERR("Can't allocate DATA rx for %s\n",
                                libcfs_nid2str(nid));
                        mxlnd_put_idle_tx(tx);
                        mxlnd_conn_decref(conn); /* for the ref taken above */
                        return -ENOMEM;
                }
                rx_data->mxc_peer = peer;
                /* NOTE no need to lock peer before adding conn ref since we took
                 * a conn ref for the tx (it cannot be freed between there and here ) */
                mxlnd_conn_addref(conn); /* for the rx_data */
                rx_data->mxc_conn = conn;

                ret = mxlnd_recv_data(ni, lntmsg, rx_data, MXLND_MSG_GET_DATA, tx->mxc_cookie);
                if (unlikely(ret != 0)) {
                        CNETERR("Can't setup GET sink for %s\n",
                                libcfs_nid2str(nid));
                        mxlnd_put_idle_rx(rx_data);
                        mxlnd_put_idle_tx(tx);
                        mxlnd_conn_decref(conn); /* for the rx_data... */
                        mxlnd_conn_decref(conn); /* and for the tx */
                        return -EIO;
                }

                tx->mxc_peer = peer;
                tx->mxc_conn = conn;
                /* conn ref taken above */
                mxlnd_init_tx_msg(tx, MXLND_MSG_GET_REQ, sizeof(kmx_getreq_msg_t), nid);
                txmsg = tx->mxc_msg;
                txmsg->mxm_u.get_req.mxgrm_hdr = *hdr;
                txmsg->mxm_u.get_req.mxgrm_cookie = tx->mxc_cookie;
                tx->mxc_match = mxlnd_create_match(tx, 0);

                mxlnd_queue_tx(tx);
                return 0;

        default:
                LBUG();
                mxlnd_conn_decref(conn); /* drop ref taken above */
                return -EIO;
        }

        /* send EAGER */

        LASSERT (offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[payload_nob])
                <= MXLND_MSG_SIZE);

        tx = mxlnd_get_idle_tx();
        if (unlikely(tx == NULL)) {
                CNETERR("Can't send %s to %s: tx descs exhausted\n",
                        mxlnd_lnetmsg_to_str(type), libcfs_nid2str(nid));
                mxlnd_conn_decref(conn); /* drop ref taken above */
                return -ENOMEM;
        }

        tx->mxc_peer = peer;
        tx->mxc_conn = conn;
        /* conn ref taken above */
        nob = offsetof(kmx_eager_msg_t, mxem_payload[payload_nob]);
        mxlnd_init_tx_msg (tx, MXLND_MSG_EAGER, nob, nid);
        tx->mxc_match = mxlnd_create_match(tx, 0);

        txmsg = tx->mxc_msg;
        txmsg->mxm_u.eager.mxem_hdr = *hdr;

        if (payload_kiov != NULL)
                lnet_copy_kiov2flat(MXLND_MSG_SIZE, txmsg,
                            offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
                            payload_niov, payload_kiov, payload_offset, payload_nob);
        else
                lnet_copy_iov2flat(MXLND_MSG_SIZE, txmsg,
                            offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
                            payload_niov, payload_iov, payload_offset, payload_nob);

        tx->mxc_lntmsg[0] = lntmsg;              /* finalise lntmsg on completion */
        mxlnd_queue_tx(tx);
        return 0;
}

/**
 * mxlnd_recv - the LND required recv function
 * @ni
 * @private
 * @lntmsg
 * @delayed
 * @niov
 * @kiov
 * @offset
 * @mlen
 * @rlen
 *
 * This must not block.
 */
int
mxlnd_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
             unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
             unsigned int offset, unsigned int mlen, unsigned int rlen)
{
        int             ret             = 0;
        int             nob             = 0;
        int             len             = 0;
        kmx_ctx_t       *rx             = private;
        kmx_msg_t       *rxmsg          = rx->mxc_msg;
        lnet_nid_t       nid            = rx->mxc_nid;
        kmx_ctx_t       *tx             = NULL;
        kmx_msg_t       *txmsg          = NULL;
        kmx_peer_t      *peer           = rx->mxc_peer;
        kmx_conn_t      *conn           = peer->mxp_conn;
        u64              cookie         = 0ULL;
        int              msg_type       = rxmsg->mxm_type;
        int              repost         = 1;
        int              credit         = 0;
        int              finalize       = 0;

        LASSERT (mlen <= rlen);
        /* Either all pages or all vaddrs */
        LASSERT (!(kiov != NULL && iov != NULL));
        LASSERT (peer && conn);

        /* conn_addref(conn) already taken for the primary rx */

        switch (msg_type) {
        case MXLND_MSG_EAGER:
                nob = offsetof(kmx_msg_t, mxm_u.eager.mxem_payload[rlen]);
                len = rx->mxc_status.xfer_length;
                if (unlikely(nob > len)) {
                        CNETERR("Eager message from %s too big: %d(%d)\n",
                                libcfs_nid2str(nid), nob, len);
                        ret = -EPROTO;
                        break;
                }

                if (kiov != NULL)
                        lnet_copy_flat2kiov(niov, kiov, offset,
                                MXLND_MSG_SIZE, rxmsg,
                                offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
                                mlen);
                else
                        lnet_copy_flat2iov(niov, iov, offset,
                                MXLND_MSG_SIZE, rxmsg,
                                offsetof(kmx_msg_t, mxm_u.eager.mxem_payload),
                                mlen);
                finalize = 1;
                credit = 1;
                break;

        case MXLND_MSG_PUT_REQ:
                /* we are going to reuse the rx, store the needed info */
                cookie = rxmsg->mxm_u.put_req.mxprm_cookie;

                /* get tx, post rx, send PUT_ACK */

                tx = mxlnd_get_idle_tx();
                if (unlikely(tx == NULL)) {
                        CNETERR("Can't allocate tx for %s\n", libcfs_nid2str(nid));
                        /* Not replying will break the connection */
                        ret = -ENOMEM;
                        break;
                }
                if (unlikely(mlen == 0)) {
                        finalize = 1;
                        tx->mxc_peer = peer;
                        tx->mxc_conn = conn;
                        mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, 0, cookie);
                        /* repost = 1 */
                        break;
                }

                mxlnd_init_tx_msg(tx, MXLND_MSG_PUT_ACK, sizeof(kmx_putack_msg_t), nid);
                tx->mxc_peer = peer;
                tx->mxc_conn = conn;
                /* no need to lock peer first since we already have a ref */
                mxlnd_conn_addref(conn); /* for the tx */
                txmsg = tx->mxc_msg;
                txmsg->mxm_u.put_ack.mxpam_src_cookie = cookie;
                txmsg->mxm_u.put_ack.mxpam_dst_cookie = tx->mxc_cookie;
                tx->mxc_cookie = cookie;
                tx->mxc_match = mxlnd_create_match(tx, 0);

                /* we must post a receive _before_ sending the PUT_ACK */
                mxlnd_ctx_init(rx);
                rx->mxc_state = MXLND_CTX_PREP;
                rx->mxc_peer = peer;
                rx->mxc_conn = conn;
                /* do not take another ref for this rx, it is already taken */
                rx->mxc_nid = peer->mxp_nid;
                ret = mxlnd_recv_data(ni, lntmsg, rx, MXLND_MSG_PUT_DATA,
                                      txmsg->mxm_u.put_ack.mxpam_dst_cookie);

                if (unlikely(ret != 0)) {
                        /* Notify peer that it's over */
                        CNETERR("Can't setup PUT_DATA rx for %s: %d\n",
                                libcfs_nid2str(nid), ret);
                        mxlnd_ctx_init(tx);
                        tx->mxc_state = MXLND_CTX_PREP;
                        tx->mxc_peer = peer;
                        tx->mxc_conn = conn;
                        /* finalize = 0, let the PUT_ACK tx finalize this */
                        tx->mxc_lntmsg[0] = rx->mxc_lntmsg[0];
                        tx->mxc_lntmsg[1] = rx->mxc_lntmsg[1];
                        /* conn ref already taken above */
                        mxlnd_send_nak(tx, nid, MXLND_MSG_PUT_ACK, ret, cookie);
                        /* repost = 1 */
                        break;
                }

                mxlnd_queue_tx(tx);
                /* do not return a credit until after PUT_DATA returns */
                repost = 0;
                break;

        case MXLND_MSG_GET_REQ:
                cookie = rxmsg->mxm_u.get_req.mxgrm_cookie;

                if (likely(lntmsg != NULL)) {
                        mxlnd_send_data(ni, lntmsg, rx->mxc_peer, MXLND_MSG_GET_DATA,
                                        cookie);
                } else {
                        /* GET didn't match anything */
                        /* The initiator has a rx mapped to [k]iov. We cannot send a nak.
                         * We have to embed the error code in the match bits.
                         * Send the error in bits 52-59 and the cookie in bits 0-51 */
                        tx = mxlnd_get_idle_tx();
                        if (unlikely(tx == NULL)) {
                                CNETERR("Can't get tx for GET NAK for %s\n",
                                        libcfs_nid2str(nid));
                                /* we can't get a tx, notify the peer that the GET failed */
                                mxlnd_send_message(conn->mxk_epa, MXLND_MSG_GET_DATA,
                                                   ENODATA, cookie);
                                ret = -ENOMEM;
                                break;
                        }
                        tx->mxc_msg_type = MXLND_MSG_GET_DATA;
                        tx->mxc_state = MXLND_CTX_PENDING;
                        tx->mxc_nid = nid;
                        tx->mxc_peer = peer;
                        tx->mxc_conn = conn;
                        /* no need to lock peer first since we already have a ref */
                        mxlnd_conn_addref(conn); /* for this tx */
                        tx->mxc_cookie = cookie;
                        tx->mxc_match = mxlnd_create_match(tx, ENODATA);
                        tx->mxc_pin_type = MX_PIN_PHYSICAL;
                        mxlnd_queue_tx(tx);
                }
                /* finalize lntmsg after tx completes */
                break;

        default:
                LBUG();
        }

        if (repost) {
                /* we received a message, increment peer's outstanding credits */
		if (credit == 1) {
			spin_lock(&conn->mxk_lock);
			conn->mxk_outstanding++;
			spin_unlock(&conn->mxk_lock);
		}
                /* we are done with the rx */
                mxlnd_put_idle_rx(rx);
                mxlnd_conn_decref(conn);
        }

        if (finalize == 1) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg, 0);

        /* we received a credit, see if we can use it to send a msg */
        if (credit) mxlnd_check_sends(peer);

        return ret;
}

void
mxlnd_sleep(unsigned long timeout)
{
	set_current_state(TASK_INTERRUPTIBLE);
	schedule_timeout(timeout);
	return;
}

/**
 * mxlnd_tx_queued - the generic send queue thread
 * @arg - thread id (as a void *)
 *
 * This thread moves send messages from the global tx_queue to the owning
 * peer's tx_[msg|data]_queue. If the peer does not exist, it creates one and adds
 * it to the global peer list.
 */
int
mxlnd_tx_queued(void *arg)
{
        long                    id      = (long) arg;
        int                     ret     = 0;
        int                     found   = 0;
        kmx_ctx_t              *tx      = NULL;
        kmx_peer_t             *peer    = NULL;
        cfs_list_t             *queue   = &kmxlnd_data.kmx_tx_queue;
	spinlock_t		*tx_q_lock = &kmxlnd_data.kmx_tx_queue_lock;
	rwlock_t		*g_lock  = &kmxlnd_data.kmx_global_lock;

	while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
		ret = down_interruptible(&kmxlnd_data.kmx_tx_queue_sem);
		if (atomic_read(&kmxlnd_data.kmx_shutdown))
			break;
		if (ret != 0) /* Should we check for -EINTR? */
			continue;
		spin_lock(tx_q_lock);
		if (cfs_list_empty(&kmxlnd_data.kmx_tx_queue)) {
			spin_unlock(tx_q_lock);
			continue;
		}
		tx = cfs_list_entry(queue->next, kmx_ctx_t, mxc_list);
		cfs_list_del_init(&tx->mxc_list);
		spin_unlock(tx_q_lock);

		found = 0;
		peer = mxlnd_find_peer_by_nid(tx->mxc_nid, 0); /* adds ref*/
		if (peer != NULL) {
			tx->mxc_peer = peer;
			write_lock(g_lock);
			if (peer->mxp_conn == NULL) {
				ret = mxlnd_conn_alloc_locked(&peer->mxp_conn,
							      peer);
				if (ret != 0) {
					/* out of memory: give up, fail tx */
					tx->mxc_errno = -ENOMEM;
					mxlnd_peer_decref(peer);
					write_unlock(g_lock);
					mxlnd_put_idle_tx(tx);
					continue;
				}
			}
			tx->mxc_conn = peer->mxp_conn;
			mxlnd_conn_addref(tx->mxc_conn); /* for this tx */
			mxlnd_peer_decref(peer); /* drop peer ref taken above */
			write_unlock(g_lock);
                        mxlnd_queue_tx(tx);
                        found = 1;
                }
                if (found == 0) {
                        int             hash    = 0;
                        kmx_peer_t     *peer    = NULL;
                        kmx_peer_t     *old     = NULL;

                        hash = mxlnd_nid_to_hash(tx->mxc_nid);

                        LASSERT(tx->mxc_msg_type != MXLND_MSG_PUT_DATA &&
                                tx->mxc_msg_type != MXLND_MSG_GET_DATA);
                        /* create peer */
                        /* adds conn ref for this function */
                        ret = mxlnd_peer_alloc(&peer, tx->mxc_nid,
                                        *kmxlnd_tunables.kmx_board,
                                        *kmxlnd_tunables.kmx_ep_id, 0ULL);
                        if (ret != 0) {
                                /* finalize message */
                                tx->mxc_errno = ret;
                                mxlnd_put_idle_tx(tx);
                                continue;
                        }
                        tx->mxc_peer = peer;
                        tx->mxc_conn = peer->mxp_conn;
                        /* this tx will keep the conn ref taken in peer_alloc() */

                        /* add peer to global peer list, but look to see
                         * if someone already created it after we released
                         * the read lock */
			write_lock(g_lock);
                        old = mxlnd_find_peer_by_nid_locked(peer->mxp_nid);
                        if (old) {
                                /* we have a peer ref on old */
                                if (old->mxp_conn) {
                                        found = 1;
                                } else {
                                        /* no conn */
                                        /* drop our ref taken above... */
                                        mxlnd_peer_decref(old);
                                        /* and delete it */
                                        mxlnd_del_peer_locked(old);
                                }
                        }

			if (found == 0) {
				cfs_list_add_tail(&peer->mxp_list,
						  &kmxlnd_data.kmx_peers[hash]);
				atomic_inc(&kmxlnd_data.kmx_npeers);
			} else {
                                tx->mxc_peer = old;
                                tx->mxc_conn = old->mxp_conn;
                                LASSERT(old->mxp_conn != NULL);
                                mxlnd_conn_addref(old->mxp_conn);
                                mxlnd_conn_decref(peer->mxp_conn); /* drop ref taken above.. */
                                mxlnd_conn_decref(peer->mxp_conn); /* drop peer's ref */
                                mxlnd_peer_decref(peer);
                        }
			write_unlock(g_lock);

                        mxlnd_queue_tx(tx);
                }
        }
        mxlnd_thread_stop(id);
        return 0;
}

/* When calling this, we must not have the peer lock. */
void
mxlnd_iconnect(kmx_peer_t *peer, u8 msg_type)
{
        mx_return_t     mxret           = MX_SUCCESS;
        mx_request_t    request;
        kmx_conn_t      *conn           = peer->mxp_conn;
        u64             match           = ((u64) msg_type) << MXLND_MSG_OFFSET;

        /* NOTE we are holding a conn ref every time we call this function,
         * we do not need to lock the peer before taking another ref */
        mxlnd_conn_addref(conn); /* hold until CONN_REQ or CONN_ACK completes */

        LASSERT(msg_type == MXLND_MSG_ICON_REQ || msg_type == MXLND_MSG_ICON_ACK);

        if (peer->mxp_reconnect_time == 0) {
                peer->mxp_reconnect_time = jiffies;
        }

        if (peer->mxp_nic_id == 0ULL) {
                int     ret     = 0;

                ret = mxlnd_ip2nic_id(LNET_NIDADDR(peer->mxp_nid),
                                      &peer->mxp_nic_id, MXLND_LOOKUP_COUNT);
                if (ret == 0) {
                        mx_nic_id_to_board_number(peer->mxp_nic_id, &peer->mxp_board);
                }
                if (peer->mxp_nic_id == 0ULL && conn->mxk_status == MXLND_CONN_WAIT) {
			/* not mapped yet, return */
			spin_lock(&conn->mxk_lock);
			mxlnd_set_conn_status(conn, MXLND_CONN_INIT);
			spin_unlock(&conn->mxk_lock);
                }
        }

        if (cfs_time_after(jiffies,
                           peer->mxp_reconnect_time + MXLND_CONNECT_TIMEOUT) &&
            conn->mxk_status != MXLND_CONN_DISCONNECT) {
                /* give up and notify LNET */
                CDEBUG(D_NET, "timeout trying to connect to %s\n",
                       libcfs_nid2str(peer->mxp_nid));
                mxlnd_conn_disconnect(conn, 0, 0);
                mxlnd_conn_decref(conn);
                return;
        }

        mxret = mx_iconnect(kmxlnd_data.kmx_endpt, peer->mxp_nic_id,
                            peer->mxp_ep_id, MXLND_MSG_MAGIC, match,
                            (void *) peer, &request);
        if (unlikely(mxret != MX_SUCCESS)) {
		spin_lock(&conn->mxk_lock);
		mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
		spin_unlock(&conn->mxk_lock);
                CNETERR("mx_iconnect() failed with %s (%d) to %s\n",
                       mx_strerror(mxret), mxret, libcfs_nid2str(peer->mxp_nid));
                mxlnd_conn_decref(conn);
        }
	mx_set_request_timeout(kmxlnd_data.kmx_endpt, request,
			       jiffies_to_msecs(MXLND_CONNECT_TIMEOUT));
	return;
}

#define MXLND_STATS 0

int
mxlnd_check_sends(kmx_peer_t *peer)
{
        int             ret             = 0;
        int             found           = 0;
        mx_return_t     mxret           = MX_SUCCESS;
        kmx_ctx_t       *tx             = NULL;
        kmx_conn_t      *conn           = NULL;
        u8              msg_type        = 0;
        int             credit          = 0;
        int             status          = 0;
        int             ntx_posted      = 0;
        int             credits         = 0;
#if MXLND_STATS
        static unsigned long last       = 0;
#endif

        if (unlikely(peer == NULL)) {
                LASSERT(peer != NULL);
                return -1;
        }
	write_lock(&kmxlnd_data.kmx_global_lock);
	conn = peer->mxp_conn;
	/* NOTE take a ref for the duration of this function since it is
	 * called when there might not be any queued txs for this peer */
	if (conn) {
		if (conn->mxk_status == MXLND_CONN_DISCONNECT) {
			write_unlock(&kmxlnd_data.kmx_global_lock);
			return -1;
		}
		mxlnd_conn_addref(conn); /* for duration of this function */
	}
	write_unlock(&kmxlnd_data.kmx_global_lock);

        /* do not add another ref for this tx */

        if (conn == NULL) {
                /* we do not have any conns */
                CNETERR("peer %s has no conn\n", libcfs_nid2str(peer->mxp_nid));
                return -1;
        }

#if MXLND_STATS
	if (cfs_time_after(jiffies, last)) {
		last = jiffies + msecs_to_jiffies(MSEC_PER_SEC);
		CDEBUG(D_NET, "status= %s credits= %d outstanding= %d ntx_msgs= %d "
			      "ntx_posted= %d ntx_data= %d data_posted= %d\n",
			      mxlnd_connstatus_to_str(conn->mxk_status), conn->mxk_credits,
			      conn->mxk_outstanding, conn->mxk_ntx_msgs, conn->mxk_ntx_posted,
			      conn->mxk_ntx_data, conn->mxk_data_posted);
	}
#endif

	spin_lock(&conn->mxk_lock);
        ntx_posted = conn->mxk_ntx_posted;
        credits = conn->mxk_credits;

        LASSERT(ntx_posted <= *kmxlnd_tunables.kmx_peercredits);
        LASSERT(ntx_posted >= 0);

        LASSERT(credits <= *kmxlnd_tunables.kmx_peercredits);
        LASSERT(credits >= 0);

        /* check number of queued msgs, ignore data */
        if (conn->mxk_outstanding >= MXLND_CREDIT_HIGHWATER()) {
                /* check if any txs queued that could return credits... */
                if (cfs_list_empty(&conn->mxk_tx_credit_queue) ||
                    conn->mxk_ntx_msgs == 0) {
                        /* if not, send a NOOP */
                        tx = mxlnd_get_idle_tx();
                        if (likely(tx != NULL)) {
                                tx->mxc_peer = peer;
                                tx->mxc_conn = peer->mxp_conn;
                                mxlnd_conn_addref(conn); /* for this tx */
                                mxlnd_init_tx_msg (tx, MXLND_MSG_NOOP, 0, peer->mxp_nid);
                                tx->mxc_match = mxlnd_create_match(tx, 0);
                                mxlnd_peer_queue_tx_locked(tx);
                                found = 1;
                                goto done_locked;
                        }
                }
        }

        /* if the peer is not ready, try to connect */
        if (unlikely(conn->mxk_status == MXLND_CONN_INIT ||
            conn->mxk_status == MXLND_CONN_FAIL)) {
                CDEBUG(D_NET, "status=%s\n", mxlnd_connstatus_to_str(conn->mxk_status));
                mxlnd_set_conn_status(conn, MXLND_CONN_WAIT);
		spin_unlock(&conn->mxk_lock);
                mxlnd_iconnect(peer, (u8) MXLND_MSG_ICON_REQ);
                goto done;
        }

        while (!cfs_list_empty(&conn->mxk_tx_free_queue) ||
               !cfs_list_empty(&conn->mxk_tx_credit_queue)) {
                /* We have something to send. If we have a queued tx that does not
                 * require a credit (free), choose it since its completion will
                 * return a credit (here or at the peer), complete a DATA or
                 * CONN_REQ or CONN_ACK. */
                cfs_list_t *tmp_tx = NULL;
                if (!cfs_list_empty(&conn->mxk_tx_free_queue)) {
                        tmp_tx = &conn->mxk_tx_free_queue;
                } else {
                        tmp_tx = &conn->mxk_tx_credit_queue;
                }
                tx = cfs_list_entry(tmp_tx->next, kmx_ctx_t, mxc_list);

                msg_type = tx->mxc_msg_type;

                /* don't try to send a rx */
                LASSERT(tx->mxc_type == MXLND_REQ_TX);

                /* ensure that it is a valid msg type */
                LASSERT(msg_type == MXLND_MSG_CONN_REQ ||
                        msg_type == MXLND_MSG_CONN_ACK ||
                        msg_type == MXLND_MSG_NOOP     ||
                        msg_type == MXLND_MSG_EAGER    ||
                        msg_type == MXLND_MSG_PUT_REQ  ||
                        msg_type == MXLND_MSG_PUT_ACK  ||
                        msg_type == MXLND_MSG_PUT_DATA ||
                        msg_type == MXLND_MSG_GET_REQ  ||
                        msg_type == MXLND_MSG_GET_DATA);
                LASSERT(tx->mxc_peer == peer);
                LASSERT(tx->mxc_nid == peer->mxp_nid);

                credit = mxlnd_tx_requires_credit(tx);
                if (credit) {

                        if (conn->mxk_ntx_posted == *kmxlnd_tunables.kmx_peercredits) {
                                CDEBUG(D_NET, "%s: posted enough\n",
                                              libcfs_nid2str(peer->mxp_nid));
                                goto done_locked;
                        }

                        if (conn->mxk_credits == 0) {
                                CDEBUG(D_NET, "%s: no credits\n",
                                              libcfs_nid2str(peer->mxp_nid));
                                goto done_locked;
                        }

                        if (conn->mxk_credits == 1 &&      /* last credit reserved for */
                            conn->mxk_outstanding == 0) {  /* giving back credits */
                                CDEBUG(D_NET, "%s: not using last credit\n",
                                              libcfs_nid2str(peer->mxp_nid));
                                goto done_locked;
                        }
                }

                if (unlikely(conn->mxk_status != MXLND_CONN_READY)) {
                        if ( ! (msg_type == MXLND_MSG_CONN_REQ ||
                                msg_type == MXLND_MSG_CONN_ACK)) {
                                CDEBUG(D_NET, "peer status is %s for tx 0x%llx (%s)\n",
                                             mxlnd_connstatus_to_str(conn->mxk_status),
                                             tx->mxc_cookie,
                                             mxlnd_msgtype_to_str(tx->mxc_msg_type));
                                if (conn->mxk_status == MXLND_CONN_DISCONNECT ||
                                    cfs_time_aftereq(jiffies, tx->mxc_deadline)) {
                                        cfs_list_del_init(&tx->mxc_list);
                                        tx->mxc_errno = -ECONNABORTED;
					spin_unlock(&conn->mxk_lock);
                                        mxlnd_put_idle_tx(tx);
                                        mxlnd_conn_decref(conn);
                                        goto done;
                                }
                                goto done_locked;
                        }
                }

                cfs_list_del_init(&tx->mxc_list);

                /* handle credits, etc now while we have the lock to avoid races */
                if (credit) {
                        conn->mxk_credits--;
                        conn->mxk_ntx_posted++;
                }
                if (msg_type != MXLND_MSG_PUT_DATA &&
                    msg_type != MXLND_MSG_GET_DATA) {
                        if (msg_type != MXLND_MSG_CONN_REQ &&
                            msg_type != MXLND_MSG_CONN_ACK) {
                                conn->mxk_ntx_msgs--;
                        }
                }
                if (tx->mxc_incarnation == 0 &&
                    conn->mxk_incarnation != 0) {
                        tx->mxc_incarnation = conn->mxk_incarnation;
                }

                /* if this is a NOOP and (1) mxp_conn->mxk_outstanding < CREDIT_HIGHWATER
                 * or (2) there is a non-DATA msg that can return credits in the
                 * queue, then drop this duplicate NOOP */
                if (unlikely(msg_type == MXLND_MSG_NOOP)) {
                        if ((conn->mxk_outstanding < MXLND_CREDIT_HIGHWATER()) ||
                            (conn->mxk_ntx_msgs >= 1)) {
                                conn->mxk_credits++;
                                conn->mxk_ntx_posted--;
				spin_unlock(&conn->mxk_lock);
                                /* redundant NOOP */
                                mxlnd_put_idle_tx(tx);
                                mxlnd_conn_decref(conn);
                                CDEBUG(D_NET, "%s: redundant noop\n",
                                              libcfs_nid2str(peer->mxp_nid));
                                found = 1;
                                goto done;
                        }
                }

                found = 1;
                if (likely((msg_type != MXLND_MSG_PUT_DATA) &&
                    (msg_type != MXLND_MSG_GET_DATA))) {
                        mxlnd_pack_msg_locked(tx);
                }

                mxret = MX_SUCCESS;

                status = conn->mxk_status;
		spin_unlock(&conn->mxk_lock);

                if (likely((status == MXLND_CONN_READY) ||
                    (msg_type == MXLND_MSG_CONN_REQ) ||
                    (msg_type == MXLND_MSG_CONN_ACK))) {
                        ret = 0;
                        if (msg_type != MXLND_MSG_CONN_REQ &&
                            msg_type != MXLND_MSG_CONN_ACK) {
                                /* add to the pending list */
                                ret = mxlnd_q_pending_ctx(tx);
                        } else {
                                /* CONN_REQ/ACK */
                                tx->mxc_state = MXLND_CTX_PENDING;
                        }

                        if (ret == 0) {
                                if (likely(msg_type != MXLND_MSG_PUT_DATA &&
                                    msg_type != MXLND_MSG_GET_DATA)) {
                                        /* send a msg style tx */
                                        LASSERT(tx->mxc_nseg == 1);
                                        LASSERT(tx->mxc_pin_type == MX_PIN_PHYSICAL);
                                        CDEBUG(D_NET, "sending %s 0x%llx\n",
                                               mxlnd_msgtype_to_str(msg_type),
                                               tx->mxc_cookie);
                                        mxret = mx_kisend(kmxlnd_data.kmx_endpt,
                                                          &tx->mxc_seg,
                                                          tx->mxc_nseg,
                                                          tx->mxc_pin_type,
                                                          conn->mxk_epa,
                                                          tx->mxc_match,
                                                          (void *) tx,
                                                          &tx->mxc_mxreq);
                                } else {
                                        /* send a DATA tx */
					spin_lock(&conn->mxk_lock);
					conn->mxk_ntx_data--;
					conn->mxk_data_posted++;
					spin_unlock(&conn->mxk_lock);
                                        CDEBUG(D_NET, "sending %s 0x%llx\n",
                                               mxlnd_msgtype_to_str(msg_type),
                                               tx->mxc_cookie);
                                        mxret = mx_kisend(kmxlnd_data.kmx_endpt,
                                                          tx->mxc_seg_list,
                                                          tx->mxc_nseg,
                                                          tx->mxc_pin_type,
                                                          conn->mxk_epa,
                                                          tx->mxc_match,
                                                          (void *) tx,
                                                          &tx->mxc_mxreq);
                                }
                        } else {
                                /* ret != 0 */
                                mxret = MX_CONNECTION_FAILED;
                        }
                        if (likely(mxret == MX_SUCCESS)) {
                                ret = 0;
                        } else {
                                CNETERR("mx_kisend() failed with %s (%d) "
                                        "sending to %s\n", mx_strerror(mxret), (int) mxret,
                                       libcfs_nid2str(peer->mxp_nid));
                                /* NOTE mx_kisend() only fails if there are not enough
                                * resources. Do not change the connection status. */
                                if (mxret == MX_NO_RESOURCES) {
                                        tx->mxc_errno = -ENOMEM;
                                } else {
                                        tx->mxc_errno = -ECONNABORTED;
                                }
                                if (credit) {
					spin_lock(&conn->mxk_lock);
					conn->mxk_ntx_posted--;
					conn->mxk_credits++;
					spin_unlock(&conn->mxk_lock);
				} else if (msg_type == MXLND_MSG_PUT_DATA ||
					   msg_type == MXLND_MSG_GET_DATA) {
					spin_lock(&conn->mxk_lock);
					conn->mxk_data_posted--;
					spin_unlock(&conn->mxk_lock);
				}
				if (msg_type != MXLND_MSG_PUT_DATA &&
				    msg_type != MXLND_MSG_GET_DATA &&
				    msg_type != MXLND_MSG_CONN_REQ &&
				    msg_type != MXLND_MSG_CONN_ACK) {
					spin_lock(&conn->mxk_lock);
					conn->mxk_outstanding +=
						tx->mxc_msg->mxm_credits;
					spin_unlock(&conn->mxk_lock);
                                }
                                if (msg_type != MXLND_MSG_CONN_REQ &&
                                    msg_type != MXLND_MSG_CONN_ACK) {
                                        /* remove from the pending list */
                                        mxlnd_deq_pending_ctx(tx);
                                }
                                mxlnd_put_idle_tx(tx);
                                mxlnd_conn_decref(conn);
                        }
                }
		spin_lock(&conn->mxk_lock);
	}
done_locked:
	spin_unlock(&conn->mxk_lock);
done:
	mxlnd_conn_decref(conn); /* drop ref taken at start of function */
	return found;
}


/**
 * mxlnd_handle_tx_completion - a tx completed, progress or complete the msg
 * @ctx - the tx descriptor
 *
 * Determine which type of send request it was and start the next step, if needed,
 * or, if done, signal completion to LNET. After we are done, put back on the
 * idle tx list.
 */
void
mxlnd_handle_tx_completion(kmx_ctx_t *tx)
{
        int             code    = tx->mxc_status.code;
        int             failed  = (code != MX_STATUS_SUCCESS || tx->mxc_errno != 0);
        kmx_msg_t       *msg    = tx->mxc_msg;
        kmx_peer_t      *peer   = tx->mxc_peer;
        kmx_conn_t      *conn   = tx->mxc_conn;
        u8              type    = tx->mxc_msg_type;
        int             credit  = mxlnd_tx_requires_credit(tx);
        u64             cookie  = tx->mxc_cookie;

        CDEBUG(D_NET, "entering %s (0x%llx):\n",
                      mxlnd_msgtype_to_str(tx->mxc_msg_type), cookie);

        LASSERT (peer != NULL);
        LASSERT (conn != NULL);

        if (type != MXLND_MSG_PUT_DATA && type != MXLND_MSG_GET_DATA) {
                LASSERT (type == msg->mxm_type);
        }

        if (failed) {
                if (tx->mxc_errno == 0) tx->mxc_errno = -EIO;
        } else {
		spin_lock(&conn->mxk_lock);
		conn->mxk_last_tx = cfs_time_current(); /* jiffies */
		spin_unlock(&conn->mxk_lock);
	}

	switch (type) {

	case MXLND_MSG_GET_DATA:
		spin_lock(&conn->mxk_lock);
		if (conn->mxk_incarnation == tx->mxc_incarnation) {
			conn->mxk_outstanding++;
			conn->mxk_data_posted--;
		}
		spin_unlock(&conn->mxk_lock);
		break;

	case MXLND_MSG_PUT_DATA:
		spin_lock(&conn->mxk_lock);
		if (conn->mxk_incarnation == tx->mxc_incarnation) {
			conn->mxk_data_posted--;
		}
		spin_unlock(&conn->mxk_lock);
		break;

        case MXLND_MSG_NOOP:
        case MXLND_MSG_PUT_REQ:
        case MXLND_MSG_PUT_ACK:
        case MXLND_MSG_GET_REQ:
        case MXLND_MSG_EAGER:
                break;

        case MXLND_MSG_CONN_ACK:
                if (peer->mxp_incompatible) {
                        /* we sent our params, now close this conn */
                        mxlnd_conn_disconnect(conn, 0, 1);
                }
        case MXLND_MSG_CONN_REQ:
                if (failed) {
                        CNETERR("%s failed with %s (%d) (errno = %d) to %s\n",
                               type == MXLND_MSG_CONN_REQ ? "CONN_REQ" : "CONN_ACK",
                               mx_strstatus(code), code, tx->mxc_errno,
                               libcfs_nid2str(tx->mxc_nid));
                        if (!peer->mxp_incompatible) {
				spin_lock(&conn->mxk_lock);
				if (code == MX_STATUS_BAD_SESSION)
					mxlnd_set_conn_status(conn,
							      MXLND_CONN_INIT);
				else
					mxlnd_set_conn_status(conn,
							      MXLND_CONN_FAIL);
				spin_unlock(&conn->mxk_lock);
                        }
                }
                break;

        default:
                CNETERR("Unknown msg type of %d\n", type);
                LBUG();
        }

        if (credit) {
		spin_lock(&conn->mxk_lock);
		if (conn->mxk_incarnation == tx->mxc_incarnation) {
			conn->mxk_ntx_posted--;
		}
		spin_unlock(&conn->mxk_lock);
        }

        mxlnd_put_idle_tx(tx);
        mxlnd_conn_decref(conn);

        mxlnd_check_sends(peer);

        CDEBUG(D_NET, "leaving\n");
        return;
}

/* Handle completion of MSG or DATA rx.
 * CONN_REQ and CONN_ACK are handled elsewhere. */
void
mxlnd_handle_rx_completion(kmx_ctx_t *rx)
{
        int             ret             = 0;
        int             repost          = 1;
        int             credit          = 1;
        u32             nob             = rx->mxc_status.xfer_length;
        u64             bits            = rx->mxc_status.match_info;
        kmx_msg_t      *msg             = rx->mxc_msg;
        kmx_peer_t     *peer            = rx->mxc_peer;
        kmx_conn_t     *conn            = rx->mxc_conn;
        u8              type            = rx->mxc_msg_type;
        u64             seq             = bits;
        lnet_msg_t     *lntmsg[2];
        int             result          = 0;
        int             peer_ref        = 0;
        int             conn_ref        = 0;

        /* NOTE We may only know the peer's nid if it is a PUT_REQ, GET_REQ,
         * failed GET reply */

        /* NOTE peer may still be NULL if it is a new peer and
         *      conn may be NULL if this is a re-connect */
        if (likely(peer != NULL && conn != NULL)) {
                /* we have a reference on the conn */
                conn_ref = 1;
        } else if (peer != NULL && conn == NULL) {
                /* we have a reference on the peer */
                peer_ref = 1;
        } else if (peer == NULL && conn != NULL) {
                /* fatal error */
                CERROR("rx 0x%llx from %s has conn but no peer\n",
                       bits, libcfs_nid2str(rx->mxc_nid));
                LBUG();
        } /* else peer and conn == NULL */

        if (conn == NULL && peer != NULL) {
		write_lock(&kmxlnd_data.kmx_global_lock);
                conn = peer->mxp_conn;
                if (conn) {
                        mxlnd_conn_addref(conn); /* conn takes ref... */
                        mxlnd_peer_decref(peer); /* from peer */
                        conn_ref = 1;
                        peer_ref = 0;
                }
		write_unlock(&kmxlnd_data.kmx_global_lock);
                rx->mxc_conn = conn;
        }

#if MXLND_DEBUG
        CDEBUG(D_NET, "receiving msg bits=0x%llx nob=%d peer=0x%p\n", bits, nob, peer);
#endif

        lntmsg[0] = NULL;
        lntmsg[1] = NULL;

        if (rx->mxc_status.code != MX_STATUS_SUCCESS &&
            rx->mxc_status.code != MX_STATUS_TRUNCATED) {
                CNETERR("rx from %s failed with %s (%d)\n",
                        libcfs_nid2str(rx->mxc_nid),
                        mx_strstatus(rx->mxc_status.code),
                        rx->mxc_status.code);
                credit = 0;
                goto cleanup;
        }

        if (nob == 0) {
                /* this may be a failed GET reply */
                if (type == MXLND_MSG_GET_DATA) {
                        /* get the error (52-59) bits from the match bits */
                        ret = (u32) MXLND_ERROR_VAL(rx->mxc_status.match_info);
                        lntmsg[0] = rx->mxc_lntmsg[0];
                        result = -ret;
                        goto cleanup;
                } else {
                        /* we had a rx complete with 0 bytes (no hdr, nothing) */
                        CNETERR("rx from %s returned with 0 bytes\n",
                                libcfs_nid2str(rx->mxc_nid));
                        goto cleanup;
                }
        }

        /* NOTE PUT_DATA and GET_DATA do not have mxc_msg, do not call unpack() */
        if (type == MXLND_MSG_PUT_DATA) {
                /* result = 0; */
                lntmsg[0] = rx->mxc_lntmsg[0];
                goto cleanup;
        } else if (type == MXLND_MSG_GET_DATA) {
                /* result = 0; */
                lntmsg[0] = rx->mxc_lntmsg[0];
                lntmsg[1] = rx->mxc_lntmsg[1];
                goto cleanup;
        }

        ret = mxlnd_unpack_msg(msg, nob);
        if (ret != 0) {
                CNETERR("Error %d unpacking rx from %s\n",
                        ret, libcfs_nid2str(rx->mxc_nid));
                goto cleanup;
        }
        rx->mxc_nob = nob;
        type = msg->mxm_type;

        if (rx->mxc_nid != msg->mxm_srcnid ||
            kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
                CNETERR("rx with mismatched NID (type %s) (my nid is "
                       "0x%llx and rx msg dst is 0x%llx)\n",
                       mxlnd_msgtype_to_str(type), kmxlnd_data.kmx_ni->ni_nid,
                       msg->mxm_dstnid);
                goto cleanup;
        }

        if ((conn != NULL && msg->mxm_srcstamp != conn->mxk_incarnation) ||
            msg->mxm_dststamp != kmxlnd_data.kmx_incarnation) {
                CNETERR("Stale rx from %s with type %s "
                       "(mxm_srcstamp (%lld) != mxk_incarnation (%lld) "
                       "|| mxm_dststamp (%lld) != kmx_incarnation (%lld))\n",
                       libcfs_nid2str(rx->mxc_nid), mxlnd_msgtype_to_str(type),
                       msg->mxm_srcstamp, conn->mxk_incarnation,
                       msg->mxm_dststamp, kmxlnd_data.kmx_incarnation);
                credit = 0;
                goto cleanup;
        }

        CDEBUG(D_NET, "Received %s with %d credits\n",
                      mxlnd_msgtype_to_str(type), msg->mxm_credits);

        LASSERT(peer != NULL && conn != NULL);
        if (msg->mxm_credits != 0) {
		spin_lock(&conn->mxk_lock);
                if (msg->mxm_srcstamp == conn->mxk_incarnation) {
                        if ((conn->mxk_credits + msg->mxm_credits) >
                             *kmxlnd_tunables.kmx_peercredits) {
                                CNETERR("mxk_credits %d  mxm_credits %d\n",
                                        conn->mxk_credits, msg->mxm_credits);
                        }
                        conn->mxk_credits += msg->mxm_credits;
                        LASSERT(conn->mxk_credits >= 0);
                        LASSERT(conn->mxk_credits <= *kmxlnd_tunables.kmx_peercredits);
                }
		spin_unlock(&conn->mxk_lock);
        }

        CDEBUG(D_NET, "switch %s for rx (0x%llx)\n", mxlnd_msgtype_to_str(type), seq);
        switch (type) {
        case MXLND_MSG_NOOP:
                break;

        case MXLND_MSG_EAGER:
                ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.eager.mxem_hdr,
                                        msg->mxm_srcnid, rx, 0);
                repost = ret < 0;
                break;

        case MXLND_MSG_PUT_REQ:
                ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.put_req.mxprm_hdr,
                                        msg->mxm_srcnid, rx, 1);
                repost = ret < 0;
                break;

        case MXLND_MSG_PUT_ACK: {
                u64  cookie = (u64) msg->mxm_u.put_ack.mxpam_dst_cookie;
                if (cookie > MXLND_MAX_COOKIE) {
                        CNETERR("NAK for msg_type %d from %s\n", rx->mxc_msg_type,
                                libcfs_nid2str(rx->mxc_nid));
                        result = -((u32) MXLND_ERROR_VAL(cookie));
                        lntmsg[0] = rx->mxc_lntmsg[0];
                } else {
                        mxlnd_send_data(kmxlnd_data.kmx_ni, rx->mxc_lntmsg[0],
                                        rx->mxc_peer, MXLND_MSG_PUT_DATA,
                                        rx->mxc_msg->mxm_u.put_ack.mxpam_dst_cookie);
                }
                /* repost == 1 */
                break;
        }
        case MXLND_MSG_GET_REQ:
                ret = lnet_parse(kmxlnd_data.kmx_ni, &msg->mxm_u.get_req.mxgrm_hdr,
                                        msg->mxm_srcnid, rx, 1);
                repost = ret < 0;
                break;

        default:
                CNETERR("Bad MXLND message type %x from %s\n", msg->mxm_type,
                        libcfs_nid2str(rx->mxc_nid));
                ret = -EPROTO;
                break;
        }

        if (ret < 0) {
                CDEBUG(D_NET, "setting PEER_CONN_FAILED\n");
		spin_lock(&conn->mxk_lock);
		mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
		spin_unlock(&conn->mxk_lock);
	}

cleanup:
	if (conn != NULL) {
		spin_lock(&conn->mxk_lock);
		conn->mxk_last_rx = cfs_time_current(); /* jiffies */
		spin_unlock(&conn->mxk_lock);
        }

        if (repost) {
                /* lnet_parse() failed, etc., repost now */
                mxlnd_put_idle_rx(rx);
                if (conn != NULL && credit == 1) {
                        if (type == MXLND_MSG_PUT_DATA ||
                            type == MXLND_MSG_EAGER ||
                            type == MXLND_MSG_PUT_REQ ||
                            type == MXLND_MSG_NOOP) {
				spin_lock(&conn->mxk_lock);
				conn->mxk_outstanding++;
				spin_unlock(&conn->mxk_lock);
                        }
                }
                if (conn_ref) mxlnd_conn_decref(conn);
                LASSERT(peer_ref == 0);
        }

        if (type == MXLND_MSG_PUT_DATA || type == MXLND_MSG_GET_DATA) {
                CDEBUG(D_NET, "leaving for rx (0x%llx)\n", bits);
        } else {
                CDEBUG(D_NET, "leaving for rx (0x%llx)\n", seq);
        }

        if (lntmsg[0] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[0], result);
        if (lntmsg[1] != NULL) lnet_finalize(kmxlnd_data.kmx_ni, lntmsg[1], result);

        if (conn != NULL && credit == 1) mxlnd_check_sends(peer);

        return;
}

void
mxlnd_handle_connect_msg(kmx_peer_t *peer, u8 msg_type, mx_status_t status)
{
        kmx_ctx_t       *tx     = NULL;
        kmx_msg_t       *txmsg  = NULL;
        kmx_conn_t      *conn   = peer->mxp_conn;
        u64             nic_id  = 0ULL;
        u32             ep_id   = 0;
        u32             sid     = 0;
        u8              type    = (msg_type == MXLND_MSG_ICON_REQ ?
                                   MXLND_MSG_CONN_REQ : MXLND_MSG_CONN_ACK);

        /* a conn ref was taken when calling mx_iconnect(),
         * hold it until CONN_REQ or CONN_ACK completes */

        CDEBUG(D_NET, "entering\n");
        if (status.code != MX_STATUS_SUCCESS) {
                int send_bye    = (msg_type == MXLND_MSG_ICON_REQ ? 0 : 1);

                CNETERR("mx_iconnect() failed for %s with %s (%d) "
                        "to %s mxp_nid = 0x%llx mxp_nic_id = 0x%0llx mxp_ep_id = %d\n",
                        mxlnd_msgtype_to_str(msg_type),
                        mx_strstatus(status.code), status.code,
                        libcfs_nid2str(peer->mxp_nid),
                        peer->mxp_nid,
                        peer->mxp_nic_id,
                        peer->mxp_ep_id);
		spin_lock(&conn->mxk_lock);
		mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
		spin_unlock(&conn->mxk_lock);

                if (cfs_time_after(jiffies, peer->mxp_reconnect_time +
                                   MXLND_CONNECT_TIMEOUT)) {
                        CNETERR("timeout, calling conn_disconnect()\n");
                        mxlnd_conn_disconnect(conn, 0, send_bye);
                }

                mxlnd_conn_decref(conn);
                return;
        }
        mx_decompose_endpoint_addr2(status.source, &nic_id, &ep_id, &sid);
	write_lock(&kmxlnd_data.kmx_global_lock);
	spin_lock(&conn->mxk_lock);
	conn->mxk_epa = status.source;
	mx_set_endpoint_addr_context(conn->mxk_epa, (void *) conn);
	if (msg_type == MXLND_MSG_ICON_ACK && likely(!peer->mxp_incompatible)) {
		mxlnd_set_conn_status(conn, MXLND_CONN_READY);
	}
	spin_unlock(&conn->mxk_lock);
	write_unlock(&kmxlnd_data.kmx_global_lock);

	/* mx_iconnect() succeeded, reset delay to 0 */
	write_lock(&kmxlnd_data.kmx_global_lock);
	peer->mxp_reconnect_time = 0;
	peer->mxp_conn->mxk_sid = sid;
	write_unlock(&kmxlnd_data.kmx_global_lock);

        /* marshal CONN_REQ or CONN_ACK msg */
        /* we are still using the conn ref from iconnect() - do not take another */
        tx = mxlnd_get_idle_tx();
        if (tx == NULL) {
                CNETERR("Can't obtain %s tx for %s\n",
                       mxlnd_msgtype_to_str(type),
                       libcfs_nid2str(peer->mxp_nid));
		spin_lock(&conn->mxk_lock);
		mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
		spin_unlock(&conn->mxk_lock);
                mxlnd_conn_decref(conn);
                return;
        }

        tx->mxc_peer = peer;
        tx->mxc_conn = conn;
        tx->mxc_deadline = jiffies + MXLND_CONNECT_TIMEOUT;
        CDEBUG(D_NET, "sending %s\n", mxlnd_msgtype_to_str(type));
        mxlnd_init_tx_msg (tx, type, sizeof(kmx_connreq_msg_t), peer->mxp_nid);
        txmsg = tx->mxc_msg;
        txmsg->mxm_u.conn_req.mxcrm_queue_depth = *kmxlnd_tunables.kmx_peercredits;
        txmsg->mxm_u.conn_req.mxcrm_eager_size = MXLND_MSG_SIZE;
        tx->mxc_match = mxlnd_create_match(tx, 0);

        mxlnd_queue_tx(tx);
        return;
}

/**
 * mxlnd_request_waitd - the MX request completion thread(s)
 * @arg - thread id (as a void *)
 *
 * This thread waits for a MX completion and then completes the request.
 * We will create one thread per CPU.
 */
int
mxlnd_request_waitd(void *arg)
{
        long                    id              = (long) arg;
        __u32                   result          = 0;
        mx_return_t             mxret           = MX_SUCCESS;
        mx_status_t             status;
        kmx_ctx_t              *ctx             = NULL;
        enum kmx_req_state      req_type        = MXLND_REQ_TX;
        kmx_peer_t             *peer            = NULL;
        kmx_conn_t             *conn            = NULL;
#if MXLND_POLLING
        int                     count           = 0;
#endif

        memset(&status, 0, sizeof(status));

        CDEBUG(D_NET, "%s starting\n", name);

	while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
		u8      msg_type        = 0;

                mxret = MX_SUCCESS;
                result = 0;
#if MXLND_POLLING
                if (id == 0 && count++ < *kmxlnd_tunables.kmx_polling) {
                        mxret = mx_test_any(kmxlnd_data.kmx_endpt, 0ULL, 0ULL,
                                            &status, &result);
                } else {
                        count = 0;
                        mxret = mx_wait_any(kmxlnd_data.kmx_endpt, MXLND_WAIT_TIMEOUT,
                                            0ULL, 0ULL, &status, &result);
                }
#else
                mxret = mx_wait_any(kmxlnd_data.kmx_endpt, MXLND_WAIT_TIMEOUT,
                                    0ULL, 0ULL, &status, &result);
#endif
		if (unlikely(atomic_read(&kmxlnd_data.kmx_shutdown)))
			break;

                if (result != 1) {
                        /* nothing completed... */
                        continue;
                }

                CDEBUG(D_NET, "wait_any() returned with %s (%d) with "
                       "match_info 0x%llx and length %d\n",
                       mx_strstatus(status.code), status.code,
                       (u64) status.match_info, status.msg_length);

                if (status.code != MX_STATUS_SUCCESS) {
                        CNETERR("wait_any() failed with %s (%d) with "
                                "match_info 0x%llx and length %d\n",
                                mx_strstatus(status.code), status.code,
                                (u64) status.match_info, status.msg_length);
                }

                msg_type = MXLND_MSG_TYPE(status.match_info);

                /* This may be a mx_iconnect() request completing,
                 * check the bit mask for CONN_REQ and CONN_ACK */
                if (msg_type == MXLND_MSG_ICON_REQ ||
                    msg_type == MXLND_MSG_ICON_ACK) {
                        peer = (kmx_peer_t*) status.context;
                        mxlnd_handle_connect_msg(peer, msg_type, status);
                        continue;
                }

                /* This must be a tx or rx */

                /* NOTE: if this is a RX from the unexpected callback, it may
                 * have very little info. If we dropped it in unexpected_recv(),
                 * it will not have a context. If so, ignore it. */
                ctx = (kmx_ctx_t *) status.context;
                if (ctx != NULL) {

                        req_type = ctx->mxc_type;
                        conn = ctx->mxc_conn; /* this may be NULL */
                        mxlnd_deq_pending_ctx(ctx);

                        /* copy status to ctx->mxc_status */
                        ctx->mxc_status = status;

                        switch (req_type) {
                        case MXLND_REQ_TX:
                                mxlnd_handle_tx_completion(ctx);
                                break;
                        case MXLND_REQ_RX:
                                mxlnd_handle_rx_completion(ctx);
                                break;
                        default:
                                CNETERR("Unknown ctx type %d\n", req_type);
                                LBUG();
                                break;
                        }

                        /* conn is always set except for the first CONN_REQ rx
                         * from a new peer */
                        if (status.code != MX_STATUS_SUCCESS && conn != NULL) {
                                mxlnd_conn_disconnect(conn, 1, 1);
                        }
                }
                CDEBUG(D_NET, "waitd() completed task\n");
        }
        CDEBUG(D_NET, "%s stopping\n", name);
        mxlnd_thread_stop(id);
        return 0;
}


unsigned long
mxlnd_check_timeouts(unsigned long now)
{
        int             i               = 0;
        int             disconnect      = 0;
        unsigned long   next            = 0; /* jiffies */
        kmx_peer_t      *peer           = NULL;
        kmx_conn_t      *conn           = NULL;
	rwlock_t	*g_lock		= &kmxlnd_data.kmx_global_lock;

	read_lock(g_lock);
	for (i = 0; i < MXLND_HASH_SIZE; i++) {
		cfs_list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i],
					mxp_list) {

			if (unlikely(atomic_read(&kmxlnd_data.kmx_shutdown))) {
				read_unlock(g_lock);
				return next;
			}

                        conn = peer->mxp_conn;
                        if (conn) {
                                mxlnd_conn_addref(conn);
                        } else {
                                continue;
                        }

			spin_lock(&conn->mxk_lock);

			/* if nothing pending (timeout == 0) or
			 * if conn is already disconnected,
			 * skip this conn */
			if (conn->mxk_timeout == 0 ||
			    conn->mxk_status == MXLND_CONN_DISCONNECT) {
				spin_unlock(&conn->mxk_lock);
                                mxlnd_conn_decref(conn);
                                continue;
                        }

                        /* we want to find the timeout that will occur first.
                         * if it is in the future, we will sleep until then.
                         * if it is in the past, then we will sleep one
                         * second and repeat the process. */
                        if ((next == 0) ||
                            (cfs_time_before(conn->mxk_timeout, next))) {
                                next = conn->mxk_timeout;
                        }

                        disconnect = 0;

			if (cfs_time_aftereq(now, conn->mxk_timeout))
				disconnect = 1;
			spin_unlock(&conn->mxk_lock);

			if (disconnect)
				mxlnd_conn_disconnect(conn, 1, 1);
			mxlnd_conn_decref(conn);
		}
	}
	read_unlock(g_lock);
	if (next == 0)
		next = now + MXLND_COMM_TIMEOUT;

	return next;
}

void
mxlnd_passive_connect(kmx_connparams_t *cp)
{
        int             ret             = 0;
        int             incompatible    = 0;
        u64             nic_id          = 0ULL;
        u32             ep_id           = 0;
        u32             sid             = 0;
        int             conn_ref        = 0;
        kmx_msg_t       *msg            = &cp->mxr_msg;
        kmx_peer_t      *peer           = cp->mxr_peer;
        kmx_conn_t      *conn           = NULL;
	rwlock_t	*g_lock		= &kmxlnd_data.kmx_global_lock;

        mx_decompose_endpoint_addr2(cp->mxr_epa, &nic_id, &ep_id, &sid);

        ret = mxlnd_unpack_msg(msg, cp->mxr_nob);
        if (ret != 0) {
                if (peer) {
                        CNETERR("Error %d unpacking CONN_REQ from %s\n",
                               ret, libcfs_nid2str(peer->mxp_nid));
                } else {
                        CNETERR("Error %d unpacking CONN_REQ from "
                               "unknown host with nic_id 0x%llx\n", ret, nic_id);
                }
                goto cleanup;
        }
        if (kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
                CNETERR("Can't accept %s: bad dst nid %s\n",
                                libcfs_nid2str(msg->mxm_srcnid),
                                libcfs_nid2str(msg->mxm_dstnid));
                goto cleanup;
        }
        if (msg->mxm_u.conn_req.mxcrm_queue_depth != *kmxlnd_tunables.kmx_peercredits) {
                CNETERR("Can't accept %s: incompatible queue depth "
                            "%d (%d wanted)\n",
                                libcfs_nid2str(msg->mxm_srcnid),
                                msg->mxm_u.conn_req.mxcrm_queue_depth,
                                *kmxlnd_tunables.kmx_peercredits);
                incompatible = 1;
        }
        if (msg->mxm_u.conn_req.mxcrm_eager_size != MXLND_MSG_SIZE) {
                CNETERR("Can't accept %s: incompatible EAGER size "
                            "%d (%d wanted)\n",
                                libcfs_nid2str(msg->mxm_srcnid),
                                msg->mxm_u.conn_req.mxcrm_eager_size,
                                (int) MXLND_MSG_SIZE);
                incompatible = 1;
        }

        if (peer == NULL) {
                peer = mxlnd_find_peer_by_nid(msg->mxm_srcnid, 0); /* adds peer ref */
                if (peer == NULL) {
                        int             hash    = 0;
                        u32             board   = 0;
                        kmx_peer_t      *existing_peer    = NULL;

                        hash = mxlnd_nid_to_hash(msg->mxm_srcnid);

                        mx_nic_id_to_board_number(nic_id, &board);

                        /* adds conn ref for peer and one for this function */
                        ret = mxlnd_peer_alloc(&peer, msg->mxm_srcnid,
                                               board, ep_id, 0ULL);
                        if (ret != 0) {
                                goto cleanup;
                        }
                        peer->mxp_conn->mxk_sid = sid;
                        LASSERT(peer->mxp_ep_id == ep_id);
			write_lock(g_lock);
                        existing_peer = mxlnd_find_peer_by_nid_locked(msg->mxm_srcnid);
                        if (existing_peer) {
                                mxlnd_conn_decref(peer->mxp_conn);
                                mxlnd_peer_decref(peer);
                                peer = existing_peer;
                                mxlnd_conn_addref(peer->mxp_conn);
                                conn = peer->mxp_conn;
			} else {
				cfs_list_add_tail(&peer->mxp_list,
						  &kmxlnd_data.kmx_peers[hash]);
				atomic_inc(&kmxlnd_data.kmx_npeers);
			}
			write_unlock(g_lock);
                } else {
                        ret = mxlnd_conn_alloc(&conn, peer); /* adds 2nd ref */
			write_lock(g_lock);
                        mxlnd_peer_decref(peer); /* drop ref taken above */
			write_unlock(g_lock);
                        if (ret != 0) {
                                CNETERR("Cannot allocate mxp_conn\n");
                                goto cleanup;
                        }
                }
                conn_ref = 1; /* peer/conn_alloc() added ref for this function */
                conn = peer->mxp_conn;
        } else { /* unexpected handler found peer */
                kmx_conn_t      *old_conn       = peer->mxp_conn;

                if (sid != peer->mxp_conn->mxk_sid) {
                        /* do not call mx_disconnect() or send a BYE */
                        mxlnd_conn_disconnect(old_conn, 0, 0);

                        /* This allocs a conn, points peer->mxp_conn to this one.
                        * The old conn is still on the peer->mxp_conns list.
                        * As the pending requests complete, they will call
                        * conn_decref() which will eventually free it. */
                        ret = mxlnd_conn_alloc(&conn, peer);
                        if (ret != 0) {
                                CNETERR("Cannot allocate peer->mxp_conn\n");
                                goto cleanup;
                        }
                        /* conn_alloc() adds one ref for the peer and one
                         * for this function */
                        conn_ref = 1;

                        peer->mxp_conn->mxk_sid = sid;
                } else {
                        /* same sid */
                        conn = peer->mxp_conn;
                }
        }
	write_lock(g_lock);
	peer->mxp_incompatible = incompatible;
	write_unlock(g_lock);
	spin_lock(&conn->mxk_lock);
	conn->mxk_incarnation = msg->mxm_srcstamp;
	mxlnd_set_conn_status(conn, MXLND_CONN_WAIT);
	spin_unlock(&conn->mxk_lock);

        /* handle_conn_ack() will create the CONN_ACK msg */
        mxlnd_iconnect(peer, (u8) MXLND_MSG_ICON_ACK);

cleanup:
        if (conn_ref) mxlnd_conn_decref(conn);

        mxlnd_connparams_free(cp);
        return;
}

void
mxlnd_check_conn_ack(kmx_connparams_t *cp)
{
        int             ret             = 0;
        int             incompatible    = 0;
        u64             nic_id          = 0ULL;
        u32             ep_id           = 0;
        u32             sid             = 0;
        kmx_msg_t       *msg            = &cp->mxr_msg;
        kmx_peer_t      *peer           = cp->mxr_peer;
        kmx_conn_t      *conn           = cp->mxr_conn;

        mx_decompose_endpoint_addr2(cp->mxr_epa, &nic_id, &ep_id, &sid);

        ret = mxlnd_unpack_msg(msg, cp->mxr_nob);
        if (ret != 0) {
                if (peer) {
                        CNETERR("Error %d unpacking CONN_ACK from %s\n",
                               ret, libcfs_nid2str(peer->mxp_nid));
                } else {
                        CNETERR("Error %d unpacking CONN_ACK from "
                               "unknown host with nic_id 0x%llx\n", ret, nic_id);
                }
                ret = -1;
                incompatible = 1;
                goto failed;
        }
        if (kmxlnd_data.kmx_ni->ni_nid != msg->mxm_dstnid) {
                CNETERR("Can't accept CONN_ACK from %s: "
                       "bad dst nid %s\n", libcfs_nid2str(msg->mxm_srcnid),
                        libcfs_nid2str(msg->mxm_dstnid));
                ret = -1;
                goto failed;
        }
        if (msg->mxm_u.conn_req.mxcrm_queue_depth != *kmxlnd_tunables.kmx_peercredits) {
                CNETERR("Can't accept CONN_ACK from %s: "
                       "incompatible queue depth %d (%d wanted)\n",
                        libcfs_nid2str(msg->mxm_srcnid),
                        msg->mxm_u.conn_req.mxcrm_queue_depth,
                        *kmxlnd_tunables.kmx_peercredits);
                incompatible = 1;
                ret = -1;
                goto failed;
        }
        if (msg->mxm_u.conn_req.mxcrm_eager_size != MXLND_MSG_SIZE) {
                CNETERR("Can't accept CONN_ACK from %s: "
                        "incompatible EAGER size %d (%d wanted)\n",
                        libcfs_nid2str(msg->mxm_srcnid),
                        msg->mxm_u.conn_req.mxcrm_eager_size,
                        (int) MXLND_MSG_SIZE);
                incompatible = 1;
                ret = -1;
                goto failed;
        }
	write_lock(&kmxlnd_data.kmx_global_lock);
	peer->mxp_incompatible = incompatible;
	write_unlock(&kmxlnd_data.kmx_global_lock);
	spin_lock(&conn->mxk_lock);
        conn->mxk_credits = *kmxlnd_tunables.kmx_peercredits;
        conn->mxk_outstanding = 0;
        conn->mxk_incarnation = msg->mxm_srcstamp;
        conn->mxk_timeout = 0;
        if (!incompatible) {
                CDEBUG(D_NET, "setting peer %s CONN_READY\n",
                       libcfs_nid2str(msg->mxm_srcnid));
                mxlnd_set_conn_status(conn, MXLND_CONN_READY);
        }
	spin_unlock(&conn->mxk_lock);

	if (!incompatible)
		mxlnd_check_sends(peer);

failed:
	if (ret < 0) {
		spin_lock(&conn->mxk_lock);
		mxlnd_set_conn_status(conn, MXLND_CONN_FAIL);
		spin_unlock(&conn->mxk_lock);
	}

	if (incompatible) mxlnd_conn_disconnect(conn, 0, 0);

	mxlnd_connparams_free(cp);
	return;
}

int
mxlnd_abort_msgs(void)
{
	int			count		= 0;
	cfs_list_t		*orphans	= &kmxlnd_data.kmx_orphan_msgs;
	spinlock_t		*g_conn_lock	= &kmxlnd_data.kmx_conn_lock;

	/* abort orphans */
	spin_lock(g_conn_lock);
	while (!cfs_list_empty(orphans)) {
		kmx_ctx_t       *ctx     = NULL;
		kmx_conn_t      *conn   = NULL;

		ctx = cfs_list_entry(orphans->next, kmx_ctx_t, mxc_list);
		cfs_list_del_init(&ctx->mxc_list);
		spin_unlock(g_conn_lock);

                ctx->mxc_errno = -ECONNABORTED;
                conn = ctx->mxc_conn;
                CDEBUG(D_NET, "aborting %s %s %s\n",
                       mxlnd_msgtype_to_str(ctx->mxc_msg_type),
                       ctx->mxc_type == MXLND_REQ_TX ? "(TX) to" : "(RX) from",
                       libcfs_nid2str(ctx->mxc_nid));
                if (ctx->mxc_type == MXLND_REQ_TX) {
                        mxlnd_put_idle_tx(ctx); /* do not hold any locks */
                        if (conn) mxlnd_conn_decref(conn); /* for this tx */
                } else {
                        ctx->mxc_state = MXLND_CTX_CANCELED;
                        mxlnd_handle_rx_completion(ctx);
                }

                count++;
		spin_lock(g_conn_lock);
	}
	spin_unlock(g_conn_lock);

	return count;
}

int
mxlnd_free_conn_zombies(void)
{
	int		count		= 0;
	cfs_list_t	*zombies	= &kmxlnd_data.kmx_conn_zombies;
	spinlock_t	*g_conn_lock	= &kmxlnd_data.kmx_conn_lock;
	rwlock_t	*g_lock		= &kmxlnd_data.kmx_global_lock;

	/* cleanup any zombies */
	spin_lock(g_conn_lock);
	while (!cfs_list_empty(zombies)) {
		kmx_conn_t	*conn   = NULL;

		conn = cfs_list_entry(zombies->next, kmx_conn_t, mxk_zombie);
		cfs_list_del_init(&conn->mxk_zombie);
		spin_unlock(g_conn_lock);

		write_lock(g_lock);
		mxlnd_conn_free_locked(conn);
		write_unlock(g_lock);

		count++;
		spin_lock(g_conn_lock);
	}
	spin_unlock(g_conn_lock);
	CDEBUG(D_NET, "%s: freed %d zombies\n", __func__, count);
	return count;
}

/**
 * mxlnd_connd - handles incoming connection requests
 * @arg - thread id (as a void *)
 *
 * This thread handles incoming connection requests
 */
int
mxlnd_connd(void *arg)
{
	long                    id              = (long) arg;

	CDEBUG(D_NET, "connd starting\n");

	while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {
		int                ret             = 0;
		kmx_connparams_t  *cp              = NULL;
		spinlock_t	  *g_conn_lock	= &kmxlnd_data.kmx_conn_lock;
		cfs_list_t	  *conn_reqs	= &kmxlnd_data.kmx_conn_reqs;

		ret = down_interruptible(&kmxlnd_data.kmx_conn_sem);

		if (atomic_read(&kmxlnd_data.kmx_shutdown))
			break;

                if (ret != 0)
                        continue;

                ret = mxlnd_abort_msgs();
                ret += mxlnd_free_conn_zombies();

		spin_lock(g_conn_lock);
		if (cfs_list_empty(conn_reqs)) {
			if (ret == 0)
				CNETERR("connd woke up but did not find a "
					"kmx_connparams_t or zombie conn\n");
			spin_unlock(g_conn_lock);
			continue;
		}
		cp = cfs_list_entry(conn_reqs->next, kmx_connparams_t,
				    mxr_list);
		cfs_list_del_init(&cp->mxr_list);
		spin_unlock(g_conn_lock);

                switch (MXLND_MSG_TYPE(cp->mxr_match)) {
                case MXLND_MSG_CONN_REQ:
                        /* We have a connection request. Handle it. */
                        mxlnd_passive_connect(cp);
                        break;
                case MXLND_MSG_CONN_ACK:
                        /* The peer is ready for messages */
                        mxlnd_check_conn_ack(cp);
                        break;
                }
        }

        mxlnd_free_conn_zombies();

        CDEBUG(D_NET, "connd stopping\n");
        mxlnd_thread_stop(id);
        return 0;
}

/**
 * mxlnd_timeoutd - enforces timeouts on messages
 * @arg - thread id (as a void *)
 *
 * This thread queries each peer for its earliest timeout. If a peer has timed out,
 * it calls mxlnd_conn_disconnect().
 *
 * After checking for timeouts, try progressing sends (call check_sends()).
 */
int
mxlnd_timeoutd(void *arg)
{
	int             i       = 0;
	long            id      = (long) arg;
	unsigned long   now     = 0;
	unsigned long   next    = 0;
	unsigned long   delay   = msecs_to_jiffies(MSEC_PER_SEC);
	kmx_peer_t     *peer    = NULL;
	kmx_peer_t     *temp    = NULL;
	kmx_conn_t     *conn    = NULL;
	rwlock_t   *g_lock  = &kmxlnd_data.kmx_global_lock;

        CDEBUG(D_NET, "timeoutd starting\n");

	while (!(atomic_read(&kmxlnd_data.kmx_shutdown))) {

                now = jiffies;
                /* if the next timeout has not arrived, go back to sleep */
                if (cfs_time_after(now, next)) {
                        next = mxlnd_check_timeouts(now);
                }

                /* try to progress peers' txs */
		write_lock(g_lock);
                for (i = 0; i < MXLND_HASH_SIZE; i++) {
                        cfs_list_t *peers = &kmxlnd_data.kmx_peers[i];

                        /* NOTE we are safe against the removal of peer, but
                         * not against the removal of temp */
                        cfs_list_for_each_entry_safe(peer, temp, peers,
                                                     mxp_list) {
				if (atomic_read(&kmxlnd_data.kmx_shutdown))
                                        break;
                                mxlnd_peer_addref(peer); /* add ref... */
                                conn = peer->mxp_conn;
                                if (conn && conn->mxk_status != MXLND_CONN_DISCONNECT) {
                                        mxlnd_conn_addref(conn); /* take ref... */
                                } else {
                                        CDEBUG(D_NET, "ignoring %s\n",
                                               libcfs_nid2str(peer->mxp_nid));
                                        mxlnd_peer_decref(peer); /* ...to here */
                                        continue;
                                }

				if ((conn->mxk_status == MXLND_CONN_READY ||
				    conn->mxk_status == MXLND_CONN_FAIL) &&
				    cfs_time_after(now,
						   conn->mxk_last_tx +
						   msecs_to_jiffies(MSEC_PER_SEC))) {
					write_unlock(g_lock);
					mxlnd_check_sends(peer);
					write_lock(g_lock);
				}
				mxlnd_conn_decref(conn); /* until here */
				mxlnd_peer_decref(peer); /* ...to here */
			}
		}
		write_unlock(g_lock);

                mxlnd_sleep(delay);
        }
        CDEBUG(D_NET, "timeoutd stopping\n");
        mxlnd_thread_stop(id);
        return 0;
}