iiblnd.h

来自「非常经典的一个分布式系统」· C头文件 代码 · 共 740 行 · 第 1/3 页

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 int
kibnal_peer_connecting(kib_peer_t *peer)
{
        /* Am I expecting a connection to materialise? */
        return (peer->ibp_connecting != 0 ||
                peer->ibp_accepting != 0 ||
                peer->ibp_passivewait);
}

static inline void
kibnal_queue_tx_locked (kib_tx_t *tx, kib_conn_t *conn)
{
        struct list_head  *q;
        
        LASSERT (tx->tx_nwrq > 0);              /* work items set up */
        LASSERT (!tx->tx_queued);               /* not queued for sending already */

        tx->tx_queued = 1;
        tx->tx_deadline = jiffies + (*kibnal_tunables.kib_timeout * HZ);

        if (tx->tx_conn == NULL) {
                kibnal_conn_addref(conn);
                tx->tx_conn = conn;
                LASSERT (tx->tx_msg->ibm_type != IBNAL_MSG_PUT_DONE);
        } else {
                LASSERT (tx->tx_conn == conn);
                LASSERT (tx->tx_msg->ibm_type == IBNAL_MSG_PUT_DONE);
        }

        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_REQ:
                case IBNAL_MSG_GET_REQ:
                        /* RDMA request: reserve a buffer for the RDMA reply
                         * before sending */
                        q = &conn->ibc_tx_queue_rsrvd;
                        break;

                case IBNAL_MSG_PUT_NAK:
                case IBNAL_MSG_PUT_ACK:
                case IBNAL_MSG_PUT_DONE:
                case IBNAL_MSG_GET_DONE:
                        /* RDMA reply/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);
}

#define KIBNAL_SERVICE_KEY_MASK  (IB_SERVICE_RECORD_COMP_SERVICENAME |          \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_1 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_2 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_3 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_4 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_5 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_6 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_7 |       \
                                  IB_SERVICE_RECORD_COMP_SERVICEDATA8_8)

static inline __u64*
kibnal_service_nid_field(IB_SERVICE_RECORD *srv)
{
        /* must be consistent with KIBNAL_SERVICE_KEY_MASK */
        return (__u64 *)srv->ServiceData8;
}

static inline void
kibnal_set_service_keys(IB_SERVICE_RECORD *srv, lnet_nid_t nid)
{
        char *svc_name = *kibnal_tunables.kib_service_name;

        LASSERT (strlen(svc_name) < sizeof(srv->ServiceName));
        memset (srv->ServiceName, 0, sizeof(srv->ServiceName));
        strcpy (srv->ServiceName, svc_name);

        *kibnal_service_nid_field(srv) = cpu_to_le64(nid);
}

/* CAVEAT EMPTOR: We rely on tx/rx descriptor alignment to allow us to use the
 * lowest 2 bits of the work request id to stash the work item type (the op
 * field is not valid when the wc completes in error). */

#define IBNAL_WID_TX    0
#define IBNAL_WID_RX    1
#define IBNAL_WID_RDMA  2
#define IBNAL_WID_MASK  3UL

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

        LASSERT ((lptr & IBNAL_WID_MASK) == 0);
        LASSERT ((type & ~IBNAL_WID_MASK) == 0);
        return (__u64)(lptr | type);
}

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

static inline int
kibnal_wreqid2type (__u64 wreqid)
{
        return (wreqid & IBNAL_WID_MASK);
}

static inline void
kibnal_set_conn_state (kib_conn_t *conn, int state)
{
        CDEBUG(D_NET,"%p state %d\n", conn, state);
        conn->ibc_state = state;
        mb();
}

#if IBNAL_USE_FMR

static inline int
kibnal_rd_size (kib_rdma_desc_t *rd) 
{
        return rd->rd_nob;
}

#else
static inline int
kibnal_rd_size (kib_rdma_desc_t *rd)
{
        int   i;
        int   size;
        
        for (i = size = 0; i < rd->rd_nfrag; i++)
                size += rd->rd_frags[i].rf_nob;
        
        return size;
}
#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 *msg,
                  int delayed, unsigned int niov,
                  struct iovec *iov, lnet_kiov_t *kiov,
                  unsigned int offset, unsigned int mlen, unsigned int rlen);
void kibnal_init_msg(kib_msg_t *msg, int type, int body_nob);
void kibnal_pack_msg(kib_msg_t *msg, __u32 version, int credits, 
                     lnet_nid_t dstnid, __u64 dststamp, __u64 seq);
void kibnal_pack_connmsg(kib_msg_t *msg, __u32 version, int nob, int type,
                         lnet_nid_t dstnid, __u64 dststamp);
int  kibnal_unpack_msg(kib_msg_t *msg, __u32 expected_version, int nob);
IB_HANDLE kibnal_create_cep(lnet_nid_t nid);
int  kibnal_create_peer (kib_peer_t **peerp, lnet_nid_t nid);
void kibnal_destroy_peer (kib_peer_t *peer);
kib_peer_t *kibnal_find_peer_locked (lnet_nid_t nid);
int  kibnal_del_peer (lnet_nid_t nid);
void kibnal_peer_alive (kib_peer_t *peer);
void kibnal_unlink_peer_locked (kib_peer_t *peer);
int  kibnal_add_persistent_peer (lnet_nid_t nid);
int  kibnal_close_stale_conns_locked (kib_peer_t *peer,
                                      __u64 incarnation);
int  kibnal_conn_rts(kib_conn_t *conn,
                     __u32 qpn, __u8 resp_res, __u8 init_depth, __u32 psn);
kib_conn_t *kibnal_create_conn (lnet_nid_t nid, int proto_version);
void kibnal_destroy_conn (kib_conn_t *conn);
void kibnal_listen_callback(IB_HANDLE cep, CM_CONN_INFO *info, void *arg);
int  kibnal_alloc_pages (kib_pages_t **pp, int npages);
void kibnal_free_pages (kib_pages_t *p);
void kibnal_queue_tx (kib_tx_t *tx, kib_conn_t *conn);
void kibnal_txlist_done (struct list_head *txlist, int status);
int  kibnal_post_receives (kib_conn_t *conn);
int  kibnal_init_rdma (kib_tx_t *tx, int type, int nob,
                       kib_rdma_desc_t *dstrd, __u64 dstcookie);
void kibnal_check_sends (kib_conn_t *conn);
void kibnal_close_conn_locked (kib_conn_t *conn, int error);
int  kibnal_thread_start (int (*fn)(void *arg), void *arg);
int  kibnal_scheduler(void *arg);
int  kibnal_connd (void *arg);
void kibnal_init_tx_msg (kib_tx_t *tx, int type, int body_nob);
void kibnal_close_conn (kib_conn_t *conn, int why);
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,
                               unsigned int offset, unsigned int nob);
void kibnal_hca_async_callback (void *hca_arg, IB_EVENT_RECORD *ev);
void kibnal_hca_callback (void *hca_arg, void *cq_arg);
int  kibnal_tunables_init (void);
void kibnal_tunables_fini (void);