openiblnd.h

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

typedef struct kib_rx                           /* receive message */
{
        struct list_head          rx_list;      /* queue for attention */
        struct kib_conn          *rx_conn;      /* owning conn */
        int                       rx_nob;       /* # bytes received (-1 while posted) */
        __u64                     rx_vaddr;     /* pre-mapped buffer (hca vaddr) */
        kib_msg_t                *rx_msg;       /* pre-mapped buffer (host vaddr) */
        struct ib_receive_param   rx_sp;        /* receive work item */
        struct ib_gather_scatter  rx_gl;        /* and it's memory */
} kib_rx_t;

typedef struct kib_tx                           /* transmit message */
{
        struct list_head          tx_list;      /* queue on idle_txs ibc_tx_queue etc. */
        struct kib_conn          *tx_conn;      /* owning conn */
        int                       tx_mapped;    /* mapped for RDMA? */
        int                       tx_sending;   /* # tx callbacks outstanding */
        int                       tx_status;    /* completion status */
        unsigned long             tx_deadline;  /* completion deadline */
        int                       tx_passive_rdma; /* peer sucks/blows */
        int                       tx_passive_rdma_wait; /* waiting for peer to complete */
        __u64                     tx_passive_rdma_cookie; /* completion cookie */
        lnet_msg_t               *tx_lntmsg[2]; /* ptl msgs to finalize on completion */
        kib_md_t                  tx_md;        /* RDMA mapping (active/passive) */
        __u64                     tx_vaddr;     /* pre-mapped buffer (hca vaddr) */
        kib_msg_t                *tx_msg;       /* pre-mapped buffer (host vaddr) */
        int                       tx_nsp;       /* # send work items */
        struct ib_send_param      tx_sp[2];     /* send work items... */
        struct ib_gather_scatter  tx_gl[2];     /* ...and their memory */
} kib_tx_t;

#define KIB_TX_UNMAPPED       0
#define KIB_TX_MAPPED         1
#define KIB_TX_MAPPED_FMR     2

typedef struct kib_connreq
{
        /* active connection-in-progress state */
        struct kib_conn           *cr_conn;
        kib_msg_t                  cr_msg;
        __u64                      cr_tid;
        tTS_IB_GID                 cr_gid;
        kib_svcrsp_t               cr_svcrsp;
        struct ib_path_record      cr_path;
        struct ib_cm_active_param  cr_connparam;
} kib_connreq_t;

typedef struct kib_conn
{
        struct kib_peer    *ibc_peer;           /* owning peer */
        struct list_head    ibc_list;           /* stash on peer's conn list */
        __u64               ibc_incarnation;    /* which instance of the peer */
        int                 ibc_version;        /* peer protocol version */
        atomic_t            ibc_refcount;       /* # users */
        int                 ibc_state;          /* what's happening */
        int                 ibc_nsends_posted;  /* # uncompleted sends */
        int                 ibc_credits;        /* # credits I have */
        int                 ibc_outstanding_credits; /* # credits to return */
        int                 ibc_reserved_credits; /* # credits for ACK/DONE msgs */
        unsigned long       ibc_last_send;      /* time of last send */
        struct list_head    ibc_tx_queue_nocred; /* sends that don't need a credit */
        struct list_head    ibc_tx_queue_rsrvd; /* sends that need a reserved cred */
        struct list_head    ibc_tx_queue;       /* send queue */
        struct list_head    ibc_active_txs;     /* active tx awaiting completion */
        spinlock_t          ibc_lock;           /* serialise */
        kib_rx_t           *ibc_rxs;            /* the rx descs */
        kib_pages_t        *ibc_rx_pages;       /* premapped rx msg pages */
        ib_qp_t            *ibc_qp;             /* queue pair */
        __u32               ibc_qpn;            /* queue pair number */
        tTS_IB_CM_COMM_ID   ibc_comm_id;        /* connection ID? */
        kib_connreq_t      *ibc_connreq;        /* connection request state */
} kib_conn_t;

