iiblnd_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]>
 *
 *   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 "iiblnd.h"

void
hexdump(char *string, void *ptr, int len)
{
        unsigned char *c = ptr;
        int i;

        return;

        if (len < 0 || len > 2048)  {
                printk("XXX what the hell? %d\n",len);
                return;
        }

        printk("%d bytes of '%s' from 0x%p\n", len, string, ptr);

        for (i = 0; i < len;) {
                printk("%02x",*(c++));
                i++;
                if (!(i & 15)) {
                        printk("\n");
                } else if (!(i&1)) {
                        printk(" ");
                }
        }

        if(len & 15) {
                printk("\n");
        }
}

void
kibnal_tx_done (kib_tx_t *tx)
{
        lnet_msg_t *lntmsg[2];
        int         rc = tx->tx_status;
        int         i;

        LASSERT (!in_interrupt());
        LASSERT (!tx->tx_queued);               /* mustn't be queued for sending */
        LASSERT (tx->tx_sending == 0);          /* mustn't be awaiting sent callback */
        LASSERT (!tx->tx_waiting);              /* mustn't be awaiting peer response */

#if IBNAL_USE_FMR
        /* Handle unmapping if required */
#endif
        /* tx may have up to 2 lnet msgs to finalise */
        lntmsg[0] = tx->tx_lntmsg[0]; tx->tx_lntmsg[0] = NULL;
        lntmsg[1] = tx->tx_lntmsg[1]; tx->tx_lntmsg[1] = NULL;
        
        if (tx->tx_conn != NULL) {
                kibnal_conn_decref(tx->tx_conn);
                tx->tx_conn = NULL;
        }

        tx->tx_nwrq = 0;
        tx->tx_status = 0;

        spin_lock(&kibnal_data.kib_tx_lock);

        list_add (&tx->tx_list, &kibnal_data.kib_idle_txs);

        spin_unlock(&kibnal_data.kib_tx_lock);

        /* delay finalize until my descs have been freed */
        for (i = 0; i < 2; i++) {
                if (lntmsg[i] == NULL)
                        continue;

                lnet_finalize (kibnal_data.kib_ni, lntmsg[i], rc);
        }
}

