viblnd_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]>
 *   Author: Frank Zago <[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 "viblnd.h"

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
        if (tx->tx_md.md_fmrcount == 0 ||
            (rc != 0 && tx->tx_md.md_active)) {
                vv_return_t      vvrc;

                /* mapping must be active (it dropped fmrcount to 0) */
                LASSERT (tx->tx_md.md_active); 

                vvrc = vv_unmap_fmr(kibnal_data.kib_hca,
                                    1, &tx->tx_md.md_fmrhandle);
                LASSERT (vvrc == vv_return_ok);

                tx->tx_md.md_fmrcount = *kibnal_tunables.kib_fmr_remaps;
        }
        tx->tx_md.md_active = 0;
#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);
        }
}

void
kibnal_txlist_done (struct list_head *txlist, int status)
{
        kib_tx_t *tx;

        while (!list_empty (txlist)) {
                tx = list_entry (txlist->next, kib_tx_t, tx_list);

                list_del (&tx->tx_list);
                /* complete now */
                tx->tx_waiting = 0;
                tx->tx_status = status;
                kibnal_tx_done (tx);
        }
}

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;
        __u64         addr = (__u64)((unsigned long)((rx)->rx_msg));
        vv_return_t   vvrc;

        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 = (vv_scatgat_t) {
                .v_address = KIBNAL_ADDR2SG(addr),
                .l_key     = rx->rx_lkey,
                .length    = IBNAL_MSG_SIZE,
        };

        rx->rx_wrq = (vv_wr_t) {
                .wr_id                   = kibnal_ptr2wreqid(rx, IBNAL_WID_RX),
                .completion_notification = 1,
                .scatgat_list            = &rx->rx_gl,
                .num_of_data_segments    = 1,
                .wr_type                 = vv_wr_receive,
        };

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

        CDEBUG(D_NET, "posting rx [%d %x "LPX64"]\n", 
               rx->rx_wrq.scatgat_list->length,
               rx->rx_wrq.scatgat_list->l_key,
               KIBNAL_SG2ADDR(rx->rx_wrq.scatgat_list->v_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 */
        
        spin_lock(&conn->ibc_lock);
        /* Serialise vv_post_receive; it's not re-entrant on the same QP */
        vvrc = vv_post_receive(kibnal_data.kib_hca,
                               conn->ibc_qp, &rx->rx_wrq);

        if (vvrc == vv_return_ok) {
                if (credit)
                        conn->ibc_outstanding_credits++;
                if (rsrvd_credit)
                        conn->ibc_reserved_credits++;

                spin_unlock(&conn->ibc_lock);

                if (credit || rsrvd_credit)
                        kibnal_check_sends(conn);

                return 0;
        }
        
        spin_unlock(&conn->ibc_lock);

        CERROR ("post rx -> %s failed %d\n", 
                libcfs_nid2str(conn->ibc_peer->ibp_nid), vvrc);
        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_ESTABLISHED);
        LASSERT (conn->ibc_comms_error == 0);

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