openiblnd_cb.c

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

        }

        kibnal_peer_alive(conn->ibc_peer);

        /* schedule for kibnal_rx() in thread context */
        spin_lock_irqsave(&kibnal_data.kib_sched_lock, flags);
        
        list_add_tail (&rx->rx_list, &kibnal_data.kib_sched_rxq);
        wake_up (&kibnal_data.kib_sched_waitq);
        
        spin_unlock_irqrestore(&kibnal_data.kib_sched_lock, flags);
        return;
        
 failed:
        CDEBUG(D_NET, "rx %p conn %p\n", rx, conn);
        kibnal_close_conn(conn, err);

        /* Don't re-post rx & drop its ref on conn */
        kibnal_conn_decref(conn);
}

void
kibnal_rx (kib_rx_t *rx)
{
        int          rc = 0;
        kib_msg_t   *msg = rx->rx_msg;

        switch (msg->ibm_type) {
        case IBNAL_MSG_GET_RDMA:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.rdma.ibrm_hdr,
                                msg->ibm_srcnid, rx, 1);
                break;
                
        case IBNAL_MSG_PUT_RDMA:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.rdma.ibrm_hdr,
                                msg->ibm_srcnid, rx, 1);
                break;

        case IBNAL_MSG_IMMEDIATE:
                rc = lnet_parse(kibnal_data.kib_ni, &msg->ibm_u.immediate.ibim_hdr,
                                msg->ibm_srcnid, rx, 0);
                break;

        default:
                LBUG();
                break;
        }

        if (rc < 0) {
                kibnal_close_conn(rx->rx_conn, rc);
                kibnal_post_rx (rx, 1, 0);
        }
}

#if 0
int
kibnal_kvaddr_to_phys (unsigned long vaddr, __u64 *physp)
{
        struct page *page;

        if (vaddr >= VMALLOC_START &&
            vaddr < VMALLOC_END)
                page = vmalloc_to_page ((void *)vaddr);
#ifdef CONFIG_HIGHMEM
        else if (vaddr >= PKMAP_BASE &&
                 vaddr < (PKMAP_BASE + LAST_PKMAP * PAGE_SIZE))
                page = vmalloc_to_page ((void *)vaddr);
        /* in 2.4 ^ just walks the page tables */
#endif
        else
                page = virt_to_page (vaddr);

        if (page == NULL ||
            !VALID_PAGE (page))
                return (-EFAULT);

        *physp = lnet_page2phys(page) + (vaddr & (PAGE_SIZE - 1));
        return (0);
}
#endif

int
kibnal_map_iov (kib_tx_t *tx, int access,
                unsigned int niov, struct iovec *iov, int offset, int nob)
                 
{
        void   *vaddr;
        int     rc;

        LASSERT (nob > 0);
        LASSERT (niov > 0);
        LASSERT (tx->tx_mapped == KIB_TX_UNMAPPED);

        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 = (void *)(((unsigned long)iov->iov_base) + offset);
        tx->tx_md.md_addr = (__u64)((unsigned long)vaddr);

        rc = ib_memory_register (kibnal_data.kib_pd,
                                 vaddr, nob,
                                 access,
                                 &tx->tx_md.md_handle.mr,
                                 &tx->tx_md.md_lkey,
                                 &tx->tx_md.md_rkey);
        
        if (rc != 0) {
                CERROR ("Can't map vaddr: %d\n", rc);
                return (rc);
        }

        tx->tx_mapped = KIB_TX_MAPPED;
        return (0);
}

