iiblnd_cb.c

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

                                         msg->ibm_u.completion.ibcm_cookie);
                break;

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

                spin_lock(&conn->ibc_lock);
                tx = kibnal_find_waiting_tx_locked(conn, IBNAL_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 = kibnal_init_rdma(tx, IBNAL_MSG_PUT_DONE, 
                                       kibnal_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);
                if (tx->tx_status == 0 && rc2 < 0)
                        tx->tx_status = rc2;
                tx->tx_waiting = 0;             /* clear waiting and queue atomically */
                kibnal_queue_tx_locked(tx, conn);
                spin_unlock(&conn->ibc_lock);
                break;
                
        case IBNAL_MSG_PUT_DONE:
                /* This buffer was pre-reserved by not returning the credit
                 * when the PUT_REQ's buffer was reposted, so I just return it
                 * now */
                kibnal_handle_completion(conn, IBNAL_MSG_PUT_ACK,
                                         msg->ibm_u.completion.ibcm_status,
                                         msg->ibm_u.completion.ibcm_cookie);
                break;

        case IBNAL_MSG_GET_REQ:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.get.ibgm_hdr,
                                msg->ibm_srcnid, rx, 1);
                repost = rc < 0;                /* repost on error */
                break;

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

                kibnal_handle_completion(conn, IBNAL_MSG_GET_REQ,
                                         msg->ibm_u.completion.ibcm_status,
                                         msg->ibm_u.completion.ibcm_cookie);
                break;
        }

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

        if (repost) {
                if (conn->ibc_version == IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD)
                        rsrvd_credit = 0;       /* peer isn't pre-reserving */

                kibnal_post_rx(rx, !rsrvd_credit, rsrvd_credit);
        }
}

void
kibnal_rx_complete (IB_WORK_COMPLETION *wc, __u64 rxseq)
{
        kib_rx_t     *rx = (kib_rx_t *)kibnal_wreqid2ptr(wc->WorkReqId);
        int           nob = wc->Length;
        kib_msg_t    *msg = rx->rx_msg;
        kib_conn_t   *conn = rx->rx_conn;
        unsigned long flags;
        int           rc;
        int           err = -EIO;

        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
         * disconnecting */
        if (conn->ibc_state > IBNAL_CONN_ESTABLISHED)
                goto ignore;

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

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

        rx->rx_nob = nob;                       /* Now I know nob > 0 */
        mb();

        if (msg->ibm_srcnid != conn->ibc_peer->ibp_nid ||
            msg->ibm_dstnid != kibnal_data.kib_ni->ni_nid ||
            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;
        }

        if (msg->ibm_seq != rxseq) {
                CERROR ("Out-of-sequence rx from %s"
                        ": got "LPD64" but expected "LPD64"\n",
                        libcfs_nid2str(conn->ibc_peer->ibp_nid),
                        msg->ibm_seq, rxseq);
                goto failed;
        }

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

        /* racing with connection establishment/teardown! */

        if (conn->ibc_state < IBNAL_CONN_ESTABLISHED) {
                write_lock_irqsave(&kibnal_data.kib_global_lock, flags);
                /* must check holding global lock to eliminate race */
                if (conn->ibc_state < IBNAL_CONN_ESTABLISHED) {
                        list_add_tail(&rx->rx_list, &conn->ibc_early_rxs);
                        write_unlock_irqrestore(&kibnal_data.kib_global_lock, 
                                                flags);
                        return;
                }
                write_unlock_irqrestore(&kibnal_data.kib_global_lock, 
                                        flags);
        }
        kibnal_handle_rx(rx);
        return;
        
 failed:
        kibnal_close_conn(conn, err);
 ignore:
        /* Don't re-post rx & drop its ref on conn */
        kibnal_conn_decref(conn);
}

