o2iblnd_cb.c

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

                        CERROR("Bad credits from %s: %d + %d > %d\n",
                               libcfs_nid2str(conn->ibc_peer->ibp_nid),
                               rc2, credits, IBLND_MSG_QUEUE_SIZE);

                        kiblnd_close_conn(conn, -EPROTO);
                        kiblnd_post_rx(rx, IBLND_POSTRX_NO_CREDIT);
                        return;
                }

                conn->ibc_credits += credits;

                spin_unlock(&conn->ibc_lock);
                kiblnd_check_sends(conn);
        }

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

        case IBLND_MSG_NOOP:
                post_credit = IBLND_POSTRX_PEER_CREDIT;
                break;

        case IBLND_MSG_IMMEDIATE:
                post_credit = IBLND_POSTRX_DONT_POST;
                rc = lnet_parse(ni, &msg->ibm_u.immediate.ibim_hdr,
                                msg->ibm_srcnid, rx, 0);
                if (rc < 0)                     /* repost on error */
                        post_credit = IBLND_POSTRX_PEER_CREDIT;
                break;

        case IBLND_MSG_PUT_REQ:
                post_credit = IBLND_POSTRX_DONT_POST;
                rc = lnet_parse(ni, &msg->ibm_u.putreq.ibprm_hdr,
                                msg->ibm_srcnid, rx, 1);
                if (rc < 0)                     /* repost on error */
                        post_credit = IBLND_POSTRX_PEER_CREDIT;
                break;

        case IBLND_MSG_PUT_NAK:
                CWARN ("PUT_NACK from %s\n", libcfs_nid2str(conn->ibc_peer->ibp_nid));
                post_credit = IBLND_POSTRX_RSRVD_CREDIT;
                kiblnd_handle_completion(conn, IBLND_MSG_PUT_REQ,
                                         msg->ibm_u.completion.ibcm_status,
                                         msg->ibm_u.completion.ibcm_cookie);
                break;

        case IBLND_MSG_PUT_ACK:
                post_credit = IBLND_POSTRX_RSRVD_CREDIT;

                spin_lock(&conn->ibc_lock);
                tx = kiblnd_find_waiting_tx_locked(conn, IBLND_MSG_PUT_REQ,
                                                   msg->ibm_u.putack.ibpam_src_cookie);
                if (tx != NULL)
                        list_del(&tx->tx_list);
                spin_unlock(&conn->ibc_lock);

                if (tx == NULL) {
                        CERROR("Unmatched PUT_ACK from %s\n",
                               libcfs_nid2str(conn->ibc_peer->ibp_nid));
                        rc = -EPROTO;
                        break;
                }

                LASSERT (tx->tx_waiting);
                /* CAVEAT EMPTOR: I could be racing with tx_complete, but...
                 * (a) I can overwrite tx_msg since my peer has received it!
                 * (b) tx_waiting set tells tx_complete() it's not done. */

                tx->tx_nwrq = 0;                /* overwrite PUT_REQ */

                rc2 = kiblnd_init_rdma(ni, tx, IBLND_MSG_PUT_DONE,
                                       kiblnd_rd_size(&msg->ibm_u.putack.ibpam_rd),
                                       &msg->ibm_u.putack.ibpam_rd,
                                       msg->ibm_u.putack.ibpam_dst_cookie);
                if (rc2 < 0)
                        CERROR("Can't setup rdma for PUT to %s: %d\n",
                               libcfs_nid2str(conn->ibc_peer->ibp_nid), rc2);

                spin_lock(&conn->ibc_lock);
                tx->tx_waiting = 0;             /* clear waiting and queue atomically */
                kiblnd_queue_tx_locked(tx, conn);
                spin_unlock(&conn->ibc_lock);
                break;

        case IBLND_MSG_PUT_DONE:
                post_credit = IBLND_POSTRX_PEER_CREDIT;
                kiblnd_handle_completion(conn, IBLND_MSG_PUT_ACK,
                                         msg->ibm_u.completion.ibcm_status,
                                         msg->ibm_u.completion.ibcm_cookie);
                break;

        case IBLND_MSG_GET_REQ:
                post_credit = IBLND_POSTRX_DONT_POST;
                rc = lnet_parse(ni, &msg->ibm_u.get.ibgm_hdr,
                                msg->ibm_srcnid, rx, 1);
                if (rc < 0)                     /* repost on error */
                        post_credit = IBLND_POSTRX_PEER_CREDIT;
                break;

        case IBLND_MSG_GET_DONE:
                post_credit = IBLND_POSTRX_RSRVD_CREDIT;
                kiblnd_handle_completion(conn, IBLND_MSG_GET_REQ,
                                         msg->ibm_u.completion.ibcm_status,
                                         msg->ibm_u.completion.ibcm_cookie);
                break;
        }

        if (rc < 0)                             /* protocol error */
                kiblnd_close_conn(conn, rc);

        if (post_credit != IBLND_POSTRX_DONT_POST)
                kiblnd_post_rx(rx, post_credit);
}

