framework.c

lustre 1.6.5 source code

        spin_lock(&sfw_data.fw_lock);

        sfw_deactivate_session();
        LASSERT (sfw_data.fw_session == NULL);
        sfw_data.fw_session = sn;

        spin_unlock(&sfw_data.fw_lock);

        reply->mksn_status  = 0;
        reply->mksn_sid     = sn->sn_id;
        reply->mksn_timeout = sn->sn_timeout;
        return 0;
}

int
sfw_remove_session (srpc_rmsn_reqst_t *request, srpc_rmsn_reply_t *reply)
{
        sfw_session_t *sn = sfw_data.fw_session;

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

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

        if (sn == NULL || !sfw_sid_equal(request->rmsn_sid, sn->sn_id)) {
                reply->rmsn_status = (sn == NULL) ? ESRCH : EBUSY;
                return 0;
        }

        spin_lock(&sfw_data.fw_lock);
        sfw_deactivate_session();
        spin_unlock(&sfw_data.fw_lock);

        reply->rmsn_status = 0;
        reply->rmsn_sid    = LST_INVALID_SID;
        LASSERT (sfw_data.fw_session == NULL);
        return 0;
}

int
sfw_debug_session (srpc_debug_reqst_t *request, srpc_debug_reply_t *reply)
{
        sfw_session_t *sn = sfw_data.fw_session;

        if (sn == NULL) {
                reply->dbg_status = ESRCH;
                reply->dbg_sid    = LST_INVALID_SID;
                return 0;
        } 

        reply->dbg_status  = 0;
        reply->dbg_sid     = sn->sn_id;      
        reply->dbg_timeout = sn->sn_timeout;
        strncpy(reply->dbg_name, &sn->sn_name[0], LST_NAME_SIZE);

        return 0;
}

void
sfw_test_rpc_fini (srpc_client_rpc_t *rpc)
{
        sfw_test_unit_t     *tsu = rpc->crpc_priv;
        sfw_test_instance_t *tsi = tsu->tsu_instance;

        /* Called with hold of tsi->tsi_lock */
        LASSERT (list_empty(&rpc->crpc_list));
        list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
}

int
sfw_load_test (sfw_test_instance_t *tsi)
{
        sfw_test_case_t *tsc = sfw_find_test_case(tsi->tsi_service);
        int              nrequired = sfw_test_buffers(tsi);
        int              nposted;

        LASSERT (tsc != NULL);

        if (tsi->tsi_is_client) {
                tsi->tsi_ops = tsc->tsc_cli_ops;
                return 0;
        }

        nposted = srpc_service_add_buffers(tsc->tsc_srv_service, nrequired);
        if (nposted != nrequired) {
                CWARN ("Failed to reserve enough buffers: "
                       "service %s, %d needed, %d reserved\n",
                       tsc->tsc_srv_service->sv_name, nrequired, nposted);
                srpc_service_remove_buffers(tsc->tsc_srv_service, nposted);
                return -ENOMEM;
        }

        CDEBUG (D_NET, "Reserved %d buffers for test %s\n",
                nposted, tsc->tsc_srv_service->sv_name);
        return 0;
}

void
sfw_unload_test (sfw_test_instance_t *tsi)
{
        sfw_test_case_t *tsc = sfw_find_test_case(tsi->tsi_service);

        LASSERT (tsc != NULL);

        if (!tsi->tsi_is_client)
                srpc_service_remove_buffers(tsc->tsc_srv_service,
                                            sfw_test_buffers(tsi));
        return;
}

void
sfw_destroy_test_instance (sfw_test_instance_t *tsi)
{
        srpc_client_rpc_t *rpc;
        sfw_test_unit_t   *tsu;

        if (!tsi->tsi_is_client) goto clean;

        tsi->tsi_ops->tso_fini(tsi);

        LASSERT (!tsi->tsi_stopping);
        LASSERT (list_empty(&tsi->tsi_active_rpcs));
        LASSERT (!sfw_test_active(tsi));

        while (!list_empty(&tsi->tsi_units)) {
                tsu = list_entry(tsi->tsi_units.next,
                                 sfw_test_unit_t, tsu_list);
                list_del(&tsu->tsu_list);
                LIBCFS_FREE(tsu, sizeof(*tsu));
        }

        while (!list_empty(&tsi->tsi_free_rpcs)) {
                rpc = list_entry(tsi->tsi_free_rpcs.next,
                                 srpc_client_rpc_t, crpc_list);
                list_del(&rpc->crpc_list);
                LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
        }

clean:
        sfw_unload_test(tsi);
        LIBCFS_FREE(tsi, sizeof(*tsi));
        return;
}