#define IBNAL_CONN_INIT_NOTHING      0          /* initial state */
#define IBNAL_CONN_INIT_QP           1          /* ibc_qp set up */
#define IBNAL_CONN_CONNECTING        2          /* started to connect */
#define IBNAL_CONN_ESTABLISHED       3          /* connection established */
#define IBNAL_CONN_DEATHROW          4          /* waiting to be closed */
#define IBNAL_CONN_ZOMBIE            5          /* waiting to be freed */

typedef struct kib_peer
{
        struct list_head    ibp_list;           /* stash on global peer list */
        struct list_head    ibp_connd_list;     /* schedule on kib_connd_peers */
        lnet_nid_t          ibp_nid;            /* who's on the other end(s) */
        __u32               ibp_ip;             /* IP to query for peer conn params */
        int                 ibp_port;           /* port to qery for peer conn params */
        __u64               ibp_incarnation;    /* peer's incarnation */
        atomic_t            ibp_refcount;       /* # users */
        int                 ibp_persistence;    /* "known" peer refs */
        struct list_head    ibp_conns;          /* all active connections */
        struct list_head    ibp_tx_queue;       /* msgs waiting for a conn */
        int                 ibp_connecting;     /* current active connection attempts */
        int                 ibp_accepting;      /* current passive connection attempts */
        unsigned long       ibp_reconnect_time; /* when reconnect may be attempted */
        unsigned long       ibp_reconnect_interval; /* exponential backoff */
        int                 ibp_error;          /* errno on closing this peer */
        cfs_time_t          ibp_last_alive;     /* when (in jiffies) I was last alive */
} kib_peer_t;

extern kib_data_t      kibnal_data;
extern kib_tunables_t  kibnal_tunables;

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

/* these are purposely avoiding using local vars so they don't increase
 * stack consumption. */

#define kibnal_conn_addref(conn)                                \
do {                                                            \
        CDEBUG(D_NET, "conn[%p] (%d)++\n",                      \
               (conn), atomic_read(&(conn)->ibc_refcount));     \
        LASSERT(atomic_read(&(conn)->ibc_refcount) > 0);        \
        atomic_inc(&(conn)->ibc_refcount);                      \
} while (0)

#define kibnal_conn_decref(conn)                                              \
do {                                                                          \
        unsigned long   flags;                                                \
                                                                              \
        CDEBUG(D_NET, "conn[%p] (%d)--\n",                                    \
               (conn), atomic_read(&(conn)->ibc_refcount));                   \
        LASSERT(atomic_read(&(conn)->ibc_refcount) > 0);                      \
        if (atomic_dec_and_test(&(conn)->ibc_refcount)) {                     \
                spin_lock_irqsave(&kibnal_data.kib_reaper_lock, flags);       \
                list_add_tail(&(conn)->ibc_list,                              \
                              &kibnal_data.kib_reaper_conns);                 \
                wake_up(&kibnal_data.kib_reaper_waitq);                       \
                spin_unlock_irqrestore(&kibnal_data.kib_reaper_lock, flags);  \
        }                                                                     \
} while (0)

#define kibnal_peer_addref(peer)                                \
do {                                                            \
        CDEBUG(D_NET, "peer[%p] -> %s (%d)++\n",                \
               (peer), libcfs_nid2str((peer)->ibp_nid),         \
               atomic_read (&(peer)->ibp_refcount));            \
        LASSERT(atomic_read(&(peer)->ibp_refcount) > 0);        \
        atomic_inc(&(peer)->ibp_refcount);                      \
} while (0)

