iiblnd_cb.c

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

                               tx->tx_msg->ibm_type, 
                               tx->tx_msg, tx->tx_msg->ibm_nob,
                               tx->tx_rd->rd_key, tx->tx_rd->rd_nfrag);
                        for (i = 0; i < tx->tx_rd->rd_nfrag; i++)
                                CDEBUG(D_NETERROR, "    [%d] "LPX64"/%d\n", i,
                                       tx->tx_rd->rd_frags[i].rf_addr,
                                       tx->tx_rd->rd_frags[i].rf_nob);
                        if (lntmsg == NULL) {
                                CDEBUG(D_NETERROR, "  No lntmsg\n");
                        } else if (lntmsg->msg_iov != NULL) {
                                CDEBUG(D_NETERROR, "  lntmsg in %d VIRT frags...\n", 
                                       lntmsg->msg_niov);
                                for (i = 0; i < lntmsg->msg_niov; i++)
                                        CDEBUG(D_NETERROR, "    [%d] %p/%d\n", i,
                                               lntmsg->msg_iov[i].iov_base,
                                               lntmsg->msg_iov[i].iov_len);
                        } else if (lntmsg->msg_kiov != NULL) {
                                CDEBUG(D_NETERROR, "  lntmsg in %d PAGE frags...\n", 
                                       lntmsg->msg_niov);
                                for (i = 0; i < lntmsg->msg_niov; i++)
                                        CDEBUG(D_NETERROR, "    [%d] %p+%d/%d\n", i,
                                               lntmsg->msg_kiov[i].kiov_page,
                                               lntmsg->msg_kiov[i].kiov_offset,
                                               lntmsg->msg_kiov[i].kiov_len);
                        } else {
                                CDEBUG(D_NETERROR, "  lntmsg in %d frags\n", 
                                       lntmsg->msg_niov);
                        }
                        
                        break;
                }
#endif
        }
        
        spin_lock(&conn->ibc_lock);

        /* I could be racing with rdma completion.  Whoever makes 'tx' idle
         * gets to free it, which also drops its ref on 'conn'. */

        tx->tx_sending--;
        conn->ibc_nsends_posted--;

        if (failed) {
                tx->tx_waiting = 0;
                tx->tx_status = -EIO;
        }
        
        idle = (tx->tx_sending == 0) &&         /* This is the final callback */
               !tx->tx_waiting &&               /* Not waiting for peer */
               !tx->tx_queued;                  /* Not re-queued (PUT_DONE) */
        if (idle)
                list_del(&tx->tx_list);

        kibnal_conn_addref(conn);               /* 1 ref for me.... */

        spin_unlock(&conn->ibc_lock);

        if (idle)
                kibnal_tx_done (tx);

        if (failed) {
                kibnal_close_conn (conn, -EIO);
        } else {
                kibnal_peer_alive(conn->ibc_peer);
                kibnal_check_sends(conn);
        }

        kibnal_conn_decref(conn);               /* ...until here */
}

void
kibnal_init_tx_msg (kib_tx_t *tx, int type, int body_nob)
{
        IB_LOCAL_DATASEGMENT *gl = &tx->tx_gl[tx->tx_nwrq];
        IB_WORK_REQ2         *wrq = &tx->tx_wrq[tx->tx_nwrq];
        int                   nob = offsetof (kib_msg_t, ibm_u) + body_nob;

        LASSERT (tx->tx_nwrq >= 0 && 
                 tx->tx_nwrq < (1 + IBNAL_MAX_RDMA_FRAGS));
        LASSERT (nob <= IBNAL_MSG_SIZE);

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

        *gl = (IB_LOCAL_DATASEGMENT) {
                .Address = tx->tx_hca_msg,
                .Length  = IBNAL_MSG_SIZE,
                .Lkey    = kibnal_data.kib_whole_mem.md_lkey,
        };

        wrq->Next           = NULL;             /* This is the last one */

        wrq->WorkReqId      = kibnal_ptr2wreqid(tx, IBNAL_WID_TX);
        wrq->Operation      = WROpSend;
        wrq->DSList         = gl;
        wrq->DSListDepth    = 1;
        wrq->MessageLen     = nob;
        wrq->Req.SendRC.ImmediateData  = 0;
        wrq->Req.SendRC.Options.s.SolicitedEvent         = 1;
        wrq->Req.SendRC.Options.s.SignaledCompletion     = 1;
        wrq->Req.SendRC.Options.s.ImmediateData          = 0;
        wrq->Req.SendRC.Options.s.Fence                  = 0; 
        /* fence only needed on RDMA reads */
        
        tx->tx_nwrq++;
}