void
sfw_destroy_batch (sfw_batch_t *tsb)
{
        sfw_test_instance_t *tsi;

        LASSERT (!sfw_batch_active(tsb));
        LASSERT (list_empty(&tsb->bat_list));

        while (!list_empty(&tsb->bat_tests)) {
                tsi = list_entry(tsb->bat_tests.next,
                                 sfw_test_instance_t, tsi_list);
                list_del_init(&tsi->tsi_list);
                sfw_destroy_test_instance(tsi);
        }

        LIBCFS_FREE(tsb, sizeof(sfw_batch_t));
        return;
}

void
sfw_destroy_session (sfw_session_t *sn)
{
        sfw_batch_t *batch;

        LASSERT (list_empty(&sn->sn_list));
        LASSERT (sn != sfw_data.fw_session);

        while (!list_empty(&sn->sn_batches)) {
                batch = list_entry(sn->sn_batches.next,
                                   sfw_batch_t, bat_list);
                list_del_init(&batch->bat_list);
                sfw_destroy_batch(batch);
        }

        LIBCFS_FREE(sn, sizeof(*sn));
        atomic_dec(&sfw_data.fw_nzombies);
        return;
}

void
sfw_unpack_test_req (srpc_msg_t *msg)
{
        srpc_test_reqst_t *req = &msg->msg_body.tes_reqst;

        LASSERT (msg->msg_type == SRPC_MSG_TEST_REQST);
        LASSERT (req->tsr_is_client);

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

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

        if (req->tsr_service == SRPC_SERVICE_BRW) {
                test_bulk_req_t *bulk = &req->tsr_u.bulk;

                __swab32s(&bulk->blk_opc);
                __swab32s(&bulk->blk_npg);
                __swab32s(&bulk->blk_flags);
                return;
        }

        if (req->tsr_service == SRPC_SERVICE_PING) {
                test_ping_req_t *ping = &req->tsr_u.ping;

                __swab32s(&ping->png_size);
                __swab32s(&ping->png_flags);
                return;
        }

        LBUG ();
        return;
}

int
sfw_add_test_instance (sfw_batch_t *tsb, srpc_server_rpc_t *rpc)
{
        srpc_msg_t          *msg = &rpc->srpc_reqstbuf->buf_msg;
        srpc_test_reqst_t   *req = &msg->msg_body.tes_reqst;
        srpc_bulk_t         *bk = rpc->srpc_bulk;
        int                  ndest = req->tsr_ndest;
        sfw_test_unit_t     *tsu;
        sfw_test_instance_t *tsi;
        int                  i;
        int                  rc;

        LIBCFS_ALLOC(tsi, sizeof(*tsi));
        if (tsi == NULL) {
                CERROR ("Can't allocate test instance for batch: "LPU64"\n",
                        tsb->bat_id.bat_id);
                return -ENOMEM;
        }

        memset(tsi, 0, sizeof(*tsi));
        spin_lock_init(&tsi->tsi_lock);
        atomic_set(&tsi->tsi_nactive, 0);
        CFS_INIT_LIST_HEAD(&tsi->tsi_units);
        CFS_INIT_LIST_HEAD(&tsi->tsi_free_rpcs);
        CFS_INIT_LIST_HEAD(&tsi->tsi_active_rpcs);

        tsi->tsi_stopping      = 0;
        tsi->tsi_batch         = tsb;
        tsi->tsi_loop          = req->tsr_loop;
        tsi->tsi_concur        = req->tsr_concur;
        tsi->tsi_service       = req->tsr_service;
        tsi->tsi_is_client     = !!(req->tsr_is_client);
        tsi->tsi_stoptsu_onerr = !!(req->tsr_stop_onerr);

        rc = sfw_load_test(tsi);
        if (rc != 0) {
                LIBCFS_FREE(tsi, sizeof(*tsi));
                return rc;
        }

        LASSERT (!sfw_batch_active(tsb));

        if (!tsi->tsi_is_client) {
                /* it's test server, just add it to tsb */
                list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
                return 0;
        }

        LASSERT (bk != NULL);
#ifndef __KERNEL__
        LASSERT (bk->bk_pages != NULL);
#endif
        LASSERT (bk->bk_niov * SFW_ID_PER_PAGE >= ndest);
        LASSERT (bk->bk_len >= sizeof(lnet_process_id_t) * ndest);

        sfw_unpack_test_req(msg);
        memcpy(&tsi->tsi_u, &req->tsr_u, sizeof(tsi->tsi_u));

        for (i = 0; i < ndest; i++) {
                lnet_process_id_t *dests;
                lnet_process_id_t  id;
                int                j;

#ifdef __KERNEL__
                dests = cfs_page_address(bk->bk_iovs[i / SFW_ID_PER_PAGE].kiov_page);
                LASSERT (dests != NULL);  /* my pages are within KVM always */
#else
                dests = cfs_page_address(bk->bk_pages[i / SFW_ID_PER_PAGE]);
#endif
                id = dests[i % SFW_ID_PER_PAGE];
                if (msg->msg_magic != SRPC_MSG_MAGIC)
                        sfw_unpack_id(id);

                for (j = 0; j < tsi->tsi_concur; j++) {
                        LIBCFS_ALLOC(tsu, sizeof(sfw_test_unit_t));
                        if (tsu == NULL) {
                                rc = -ENOMEM;
                                CERROR ("Can't allocate tsu for %d\n",
                                        tsi->tsi_service);
                                goto error;
                        }

                        tsu->tsu_dest     = id;
                        tsu->tsu_instance = tsi;
                        tsu->tsu_private  = NULL;
                        list_add_tail(&tsu->tsu_list, &tsi->tsi_units);
                }
        }

        rc = tsi->tsi_ops->tso_init(tsi);
        if (rc == 0) {
                list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
                return 0;
        }

error:
        LASSERT (rc != 0);
        sfw_destroy_test_instance(tsi);
        return rc;
}