#define kibnal_peer_decref(peer)                                \
do {                                                            \
        CDEBUG(D_NET, "peer[%p] -> %s (%d)--\n",                \
               (peer), libcfs_nid2str((peer)->ibp_nid),         \
               atomic_read (&(peer)->ibp_refcount));            \
        LASSERT(atomic_read(&(peer)->ibp_refcount) > 0);        \
        if (atomic_dec_and_test(&(peer)->ibp_refcount))         \
                kibnal_destroy_peer(peer);                      \
} while (0)

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

static inline struct list_head *
kibnal_nid2peerlist (lnet_nid_t nid)
{
        unsigned int hash = ((unsigned int)nid) % kibnal_data.kib_peer_hash_size;

        return (&kibnal_data.kib_peers [hash]);
}

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

static inline void
kibnal_queue_tx_locked (kib_tx_t *tx, kib_conn_t *conn)
{
        struct list_head      *q;

        LASSERT (tx->tx_nsp > 0);               /* work items set up */
        LASSERT (tx->tx_conn == NULL);          /* only set here */

        kibnal_conn_addref(conn);
        tx->tx_conn = conn;
        tx->tx_deadline = jiffies + *kibnal_tunables.kib_timeout * HZ;

        if (conn->ibc_version == IBNAL_MSG_VERSION_RDMAREPLYNOTRSRVD) {
                /* All messages have simple credit control */
                q = &conn->ibc_tx_queue;
        } else {
                LASSERT (conn->ibc_version == IBNAL_MSG_VERSION);
                
                switch (tx->tx_msg->ibm_type) {
                case IBNAL_MSG_PUT_RDMA:
                case IBNAL_MSG_GET_RDMA:
                        /* RDMA request: reserve a buffer for the RDMA reply
                         * before sending */
                        q = &conn->ibc_tx_queue_rsrvd;
                        break;

                case IBNAL_MSG_PUT_DONE:
                case IBNAL_MSG_GET_DONE:
                        /* RDMA completion: no credits; peer has reserved a
                         * reply buffer */
                        q = &conn->ibc_tx_queue_nocred;
                        break;
                
                case IBNAL_MSG_NOOP:
                case IBNAL_MSG_IMMEDIATE:
                        /* Otherwise: consume a credit before sending */
                        q = &conn->ibc_tx_queue;
                        break;
                
                default:
                        LBUG();
                        q = NULL;
                }
        }

        list_add_tail(&tx->tx_list, q);
}

static inline int
kibnal_send_keepalive(kib_conn_t *conn) 
{
        return (*kibnal_tunables.kib_keepalive > 0) &&
                time_after(jiffies, conn->ibc_last_send +
                           *kibnal_tunables.kib_keepalive*HZ);
}

/* CAVEAT EMPTOR:
 * We rely on tx/rx descriptor alignment to allow us to use the lowest bit
 * of the work request id as a flag to determine if the completion is for a
 * transmit or a receive.  It seems that that the CQ entry's 'op' field
 * isn't always set correctly on completions that occur after QP teardown. */

static inline __u64
kibnal_ptr2wreqid (void *ptr, int isrx)
{
        unsigned long lptr = (unsigned long)ptr;

        LASSERT ((lptr & 1) == 0);
        return (__u64)(lptr | (isrx ? 1 : 0));
}

static inline void *
kibnal_wreqid2ptr (__u64 wreqid)
{
        return (void *)(((unsigned long)wreqid) & ~1UL);
}

static inline int
kibnal_wreqid_is_rx (__u64 wreqid)
{
        return (wreqid & 1) != 0;
}

#if (IB_NTXRXPARAMS == 3)
static inline int
kibnal_ib_send(ib_qp_t *qp, struct ib_send_param *p)
{
        return ib_send(qp, p, 1);
}

static inline int
kibnal_ib_receive(ib_qp_t *qp, struct ib_receive_param *p)
{
        return ib_receive(qp, p, 1);
}
#elif (IB_NTXRXPARAMS == 4)
static inline int
kibnal_ib_send(ib_qp_t *qp, struct ib_send_param *p)
{
        return ib_send(qp, p, 1, NULL);
}

