openiblnd_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 "openiblnd.h"

/*
 *  LIB functions follow
 *
 */
void
kibnal_schedule_tx_done (kib_tx_t *tx)
{
        unsigned long flags;

        spin_lock_irqsave (&kibnal_data.kib_sched_lock, flags);

        list_add_tail(&tx->tx_list, &kibnal_data.kib_sched_txq);
        wake_up (&kibnal_data.kib_sched_waitq);

        spin_unlock_irqrestore(&kibnal_data.kib_sched_lock, flags);
}

void
kibnal_tx_done (kib_tx_t *tx)
{
        lnet_msg_t      *lntmsg[2];
        unsigned long    flags;
        int              i;
        int              rc;

        LASSERT (tx->tx_sending == 0);          /* mustn't be awaiting callback */
        LASSERT (!tx->tx_passive_rdma_wait);    /* mustn't be awaiting RDMA */

        if (in_interrupt()) {
                /* can't deregister memory/flush FMAs/finalize in IRQ context... */
                kibnal_schedule_tx_done(tx);
                return;
        }

        switch (tx->tx_mapped) {
        default:
                LBUG();

        case KIB_TX_UNMAPPED:
                break;
                
        case KIB_TX_MAPPED:
                rc = ib_memory_deregister(tx->tx_md.md_handle.mr);
                LASSERT (rc == 0);
                tx->tx_mapped = KIB_TX_UNMAPPED;
                break;

#if IBNAL_FMR
        case KIB_TX_MAPPED_FMR:
                rc = ib_fmr_deregister(tx->tx_md.md_handle.fmr);
                LASSERT (rc == 0);

#ifndef USING_TSAPI
                /* Somewhat belt-and-braces since the tx's conn has closed if
                 * this was a passive RDMA waiting to complete... */
                if (tx->tx_status != 0)
                        ib_fmr_pool_force_flush(kibnal_data.kib_fmr_pool);
#endif
                tx->tx_mapped = KIB_TX_UNMAPPED;
                break;
#endif
        }

        /* tx may have up to 2 ptlmsgs to finalise */
        lntmsg[0] = tx->tx_lntmsg[0]; tx->tx_lntmsg[0] = NULL;
        lntmsg[1] = tx->tx_lntmsg[1]; tx->tx_lntmsg[1] = NULL;
        rc = tx->tx_status;

        if (tx->tx_conn != NULL) {
                kibnal_conn_decref(tx->tx_conn);
                tx->tx_conn = NULL;
        }

        tx->tx_nsp = 0;
        tx->tx_passive_rdma = 0;
        tx->tx_status = 0;

        spin_lock_irqsave (&kibnal_data.kib_tx_lock, flags);

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

        spin_unlock_irqrestore (&kibnal_data.kib_tx_lock, flags);

        /* 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) 
{
        unsigned long  flags;
        kib_tx_t      *tx;
        
        spin_lock_irqsave (&kibnal_data.kib_tx_lock, flags);

        if (list_empty (&kibnal_data.kib_idle_txs)) {
                spin_unlock_irqrestore (&kibnal_data.kib_tx_lock, flags);
                return NULL;
        }

        tx = list_entry (kibnal_data.kib_idle_txs.next, kib_tx_t, tx_list);
        list_del (&tx->tx_list);

        /* Allocate a new passive RDMA completion cookie.  It might not be
         * needed, but we've got a lock right now and we're unlikely to
         * wrap... */
        tx->tx_passive_rdma_cookie = kibnal_data.kib_next_tx_cookie++;

        spin_unlock_irqrestore (&kibnal_data.kib_tx_lock, flags);

        LASSERT (tx->tx_mapped == KIB_TX_UNMAPPED);
        LASSERT (tx->tx_nsp == 0);
        LASSERT (tx->tx_sending == 0);
        LASSERT (tx->tx_status == 0);
        LASSERT (tx->tx_conn == NULL);
        LASSERT (!tx->tx_passive_rdma);
        LASSERT (!tx->tx_passive_rdma_wait);
        LASSERT (tx->tx_lntmsg[0] == NULL);
        LASSERT (tx->tx_lntmsg[1] == NULL);

        return tx;
}

