selftest.h

lustre 1.6.5 source code

        const char        *sv_name;          /* human readable name */
        int                sv_nprune;        /* # posted RPC to be pruned */
        int                sv_concur;        /* max # concurrent RPCs */

        spinlock_t         sv_lock;
        int                sv_shuttingdown;
        srpc_event_t       sv_ev;            /* LNet event */
        int                sv_nposted_msg;   /* # posted message buffers */
        struct list_head   sv_free_rpcq;     /* free RPC descriptors */
        struct list_head   sv_active_rpcq;   /* in-flight RPCs */
        struct list_head   sv_posted_msgq;   /* posted message buffers */
        struct list_head   sv_blocked_msgq;  /* blocked for RPC descriptor */

        /* Service callbacks:
         * - sv_handler: process incoming RPC request
         * - sv_bulk_ready: notify bulk data
         */
        int                (*sv_handler) (srpc_server_rpc_t *);
        int                (*sv_bulk_ready) (srpc_server_rpc_t *, int);
} srpc_service_t;

#define SFW_POST_BUFFERS         8
#define SFW_SERVICE_CONCURRENCY  (SFW_POST_BUFFERS/2)

typedef struct {
        struct list_head  sn_list;    /* chain on fw_zombie_sessions */
        lst_sid_t         sn_id;      /* unique identifier */
        unsigned int      sn_timeout; /* # seconds' inactivity to expire */
        int               sn_timer_active;
        stt_timer_t       sn_timer;
        struct list_head  sn_batches; /* list of batches */
        char              sn_name[LST_NAME_SIZE];
        atomic_t          sn_brw_errors;
        atomic_t          sn_ping_errors;
} sfw_session_t;

#define sfw_sid_equal(sid0, sid1)     ((sid0).ses_nid == (sid1).ses_nid && \
                                       (sid0).ses_stamp == (sid1).ses_stamp)

typedef struct {
        struct list_head  bat_list;      /* chain on sn_batches */
        lst_bid_t         bat_id;        /* batch id */
        int               bat_error;     /* error code of batch */
        sfw_session_t    *bat_session;   /* batch's session */
        atomic_t          bat_nactive;   /* # of active tests */
        struct list_head  bat_tests;     /* test instances */
} sfw_batch_t;

typedef struct {
        int  (*tso_init)(struct sfw_test_instance *tsi); /* intialize test client */
        void (*tso_fini)(struct sfw_test_instance *tsi); /* finalize test client */
        int  (*tso_prep_rpc)(struct sfw_test_unit *tsu,     
                             lnet_process_id_t dest,
                             srpc_client_rpc_t **rpc);   /* prep a tests rpc */
        void (*tso_done_rpc)(struct sfw_test_unit *tsu,
                             srpc_client_rpc_t *rpc);    /* done a test rpc */
} sfw_test_client_ops_t;

typedef struct sfw_test_instance {
        struct list_head        tsi_list;         /* chain on batch */
        int                     tsi_service;      /* test type */
        sfw_batch_t            *tsi_batch;        /* batch */
        sfw_test_client_ops_t  *tsi_ops;          /* test client operations */

        /* public parameter for all test units */
        int                     tsi_is_client:1;     /* is test client */
        int                     tsi_stoptsu_onerr:1; /* stop tsu on error */
        int                     tsi_concur;          /* concurrency */
        int                     tsi_loop;            /* loop count */

        /* status of test instance */
        spinlock_t              tsi_lock;         /* serialize */
        int                     tsi_stopping:1;   /* test is stopping */
        atomic_t                tsi_nactive;      /* # of active test unit */
        struct list_head        tsi_units;        /* test units */
        struct list_head        tsi_free_rpcs;    /* free rpcs */
        struct list_head        tsi_active_rpcs;  /* active rpcs */

        union {
                test_bulk_req_t bulk;             /* bulk parameter */
                test_ping_req_t ping;             /* ping parameter */
        } tsi_u;
} sfw_test_instance_t;

/* XXX: trailing (CFS_PAGE_SIZE % sizeof(lnet_process_id_t)) bytes at 
 * the end of pages are not used */