static inline int
kibnal_ib_receive(ib_qp_t *qp, struct ib_receive_param *p)
{
        return ib_receive(qp, p, 1, NULL);
}
#else
 #error "IB_NTXRXPARAMS not set correctly"
#endif

int kibnal_startup (lnet_ni_t *ni);
void kibnal_shutdown (lnet_ni_t *ni);
int kibnal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg);
int kibnal_send (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg);
int kibnal_eager_recv (lnet_ni_t *ni, void *private, 
                       lnet_msg_t *lntmsg, void **new_private);
int kibnal_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 kibnal_accept(lnet_ni_t *ni, struct socket *sock);

extern void kibnal_init_msg(kib_msg_t *msg, int type, int body_nob);
extern void kibnal_pack_msg(kib_msg_t *msg, int version, int credits, 
                            lnet_nid_t dstnid, __u64 dststamp);
extern int kibnal_unpack_msg(kib_msg_t *msg, int expected_version, int nob);
extern void kibnal_handle_svcqry (struct socket *sock);
extern int kibnal_make_svcqry (kib_conn_t *conn);
extern void kibnal_free_acceptsock (kib_acceptsock_t *as);
extern int kibnal_create_peer (kib_peer_t **peerp, lnet_nid_t nid);
extern void kibnal_destroy_peer (kib_peer_t *peer);
extern int kibnal_add_persistent_peer(lnet_nid_t nid, __u32 ip, int port);
extern int kibnal_del_peer (lnet_nid_t nid);
extern kib_peer_t *kibnal_find_peer_locked (lnet_nid_t nid);
extern void kibnal_unlink_peer_locked (kib_peer_t *peer);
extern void kibnal_peer_alive(kib_peer_t *peer);
extern int  kibnal_close_stale_conns_locked (kib_peer_t *peer,
                                              __u64 incarnation);
extern kib_conn_t *kibnal_create_conn (void);
extern void kibnal_destroy_conn (kib_conn_t *conn);
extern int kibnal_alloc_pages (kib_pages_t **pp, int npages, int access);
extern void kibnal_free_pages (kib_pages_t *p);

extern void kibnal_check_sends (kib_conn_t *conn);

extern tTS_IB_CM_CALLBACK_RETURN
kibnal_bad_conn_callback (tTS_IB_CM_EVENT event, tTS_IB_CM_COMM_ID cid,
                          void *param, void *arg);
extern tTS_IB_CM_CALLBACK_RETURN
kibnal_conn_callback (tTS_IB_CM_EVENT event, tTS_IB_CM_COMM_ID cid,
                       void *param, void *arg);
extern tTS_IB_CM_CALLBACK_RETURN
kibnal_passive_conn_callback (tTS_IB_CM_EVENT event, tTS_IB_CM_COMM_ID cid,
                               void *param, void *arg);

extern void kibnal_close_conn_locked (kib_conn_t *conn, int error);
extern void kibnal_destroy_conn (kib_conn_t *conn);
extern int  kibnal_thread_start (int (*fn)(void *arg), void *arg);
extern int  kibnal_scheduler(void *arg);
extern int  kibnal_connd (void *arg);
extern int  kibnal_reaper (void *arg);
extern void kibnal_callback (ib_cq_t *cq, struct ib_cq_entry *e, void *arg);
extern void kibnal_txlist_done (struct list_head *txlist, int status);
extern void kibnal_init_tx_msg (kib_tx_t *tx, int type, int body_nob);
extern int  kibnal_close_conn (kib_conn_t *conn, int why);
extern void kibnal_start_active_rdma (int type, int status,
                                      kib_rx_t *rx, lnet_msg_t *lntmsg,
                                      unsigned int niov,
                                      struct iovec *iov, lnet_kiov_t *kiov,
                                      int offset, int nob);

extern int  kibnal_tunables_init(void);
extern void kibnal_tunables_fini(void);