lnet/selftest/framework.c

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