lnet/selftest/framework.c

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