#define SFW_MAX_CONCUR     LST_MAX_CONCUR
#define SFW_ID_PER_PAGE    (CFS_PAGE_SIZE / sizeof(lnet_process_id_t))
#define SFW_MAX_NDESTS     (LNET_MAX_IOV * SFW_ID_PER_PAGE)
#define sfw_id_pages(n)    (((n) + SFW_ID_PER_PAGE - 1) / SFW_ID_PER_PAGE)

typedef struct sfw_test_unit {
        struct list_head        tsu_list;         /* chain on lst_test_instance */
        lnet_process_id_t       tsu_dest;         /* id of dest node */
        int                     tsu_loop;         /* loop count of the test */
        sfw_test_instance_t    *tsu_instance;     /* pointer to test instance */
        void                   *tsu_private;      /* private data */
        swi_workitem_t          tsu_worker;       /* workitem of the test unit */
} sfw_test_unit_t;

typedef struct {
        struct list_head        tsc_list;         /* chain on fw_tests */
        srpc_service_t         *tsc_srv_service;  /* test service */
        sfw_test_client_ops_t  *tsc_cli_ops;      /* ops of test client */
} sfw_test_case_t;


srpc_client_rpc_t *
sfw_create_rpc(lnet_process_id_t peer, int service, int nbulkiov, int bulklen,
               void (*done) (srpc_client_rpc_t *), void *priv);
int sfw_create_test_rpc(sfw_test_unit_t *tsu, lnet_process_id_t peer,
                        int nblk, int blklen, srpc_client_rpc_t **rpc);
void sfw_abort_rpc(srpc_client_rpc_t *rpc);
void sfw_post_rpc(srpc_client_rpc_t *rpc);
void sfw_client_rpc_done(srpc_client_rpc_t *rpc);
void sfw_unpack_message(srpc_msg_t *msg);
void sfw_free_pages(srpc_server_rpc_t *rpc);
void sfw_add_bulk_page(srpc_bulk_t *bk, cfs_page_t *pg, int i);
int sfw_alloc_pages(srpc_server_rpc_t *rpc, int npages, int sink);

srpc_client_rpc_t *
srpc_create_client_rpc(lnet_process_id_t peer, int service, 
                       int nbulkiov, int bulklen,
                       void (*rpc_done)(srpc_client_rpc_t *),
                       void (*rpc_fini)(srpc_client_rpc_t *), void *priv);
void srpc_post_rpc(srpc_client_rpc_t *rpc);
void srpc_abort_rpc(srpc_client_rpc_t *rpc, int why);
void srpc_free_bulk(srpc_bulk_t *bk);
srpc_bulk_t *srpc_alloc_bulk(int npages, int sink);
int srpc_send_rpc(swi_workitem_t *wi);
int srpc_send_reply(srpc_server_rpc_t *rpc);
int srpc_add_service(srpc_service_t *sv);
int srpc_remove_service(srpc_service_t *sv);
void srpc_shutdown_service(srpc_service_t *sv);
int srpc_finish_service(srpc_service_t *sv);
int srpc_service_add_buffers(srpc_service_t *sv, int nbuffer);
void srpc_service_remove_buffers(srpc_service_t *sv, int nbuffer);
void srpc_get_counters(srpc_counters_t *cnt);
void srpc_set_counters(const srpc_counters_t *cnt);

void swi_kill_workitem(swi_workitem_t *wi);
void swi_schedule_workitem(swi_workitem_t *wi);
void swi_schedule_serial_workitem(swi_workitem_t *wi);
int swi_startup(void);
int sfw_startup(void);
int srpc_startup(void);
void swi_shutdown(void);
void sfw_shutdown(void);
void srpc_shutdown(void);

static inline void
srpc_destroy_client_rpc (srpc_client_rpc_t *rpc)
{
        LASSERT (rpc != NULL);
        LASSERT (!srpc_event_pending(rpc));
        LASSERT (list_empty(&rpc->crpc_privl));
        LASSERT (atomic_read(&rpc->crpc_refcount) == 0);
#ifndef __KERNEL__
        LASSERT (rpc->crpc_bulk.bk_pages == NULL);
#endif

        if (rpc->crpc_fini == NULL) {
                LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
        } else {
                (*rpc->crpc_fini) (rpc);
        }

        return;
}

