iiblnd_cb.c

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

        LASSERT (niov > 0);

        while (offset >= iov->iov_len) {
                offset -= iov->iov_len;
                niov--;
                iov++;
                LASSERT (niov > 0);
        }

        if (nob > iov->iov_len - offset) {
                CERROR ("Can't map multiple vaddr fragments\n");
                return (-EMSGSIZE);
        }

        vaddr = ((unsigned long)iov->iov_base) + offset;
        
        page_offset = vaddr & (PAGE_SIZE - 1);
        resid = nob;
        npages = 0;

        do {
                LASSERT (npages < LNET_MAX_IOV);

                page = kibnal_kvaddr_to_page(vaddr);
                if (page == NULL) {
                        CERROR("Can't find page for %lu\n", vaddr);
                        return -EFAULT;
                }

                tx->tx_pages[npages++] = lnet_page2phys(page);

                fragnob = PAGE_SIZE - (vaddr & (PAGE_SIZE - 1));
                vaddr += fragnob;
                resid -= fragnob;

        } while (resid > 0);

        return kibnal_map_tx(tx, rd, active, npages, page_offset, nob);
}

int
kibnal_setup_rd_kiov (kib_tx_t *tx, kib_rdma_desc_t *rd, int active,
                      int nkiov, lnet_kiov_t *kiov, int offset, int nob)
{
        int            resid;
        int            npages;
        unsigned long  page_offset;
        
        CDEBUG(D_NET, "niov %d offset %d nob %d\n", nkiov, offset, nob);

        LASSERT (nob > 0);
        LASSERT (nkiov > 0);
        LASSERT (nkiov <= LNET_MAX_IOV);
        LASSERT (!tx->tx_md.md_active);
        LASSERT ((rd != tx->tx_rd) == !active);

        while (offset >= kiov->kiov_len) {
                offset -= kiov->kiov_len;
                nkiov--;
                kiov++;
                LASSERT (nkiov > 0);
        }

        page_offset = kiov->kiov_offset + offset;
        
        resid = offset + nob;
        npages = 0;

        do {
                LASSERT (npages < LNET_MAX_IOV);
                LASSERT (nkiov > 0);

                if ((npages > 0 && kiov->kiov_offset != 0) ||
                    (resid > kiov->kiov_len && 
                     (kiov->kiov_offset + kiov->kiov_len) != PAGE_SIZE)) {
                        /* Can't have gaps */
                        CERROR ("Can't make payload contiguous in I/O VM:"
                                "page %d, offset %d, len %d \n",
                                npages, kiov->kiov_offset, kiov->kiov_len);
                        
                        return -EINVAL;
                }

                tx->tx_pages[npages++] = lnet_page2phys(kiov->kiov_page);
                resid -= kiov->kiov_len;
                kiov++;
                nkiov--;
        } while (resid > 0);

        return kibnal_map_tx(tx, rd, active, npages, page_offset, nob);
}
#endif

kib_conn_t *
kibnal_find_conn_locked (kib_peer_t *peer)
{
        struct list_head *tmp;

        /* just return the first connection */
        list_for_each (tmp, &peer->ibp_conns) {
                return (list_entry(tmp, kib_conn_t, ibc_list));
        }

        return (NULL);
}

