mxlnd_cb.c

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

        int                      ret    = -ENOENT;
        struct kmx_peer         *peer   = NULL;
        struct kmx_conn         *conn   = NULL;

        read_lock(&kmxlnd_data.kmx_peers_lock);
        for (i = 0; i < MXLND_HASH_SIZE; i++) {
                list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i], mxp_peers) {
                        conn = peer->mxp_conn;
                        if (index-- > 0)
                                continue;

                        *nidp = peer->mxp_nid;
                        *count = atomic_read(&peer->mxp_refcount);
                        ret = 0;
                        break;
                }
        }
        read_unlock(&kmxlnd_data.kmx_peers_lock);
        
        return ret;
}

void
mxlnd_del_peer_locked(struct kmx_peer *peer)
{
        list_del_init(&peer->mxp_peers); /* remove from the global list */
        if (peer->mxp_conn) mxlnd_conn_disconnect(peer->mxp_conn, 0, 0);
        mxlnd_peer_decref(peer); /* drop global list ref */
        return;
}

int
mxlnd_del_peer(lnet_nid_t nid)
{
        int             i       = 0;
        int             ret     = 0;
        struct kmx_peer *peer   = NULL;
        struct kmx_peer *next   = NULL;

        if (nid != LNET_NID_ANY) {
                peer = mxlnd_find_peer_by_nid(nid);
        }
        write_lock(&kmxlnd_data.kmx_peers_lock);
        if (nid != LNET_NID_ANY) {
                if (peer == NULL) {
                        ret = -ENOENT;
                } else {
                        mxlnd_del_peer_locked(peer);
                }
        } else { /* LNET_NID_ANY */
                for (i = 0; i < MXLND_HASH_SIZE; i++) {
                        list_for_each_entry_safe(peer, next, 
                                                 &kmxlnd_data.kmx_peers[i], mxp_peers) {
                                mxlnd_del_peer_locked(peer);
                        }
                }
        }
        write_unlock(&kmxlnd_data.kmx_peers_lock);

        return ret;
}

struct kmx_conn *
mxlnd_get_conn_by_idx(int index)
{
        int                      i      = 0;
        struct kmx_peer         *peer   = NULL;
        struct kmx_conn         *conn   = NULL;

        read_lock(&kmxlnd_data.kmx_peers_lock);
        for (i = 0; i < MXLND_HASH_SIZE; i++) {
                list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i], mxp_peers) {
                        list_for_each_entry(conn, &peer->mxp_conns, mxk_list) {
                                if (index-- > 0)
                                        continue;

                                mxlnd_conn_addref(conn); /* add ref here, dec in ctl() */
                                read_unlock(&kmxlnd_data.kmx_peers_lock);
                                return conn;
                        }
                }
        }
        read_unlock(&kmxlnd_data.kmx_peers_lock);
        
        return NULL;
}

void
mxlnd_close_matching_conns_locked(struct kmx_peer *peer)
{
        struct kmx_conn *conn   = NULL;
        struct kmx_conn *next   = NULL;

        list_for_each_entry_safe(conn, next, &peer->mxp_conns, mxk_list) {
                mxlnd_conn_disconnect(conn, 0 , 0);
        }
        return;
}

int
mxlnd_close_matching_conns(lnet_nid_t nid)
{
        int             i       = 0;
        int             ret     = 0;
        struct kmx_peer *peer   = NULL;

        read_lock(&kmxlnd_data.kmx_peers_lock);
        if (nid != LNET_NID_ANY) {
                peer = mxlnd_find_peer_by_nid(nid);
                if (peer == NULL) {
                        ret = -ENOENT;
                } else {
                        mxlnd_close_matching_conns_locked(peer);
                }
        } else { /* LNET_NID_ANY */
                for (i = 0; i < MXLND_HASH_SIZE; i++) {
                        list_for_each_entry(peer, &kmxlnd_data.kmx_peers[i], mxp_peers)
                                mxlnd_close_matching_conns_locked(peer);
                }
        }
        read_unlock(&kmxlnd_data.kmx_peers_lock);

        return ret;
}

