framework.c

lustre 1.6.5 source code

        sfw_add_session_timer();
#endif

        sfw_data.fw_active_srpc = NULL;
        spin_unlock(&sfw_data.fw_lock);
        return rc;
}

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)
{
        srpc_client_rpc_t *rpc;

        spin_lock(&sfw_data.fw_lock);

        LASSERT (!sfw_data.fw_shuttingdown);
        LASSERT (service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);

        if (nbulkiov == 0 && !list_empty(&sfw_data.fw_zombie_rpcs)) {
                rpc = list_entry(sfw_data.fw_zombie_rpcs.next,
                                 srpc_client_rpc_t, crpc_list);
                list_del(&rpc->crpc_list);
                spin_unlock(&sfw_data.fw_lock);

                srpc_init_client_rpc(rpc, peer, service, 0, 0,
                                     done, sfw_client_rpc_fini, priv);
                return rpc;
        }

        spin_unlock(&sfw_data.fw_lock);

        rpc = srpc_create_client_rpc(peer, service, nbulkiov, bulklen, done,
                                     nbulkiov != 0 ? NULL : sfw_client_rpc_fini,
                                     priv);
        return rpc;
}

void
sfw_unpack_message (srpc_msg_t *msg)
{
        if (msg->msg_magic == SRPC_MSG_MAGIC)
                return; /* no flipping needed */

        LASSERT (msg->msg_magic == __swab32(SRPC_MSG_MAGIC));

        __swab32s(&msg->msg_type);

        if (msg->msg_type == SRPC_MSG_STAT_REQST) {
                srpc_stat_reqst_t *req = &msg->msg_body.stat_reqst;

                __swab32s(&req->str_type);
                __swab64s(&req->str_rpyid);
                sfw_unpack_sid(req->str_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_STAT_REPLY) {
                srpc_stat_reply_t *rep = &msg->msg_body.stat_reply;

                __swab32s(&rep->str_status);
                sfw_unpack_sid(rep->str_sid);
                sfw_unpack_fw_counters(rep->str_fw);
                sfw_unpack_rpc_counters(rep->str_rpc);
                sfw_unpack_lnet_counters(rep->str_lnet);
                return;
        }

        if (msg->msg_type == SRPC_MSG_MKSN_REQST) {
                srpc_mksn_reqst_t *req = &msg->msg_body.mksn_reqst;

                __swab64s(&req->mksn_rpyid);
                __swab32s(&req->mksn_force);
                sfw_unpack_sid(req->mksn_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_MKSN_REPLY) {
                srpc_mksn_reply_t *rep = &msg->msg_body.mksn_reply;

                __swab32s(&rep->mksn_status);
                __swab32s(&rep->mksn_timeout);
                sfw_unpack_sid(rep->mksn_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_RMSN_REQST) {
                srpc_rmsn_reqst_t *req = &msg->msg_body.rmsn_reqst;

                __swab64s(&req->rmsn_rpyid);
                sfw_unpack_sid(req->rmsn_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_RMSN_REPLY) {
                srpc_rmsn_reply_t *rep = &msg->msg_body.rmsn_reply;

                __swab32s(&rep->rmsn_status);
                sfw_unpack_sid(rep->rmsn_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_DEBUG_REQST) {
                srpc_debug_reqst_t *req = &msg->msg_body.dbg_reqst;

                __swab64s(&req->dbg_rpyid);
                __swab32s(&req->dbg_flags);
                sfw_unpack_sid(req->dbg_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_DEBUG_REPLY) {
                srpc_debug_reply_t *rep = &msg->msg_body.dbg_reply;

                __swab32s(&rep->dbg_nbatch);
                __swab32s(&rep->dbg_timeout);
                sfw_unpack_sid(rep->dbg_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_BATCH_REQST) {
                srpc_batch_reqst_t *req = &msg->msg_body.bat_reqst;

                __swab32s(&req->bar_opc);
                __swab64s(&req->bar_rpyid);
                __swab32s(&req->bar_testidx);
                __swab32s(&req->bar_arg);
                sfw_unpack_sid(req->bar_sid);
                __swab64s(&req->bar_bid.bat_id);
                return;
        }

        if (msg->msg_type == SRPC_MSG_BATCH_REPLY) {
                srpc_batch_reply_t *rep = &msg->msg_body.bat_reply;

                __swab32s(&rep->bar_status);
                sfw_unpack_sid(rep->bar_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_TEST_REQST) {
                srpc_test_reqst_t *req = &msg->msg_body.tes_reqst;

                __swab64s(&req->tsr_rpyid);
                __swab64s(&req->tsr_bulkid);
                __swab32s(&req->tsr_loop);
                __swab32s(&req->tsr_ndest);
                __swab32s(&req->tsr_concur);
                __swab32s(&req->tsr_service);
                sfw_unpack_sid(req->tsr_sid);
                __swab64s(&req->tsr_bid.bat_id);
                return;
        }

        if (msg->msg_type == SRPC_MSG_TEST_REPLY) {
                srpc_test_reply_t *rep = &msg->msg_body.tes_reply;

                __swab32s(&rep->tsr_status);
                sfw_unpack_sid(rep->tsr_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_JOIN_REQST) {
                srpc_join_reqst_t *req = &msg->msg_body.join_reqst;

                __swab64s(&req->join_rpyid);
                sfw_unpack_sid(req->join_sid);
                return;
        }

        if (msg->msg_type == SRPC_MSG_JOIN_REPLY) {
                srpc_join_reply_t *rep = &msg->msg_body.join_reply;

                __swab32s(&rep->join_status);
                __swab32s(&rep->join_timeout);
                sfw_unpack_sid(rep->join_sid);
                return;
        }

        LBUG ();
        return;
}

void
sfw_abort_rpc (srpc_client_rpc_t *rpc)
{
        LASSERT (atomic_read(&rpc->crpc_refcount) > 0);
        LASSERT (rpc->crpc_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);

        spin_lock(&rpc->crpc_lock);
        srpc_abort_rpc(rpc, -EINTR);
        spin_unlock(&rpc->crpc_lock);
        return;
}

void
sfw_post_rpc (srpc_client_rpc_t *rpc)
{
        spin_lock(&rpc->crpc_lock);

        LASSERT (!rpc->crpc_closed);
        LASSERT (!rpc->crpc_aborted);
        LASSERT (list_empty(&rpc->crpc_list));
        LASSERT (!sfw_data.fw_shuttingdown);

        rpc->crpc_timeout = SFW_CLIENT_RPC_TIMEOUT;
        srpc_post_rpc(rpc);

        spin_unlock(&rpc->crpc_lock);
        return;
}

static srpc_service_t sfw_services[] = 
{
        {
                .sv_name = "debug",
                .sv_id   = SRPC_SERVICE_DEBUG,
        },
        {
                .sv_name = "query stats",
                .sv_id   = SRPC_SERVICE_QUERY_STAT,
        },
        {
                .sv_name = "make sessin",
                .sv_id   = SRPC_SERVICE_MAKE_SESSION,
        },
        {
                .sv_name = "remove session",
                .sv_id   = SRPC_SERVICE_REMOVE_SESSION,
        },
        {
                .sv_name = "batch service",
                .sv_id   = SRPC_SERVICE_BATCH,
        },
        {
                .sv_name = "test service",
                .sv_id   = SRPC_SERVICE_TEST,
        },
        {       .sv_name = NULL, }
};

extern sfw_test_client_ops_t ping_test_client;
extern srpc_service_t        ping_test_service;

extern sfw_test_client_ops_t brw_test_client;
extern srpc_service_t        brw_test_service;

int
sfw_startup (void)
{
        int              i;
        int              rc;
        int              error;
        srpc_service_t  *sv;
        sfw_test_case_t *tsc;

#ifndef __KERNEL__
        char *s;

        s = getenv("SESSION_TIMEOUT");
        session_timeout = s != NULL ? atoi(s) : session_timeout;

        s = getenv("BRW_INJECT_ERRORS");
        brw_inject_errors = s != NULL ? atoi(s) : brw_inject_errors;
#endif

        if (session_timeout < 0) {
                CERROR ("Session timeout must be non-negative: %d\n",
                        session_timeout);
                return -EINVAL;
        }

        if (session_timeout == 0)
                CWARN ("Zero session_timeout specified "
                       "- test sessions never expire.\n");

        memset(&sfw_data, 0, sizeof(struct smoketest_framework));

        sfw_data.fw_session     = NULL;
        sfw_data.fw_active_srpc = NULL;
        spin_lock_init(&sfw_data.fw_lock);
        atomic_set(&sfw_data.fw_nzombies, 0);
        CFS_INIT_LIST_HEAD(&sfw_data.fw_tests);
        CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_rpcs);
        CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_sessions);

        rc = sfw_register_test(&brw_test_service, &brw_test_client);
        LASSERT (rc == 0);
        rc = sfw_register_test(&ping_test_service, &ping_test_client);
        LASSERT (rc == 0);

        error = 0;
        list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
                sv = tsc->tsc_srv_service;
                sv->sv_concur = SFW_TEST_CONCURRENCY;

                rc = srpc_add_service(sv);
                LASSERT (rc != -EBUSY);
                if (rc != 0) {
                        CWARN ("Failed to add %s service: %d\n",
                               sv->sv_name, rc);
                        error = rc;
                }
        }

        for (i = 0; ; i++) {
                sv = &sfw_services[i];
                if (sv->sv_name == NULL) break;

                sv->sv_bulk_ready = NULL;
                sv->sv_handler    = sfw_handle_server_rpc;
                sv->sv_concur     = SFW_SERVICE_CONCURRENCY;
                if (sv->sv_id == SRPC_SERVICE_TEST)
                        sv->sv_bulk_ready = sfw_bulk_ready;

                rc = srpc_add_service(sv);
                LASSERT (rc != -EBUSY);
                if (rc != 0) {
                        CWARN ("Failed to add %s service: %d\n",
                               sv->sv_name, rc);
                        error = rc;
                }

                /* about to sfw_shutdown, no need to add buffer */
                if (error) continue; 

                rc = srpc_service_add_buffers(sv, SFW_POST_BUFFERS);
                if (rc != SFW_POST_BUFFERS) {
                        CWARN ("Failed to reserve enough buffers: "
                               "service %s, %d needed, %d reserved\n",
                               sv->sv_name, SFW_POST_BUFFERS, rc);
                        error = -ENOMEM;
                }
        }

        if (error != 0)
                sfw_shutdown();
        return error;
}

void
sfw_shutdown (void)
{
        srpc_service_t  *sv;
        sfw_test_case_t *tsc;
        int              i;

        spin_lock(&sfw_data.fw_lock);

        sfw_data.fw_shuttingdown = 1;
#ifdef __KERNEL__
        lst_wait_until(sfw_data.fw_active_srpc == NULL, sfw_data.fw_lock,
                       "waiting for active RPC to finish.\n");
#else
        LASSERT (sfw_data.fw_active_srpc == NULL);
#endif

        if (sfw_del_session_timer() != 0)
                lst_wait_until(sfw_data.fw_session == NULL, sfw_data.fw_lock,
                               "waiting for session timer to explode.\n");

        sfw_deactivate_session();
        lst_wait_until(atomic_read(&sfw_data.fw_nzombies) == 0,
                       sfw_data.fw_lock,
                       "waiting for %d zombie sessions to die.\n",
                       atomic_read(&sfw_data.fw_nzombies));

        spin_unlock(&sfw_data.fw_lock);

        for (i = 0; ; i++) {
                sv = &sfw_services[i];
                if (sv->sv_name == NULL)
                        break;

                srpc_shutdown_service(sv);
                srpc_remove_service(sv);
        }

        list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
                sv = tsc->tsc_srv_service;
                srpc_shutdown_service(sv);
                srpc_remove_service(sv);
        }

        while (!list_empty(&sfw_data.fw_zombie_rpcs)) {
                srpc_client_rpc_t *rpc;

                rpc = list_entry(sfw_data.fw_zombie_rpcs.next, 
                                 srpc_client_rpc_t, crpc_list);
                list_del(&rpc->crpc_list);

                LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
        }

        for (i = 0; ; i++) {
                sv = &sfw_services[i];
                if (sv->sv_name == NULL)
                        break;

                srpc_wait_service_shutdown(sv);
        }

        while (!list_empty(&sfw_data.fw_tests)) {
                tsc = list_entry(sfw_data.fw_tests.next,
                                 sfw_test_case_t, tsc_list);
                
                srpc_wait_service_shutdown(tsc->tsc_srv_service);

                list_del(&tsc->tsc_list);
                LIBCFS_FREE(tsc, sizeof(*tsc));
        }

        return;
}