void
kibnal_complete_passive_rdma(kib_conn_t *conn, __u64 cookie, int status)
{
        struct list_head *ttmp;
        unsigned long     flags;
        int               idle;

        spin_lock_irqsave (&conn->ibc_lock, flags);

        list_for_each (ttmp, &conn->ibc_active_txs) {
                kib_tx_t *tx = list_entry(ttmp, kib_tx_t, tx_list);

                LASSERT (tx->tx_passive_rdma ||
                         !tx->tx_passive_rdma_wait);

                LASSERT (tx->tx_passive_rdma_wait ||
                         tx->tx_sending != 0);

                if (!tx->tx_passive_rdma_wait ||
                    tx->tx_passive_rdma_cookie != cookie)
                        continue;

                CDEBUG(D_NET, "Complete %p "LPD64": %d\n", tx, cookie, status);

                /* XXX Set mlength of reply here */

                tx->tx_status = status;
                tx->tx_passive_rdma_wait = 0;
                idle = (tx->tx_sending == 0);

                if (idle)
                        list_del (&tx->tx_list);

                spin_unlock_irqrestore (&conn->ibc_lock, flags);

                /* I could be racing with tx callbacks.  It's whoever
                 * _makes_ tx idle that frees it */
                if (idle)
                        kibnal_tx_done (tx);
                return;
        }
                
        spin_unlock_irqrestore (&conn->ibc_lock, flags);

        CERROR ("Unmatched (late?) RDMA completion "LPX64" from %s\n",
                cookie, libcfs_nid2str(conn->ibc_peer->ibp_nid));
}

void
kibnal_post_rx (kib_rx_t *rx, int credit, int rsrvd_credit)
{
        kib_conn_t   *conn = rx->rx_conn;
        int           rc;
        unsigned long flags;

        LASSERT(!rsrvd_credit ||
                conn->ibc_version != IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD);

        rx->rx_gl = (struct ib_gather_scatter) {
                .address = rx->rx_vaddr,
                .length  = IBNAL_MSG_SIZE,
                .key     = conn->ibc_rx_pages->ibp_lkey,
        };

        rx->rx_sp = (struct ib_receive_param) {
                .work_request_id        = kibnal_ptr2wreqid(rx, 1),
                .scatter_list           = &rx->rx_gl,
                .num_scatter_entries    = 1,
                .device_specific        = NULL,
                .signaled               = 1,
        };

        LASSERT (conn->ibc_state >= IBNAL_CONN_ESTABLISHED);
        LASSERT (rx->rx_nob >= 0);              /* not posted */
        rx->rx_nob = -1;                        /* is now */
        mb();

        if (conn->ibc_state != IBNAL_CONN_ESTABLISHED)
                rc = -ECONNABORTED;
        else
                rc = kibnal_ib_receive(conn->ibc_qp, &rx->rx_sp);

        if (rc == 0) {
                if (credit || rsrvd_credit) {
                        spin_lock_irqsave(&conn->ibc_lock, flags);

                        if (credit)
                                conn->ibc_outstanding_credits++;
                        if (rsrvd_credit)
                                conn->ibc_reserved_credits++;
                        
                        spin_unlock_irqrestore(&conn->ibc_lock, flags);

                        kibnal_check_sends(conn);
                }
                return;
        }

        if (conn->ibc_state == IBNAL_CONN_ESTABLISHED) {
                CERROR ("Error posting receive -> %s: %d\n",
                        libcfs_nid2str(conn->ibc_peer->ibp_nid), rc);
                kibnal_close_conn (rx->rx_conn, rc);
        } else {
                CDEBUG (D_NET, "Error posting receive -> %s: %d\n",
                        libcfs_nid2str(conn->ibc_peer->ibp_nid), rc);
        }

        /* Drop rx's ref */
        kibnal_conn_decref(conn);
}