/**
 * mxlnd_ctl - modify MXLND parameters
 * @ni - LNET interface handle
 * @cmd - command to change
 * @arg - the ioctl data
 *
 * Not implemented yet.
 */
int
mxlnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
{
        struct libcfs_ioctl_data *data  = arg;
        int                       ret   = -EINVAL;

        LASSERT (ni == kmxlnd_data.kmx_ni);

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

                ret = mxlnd_get_peer_info(data->ioc_count, &nid, &count);
                data->ioc_nid    = nid;
                data->ioc_count  = count;
                break;
        }
        case IOC_LIBCFS_DEL_PEER: {
                ret = mxlnd_del_peer(data->ioc_nid);
                break;
        }
        case IOC_LIBCFS_GET_CONN: {
                struct kmx_conn *conn = NULL;

                conn = mxlnd_get_conn_by_idx(data->ioc_count);
                if (conn == NULL) {
                        ret = -ENOENT;
                } else {
                        ret = 0;
                        data->ioc_nid = conn->mxk_peer->mxp_nid;
                        mxlnd_conn_decref(conn); /* dec ref taken in get_conn_by_idx() */
                }
                break;
        }
        case IOC_LIBCFS_CLOSE_CONNECTION: {
                ret = mxlnd_close_matching_conns(data->ioc_nid);
                break;
        }
        default:
                CDEBUG(D_NETERROR, "unknown ctl(%d)\n", cmd);
                break;
        }
        
        return ret;
}

/**
 * mxlnd_peer_queue_tx_locked - add the tx to the global tx queue
 * @tx
 *
 * Add the tx to the peer's msg or data queue. The caller has locked the peer.
 */
void
mxlnd_peer_queue_tx_locked(struct kmx_ctx *tx)
{
        u8                      msg_type        = tx->mxc_msg_type;
        //struct kmx_peer         *peer           = tx->mxc_peer;
        struct kmx_conn         *conn           = tx->mxc_conn;

        LASSERT (msg_type != 0);
        LASSERT (tx->mxc_nid != 0);
        LASSERT (tx->mxc_peer != NULL);
        LASSERT (tx->mxc_conn != NULL);

        tx->mxc_incarnation = conn->mxk_incarnation;

        if (msg_type != MXLND_MSG_PUT_DATA &&
            msg_type != MXLND_MSG_GET_DATA) {
                /* msg style tx */
                if (mxlnd_tx_requires_credit(tx)) {
                        list_add_tail(&tx->mxc_list, &conn->mxk_tx_credit_queue);
                        conn->mxk_ntx_msgs++;
                } else if (msg_type == MXLND_MSG_CONN_REQ ||
                           msg_type == MXLND_MSG_CONN_ACK) {
                        /* put conn msgs at the front of the queue */
                        list_add(&tx->mxc_list, &conn->mxk_tx_free_queue);
                } else {
                        /* PUT_ACK, PUT_NAK */
                        list_add_tail(&tx->mxc_list, &conn->mxk_tx_free_queue);
                        conn->mxk_ntx_msgs++;
                }
        } else {
                /* data style tx */
                list_add_tail(&tx->mxc_list, &conn->mxk_tx_free_queue);
                conn->mxk_ntx_data++;
        }

        return;
}

/**
 * mxlnd_peer_queue_tx - add the tx to the global tx queue
 * @tx
 *
 * Add the tx to the peer's msg or data queue
 */
static inline void
mxlnd_peer_queue_tx(struct kmx_ctx *tx)
{
        LASSERT(tx->mxc_peer != NULL);
        LASSERT(tx->mxc_conn != NULL);
        spin_lock(&tx->mxc_conn->mxk_lock);
        mxlnd_peer_queue_tx_locked(tx);
        spin_unlock(&tx->mxc_conn->mxk_lock);

        return;
}

/**
 * mxlnd_queue_tx - add the tx to the global tx queue
 * @tx
 *
 * Add the tx to the global queue and up the tx_queue_sem
 */