static void
sfw_test_unit_done (sfw_test_unit_t *tsu)
{
        sfw_test_instance_t *tsi = tsu->tsu_instance;
        sfw_batch_t         *tsb = tsi->tsi_batch;
        sfw_session_t       *sn = tsb->bat_session;

        LASSERT (sfw_test_active(tsi));

        if (!atomic_dec_and_test(&tsi->tsi_nactive))
                return;
        
        /* the test instance is done */
        spin_lock(&tsi->tsi_lock);

        tsi->tsi_stopping = 0;

        spin_unlock(&tsi->tsi_lock);

        spin_lock(&sfw_data.fw_lock);

        if (!atomic_dec_and_test(&tsb->bat_nactive) || /* tsb still active */
            sn == sfw_data.fw_session) {               /* sn also active */
                spin_unlock(&sfw_data.fw_lock);
                return;
        }
        
        LASSERT (!list_empty(&sn->sn_list)); /* I'm a zombie! */

        list_for_each_entry (tsb, &sn->sn_batches, bat_list) {
                if (sfw_batch_active(tsb)) {
                        spin_unlock(&sfw_data.fw_lock);
                        return;
                }
        }

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

        sfw_destroy_session(sn);
        return;
}

void
sfw_test_rpc_done (srpc_client_rpc_t *rpc)
{
        sfw_test_unit_t     *tsu = rpc->crpc_priv;
        sfw_test_instance_t *tsi = tsu->tsu_instance;
        int                  done = 0;

        tsi->tsi_ops->tso_done_rpc(tsu, rpc);
                      
        spin_lock(&tsi->tsi_lock);

        LASSERT (sfw_test_active(tsi));
        LASSERT (!list_empty(&rpc->crpc_list));

        list_del_init(&rpc->crpc_list);

        /* batch is stopping or loop is done or get error */
        if (tsi->tsi_stopping ||
            tsu->tsu_loop == 0 ||
            (rpc->crpc_status != 0 && tsi->tsi_stoptsu_onerr))
                done = 1;

        /* dec ref for poster */
        srpc_client_rpc_decref(rpc);

        spin_unlock(&tsi->tsi_lock);

        if (!done) {
                swi_schedule_workitem(&tsu->tsu_worker);
                return;
        }

        sfw_test_unit_done(tsu);
        return;
}

int
sfw_create_test_rpc (sfw_test_unit_t *tsu, lnet_process_id_t peer,
                     int nblk, int blklen, srpc_client_rpc_t **rpcpp)
{
        srpc_client_rpc_t   *rpc = NULL;
        sfw_test_instance_t *tsi = tsu->tsu_instance;
        
        spin_lock(&tsi->tsi_lock);

        LASSERT (sfw_test_active(tsi));

        if (!list_empty(&tsi->tsi_free_rpcs)) {
                /* pick request from buffer */
                rpc = list_entry(tsi->tsi_free_rpcs.next,
                                 srpc_client_rpc_t, crpc_list);
                LASSERT (nblk == rpc->crpc_bulk.bk_niov);