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 (c) 2007, 2010, Oracle and/or its affiliates. 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_packed_t *dests;
 765                 lnet_process_id_packed_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.nid = id.nid;
 788                         tsu->tsu_dest.pid = id.pid;
 789                         tsu->tsu_instance = tsi;
 790                         tsu->tsu_private  = NULL;
 791                         cfs_list_add_tail(&tsu->tsu_list, &tsi->tsi_units);
 792                 }
 793         }
 794
 795         rc = tsi->tsi_ops->tso_init(tsi);
 796         if (rc == 0) {
 797                 cfs_list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
 798                 return 0;
 799         }
 800
 801 error:
 802         LASSERT (rc != 0);
 803         sfw_destroy_test_instance(tsi);
 804         return rc;
 805 }
 806
 807 static void
 808 sfw_test_unit_done (sfw_test_unit_t *tsu)
 809 {
 810         sfw_test_instance_t *tsi = tsu->tsu_instance;
 811         sfw_batch_t         *tsb = tsi->tsi_batch;
 812         sfw_session_t       *sn = tsb->bat_session;
 813
 814         LASSERT (sfw_test_active(tsi));
 815
 816         if (!cfs_atomic_dec_and_test(&tsi->tsi_nactive))
 817                 return;
 818
 819         /* the test instance is done */
 820         cfs_spin_lock(&tsi->tsi_lock);
 821
 822         tsi->tsi_stopping = 0;
 823
 824         cfs_spin_unlock(&tsi->tsi_lock);
 825
 826         cfs_spin_lock(&sfw_data.fw_lock);
 827
 828         if (!cfs_atomic_dec_and_test(&tsb->bat_nactive) ||/* tsb still active */
 829             sn == sfw_data.fw_session) {               /* sn also active */
 830                 cfs_spin_unlock(&sfw_data.fw_lock);
 831                 return;
 832         }
 833
 834         LASSERT (!cfs_list_empty(&sn->sn_list)); /* I'm a zombie! */
 835
 836         cfs_list_for_each_entry_typed (tsb, &sn->sn_batches,
 837                                        sfw_batch_t, bat_list) {
 838                 if (sfw_batch_active(tsb)) {
 839                         cfs_spin_unlock(&sfw_data.fw_lock);
 840                         return;
 841                 }
 842         }
 843
 844         cfs_list_del_init(&sn->sn_list);
 845         cfs_spin_unlock(&sfw_data.fw_lock);
 846
 847         sfw_destroy_session(sn);
 848         return;
 849 }
 850
 851 void
 852 sfw_test_rpc_done (srpc_client_rpc_t *rpc)
 853 {
 854         sfw_test_unit_t     *tsu = rpc->crpc_priv;
 855         sfw_test_instance_t *tsi = tsu->tsu_instance;
 856         int                  done = 0;
 857
 858         tsi->tsi_ops->tso_done_rpc(tsu, rpc);
 859
 860         cfs_spin_lock(&tsi->tsi_lock);
 861
 862         LASSERT (sfw_test_active(tsi));
 863         LASSERT (!cfs_list_empty(&rpc->crpc_list));
 864
 865         cfs_list_del_init(&rpc->crpc_list);
 866
 867         /* batch is stopping or loop is done or get error */
 868         if (tsi->tsi_stopping ||
 869             tsu->tsu_loop == 0 ||
 870             (rpc->crpc_status != 0 && tsi->tsi_stoptsu_onerr))
 871                 done = 1;
 872
 873         /* dec ref for poster */
 874         srpc_client_rpc_decref(rpc);
 875
 876         cfs_spin_unlock(&tsi->tsi_lock);
 877
 878         if (!done) {
 879                 swi_schedule_workitem(&tsu->tsu_worker);
 880                 return;
 881         }
 882
 883         sfw_test_unit_done(tsu);
 884         return;
 885 }
 886
 887 int
 888 sfw_create_test_rpc (sfw_test_unit_t *tsu, lnet_process_id_t peer,
 889                      int nblk, int blklen, srpc_client_rpc_t **rpcpp)
 890 {
 891         srpc_client_rpc_t   *rpc = NULL;
 892         sfw_test_instance_t *tsi = tsu->tsu_instance;
 893
 894         cfs_spin_lock(&tsi->tsi_lock);
 895
 896         LASSERT (sfw_test_active(tsi));
 897
 898         if (!cfs_list_empty(&tsi->tsi_free_rpcs)) {
 899                 /* pick request from buffer */
 900                 rpc = cfs_list_entry(tsi->tsi_free_rpcs.next,
 901                                      srpc_client_rpc_t, crpc_list);
 902                 LASSERT (nblk == rpc->crpc_bulk.bk_niov);
 903                 cfs_list_del_init(&rpc->crpc_list);
 904
 905                 srpc_init_client_rpc(rpc, peer, tsi->tsi_service, nblk,
 906                                      blklen, sfw_test_rpc_done,
 907                                      sfw_test_rpc_fini, tsu);
 908         }
 909
 910         cfs_spin_unlock(&tsi->tsi_lock);
 911
 912         if (rpc == NULL)
 913                 rpc = srpc_create_client_rpc(peer, tsi->tsi_service, nblk,
 914                                              blklen, sfw_test_rpc_done,
 915                                              sfw_test_rpc_fini, tsu);
 916         if (rpc == NULL) {
 917                 CERROR ("Can't create rpc for test %d\n", tsi->tsi_service);
 918                 return -ENOMEM;
 919         }
 920
 921         *rpcpp = rpc;
 922         return 0;
 923 }
 924
 925 int
 926 sfw_run_test (swi_workitem_t *wi)
 927 {
 928         sfw_test_unit_t     *tsu = wi->wi_data;
 929         sfw_test_instance_t *tsi = tsu->tsu_instance;
 930         srpc_client_rpc_t   *rpc = NULL;
 931
 932         LASSERT (wi == &tsu->tsu_worker);
 933
 934         if (tsi->tsi_ops->tso_prep_rpc(tsu, tsu->tsu_dest, &rpc) != 0) {
 935                 LASSERT (rpc == NULL);
 936                 goto test_done;
 937         }
 938
 939         LASSERT (rpc != NULL);
 940
 941         cfs_spin_lock(&tsi->tsi_lock);
 942
 943         if (tsi->tsi_stopping) {
 944                 cfs_list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
 945                 cfs_spin_unlock(&tsi->tsi_lock);
 946                 goto test_done;
 947         }
 948
 949         if (tsu->tsu_loop > 0)
 950                 tsu->tsu_loop--;
 951
 952         cfs_list_add_tail(&rpc->crpc_list, &tsi->tsi_active_rpcs);
 953         cfs_spin_unlock(&tsi->tsi_lock);
 954
 955         rpc->crpc_timeout = rpc_timeout;
 956
 957         cfs_spin_lock(&rpc->crpc_lock);
 958         srpc_post_rpc(rpc);
 959         cfs_spin_unlock(&rpc->crpc_lock);
 960         return 0;
 961
 962 test_done:
 963         /*
 964          * No one can schedule me now since:
 965          * - previous RPC, if any, has done and
 966          * - no new RPC is initiated.
 967          * - my batch is still active; no one can run it again now.
 968          * Cancel pending schedules and prevent future schedule attempts:
 969          */
 970         swi_kill_workitem(wi);
 971         sfw_test_unit_done(tsu);
 972         return 1;
 973 }
 974
 975 int
 976 sfw_run_batch (sfw_batch_t *tsb)
 977 {
 978         swi_workitem_t      *wi;
 979         sfw_test_unit_t     *tsu;
 980         sfw_test_instance_t *tsi;
 981
 982         if (sfw_batch_active(tsb)) {
 983                 CDEBUG(D_NET, "Batch already active: "LPU64" (%d)\n",
 984                        tsb->bat_id.bat_id, cfs_atomic_read(&tsb->bat_nactive));
 985                 return 0;
 986         }
 987
 988         cfs_list_for_each_entry_typed (tsi, &tsb->bat_tests,
 989                                        sfw_test_instance_t, tsi_list) {
 990                 if (!tsi->tsi_is_client) /* skip server instances */
 991                         continue;
 992
 993                 LASSERT (!tsi->tsi_stopping);
 994                 LASSERT (!sfw_test_active(tsi));
 995
 996                 cfs_atomic_inc(&tsb->bat_nactive);
 997
 998                 cfs_list_for_each_entry_typed (tsu, &tsi->tsi_units,
 999                                                sfw_test_unit_t, tsu_list) {
1000                         cfs_atomic_inc(&tsi->tsi_nactive);
1001                         tsu->tsu_loop = tsi->tsi_loop;
1002                         wi = &tsu->tsu_worker;
1003                         swi_init_workitem(wi, tsu, sfw_run_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 }