void
kiblnd_rx_complete (kib_rx_t *rx, int status, int nob)
{
        kib_msg_t    *msg = rx->rx_msg;
        kib_conn_t   *conn = rx->rx_conn;
        lnet_ni_t    *ni = conn->ibc_peer->ibp_ni;
        kib_net_t    *net = ni->ni_data;
        unsigned long flags;
        int           rc;
        int           err = -EIO;

        LASSERT (net != NULL);
        LASSERT (rx->rx_nob < 0);               /* was posted */
        rx->rx_nob = 0;                         /* isn't now */
        
        if (conn->ibc_state > IBLND_CONN_ESTABLISHED)
                goto ignore;

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

        LASSERT (nob >= 0);
        rx->rx_nob = nob;

        rc = kiblnd_unpack_msg(msg, 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 (msg->ibm_srcnid != conn->ibc_peer->ibp_nid ||
            msg->ibm_dstnid != ni->ni_nid ||
            msg->ibm_srcstamp != conn->ibc_incarnation ||
            msg->ibm_dststamp != net->ibn_incarnation) {
                CERROR ("Stale rx from %s\n",
                        libcfs_nid2str(conn->ibc_peer->ibp_nid));
                err = -ESTALE;
                goto failed;
        }

        /* set time last known alive */
        kiblnd_peer_alive(conn->ibc_peer);

        /* racing with connection establishment/teardown! */

        if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
                write_lock_irqsave(&kiblnd_data.kib_global_lock, flags);
                /* must check holding global lock to eliminate race */
                if (conn->ibc_state < IBLND_CONN_ESTABLISHED) {
                        list_add_tail(&rx->rx_list, &conn->ibc_early_rxs);
                        write_unlock_irqrestore(&kiblnd_data.kib_global_lock,
                                                flags);
                        return;
                }
                write_unlock_irqrestore(&kiblnd_data.kib_global_lock,
                                        flags);
        }
        kiblnd_handle_rx(rx);
        return;

 failed:
        CDEBUG(D_NET, "rx %p conn %p\n", rx, conn);
        kiblnd_close_conn(conn, err);
 ignore:
        kiblnd_drop_rx(rx);                     /* Don't re-post rx. */
}

struct page *
kiblnd_kvaddr_to_page (unsigned long vaddr)
{
        struct page *page;

        if (vaddr >= VMALLOC_START &&
            vaddr < VMALLOC_END) {
                page = vmalloc_to_page ((void *)vaddr);
                LASSERT (page != NULL);
                return page;
        }
#ifdef CONFIG_HIGHMEM
        if (vaddr >= PKMAP_BASE &&
            vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE)) {
                /* No highmem pages only used for bulk (kiov) I/O */
                CERROR("find page for address in highmem\n");
                LBUG();
        }
#endif
        page = virt_to_page (vaddr);
        LASSERT (page != NULL);
        return page;
}

#if !IBLND_MAP_ON_DEMAND
int
kiblnd_setup_rd_iov(lnet_ni_t *ni, kib_tx_t *tx, kib_rdma_desc_t *rd, 
                    unsigned int niov, struct iovec *iov, int offset, int nob)
                 
