viblnd.c

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

        reqattr.create.receive_max_outstand_wr    = IBNAL_RX_MSGS;
        reqattr.create.max_scatgat_per_send_wr    = 1;
        reqattr.create.max_scatgat_per_receive_wr = 1;
        reqattr.create.signaling_type             = vv_selectable_signaling;
        reqattr.create.pd_h                       = kibnal_data.kib_pd;
        reqattr.create.recv_solicited_events      = vv_selectable_signaling; // vv_signal_all;

        vvrc = vv_qp_create(kibnal_data.kib_hca, &reqattr, NULL,
                            &conn->ibc_qp, &rspattr);
        if (vvrc != vv_return_ok) {
                CERROR ("Failed to create queue pair: %d\n", vvrc);
                goto failed;
        }

        /* Mark QP created */
        conn->ibc_state = IBNAL_CONN_INIT_QP;
        conn->ibc_connvars->cv_local_qpn = rspattr.create_return.qp_num;

        if (rspattr.create_return.receive_max_outstand_wr < 
            IBNAL_RX_MSGS ||
            rspattr.create_return.send_max_outstand_wr < 
            (1 + IBNAL_MAX_RDMA_FRAGS) * (*kibnal_tunables.kib_concurrent_sends)) {
                CERROR("Insufficient rx/tx work items: wanted %d/%d got %d/%d\n",
                       IBNAL_RX_MSGS, 
                       (1 + IBNAL_MAX_RDMA_FRAGS) * 
                       (*kibnal_tunables.kib_concurrent_sends),
                       rspattr.create_return.receive_max_outstand_wr,
                       rspattr.create_return.send_max_outstand_wr);
                goto failed;
        }

        /* Mark init complete */
        conn->ibc_state = IBNAL_CONN_INIT;

        /* 1 ref for caller */
        atomic_set (&conn->ibc_refcount, 1);
        return (conn);
        
 failed:
        kibnal_destroy_conn (conn);
        return (NULL);
}

void
kibnal_destroy_conn (kib_conn_t *conn)
{
        vv_return_t vvrc;

        /* Only the connd does this (i.e. single threaded) */
        LASSERT (!in_interrupt());
        LASSERT (current == kibnal_data.kib_connd);
        
        CDEBUG (D_NET, "connection %p\n", conn);

        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) {
        default:
                /* conn must be completely disengaged from the network */
                LBUG();

        case IBNAL_CONN_DISCONNECTED:
                /* connvars should have been freed already */
                LASSERT (conn->ibc_connvars == NULL);
                /* fall through */

        case IBNAL_CONN_INIT:
                vvrc = cm_destroy_cep(conn->ibc_cep);
                LASSERT (vvrc == vv_return_ok);
                /* fall through */

        case IBNAL_CONN_INIT_QP:
                kibnal_set_qp_state(conn, vv_qp_state_reset);
                vvrc = vv_qp_destroy(kibnal_data.kib_hca, conn->ibc_qp);
                if (vvrc != vv_return_ok)
                        CERROR("Can't destroy QP: %d\n", vvrc);
                /* fall through */
                
        case IBNAL_CONN_INIT_NOTHING:
                break;
        }

        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_connvars != NULL)
                LIBCFS_FREE(conn->ibc_connvars, sizeof(*conn->ibc_connvars));

        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 -> %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)
{
        kib_peer_t         *peer;
        struct list_head   *ptmp;
        struct list_head   *pnxt;
        int                 lo;
        int                 hi;
        int                 i;
        unsigned long       flags;
        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 ||
                                 peer->ibp_connecting != 0 ||
                                 peer->ibp_accepting != 0 ||
                                 !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;

        LASSERT (ni == kibnal_data.kib_ni);

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

                rc = kibnal_get_peer_info(data->ioc_count,
                                          &nid, &ip, &share_count);
                data->ioc_nid    = nid;
                data->ioc_count  = share_count;
                data->ioc_u32[0] = ip;
                data->ioc_u32[1] = *kibnal_tunables.kib_service_number; /* port */
                break;
        }
        case IOC_LIBCFS_ADD_PEER: {
                rc = kibnal_add_persistent_peer (data->ioc_nid,
                                                 data->ioc_u32[0]); /* IP */
                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 {
                        // kibnal_debug_conn(conn);
                        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, int allow_write)
{
        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;

        if (kibnal_data.kib_tx_descs == NULL)
                return;