ralnd.h

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

                kra_completion_msg_t  completion;
        }                    ram_u;
        __u32             ram_seq;              /* incrementing sequence number */
} kra_msg_t;

#define RANAL_MSG_MAGIC     LNET_PROTO_RA_MAGIC /* unique magic */
#define RANAL_MSG_VERSION              1        /* current protocol version */

#define RANAL_MSG_FENCE             0x80        /* fence RDMA */

#define RANAL_MSG_NONE              0x00        /* illegal message */
#define RANAL_MSG_NOOP              0x01        /* empty ram_u (keepalive) */
#define RANAL_MSG_IMMEDIATE         0x02        /* ram_u.immediate */
#define RANAL_MSG_PUT_REQ           0x03        /* ram_u.putreq (src->sink) */
#define RANAL_MSG_PUT_NAK           0x04        /* ram_u.completion (no PUT match: sink->src) */
#define RANAL_MSG_PUT_ACK           0x05        /* ram_u.putack (PUT matched: sink->src) */
#define RANAL_MSG_PUT_DONE          0x86        /* ram_u.completion (src->sink) */
#define RANAL_MSG_GET_REQ           0x07        /* ram_u.get (sink->src) */
#define RANAL_MSG_GET_NAK           0x08        /* ram_u.completion (no GET match: src->sink) */
#define RANAL_MSG_GET_DONE          0x89        /* ram_u.completion (src->sink) */
#define RANAL_MSG_CLOSE             0x8a        /* empty ram_u */

/***********************************************************************/

typedef struct kra_tx                           /* message descriptor */
{
        struct list_head          tx_list;      /* queue on idle_txs/rac_sendq/rac_waitq */
        struct kra_conn          *tx_conn;      /* owning conn */
        lnet_msg_t               *tx_lntmsg[2]; /* ptl msgs to finalize on completion */
        unsigned long             tx_qtime;     /* when tx started to wait for something (jiffies) */
        int                       tx_nob;       /* # bytes of payload */
        int                       tx_buftype;   /* payload buffer type */
        void                     *tx_buffer;    /* source/sink buffer */
        int                       tx_phys_offset; /* first page offset (if phys) */
        int                       tx_phys_npages; /* # physical pages */
        RAP_PHYS_REGION          *tx_phys;      /* page descriptors */
        RAP_MEM_KEY               tx_map_key;   /* mapping key */
        RAP_RDMA_DESCRIPTOR       tx_rdma_desc; /* rdma descriptor */
        __u64                     tx_cookie;    /* identify this tx to peer */
        kra_msg_t                 tx_msg;       /* FMA message buffer */
} kra_tx_t;

#define RANAL_BUF_NONE           0              /* buffer type not set */
#define RANAL_BUF_IMMEDIATE      1              /* immediate data */
#define RANAL_BUF_PHYS_UNMAPPED  2              /* physical: not mapped yet */
#define RANAL_BUF_PHYS_MAPPED    3              /* physical: mapped already */
#define RANAL_BUF_VIRT_UNMAPPED  4              /* virtual: not mapped yet */
#define RANAL_BUF_VIRT_MAPPED    5              /* virtual: mapped already */

typedef struct kra_conn
{
        struct kra_peer    *rac_peer;           /* owning peer */
        struct list_head    rac_list;           /* stash on peer's conn list */
        struct list_head    rac_hashlist;       /* stash in connection hash table */
        struct list_head    rac_schedlist;      /* schedule (on rad_???_conns) for attention */
        struct list_head    rac_fmaq;           /* txs queued for FMA */
        struct list_head    rac_rdmaq;          /* txs awaiting RDMA completion */
        struct list_head    rac_replyq;         /* txs awaiting replies */
        __u64               rac_peerstamp;      /* peer's unique stamp */
        __u64               rac_peer_connstamp; /* peer's unique connection stamp */
        __u64               rac_my_connstamp;   /* my unique connection stamp */
        unsigned long       rac_last_tx;        /* when I last sent an FMA message (jiffies) */
        unsigned long       rac_last_rx;        /* when I last received an FMA messages (jiffies) */
        long                rac_keepalive;      /* keepalive interval (seconds) */
        long                rac_timeout;        /* infer peer death if no rx for this many seconds */
        __u32               rac_cqid;           /* my completion callback id (non-unique) */
        __u32               rac_tx_seq;         /* tx msg sequence number */
        __u32               rac_rx_seq;         /* rx msg sequence number */
        atomic_t            rac_refcount;       /* # users */
        unsigned int        rac_close_sent;     /* I've sent CLOSE */
        unsigned int        rac_close_recvd;    /* I've received CLOSE */
        unsigned int        rac_state;          /* connection state */
        unsigned int        rac_scheduled;      /* being attented to */
        spinlock_t          rac_lock;           /* serialise */
        kra_device_t       *rac_device;         /* which device */
        RAP_PVOID           rac_rihandle;       /* RA endpoint */
        kra_msg_t          *rac_rxmsg;          /* incoming message (FMA prefix) */
        kra_msg_t           rac_msg;            /* keepalive/CLOSE message buffer */
} kra_conn_t;

#define RANAL_CONN_ESTABLISHED     0
#define RANAL_CONN_CLOSING         1
#define RANAL_CONN_CLOSED          2