void
mxlnd_queue_tx(struct kmx_ctx *tx)
{
        int             ret     = 0;
        struct kmx_peer *peer   = tx->mxc_peer;
        LASSERT (tx->mxc_nid != 0);

        if (peer != NULL) {
                if (peer->mxp_incompatible &&
                    tx->mxc_msg_type != MXLND_MSG_CONN_ACK) {
                        /* let this fail now */
                        tx->mxc_status.code = -ECONNABORTED;
                        mxlnd_put_idle_tx(tx);
                        return;
                }
                if (tx->mxc_conn == NULL) {
                        mxlnd_conn_alloc(&tx->mxc_conn, peer);
                }
                LASSERT(tx->mxc_conn != NULL);
                mxlnd_peer_queue_tx(tx);
                ret = mxlnd_check_sends(peer);
        } else {
                spin_lock(&kmxlnd_data.kmx_tx_queue_lock);
                list_add_tail(&tx->mxc_list, &kmxlnd_data.kmx_tx_queue);
                spin_unlock(&kmxlnd_data.kmx_tx_queue_lock);
                up(&kmxlnd_data.kmx_tx_queue_sem);
        }
        return;
}

int
mxlnd_setup_iov(struct kmx_ctx *ctx, u32 niov, struct iovec *iov, u32 offset, u32 nob)
{
        int             i                       = 0;
        int             sum                     = 0;
        int             old_sum                 = 0;
        int             nseg                    = 0;
        int             first_iov               = -1;
        int             first_iov_offset        = 0;
        int             first_found             = 0;
        int             last_iov                = -1;
        int             last_iov_length         = 0;
        mx_ksegment_t  *seg                     = NULL;

        if (niov == 0) return 0;
        LASSERT(iov != NULL);

        for (i = 0; i < niov; i++) {
                sum = old_sum + (u32) iov[i].iov_len;
                if (!first_found && (sum > offset)) {
                        first_iov = i;
                        first_iov_offset = offset - old_sum;
                        first_found = 1;
                        sum = (u32) iov[i].iov_len - first_iov_offset;
                        old_sum = 0;
                }
                if (sum >= nob) {
                        last_iov = i;
                        last_iov_length = (u32) iov[i].iov_len - (sum - nob);
                        if (first_iov == last_iov) last_iov_length -= first_iov_offset;
                        break;
                }
                old_sum = sum;
        }
        LASSERT(first_iov >= 0 && last_iov >= first_iov);
        nseg = last_iov - first_iov + 1;
        LASSERT(nseg > 0);
        
        MXLND_ALLOC (seg, nseg * sizeof(*seg));
        if (seg == NULL) {
                CDEBUG(D_NETERROR, "MXLND_ALLOC() failed\n");
                return -1;
        }
        memset(seg, 0, nseg * sizeof(*seg));
        ctx->mxc_nseg = nseg;
        sum = 0;
        for (i = 0; i < nseg; i++) {
                seg[i].segment_ptr = MX_KVA_TO_U64(iov[first_iov + i].iov_base);
                seg[i].segment_length = (u32) iov[first_iov + i].iov_len;
                if (i == 0) {
                        seg[i].segment_ptr += (u64) first_iov_offset;
                        seg[i].segment_length -= (u32) first_iov_offset;
                }
                if (i == (nseg - 1)) {
                        seg[i].segment_length = (u32) last_iov_length;
                }
                sum += seg[i].segment_length;
        }
        ctx->mxc_seg_list = seg;
        ctx->mxc_pin_type = MX_PIN_KERNEL;
#ifdef MX_PIN_FULLPAGES
        ctx->mxc_pin_type |= MX_PIN_FULLPAGES;
#endif
        LASSERT(nob == sum);
        return 0;
}

int
mxlnd_setup_kiov(struct kmx_ctx *ctx, u32 niov, lnet_kiov_t *kiov, u32 offset, u32 nob)
{
        int             i                       = 0;
        int             sum                     = 0;
        int             old_sum                 = 0;
        int             nseg                    = 0;
        int             first_kiov              = -1;
        int             first_kiov_offset       = 0;
        int             first_found             = 0;
        int             last_kiov               = -1;
        int             last_kiov_length        = 0;
        mx_ksegment_t  *seg                     = NULL;

        if (niov == 0) return 0;
        LASSERT(kiov != NULL);

        for (i = 0; i < niov; i++) {