openiblnd_cb.c

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

kibnal_peer_alive (kib_peer_t *peer)
{
        /* This is racy, but everyone's only writing cfs_time_current() */
        peer->ibp_last_alive = cfs_time_current();
        mb();
}

void
kibnal_peer_notify (kib_peer_t *peer)
{
        time_t        last_alive = 0;
        int           error = 0;
        unsigned long flags;
        
        read_lock_irqsave(&kibnal_data.kib_global_lock, flags);

        if (list_empty(&peer->ibp_conns) &&
            peer->ibp_accepting == 0 &&
            peer->ibp_connecting == 0 &&
            peer->ibp_error != 0) {
                error = peer->ibp_error;
                peer->ibp_error = 0;
                last_alive = cfs_time_current_sec() -
                             cfs_duration_sec(cfs_time_current() -
                                              peer->ibp_last_alive);
        }
        
        read_unlock_irqrestore(&kibnal_data.kib_global_lock, flags);
        
        if (error != 0)
                lnet_notify(kibnal_data.kib_ni, peer->ibp_nid, 0, last_alive);
}

void
kibnal_close_conn_locked (kib_conn_t *conn, int error)
{
        /* This just does the immmediate housekeeping, and schedules the
         * connection for the reaper to finish off.
         * Caller holds kib_global_lock exclusively in irq context */
        kib_peer_t   *peer = conn->ibc_peer;

        CDEBUG (error == 0 ? D_NET : D_NETERROR,
                "closing conn to %s: error %d\n", 
                libcfs_nid2str(peer->ibp_nid), error);
        
        LASSERT (conn->ibc_state == IBNAL_CONN_ESTABLISHED ||
                 conn->ibc_state == IBNAL_CONN_CONNECTING);

        if (conn->ibc_state == IBNAL_CONN_ESTABLISHED) {
                /* kib_reaper_conns takes ibc_list's ref */
                list_del (&conn->ibc_list);
        } else {
                /* new ref for kib_reaper_conns */
                kibnal_conn_addref(conn);
        }
        
        if (list_empty (&peer->ibp_conns)) {   /* no more conns */
                if (peer->ibp_persistence == 0 && /* non-persistent peer */
                    kibnal_peer_active(peer))     /* still in peer table */
                        kibnal_unlink_peer_locked (peer);

                peer->ibp_error = error; /* set/clear error on last conn */
        }

        conn->ibc_state = IBNAL_CONN_DEATHROW;

        /* Schedule conn for closing/destruction */
        spin_lock (&kibnal_data.kib_reaper_lock);

        list_add_tail (&conn->ibc_list, &kibnal_data.kib_reaper_conns);
        wake_up (&kibnal_data.kib_reaper_waitq);
                
        spin_unlock (&kibnal_data.kib_reaper_lock);
}

int
kibnal_close_conn (kib_conn_t *conn, int why)
{
        unsigned long     flags;
        int               count = 0;

        write_lock_irqsave (&kibnal_data.kib_global_lock, flags);

        LASSERT (conn->ibc_state >= IBNAL_CONN_CONNECTING);
        
        if (conn->ibc_state <= IBNAL_CONN_ESTABLISHED) {
                count = 1;
                kibnal_close_conn_locked (conn, why);
        }
        
        write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
        return (count);
}

