iiblnd.c

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

                
                page_offset += IBNAL_MSG_SIZE;
                LASSERT (page_offset <= PAGE_SIZE);

                if (page_offset == PAGE_SIZE) {
                        page_offset = 0;
                        ipage++;
                        LASSERT (ipage <= IBNAL_RX_MSG_PAGES);
                }
        }

        params.qp_create = (IB_QP_ATTRIBUTES_CREATE) {
                .Type                    = QPTypeReliableConnected,
                .SendQDepth              = (1 + IBNAL_MAX_RDMA_FRAGS) *
                                           (*kibnal_tunables.kib_concurrent_sends),
                .RecvQDepth              = IBNAL_RX_MSGS,
                .SendDSListDepth         = 1,
                .RecvDSListDepth         = 1,
                .SendCQHandle            = kibnal_data.kib_cq,
                .RecvCQHandle            = kibnal_data.kib_cq,
                .PDHandle                = kibnal_data.kib_pd,
                .SendSignaledCompletions = TRUE,
        };
        frc = iba_create_qp(kibnal_data.kib_hca, &params.qp_create, NULL,
                            &conn->ibc_qp, &conn->ibc_cvars->cv_qpattrs);
        if (frc != 0) {
                CERROR ("Can't create QP %s: %d\n", libcfs_nid2str(nid), frc);
                goto failed;
        }

        /* Mark QP created */
        kibnal_set_conn_state(conn, IBNAL_CONN_INIT_QP);

        params.qp_attr = (IB_QP_ATTRIBUTES_MODIFY) {
                .RequestState             = QPStateInit,
                .Attrs                    = (IB_QP_ATTR_PORTGUID |
                                             IB_QP_ATTR_PKEYINDEX |
                                             IB_QP_ATTR_ACCESSCONTROL),
                .PortGUID                 = kibnal_data.kib_port_guid,
                .PkeyIndex                = 0,
                .AccessControl = { 
                        .s = {
                                .RdmaWrite = 1,
                                .RdmaRead  = 1,
                        },
                },
        };
        frc = iba_modify_qp(conn->ibc_qp, &params.qp_attr, NULL);
        if (frc != 0) {
                CERROR ("Can't set QP %s state to INIT: %d\n",
                        libcfs_nid2str(nid), frc);
                goto failed;
        }

        frc = iba_query_qp(conn->ibc_qp, &conn->ibc_cvars->cv_qpattrs, NULL);
        if (frc != FSUCCESS) {
                CERROR ("Can't query QP %s attributes: %d\n",
                        libcfs_nid2str(nid), frc);
                goto failed;
        }

        /* 1 ref for caller */
        atomic_set (&conn->ibc_refcount, 1);
        CDEBUG(D_NET, "New conn %p\n", conn);
        return (conn);
        
 failed:
        kibnal_destroy_conn (conn);
        return (NULL);
}

void
kibnal_destroy_conn (kib_conn_t *conn)
{
        FSTATUS frc;

        LASSERT (!in_interrupt());
        
        CDEBUG (D_NET, "connection %s\n", 
                (conn->ibc_peer) == NULL ? "<ANON>" :
                libcfs_nid2str(conn->ibc_peer->ibp_nid));

        LASSERT (atomic_read (&conn->ibc_refcount) == 0);
        LASSERT (list_empty(&conn->ibc_early_rxs));
        LASSERT (list_empty(&conn->ibc_tx_queue));
        LASSERT (list_empty(&conn->ibc_tx_queue_rsrvd));
        LASSERT (list_empty(&conn->ibc_tx_queue_nocred));
        LASSERT (list_empty(&conn->ibc_active_txs));
        LASSERT (conn->ibc_nsends_posted == 0);

        switch (conn->ibc_state) {
        case IBNAL_CONN_INIT_NOTHING:
        case IBNAL_CONN_INIT_QP:
        case IBNAL_CONN_DISCONNECTED:
                break;

        default:
                /* conn must either have never engaged with the CM, or have
                 * completely disengaged from it */
                CERROR("Bad conn %s state %d\n",
                       (conn->ibc_peer) == NULL ? "<anon>" :
                       libcfs_nid2str(conn->ibc_peer->ibp_nid), conn->ibc_state);
                LBUG();
        }

        if (conn->ibc_cep != NULL) {
                frc = iba_cm_destroy_cep(conn->ibc_cep);
                if (frc != FSUCCESS)
                        CERROR("Error destroying CEP %p: %d\n",
                               conn->ibc_cep, frc);
        }

        if (conn->ibc_qp != NULL) {
                frc = iba_destroy_qp(conn->ibc_qp);
                if (frc != FSUCCESS)
                        CERROR("Error destroying QP %p: %d\n",
                               conn->ibc_qp, frc);
        }

        if (conn->ibc_rx_pages != NULL) 
                kibnal_free_pages(conn->ibc_rx_pages);
        
        if (conn->ibc_rxs != NULL)
                LIBCFS_FREE(conn->ibc_rxs, 
                            IBNAL_RX_MSGS * sizeof(kib_rx_t));

        if (conn->ibc_cvars != NULL)
                LIBCFS_FREE(conn->ibc_cvars, sizeof(*conn->ibc_cvars));

        if (conn->ibc_peer != NULL)
                kibnal_peer_decref(conn->ibc_peer);

        LIBCFS_FREE(conn, sizeof (*conn));

        atomic_dec(&kibnal_data.kib_nconns);
}