void
kibnal_check_sends (kib_conn_t *conn)
{
        kib_tx_t       *tx;
        FSTATUS         frc;
        int             rc;
        int             consume_cred;
        int             done;

        LASSERT (conn->ibc_state >= IBNAL_CONN_ESTABLISHED);
        
        spin_lock(&conn->ibc_lock);

        LASSERT (conn->ibc_nsends_posted <=
                *kibnal_tunables.kib_concurrent_sends);
        LASSERT (conn->ibc_reserved_credits >= 0);
        
        while (conn->ibc_reserved_credits > 0 &&
               !list_empty(&conn->ibc_tx_queue_rsrvd)) {
                LASSERT (conn->ibc_version != 
                         IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD);
                tx = list_entry(conn->ibc_tx_queue_rsrvd.next,
                                kib_tx_t, tx_list);
                list_del(&tx->tx_list);
                list_add_tail(&tx->tx_list, &conn->ibc_tx_queue);
                conn->ibc_reserved_credits--;
        }

        if (list_empty(&conn->ibc_tx_queue) &&
            list_empty(&conn->ibc_tx_queue_nocred) &&
            (conn->ibc_outstanding_credits >= IBNAL_CREDIT_HIGHWATER ||
             kibnal_send_keepalive(conn))) {
                spin_unlock(&conn->ibc_lock);
                
                tx = kibnal_get_idle_tx();
                if (tx != NULL)
                        kibnal_init_tx_msg(tx, IBNAL_MSG_NOOP, 0);

                spin_lock(&conn->ibc_lock);
                
                if (tx != NULL)
                        kibnal_queue_tx_locked(tx, conn);
        }

        for (;;) {
                if (!list_empty(&conn->ibc_tx_queue_nocred)) {
                        LASSERT (conn->ibc_version != 
                                 IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD);
                        tx = list_entry (conn->ibc_tx_queue_nocred.next, 
                                         kib_tx_t, tx_list);
                        consume_cred = 0;
                } else if (!list_empty (&conn->ibc_tx_queue)) {
                        tx = list_entry (conn->ibc_tx_queue.next, 
                                         kib_tx_t, tx_list);
                        consume_cred = 1;
                } else {
                        /* nothing waiting */
                        break;
                }

                LASSERT (tx->tx_queued);
                /* We rely on this for QP sizing */
                LASSERT (tx->tx_nwrq > 0 && tx->tx_nwrq <= 1 + IBNAL_MAX_RDMA_FRAGS);

                LASSERT (conn->ibc_outstanding_credits >= 0);
                LASSERT (conn->ibc_outstanding_credits <= IBNAL_MSG_QUEUE_SIZE);
                LASSERT (conn->ibc_credits >= 0);
                LASSERT (conn->ibc_credits <= IBNAL_MSG_QUEUE_SIZE);

                if (conn->ibc_nsends_posted ==
                    *kibnal_tunables.kib_concurrent_sends) {
                        /* We've got some tx completions outstanding... */
                        CDEBUG(D_NET, "%s: posted enough\n",
                               libcfs_nid2str(conn->ibc_peer->ibp_nid));
                        break;
                }

                if (consume_cred) {
                        if (conn->ibc_credits == 0) {   /* no credits */
                                CDEBUG(D_NET, "%s: no credits\n",
                                       libcfs_nid2str(conn->ibc_peer->ibp_nid));
                                break;
                        }
                        
                        if (conn->ibc_credits == 1 &&   /* last credit reserved for */
                            conn->ibc_outstanding_credits == 0) { /* giving back credits */
                                CDEBUG(D_NET, "%s: not using last credit\n",
                                       libcfs_nid2str(conn->ibc_peer->ibp_nid));
                                break;
                        }
                }
                
                list_del (&tx->tx_list);
                tx->tx_queued = 0;

                /* NB don't drop ibc_lock before bumping tx_sending */

                if (tx->tx_msg->ibm_type == IBNAL_MSG_NOOP &&
                    (!list_empty(&conn->ibc_tx_queue) ||
                     !list_empty(&conn->ibc_tx_queue_nocred) ||
                     (conn->ibc_outstanding_credits < IBNAL_CREDIT_HIGHWATER &&
                      !kibnal_send_keepalive(conn)))) {
                        /* redundant NOOP */
                        spin_unlock(&conn->ibc_lock);
                        kibnal_tx_done(tx);
                        spin_lock(&conn->ibc_lock);
                        CDEBUG(D_NET, "%s: redundant noop\n",
                               libcfs_nid2str(conn->ibc_peer->ibp_nid));
                        continue;
                }

                kibnal_pack_msg(tx->tx_msg, conn->ibc_version,
                                conn->ibc_outstanding_credits,
                                conn->ibc_peer->ibp_nid, conn->ibc_incarnation,
                                conn->ibc_txseq);

                conn->ibc_txseq++;
                conn->ibc_outstanding_credits = 0;
                conn->ibc_nsends_posted++;
                if (consume_cred)
                        conn->ibc_credits--;

                /* CAVEAT EMPTOR!  This tx could be the PUT_DONE of an RDMA
                 * PUT.  If so, it was first queued here as a PUT_REQ, sent and
                 * stashed on ibc_active_txs, matched by an incoming PUT_ACK,
                 * and then re-queued here.  It's (just) possible that
                 * tx_sending is non-zero if we've not done the tx_complete() from
                 * the first send; hence the ++ rather than = below. */
                tx->tx_sending++;

                list_add (&tx->tx_list, &conn->ibc_active_txs);

                LASSERT (tx->tx_nwrq > 0);

                rc = 0;
                frc = FSUCCESS;
                if (conn->ibc_state != IBNAL_CONN_ESTABLISHED) {
                        rc = -ECONNABORTED;
                } else {
                        frc = iba_post_send2(conn->ibc_qp, tx->tx_wrq, NULL);
                        if (frc != FSUCCESS)
                                rc = -EIO;
                }

                conn->ibc_last_send = jiffies;

                if (rc != 0) {
                        /* NB credits are transferred in the actual
                         * message, which can only be the last work item */
                        conn->ibc_outstanding_credits += tx->tx_msg->ibm_credits;
                        if (consume_cred)
                                conn->ibc_credits++;
                        conn->ibc_nsends_posted--;

                        tx->tx_status = rc;
                        tx->tx_waiting = 0;
                        tx->tx_sending--;
                        
                        done = (tx->tx_sending == 0);
                        if (done)
                                list_del (&tx->tx_list);
                        
                        spin_unlock(&conn->ibc_lock);
                        
                        if (conn->ibc_state == IBNAL_CONN_ESTABLISHED)
                                CERROR ("Error %d posting transmit to %s\n", 
                                        frc, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                        else
                                CDEBUG (D_NET, "Error %d posting transmit to %s\n",
                                        rc, libcfs_nid2str(conn->ibc_peer->ibp_nid));

                        kibnal_close_conn (conn, rc);

                        if (done)
                                kibnal_tx_done (tx);
                        return;
                }
        }

        spin_unlock(&conn->ibc_lock);
}

void
kibnal_tx_complete (IB_WORK_COMPLETION *wc)
{
        kib_tx_t     *tx = (kib_tx_t *)kibnal_wreqid2ptr(wc->WorkReqId);
        kib_conn_t   *conn = tx->tx_conn;
        int           failed = wc->Status != WRStatusSuccess;
        int           idle;

        CDEBUG(D_NET, "%s: sending %d nwrq %d status %d\n", 
               libcfs_nid2str(conn->ibc_peer->ibp_nid),
               tx->tx_sending, tx->tx_nwrq, wc->Status);

        LASSERT (tx->tx_sending > 0);

        if (failed &&
            tx->tx_status == 0 &&
            conn->ibc_state == IBNAL_CONN_ESTABLISHED) {
#if KIBLND_DETAILED_DEBUG
                int                   i;
                IB_WORK_REQ2         *wrq = &tx->tx_wrq[0];
                IB_LOCAL_DATASEGMENT *gl = &tx->tx_gl[0];
                lnet_msg_t           *lntmsg = tx->tx_lntmsg[0];
#endif
                CDEBUG(D_NETERROR, "tx -> %s type %x cookie "LPX64
                       " sending %d waiting %d failed %d nwrk %d\n", 
                       libcfs_nid2str(conn->ibc_peer->ibp_nid),
                       tx->tx_msg->ibm_type, tx->tx_cookie,
                       tx->tx_sending, tx->tx_waiting, wc->Status,
                       tx->tx_nwrq);
#if KIBLND_DETAILED_DEBUG
                for (i = 0; i < tx->tx_nwrq; i++, wrq++, gl++) {
                        switch (wrq->Operation) {
                        default:
                                CDEBUG(D_NETERROR, "    [%3d] Addr %p Next %p OP %d "
                                       "DSList %p(%p)/%d: "LPX64"/%d K %x\n",
                                       i, wrq, wrq->Next, wrq->Operation,
                                       wrq->DSList, gl, wrq->DSListDepth,
                                       gl->Address, gl->Length, gl->Lkey);
                                break;
                        case WROpSend:
                                CDEBUG(D_NETERROR, "    [%3d] Addr %p Next %p SEND "
                                       "DSList %p(%p)/%d: "LPX64"/%d K %x\n",
                                       i, wrq, wrq->Next, 
                                       wrq->DSList, gl, wrq->DSListDepth,
                                       gl->Address, gl->Length, gl->Lkey);
                                break;
                        case WROpRdmaWrite:
                                CDEBUG(D_NETERROR, "    [%3d] Addr %p Next %p DMA "
                                       "DSList: %p(%p)/%d "LPX64"/%d K %x -> "
                                       LPX64" K %x\n",
                                       i, wrq, wrq->Next, 
                                       wrq->DSList, gl, wrq->DSListDepth,
                                       gl->Address, gl->Length, gl->Lkey,
                                       wrq->Req.SendRC.RemoteDS.Address,
                                       wrq->Req.SendRC.RemoteDS.Rkey);
                                break;
                        }
                }
                
                switch (tx->tx_msg->ibm_type) {
                default:
                        CDEBUG(D_NETERROR, "  msg type %x %p/%d, No RDMA\n", 
                               tx->tx_msg->ibm_type, 
                               tx->tx_msg, tx->tx_msg->ibm_nob);
                        break;

                case IBNAL_MSG_PUT_DONE:
                case IBNAL_MSG_GET_DONE:
                        CDEBUG(D_NETERROR, "  msg type %x %p/%d, RDMA key %x frags %d...\n",