kib_tx_t *
kibnal_get_idle_tx (void) 
{
        kib_tx_t      *tx;
        
        spin_lock(&kibnal_data.kib_tx_lock);

        if (list_empty (&kibnal_data.kib_idle_txs)) {
                spin_unlock(&kibnal_data.kib_tx_lock);
                return NULL;
        }

        tx = list_entry (kibnal_data.kib_idle_txs.next, kib_tx_t, tx_list);
        list_del (&tx->tx_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->tx_cookie = kibnal_data.kib_next_tx_cookie++;

        spin_unlock(&kibnal_data.kib_tx_lock);

        LASSERT (tx->tx_nwrq == 0);
        LASSERT (!tx->tx_queued);
        LASSERT (tx->tx_sending == 0);
        LASSERT (!tx->tx_waiting);
        LASSERT (tx->tx_status == 0);
        LASSERT (tx->tx_conn == NULL);
        LASSERT (tx->tx_lntmsg[0] == NULL);
        LASSERT (tx->tx_lntmsg[1] == NULL);
        
        return tx;
}

int
kibnal_post_rx (kib_rx_t *rx, int credit, int rsrvd_credit)
{
        kib_conn_t   *conn = rx->rx_conn;
        int           rc = 0;
        FSTATUS       frc;

        LASSERT (!in_interrupt());
        /* old peers don't reserve rxs for RDMA replies */
        LASSERT (!rsrvd_credit ||
                 conn->ibc_version != IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD);
        
        rx->rx_gl = (IB_LOCAL_DATASEGMENT) {
                .Address = rx->rx_hca_msg,
                .Lkey    = kibnal_data.kib_whole_mem.md_lkey,
                .Length  = IBNAL_MSG_SIZE,
        };

        rx->rx_wrq = (IB_WORK_REQ2) {
                .Next          = NULL,
                .WorkReqId     = kibnal_ptr2wreqid(rx, IBNAL_WID_RX),
                .MessageLen    = IBNAL_MSG_SIZE,
                .DSList        = &rx->rx_gl,
                .DSListDepth   = 1,
                .Operation     = WROpRecv,
        };

        LASSERT (conn->ibc_state >= IBNAL_CONN_CONNECTING);
        LASSERT (rx->rx_nob >= 0);              /* not posted */

        CDEBUG(D_NET, "posting rx [%d %x "LPX64"]\n", 
               rx->rx_wrq.DSList->Length,
               rx->rx_wrq.DSList->Lkey,
               rx->rx_wrq.DSList->Address);

        if (conn->ibc_state > IBNAL_CONN_ESTABLISHED) {
                /* No more posts for this rx; so lose its ref */
                kibnal_conn_decref(conn);
                return 0;
        }
        
        rx->rx_nob = -1;                        /* flag posted */
        mb();

        frc = iba_post_recv2(conn->ibc_qp, &rx->rx_wrq, NULL);
        if (frc == FSUCCESS) {
                if (credit || rsrvd_credit) {
                        spin_lock(&conn->ibc_lock);

                        if (credit)
                                conn->ibc_outstanding_credits++;
                        if (rsrvd_credit)
                                conn->ibc_reserved_credits++;

                        spin_unlock(&conn->ibc_lock);

                        kibnal_check_sends(conn);
                }
                return 0;
        }
        
        CERROR ("post rx -> %s failed %d\n", 
                libcfs_nid2str(conn->ibc_peer->ibp_nid), frc);
        rc = -EIO;
        kibnal_close_conn(rx->rx_conn, rc);
        /* No more posts for this rx; so lose its ref */
        kibnal_conn_decref(conn);
        return rc;
}

int
kibnal_post_receives (kib_conn_t *conn)
{
        int    i;
        int    rc;

        LASSERT (conn->ibc_state == IBNAL_CONN_CONNECTING);

        for (i = 0; i < IBNAL_RX_MSGS; i++) {
                /* +1 ref for rx desc.  This ref remains until kibnal_post_rx
                 * fails (i.e. actual failure or we're disconnecting) */
                kibnal_conn_addref(conn);
                rc = kibnal_post_rx (&conn->ibc_rxs[i], 0, 0);
                if (rc != 0)
                        return rc;
        }

        return 0;
}

kib_tx_t *
kibnal_find_waiting_tx_locked(kib_conn_t *conn, int txtype, __u64 cookie)
{
        struct list_head   *tmp;
        
        list_for_each(tmp, &conn->ibc_active_txs) {
                kib_tx_t *tx = list_entry(tmp, kib_tx_t, tx_list);
                
                LASSERT (!tx->tx_queued);
                LASSERT (tx->tx_sending != 0 || tx->tx_waiting);

                if (tx->tx_cookie != cookie)
                        continue;

                if (tx->tx_waiting &&
                    tx->tx_msg->ibm_type == txtype)
                        return tx;

                CWARN("Bad completion: %swaiting, type %x (wanted %x)\n",
                      tx->tx_waiting ? "" : "NOT ",
                      tx->tx_msg->ibm_type, txtype);
        }
        return NULL;
}

void
kibnal_handle_completion(kib_conn_t *conn, int txtype, int status, __u64 cookie)
{
        kib_tx_t    *tx;
        int          idle;

        spin_lock(&conn->ibc_lock);

        tx = kibnal_find_waiting_tx_locked(conn, txtype, cookie);
        if (tx == NULL) {
                spin_unlock(&conn->ibc_lock);

                CWARN("Unmatched completion type %x cookie "LPX64" from %s\n",
                      txtype, cookie, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                kibnal_close_conn (conn, -EPROTO);
                return;
        }

        if (tx->tx_status == 0) {               /* success so far */
                if (status < 0) {               /* failed? */
                        tx->tx_status = status;
                } else if (txtype == IBNAL_MSG_GET_REQ) {
                        lnet_set_reply_msg_len(kibnal_data.kib_ni,
                                               tx->tx_lntmsg[1], status);
                }
        }
        
        tx->tx_waiting = 0;

        idle = !tx->tx_queued && (tx->tx_sending == 0);
        if (idle)
                list_del(&tx->tx_list);

        spin_unlock(&conn->ibc_lock);
        
        if (idle)
                kibnal_tx_done(tx);
}

void
kibnal_send_completion (kib_conn_t *conn, int type, int status, __u64 cookie) 
{
        kib_tx_t    *tx = kibnal_get_idle_tx();
        
        if (tx == NULL) {
                CERROR("Can't get tx for completion %x for %s\n",
                       type, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                return;
        }
        
        tx->tx_msg->ibm_u.completion.ibcm_status = status;
        tx->tx_msg->ibm_u.completion.ibcm_cookie = cookie;
        kibnal_init_tx_msg(tx, type, sizeof(kib_completion_msg_t));
        
        kibnal_queue_tx(tx, conn);
}

void
kibnal_handle_rx (kib_rx_t *rx)
{
        kib_msg_t    *msg = rx->rx_msg;
        kib_conn_t   *conn = rx->rx_conn;
        int           credits = msg->ibm_credits;
        kib_tx_t     *tx;
        int           rc = 0;
        int           repost = 1;
        int           rsrvd_credit = 0;
        int           rc2;

        LASSERT (conn->ibc_state >= IBNAL_CONN_ESTABLISHED);

        CDEBUG (D_NET, "Received %x[%d] from %s\n",
                msg->ibm_type, credits, libcfs_nid2str(conn->ibc_peer->ibp_nid));
        
        if (credits != 0) {
                /* Have I received credits that will let me send? */
                spin_lock(&conn->ibc_lock);
                conn->ibc_credits += credits;
                spin_unlock(&conn->ibc_lock);

                kibnal_check_sends(conn);
        }

        switch (msg->ibm_type) {
        default:
                CERROR("Bad IBNAL message type %x from %s\n",
                       msg->ibm_type, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                rc = -EPROTO;
                break;

        case IBNAL_MSG_NOOP:
                break;

        case IBNAL_MSG_IMMEDIATE:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.immediate.ibim_hdr,
                                msg->ibm_srcnid, rx, 0);
                repost = rc < 0;                /* repost on error */
                break;
                
        case IBNAL_MSG_PUT_REQ:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.putreq.ibprm_hdr,
                                msg->ibm_srcnid, rx, 1);
                repost = rc < 0;                /* repost on error */
                break;

        case IBNAL_MSG_PUT_NAK:
                rsrvd_credit = 1;               /* rdma reply (was pre-reserved) */

                CWARN ("PUT_NACK from %s\n", libcfs_nid2str(conn->ibc_peer->ibp_nid));
                kibnal_handle_completion(conn, IBNAL_MSG_PUT_REQ, 
                                         msg->ibm_u.completion.ibcm_status,