lnet/selftest/framework.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * Copyright (C) 2001, 2002 Cluster File Systems, Inc.
   5  *   Authors: Isaac Huang <isaac@clusterfs.com>
   6  *            Liang Zhen  <liangzhen@clusterfs.com>
   7  */
   8
   9 #define DEBUG_SUBSYSTEM S_LNET
  10
  11 #include <libcfs/kp30.h>
  12 #include <libcfs/libcfs.h>
  13 #include <lnet/lib-lnet.h>
  14
  15 #include "selftest.h"
  16
  17
  18 static int session_timeout = 100;
  19 CFS_MODULE_PARM(session_timeout, "i", int, 0444,
  20                 "test session timeout in seconds (100 by default, 0 == never)");
  21
  22 #define SFW_TEST_CONCURRENCY     128
  23 #define SFW_TEST_RPC_TIMEOUT     64
  24 #define SFW_CLIENT_RPC_TIMEOUT   64  /* in seconds */
  25 #define SFW_EXTRA_TEST_BUFFERS   8 /* tolerate buggy peers with extra buffers */
  26
  27 #define sfw_test_buffers(tsi)    ((tsi)->tsi_loop + SFW_EXTRA_TEST_BUFFERS)
  28
  29 #define sfw_unpack_id(id)               \
  30 do {                                    \
  31         __swab64s(&(id).nid);           \
  32         __swab32s(&(id).pid);           \
  33 } while (0)
  34
  35 #define sfw_unpack_sid(sid)             \
  36 do {                                    \
  37         __swab64s(&(sid).ses_nid);      \
  38         __swab64s(&(sid).ses_stamp);    \
  39 } while (0)
  40
  41 #define sfw_unpack_fw_counters(fc)        \
  42 do {                                      \
  43         __swab32s(&(fc).brw_errors);      \
  44         __swab32s(&(fc).active_tests);    \
  45         __swab32s(&(fc).active_batches);  \
  46         __swab32s(&(fc).zombie_sessions); \
  47 } while (0)
  48
  49 #define sfw_unpack_rpc_counters(rc)     \
  50 do {                                    \
  51         __swab32s(&(rc).errors);        \
  52         __swab32s(&(rc).rpcs_sent);     \
  53         __swab32s(&(rc).rpcs_rcvd);     \
  54         __swab32s(&(rc).rpcs_dropped);  \
  55         __swab32s(&(rc).rpcs_expired);  \
  56         __swab64s(&(rc).bulk_get);      \
  57         __swab64s(&(rc).bulk_put);      \
  58 } while (0)
  59
  60 #define sfw_unpack_lnet_counters(lc)    \
  61 do {                                    \
  62         __swab32s(&(lc).errors);        \
  63         __swab32s(&(lc).msgs_max);      \
  64         __swab32s(&(lc).msgs_alloc);    \
  65         __swab32s(&(lc).send_count);    \
  66         __swab32s(&(lc).recv_count);    \
  67         __swab32s(&(lc).drop_count);    \
  68         __swab32s(&(lc).route_count);   \
  69         __swab64s(&(lc).send_length);   \
  70         __swab64s(&(lc).recv_length);   \
  71         __swab64s(&(lc).drop_length);   \
  72         __swab64s(&(lc).route_length);  \
  73 } while (0)
  74
  75 #define sfw_test_active(t)      (atomic_read(&(t)->tsi_nactive) != 0)
  76 #define sfw_batch_active(b)     (atomic_read(&(b)->bat_nactive) != 0)
  77
  78 struct smoketest_framework {
  79         struct list_head   fw_zombie_rpcs;     /* RPCs to be recycled */
  80         struct list_head   fw_zombie_sessions; /* stopping sessions */
  81         struct list_head   fw_tests;           /* registered test cases */
  82         atomic_t           fw_nzombies;        /* # zombie sessions */
  83         spinlock_t         fw_lock;            /* serialise */
  84         sfw_session_t     *fw_session;         /* _the_ session */
  85         int                fw_shuttingdown;    /* shutdown in progress */
  86         srpc_server_rpc_t *fw_active_srpc;     /* running RPC */
  87 } sfw_data;
  88
  89 /* forward ref's */
  90 int sfw_stop_batch (sfw_batch_t *tsb, int force);
  91 void sfw_destroy_session (sfw_session_t *sn);
  92
  93 static inline sfw_test_case_t *
  94 sfw_find_test_case(int id)
  95 {
  96         sfw_test_case_t *tsc;
  97
  98         LASSERT (id <= SRPC_SERVICE_MAX_ID);
  99         LASSERT (id > SRPC_FRAMEWORK_SERVICE_MAX_ID);
 100
 101         list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
 102                 if (tsc->tsc_srv_service->sv_id == id)
 103                         return tsc;
 104         }
 105
 106         return NULL;
 107 }
 108
 109 static int
 110 sfw_register_test (srpc_service_t *service, sfw_test_client_ops_t *cliops)
 111 {
 112         sfw_test_case_t *tsc;
 113
 114         if (sfw_find_test_case(service->sv_id) != NULL) {
 115                 CERROR ("Failed to register test %s (%d)\n",
 116                         service->sv_name, service->sv_id);
 117                 return -EEXIST;
 118         }
 119
 120         LIBCFS_ALLOC(tsc, sizeof(sfw_test_case_t));
 121         if (tsc == NULL)
 122                 return -ENOMEM;
 123
 124         memset(tsc, 0, sizeof(sfw_test_case_t));
 125         tsc->tsc_cli_ops     = cliops;
 126         tsc->tsc_srv_service = service;
 127
 128         list_add_tail(&tsc->tsc_list, &sfw_data.fw_tests);
 129         return 0;
 130 }
 131
 132 void
 133 sfw_add_session_timer (void)
 134 {
 135         sfw_session_t *sn = sfw_data.fw_session;
 136         stt_timer_t   *timer = &sn->sn_timer;
 137
 138         LASSERT (!sfw_data.fw_shuttingdown);
 139
 140         if (sn == NULL || sn->sn_timeout == 0)
 141                 return;
 142
 143         LASSERT (!sn->sn_timer_active);
 144
 145         sn->sn_timer_active = 1;
 146         timer->stt_expires = cfs_time_add(sn->sn_timeout,
 147                                           cfs_time_current_sec());
 148         stt_add_timer(timer);
 149         return;
 150 }
 151
 152 int
 153 sfw_del_session_timer (void)
 154 {
 155         sfw_session_t *sn = sfw_data.fw_session;
 156
 157         if (sn == NULL || !sn->sn_timer_active)
 158                 return 0;
 159
 160         LASSERT (sn->sn_timeout != 0);
 161
 162         if (stt_del_timer(&sn->sn_timer)) { /* timer defused */
 163                 sn->sn_timer_active = 0;
 164                 return 0;
 165         }
 166
 167 #ifndef __KERNEL__
 168         /* Racing is impossible in single-threaded userland selftest */
 169         LBUG();
 170 #endif
 171         return EBUSY; /* racing with sfw_session_expired() */
 172 }
 173
 174 /* called with sfw_data.fw_lock held */
 175 static void
 176 sfw_deactivate_session (void)
 177 {
 178         sfw_session_t *sn = sfw_data.fw_session;
 179         int            nactive = 0;
 180         sfw_batch_t   *tsb;
 181
 182         if (sn == NULL)
 183                 return;
 184
 185         LASSERT (!sn->sn_timer_active);
 186
 187         sfw_data.fw_session = NULL;
 188         atomic_inc(&sfw_data.fw_nzombies);
 189         list_add(&sn->sn_list, &sfw_data.fw_zombie_sessions);
 190
 191         list_for_each_entry (tsb, &sn->sn_batches, bat_list) {
 192                 if (sfw_batch_active(tsb)) {
 193                         nactive++;
 194                         sfw_stop_batch(tsb, 1);
 195                 }
 196         }
 197
 198         if (nactive != 0)
 199                 return;   /* wait for active batches to stop */
 200
 201         list_del_init(&sn->sn_list);
 202         spin_unlock(&sfw_data.fw_lock);
 203
 204         sfw_destroy_session(sn);
 205
 206         spin_lock(&sfw_data.fw_lock);
 207         return;
 208 }
 209
 210 #ifndef __KERNEL__
 211
 212 int
 213 sfw_session_removed (void)
 214 {
 215         return (sfw_data.fw_session == NULL) ? 1 : 0;
 216 }
 217
 218 void
 219 sfw_set_session_timeout (int timeout)
 220 {
 221         session_timeout = timeout;
 222 }
 223
 224 #endif
 225
 226 void
 227 sfw_session_expired (void *data)
 228 {
 229         sfw_session_t *sn = data;
 230
 231         spin_lock(&sfw_data.fw_lock);
 232
 233         LASSERT (sn->sn_timer_active);
 234         LASSERT (sn == sfw_data.fw_session);
 235
 236         CWARN ("Session expired! sid: %s-"LPU64", name: %s\n",
 237                libcfs_nid2str(sn->sn_id.ses_nid),
 238                sn->sn_id.ses_stamp, &sn->sn_name[0]);
 239
 240         sn->sn_timer_active = 0;
 241         sfw_deactivate_session();
 242
 243         spin_unlock(&sfw_data.fw_lock);
 244         return;
 245 }
 246
 247 static inline void
 248 sfw_init_session (sfw_session_t *sn, lst_sid_t sid, const char *name)
 249 {
 250         stt_timer_t *timer = &sn->sn_timer;
 251
 252         memset(sn, 0, sizeof(sfw_session_t));
 253         CFS_INIT_LIST_HEAD(&sn->sn_list);
 254         CFS_INIT_LIST_HEAD(&sn->sn_batches);
 255         atomic_set(&sn->sn_brw_errors, 0);
 256         strncpy(&sn->sn_name[0], name, LST_NAME_SIZE);
 257
 258         sn->sn_timer_active = 0;
 259         sn->sn_id           = sid;
 260         sn->sn_timeout      = session_timeout;
 261
 262         timer->stt_data = sn;
 263         timer->stt_func = sfw_session_expired;
 264         CFS_INIT_LIST_HEAD(&timer->stt_list);
 265 }
 266
 267 /* completion handler for incoming framework RPCs */
 268 void
 269 sfw_server_rpc_done (srpc_server_rpc_t *rpc)
 270 {
 271         srpc_service_t *sv = rpc->srpc_service;
 272         int             status = rpc->srpc_status;
 273
 274         CDEBUG (D_NET,
 275                 "Incoming framework RPC done: "
 276                 "service %s, peer %s, status %s:%d\n",
 277                 sv->sv_name, libcfs_id2str(rpc->srpc_peer),
 278                 swi_state2str(rpc->srpc_wi.wi_state),
 279                 status);
 280
 281         if (rpc->srpc_bulk != NULL)
 282                 sfw_free_pages(rpc);
 283         return;
 284 }
 285
 286 void
 287 sfw_client_rpc_fini (srpc_client_rpc_t *rpc)
 288 {
 289         LASSERT (rpc->crpc_bulk.bk_niov == 0);
 290         LASSERT (list_empty(&rpc->crpc_list));
 291         LASSERT (atomic_read(&rpc->crpc_refcount) == 0);
 292 #ifndef __KERNEL__
 293         LASSERT (rpc->crpc_bulk.bk_pages == NULL);
 294 #endif
 295
 296         CDEBUG (D_NET,
 297                 "Outgoing framework RPC done: "
 298                 "service %d, peer %s, status %s:%d:%d\n",
 299                 rpc->crpc_service, libcfs_id2str(rpc->crpc_dest),
 300                 swi_state2str(rpc->crpc_wi.wi_state),
 301                 rpc->crpc_aborted, rpc->crpc_status);
 302
 303         spin_lock(&sfw_data.fw_lock);
 304
 305         /* my callers must finish all RPCs before shutting me down */
 306         LASSERT (!sfw_data.fw_shuttingdown);
 307         list_add(&rpc->crpc_list, &sfw_data.fw_zombie_rpcs);
 308
 309         spin_unlock(&sfw_data.fw_lock);
 310         return;
 311 }
 312
 313 sfw_batch_t *
 314 sfw_find_batch (lst_bid_t bid)
 315 {
 316         sfw_session_t *sn = sfw_data.fw_session;
 317         sfw_batch_t   *bat;
 318
 319         LASSERT (sn != NULL);
 320
 321         list_for_each_entry (bat, &sn->sn_batches, bat_list) {
 322                 if (bat->bat_id.bat_id == bid.bat_id)
 323                         return bat;
 324         }
 325
 326         return NULL;
 327 }
 328
 329 sfw_batch_t *
 330 sfw_bid2batch (lst_bid_t bid)
 331 {
 332         sfw_session_t *sn = sfw_data.fw_session;
 333         sfw_batch_t   *bat;
 334
 335         LASSERT (sn != NULL);
 336
 337         bat = sfw_find_batch(bid);
 338         if (bat != NULL)
 339                 return bat;
 340
 341         LIBCFS_ALLOC(bat, sizeof(sfw_batch_t));
 342         if (bat == NULL)
 343                 return NULL;
 344
 345         bat->bat_error    = 0;
 346         bat->bat_session  = sn;
 347         bat->bat_id       = bid;
 348         atomic_set(&bat->bat_nactive, 0);
 349         CFS_INIT_LIST_HEAD(&bat->bat_tests);
 350
 351         list_add_tail(&bat->bat_list, &sn->sn_batches);
 352         return bat;
 353 }
 354
 355 int
 356 sfw_get_stats (srpc_stat_reqst_t *request, srpc_stat_reply_t *reply)
 357 {
 358         sfw_session_t  *sn = sfw_data.fw_session;
 359         sfw_counters_t *cnt = &reply->str_fw;
 360         sfw_batch_t    *bat;
 361
 362         reply->str_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
 363
 364         if (request->str_sid.ses_nid == LNET_NID_ANY) {
 365                 reply->str_status = EINVAL;
 366                 return 0;
 367         }
 368
 369         if (sn == NULL || !sfw_sid_equal(request->str_sid, sn->sn_id)) {
 370                 reply->str_status = ESRCH;
 371                 return 0;
 372         }
 373
 374         LNET_LOCK();
 375         reply->str_lnet = the_lnet.ln_counters;
 376         LNET_UNLOCK();
 377
 378         srpc_get_counters(&reply->str_rpc);
 379
 380         cnt->brw_errors      = atomic_read(&sn->sn_brw_errors);
 381         cnt->zombie_sessions = atomic_read(&sfw_data.fw_nzombies);
 382
 383         cnt->active_tests = cnt->active_batches = 0;
 384         list_for_each_entry (bat, &sn->sn_batches, bat_list) {
 385                 int n = atomic_read(&bat->bat_nactive);
 386
 387                 if (n > 0) {
 388                         cnt->active_batches++;
 389                         cnt->active_tests += n;
 390                 }
 391         }
 392
 393         reply->str_status = 0;
 394         return 0;
 395 }
 396
 397 int
 398 sfw_make_session (srpc_mksn_reqst_t *request, srpc_mksn_reply_t *reply)
 399 {
 400         sfw_session_t *sn = sfw_data.fw_session;
 401
 402         if (request->mksn_sid.ses_nid == LNET_NID_ANY) {
 403                 reply->mksn_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
 404                 reply->mksn_status = EINVAL;
 405                 return 0;
 406         }
 407
 408         if (sn != NULL && !request->mksn_force) {
 409                 reply->mksn_sid    = sn->sn_id;
 410                 reply->mksn_status = EBUSY;
 411                 strncpy(&reply->mksn_name[0], &sn->sn_name[0], LST_NAME_SIZE);
 412                 return 0;
 413         }
 414
 415         LIBCFS_ALLOC(sn, sizeof(sfw_session_t));
 416         if (sn == NULL) {
 417                 CERROR ("Dropping RPC (mksn) under memory pressure.\n");
 418                 return -ENOMEM;
 419         }
 420
 421         sfw_init_session(sn, request->mksn_sid, &request->mksn_name[0]);
 422
 423         spin_lock(&sfw_data.fw_lock);
 424
 425         sfw_deactivate_session();
 426         LASSERT (sfw_data.fw_session == NULL);
 427         sfw_data.fw_session = sn;
 428
 429         spin_unlock(&sfw_data.fw_lock);
 430
 431         reply->mksn_status  = 0;
 432         reply->mksn_sid     = sn->sn_id;
 433         reply->mksn_timeout = sn->sn_timeout;
 434         return 0;
 435 }
 436
 437 int
 438 sfw_remove_session (srpc_rmsn_reqst_t *request, srpc_rmsn_reply_t *reply)
 439 {
 440         sfw_session_t *sn = sfw_data.fw_session;
 441
 442         reply->rmsn_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
 443
 444         if (request->rmsn_sid.ses_nid == LNET_NID_ANY) {
 445                 reply->rmsn_status = EINVAL;
 446                 return 0;
 447         }
 448
 449         if (sn == NULL || !sfw_sid_equal(request->rmsn_sid, sn->sn_id)) {
 450                 reply->rmsn_status = (sn == NULL) ? ESRCH : EBUSY;
 451                 return 0;
 452         }
 453
 454         spin_lock(&sfw_data.fw_lock);
 455         sfw_deactivate_session();
 456         spin_unlock(&sfw_data.fw_lock);
 457
 458         reply->rmsn_status = 0;
 459         reply->rmsn_sid    = LST_INVALID_SID;
 460         LASSERT (sfw_data.fw_session == NULL);
 461         return 0;
 462 }
 463
 464 int
 465 sfw_debug_session (srpc_debug_reqst_t *request, srpc_debug_reply_t *reply)
 466 {
 467         sfw_session_t *sn = sfw_data.fw_session;
 468
 469         if (sn == NULL) {
 470                 reply->dbg_status = ESRCH;
 471                 reply->dbg_sid    = LST_INVALID_SID;
 472                 return 0;
 473         }
 474
 475         reply->dbg_status  = 0;
 476         reply->dbg_sid     = sn->sn_id;
 477         reply->dbg_timeout = sn->sn_timeout;
 478         strncpy(reply->dbg_name, &sn->sn_name[0], LST_NAME_SIZE);
 479
 480         return 0;
 481 }
 482
 483 void
 484 sfw_test_rpc_fini (srpc_client_rpc_t *rpc)
 485 {
 486         sfw_test_unit_t     *tsu = rpc->crpc_priv;
 487         sfw_test_instance_t *tsi = tsu->tsu_instance;
 488
 489         /* Called with hold of tsi->tsi_lock */
 490         LASSERT (list_empty(&rpc->crpc_list));
 491         list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
 492 }
 493
 494 int
 495 sfw_load_test (sfw_test_instance_t *tsi)
 496 {
 497         sfw_test_case_t *tsc = sfw_find_test_case(tsi->tsi_service);
 498         int              nrequired = sfw_test_buffers(tsi);
 499         int              nposted;
 500
 501         LASSERT (tsc != NULL);
 502
 503         if (tsi->tsi_is_client) {
 504                 tsi->tsi_ops = tsc->tsc_cli_ops;
 505                 return 0;
 506         }
 507
 508         nposted = srpc_service_add_buffers(tsc->tsc_srv_service, nrequired);
 509         if (nposted != nrequired) {
 510                 CWARN ("Failed to reserve enough buffers: "
 511                        "service %s, %d needed, %d reserved\n",
 512                        tsc->tsc_srv_service->sv_name, nrequired, nposted);
 513                 srpc_service_remove_buffers(tsc->tsc_srv_service, nposted);
 514                 return -ENOMEM;
 515         }
 516
 517         CDEBUG (D_NET, "Reserved %d buffers for test %s\n",
 518                 nposted, tsc->tsc_srv_service->sv_name);
 519         return 0;
 520 }
 521
 522 void
 523 sfw_unload_test (sfw_test_instance_t *tsi)
 524 {
 525         sfw_test_case_t *tsc = sfw_find_test_case(tsi->tsi_service);
 526
 527         LASSERT (tsc != NULL);
 528
 529         if (!tsi->tsi_is_client)
 530                 srpc_service_remove_buffers(tsc->tsc_srv_service,
 531                                             sfw_test_buffers(tsi));
 532         return;
 533 }
 534
 535 void
 536 sfw_destroy_test_instance (sfw_test_instance_t *tsi)
 537 {
 538         srpc_client_rpc_t *rpc;
 539         sfw_test_unit_t   *tsu;
 540
 541         if (!tsi->tsi_is_client)
 542                 goto clean;
 543
 544         tsi->tsi_ops->tso_fini(tsi);
 545
 546         LASSERT (!tsi->tsi_stopping);
 547         LASSERT (list_empty(&tsi->tsi_active_rpcs));
 548         LASSERT (!sfw_test_active(tsi));
 549
 550         while (!list_empty(&tsi->tsi_units)) {
 551                 tsu = list_entry(tsi->tsi_units.next,
 552                                  sfw_test_unit_t, tsu_list);
 553                 list_del(&tsu->tsu_list);
 554                 LIBCFS_FREE(tsu, sizeof(*tsu));
 555         }
 556
 557         while (!list_empty(&tsi->tsi_free_rpcs)) {
 558                 rpc = list_entry(tsi->tsi_free_rpcs.next,
 559                                  srpc_client_rpc_t, crpc_list);
 560                 list_del(&rpc->crpc_list);
 561                 LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
 562         }
 563
 564 clean:
 565         sfw_unload_test(tsi);
 566         LIBCFS_FREE(tsi, sizeof(*tsi));
 567         return;
 568 }
 569
 570 void
 571 sfw_destroy_batch (sfw_batch_t *tsb)
 572 {
 573         sfw_test_instance_t *tsi;
 574
 575         LASSERT (!sfw_batch_active(tsb));
 576         LASSERT (list_empty(&tsb->bat_list));
 577
 578         while (!list_empty(&tsb->bat_tests)) {
 579                 tsi = list_entry(tsb->bat_tests.next,
 580                                  sfw_test_instance_t, tsi_list);
 581                 list_del_init(&tsi->tsi_list);
 582                 sfw_destroy_test_instance(tsi);
 583         }
 584
 585         LIBCFS_FREE(tsb, sizeof(sfw_batch_t));
 586         return;
 587 }
 588
 589 void
 590 sfw_destroy_session (sfw_session_t *sn)
 591 {
 592         sfw_batch_t *batch;
 593
 594         LASSERT (list_empty(&sn->sn_list));
 595         LASSERT (sn != sfw_data.fw_session);
 596
 597         while (!list_empty(&sn->sn_batches)) {
 598                 batch = list_entry(sn->sn_batches.next,
 599                                    sfw_batch_t, bat_list);
 600                 list_del_init(&batch->bat_list);
 601                 sfw_destroy_batch(batch);
 602         }
 603
 604         LIBCFS_FREE(sn, sizeof(*sn));
 605         atomic_dec(&sfw_data.fw_nzombies);
 606         return;
 607 }
 608
 609 void
 610 sfw_unpack_test_req (srpc_msg_t *msg)
 611 {
 612         srpc_test_reqst_t *req = &msg->msg_body.tes_reqst;
 613
 614         LASSERT (msg->msg_type == SRPC_MSG_TEST_REQST);
 615         LASSERT (req->tsr_is_client);
 616
 617         if (msg->msg_magic == SRPC_MSG_MAGIC)
 618                 return; /* no flipping needed */
 619
 620         LASSERT (msg->msg_magic == __swab32(SRPC_MSG_MAGIC));
 621
 622         if (req->tsr_service == SRPC_SERVICE_BRW) {
 623                 test_bulk_req_t *bulk = &req->tsr_u.bulk;
 624
 625                 __swab32s(&bulk->blk_opc);
 626                 __swab32s(&bulk->blk_npg);
 627                 __swab32s(&bulk->blk_flags);
 628                 return;
 629         }
 630
 631         if (req->tsr_service == SRPC_SERVICE_PING) {
 632                 test_ping_req_t *ping = &req->tsr_u.ping;
 633
 634                 __swab32s(&ping->png_size);
 635                 __swab32s(&ping->png_flags);
 636                 return;
 637         }
 638
 639         LBUG ();
 640         return;
 641 }
 642
 643 int
 644 sfw_add_test_instance (sfw_batch_t *tsb, srpc_server_rpc_t *rpc)
 645 {
 646         srpc_msg_t          *msg = &rpc->srpc_reqstbuf->buf_msg;
 647         srpc_test_reqst_t   *req = &msg->msg_body.tes_reqst;
 648         srpc_bulk_t         *bk = rpc->srpc_bulk;
 649         int                  ndest = req->tsr_ndest;
 650         sfw_test_unit_t     *tsu;
 651         sfw_test_instance_t *tsi;
 652         int                  i;
 653         int                  rc;
 654
 655         LIBCFS_ALLOC(tsi, sizeof(*tsi));
 656         if (tsi == NULL) {
 657                 CERROR ("Can't allocate test instance for batch: "LPU64"\n",
 658                         tsb->bat_id.bat_id);
 659                 return -ENOMEM;
 660         }
 661
 662         memset(tsi, 0, sizeof(*tsi));
 663         spin_lock_init(&tsi->tsi_lock);
 664         atomic_set(&tsi->tsi_nactive, 0);
 665         CFS_INIT_LIST_HEAD(&tsi->tsi_units);
 666         CFS_INIT_LIST_HEAD(&tsi->tsi_free_rpcs);
 667         CFS_INIT_LIST_HEAD(&tsi->tsi_active_rpcs);
 668
 669         tsi->tsi_stopping   = 0;
 670         tsi->tsi_batch      = tsb;
 671         tsi->tsi_loop       = req->tsr_loop;
 672         tsi->tsi_concur     = req->tsr_concur;
 673         tsi->tsi_service    = req->tsr_service;
 674         tsi->tsi_is_client  = !!(req->tsr_is_client);
 675         tsi->tsi_stop_onerr = !!(req->tsr_stop_onerr);
 676
 677         rc = sfw_load_test(tsi);
 678         if (rc != 0) {
 679                 LIBCFS_FREE(tsi, sizeof(*tsi));
 680                 return rc;
 681         }
 682
 683         LASSERT (!sfw_batch_active(tsb));
 684
 685         if (!tsi->tsi_is_client) {
 686                 /* it's test server, just add it to tsb */
 687                 list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
 688                 return 0;
 689         }
 690
 691         LASSERT (bk != NULL);
 692 #ifndef __KERNEL__
 693         LASSERT (bk->bk_pages != NULL);
 694 #endif
 695         LASSERT (bk->bk_niov * SFW_ID_PER_PAGE >= ndest);
 696         LASSERT (bk->bk_len >= sizeof(lnet_process_id_t) * ndest);
 697
 698         sfw_unpack_test_req(msg);
 699         memcpy(&tsi->tsi_u, &req->tsr_u, sizeof(tsi->tsi_u));
 700
 701         for (i = 0; i < ndest; i++) {
 702                 lnet_process_id_t *dests;
 703                 lnet_process_id_t  id;
 704                 int                j;
 705
 706 #ifdef __KERNEL__
 707                 dests = cfs_page_address(bk->bk_iovs[i / SFW_ID_PER_PAGE].kiov_page);
 708                 LASSERT (dests != NULL);  /* my pages are within KVM always */
 709 #else
 710                 dests = cfs_page_address(bk->bk_pages[i / SFW_ID_PER_PAGE]);
 711 #endif
 712                 id = dests[i % SFW_ID_PER_PAGE];
 713                 if (msg->msg_magic != SRPC_MSG_MAGIC)
 714                         sfw_unpack_id(id);
 715
 716                 for (j = 0; j < tsi->tsi_concur; j++) {
 717                         LIBCFS_ALLOC(tsu, sizeof(sfw_test_unit_t));
 718                         if (tsu == NULL) {
 719                                 rc = -ENOMEM;
 720                                 CERROR ("Can't allocate tsu for %d\n",
 721                                         tsi->tsi_service);
 722                                 goto error;
 723                         }
 724
 725                         tsu->tsu_dest     = id;
 726                         tsu->tsu_instance = tsi;
 727                         tsu->tsu_private  = NULL;
 728                         list_add_tail(&tsu->tsu_list, &tsi->tsi_units);
 729                 }
 730         }
 731
 732         rc = tsi->tsi_ops->tso_init(tsi);
 733         if (rc == 0) {
 734                 list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
 735                 return 0;
 736         }
 737
 738 error:
 739         LASSERT (rc != 0);
 740         sfw_destroy_test_instance(tsi);
 741         return rc;
 742 }
 743
 744 static void
 745 sfw_test_unit_done (sfw_test_unit_t *tsu)
 746 {
 747         sfw_test_instance_t *tsi = tsu->tsu_instance;
 748         sfw_batch_t         *tsb = tsi->tsi_batch;
 749         sfw_session_t       *sn = tsb->bat_session;
 750
 751         LASSERT (sfw_test_active(tsi));
 752
 753         if (!atomic_dec_and_test(&tsi->tsi_nactive))
 754                 return;
 755
 756         /* the test instance is done */
 757         spin_lock(&tsi->tsi_lock);
 758
 759         tsi->tsi_stopping = 0;
 760
 761         spin_unlock(&tsi->tsi_lock);
 762
 763         spin_lock(&sfw_data.fw_lock);
 764
 765         if (!atomic_dec_and_test(&tsb->bat_nactive) || /* tsb still active */
 766             sn == sfw_data.fw_session) {               /* sn also active */
 767                 spin_unlock(&sfw_data.fw_lock);
 768                 return;
 769         }
 770
 771         LASSERT (!list_empty(&sn->sn_list)); /* I'm a zombie! */
 772
 773         list_for_each_entry (tsb, &sn->sn_batches, bat_list) {
 774                 if (sfw_batch_active(tsb)) {
 775                         spin_unlock(&sfw_data.fw_lock);
 776                         return;
 777                 }
 778         }
 779
 780         list_del_init(&sn->sn_list);
 781         spin_unlock(&sfw_data.fw_lock);
 782
 783         sfw_destroy_session(sn);
 784         return;
 785 }
 786
 787 void
 788 sfw_test_rpc_done (srpc_client_rpc_t *rpc)
 789 {
 790         sfw_test_unit_t     *tsu = rpc->crpc_priv;
 791         sfw_test_instance_t *tsi = tsu->tsu_instance;
 792         int                  done = 0;
 793
 794         if (rpc->crpc_status != 0 && tsu->tsu_error == 0 &&
 795             (rpc->crpc_status != -EINTR || !tsi->tsi_stopping))
 796                 tsu->tsu_error = rpc->crpc_status;
 797
 798         tsi->tsi_ops->tso_done_rpc(tsu, rpc);
 799
 800         spin_lock(&tsi->tsi_lock);
 801
 802         LASSERT (sfw_test_active(tsi));
 803         LASSERT (!list_empty(&rpc->crpc_list));
 804
 805         list_del_init(&rpc->crpc_list);
 806
 807         /* batch is stopping or loop is done or get error */
 808         if (tsi->tsi_stopping ||
 809             tsu->tsu_loop == 0 ||
 810             (tsu->tsu_error != 0 && tsi->tsi_stop_onerr))
 811                 done = 1;
 812
 813         /* dec ref for poster */
 814         srpc_client_rpc_decref(rpc);
 815
 816         spin_unlock(&tsi->tsi_lock);
 817
 818         if (!done) {
 819                 swi_schedule_workitem(&tsu->tsu_worker);
 820                 return;
 821         }
 822
 823         sfw_test_unit_done(tsu);
 824         return;
 825 }
 826
 827 int
 828 sfw_create_test_rpc (sfw_test_unit_t *tsu, lnet_process_id_t peer,
 829                      int nblk, int blklen, srpc_client_rpc_t **rpcpp)
 830 {
 831         srpc_client_rpc_t   *rpc = NULL;
 832         sfw_test_instance_t *tsi = tsu->tsu_instance;
 833
 834         spin_lock(&tsi->tsi_lock);
 835
 836         LASSERT (sfw_test_active(tsi));
 837
 838         if (!list_empty(&tsi->tsi_free_rpcs)) {
 839                 /* pick request from buffer */
 840                 rpc = list_entry(tsi->tsi_free_rpcs.next,
 841                                  srpc_client_rpc_t, crpc_list);
 842                 LASSERT (nblk == rpc->crpc_bulk.bk_niov);
 843                 list_del_init(&rpc->crpc_list);
 844
 845                 srpc_init_client_rpc(rpc, peer, tsi->tsi_service, nblk,
 846                                      blklen, sfw_test_rpc_done,
 847                                      sfw_test_rpc_fini, tsu);
 848         }
 849
 850         spin_unlock(&tsi->tsi_lock);
 851
 852         if (rpc == NULL)
 853                 rpc = srpc_create_client_rpc(peer, tsi->tsi_service, nblk,
 854                                              blklen, sfw_test_rpc_done,
 855                                              sfw_test_rpc_fini, tsu);
 856         if (rpc == NULL) {
 857                 CERROR ("Can't create rpc for test %d\n", tsi->tsi_service);
 858                 return -ENOMEM;
 859         }
 860
 861         *rpcpp = rpc;
 862         return 0;
 863 }
 864
 865 int
 866 sfw_run_test (swi_workitem_t *wi)
 867 {
 868         sfw_test_unit_t     *tsu = wi->wi_data;
 869         sfw_test_instance_t *tsi = tsu->tsu_instance;
 870         srpc_client_rpc_t   *rpc = NULL;
 871
 872         LASSERT (wi == &tsu->tsu_worker);
 873
 874         tsu->tsu_error = tsi->tsi_ops->tso_prep_rpc(tsu, tsu->tsu_dest, &rpc);
 875         if (tsu->tsu_error != 0) {
 876                 LASSERT (rpc == NULL);
 877                 goto test_done;
 878         }
 879
 880         LASSERT (rpc != NULL);
 881
 882         spin_lock(&tsi->tsi_lock);
 883
 884         if (tsi->tsi_stopping) {
 885                 list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
 886                 spin_unlock(&tsi->tsi_lock);
 887                 goto test_done;
 888         }
 889
 890         if (tsu->tsu_loop > 0)
 891                 tsu->tsu_loop--;
 892
 893         list_add_tail(&rpc->crpc_list, &tsi->tsi_active_rpcs);
 894         spin_unlock(&tsi->tsi_lock);
 895
 896         rpc->crpc_timeout = SFW_TEST_RPC_TIMEOUT;
 897
 898         spin_lock(&rpc->crpc_lock);
 899         srpc_post_rpc(rpc);
 900         spin_unlock(&rpc->crpc_lock);
 901         return 0;
 902
 903 test_done:
 904         /*
 905          * No one can schedule me now since:
 906          * - previous RPC, if any, has done and
 907          * - no new RPC is initiated.
 908          * - my batch is still active; no one can run it again now.
 909          * Cancel pending schedules and prevent future schedule attempts:
 910          */
 911         swi_kill_workitem(wi);
 912         sfw_test_unit_done(tsu);
 913         return 1;
 914 }
 915
 916 int
 917 sfw_run_batch (sfw_batch_t *tsb)
 918 {
 919         swi_workitem_t      *wi;
 920         sfw_test_unit_t     *tsu;
 921         sfw_test_instance_t *tsi;
 922
 923         if (sfw_batch_active(tsb)) {
 924                 CDEBUG (D_NET, "Can't start active batch: "LPU64" (%d)\n",
 925                         tsb->bat_id.bat_id, atomic_read(&tsb->bat_nactive));
 926                 return -EPERM;
 927         }
 928
 929         list_for_each_entry (tsi, &tsb->bat_tests, tsi_list) {
 930                 if (!tsi->tsi_is_client) /* skip server instances */
 931                         continue;
 932
 933                 LASSERT (!tsi->tsi_stopping);
 934                 LASSERT (!sfw_test_active(tsi));
 935
 936                 atomic_inc(&tsb->bat_nactive);
 937
 938                 list_for_each_entry (tsu, &tsi->tsi_units, tsu_list) {
 939                         atomic_inc(&tsi->tsi_nactive);
 940
 941                         tsu->tsu_error = 0;
 942                         tsu->tsu_loop  = tsi->tsi_loop;
 943
 944                         wi = &tsu->tsu_worker;
 945                         swi_init_workitem(wi, tsu, sfw_run_test);
 946                         swi_schedule_workitem(wi);
 947                 }
 948         }
 949
 950         return 0;
 951 }
 952
 953 int
 954 sfw_stop_batch (sfw_batch_t *tsb, int force)
 955 {
 956         sfw_test_instance_t *tsi;
 957         srpc_client_rpc_t   *rpc;
 958
 959         if (!sfw_batch_active(tsb))
 960                 return -EPERM;
 961
 962         list_for_each_entry (tsi, &tsb->bat_tests, tsi_list) {
 963                 spin_lock(&tsi->tsi_lock);
 964
 965                 if (!tsi->tsi_is_client ||
 966                     !sfw_test_active(tsi) || tsi->tsi_stopping) {
 967                         spin_unlock(&tsi->tsi_lock);
 968                         continue;
 969                 }
 970
 971                 tsi->tsi_stopping = 1;
 972
 973                 if (!force) {
 974                         spin_unlock(&tsi->tsi_lock);
 975                         continue;
 976                 }
 977
 978                 /* abort launched rpcs in the test */
 979                 list_for_each_entry (rpc, &tsi->tsi_active_rpcs, crpc_list) {
 980                         spin_lock(&rpc->crpc_lock);
 981
 982                         srpc_abort_rpc(rpc, -EINTR);
 983
 984                         spin_unlock(&rpc->crpc_lock);
 985                 }
 986
 987                 spin_unlock(&tsi->tsi_lock);
 988         }
 989
 990         return 0;
 991 }
 992
 993 int
 994 sfw_query_batch (sfw_batch_t *tsb, int testidx, srpc_batch_reply_t *reply)
 995 {
 996         sfw_test_instance_t *tsi;
 997
 998         if (testidx < 0)
 999                 return -EINVAL;
1000
1001         if (testidx == 0) {
1002                 reply->bar_active = atomic_read(&tsb->bat_nactive);
1003                 return 0;
1004         }
1005
1006         list_for_each_entry (tsi, &tsb->bat_tests, tsi_list) {
1007                 if (testidx-- > 1)
1008                         continue;
1009
1010                 reply->bar_active = atomic_read(&tsi->tsi_nactive);
1011                 return 0;
1012         }
1013
1014         return -ENOENT;
1015 }
1016
1017 void
1018 sfw_free_pages (srpc_server_rpc_t *rpc)
1019 {
1020         srpc_free_bulk(rpc->srpc_bulk);
1021         rpc->srpc_bulk = NULL;
1022 }
1023
1024 int
1025 sfw_alloc_pages (srpc_server_rpc_t *rpc, int npages, int sink)
1026 {
1027         LASSERT (rpc->srpc_bulk == NULL);
1028         LASSERT (npages > 0 && npages <= LNET_MAX_IOV);
1029
1030         rpc->srpc_bulk = srpc_alloc_bulk(npages, sink);
1031         if (rpc->srpc_bulk == NULL) return -ENOMEM;
1032
1033         return 0;
1034 }
1035
1036 int
1037 sfw_add_test (srpc_server_rpc_t *rpc)
1038 {
1039         sfw_session_t     *sn = sfw_data.fw_session;
1040         srpc_test_reply_t *reply = &rpc->srpc_replymsg.msg_body.tes_reply;
1041         srpc_test_reqst_t *request;
1042         int                rc;
1043         sfw_batch_t       *bat;
1044
1045         request = &rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst;
1046         reply->tsr_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
1047
1048         if (request->tsr_loop == 0 ||
1049             request->tsr_concur == 0 ||
1050             request->tsr_sid.ses_nid == LNET_NID_ANY ||
1051             request->tsr_ndest > SFW_MAX_NDESTS ||
1052             (request->tsr_is_client && request->tsr_ndest == 0) ||
1053             request->tsr_concur > SFW_MAX_CONCUR ||
1054             request->tsr_service > SRPC_SERVICE_MAX_ID ||
1055             request->tsr_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID) {
1056                 reply->tsr_status = EINVAL;
1057                 return 0;
1058         }
1059
1060         if (sn == NULL || !sfw_sid_equal(request->tsr_sid, sn->sn_id) ||
1061             sfw_find_test_case(request->tsr_service) == NULL) {
1062                 reply->tsr_status = ENOENT;
1063                 return 0;
1064         }
1065
1066         bat = sfw_bid2batch(request->tsr_bid);
1067         if (bat == NULL) {
1068                 CERROR ("Dropping RPC (%s) from %s under memory pressure.\n",
1069                         rpc->srpc_service->sv_name,
1070                         libcfs_id2str(rpc->srpc_peer));
1071                 return -ENOMEM;
1072         }
1073
1074         if (sfw_batch_active(bat)) {
1075                 reply->tsr_status = EBUSY;
1076                 return 0;
1077         }
1078
1079         if (request->tsr_is_client && rpc->srpc_bulk == NULL) {
1080                 /* rpc will be resumed later in sfw_bulk_ready */
1081                 return sfw_alloc_pages(rpc,
1082                                        sfw_id_pages(request->tsr_ndest), 1);
1083         }
1084
1085         rc = sfw_add_test_instance(bat, rpc);
1086         CDEBUG (rc == 0 ? D_NET : D_WARNING,
1087                 "%s test: sv %d %s, loop %d, concur %d, ndest %d\n",
1088                 rc == 0 ? "Added" : "Failed to add", request->tsr_service,
1089                 request->tsr_is_client ? "client" : "server",
1090                 request->tsr_loop, request->tsr_concur, request->tsr_ndest);
1091
1092         reply->tsr_status = (rc < 0) ? -rc : rc;
1093         return 0;
1094 }
1095
1096 int
1097 sfw_control_batch (srpc_batch_reqst_t *request, srpc_batch_reply_t *reply)
1098 {
1099         sfw_session_t *sn = sfw_data.fw_session;
1100         int            rc = 0;
1101         sfw_batch_t   *bat;
1102
1103         reply->bar_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
1104
1105         if (sn == NULL || !sfw_sid_equal(request->bar_sid, sn->sn_id)) {
1106                 reply->bar_status = ESRCH;
1107                 return 0;
1108         }
1109
1110         bat = sfw_find_batch(request->bar_bid);
1111         if (bat == NULL) {
1112                 reply->bar_status = ENOENT;
1113                 return 0;
1114         }
1115
1116         switch (request->bar_opc) {
1117         case SRPC_BATCH_OPC_RUN:
1118                 rc = sfw_run_batch(bat);
1119                 break;
1120
1121         case SRPC_BATCH_OPC_STOP:
1122                 rc = sfw_stop_batch(bat, request->bar_arg);
1123                 break;
1124
1125         case SRPC_BATCH_OPC_QUERY:
1126                 rc = sfw_query_batch(bat, request->bar_testidx, reply);
1127                 break;
1128
1129         default:
1130                 return -EINVAL; /* drop it */
1131         }
1132
1133         reply->bar_status = (rc < 0) ? -rc : rc;
1134         return 0;
1135 }
1136
1137 int
1138 sfw_handle_server_rpc (srpc_server_rpc_t *rpc)
1139 {
1140         srpc_service_t *sv = rpc->srpc_service;
1141         srpc_msg_t     *reply = &rpc->srpc_replymsg;
1142         srpc_msg_t     *request = &rpc->srpc_reqstbuf->buf_msg;
1143         int             rc = 0;
1144
1145         LASSERT (sfw_data.fw_active_srpc == NULL);
1146         LASSERT (sv->sv_id <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1147
1148         spin_lock(&sfw_data.fw_lock);
1149
1150         if (sfw_data.fw_shuttingdown) {
1151                 spin_unlock(&sfw_data.fw_lock);
1152                 return -ESHUTDOWN;
1153         }
1154
1155         /* Remove timer to avoid racing with it or expiring active session */
1156         if (sfw_del_session_timer() != 0) {
1157                 CERROR ("Dropping RPC (%s) from %s: racing with expiry timer.",
1158                         sv->sv_name, libcfs_id2str(rpc->srpc_peer));
1159                 spin_unlock(&sfw_data.fw_lock);
1160                 return -EAGAIN;
1161         }
1162
1163         sfw_data.fw_active_srpc = rpc;
1164         spin_unlock(&sfw_data.fw_lock);
1165
1166         sfw_unpack_message(request);
1167         LASSERT (request->msg_type == srpc_service2request(sv->sv_id));
1168
1169         switch(sv->sv_id) {
1170         default:
1171                 LBUG ();
1172         case SRPC_SERVICE_TEST:
1173                 rc = sfw_add_test(rpc);
1174                 break;
1175
1176         case SRPC_SERVICE_BATCH:
1177                 rc = sfw_control_batch(&request->msg_body.bat_reqst,
1178                                        &reply->msg_body.bat_reply);
1179                 break;
1180
1181         case SRPC_SERVICE_QUERY_STAT:
1182                 rc = sfw_get_stats(&request->msg_body.stat_reqst,
1183                                    &reply->msg_body.stat_reply);
1184                 break;
1185
1186         case SRPC_SERVICE_DEBUG:
1187                 rc = sfw_debug_session(&request->msg_body.dbg_reqst,
1188                                        &reply->msg_body.dbg_reply);
1189                 break;
1190
1191         case SRPC_SERVICE_MAKE_SESSION:
1192                 rc = sfw_make_session(&request->msg_body.mksn_reqst,
1193                                       &reply->msg_body.mksn_reply);
1194                 break;
1195
1196         case SRPC_SERVICE_REMOVE_SESSION:
1197                 rc = sfw_remove_session(&request->msg_body.rmsn_reqst,
1198                                         &reply->msg_body.rmsn_reply);
1199                 break;
1200         }
1201
1202         rpc->srpc_done = sfw_server_rpc_done;
1203         spin_lock(&sfw_data.fw_lock);
1204
1205 #ifdef __KERNEL__
1206         if (!sfw_data.fw_shuttingdown)
1207                 sfw_add_session_timer();
1208 #else
1209         LASSERT (!sfw_data.fw_shuttingdown);
1210         sfw_add_session_timer();
1211 #endif
1212
1213         sfw_data.fw_active_srpc = NULL;
1214         spin_unlock(&sfw_data.fw_lock);
1215         return rc;
1216 }
1217
1218 int
1219 sfw_bulk_ready (srpc_server_rpc_t *rpc, int status)
1220 {
1221         srpc_service_t *sv = rpc->srpc_service;
1222         int             rc;
1223
1224         LASSERT (rpc->srpc_bulk != NULL);
1225         LASSERT (sv->sv_id == SRPC_SERVICE_TEST);
1226         LASSERT (sfw_data.fw_active_srpc == NULL);
1227         LASSERT (rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst.tsr_is_client);
1228
1229         spin_lock(&sfw_data.fw_lock);
1230
1231         if (status != 0) {
1232                 CERROR ("Bulk transfer failed for RPC: "
1233                         "service %s, peer %s, status %d\n",
1234                         sv->sv_name, libcfs_id2str(rpc->srpc_peer), status);
1235                 spin_unlock(&sfw_data.fw_lock);
1236                 return -EIO;
1237         }
1238
1239         if (sfw_data.fw_shuttingdown) {
1240                 spin_unlock(&sfw_data.fw_lock);
1241                 return -ESHUTDOWN;
1242         }
1243
1244         if (sfw_del_session_timer() != 0) {
1245                 CERROR ("Dropping RPC (%s) from %s: racing with expiry timer",
1246                         sv->sv_name, libcfs_id2str(rpc->srpc_peer));
1247                 spin_unlock(&sfw_data.fw_lock);
1248                 return -EAGAIN;
1249         }
1250
1251         sfw_data.fw_active_srpc = rpc;
1252         spin_unlock(&sfw_data.fw_lock);
1253
1254         rc = sfw_add_test(rpc);
1255
1256         spin_lock(&sfw_data.fw_lock);
1257
1258 #ifdef __KERNEL__
1259         if (!sfw_data.fw_shuttingdown)
1260                 sfw_add_session_timer();
1261 #else
1262         LASSERT (!sfw_data.fw_shuttingdown);
1263         sfw_add_session_timer();
1264 #endif
1265
1266         sfw_data.fw_active_srpc = NULL;
1267         spin_unlock(&sfw_data.fw_lock);
1268         return rc;
1269 }
1270
1271 srpc_client_rpc_t *
1272 sfw_create_rpc (lnet_process_id_t peer, int service,
1273                 int nbulkiov, int bulklen,
1274                 void (*done) (srpc_client_rpc_t *), void *priv)
1275 {
1276         srpc_client_rpc_t *rpc;
1277
1278         spin_lock(&sfw_data.fw_lock);
1279
1280         LASSERT (!sfw_data.fw_shuttingdown);
1281         LASSERT (service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1282
1283         if (nbulkiov == 0 && !list_empty(&sfw_data.fw_zombie_rpcs)) {
1284                 rpc = list_entry(sfw_data.fw_zombie_rpcs.next,
1285                                  srpc_client_rpc_t, crpc_list);
1286                 list_del(&rpc->crpc_list);
1287                 spin_unlock(&sfw_data.fw_lock);
1288
1289                 srpc_init_client_rpc(rpc, peer, service, 0, 0,
1290                                      done, sfw_client_rpc_fini, priv);
1291                 return rpc;
1292         }
1293
1294         spin_unlock(&sfw_data.fw_lock);
1295
1296         rpc = srpc_create_client_rpc(peer, service, nbulkiov, bulklen, done,
1297                                      nbulkiov != 0 ? NULL : sfw_client_rpc_fini,
1298                                      priv);
1299         return rpc;
1300 }
1301
1302 void
1303 sfw_unpack_message (srpc_msg_t *msg)
1304 {
1305         if (msg->msg_magic == SRPC_MSG_MAGIC)
1306                 return; /* no flipping needed */
1307
1308         LASSERT (msg->msg_magic == __swab32(SRPC_MSG_MAGIC));
1309
1310         __swab32s(&msg->msg_type);
1311
1312         if (msg->msg_type == SRPC_MSG_STAT_REQST) {
1313                 srpc_stat_reqst_t *req = &msg->msg_body.stat_reqst;
1314
1315                 __swab32s(&req->str_type);
1316                 __swab64s(&req->str_rpyid);
1317                 sfw_unpack_sid(req->str_sid);
1318                 return;
1319         }
1320
1321         if (msg->msg_type == SRPC_MSG_STAT_REPLY) {
1322                 srpc_stat_reply_t *rep = &msg->msg_body.stat_reply;
1323
1324                 __swab32s(&rep->str_status);
1325                 sfw_unpack_sid(rep->str_sid);
1326                 sfw_unpack_fw_counters(rep->str_fw);
1327                 sfw_unpack_rpc_counters(rep->str_rpc);
1328                 sfw_unpack_lnet_counters(rep->str_lnet);
1329                 return;
1330         }
1331
1332         if (msg->msg_type == SRPC_MSG_MKSN_REQST) {
1333                 srpc_mksn_reqst_t *req = &msg->msg_body.mksn_reqst;
1334
1335                 __swab64s(&req->mksn_rpyid);
1336                 __swab32s(&req->mksn_force);
1337                 sfw_unpack_sid(req->mksn_sid);
1338                 return;
1339         }
1340
1341         if (msg->msg_type == SRPC_MSG_MKSN_REPLY) {
1342                 srpc_mksn_reply_t *rep = &msg->msg_body.mksn_reply;
1343
1344                 __swab32s(&rep->mksn_status);
1345                 __swab32s(&rep->mksn_timeout);
1346                 sfw_unpack_sid(rep->mksn_sid);
1347                 return;
1348         }
1349
1350         if (msg->msg_type == SRPC_MSG_RMSN_REQST) {
1351                 srpc_rmsn_reqst_t *req = &msg->msg_body.rmsn_reqst;
1352
1353                 __swab64s(&req->rmsn_rpyid);
1354                 sfw_unpack_sid(req->rmsn_sid);
1355                 return;
1356         }
1357
1358         if (msg->msg_type == SRPC_MSG_RMSN_REPLY) {
1359                 srpc_rmsn_reply_t *rep = &msg->msg_body.rmsn_reply;
1360
1361                 __swab32s(&rep->rmsn_status);
1362                 sfw_unpack_sid(rep->rmsn_sid);
1363                 return;
1364         }
1365
1366         if (msg->msg_type == SRPC_MSG_DEBUG_REQST) {
1367                 srpc_debug_reqst_t *req = &msg->msg_body.dbg_reqst;
1368
1369                 __swab64s(&req->dbg_rpyid);
1370                 __swab32s(&req->dbg_flags);
1371                 sfw_unpack_sid(req->dbg_sid);
1372                 return;
1373         }
1374
1375         if (msg->msg_type == SRPC_MSG_DEBUG_REPLY) {
1376                 srpc_debug_reply_t *rep = &msg->msg_body.dbg_reply;
1377
1378                 __swab32s(&rep->dbg_nbatch);
1379                 __swab32s(&rep->dbg_timeout);
1380                 sfw_unpack_sid(rep->dbg_sid);
1381                 return;
1382         }
1383
1384         if (msg->msg_type == SRPC_MSG_BATCH_REQST) {
1385                 srpc_batch_reqst_t *req = &msg->msg_body.bat_reqst;
1386
1387                 __swab32s(&req->bar_opc);
1388                 __swab64s(&req->bar_rpyid);
1389                 __swab32s(&req->bar_testidx);
1390                 __swab32s(&req->bar_arg);
1391                 sfw_unpack_sid(req->bar_sid);
1392                 __swab64s(&req->bar_bid.bat_id);
1393                 return;
1394         }
1395
1396         if (msg->msg_type == SRPC_MSG_BATCH_REPLY) {
1397                 srpc_batch_reply_t *rep = &msg->msg_body.bat_reply;
1398
1399                 __swab32s(&rep->bar_status);
1400                 sfw_unpack_sid(rep->bar_sid);
1401                 return;
1402         }
1403
1404         if (msg->msg_type == SRPC_MSG_TEST_REQST) {
1405                 srpc_test_reqst_t *req = &msg->msg_body.tes_reqst;
1406
1407                 __swab64s(&req->tsr_rpyid);
1408                 __swab64s(&req->tsr_bulkid);
1409                 __swab32s(&req->tsr_loop);
1410                 __swab32s(&req->tsr_ndest);
1411                 __swab32s(&req->tsr_concur);
1412                 __swab32s(&req->tsr_service);
1413                 sfw_unpack_sid(req->tsr_sid);
1414                 __swab64s(&req->tsr_bid.bat_id);
1415                 return;
1416         }
1417
1418         if (msg->msg_type == SRPC_MSG_TEST_REPLY) {
1419                 srpc_test_reply_t *rep = &msg->msg_body.tes_reply;
1420
1421                 __swab32s(&rep->tsr_status);
1422                 sfw_unpack_sid(rep->tsr_sid);
1423                 return;
1424         }
1425
1426         if (msg->msg_type == SRPC_MSG_JOIN_REQST) {
1427                 srpc_join_reqst_t *req = &msg->msg_body.join_reqst;
1428
1429                 __swab64s(&req->join_rpyid);
1430                 sfw_unpack_sid(req->join_sid);
1431                 return;
1432         }
1433
1434         if (msg->msg_type == SRPC_MSG_JOIN_REPLY) {
1435                 srpc_join_reply_t *rep = &msg->msg_body.join_reply;
1436
1437                 __swab32s(&rep->join_status);
1438                 __swab32s(&rep->join_timeout);
1439                 sfw_unpack_sid(rep->join_sid);
1440                 return;
1441         }
1442
1443         LBUG ();
1444         return;
1445 }
1446
1447 void
1448 sfw_abort_rpc (srpc_client_rpc_t *rpc)
1449 {
1450         LASSERT (atomic_read(&rpc->crpc_refcount) > 0);
1451         LASSERT (rpc->crpc_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1452
1453         spin_lock(&rpc->crpc_lock);
1454         srpc_abort_rpc(rpc, -EINTR);
1455         spin_unlock(&rpc->crpc_lock);
1456         return;
1457 }
1458
1459 void
1460 sfw_post_rpc (srpc_client_rpc_t *rpc)
1461 {
1462         spin_lock(&rpc->crpc_lock);
1463
1464         LASSERT (!rpc->crpc_closed);
1465         LASSERT (!rpc->crpc_aborted);
1466         LASSERT (list_empty(&rpc->crpc_list));
1467         LASSERT (!sfw_data.fw_shuttingdown);
1468
1469         rpc->crpc_timeout = SFW_CLIENT_RPC_TIMEOUT;
1470         srpc_post_rpc(rpc);
1471
1472         spin_unlock(&rpc->crpc_lock);
1473         return;
1474 }
1475
1476 static srpc_service_t sfw_services[] =
1477 {
1478         {
1479                 .sv_name = "debug",
1480                 .sv_id   = SRPC_SERVICE_DEBUG,
1481         },
1482         {
1483                 .sv_name = "query stats",
1484                 .sv_id   = SRPC_SERVICE_QUERY_STAT,
1485         },
1486         {
1487                 .sv_name = "make sessin",
1488                 .sv_id   = SRPC_SERVICE_MAKE_SESSION,
1489         },
1490         {
1491                 .sv_name = "remove session",
1492                 .sv_id   = SRPC_SERVICE_REMOVE_SESSION,
1493         },
1494         {
1495                 .sv_name = "batch service",
1496                 .sv_id   = SRPC_SERVICE_BATCH,
1497         },
1498         {
1499                 .sv_name = "test service",
1500                 .sv_id   = SRPC_SERVICE_TEST,
1501         },
1502         {       .sv_name = NULL, }
1503 };
1504
1505 extern sfw_test_client_ops_t ping_test_client;
1506 extern srpc_service_t        ping_test_service;
1507
1508 extern sfw_test_client_ops_t brw_test_client;
1509 extern srpc_service_t        brw_test_service;
1510
1511 int
1512 sfw_startup (void)
1513 {
1514         int              i;
1515         int              rc;
1516         int              error;
1517         srpc_service_t  *sv;
1518         sfw_test_case_t *tsc;
1519
1520         if (session_timeout < 0) {
1521                 CERROR ("Session timeout must be non-negative: %d\n",
1522                         session_timeout);
1523                 return -EINVAL;
1524         }
1525
1526         if (session_timeout == 0)
1527                 CWARN ("Zero session_timeout specified "
1528                        "- test sessions never timeout.\n");
1529
1530         memset(&sfw_data, 0, sizeof(struct smoketest_framework));
1531
1532         sfw_data.fw_session     = NULL;
1533         sfw_data.fw_active_srpc = NULL;
1534         spin_lock_init(&sfw_data.fw_lock);
1535         atomic_set(&sfw_data.fw_nzombies, 0);
1536         CFS_INIT_LIST_HEAD(&sfw_data.fw_tests);
1537         CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_rpcs);
1538         CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_sessions);
1539
1540         rc = sfw_register_test(&brw_test_service, &brw_test_client);
1541         LASSERT (rc == 0);
1542         rc = sfw_register_test(&ping_test_service, &ping_test_client);
1543         LASSERT (rc == 0);
1544
1545         error = 0;
1546         list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
1547                 sv = tsc->tsc_srv_service;
1548                 sv->sv_concur = SFW_TEST_CONCURRENCY;
1549
1550                 rc = srpc_add_service(sv);
1551                 LASSERT (rc != -EBUSY);
1552                 if (rc != 0) {
1553                         CWARN ("Failed to add %s service: %d\n",
1554                                sv->sv_name, rc);
1555                         error = rc;
1556                 }
1557         }
1558
1559         for (i = 0; ; i++) {
1560                 sv = &sfw_services[i];
1561                 if (sv->sv_name == NULL) break;
1562
1563                 sv->sv_bulk_ready = NULL;
1564                 sv->sv_handler    = sfw_handle_server_rpc;
1565                 sv->sv_concur     = SFW_SERVICE_CONCURRENCY;
1566                 if (sv->sv_id == SRPC_SERVICE_TEST)
1567                         sv->sv_bulk_ready = sfw_bulk_ready;
1568
1569                 rc = srpc_add_service(sv);
1570                 LASSERT (rc != -EBUSY);
1571                 if (rc != 0) {
1572                         CWARN ("Failed to add %s service: %d\n",
1573                                sv->sv_name, rc);
1574                         error = rc;
1575                 }
1576
1577                 /* about to sfw_shutdown, no need to add buffer */
1578                 if (error) continue;
1579
1580                 rc = srpc_service_add_buffers(sv, SFW_POST_BUFFERS);
1581                 if (rc != SFW_POST_BUFFERS) {
1582                         CWARN ("Failed to reserve enough buffers: "
1583                                "service %s, %d needed, %d reserved\n",
1584                                sv->sv_name, SFW_POST_BUFFERS, rc);
1585                         error = -ENOMEM;
1586                 }
1587         }
1588
1589         if (error != 0)
1590                 sfw_shutdown();
1591         return error;
1592 }
1593
1594 void
1595 sfw_shutdown (void)
1596 {
1597         srpc_service_t  *sv;
1598         sfw_test_case_t *tsc;
1599         int              i;
1600
1601         spin_lock(&sfw_data.fw_lock);
1602
1603         sfw_data.fw_shuttingdown = 1;
1604 #ifdef __KERNEL__
1605         lst_wait_until(sfw_data.fw_active_srpc == NULL, sfw_data.fw_lock,
1606                        "waiting for active RPC to finish.\n");
1607 #else
1608         LASSERT (sfw_data.fw_active_srpc == NULL);
1609 #endif
1610
1611         if (sfw_del_session_timer() != 0)
1612                 lst_wait_until(sfw_data.fw_session == NULL, sfw_data.fw_lock,
1613                                "waiting for session timer to explode.\n");
1614
1615         sfw_deactivate_session();
1616         lst_wait_until(atomic_read(&sfw_data.fw_nzombies) == 0,
1617                        sfw_data.fw_lock,
1618                        "waiting for %d zombie sessions to die.\n",
1619                        atomic_read(&sfw_data.fw_nzombies));
1620
1621         spin_unlock(&sfw_data.fw_lock);
1622
1623         for (i = 0; ; i++) {
1624                 sv = &sfw_services[i];
1625                 if (sv->sv_name == NULL)
1626                         break;
1627
1628                 srpc_shutdown_service(sv);
1629                 srpc_remove_service(sv);
1630         }
1631
1632         list_for_each_entry (tsc, &sfw_data.fw_tests, tsc_list) {
1633                 sv = tsc->tsc_srv_service;
1634                 srpc_shutdown_service(sv);
1635                 srpc_remove_service(sv);
1636         }
1637
1638         while (!list_empty(&sfw_data.fw_zombie_rpcs)) {
1639                 srpc_client_rpc_t *rpc;
1640
1641                 rpc = list_entry(sfw_data.fw_zombie_rpcs.next,
1642                                  srpc_client_rpc_t, crpc_list);
1643                 list_del(&rpc->crpc_list);
1644
1645                 LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
1646         }
1647
1648         for (i = 0; ; i++) {
1649                 sv = &sfw_services[i];
1650                 if (sv->sv_name == NULL)
1651                         break;
1652
1653                 srpc_wait_service_shutdown(sv);
1654         }
1655
1656         while (!list_empty(&sfw_data.fw_tests)) {
1657                 tsc = list_entry(sfw_data.fw_tests.next,
1658                                  sfw_test_case_t, tsc_list);
1659
1660                 srpc_wait_service_shutdown(tsc->tsc_srv_service);
1661
1662                 list_del(&tsc->tsc_list);
1663                 LIBCFS_FREE(tsc, sizeof(*tsc));
1664         }
1665
1666         return;
1667 }