int
kibnal_close_peer_conns_locked (kib_peer_t *peer, int why)
{
        kib_conn_t         *conn;
        struct list_head   *ctmp;
        struct list_head   *cnxt;
        int                 count = 0;

        list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
                conn = list_entry (ctmp, kib_conn_t, ibc_list);

                count++;
                kibnal_close_conn_locked (conn, why);
        }

        return (count);
}

int
kibnal_close_stale_conns_locked (kib_peer_t *peer, __u64 incarnation)
{
        kib_conn_t         *conn;
        struct list_head   *ctmp;
        struct list_head   *cnxt;
        int                 count = 0;

        list_for_each_safe (ctmp, cnxt, &peer->ibp_conns) {
                conn = list_entry (ctmp, kib_conn_t, ibc_list);

                if (conn->ibc_incarnation == incarnation)
                        continue;

                CDEBUG(D_NET, "Closing stale conn nid:%s incarnation:"LPX64"("LPX64")\n",
                       libcfs_nid2str(peer->ibp_nid), 
                       conn->ibc_incarnation, incarnation);
                
                count++;
                kibnal_close_conn_locked (conn, -ESTALE);
        }

        return (count);
}

int
kibnal_close_matching_conns (lnet_nid_t nid)
{
        unsigned long       flags;
        kib_peer_t         *peer;
        struct list_head   *ptmp;
        struct list_head   *pnxt;
        int                 lo;
        int                 hi;
        int                 i;
        int                 count = 0;

        write_lock_irqsave (&kibnal_data.kib_global_lock, flags);

        if (nid != LNET_NID_ANY)
                lo = hi = kibnal_nid2peerlist(nid) - kibnal_data.kib_peers;
        else {
                lo = 0;
                hi = kibnal_data.kib_peer_hash_size - 1;
        }

        for (i = lo; i <= hi; i++) {
                list_for_each_safe (ptmp, pnxt, &kibnal_data.kib_peers[i]) {

                        peer = list_entry (ptmp, kib_peer_t, ibp_list);
                        LASSERT (peer->ibp_persistence != 0 ||
                                 kibnal_peer_connecting(peer) ||
                                 !list_empty (&peer->ibp_conns));

                        if (!(nid == LNET_NID_ANY || nid == peer->ibp_nid))
                                continue;

                        count += kibnal_close_peer_conns_locked (peer, 0);
                }
        }

        write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);

        /* wildcards always succeed */
        if (nid == LNET_NID_ANY)
                return (0);
        
        return (count == 0 ? -ENOENT : 0);
}

