mxlnd_cb.c

非常经典的一个分布式系统

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2004 Cluster File Systems, Inc.
 *   Author: Eric Barton <[e-mail]>
 * Copyright (C) 2006 Myricom, Inc.
 *   Author: Myricom, Inc. <help at myri.com>
 *
 *   This file is part of Lustre, http://www.lustre.org.
 *
 *   Lustre is free software; you can redistribute it and/or
 *   modify it under the terms of version 2 of the GNU General Public
 *   License as published by the Free Software Foundation.
 *
 *   Lustre 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 for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Lustre; if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

#include "mxlnd.h"

inline void mxlnd_noop(char *s, ...)
{
        return;
}

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_REQ:
                return "MXLND_CONN_REQ";
        case MXLND_CONN_ACK:
                return "MXLND_CONN_ACK";
        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_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:
                return "*unknown*";
        }
}

static inline u64
//mxlnd_create_match(u8 msg_type, u8 error, u64 cookie)
mxlnd_create_match(struct kmx_ctx *ctx, u8 error)
{
        u64 type        = (u64) ctx->mxc_msg_type;
        u64 err         = (u64) error;
        u64 match       = 0LL;

        LASSERT(ctx->mxc_msg_type != 0);
        LASSERT(ctx->mxc_cookie >> 52 == 0);
        match = (type << 60) | (err << 52) | ctx->mxc_cookie;
        return match;
}

static inline void
mxlnd_parse_match(u64 match, u8 *msg_type, u8 *error, u64 *cookie)
{
        *msg_type = (u8) (match >> 60);
        *error    = (u8) ((match >> 52) & 0xFF);
        *cookie   = match & 0xFFFFFFFFFFFFFLL;
        LASSERT(match == (MXLND_MASK_ICON_REQ & 0xF000000000000000LL) ||
                match == (MXLND_MASK_ICON_ACK & 0xF000000000000000LL) ||
                *msg_type == MXLND_MSG_EAGER    ||
                *msg_type == MXLND_MSG_CONN_REQ ||
                *msg_type == MXLND_MSG_CONN_ACK ||
                *msg_type == MXLND_MSG_NOOP     ||
                *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);
        return;
}