{
        struct scatterlist *sg;
        int                 i;
        int                 fragnob;
        unsigned long       vaddr;
        struct page        *page;
        int                 page_offset;
        kib_net_t          *net = ni->ni_data;

        LASSERT (nob > 0);
        LASSERT (niov > 0);
        LASSERT (net != NULL);

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

        sg = tx->tx_frags;
        do {
                LASSERT (niov > 0);

                vaddr = ((unsigned long)iov->iov_base) + offset;
                page_offset = vaddr & (PAGE_SIZE - 1);
                page = kiblnd_kvaddr_to_page(vaddr);
                if (page == NULL) {
                        CERROR ("Can't find page\n");
                        return -EFAULT;
                }

                fragnob = min((int)(iov->iov_len - offset), nob);
                fragnob = min(fragnob, (int)PAGE_SIZE - page_offset);

                sg->page = page;
                sg->offset = page_offset;
                sg->length = fragnob;
                sg++;

                if (offset + fragnob < iov->iov_len) {
                        offset += fragnob;
                } else {
                        offset = 0;
                        iov++;
                        niov--;
                }
                nob -= fragnob;
        } while (nob > 0);
        
        /* If rd is not tx_rd, it's going to get sent to a peer and I'm the
         * RDMA sink */
        tx->tx_nfrags = sg - tx->tx_frags;
        tx->tx_dmadir = (rd != tx->tx_rd) ? DMA_FROM_DEVICE : DMA_TO_DEVICE;

        rd->rd_nfrags = kiblnd_dma_map_sg(net->ibn_dev->ibd_cmid->device,
                                          tx->tx_frags, tx->tx_nfrags,
                                          tx->tx_dmadir);
        rd->rd_key    = (rd != tx->tx_rd) ? 
                        net->ibn_dev->ibd_mr->rkey : net->ibn_dev->ibd_mr->lkey;

        for (i = 0; i < rd->rd_nfrags; i++) {
                rd->rd_frags[i].rf_nob  = kiblnd_sg_dma_len(
                        net->ibn_dev->ibd_cmid->device, &tx->tx_frags[i]);
                rd->rd_frags[i].rf_addr = kiblnd_sg_dma_address(
                        net->ibn_dev->ibd_cmid->device, &tx->tx_frags[i]);
        }
        
        return 0;
}

int
kiblnd_setup_rd_kiov (lnet_ni_t *ni, kib_tx_t *tx, kib_rdma_desc_t *rd, 
                      int nkiov, lnet_kiov_t *kiov, int offset, int nob)
{
        struct scatterlist *sg;
        int                 i;
        int                 fragnob;
        kib_net_t          *net = ni->ni_data;

        CDEBUG(D_NET, "niov %d offset %d nob %d\n", nkiov, offset, nob);

        LASSERT (nob > 0);
        LASSERT (nkiov > 0);
        LASSERT (net != NULL);

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

        sg = tx->tx_frags;
        do {
                LASSERT (nkiov > 0);

                fragnob = min((int)(kiov->kiov_len - offset), nob);

                memset(sg, 0, sizeof(*sg));
                sg->page = kiov->kiov_page;
                sg->offset = kiov->kiov_offset + offset;
                sg->length = fragnob;
                sg++;
                
                offset = 0;
                kiov++;
                nkiov--;
                nob -= fragnob;
        } while (nob > 0);

        /* If rd is not tx_rd, it's going to get sent to a peer and I'm the
         * RDMA sink */
        tx->tx_nfrags = sg - tx->tx_frags;
        tx->tx_dmadir = (rd != tx->tx_rd) ? DMA_FROM_DEVICE : DMA_TO_DEVICE;

        rd->rd_nfrags = kiblnd_dma_map_sg(net->ibn_dev->ibd_cmid->device,
                                          tx->tx_frags, tx->tx_nfrags, tx->tx_dmadir);
        rd->rd_key    = (rd != tx->tx_rd) ? 
                        net->ibn_dev->ibd_mr->rkey : net->ibn_dev->ibd_mr->lkey;

        for (i = 0; i < tx->tx_nfrags; i++) {
                rd->rd_frags[i].rf_nob  = kiblnd_sg_dma_len(
                        net->ibn_dev->ibd_cmid->device, &tx->tx_frags[i]);
                rd->rd_frags[i].rf_addr = kiblnd_sg_dma_address(
                        net->ibn_dev->ibd_cmid->device, &tx->tx_frags[i]);
#if 0
                CDEBUG(D_WARNING,"frag[%d]: "LPX64" for %d\n",
                       i, rd->rd_frags[i].rf_addr, rd->rd_frags[i].rf_nob);
#endif
        }
        
        return 0;
}
#else
int
kiblnd_map_tx (lnet_ni_t *ni, kib_tx_t *tx, kib_rdma_desc_t *rd,
               int npages, unsigned long page_offset, int nob)
{
        struct ib_pool_fmr *fmr;
        kib_net_t          *net = ni->ni_data;

        LASSERT (net != NULL);
        LASSERT (tx->tx_fmr == NULL);
        LASSERT (page_offset < PAGE_SIZE);
        LASSERT (npages >= (1 + ((page_offset + nob - 1)>>PAGE_SHIFT)));
        LASSERT (npages <= LNET_MAX_IOV);