mxlnd.h

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

        struct list_head    kmx_peers[MXLND_HASH_SIZE];
                                                /* list of all known peers */
        rwlock_t            kmx_peers_lock;     /* peer list rw lock */
        atomic_t            kmx_npeers;         /* number of peers */

        struct list_head    kmx_txs;            /* all tx descriptors */
        struct list_head    kmx_tx_idle;        /* list of idle tx */
        spinlock_t          kmx_tx_idle_lock;   /* lock for idle tx list */
        s32                 kmx_tx_used;        /* txs in use */
        u64                 kmx_tx_next_cookie; /* unique id for tx */
        struct list_head    kmx_tx_queue;       /* generic send queue */
        spinlock_t          kmx_tx_queue_lock;  /* lock for generic sends */
        struct semaphore    kmx_tx_queue_sem;   /* semaphore for tx queue */

        struct list_head    kmx_rxs;            /* all rx descriptors */
        spinlock_t          kmx_rxs_lock;       /* lock for rxs list */
        struct list_head    kmx_rx_idle;        /* list of idle tx */
        spinlock_t          kmx_rx_idle_lock;   /* lock for idle rx list */
} kmx_data_t;

#define MXLND_INIT_NOTHING      0       /* in the beginning, there was nothing... */
#define MXLND_INIT_DATA         1       /* main data structures created */
#define MXLND_INIT_TXS          2       /* tx descriptors created */
#define MXLND_INIT_RXS          3       /* initial rx descriptors created */
#define MXLND_INIT_MX           4       /* initiate MX library, open endpoint, get NIC id */
#define MXLND_INIT_THREADS      5       /* waitd, timeoutd, tx_queued threads */
#define MXLND_INIT_ALL          6       /* startup completed */

#include "mxlnd_wire.h"

enum kmx_req_type {
        MXLND_REQ_TX    = 0,
        MXLND_REQ_RX    = 1,
};

/* The life cycle of a request */
enum kmx_req_state {
        MXLND_CTX_INIT       = 0,               /* just created */
        MXLND_CTX_IDLE       = 1,               /* available for use */
        MXLND_CTX_PREP       = 2,               /* getting ready for send/recv */
        MXLND_CTX_PENDING    = 3,               /* mx_isend() or mx_irecv() called */
        MXLND_CTX_COMPLETED  = 4,               /* cleaning up after completion or timeout */
        MXLND_CTX_CANCELED   = 5,               /* timed out but still in ctx list */
};

/* Context Structure - generic tx/rx descriptor
 * It represents the context (or state) of each send or receive request.
 * In other LNDs, they have separate TX and RX descriptors and this replaces both.
 *
 * We will keep the these on the global kmx_rxs and kmx_txs lists for cleanup
 * during shutdown(). We will move them between the rx/tx idle lists and the
 * pending list which is monitored by mxlnd_timeoutd().
 */
struct kmx_ctx {
        enum kmx_req_type   mxc_type;           /* TX or RX */
        u64                 mxc_incarnation;    /* store the peer's incarnation here
                                                   to verify before changing flow
                                                   control credits after completion */
        unsigned long       mxc_deadline;       /* request time out in absolute jiffies */
        enum kmx_req_state  mxc_state;          /* what is the state of the request? */
        struct list_head    mxc_global_list;    /* place on kmx_rxs or kmx_txs */
        struct list_head    mxc_list;           /* place on rx/tx idle list, tx q, peer tx */
        struct list_head    mxc_rx_list;        /* place on mxp_rx_posted list */
        spinlock_t          mxc_lock;           /* lock */

        lnet_nid_t          mxc_nid;            /* dst's NID if peer is not known */
        struct kmx_peer    *mxc_peer;           /* owning peer */
        struct kmx_conn    *mxc_conn;           /* owning conn */
        struct kmx_msg     *mxc_msg;            /* msg hdr mapped to mxc_page */
        struct page        *mxc_page;           /* buffer for eager msgs */
        lnet_msg_t         *mxc_lntmsg[2];      /* lnet msgs to finalize */

        u8                  mxc_msg_type;       /* what type of message is this? */
        u64                 mxc_cookie;         /* completion cookie */
        u64                 mxc_match;          /* MX match info */
        mx_ksegment_t       mxc_seg;            /* local MX ksegment for non-DATA */
        mx_ksegment_t      *mxc_seg_list;       /* MX ksegment array for DATA */
        int                 mxc_nseg;           /* number of segments */
        unsigned long       mxc_pin_type;       /* MX_PIN_KERNEL or MX_PIN_PHYSICAL */
        u32                 mxc_nob;            /* number of bytes sent/received */
        mx_request_t        mxc_mxreq;          /* MX request */
        mx_status_t         mxc_status;         /* MX status */
        s64                 mxc_get;            /* # of times returned from idle list */
        s64                 mxc_put;            /* # of times returned from idle list */
};

#define MXLND_CONN_DISCONNECT  -2       /* conn is being destroyed - do not add txs */
#define MXLND_CONN_FAIL        -1       /* connect failed (bad handshake, unavail, etc.) */
#define MXLND_CONN_INIT         0       /* in the beginning, there was nothing... */
#define MXLND_CONN_REQ          1       /* a connection request message is needed */
#define MXLND_CONN_ACK          2       /* a connection ack is needed */
#define MXLND_CONN_WAIT         3       /* waiting for req or ack to complete */
#define MXLND_CONN_READY        4       /* ready to send */