struct kmx_ctx *
mxlnd_get_idle_rx(void)
{
        struct list_head        *tmp    = NULL;
        struct kmx_ctx          *rx     = NULL;

        spin_lock(&kmxlnd_data.kmx_rx_idle_lock);

        if (list_empty (&kmxlnd_data.kmx_rx_idle)) {
                spin_unlock(&kmxlnd_data.kmx_rx_idle_lock);
                return NULL;
        }

        tmp = &kmxlnd_data.kmx_rx_idle;
        rx = list_entry (tmp->next, struct kmx_ctx, mxc_list);
        list_del_init(&rx->mxc_list);
        spin_unlock(&kmxlnd_data.kmx_rx_idle_lock);

#if MXLND_DEBUG
        if (rx->mxc_get != rx->mxc_put) {
                CDEBUG(D_NETERROR, "*** RX get (%lld) != put (%lld) ***\n", rx->mxc_get, rx->mxc_put);
                CDEBUG(D_NETERROR, "*** incarnation= %lld ***\n", rx->mxc_incarnation);
                CDEBUG(D_NETERROR, "*** deadline= %ld ***\n", rx->mxc_deadline);
                CDEBUG(D_NETERROR, "*** state= %s ***\n", mxlnd_ctxstate_to_str(rx->mxc_state));
                CDEBUG(D_NETERROR, "*** listed?= %d ***\n", !list_empty(&rx->mxc_list));
                CDEBUG(D_NETERROR, "*** nid= 0x%llx ***\n", rx->mxc_nid);
                CDEBUG(D_NETERROR, "*** peer= 0x%p ***\n", rx->mxc_peer);
                CDEBUG(D_NETERROR, "*** msg_type= %s ***\n", mxlnd_msgtype_to_str(rx->mxc_msg_type));
                CDEBUG(D_NETERROR, "*** cookie= 0x%llx ***\n", rx->mxc_cookie);
                CDEBUG(D_NETERROR, "*** 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;

        return rx;
}

int
mxlnd_put_idle_rx(struct kmx_ctx *rx)
{
        if (rx == NULL) {
                CDEBUG(D_NETERROR, "called with NULL pointer\n");
                return -EINVAL;
        } else if (rx->mxc_type != MXLND_REQ_RX) {
                CDEBUG(D_NETERROR, "called with tx\n");
                return -EINVAL;
        }
        LASSERT(rx->mxc_get == rx->mxc_put + 1);
        mxlnd_ctx_init(rx);
        rx->mxc_put++;
        spin_lock(&kmxlnd_data.kmx_rx_idle_lock);
        list_add_tail(&rx->mxc_list, &kmxlnd_data.kmx_rx_idle);
        spin_unlock(&kmxlnd_data.kmx_rx_idle_lock);
        return 0;
}

int
mxlnd_reduce_idle_rxs(__u32 count)
{
        __u32                   i       = 0;
        struct kmx_ctx          *rx     = NULL;

        spin_lock(&kmxlnd_data.kmx_rxs_lock);
        for (i = 0; i < count; i++) {
                rx = mxlnd_get_idle_rx();
                if (rx != NULL) {
                        struct list_head *tmp = &rx->mxc_global_list;
                        list_del_init(tmp);
                        mxlnd_ctx_free(rx);
                } else {
                        CDEBUG(D_NETERROR, "only reduced %d out of %d rxs\n", i, count);
                        break;
                }
        }
        spin_unlock(&kmxlnd_data.kmx_rxs_lock);
        return 0;
}

struct kmx_ctx *
mxlnd_get_idle_tx(void)
{
        struct list_head        *tmp    = NULL;
        struct kmx_ctx          *tx     = NULL;

        spin_lock(&kmxlnd_data.kmx_tx_idle_lock);

        if (list_empty (&kmxlnd_data.kmx_tx_idle)) {
                CDEBUG(D_NETERROR, "%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 = list_entry (tmp->next, struct kmx_ctx, mxc_list);
        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) {
                tx->mxc_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;

        return tx;
}

int
mxlnd_put_idle_tx(struct kmx_ctx *tx)
{
        int             failed  = (tx->mxc_status.code != MX_STATUS_SUCCESS && tx->mxc_status.code != MX_STATUS_TRUNCATED);
        int             result  = failed ? -EIO : 0;
        lnet_msg_t      *lntmsg[2];

        if (tx == NULL) {
                CDEBUG(D_NETERROR, "called with NULL pointer\n");
                return -EINVAL;
        } else if (tx->mxc_type != MXLND_REQ_TX) {
                CDEBUG(D_NETERROR, "called with rx\n");
                return -EINVAL;
        }

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

        LASSERT(tx->mxc_get == tx->mxc_put + 1);
        mxlnd_ctx_init(tx);
        tx->mxc_put++;
        spin_lock(&kmxlnd_data.kmx_tx_idle_lock);
        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;
}

/**
 * mxlnd_conn_free - free the conn
 * @conn - a kmx_conn pointer
 *
 * The calling function should remove the conn from the conns list first
 * then destroy it.
 */
void
mxlnd_conn_free(struct kmx_conn *conn)
{
        struct kmx_peer *peer   = conn->mxk_peer;

        CDEBUG(D_NET, "freeing conn 0x%p *****\n", conn);
        LASSERT (list_empty (&conn->mxk_tx_credit_queue) &&
                 list_empty (&conn->mxk_tx_free_queue) &&
                 list_empty (&conn->mxk_pending));
        if (!list_empty(&conn->mxk_list)) {
                spin_lock(&peer->mxp_lock);
                list_del_init(&conn->mxk_list);
                if (peer->mxp_conn == conn) {
                        peer->mxp_conn = NULL;
                        if (!(conn->mxk_epa.stuff[0] == 0 && conn->mxk_epa.stuff[1] == 0)) {
                                mx_set_endpoint_addr_context(conn->mxk_epa,
                                                             (void *) NULL);
                        }
                }
                spin_unlock(&peer->mxp_lock);
        }
        mxlnd_peer_decref(conn->mxk_peer); /* drop conn's ref to peer */
        MXLND_FREE (conn, sizeof (*conn));
        return;
}


void
mxlnd_conn_cancel_pending_rxs(struct kmx_conn *conn)
{
        int                     found   = 0;
        struct kmx_ctx          *ctx    = NULL;
        struct kmx_ctx          *next   = NULL;
        mx_return_t             mxret   = MX_SUCCESS;
        u32                     result  = 0;

        do {
                found = 0;
                spin_lock(&conn->mxk_lock);
                list_for_each_entry_safe(ctx, next, &conn->mxk_pending, mxc_list) {
                        /* we will delete all including txs */
                        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) {