struct page *
kibnal_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 !IBNAL_USE_FMR
int
kibnal_append_rdfrag(kib_rdma_desc_t *rd, int active, struct page *page, 
                     unsigned long page_offset, unsigned long len)
{
        kib_rdma_frag_t *frag = &rd->rd_frags[rd->rd_nfrag];

        if (rd->rd_nfrag >= IBNAL_MAX_RDMA_FRAGS) {
                CERROR ("Too many RDMA fragments\n");
                return -EMSGSIZE;
        }

        if (active) {
                if (rd->rd_nfrag == 0)
                        rd->rd_key = kibnal_data.kib_whole_mem.md_lkey;
        } else {
                if (rd->rd_nfrag == 0)
                        rd->rd_key = kibnal_data.kib_whole_mem.md_rkey;
        }

        frag->rf_nob  = len;
        frag->rf_addr = kibnal_data.kib_whole_mem.md_addr +
                        lnet_page2phys(page) + page_offset;

        CDEBUG(D_NET,"map key %x frag [%d]["LPX64" for %d]\n", 
               rd->rd_key, rd->rd_nfrag, frag->rf_addr, frag->rf_nob);

        rd->rd_nfrag++;
        return 0;
}

int
kibnal_setup_rd_iov(kib_tx_t *tx, kib_rdma_desc_t *rd, int active,
                    unsigned int niov, struct iovec *iov, int offset, int nob)
                 
{
        int           fragnob;
        int           rc;
        unsigned long vaddr;
        struct page  *page;
        int           page_offset;

        LASSERT (nob > 0);
        LASSERT (niov > 0);
        LASSERT ((rd != tx->tx_rd) == !active);

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

        rd->rd_nfrag = 0;
        do {
                LASSERT (niov > 0);

                vaddr = ((unsigned long)iov->iov_base) + offset;
                page_offset = vaddr & (PAGE_SIZE - 1);
                page = kibnal_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);

                rc = kibnal_append_rdfrag(rd, active, page, 
                                          page_offset, fragnob);
                if (rc != 0)
                        return rc;

                if (offset + fragnob < iov->iov_len) {
                        offset += fragnob;
                } else {
                        offset = 0;
                        iov++;
                        niov--;
                }
                nob -= fragnob;
        } while (nob > 0);
        
        return 0;
}

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            fragnob;
        int            rc;

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

        LASSERT (nob > 0);
        LASSERT (nkiov > 0);
        LASSERT ((rd != tx->tx_rd) == !active);

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

        rd->rd_nfrag = 0;
        do {
                LASSERT (nkiov > 0);
                fragnob = min((int)(kiov->kiov_len - offset), nob);
                
                rc = kibnal_append_rdfrag(rd, active, kiov->kiov_page,
                                          kiov->kiov_offset + offset,
                                          fragnob);
                if (rc != 0)
                        return rc;

                offset = 0;
                kiov++;
                nkiov--;
                nob -= fragnob;
        } while (nob > 0);

        return 0;
}
#else
int
kibnal_map_tx (kib_tx_t *tx, kib_rdma_desc_t *rd, int active,
               int npages, unsigned long page_offset, int nob)
{
        IB_ACCESS_CONTROL access = {0,};
        FSTATUS           frc;

        LASSERT ((rd != tx->tx_rd) == !active);
        LASSERT (!tx->tx_md.md_active);
        LASSERT (tx->tx_md.md_fmrcount > 0);
        LASSERT (page_offset < PAGE_SIZE);
        LASSERT (npages >= (1 + ((page_offset + nob - 1)>>PAGE_SHIFT)));
        LASSERT (npages <= LNET_MAX_IOV);

        if (!active) {
                // access.s.MWBindable = 1;
                access.s.LocalWrite = 1;
                access.s.RdmaWrite = 1;
        }

        /* Map the memory described by tx->tx_pages
        frc = iibt_register_physical_memory(kibnal_data.kib_hca,
                                            IBNAL_RDMA_BASE,
                                            tx->tx_pages, npages,
                                            page_offset,
                                            kibnal_data.kib_pd,
                                            access,
                                            &tx->tx_md.md_handle,
                                            &tx->tx_md.md_addr,
                                            &tx->tx_md.md_lkey,
                                            &tx->tx_md.md_rkey);
        */
        return -EINVAL;
}

int
kibnal_setup_rd_iov (kib_tx_t *tx, kib_rdma_desc_t *rd, int active,
                     unsigned int niov, struct iovec *iov, int offset, int nob)
                 
{
        int           resid;
        int           fragnob;
        struct page  *page;
        int           npages;
        unsigned long page_offset;
        unsigned long vaddr;

        LASSERT (nob > 0);