void
kibnal_peer_connect_failed (kib_peer_t *peer, int active, int error)
{
        LIST_HEAD        (zombies);
        unsigned long     flags;

        LASSERT(error != 0);

        write_lock_irqsave (&kibnal_data.kib_global_lock, flags);

        if (active) {
                LASSERT (peer->ibp_connecting != 0);
                peer->ibp_connecting--;
        } else {
                LASSERT (peer->ibp_accepting != 0);
                peer->ibp_accepting--;
        }

        if (peer->ibp_connecting != 0 ||
            peer->ibp_accepting != 0) {
                /* another connection attempt under way... */
                write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
                return;
        }

        if (list_empty(&peer->ibp_conns)) {
                /* Say when active connection can be re-attempted */
                peer->ibp_reconnect_interval *= 2;
                peer->ibp_reconnect_interval =
                        MAX(peer->ibp_reconnect_interval,
                            *kibnal_tunables.kib_min_reconnect_interval);
                peer->ibp_reconnect_interval =
                        MIN(peer->ibp_reconnect_interval,
                            *kibnal_tunables.kib_max_reconnect_interval);
                
                peer->ibp_reconnect_time = jiffies + 
                                           peer->ibp_reconnect_interval * HZ;
        
                /* Take peer's blocked transmits; I'll complete
                 * them with error */
                list_add(&zombies, &peer->ibp_tx_queue);
                list_del_init(&peer->ibp_tx_queue);
                
                if (kibnal_peer_active(peer) &&
                    (peer->ibp_persistence == 0)) {
                        /* failed connection attempt on non-persistent peer */
                        kibnal_unlink_peer_locked (peer);
                }

                peer->ibp_error = error;
        } else {
                /* Can't have blocked transmits if there are connections */
                LASSERT (list_empty(&peer->ibp_tx_queue));
        }
        
        write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);

        kibnal_peer_notify(peer);
        
        if (!list_empty (&zombies))
                CDEBUG (D_NETERROR, "Deleting messages for %s: connection failed\n",
                        libcfs_nid2str(peer->ibp_nid));

        kibnal_txlist_done(&zombies, -EHOSTUNREACH);
}

void
kibnal_connreq_done (kib_conn_t *conn, int active, int status)
{
        int               state = conn->ibc_state;
        kib_peer_t       *peer = conn->ibc_peer;
        kib_tx_t         *tx;
        unsigned long     flags;
        int               rc;
        int               i;

        if (conn->ibc_connreq != NULL) {
                LIBCFS_FREE (conn->ibc_connreq, sizeof (*conn->ibc_connreq));
                conn->ibc_connreq = NULL;
        }

        switch (state) {
        case IBNAL_CONN_CONNECTING:
                /* conn has a CM comm_id */
                if (status == 0) {
                        /* Install common (active/passive) callback for
                         * disconnect/idle notification */
                        rc = tsIbCmCallbackModify(conn->ibc_comm_id, 
                                                  kibnal_conn_callback,
                                                  conn);
                        LASSERT (rc == 0);
                } else {
                        /* LASSERT (no more CM callbacks) */
                        rc = tsIbCmCallbackModify(conn->ibc_comm_id,
                                                  kibnal_bad_conn_callback,
                                                  conn);
                        LASSERT (rc == 0);
                }
                break;
                
        case IBNAL_CONN_INIT_QP:
                LASSERT (status != 0);
                break;
                
        default:
                LBUG();
        }
        
        write_lock_irqsave (&kibnal_data.kib_global_lock, flags);

        if (active)
                LASSERT (peer->ibp_connecting != 0);
        else
                LASSERT (peer->ibp_accepting != 0);
        
        if (status == 0 &&                      /* connection established */
            kibnal_peer_active(peer)) {         /* peer not deleted */

                if (active)
                        peer->ibp_connecting--;
                else
                        peer->ibp_accepting--;

                conn->ibc_last_send = jiffies;
                conn->ibc_state = IBNAL_CONN_ESTABLISHED;
                kibnal_peer_alive(peer);

                /* +1 ref for ibc_list; caller(== CM)'s ref remains until
                 * the IB_CM_IDLE callback */
                kibnal_conn_addref(conn);
                list_add (&conn->ibc_list, &peer->ibp_conns);

                peer->ibp_reconnect_interval = 0; /* OK to reconnect at any time */

                /* post blocked sends to the new connection */
                spin_lock (&conn->ibc_lock);
                
                while (!list_empty (&peer->ibp_tx_queue)) {
                        tx = list_entry (peer->ibp_tx_queue.next, 
                                         kib_tx_t, tx_list);
                        
                        list_del (&tx->tx_list);

                        kibnal_queue_tx_locked (tx, conn);
                }
                
                spin_unlock (&conn->ibc_lock);

                /* Nuke any dangling conns from a different peer instance... */
                kibnal_close_stale_conns_locked (conn->ibc_peer,
                                                 conn->ibc_incarnation);

                write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);

                /* queue up all the receives */
                for (i = 0; i < IBNAL_RX_MSGS; i++) {
                        /* +1 ref for rx desc */
                        kibnal_conn_addref(conn);

                        CDEBUG(D_NET, "RX[%d] %p->%p - "LPX64"\n",
                               i, &conn->ibc_rxs[i], conn->ibc_rxs[i].rx_msg,
                               conn->ibc_rxs[i].rx_vaddr);

                        kibnal_post_rx (&conn->ibc_rxs[i], 0, 0);
                }

                kibnal_check_sends (conn);
                return;
        }

        if (status == 0) {
                /* connection established, but peer was deleted.  Schedule for
                 * reaper to cm_disconnect... */
                status = -ECONNABORTED;
                kibnal_close_conn_locked (conn, status);
        } else {
                /* just waiting for refs to drain */
                conn->ibc_state = IBNAL_CONN_ZOMBIE;
        } 

        write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);

        kibnal_peer_connect_failed (conn->ibc_peer, active, status);
}