int
kibnal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
{
        struct libcfs_ioctl_data *data = arg;
        int                       rc = -EINVAL;
        ENTRY;

        LASSERT (ni == kibnal_data.kib_ni);

        switch(cmd) {
        case IOC_LIBCFS_GET_PEER: {
                lnet_nid_t   nid = 0;
                int          share_count = 0;

                rc = kibnal_get_peer_info(data->ioc_count,
                                          &nid, &share_count);
                data->ioc_nid   = nid;
                data->ioc_count = share_count;
                break;
        }
        case IOC_LIBCFS_ADD_PEER: {
                rc = kibnal_add_persistent_peer (data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_DEL_PEER: {
                rc = kibnal_del_peer (data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_GET_CONN: {
                kib_conn_t *conn = kibnal_get_conn_by_idx (data->ioc_count);

                if (conn == NULL)
                        rc = -ENOENT;
                else {
                        rc = 0;
                        data->ioc_nid = conn->ibc_peer->ibp_nid;
                        kibnal_conn_decref(conn);
                }
                break;
        }
        case IOC_LIBCFS_CLOSE_CONNECTION: {
                rc = kibnal_close_matching_conns (data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_REGISTER_MYNID: {
                if (ni->ni_nid == data->ioc_nid) {
                        rc = 0;
                } else {
                        CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID: %s(%s)\n",
                               libcfs_nid2str(data->ioc_nid),
                               libcfs_nid2str(ni->ni_nid));
                        rc = -EINVAL;
                }
                break;
        }
        }

        RETURN(rc);
}

void
kibnal_free_pages (kib_pages_t *p)
{
        int     npages = p->ibp_npages;
        int     i;
        
        for (i = 0; i < npages; i++)
                if (p->ibp_pages[i] != NULL)
                        __free_page(p->ibp_pages[i]);
        
        LIBCFS_FREE (p, offsetof(kib_pages_t, ibp_pages[npages]));
}

int
kibnal_alloc_pages (kib_pages_t **pp, int npages)
{
        kib_pages_t   *p;
        int            i;

        LIBCFS_ALLOC(p, offsetof(kib_pages_t, ibp_pages[npages]));
        if (p == NULL) {
                CERROR ("Can't allocate buffer %d\n", npages);
                return (-ENOMEM);
        }

        memset (p, 0, offsetof(kib_pages_t, ibp_pages[npages]));
        p->ibp_npages = npages;
        
        for (i = 0; i < npages; i++) {
                p->ibp_pages[i] = alloc_page (GFP_KERNEL);
                if (p->ibp_pages[i] == NULL) {
                        CERROR ("Can't allocate page %d of %d\n", i, npages);
                        kibnal_free_pages(p);
                        return (-ENOMEM);
                }
        }

        *pp = p;
        return (0);
}

int
kibnal_alloc_tx_descs (void) 
{
        int    i;
        
        LIBCFS_ALLOC (kibnal_data.kib_tx_descs,
                      IBNAL_TX_MSGS() * sizeof(kib_tx_t));
        if (kibnal_data.kib_tx_descs == NULL)
                return -ENOMEM;
        
        memset(kibnal_data.kib_tx_descs, 0,
               IBNAL_TX_MSGS() * sizeof(kib_tx_t));

        for (i = 0; i < IBNAL_TX_MSGS(); i++) {
                kib_tx_t *tx = &kibnal_data.kib_tx_descs[i];

#if IBNAL_USE_FMR
                LIBCFS_ALLOC(tx->tx_pages, LNET_MAX_IOV *
                             sizeof(*tx->tx_pages));
                if (tx->tx_pages == NULL)
                        return -ENOMEM;
#else
                LIBCFS_ALLOC(tx->tx_wrq, 
                             (1 + IBNAL_MAX_RDMA_FRAGS) * 
                             sizeof(*tx->tx_wrq));
                if (tx->tx_wrq == NULL)
                        return -ENOMEM;
                
                LIBCFS_ALLOC(tx->tx_gl, 
                             (1 + IBNAL_MAX_RDMA_FRAGS) * 
                             sizeof(*tx->tx_gl));
                if (tx->tx_gl == NULL)
                        return -ENOMEM;
                
                LIBCFS_ALLOC(tx->tx_rd, 
                             offsetof(kib_rdma_desc_t, 
                                      rd_frags[IBNAL_MAX_RDMA_FRAGS]));
                if (tx->tx_rd == NULL)
                        return -ENOMEM;
#endif
        }

        return 0;
}

void
kibnal_free_tx_descs (void) 
{
        int    i;