ptllnd.h

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

enum 
{
        PTLLND_INIT_NOTHING = 0,
        PTLLND_INIT_DATA,
        PTLLND_INIT_ALL,
};

extern kptl_tunables_t  kptllnd_tunables;
extern kptl_data_t      kptllnd_data;

static inline lnet_nid_t 
kptllnd_ptl2lnetnid(ptl_nid_t ptl_nid)
{
#ifdef _USING_LUSTRE_PORTALS_
        return LNET_MKNID(LNET_NIDNET(kptllnd_data.kptl_ni->ni_nid), 
                          LNET_NIDADDR(ptl_nid));
#else
	return LNET_MKNID(LNET_NIDNET(kptllnd_data.kptl_ni->ni_nid), 
                          ptl_nid);
#endif
}

static inline ptl_nid_t 
kptllnd_lnet2ptlnid(lnet_nid_t lnet_nid)
{
#ifdef _USING_LUSTRE_PORTALS_
        return LNET_MKNID(LNET_NIDNET(kptllnd_data.kptl_portals_id.nid),
                          LNET_NIDADDR(lnet_nid));
#else
	return LNET_NIDADDR(lnet_nid);
#endif
}

int  kptllnd_startup(lnet_ni_t *ni);
void kptllnd_shutdown(lnet_ni_t *ni);
int  kptllnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg);
int  kptllnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg);
int  kptllnd_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  kptllnd_eager_recv(struct lnet_ni *ni, void *private, 
                        lnet_msg_t *msg, void **new_privatep);
void kptllnd_eq_callback(ptl_event_t *evp);
int  kptllnd_scheduler(void *arg);
int  kptllnd_watchdog(void *arg);
int  kptllnd_thread_start(int (*fn)(void *arg), void *arg);
int  kptllnd_tunables_init(void);
void kptllnd_tunables_fini(void);

const char *kptllnd_evtype2str(int evtype);
const char *kptllnd_msgtype2str(int msgtype);
const char *kptllnd_errtype2str(int errtype);

static inline void *
kptllnd_eventarg2obj (kptl_eventarg_t *eva)
{
        switch (eva->eva_type) {
        default:
                LBUG();
        case PTLLND_EVENTARG_TYPE_BUF:
                return list_entry(eva, kptl_rx_buffer_t, rxb_eventarg);
        case PTLLND_EVENTARG_TYPE_RDMA:
                return list_entry(eva, kptl_tx_t, tx_rdma_eventarg);
        case PTLLND_EVENTARG_TYPE_MSG:
                return list_entry(eva, kptl_tx_t, tx_msg_eventarg);
        }
}

/*
 * RX BUFFER SUPPORT FUNCTIONS
 */
void kptllnd_rx_buffer_pool_init(kptl_rx_buffer_pool_t *rxbp);
void kptllnd_rx_buffer_pool_fini(kptl_rx_buffer_pool_t *rxbp);
int  kptllnd_rx_buffer_pool_reserve(kptl_rx_buffer_pool_t *rxbp, int count);
void kptllnd_rx_buffer_pool_unreserve(kptl_rx_buffer_pool_t *rxbp, int count);
void kptllnd_rx_buffer_callback(ptl_event_t *ev);
void kptllnd_rx_buffer_post(kptl_rx_buffer_t *rxb);

static inline int
kptllnd_rx_buffer_size(void)
{
        return PAGE_SIZE * (*kptllnd_tunables.kptl_rxb_npages);
}

static inline void
kptllnd_rx_buffer_addref(kptl_rx_buffer_t *rxb)
{
        unsigned long flags;
        
        spin_lock_irqsave(&rxb->rxb_pool->rxbp_lock, flags);
        rxb->rxb_refcount++;
        spin_unlock_irqrestore(&rxb->rxb_pool->rxbp_lock, flags);
}

static inline void
kptllnd_rx_buffer_decref_locked(kptl_rx_buffer_t *rxb)
{
        if (--(rxb->rxb_refcount) == 0) {
                spin_lock(&kptllnd_data.kptl_sched_lock);
        
                list_add_tail(&rxb->rxb_repost_list,
                              &kptllnd_data.kptl_sched_rxbq);
                wake_up(&kptllnd_data.kptl_sched_waitq);

                spin_unlock(&kptllnd_data.kptl_sched_lock);
        }
}

static inline void
kptllnd_rx_buffer_decref(kptl_rx_buffer_t *rxb)
{
        unsigned long flags;
        int           count;
        
        spin_lock_irqsave(&rxb->rxb_pool->rxbp_lock, flags);
        count = --(rxb->rxb_refcount);
        spin_unlock_irqrestore(&rxb->rxb_pool->rxbp_lock, flags);

        if (count == 0)
                kptllnd_rx_buffer_post(rxb);
}

/*
 * RX SUPPORT FUNCTIONS
 */
void kptllnd_rx_done(kptl_rx_t *rx);
void kptllnd_rx_parse(kptl_rx_t *rx);

/*
 * PEER SUPPORT FUNCTIONS
 */
int kptllnd_get_peer_info(int index,
                          lnet_process_id_t *id, 
                          int *state, int *sent_hello,
                          int *refcount, __u64 *incarnation,
                          __u64 *next_matchbits, __u64 *last_matchbits_seen,
                          int *nsendq, int *nactiveq,
                          int *credits, int *outstanding_credits);