static inline void
srpc_init_client_rpc (srpc_client_rpc_t *rpc, lnet_process_id_t peer,
                      int service, int nbulkiov, int bulklen,
                      void (*rpc_done)(srpc_client_rpc_t *),
                      void (*rpc_fini)(srpc_client_rpc_t *), void *priv)
{
        LASSERT (nbulkiov <= LNET_MAX_IOV);

        memset(rpc, 0, offsetof(srpc_client_rpc_t,
                                crpc_bulk.bk_iovs[nbulkiov]));

        CFS_INIT_LIST_HEAD(&rpc->crpc_list);
        CFS_INIT_LIST_HEAD(&rpc->crpc_privl);
        swi_init_workitem(&rpc->crpc_wi, rpc, srpc_send_rpc);
        spin_lock_init(&rpc->crpc_lock);
        atomic_set(&rpc->crpc_refcount, 1); /* 1 ref for caller */

        rpc->crpc_dest         = peer;
        rpc->crpc_priv         = priv;
        rpc->crpc_service      = service;
        rpc->crpc_bulk.bk_len  = bulklen;
        rpc->crpc_bulk.bk_niov = nbulkiov;
        rpc->crpc_done         = rpc_done;
        rpc->crpc_fini         = rpc_fini;
        rpc->crpc_reqstmdh     =
        rpc->crpc_replymdh     =
        rpc->crpc_bulk.bk_mdh  = LNET_INVALID_HANDLE;

        /* no event is expected at this point */
        rpc->crpc_bulkev.ev_fired  =
        rpc->crpc_reqstev.ev_fired =
        rpc->crpc_replyev.ev_fired = 1;

        rpc->crpc_reqstmsg.msg_magic   = SRPC_MSG_MAGIC;
        rpc->crpc_reqstmsg.msg_version = SRPC_MSG_VERSION;
        rpc->crpc_reqstmsg.msg_type    = srpc_service2request(service);
        return;
}

static inline const char * 
swi_state2str (int state)
{
#define STATE2STR(x) case x: return #x
        switch(state) {
                default: 
                        LBUG();
                STATE2STR(SWI_STATE_NEWBORN);
                STATE2STR(SWI_STATE_REPLY_SUBMITTED);
                STATE2STR(SWI_STATE_REPLY_SENT);
                STATE2STR(SWI_STATE_REQUEST_SUBMITTED);
                STATE2STR(SWI_STATE_REQUEST_SENT);
                STATE2STR(SWI_STATE_REPLY_RECEIVED);
                STATE2STR(SWI_STATE_BULK_STARTED);
                STATE2STR(SWI_STATE_DONE);
        }
#undef STATE2STR
}

#define UNUSED(x)       ( (void)(x) )

#ifndef __KERNEL__

int stt_poll_interval(void);
int sfw_session_removed(void);

int stt_check_events(void);
int swi_check_events(void);
int srpc_check_event(int timeout);

int lnet_selftest_init(void);
void lnet_selftest_fini(void);
int selftest_wait_events(void);

#else

#define selftest_wait_events()    cfs_pause(cfs_time_seconds(1))

#endif

#define lst_wait_until(cond, lock, fmt, a...)                           \
do {                                                                    \
        int __I = 2;                                                    \
        while (!(cond)) {                                               \
                __I++;                                                  \
                CDEBUG(((__I & (-__I)) == __I) ? D_WARNING :            \
                                                 D_NET,     /* 2**n? */ \
                       fmt, ## a);                                      \
                spin_unlock(&(lock));                                   \
                                                                        \
                selftest_wait_events();                                 \
                                                                        \
                spin_lock(&(lock));                                     \
        }                                                               \
} while (0)

static inline void
srpc_wait_service_shutdown (srpc_service_t *sv)
{
        int i = 2;

        spin_lock(&sv->sv_lock);
        LASSERT (sv->sv_shuttingdown);
        spin_unlock(&sv->sv_lock);

        while (srpc_finish_service(sv) == 0) {
                i++;
                CDEBUG (((i & -i) == i) ? D_WARNING : D_NET,
                        "Waiting for %s service to shutdown...\n",
                        sv->sv_name);
                selftest_wait_events();
        }
}

#endif /* __SELFTEST_SELFTEST_H__ */