openiblnd_cb.c

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

                tx->tx_sending = tx->tx_nsp;
                tx->tx_passive_rdma_wait = tx->tx_passive_rdma;
                list_add (&tx->tx_list, &conn->ibc_active_txs);

                spin_unlock_irqrestore (&conn->ibc_lock, flags);

                /* NB the gap between removing tx from the queue and sending it
                 * allows message re-ordering to occur */

                LASSERT (tx->tx_nsp > 0);

                rc = -ECONNABORTED;
                nwork = 0;
                if (conn->ibc_state == IBNAL_CONN_ESTABLISHED) {
                        tx->tx_status = 0;
                        /* Driver only accepts 1 item at a time */
                        for (i = 0; i < tx->tx_nsp; i++) {
                                rc = kibnal_ib_send(conn->ibc_qp, &tx->tx_sp[i]);
                                if (rc != 0)
                                        break;
                                nwork++;
                        }
                }

                conn->ibc_last_send = jiffies;

                spin_lock_irqsave (&conn->ibc_lock, flags);
                if (rc != 0) {
                        /* NB credits are transferred in the actual
                         * message, which can only be the last work item */
                        conn->ibc_outstanding_credits += tx->tx_msg->ibm_credits;
                        if (consume_credit)
                                conn->ibc_credits++;
                        conn->ibc_nsends_posted--;

                        tx->tx_status = rc;
                        tx->tx_passive_rdma_wait = 0;
                        tx->tx_sending -= tx->tx_nsp - nwork;

                        done = (tx->tx_sending == 0);
                        if (done)
                                list_del (&tx->tx_list);
                        
                        spin_unlock_irqrestore (&conn->ibc_lock, flags);
                        
                        if (conn->ibc_state == IBNAL_CONN_ESTABLISHED)
                                CERROR ("Error %d posting transmit to %s\n", 
                                        rc, libcfs_nid2str(conn->ibc_peer->ibp_nid));
                        else
                                CDEBUG (D_NET, "Error %d posting transmit to %s\n",
                                        rc, libcfs_nid2str(conn->ibc_peer->ibp_nid));

                        kibnal_close_conn (conn, rc);

                        if (done)
                                kibnal_tx_done (tx);
                        return;
                }
                
        }

        spin_unlock_irqrestore (&conn->ibc_lock, flags);
}

void
kibnal_tx_callback (struct ib_cq_entry *e)
{
        kib_tx_t     *tx = (kib_tx_t *)kibnal_wreqid2ptr(e->work_request_id);
        kib_conn_t   *conn;
        unsigned long flags;
        int           idle;

        conn = tx->tx_conn;
        LASSERT (conn != NULL);
        LASSERT (tx->tx_sending != 0);

        spin_lock_irqsave(&conn->ibc_lock, flags);

        CDEBUG(D_NET, "conn %p tx %p [%d/%d]: %d\n", conn, tx,
               tx->tx_nsp - tx->tx_sending, tx->tx_nsp,
               e->status);

        /* I could be racing with rdma completion.  Whoever makes 'tx' idle
         * gets to free it, which also drops its ref on 'conn'.  If it's
         * not me, then I take an extra ref on conn so it can't disappear
         * under me. */

        tx->tx_sending--;
        idle = (tx->tx_sending == 0) &&         /* This is the final callback */
               (!tx->tx_passive_rdma_wait);     /* Not waiting for RDMA completion */
        if (idle)
                list_del(&tx->tx_list);

        kibnal_conn_addref(conn);

        if (tx->tx_sending == 0)
                conn->ibc_nsends_posted--;

        if (e->status != IB_COMPLETION_STATUS_SUCCESS &&
            tx->tx_status == 0)
                tx->tx_status = -ECONNABORTED;
                
        spin_unlock_irqrestore(&conn->ibc_lock, flags);

        if (idle)
                kibnal_tx_done (tx);

        if (e->status != IB_COMPLETION_STATUS_SUCCESS) {
                CDEBUG (D_NETERROR, "Tx completion to %s failed: %d\n", 
                        libcfs_nid2str(conn->ibc_peer->ibp_nid), e->status);
                kibnal_close_conn (conn, -ENETDOWN);
        } else {
                kibnal_peer_alive(conn->ibc_peer);
                /* can I shovel some more sends out the door? */
                kibnal_check_sends(conn);
        }

        kibnal_conn_decref(conn);
}