void kptllnd_peer_destroy(kptl_peer_t *peer);
int  kptllnd_peer_del(lnet_process_id_t id);
void kptllnd_peer_close_locked(kptl_peer_t *peer, int why);
void kptllnd_peer_close(kptl_peer_t *peer, int why);
void kptllnd_handle_closing_peers(void);
int  kptllnd_peer_connect(kptl_tx_t *tx, lnet_nid_t nid);
void kptllnd_peer_check_sends(kptl_peer_t *peer);
void kptllnd_peer_check_bucket(int idx, int stamp);
void kptllnd_tx_launch(kptl_peer_t *peer, kptl_tx_t *tx, int nfrag);
int  kptllnd_find_target(kptl_peer_t **peerp, lnet_process_id_t target);
kptl_peer_t *kptllnd_peer_handle_hello(ptl_process_id_t initiator,
                                       kptl_msg_t *msg);
kptl_peer_t *kptllnd_id2peer_locked(lnet_process_id_t id);
void kptllnd_peer_alive(kptl_peer_t *peer);

static inline void
kptllnd_peer_addref (kptl_peer_t *peer)
{
        atomic_inc(&peer->peer_refcount);
}

static inline void
kptllnd_peer_decref (kptl_peer_t *peer)
{
        if (atomic_dec_and_test(&peer->peer_refcount))
                kptllnd_peer_destroy(peer);
}

static inline void
kptllnd_set_tx_peer(kptl_tx_t *tx, kptl_peer_t *peer) 
{
        LASSERT (tx->tx_peer == NULL);
        
        kptllnd_peer_addref(peer);
        tx->tx_peer = peer;
}

static inline struct list_head *
kptllnd_nid2peerlist(lnet_nid_t nid)
{
        unsigned int hash = ((unsigned int)nid) %
                            kptllnd_data.kptl_peer_hash_size;

        return &kptllnd_data.kptl_peers[hash];
}

static inline kptl_peer_t *
kptllnd_id2peer(lnet_process_id_t id)
{
        kptl_peer_t   *peer;
        unsigned long  flags;

        read_lock_irqsave(&kptllnd_data.kptl_peer_rw_lock, flags);
        peer = kptllnd_id2peer_locked(id);
        read_unlock_irqrestore(&kptllnd_data.kptl_peer_rw_lock, flags);

        return peer;
}

static inline int
kptllnd_reserve_buffers(int n)
{
        return kptllnd_rx_buffer_pool_reserve(&kptllnd_data.kptl_rx_buffer_pool,
                                              n);
}

static inline int
kptllnd_peer_reserve_buffers(void)
{
        return kptllnd_reserve_buffers(*kptllnd_tunables.kptl_peercredits);
}

static inline void
kptllnd_peer_unreserve_buffers(void)
{
        kptllnd_rx_buffer_pool_unreserve(&kptllnd_data.kptl_rx_buffer_pool,
                                         *kptllnd_tunables.kptl_peercredits);
}

/*
 * TX SUPPORT FUNCTIONS
 */
int  kptllnd_setup_tx_descs(void);
void kptllnd_cleanup_tx_descs(void);
void kptllnd_tx_fini(kptl_tx_t *tx);
kptl_tx_t *kptllnd_get_idle_tx(enum kptl_tx_type purpose);
void kptllnd_tx_callback(ptl_event_t *ev);
const char *kptllnd_tx_typestr(int type);

static inline void
kptllnd_tx_addref(kptl_tx_t *tx)
{
        atomic_inc(&tx->tx_refcount);
}

static inline void 
kptllnd_tx_decref(kptl_tx_t *tx)
{
        LASSERT (!in_interrupt());        /* Thread context only */

        if (atomic_dec_and_test(&tx->tx_refcount))
                kptllnd_tx_fini(tx);
}

/*
 * MESSAGE SUPPORT FUNCTIONS
 */
void kptllnd_init_msg(kptl_msg_t *msg, int type, int body_nob);
void kptllnd_msg_pack(kptl_msg_t *msg, kptl_peer_t *peer);
int  kptllnd_msg_unpack(kptl_msg_t *msg, int nob);

/*
 * MISC SUPPORT FUNCTIONS
 */
void kptllnd_init_rdma_md(kptl_tx_t *tx, unsigned int niov,
                          struct iovec *iov, lnet_kiov_t *kiov,
                          unsigned int offset, unsigned int nob);
char *kptllnd_ptlid2str(ptl_process_id_t id);

void kptllnd_init_ptltrace(void);
void kptllnd_dump_ptltrace(void);

#ifdef PJK_DEBUGGING
#define SIMULATION_FAIL_TX_PUT_ALLOC   0       /* 0x00000001 */
#define SIMULATION_FAIL_TX_GET_ALLOC   1       /* 0x00000002 */
#define SIMULATION_FAIL_TX             2       /* 0x00000004 */
#define SIMULATION_FAIL_RX_ALLOC       3       /* 0x00000008 */

#define IS_SIMULATION_ENABLED(x) \
        (((*kptllnd_tunables.kptl_simulation_bitmap) & 1<< SIMULATION_##x) != 0)
#else
#define IS_SIMULATION_ENABLED(x)       0
#endif