framework.c

lustre 1.6.5 source code

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
 *   Authors: Isaac Huang <[e-mail]>
 *            Liang Zhen  <[e-mail]>
 */

#define DEBUG_SUBSYSTEM S_LNET

#include "selftest.h"

int brw_inject_errors = 0;
CFS_MODULE_PARM(brw_inject_errors, "i", int, 0644,
                "# data errors to inject randomly, zero by default");

static int session_timeout = 100;
CFS_MODULE_PARM(session_timeout, "i", int, 0444,
                "test session timeout in seconds (100 by default, 0 == never)");

#define SFW_TEST_CONCURRENCY     128
#define SFW_TEST_RPC_TIMEOUT     64
#define SFW_CLIENT_RPC_TIMEOUT   64  /* in seconds */
#define SFW_EXTRA_TEST_BUFFERS   8 /* tolerate buggy peers with extra buffers */

#define sfw_test_buffers(tsi)    ((tsi)->tsi_loop + SFW_EXTRA_TEST_BUFFERS)

#define sfw_unpack_id(id)               \
do {                                    \
        __swab64s(&(id).nid);           \
        __swab32s(&(id).pid);           \
} while (0)

#define sfw_unpack_sid(sid)             \
do {                                    \
        __swab64s(&(sid).ses_nid);      \
        __swab64s(&(sid).ses_stamp);    \
} while (0)

#define sfw_unpack_fw_counters(fc)        \
do {                                      \
        __swab32s(&(fc).brw_errors);      \
        __swab32s(&(fc).ping_errors);     \
        __swab32s(&(fc).active_tests);    \
        __swab32s(&(fc).active_batches);  \
        __swab32s(&(fc).zombie_sessions); \
} while (0)

#define sfw_unpack_rpc_counters(rc)     \
do {                                    \
        __swab32s(&(rc).errors);        \
        __swab32s(&(rc).rpcs_sent);     \
        __swab32s(&(rc).rpcs_rcvd);     \
        __swab32s(&(rc).rpcs_dropped);  \
        __swab32s(&(rc).rpcs_expired);  \
        __swab64s(&(rc).bulk_get);      \
        __swab64s(&(rc).bulk_put);      \
} while (0)

#define sfw_unpack_lnet_counters(lc)    \
do {                                    \
        __swab32s(&(lc).errors);        \
        __swab32s(&(lc).msgs_max);      \
        __swab32s(&(lc).msgs_alloc);    \
        __swab32s(&(lc).send_count);    \
        __swab32s(&(lc).recv_count);    \
        __swab32s(&(lc).drop_count);    \
        __swab32s(&(lc).route_count);   \
        __swab64s(&(lc).send_length);   \
        __swab64s(&(lc).recv_length);   \
        __swab64s(&(lc).drop_length);   \
        __swab64s(&(lc).route_length);  \
} while (0)

#define sfw_test_active(t)      (atomic_read(&(t)->tsi_nactive) != 0)
#define sfw_batch_active(b)     (atomic_read(&(b)->bat_nactive) != 0)

struct smoketest_framework {
        struct list_head   fw_zombie_rpcs;     /* RPCs to be recycled */
        struct list_head   fw_zombie_sessions; /* stopping sessions */
        struct list_head   fw_tests;           /* registered test cases */
        atomic_t           fw_nzombies;        /* # zombie sessions */
        spinlock_t         fw_lock;            /* serialise */
        sfw_session_t     *fw_session;         /* _the_ session */
        int                fw_shuttingdown;    /* shutdown in progress */
        srpc_server_rpc_t *fw_active_srpc;     /* running RPC */
} sfw_data;

/* forward ref's */
int sfw_stop_batch (sfw_batch_t *tsb, int force);
void sfw_destroy_session (sfw_session_t *sn);

static inline sfw_test_case_t *
sfw_find_test_case(int id)
{
        sfw_test_case_t *tsc;

        LASSERT (id <= SRPC_SERVICE_MAX_ID);
        LASSERT (id > SRPC_FRAMEWORK_SERVICE_MAX_ID);

        list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
                if (tsc->tsc_srv_service->sv_id == id)
                        return tsc;
        }

        return NULL;
}

static int
sfw_register_test (srpc_service_t *service, sfw_test_client_ops_t *cliops)
{
        sfw_test_case_t *tsc;

        if (sfw_find_test_case(service->sv_id) != NULL) {
                CERROR ("Failed to register test %s (%d)\n",
                        service->sv_name, service->sv_id);
                return -EEXIST;
        }

        LIBCFS_ALLOC(tsc, sizeof(sfw_test_case_t));
        if (tsc == NULL)
                return -ENOMEM;

        memset(tsc, 0, sizeof(sfw_test_case_t));
        tsc->tsc_cli_ops     = cliops;
        tsc->tsc_srv_service = service;

        list_add_tail(&tsc->tsc_list, &sfw_data.fw_tests);
        return 0;
}