void
kibnal_callback (ib_cq_t *cq, struct ib_cq_entry *e, void *arg)
{
        if (kibnal_wreqid_is_rx(e->work_request_id))
                kibnal_rx_callback (e);
        else
                kibnal_tx_callback (e);
}

void
kibnal_init_tx_msg (kib_tx_t *tx, int type, int body_nob)
{
        struct ib_gather_scatter *gl = &tx->tx_gl[tx->tx_nsp];
        struct ib_send_param     *sp = &tx->tx_sp[tx->tx_nsp];
        int                       fence;
        int                       nob = offsetof (kib_msg_t, ibm_u) + body_nob;

        LASSERT (tx->tx_nsp >= 0 && 
                 tx->tx_nsp < sizeof(tx->tx_sp)/sizeof(tx->tx_sp[0]));
        LASSERT (nob <= IBNAL_MSG_SIZE);

        kibnal_init_msg(tx->tx_msg, type, body_nob);

        /* Fence the message if it's bundled with an RDMA read */
        fence = (tx->tx_nsp > 0) &&
                (type == IBNAL_MSG_PUT_DONE);

        *gl = (struct ib_gather_scatter) {
                .address = tx->tx_vaddr,
                .length  = nob,
                .key     = kibnal_data.kib_tx_pages->ibp_lkey,
        };

        /* NB If this is an RDMA read, the completion message must wait for
         * the RDMA to complete.  Sends wait for previous RDMA writes
         * anyway... */
        *sp = (struct ib_send_param) {
                .work_request_id      = kibnal_ptr2wreqid(tx, 0),
                .op                   = IB_OP_SEND,
                .gather_list          = gl,
                .num_gather_entries   = 1,
                .device_specific      = NULL,
                .solicited_event      = 1,
                .signaled             = 1,
                .immediate_data_valid = 0,
                .fence                = fence,
                .inline_data          = 0,
        };

        tx->tx_nsp++;
}

void
kibnal_queue_tx (kib_tx_t *tx, kib_conn_t *conn)
{
        unsigned long         flags;

        spin_lock_irqsave(&conn->ibc_lock, flags);

        kibnal_queue_tx_locked (tx, conn);
        
        spin_unlock_irqrestore(&conn->ibc_lock, flags);
        
        kibnal_check_sends(conn);
}

void
kibnal_schedule_active_connect_locked (kib_peer_t *peer)
{
        /* Called with exclusive kib_global_lock */

        peer->ibp_connecting++;
        kibnal_peer_addref(peer); /* extra ref for connd */
        
        spin_lock (&kibnal_data.kib_connd_lock);
        
        LASSERT (list_empty(&peer->ibp_connd_list));
        list_add_tail (&peer->ibp_connd_list,
                       &kibnal_data.kib_connd_peers);
        wake_up (&kibnal_data.kib_connd_waitq);
        
        spin_unlock (&kibnal_data.kib_connd_lock);
}

