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