int
kibnal_init_rdma (kib_tx_t *tx, int type, int nob,
                  kib_rdma_desc_t *dstrd, __u64 dstcookie)
{
        kib_msg_t            *ibmsg = tx->tx_msg;
        kib_rdma_desc_t      *srcrd = tx->tx_rd;
        IB_LOCAL_DATASEGMENT *gl;
        IB_WORK_REQ2         *wrq;
        int                   rc;

#if IBNAL_USE_FMR
        LASSERT (tx->tx_nwrq == 0);

        gl = &tx->tx_gl[0];
        gl->Length  = nob;
        gl->Address = srcrd->rd_addr;
        gl->Lkey    = srcrd->rd_key;

        wrq = &tx->tx_wrq[0];

        wrq->Next           = wrq + 1;
        wrq->WorkReqId      = kibnal_ptr2wreqid(tx, IBNAL_WID_RDMA);
        wrq->Operation      = WROpRdmaWrite;
        wrq->DSList         = gl;
        wrq->DSListDepth    = 1;
        wrq->MessageLen     = nob;

        wrq->Req.SendRC.ImmediateData                = 0;
        wrq->Req.SendRC.Options.s.SolicitedEvent     = 0;
        wrq->Req.SendRC.Options.s.SignaledCompletion = 0;
        wrq->Req.SendRC.Options.s.ImmediateData      = 0;
        wrq->Req.SendRC.Options.s.Fence              = 0; 

        wrq->Req.SendRC.RemoteDS.Address = dstrd->rd_addr;
        wrq->Req.SendRC.RemoteDS.Rkey    = dstrd->rd_key;

        tx->tx_nwrq = 1;
        rc = nob;
#else
        /* CAVEAT EMPTOR: this 'consumes' the frags in 'dstrd' */
        int              resid = nob;
        kib_rdma_frag_t *srcfrag;
        int              srcidx;
        kib_rdma_frag_t *dstfrag;
        int              dstidx;
        int              wrknob;

        /* Called by scheduler */
        LASSERT (!in_interrupt());

        LASSERT (type == IBNAL_MSG_GET_DONE ||
                 type == IBNAL_MSG_PUT_DONE);

        srcidx = dstidx = 0;
        srcfrag = &srcrd->rd_frags[0];
        dstfrag = &dstrd->rd_frags[0];
        rc = resid;

        while (resid > 0) {
                if (srcidx >= srcrd->rd_nfrag) {
                        CERROR("Src buffer exhausted: %d frags\n", srcidx);
                        rc = -EPROTO;
                        break;
                }
                
                if (dstidx == dstrd->rd_nfrag) {
                        CERROR("Dst buffer exhausted: %d frags\n", dstidx);
                        rc = -EPROTO;
                        break;
                }

                if (tx->tx_nwrq == IBNAL_MAX_RDMA_FRAGS) {
                        CERROR("RDMA too fragmented: %d/%d src %d/%d dst frags\n",
                               srcidx, srcrd->rd_nfrag,
                               dstidx, dstrd->rd_nfrag);
                        rc = -EMSGSIZE;
                        break;
                }

                wrknob = MIN(MIN(srcfrag->rf_nob, dstfrag->rf_nob), resid);

                gl = &tx->tx_gl[tx->tx_nwrq];
                gl->Length  = wrknob;
                gl->Address = srcfrag->rf_addr;
                gl->Lkey    = srcrd->rd_key;

                wrq = &tx->tx_wrq[tx->tx_nwrq];

                wrq->Next           = wrq + 1;
                wrq->WorkReqId      = kibnal_ptr2wreqid(tx, IBNAL_WID_RDMA);
                wrq->Operation      = WROpRdmaWrite;
                wrq->DSList         = gl;
                wrq->DSListDepth    = 1;
                wrq->MessageLen     = nob;

                wrq->Req.SendRC.ImmediateData                = 0;
                wrq->Req.SendRC.Options.s.SolicitedEvent     = 0;
                wrq->Req.SendRC.Options.s.SignaledCompletion = 0;
                wrq->Req.SendRC.Options.s.ImmediateData      = 0;
                wrq->Req.SendRC.Options.s.Fence              = 0; 

                wrq->Req.SendRC.RemoteDS.Address = dstfrag->rf_addr;
                wrq->Req.SendRC.RemoteDS.Rkey    = dstrd->rd_key;

                resid -= wrknob;
                if (wrknob < srcfrag->rf_nob) {
                        srcfrag->rf_addr += wrknob;
                        srcfrag->rf_nob -= wrknob;
                } else {
                        srcfrag++;
                        srcidx++;
                }
                
                if (wrknob < dstfrag->rf_nob) {
                        dstfrag->rf_addr += wrknob;
                        dstfrag->rf_nob -= wrknob;
                } else {
                        dstfrag++;
                        dstidx++;
                }
                
                tx->tx_nwrq++;
        }

        if (rc < 0)                             /* no RDMA if completing with failure */
                tx->tx_nwrq = 0;
#endif
        
        ibmsg->ibm_u.completion.ibcm_status = rc;
        ibmsg->ibm_u.completion.ibcm_cookie = dstcookie;
        kibnal_init_tx_msg(tx, type, sizeof (kib_completion_msg_t));

        return rc;
}