void
kibnal_launch_tx (kib_tx_t *tx, lnet_nid_t nid)
{
        unsigned long    flags;
        kib_peer_t      *peer;
        kib_conn_t      *conn;
        int              retry;
        int              rc;
        rwlock_t        *g_lock = &kibnal_data.kib_global_lock;

        /* If I get here, I've committed to send, so I complete the tx with
         * failure on any problems */
        
        LASSERT (tx->tx_conn == NULL);          /* only set when assigned a conn */
        LASSERT (tx->tx_nsp > 0);               /* work items have been set up */

        for (retry = 0; ; retry = 1) {
                read_lock_irqsave(g_lock, flags);
        
                peer = kibnal_find_peer_locked (nid);
                if (peer != NULL) {
                        conn = kibnal_find_conn_locked (peer);
                        if (conn != NULL) {
                                kibnal_conn_addref(conn); /* 1 ref for me...*/
                                read_unlock_irqrestore(g_lock, flags);
                
                                kibnal_queue_tx (tx, conn);
                                kibnal_conn_decref(conn); /* ...until here */
                                return;
                        }
                }
                
                /* Making one or more connections; I'll need a write lock... */
                read_unlock(g_lock);
                write_lock(g_lock);

                peer = kibnal_find_peer_locked (nid);
                if (peer != NULL)
                        break;
                
                write_unlock_irqrestore (g_lock, flags);

                if (retry) {
                        CERROR("Can't find peer %s\n", libcfs_nid2str(nid));
                        tx->tx_status = -EHOSTUNREACH;
                        kibnal_tx_done (tx);
                        return;
                }

                rc = kibnal_add_persistent_peer(nid, LNET_NIDADDR(nid),
                                                lnet_acceptor_port());
                if (rc != 0) {
                        CERROR("Can't add peer %s: %d\n",
                               libcfs_nid2str(nid), rc);
                        tx->tx_status = rc;
                        kibnal_tx_done(tx);
                        return;
                }
        }

        conn = kibnal_find_conn_locked (peer);
        if (conn != NULL) {
                /* Connection exists; queue message on it */
                kibnal_conn_addref(conn);       /* +1 ref from me... */
                write_unlock_irqrestore (g_lock, flags);
                
                kibnal_queue_tx (tx, conn);
                kibnal_conn_decref(conn);       /* ...until here */
                return;
        }

        if (peer->ibp_connecting == 0 &&
            peer->ibp_accepting == 0) {
                if (!(peer->ibp_reconnect_interval == 0 || /* first attempt */
                      time_after_eq(jiffies, peer->ibp_reconnect_time))) {
                        write_unlock_irqrestore (g_lock, flags);
                        tx->tx_status = -EHOSTUNREACH;
                        kibnal_tx_done (tx);
                        return;
                }
        
                kibnal_schedule_active_connect_locked(peer);
        }
        
        /* A connection is being established; queue the message... */
        list_add_tail (&tx->tx_list, &peer->ibp_tx_queue);

        write_unlock_irqrestore (g_lock, flags);
}

void
kibnal_txlist_done (struct list_head *txlist, int status)
{
        kib_tx_t *tx;

        while (!list_empty(txlist)) {
                tx = list_entry (txlist->next, kib_tx_t, tx_list);

                list_del (&tx->tx_list);
                /* complete now */
                tx->tx_status = status;
                kibnal_tx_done (tx);
        }
}

int
kibnal_start_passive_rdma (int type, lnet_msg_t *lntmsg,
                           int niov, struct iovec *iov, lnet_kiov_t *kiov,
                           int nob)
{
        lnet_nid_t  nid = lntmsg->msg_target.nid;
        kib_tx_t   *tx;
        kib_msg_t  *ibmsg;
        int         rc;
        int         access;
        
        LASSERT (type == IBNAL_MSG_PUT_RDMA || 
                 type == IBNAL_MSG_GET_RDMA);
        LASSERT (nob > 0);
        LASSERT (!in_interrupt());              /* Mapping could block */

        if (type == IBNAL_MSG_PUT_RDMA) {
                access = IB_ACCESS_REMOTE_READ;
        } else {
                access = IB_ACCESS_REMOTE_WRITE |
                         IB_ACCESS_LOCAL_WRITE;
        }

        tx = kibnal_get_idle_tx ();
        if (tx == NULL) {
                CERROR("Can't allocate %s txd for %s\n",
                       (type == IBNAL_MSG_PUT_RDMA) ? "PUT/REPLY" : "GET",
                       libcfs_nid2str(nid));
                return -ENOMEM;
        }

        
        if (iov != NULL) 
                rc = kibnal_map_iov (tx, access, niov, iov, 0, nob);
        else
                rc = kibnal_map_kiov (tx, access, niov, kiov, 0, nob);

        if (rc != 0) {
                CERROR ("Can't map RDMA for %s: %d\n", 
                        libcfs_nid2str(nid), rc);
                goto failed;
        }
        
        if (type == IBNAL_MSG_GET_RDMA) {
                /* reply gets finalized when tx completes */
                tx->tx_lntmsg[1] = lnet_create_reply_msg(kibnal_data.kib_ni, 
                                                         lntmsg);
                if (tx->tx_lntmsg[1] == NULL) {
                        CERROR ("Can't create reply for GET -> %s\n",
                                libcfs_nid2str(nid));
                        rc = -ENOMEM;
                        goto failed;
                }
        }
        
        tx->tx_passive_rdma = 1;