void
kibnal_rx_callback (struct ib_cq_entry *e)
{
        kib_rx_t     *rx = (kib_rx_t *)kibnal_wreqid2ptr(e->work_request_id);
        kib_msg_t    *msg = rx->rx_msg;
        kib_conn_t   *conn = rx->rx_conn;
        int           credits;
        unsigned long flags;
        int           rc;
        int           err = -ECONNABORTED;

        CDEBUG (D_NET, "rx %p conn %p\n", rx, conn);
        LASSERT (rx->rx_nob < 0);               /* was posted */
        rx->rx_nob = 0;                         /* isn't now */
        mb();

        /* receives complete with error in any case after we've started
         * closing the QP */
        if (conn->ibc_state >= IBNAL_CONN_DEATHROW)
                goto failed;

        /* We don't post receives until the conn is established */
        LASSERT (conn->ibc_state == IBNAL_CONN_ESTABLISHED);

        if (e->status != IB_COMPLETION_STATUS_SUCCESS) {
                CERROR("Rx from %s failed: %d\n", 
                       libcfs_nid2str(conn->ibc_peer->ibp_nid), e->status);
                goto failed;
        }

        LASSERT (e->bytes_transferred >= 0);
        rx->rx_nob = e->bytes_transferred;
        mb();

        rc = kibnal_unpack_msg(msg, conn->ibc_version, rx->rx_nob);
        if (rc != 0) {
                CERROR ("Error %d unpacking rx from %s\n",
                        rc, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                goto failed;
        }

        if (!lnet_ptlcompat_matchnid(conn->ibc_peer->ibp_nid,
                                     msg->ibm_srcnid) ||
            !lnet_ptlcompat_matchnid(kibnal_data.kib_ni->ni_nid,
                                     msg->ibm_dstnid) ||
            msg->ibm_srcstamp != conn->ibc_incarnation ||
            msg->ibm_dststamp != kibnal_data.kib_incarnation) {
                CERROR ("Stale rx from %s\n",
                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
                err = -ESTALE;
                goto failed;
        }

        /* Have I received credits that will let me send? */
        credits = msg->ibm_credits;
        if (credits != 0) {
                spin_lock_irqsave(&conn->ibc_lock, flags);
                conn->ibc_credits += credits;
                spin_unlock_irqrestore(&conn->ibc_lock, flags);
                
                kibnal_check_sends(conn);
        }

        switch (msg->ibm_type) {
        case IBNAL_MSG_NOOP:
                kibnal_post_rx (rx, 1, 0);
                return;

        case IBNAL_MSG_IMMEDIATE:
                break;
                
        case IBNAL_MSG_PUT_RDMA:
        case IBNAL_MSG_GET_RDMA:
                CDEBUG(D_NET, "%d RDMA: cookie "LPX64", key %x, addr "LPX64", nob %d\n",
                       msg->ibm_type, msg->ibm_u.rdma.ibrm_cookie,
                       msg->ibm_u.rdma.ibrm_desc.rd_key,
                       msg->ibm_u.rdma.ibrm_desc.rd_addr,
                       msg->ibm_u.rdma.ibrm_desc.rd_nob);
                break;
                
        case IBNAL_MSG_PUT_DONE:
        case IBNAL_MSG_GET_DONE:
                CDEBUG(D_NET, "%d DONE: cookie "LPX64", status %d\n",
                       msg->ibm_type, msg->ibm_u.completion.ibcm_cookie,
                       msg->ibm_u.completion.ibcm_status);

                kibnal_complete_passive_rdma (conn, 
                                              msg->ibm_u.completion.ibcm_cookie,
                                              msg->ibm_u.completion.ibcm_status);

                if (conn->ibc_version == IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD) {
                        kibnal_post_rx (rx, 1, 0);
                } else {
                        /* this reply buffer was pre-reserved */
                        kibnal_post_rx (rx, 0, 1);
                }
                return;
                        
        default:
                CERROR ("Bad msg type %x from %s\n",
                        msg->ibm_type, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                goto failed;