/* connection state - queues for queued and pending msgs */
struct kmx_conn
{
        u64                 mxk_incarnation;    /* connections's incarnation value */
        atomic_t            mxk_refcount;       /* reference counting */

        struct kmx_peer    *mxk_peer;           /* owning peer */
        mx_endpoint_addr_t  mxk_epa;            /* peer's endpoint address */

        struct list_head    mxk_list;           /* for placing on mxp_conns */
        spinlock_t          mxk_lock;           /* lock */
        unsigned long       mxk_timeout;        /* expiration of oldest pending tx/rx */
        unsigned long       mxk_last_tx;        /* when last tx completed with success */
        unsigned long       mxk_last_rx;        /* when last rx completed */

        int                 mxk_credits;        /* # of my credits for sending to peer */
        int                 mxk_outstanding;    /* # of credits to return */

        int                 mxk_status;         /* can we send messages? MXLND_CONN_* */
        struct list_head    mxk_tx_credit_queue;   /* send queue for peer */
        struct list_head    mxk_tx_free_queue;  /* send queue for peer */
        int                 mxk_ntx_msgs;       /* # of msgs on tx queues */
        int                 mxk_ntx_data ;      /* # of DATA on tx queues */
        int                 mxk_ntx_posted;     /* # of tx msgs in flight */
        int                 mxk_data_posted;    /* # of tx data payloads in flight */

        struct list_head    mxk_pending;        /* in flight rxs and txs */
};

/* peer state */
struct kmx_peer
{
        lnet_nid_t          mxp_nid;            /* peer's LNET NID */
        u64                 mxp_incarnation;    /* peer's incarnation value */
        atomic_t            mxp_refcount;       /* reference counts */

        struct kmx_host    *mxp_host;           /* peer lookup info */
        u64                 mxp_nic_id;         /* remote's MX nic_id for mx_connect() */

        struct list_head    mxp_peers;          /* for placing on kmx_peers */
        spinlock_t          mxp_lock;           /* lock */

        struct list_head    mxp_conns;          /* list of connections */
        struct kmx_conn    *mxp_conn;           /* current connection */

        unsigned long       mxp_reconnect_time;  /* when to retry connect */
        int                 mxp_incompatible;   /* incorrect conn_req values */
};

extern kmx_data_t       kmxlnd_data;
extern kmx_tunables_t   kmxlnd_tunables;

/* required for the LNET API */
int  mxlnd_startup(lnet_ni_t *ni);
void mxlnd_shutdown(lnet_ni_t *ni);
int  mxlnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg);
int  mxlnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg);
int  mxlnd_recv(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg, int delayed,
                unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov, 
                unsigned int offset, unsigned int mlen, unsigned int rlen);

/* in mxlnd.c */
extern void mxlnd_thread_stop(long id);
extern int  mxlnd_ctx_alloc(struct kmx_ctx **ctxp, enum kmx_req_type type);
extern void mxlnd_ctx_free(struct kmx_ctx *ctx);
extern void mxlnd_ctx_init(struct kmx_ctx *ctx);
extern lnet_nid_t mxlnd_nic_id2nid(lnet_ni_t *ni, u64 nic_id);
extern u64 mxlnd_nid2nic_id(lnet_nid_t nid);

/* in mxlnd_cb.c */
void mxlnd_eager_recv(void *context, __u64 match_value, __u32 length);
extern mx_unexp_handler_action_t mxlnd_unexpected_recv(void *context,
                mx_endpoint_addr_t source, __u64 match_value, __u64 length, 
                void *data_if_available);
extern void mxlnd_peer_free(struct kmx_peer *peer);
extern void mxlnd_conn_free(struct kmx_conn *conn);
extern void mxlnd_sleep(unsigned long timeout);
extern int  mxlnd_tx_queued(void *arg);
extern void mxlnd_handle_rx_completion(struct kmx_ctx *rx);
extern int  mxlnd_check_sends(struct kmx_peer *peer);
extern int  mxlnd_tx_peer_queued(void *arg);
extern int  mxlnd_request_waitd(void *arg);
extern int  mxlnd_unex_recvd(void *arg);
extern int  mxlnd_timeoutd(void *arg);
extern int  mxlnd_connd(void *arg);

#define mxlnd_peer_addref(peer)                                 \
do {                                                            \
        LASSERT(atomic_read(&(peer)->mxp_refcount) > 0);        \
        atomic_inc(&(peer)->mxp_refcount);                      \
} while (0)


#define mxlnd_peer_decref(peer)                                 \
do {                                                            \
        LASSERT(atomic_read(&(peer)->mxp_refcount) > 0);        \
        if (atomic_dec_and_test(&(peer)->mxp_refcount))         \
                mxlnd_peer_free(peer);                          \
} while (0)

#define mxlnd_conn_addref(conn)                                 \
do {                                                            \
        LASSERT(atomic_read(&(conn)->mxk_refcount) > 0);        \
        atomic_inc(&(conn)->mxk_refcount);                      \
} while (0)


#define mxlnd_conn_decref(conn)                                 \
do {                                                            \
        LASSERT(atomic_read(&(conn)->mxk_refcount) > 0);        \
        if (atomic_dec_and_test(&(conn)->mxk_refcount))         \
                mxlnd_conn_free(conn);                          \
} while (0)