void
sfw_add_session_timer (void)
{
        sfw_session_t *sn = sfw_data.fw_session;
        stt_timer_t   *timer = &sn->sn_timer;

        LASSERT (!sfw_data.fw_shuttingdown);

        if (sn == NULL || sn->sn_timeout == 0)
                return;

        LASSERT (!sn->sn_timer_active);

        sn->sn_timer_active = 1;
        timer->stt_expires = cfs_time_add(sn->sn_timeout,
                                          cfs_time_current_sec());
        stt_add_timer(timer);
        return;
}

int
sfw_del_session_timer (void)
{
        sfw_session_t *sn = sfw_data.fw_session;

        if (sn == NULL || !sn->sn_timer_active)
                return 0;

        LASSERT (sn->sn_timeout != 0);

        if (stt_del_timer(&sn->sn_timer)) { /* timer defused */
                sn->sn_timer_active = 0;
                return 0;
        }

#ifndef __KERNEL__
        /* Racing is impossible in single-threaded userland selftest */
        LBUG();
#endif
        return EBUSY; /* racing with sfw_session_expired() */
}

/* called with sfw_data.fw_lock held */
static void
sfw_deactivate_session (void)
{
        sfw_session_t *sn = sfw_data.fw_session;
        int            nactive = 0;
        sfw_batch_t   *tsb;

        if (sn == NULL) return;

        LASSERT (!sn->sn_timer_active);

        sfw_data.fw_session = NULL;
        atomic_inc(&sfw_data.fw_nzombies);
        list_add(&sn->sn_list, &sfw_data.fw_zombie_sessions);

        list_for_each_entry (tsb, &sn->sn_batches, bat_list) {
                if (sfw_batch_active(tsb)) {
                        nactive++;
                        sfw_stop_batch(tsb, 1);
                }
        }

        if (nactive != 0)
                return;   /* wait for active batches to stop */

        list_del_init(&sn->sn_list);
        spin_unlock(&sfw_data.fw_lock);

        sfw_destroy_session(sn);

        spin_lock(&sfw_data.fw_lock);
        return;
}

#ifndef __KERNEL__

int
sfw_session_removed (void)
{
        return (sfw_data.fw_session == NULL) ? 1 : 0;
}

#endif

void
sfw_session_expired (void *data)
{
        sfw_session_t *sn = data;

        spin_lock(&sfw_data.fw_lock);

        LASSERT (sn->sn_timer_active);
        LASSERT (sn == sfw_data.fw_session);

        CWARN ("Session expired! sid: %s-"LPU64", name: %s\n",
               libcfs_nid2str(sn->sn_id.ses_nid),
               sn->sn_id.ses_stamp, &sn->sn_name[0]);

        sn->sn_timer_active = 0;
        sfw_deactivate_session();

        spin_unlock(&sfw_data.fw_lock);
        return;
}

static inline void
sfw_init_session (sfw_session_t *sn, lst_sid_t sid, const char *name)
{
        stt_timer_t *timer = &sn->sn_timer;

        memset(sn, 0, sizeof(sfw_session_t));
        CFS_INIT_LIST_HEAD(&sn->sn_list);
        CFS_INIT_LIST_HEAD(&sn->sn_batches);
        atomic_set(&sn->sn_brw_errors, 0);
        atomic_set(&sn->sn_ping_errors, 0);
        strncpy(&sn->sn_name[0], name, LST_NAME_SIZE);

        sn->sn_timer_active = 0;
        sn->sn_id           = sid;
        sn->sn_timeout      = session_timeout;

        timer->stt_data = sn;
        timer->stt_func = sfw_session_expired;
        CFS_INIT_LIST_HEAD(&timer->stt_list);
}

/* completion handler for incoming framework RPCs */
void
sfw_server_rpc_done (srpc_server_rpc_t *rpc)
{
        srpc_service_t *sv = rpc->srpc_service;
        int             status = rpc->srpc_status;

        CDEBUG (D_NET,
                "Incoming framework RPC done: "
                "service %s, peer %s, status %s:%d\n",
                sv->sv_name, libcfs_id2str(rpc->srpc_peer),
                swi_state2str(rpc->srpc_wi.wi_state),
                status);

        if (rpc->srpc_bulk != NULL)
                sfw_free_pages(rpc);
        return;
}