void
kibnal_queue_tx (kib_tx_t *tx, kib_conn_t *conn)
{
        spin_lock(&conn->ibc_lock);
        kibnal_queue_tx_locked (tx, conn);
        spin_unlock(&conn->ibc_lock);
        
        kibnal_check_sends(conn);
}

void
kibnal_schedule_active_connect_locked (kib_peer_t *peer, int proto_version)
{
        /* Called holding kib_global_lock exclusive with IRQs disabled */

        peer->ibp_version = proto_version;      /* proto version for new conn */
        peer->ibp_connecting++;                 /* I'm connecting */
        kibnal_peer_addref(peer);               /* extra ref for connd */

        spin_lock(&kibnal_data.kib_connd_lock);

        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_schedule_active_connect (kib_peer_t *peer, int proto_version)
{
        unsigned long flags;

        write_lock_irqsave(&kibnal_data.kib_global_lock, flags);

        kibnal_schedule_active_connect_locked(peer, proto_version);

        write_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
}

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

        /* 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_nwrq > 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); /* ...to 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;
                        tx->tx_waiting = 0;
                        kibnal_tx_done (tx);
                        return;
                }

                rc = kibnal_add_persistent_peer(nid);
                if (rc != 0) {
                        CERROR("Can't add peer %s: %d\n",
                               libcfs_nid2str(nid), rc);
                        
                        tx->tx_status = -EHOSTUNREACH;
                        tx->tx_waiting = 0;
                        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 for me... */
                write_unlock_irqrestore(g_lock, flags);
                
                kibnal_queue_tx (tx, conn);
                kibnal_conn_decref(conn);       /* ...until here */
                return;
        }

        if (!kibnal_peer_connecting(peer)) {
                if (!(peer->ibp_reconnect_interval == 0 || /* first attempt */
                      time_after_eq(jiffies, peer->ibp_reconnect_time))) {