int
kibnal_map_kiov (kib_tx_t *tx, int access,
                  int nkiov, lnet_kiov_t *kiov,
                  int offset, int nob)
{
#if IBNAL_FMR
        __u64                      *phys;
        const int                   mapped = KIB_TX_MAPPED_FMR;
#else
        struct ib_physical_buffer  *phys;
        const int                   mapped = KIB_TX_MAPPED;
#endif
        int                         page_offset;
        int                         nphys;
        int                         resid;
        int                         phys_size;
        int                         rc;

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

        LASSERT (nob > 0);
        LASSERT (nkiov > 0);
        LASSERT (tx->tx_mapped == KIB_TX_UNMAPPED);

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

        phys_size = nkiov * sizeof (*phys);
        LIBCFS_ALLOC(phys, phys_size);
        if (phys == NULL) {
                CERROR ("Can't allocate tmp phys\n");
                return (-ENOMEM);
        }

        page_offset = kiov->kiov_offset + offset;
#if IBNAL_FMR
        phys[0] = lnet_page2phys(kiov->kiov_page);
#else
        phys[0].address = lnet_page2phys(kiov->kiov_page);
        phys[0].size = PAGE_SIZE;
#endif
        nphys = 1;
        resid = nob - (kiov->kiov_len - offset);

        while (resid > 0) {
                kiov++;
                nkiov--;
                LASSERT (nkiov > 0);

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

                        for (i = -nphys; i < nkiov; i++) 
                        {
                                CERROR("kiov[%d] %p +%d for %d\n",
                                       i, kiov[i].kiov_page, kiov[i].kiov_offset, kiov[i].kiov_len);
                        }
                        
                        rc = -EINVAL;
                        goto out;
                }

                if (nphys == LNET_MAX_IOV) {
                        CERROR ("payload too big (%d)\n", nphys);
                        rc = -EMSGSIZE;
                        goto out;
                }

                LASSERT (nphys * sizeof (*phys) < phys_size);
#if IBNAL_FMR
                phys[nphys] = lnet_page2phys(kiov->kiov_page);
#else
                phys[nphys].address = lnet_page2phys(kiov->kiov_page);
                phys[nphys].size = PAGE_SIZE;
#endif
                nphys++;

                resid -= PAGE_SIZE;
        }

        tx->tx_md.md_addr = IBNAL_RDMA_BASE;

#if IBNAL_FMR
        rc = ib_fmr_register_physical (kibnal_data.kib_fmr_pool,
                                       phys, nphys,
                                       &tx->tx_md.md_addr,
                                       page_offset,
                                       &tx->tx_md.md_handle.fmr,
                                       &tx->tx_md.md_lkey,
                                       &tx->tx_md.md_rkey);
#else
        rc = ib_memory_register_physical (kibnal_data.kib_pd,
                                          phys, nphys,
                                          &tx->tx_md.md_addr,
                                          nob, page_offset,
                                          access,
                                          &tx->tx_md.md_handle.mr,
                                          &tx->tx_md.md_lkey,
                                          &tx->tx_md.md_rkey);
#endif
        if (rc == 0) {
                CDEBUG(D_NET, "Mapped %d pages %d bytes @ offset %d: lkey %x, rkey %x\n",
                       nphys, nob, page_offset, tx->tx_md.md_lkey, tx->tx_md.md_rkey);
                tx->tx_mapped = mapped;
        } else {
                CERROR ("Can't map phys: %d\n", rc);
                rc = -EFAULT;
        }

 out:
        LIBCFS_FREE(phys, phys_size);
        return (rc);
}

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)
{
        unsigned long   flags;
        kib_tx_t       *tx;
        int             rc;
        int             i;
        int             consume_credit;
        int             done;
        int             nwork;

        spin_lock_irqsave (&conn->ibc_lock, flags);

        LASSERT (conn->ibc_nsends_posted <= IBNAL_RX_MSGS);
        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_irqrestore(&conn->ibc_lock, flags);
                
                tx = kibnal_get_idle_tx();
                if (tx != NULL)
                        kibnal_init_tx_msg(tx, IBNAL_MSG_NOOP, 0);

                spin_lock_irqsave(&conn->ibc_lock, flags);
                
                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_credit = 0;
                } else if (!list_empty (&conn->ibc_tx_queue)) {
                        tx = list_entry (conn->ibc_tx_queue.next, 
                                         kib_tx_t, tx_list);
                        consume_credit = 1;
                } else {
                        /* nothing waiting */
                        break;
                }

                /* We rely on this for QP sizing */
                LASSERT (tx->tx_nsp > 0 && tx->tx_nsp <= 2);

                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);

                /* Not on ibc_rdma_queue */
                LASSERT (!tx->tx_passive_rdma_wait);

                if (conn->ibc_nsends_posted == IBNAL_RX_MSGS)
                        break;

                if (consume_credit) {
                        if (conn->ibc_credits == 0)     /* no credits */
                                break;
                
                        if (conn->ibc_credits == 1 &&   /* last credit reserved for */
                            conn->ibc_outstanding_credits == 0) /* giving back credits */
                                break;
                }
                
                list_del (&tx->tx_list);

                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_irqrestore(&conn->ibc_lock, flags);
                        kibnal_tx_done(tx);
                        spin_lock_irqsave(&conn->ibc_lock, flags);
                        continue;
                }

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

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