typedef struct kra_peer
{
        struct list_head    rap_list;           /* stash on global peer list */
        struct list_head    rap_connd_list;     /* schedule on kra_connd_peers */
        struct list_head    rap_conns;          /* all active connections */
        struct list_head    rap_tx_queue;       /* msgs waiting for a conn */
        lnet_nid_t           rap_nid;            /* who's on the other end(s) */
        __u32               rap_ip;             /* IP address of peer */
        int                 rap_port;           /* port on which peer listens */
        atomic_t            rap_refcount;       /* # users */
        int                 rap_persistence;    /* "known" peer refs */
        int                 rap_connecting;     /* connection forming */
        unsigned long       rap_reconnect_time; /* CURRENT_SECONDS when reconnect OK */
        unsigned long       rap_reconnect_interval; /* exponential backoff */
} kra_peer_t;

extern kra_data_t      kranal_data;
extern kra_tunables_t  kranal_tunables;

extern void kranal_destroy_peer(kra_peer_t *peer);
extern void kranal_destroy_conn(kra_conn_t *conn);

static inline void
kranal_peer_addref(kra_peer_t *peer)
{
        CDEBUG(D_NET, "%p->%s\n", peer, libcfs_nid2str(peer->rap_nid));
        LASSERT(atomic_read(&peer->rap_refcount) > 0);
        atomic_inc(&peer->rap_refcount);
}

static inline void
kranal_peer_decref(kra_peer_t *peer)
{
        CDEBUG(D_NET, "%p->%s\n", peer, libcfs_nid2str(peer->rap_nid));
        LASSERT(atomic_read(&peer->rap_refcount) > 0);
        if (atomic_dec_and_test(&peer->rap_refcount))
                kranal_destroy_peer(peer);
}

static inline struct list_head *
kranal_nid2peerlist (lnet_nid_t nid)
{
        unsigned int hash = ((unsigned int)nid) % kranal_data.kra_peer_hash_size;

        return (&kranal_data.kra_peers[hash]);
}

static inline int
kranal_peer_active(kra_peer_t *peer)
{
        /* Am I in the peer hash table? */
        return (!list_empty(&peer->rap_list));
}

static inline void
kranal_conn_addref(kra_conn_t *conn)
{
        CDEBUG(D_NET, "%p->%s\n", conn, 
               libcfs_nid2str(conn->rac_peer->rap_nid));
        LASSERT(atomic_read(&conn->rac_refcount) > 0);
        atomic_inc(&conn->rac_refcount);
}

static inline void
kranal_conn_decref(kra_conn_t *conn)
{
        CDEBUG(D_NET, "%p->%s\n", conn,
               libcfs_nid2str(conn->rac_peer->rap_nid));
        LASSERT(atomic_read(&conn->rac_refcount) > 0);
        if (atomic_dec_and_test(&conn->rac_refcount))
                kranal_destroy_conn(conn);
}

static inline struct list_head *
kranal_cqid2connlist (__u32 cqid)
{
        unsigned int hash = cqid % kranal_data.kra_conn_hash_size;

        return (&kranal_data.kra_conns [hash]);
}

static inline kra_conn_t *
kranal_cqid2conn_locked (__u32 cqid)
{
        struct list_head *conns = kranal_cqid2connlist(cqid);
        struct list_head *tmp;
        kra_conn_t       *conn;

        list_for_each(tmp, conns) {
                conn = list_entry(tmp, kra_conn_t, rac_hashlist);

                if (conn->rac_cqid == cqid)
                        return conn;
        }

        return NULL;
}

static inline int
kranal_tx_mapped (kra_tx_t *tx)
{
        return (tx->tx_buftype == RANAL_BUF_VIRT_MAPPED ||
                tx->tx_buftype == RANAL_BUF_PHYS_MAPPED);
}

int kranal_startup (lnet_ni_t *ni);
void kranal_shutdown (lnet_ni_t *ni);
int kranal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg);
int kranal_send (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg);
int kranal_eager_recv(lnet_ni_t *ni, void *private, 
                        lnet_msg_t *lntmsg, void **new_private);
int kranal_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);
int kranal_accept(lnet_ni_t *ni, struct socket *sock);

extern void kranal_free_acceptsock (kra_acceptsock_t *ras);
extern int kranal_listener_procint (cfs_sysctl_table_t *table,
                                    int write, struct file *filp,
                                    void *buffer, size_t *lenp);
extern void kranal_update_reaper_timeout (long timeout);
extern void kranal_tx_done (kra_tx_t *tx, int completion);
extern void kranal_unlink_peer_locked (kra_peer_t *peer);
extern void kranal_schedule_conn (kra_conn_t *conn);
extern int kranal_create_peer (kra_peer_t **peerp, lnet_nid_t nid);
extern int kranal_add_persistent_peer (lnet_nid_t nid, __u32 ip, int port);
extern kra_peer_t *kranal_find_peer_locked (lnet_nid_t nid);
extern void kranal_post_fma (kra_conn_t *conn, kra_tx_t *tx);
extern int kranal_del_peer (lnet_nid_t nid);
extern void kranal_device_callback (RAP_INT32 devid, RAP_PVOID arg);
extern int kranal_thread_start (int(*fn)(void *arg), void *arg);
extern int kranal_connd (void *arg);
extern int kranal_reaper (void *arg);
extern int kranal_scheduler (void *arg);
extern void kranal_close_conn_locked (kra_conn_t *conn, int error);
extern void kranal_close_conn (kra_conn_t *conn, int error);
extern void kranal_terminate_conn_locked (kra_conn_t *conn);
extern void kranal_connect (kra_peer_t *peer);
extern int kranal_conn_handshake (struct socket *sock, kra_peer_t *peer);
extern int kranal_tunables_init(void);
extern void kranal_tunables_fini(void);
extern void kranal_init_msg(kra_msg_t *msg, int type);