void
sfw_client_rpc_fini (srpc_client_rpc_t *rpc)
{
        LASSERT (rpc->crpc_bulk.bk_niov == 0);
        LASSERT (list_empty(&rpc->crpc_list));
        LASSERT (atomic_read(&rpc->crpc_refcount) == 0);
#ifndef __KERNEL__
        LASSERT (rpc->crpc_bulk.bk_pages == NULL);
#endif

        CDEBUG (D_NET,
                "Outgoing framework RPC done: "
                "service %d, peer %s, status %s:%d:%d\n",
                rpc->crpc_service, libcfs_id2str(rpc->crpc_dest),
                swi_state2str(rpc->crpc_wi.wi_state),
                rpc->crpc_aborted, rpc->crpc_status);

        spin_lock(&sfw_data.fw_lock);

        /* my callers must finish all RPCs before shutting me down */
        LASSERT (!sfw_data.fw_shuttingdown);
        list_add(&rpc->crpc_list, &sfw_data.fw_zombie_rpcs);

        spin_unlock(&sfw_data.fw_lock);
        return;
}

sfw_batch_t *
sfw_find_batch (lst_bid_t bid)
{
        sfw_session_t *sn = sfw_data.fw_session;
        sfw_batch_t   *bat;

        LASSERT (sn != NULL);

        list_for_each_entry (bat, &sn->sn_batches, bat_list) {
                if (bat->bat_id.bat_id == bid.bat_id)
                        return bat;
        }

        return NULL;
}

sfw_batch_t *
sfw_bid2batch (lst_bid_t bid)
{
        sfw_session_t *sn = sfw_data.fw_session;
        sfw_batch_t   *bat;

        LASSERT (sn != NULL);

        bat = sfw_find_batch(bid);
        if (bat != NULL)
                return bat;

        LIBCFS_ALLOC(bat, sizeof(sfw_batch_t));
        if (bat == NULL) 
                return NULL;

        bat->bat_error    = 0;
        bat->bat_session  = sn;
        bat->bat_id       = bid;
        atomic_set(&bat->bat_nactive, 0);
        CFS_INIT_LIST_HEAD(&bat->bat_tests);

        list_add_tail(&bat->bat_list, &sn->sn_batches);
        return bat;
}

int
sfw_get_stats (srpc_stat_reqst_t *request, srpc_stat_reply_t *reply)
{
        sfw_session_t  *sn = sfw_data.fw_session;
        sfw_counters_t *cnt = &reply->str_fw;
        sfw_batch_t    *bat;

        reply->str_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;

        if (request->str_sid.ses_nid == LNET_NID_ANY) {
                reply->str_status = EINVAL;
                return 0;
        }

        if (sn == NULL || !sfw_sid_equal(request->str_sid, sn->sn_id)) {
                reply->str_status = ESRCH;
                return 0;
        }

        LNET_LOCK();
        reply->str_lnet = the_lnet.ln_counters;
        LNET_UNLOCK();

        srpc_get_counters(&reply->str_rpc);

        cnt->brw_errors      = atomic_read(&sn->sn_brw_errors);
        cnt->ping_errors     = atomic_read(&sn->sn_ping_errors);
        cnt->zombie_sessions = atomic_read(&sfw_data.fw_nzombies);

        cnt->active_tests = cnt->active_batches = 0;
        list_for_each_entry (bat, &sn->sn_batches, bat_list) {
                int n = atomic_read(&bat->bat_nactive);

                if (n > 0) {
                        cnt->active_batches++;
                        cnt->active_tests += n;
                }
        }

        reply->str_status = 0;
        return 0;
}

int
sfw_make_session (srpc_mksn_reqst_t *request, srpc_mksn_reply_t *reply)
{
        sfw_session_t *sn = sfw_data.fw_session;

        if (request->mksn_sid.ses_nid == LNET_NID_ANY) {
                reply->mksn_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
                reply->mksn_status = EINVAL;
                return 0;
        }

        if (sn != NULL && !request->mksn_force) {
                reply->mksn_sid    = sn->sn_id;
                reply->mksn_status = EBUSY;
                strncpy(&reply->mksn_name[0], &sn->sn_name[0], LST_NAME_SIZE);
                return 0;
        }
        
        LIBCFS_ALLOC(sn, sizeof(sfw_session_t));
        if (sn == NULL) {
                CERROR ("Dropping RPC (mksn) under memory pressure.\n");
                return -ENOMEM;
        }

        sfw_init_session(sn, request->mksn_sid, &request->mksn_name[0]);