int
kibnal_accept_connreq (kib_conn_t **connp, tTS_IB_CM_COMM_ID cid,
                       kib_msg_t *msg, int nob)
{
        kib_conn_t    *conn;
        kib_peer_t    *peer;
        kib_peer_t    *peer2;
        unsigned long  flags;
        int            rc;

        rc = kibnal_unpack_msg(msg, 0, nob);
        if (rc != 0) {
                CERROR("Can't unpack connreq msg: %d\n", rc);
                return -EPROTO;
        }

        CDEBUG(D_NET, "connreq from %s\n", libcfs_nid2str(msg->ibm_srcnid));

        if (msg->ibm_type != IBNAL_MSG_CONNREQ) {
                CERROR("Unexpected connreq msg type: %x from %s\n",
                       msg->ibm_type, libcfs_nid2str(msg->ibm_srcnid));
                return -EPROTO;
        }
                
        if (msg->ibm_u.connparams.ibcp_queue_depth != IBNAL_MSG_QUEUE_SIZE) {
                CERROR("Can't accept %s: bad queue depth %d (%d expected)\n",
                       libcfs_nid2str(msg->ibm_srcnid), 
                       msg->ibm_u.connparams.ibcp_queue_depth, 
                       IBNAL_MSG_QUEUE_SIZE);
                return (-EPROTO);
        }
        
        conn = kibnal_create_conn();
        if (conn == NULL)
                return (-ENOMEM);

        /* assume 'nid' is a new peer */
        rc = kibnal_create_peer(&peer, msg->ibm_srcnid);
        if (rc != 0) {
                kibnal_conn_decref(conn);
                return (-ENOMEM);
        }
        
        write_lock_irqsave (&kibnal_data.kib_global_lock, flags);

        if (kibnal_data.kib_nonewpeers) {
                write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
                
                CERROR ("Shutdown has started, drop connreq from %s\n",
                        libcfs_nid2str(msg->ibm_srcnid));
                kibnal_conn_decref(conn);
                kibnal_peer_decref(peer);
                return -ESHUTDOWN;
        }

        /* Check I'm the same instance that gave the connection parameters.  
         * NB If my incarnation changes after this, the peer will get nuked and
         * we'll spot that when the connection is finally added into the peer's
         * connlist */
        if (!lnet_ptlcompat_matchnid(kibnal_data.kib_ni->ni_nid,
                                     msg->ibm_dstnid) ||
            msg->ibm_dststamp != kibnal_data.kib_incarnation) {
                write_unlock_irqrestore (&kibnal_data.kib_global_lock, flags);
                
                CERROR("Stale connection params from %s\n",
                       libcfs_nid2str(msg->ibm_srcnid));
                kibnal_conn_decref(conn);
                kibnal_peer_decref(peer);
                return -ESTALE;
        }

        peer2 = kibnal_find_peer_locked(msg->ibm_srcnid);
        if (peer2 == NULL) {
                /* Brand new peer */
                LASSERT (peer->ibp_accepting == 0);

                /* peer table takes my ref on peer */
                list_add_tail (&peer->ibp_list,
                               kibnal_nid2peerlist(msg->ibm_srcnid));
        } else {
                /* tie-break connection race in favour of the higher NID */                
                if (peer2->ibp_connecting != 0 &&
                    msg->ibm_srcnid < kibnal_data.kib_ni->ni_nid) {