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         rpc->crpc_timeout = rpc_timeout;
1009
1010         spin_lock(&rpc->crpc_lock);
1011         srpc_post_rpc(rpc);
1012         spin_unlock(&rpc->crpc_lock);
1013         return 0;
1014
1015 test_done:
1016         /*
1017          * No one can schedule me now since:
1018          * - previous RPC, if any, has done and
1019          * - no new RPC is initiated.
1020          * - my batch is still active; no one can run it again now.
1021          * Cancel pending schedules and prevent future schedule attempts:
1022          */
1023         swi_exit_workitem(wi);
1024         sfw_test_unit_done(tsu);
1025         return 1;
1026 }
1027
1028 int
1029 sfw_run_batch (sfw_batch_t *tsb)
1030 {
1031         swi_workitem_t      *wi;
1032         sfw_test_unit_t     *tsu;
1033         sfw_test_instance_t *tsi;
1034
1035         if (sfw_batch_active(tsb)) {
1036                 CDEBUG(D_NET, "Batch already active: "LPU64" (%d)\n",
1037                        tsb->bat_id.bat_id, cfs_atomic_read(&tsb->bat_nactive));
1038                 return 0;
1039         }
1040
1041         cfs_list_for_each_entry_typed (tsi, &tsb->bat_tests,
1042                                        sfw_test_instance_t, tsi_list) {
1043                 if (!tsi->tsi_is_client) /* skip server instances */
1044                         continue;
1045
1046                 LASSERT (!tsi->tsi_stopping);
1047                 LASSERT (!sfw_test_active(tsi));
1048
1049                 cfs_atomic_inc(&tsb->bat_nactive);
1050
1051                 cfs_list_for_each_entry_typed (tsu, &tsi->tsi_units,
1052                                                sfw_test_unit_t, tsu_list) {
1053                         cfs_atomic_inc(&tsi->tsi_nactive);
1054                         tsu->tsu_loop = tsi->tsi_loop;
1055                         wi = &tsu->tsu_worker;
1056                         swi_init_workitem(wi, tsu, sfw_run_test,
1057                                           lst_sched_test[\
1058                                           lnet_cpt_of_nid(tsu->tsu_dest.nid)]);
1059                         swi_schedule_workitem(wi);
1060                 }
1061         }
1062
1063         return 0;
1064 }
1065
1066 int
1067 sfw_stop_batch (sfw_batch_t *tsb, int force)
1068 {
1069         sfw_test_instance_t *tsi;
1070         srpc_client_rpc_t   *rpc;
1071
1072         if (!sfw_batch_active(tsb)) {
1073                 CDEBUG(D_NET, "Batch "LPU64" inactive\n", tsb->bat_id.bat_id);
1074                 return 0;
1075         }
1076
1077         cfs_list_for_each_entry_typed (tsi, &tsb->bat_tests,
1078                                        sfw_test_instance_t, tsi_list) {
1079                 spin_lock(&tsi->tsi_lock);
1080
1081                 if (!tsi->tsi_is_client ||
1082                     !sfw_test_active(tsi) || tsi->tsi_stopping) {
1083                         spin_unlock(&tsi->tsi_lock);
1084                         continue;
1085                 }
1086
1087                 tsi->tsi_stopping = 1;
1088
1089                 if (!force) {
1090                         spin_unlock(&tsi->tsi_lock);
1091                         continue;
1092                 }
1093
1094                 /* abort launched rpcs in the test */
1095                 cfs_list_for_each_entry_typed(rpc, &tsi->tsi_active_rpcs,
1096                                               srpc_client_rpc_t, crpc_list) {
1097                         spin_lock(&rpc->crpc_lock);
1098
1099                         srpc_abort_rpc(rpc, -EINTR);
1100
1101                         spin_unlock(&rpc->crpc_lock);
1102                 }
1103
1104                 spin_unlock(&tsi->tsi_lock);
1105         }
1106
1107         return 0;
1108 }
1109
1110 int
1111 sfw_query_batch (sfw_batch_t *tsb, int testidx, srpc_batch_reply_t *reply)
1112 {
1113         sfw_test_instance_t *tsi;
1114
1115         if (testidx < 0)
1116                 return -EINVAL;
1117
1118         if (testidx == 0) {
1119                 reply->bar_active = cfs_atomic_read(&tsb->bat_nactive);
1120                 return 0;
1121         }
1122
1123         cfs_list_for_each_entry_typed (tsi, &tsb->bat_tests,
1124                                        sfw_test_instance_t, tsi_list) {
1125                 if (testidx-- > 1)
1126                         continue;
1127
1128                 reply->bar_active = cfs_atomic_read(&tsi->tsi_nactive);
1129                 return 0;
1130         }
1131
1132         return -ENOENT;
1133 }
1134
1135 void
1136 sfw_free_pages (srpc_server_rpc_t *rpc)
1137 {
1138         srpc_free_bulk(rpc->srpc_bulk);
1139         rpc->srpc_bulk = NULL;
1140 }
1141
1142 int
1143 sfw_alloc_pages(struct srpc_server_rpc *rpc, int cpt, int npages, int len,
1144                 int sink)
1145 {
1146         LASSERT(rpc->srpc_bulk == NULL);
1147         LASSERT(npages > 0 && npages <= LNET_MAX_IOV);
1148
1149         rpc->srpc_bulk = srpc_alloc_bulk(cpt, npages, len, sink);
1150         if (rpc->srpc_bulk == NULL)
1151                 return -ENOMEM;
1152
1153         return 0;
1154 }
1155
1156 int
1157 sfw_add_test (srpc_server_rpc_t *rpc)
1158 {
1159         sfw_session_t     *sn = sfw_data.fw_session;
1160         srpc_test_reply_t *reply = &rpc->srpc_replymsg.msg_body.tes_reply;
1161         srpc_test_reqst_t *request;
1162         int                rc;
1163         sfw_batch_t       *bat;
1164
1165         request = &rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst;
1166         reply->tsr_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
1167
1168         if (request->tsr_loop == 0 ||
1169             request->tsr_concur == 0 ||
1170             request->tsr_sid.ses_nid == LNET_NID_ANY ||
1171             request->tsr_ndest > SFW_MAX_NDESTS ||
1172             (request->tsr_is_client && request->tsr_ndest == 0) ||
1173             request->tsr_concur > SFW_MAX_CONCUR ||
1174             request->tsr_service > SRPC_SERVICE_MAX_ID ||
1175             request->tsr_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID) {
1176                 reply->tsr_status = EINVAL;
1177                 return 0;
1178         }
1179
1180         if (sn == NULL || !sfw_sid_equal(request->tsr_sid, sn->sn_id) ||
1181             sfw_find_test_case(request->tsr_service) == NULL) {
1182                 reply->tsr_status = ENOENT;
1183                 return 0;
1184         }
1185
1186         bat = sfw_bid2batch(request->tsr_bid);
1187         if (bat == NULL) {
1188                 CERROR ("Dropping RPC (%s) from %s under memory pressure.\n",
1189                         rpc->srpc_scd->scd_svc->sv_name,
1190                         libcfs_id2str(rpc->srpc_peer));
1191                 return -ENOMEM;
1192         }
1193
1194         if (sfw_batch_active(bat)) {
1195                 reply->tsr_status = EBUSY;
1196                 return 0;
1197         }
1198
1199         if (request->tsr_is_client && rpc->srpc_bulk == NULL) {
1200                 /* rpc will be resumed later in sfw_bulk_ready */
1201                 int     npg = sfw_id_pages(request->tsr_ndest);
1202                 int     len;
1203
1204                 if ((sn->sn_features & LST_FEAT_BULK_LEN) == 0) {
1205                         len = npg * PAGE_CACHE_SIZE;
1206
1207                 } else  {
1208                         len = sizeof(lnet_process_id_packed_t) *
1209                               request->tsr_ndest;
1210                 }
1211
1212                 return sfw_alloc_pages(rpc, CFS_CPT_ANY, npg, len, 1);
1213         }
1214
1215         rc = sfw_add_test_instance(bat, rpc);
1216         CDEBUG (rc == 0 ? D_NET : D_WARNING,
1217                 "%s test: sv %d %s, loop %d, concur %d, ndest %d\n",
1218                 rc == 0 ? "Added" : "Failed to add", request->tsr_service,
1219                 request->tsr_is_client ? "client" : "server",
1220                 request->tsr_loop, request->tsr_concur, request->tsr_ndest);
1221
1222         reply->tsr_status = (rc < 0) ? -rc : rc;
1223         return 0;
1224 }
1225
1226 int
1227 sfw_control_batch (srpc_batch_reqst_t *request, srpc_batch_reply_t *reply)
1228 {
1229         sfw_session_t *sn = sfw_data.fw_session;
1230         int            rc = 0;
1231         sfw_batch_t   *bat;
1232
1233         reply->bar_sid = (sn == NULL) ? LST_INVALID_SID : sn->sn_id;
1234
1235         if (sn == NULL || !sfw_sid_equal(request->bar_sid, sn->sn_id)) {
1236                 reply->bar_status = ESRCH;
1237                 return 0;
1238         }
1239
1240         bat = sfw_find_batch(request->bar_bid);
1241         if (bat == NULL) {
1242                 reply->bar_status = ENOENT;
1243                 return 0;
1244         }
1245
1246         switch (request->bar_opc) {
1247         case SRPC_BATCH_OPC_RUN:
1248                 rc = sfw_run_batch(bat);
1249                 break;
1250
1251         case SRPC_BATCH_OPC_STOP:
1252                 rc = sfw_stop_batch(bat, request->bar_arg);
1253                 break;
1254
1255         case SRPC_BATCH_OPC_QUERY:
1256                 rc = sfw_query_batch(bat, request->bar_testidx, reply);
1257                 break;
1258
1259         default:
1260                 return -EINVAL; /* drop it */
1261         }
1262
1263         reply->bar_status = (rc < 0) ? -rc : rc;
1264         return 0;
1265 }
1266
1267 int
1268 sfw_handle_server_rpc(struct srpc_server_rpc *rpc)
1269 {
1270         struct srpc_service     *sv = rpc->srpc_scd->scd_svc;
1271         srpc_msg_t     *reply   = &rpc->srpc_replymsg;
1272         srpc_msg_t     *request = &rpc->srpc_reqstbuf->buf_msg;
1273         unsigned        features = LST_FEATS_MASK;
1274         int             rc = 0;
1275
1276         LASSERT(sfw_data.fw_active_srpc == NULL);
1277         LASSERT(sv->sv_id <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1278
1279         spin_lock(&sfw_data.fw_lock);
1280
1281         if (sfw_data.fw_shuttingdown) {
1282                 spin_unlock(&sfw_data.fw_lock);
1283                 return -ESHUTDOWN;
1284         }
1285
1286         /* Remove timer to avoid racing with it or expiring active session */
1287         if (sfw_del_session_timer() != 0) {
1288                 CERROR("Dropping RPC (%s) from %s: racing with expiry timer.",
1289                        sv->sv_name, libcfs_id2str(rpc->srpc_peer));
1290                 spin_unlock(&sfw_data.fw_lock);
1291                 return -EAGAIN;
1292         }
1293
1294         sfw_data.fw_active_srpc = rpc;
1295         spin_unlock(&sfw_data.fw_lock);
1296
1297         sfw_unpack_message(request);
1298         LASSERT(request->msg_type == srpc_service2request(sv->sv_id));
1299
1300         /* rpc module should have checked this */
1301         LASSERT(request->msg_version == SRPC_MSG_VERSION);
1302
1303         if (sv->sv_id != SRPC_SERVICE_MAKE_SESSION &&
1304             sv->sv_id != SRPC_SERVICE_DEBUG) {
1305                 sfw_session_t *sn = sfw_data.fw_session;
1306
1307                 if (sn != NULL &&
1308                     sn->sn_features != request->msg_ses_feats) {
1309                         CNETERR("Features of framework RPC don't match "
1310                                 "features of current session: %x/%x\n",
1311                                 request->msg_ses_feats, sn->sn_features);
1312                         reply->msg_body.reply.status = EPROTO;
1313                         reply->msg_body.reply.sid    = sn->sn_id;
1314                         goto out;
1315                 }
1316
1317         } else if ((request->msg_ses_feats & ~LST_FEATS_MASK) != 0) {
1318                 /* NB: at this point, old version will ignore features and
1319                  * create new session anyway, so console should be able
1320                  * to handle this */
1321                 reply->msg_body.reply.status = EPROTO;
1322                 goto out;
1323         }
1324
1325         switch(sv->sv_id) {
1326         default:
1327                 LBUG ();
1328         case SRPC_SERVICE_TEST:
1329                 rc = sfw_add_test(rpc);
1330                 break;
1331
1332         case SRPC_SERVICE_BATCH:
1333                 rc = sfw_control_batch(&request->msg_body.bat_reqst,
1334                                        &reply->msg_body.bat_reply);
1335                 break;
1336
1337         case SRPC_SERVICE_QUERY_STAT:
1338                 rc = sfw_get_stats(&request->msg_body.stat_reqst,
1339                                    &reply->msg_body.stat_reply);
1340                 break;
1341
1342         case SRPC_SERVICE_DEBUG:
1343                 rc = sfw_debug_session(&request->msg_body.dbg_reqst,
1344                                        &reply->msg_body.dbg_reply);
1345                 break;
1346
1347         case SRPC_SERVICE_MAKE_SESSION:
1348                 rc = sfw_make_session(&request->msg_body.mksn_reqst,
1349                                       &reply->msg_body.mksn_reply);
1350                 break;
1351
1352         case SRPC_SERVICE_REMOVE_SESSION:
1353                 rc = sfw_remove_session(&request->msg_body.rmsn_reqst,
1354                                         &reply->msg_body.rmsn_reply);
1355                 break;
1356         }
1357
1358         if (sfw_data.fw_session != NULL)
1359                 features = sfw_data.fw_session->sn_features;
1360  out:
1361         reply->msg_ses_feats = features;
1362         rpc->srpc_done = sfw_server_rpc_done;
1363         spin_lock(&sfw_data.fw_lock);
1364
1365 #ifdef __KERNEL__
1366         if (!sfw_data.fw_shuttingdown)
1367                 sfw_add_session_timer();
1368 #else
1369         LASSERT(!sfw_data.fw_shuttingdown);
1370         sfw_add_session_timer();
1371 #endif
1372
1373         sfw_data.fw_active_srpc = NULL;
1374         spin_unlock(&sfw_data.fw_lock);
1375         return rc;
1376 }
1377
1378 int
1379 sfw_bulk_ready(struct srpc_server_rpc *rpc, int status)
1380 {
1381         struct srpc_service     *sv = rpc->srpc_scd->scd_svc;
1382         int                     rc;
1383
1384         LASSERT(rpc->srpc_bulk != NULL);
1385         LASSERT(sv->sv_id == SRPC_SERVICE_TEST);
1386         LASSERT(sfw_data.fw_active_srpc == NULL);
1387         LASSERT(rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst.tsr_is_client);
1388
1389         spin_lock(&sfw_data.fw_lock);
1390
1391         if (status != 0) {
1392                 CERROR("Bulk transfer failed for RPC: "
1393                        "service %s, peer %s, status %d\n",
1394                        sv->sv_name, libcfs_id2str(rpc->srpc_peer), status);
1395                 spin_unlock(&sfw_data.fw_lock);
1396                 return -EIO;
1397         }
1398
1399         if (sfw_data.fw_shuttingdown) {
1400                 spin_unlock(&sfw_data.fw_lock);
1401                 return -ESHUTDOWN;
1402         }
1403
1404         if (sfw_del_session_timer() != 0) {
1405                 CERROR("Dropping RPC (%s) from %s: racing with expiry timer",
1406                        sv->sv_name, libcfs_id2str(rpc->srpc_peer));
1407                 spin_unlock(&sfw_data.fw_lock);
1408                 return -EAGAIN;
1409         }
1410
1411         sfw_data.fw_active_srpc = rpc;
1412         spin_unlock(&sfw_data.fw_lock);
1413
1414         rc = sfw_add_test(rpc);
1415
1416         spin_lock(&sfw_data.fw_lock);
1417
1418 #ifdef __KERNEL__
1419         if (!sfw_data.fw_shuttingdown)
1420                 sfw_add_session_timer();
1421 #else
1422         LASSERT(!sfw_data.fw_shuttingdown);
1423         sfw_add_session_timer();
1424 #endif
1425
1426         sfw_data.fw_active_srpc = NULL;
1427         spin_unlock(&sfw_data.fw_lock);
1428         return rc;
1429 }
1430
1431 srpc_client_rpc_t *
1432 sfw_create_rpc(lnet_process_id_t peer, int service,
1433                unsigned features, int nbulkiov, int bulklen,
1434                void (*done)(srpc_client_rpc_t *), void *priv)
1435 {
1436         srpc_client_rpc_t *rpc = NULL;
1437
1438         spin_lock(&sfw_data.fw_lock);
1439
1440         LASSERT (!sfw_data.fw_shuttingdown);
1441         LASSERT (service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1442
1443         if (nbulkiov == 0 && !cfs_list_empty(&sfw_data.fw_zombie_rpcs)) {
1444                 rpc = cfs_list_entry(sfw_data.fw_zombie_rpcs.next,
1445                                      srpc_client_rpc_t, crpc_list);
1446                 cfs_list_del(&rpc->crpc_list);
1447
1448                 srpc_init_client_rpc(rpc, peer, service, 0, 0,
1449                                      done, sfw_client_rpc_fini, priv);
1450         }
1451
1452         spin_unlock(&sfw_data.fw_lock);
1453
1454         if (rpc == NULL) {
1455                 rpc = srpc_create_client_rpc(peer, service,
1456                                              nbulkiov, bulklen, done,
1457                                              nbulkiov != 0 ?  NULL :
1458                                              sfw_client_rpc_fini,
1459                                              priv);
1460         }
1461
1462         if (rpc != NULL) /* "session" is concept in framework */
1463                 rpc->crpc_reqstmsg.msg_ses_feats = features;
1464
1465         return rpc;
1466 }
1467
1468 void
1469 sfw_unpack_message (srpc_msg_t *msg)
1470 {
1471         if (msg->msg_magic == SRPC_MSG_MAGIC)
1472                 return; /* no flipping needed */
1473
1474         /* srpc module should guarantee I wouldn't get crap */
1475         LASSERT (msg->msg_magic == __swab32(SRPC_MSG_MAGIC));
1476
1477         if (msg->msg_type == SRPC_MSG_STAT_REQST) {
1478                 srpc_stat_reqst_t *req = &msg->msg_body.stat_reqst;
1479
1480                 __swab32s(&req->str_type);
1481                 __swab64s(&req->str_rpyid);
1482                 sfw_unpack_sid(req->str_sid);
1483                 return;
1484         }
1485
1486         if (msg->msg_type == SRPC_MSG_STAT_REPLY) {
1487                 srpc_stat_reply_t *rep = &msg->msg_body.stat_reply;
1488
1489                 __swab32s(&rep->str_status);
1490                 sfw_unpack_sid(rep->str_sid);
1491                 sfw_unpack_fw_counters(rep->str_fw);
1492                 sfw_unpack_rpc_counters(rep->str_rpc);
1493                 sfw_unpack_lnet_counters(rep->str_lnet);
1494                 return;
1495         }
1496
1497         if (msg->msg_type == SRPC_MSG_MKSN_REQST) {
1498                 srpc_mksn_reqst_t *req = &msg->msg_body.mksn_reqst;
1499
1500                 __swab64s(&req->mksn_rpyid);
1501                 __swab32s(&req->mksn_force);
1502                 sfw_unpack_sid(req->mksn_sid);
1503                 return;
1504         }
1505
1506         if (msg->msg_type == SRPC_MSG_MKSN_REPLY) {
1507                 srpc_mksn_reply_t *rep = &msg->msg_body.mksn_reply;
1508
1509                 __swab32s(&rep->mksn_status);
1510                 __swab32s(&rep->mksn_timeout);
1511                 sfw_unpack_sid(rep->mksn_sid);
1512                 return;
1513         }
1514
1515         if (msg->msg_type == SRPC_MSG_RMSN_REQST) {
1516                 srpc_rmsn_reqst_t *req = &msg->msg_body.rmsn_reqst;
1517
1518                 __swab64s(&req->rmsn_rpyid);
1519                 sfw_unpack_sid(req->rmsn_sid);
1520                 return;
1521         }
1522
1523         if (msg->msg_type == SRPC_MSG_RMSN_REPLY) {
1524                 srpc_rmsn_reply_t *rep = &msg->msg_body.rmsn_reply;
1525
1526                 __swab32s(&rep->rmsn_status);
1527                 sfw_unpack_sid(rep->rmsn_sid);
1528                 return;
1529         }
1530
1531         if (msg->msg_type == SRPC_MSG_DEBUG_REQST) {
1532                 srpc_debug_reqst_t *req = &msg->msg_body.dbg_reqst;
1533
1534                 __swab64s(&req->dbg_rpyid);
1535                 __swab32s(&req->dbg_flags);
1536                 sfw_unpack_sid(req->dbg_sid);
1537                 return;
1538         }
1539
1540         if (msg->msg_type == SRPC_MSG_DEBUG_REPLY) {
1541                 srpc_debug_reply_t *rep = &msg->msg_body.dbg_reply;
1542
1543                 __swab32s(&rep->dbg_nbatch);
1544                 __swab32s(&rep->dbg_timeout);
1545                 sfw_unpack_sid(rep->dbg_sid);
1546                 return;
1547         }
1548
1549         if (msg->msg_type == SRPC_MSG_BATCH_REQST) {
1550                 srpc_batch_reqst_t *req = &msg->msg_body.bat_reqst;
1551
1552                 __swab32s(&req->bar_opc);
1553                 __swab64s(&req->bar_rpyid);
1554                 __swab32s(&req->bar_testidx);
1555                 __swab32s(&req->bar_arg);
1556                 sfw_unpack_sid(req->bar_sid);
1557                 __swab64s(&req->bar_bid.bat_id);
1558                 return;
1559         }
1560
1561         if (msg->msg_type == SRPC_MSG_BATCH_REPLY) {
1562                 srpc_batch_reply_t *rep = &msg->msg_body.bat_reply;
1563
1564                 __swab32s(&rep->bar_status);
1565                 sfw_unpack_sid(rep->bar_sid);
1566                 return;
1567         }
1568
1569         if (msg->msg_type == SRPC_MSG_TEST_REQST) {
1570                 srpc_test_reqst_t *req = &msg->msg_body.tes_reqst;
1571
1572                 __swab64s(&req->tsr_rpyid);
1573                 __swab64s(&req->tsr_bulkid);
1574                 __swab32s(&req->tsr_loop);
1575                 __swab32s(&req->tsr_ndest);
1576                 __swab32s(&req->tsr_concur);
1577                 __swab32s(&req->tsr_service);
1578                 sfw_unpack_sid(req->tsr_sid);
1579                 __swab64s(&req->tsr_bid.bat_id);
1580                 return;
1581         }
1582
1583         if (msg->msg_type == SRPC_MSG_TEST_REPLY) {
1584                 srpc_test_reply_t *rep = &msg->msg_body.tes_reply;
1585
1586                 __swab32s(&rep->tsr_status);
1587                 sfw_unpack_sid(rep->tsr_sid);
1588                 return;
1589         }
1590
1591         if (msg->msg_type == SRPC_MSG_JOIN_REQST) {
1592                 srpc_join_reqst_t *req = &msg->msg_body.join_reqst;
1593
1594                 __swab64s(&req->join_rpyid);
1595                 sfw_unpack_sid(req->join_sid);
1596                 return;
1597         }
1598
1599         if (msg->msg_type == SRPC_MSG_JOIN_REPLY) {
1600                 srpc_join_reply_t *rep = &msg->msg_body.join_reply;
1601
1602                 __swab32s(&rep->join_status);
1603                 __swab32s(&rep->join_timeout);
1604                 sfw_unpack_sid(rep->join_sid);
1605                 return;
1606         }
1607
1608         LBUG ();
1609         return;
1610 }
1611
1612 void
1613 sfw_abort_rpc (srpc_client_rpc_t *rpc)
1614 {
1615         LASSERT(cfs_atomic_read(&rpc->crpc_refcount) > 0);
1616         LASSERT(rpc->crpc_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1617
1618         spin_lock(&rpc->crpc_lock);
1619         srpc_abort_rpc(rpc, -EINTR);
1620         spin_unlock(&rpc->crpc_lock);
1621         return;
1622 }
1623
1624 void
1625 sfw_post_rpc (srpc_client_rpc_t *rpc)
1626 {
1627         spin_lock(&rpc->crpc_lock);
1628
1629         LASSERT (!rpc->crpc_closed);
1630         LASSERT (!rpc->crpc_aborted);
1631         LASSERT (cfs_list_empty(&rpc->crpc_list));
1632         LASSERT (!sfw_data.fw_shuttingdown);
1633
1634         rpc->crpc_timeout = rpc_timeout;
1635         srpc_post_rpc(rpc);
1636
1637         spin_unlock(&rpc->crpc_lock);
1638         return;
1639 }
1640
1641 static srpc_service_t sfw_services[] =
1642 {
1643         {
1644                 /* sv_id */    SRPC_SERVICE_DEBUG,
1645                 /* sv_name */  "debug",
1646                 0
1647         },
1648         {
1649                 /* sv_id */    SRPC_SERVICE_QUERY_STAT,
1650                 /* sv_name */  "query stats",
1651                 0
1652         },
1653         {
1654                 /* sv_id */    SRPC_SERVICE_MAKE_SESSION,
1655                 /* sv_name */  "make session",
1656                 0
1657         },
1658         {
1659                 /* sv_id */    SRPC_SERVICE_REMOVE_SESSION,
1660                 /* sv_name */  "remove session",
1661                 0
1662         },
1663         {
1664                 /* sv_id */    SRPC_SERVICE_BATCH,
1665                 /* sv_name */  "batch service",
1666                 0
1667         },
1668         {
1669                 /* sv_id */    SRPC_SERVICE_TEST,
1670                 /* sv_name */  "test service",
1671                 0
1672         },
1673         {
1674                 /* sv_id */    0,
1675                 /* sv_name */  NULL,
1676                 0
1677         }
1678 };
1679
1680 extern sfw_test_client_ops_t ping_test_client;
1681 extern srpc_service_t        ping_test_service;
1682 extern void ping_init_test_client(void);
1683 extern void ping_init_test_service(void);
1684
1685 extern sfw_test_client_ops_t brw_test_client;
1686 extern srpc_service_t        brw_test_service;
1687 extern void brw_init_test_client(void);
1688 extern void brw_init_test_service(void);
1689
1690
1691 int
1692 sfw_startup (void)
1693 {
1694         int              i;
1695         int              rc;
1696         int              error;
1697         srpc_service_t  *sv;
1698         sfw_test_case_t *tsc;
1699
1700 #ifndef __KERNEL__
1701         char *s;
1702
1703         s = getenv("SESSION_TIMEOUT");
1704         session_timeout = s != NULL ? atoi(s) : session_timeout;
1705
1706         s = getenv("RPC_TIMEOUT");
1707         rpc_timeout = s != NULL ? atoi(s) : rpc_timeout;
1708 #endif
1709
1710         if (session_timeout < 0) {
1711                 CERROR ("Session timeout must be non-negative: %d\n",
1712                         session_timeout);
1713                 return -EINVAL;
1714         }
1715
1716         if (rpc_timeout < 0) {
1717                 CERROR ("RPC timeout must be non-negative: %d\n",
1718                         rpc_timeout);
1719                 return -EINVAL;
1720         }
1721
1722         if (session_timeout == 0)
1723                 CWARN ("Zero session_timeout specified "
1724                        "- test sessions never expire.\n");
1725
1726         if (rpc_timeout == 0)
1727                 CWARN ("Zero rpc_timeout specified "
1728                        "- test RPC never expire.\n");
1729
1730         memset(&sfw_data, 0, sizeof(struct smoketest_framework));
1731
1732         sfw_data.fw_session     = NULL;
1733         sfw_data.fw_active_srpc = NULL;
1734         spin_lock_init(&sfw_data.fw_lock);
1735         cfs_atomic_set(&sfw_data.fw_nzombies, 0);
1736         CFS_INIT_LIST_HEAD(&sfw_data.fw_tests);
1737         CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_rpcs);
1738         CFS_INIT_LIST_HEAD(&sfw_data.fw_zombie_sessions);
1739
1740         brw_init_test_client();
1741         brw_init_test_service();
1742         rc = sfw_register_test(&brw_test_service, &brw_test_client);
1743         LASSERT (rc == 0);
1744
1745         ping_init_test_client();
1746         ping_init_test_service();
1747         rc = sfw_register_test(&ping_test_service, &ping_test_client);
1748         LASSERT (rc == 0);
1749
1750         error = 0;
1751         cfs_list_for_each_entry_typed (tsc, &sfw_data.fw_tests,
1752                                        sfw_test_case_t, tsc_list) {
1753                 sv = tsc->tsc_srv_service;
1754
1755                 rc = srpc_add_service(sv);
1756                 LASSERT (rc != -EBUSY);
1757                 if (rc != 0) {
1758                         CWARN ("Failed to add %s service: %d\n",
1759                                sv->sv_name, rc);
1760                         error = rc;
1761                 }
1762         }
1763
1764         for (i = 0; ; i++) {
1765                 sv = &sfw_services[i];
1766                 if (sv->sv_name == NULL) break;
1767
1768                 sv->sv_bulk_ready = NULL;
1769                 sv->sv_handler    = sfw_handle_server_rpc;
1770                 sv->sv_wi_total   = SFW_FRWK_WI_MAX;
1771                 if (sv->sv_id == SRPC_SERVICE_TEST)
1772                         sv->sv_bulk_ready = sfw_bulk_ready;
1773
1774                 rc = srpc_add_service(sv);
1775                 LASSERT (rc != -EBUSY);
1776                 if (rc != 0) {
1777                         CWARN ("Failed to add %s service: %d\n",
1778                                sv->sv_name, rc);
1779                         error = rc;
1780                 }
1781
1782                 /* about to sfw_shutdown, no need to add buffer */
1783                 if (error) continue;
1784
1785                 rc = srpc_service_add_buffers(sv, sv->sv_wi_total);
1786                 if (rc != 0) {
1787                         CWARN("Failed to reserve enough buffers: "
1788                               "service %s, %d needed: %d\n",
1789                               sv->sv_name, sv->sv_wi_total, rc);
1790                         error = -ENOMEM;
1791                 }
1792         }
1793
1794         if (error != 0)
1795                 sfw_shutdown();
1796         return error;
1797 }
1798
1799 void
1800 sfw_shutdown (void)
1801 {
1802         srpc_service_t  *sv;
1803         sfw_test_case_t *tsc;
1804         int              i;
1805
1806         spin_lock(&sfw_data.fw_lock);
1807
1808         sfw_data.fw_shuttingdown = 1;
1809 #ifdef __KERNEL__
1810         lst_wait_until(sfw_data.fw_active_srpc == NULL, sfw_data.fw_lock,
1811                        "waiting for active RPC to finish.\n");
1812 #else
1813         LASSERT (sfw_data.fw_active_srpc == NULL);
1814 #endif
1815
1816         if (sfw_del_session_timer() != 0)
1817                 lst_wait_until(sfw_data.fw_session == NULL, sfw_data.fw_lock,
1818                                "waiting for session timer to explode.\n");
1819
1820         sfw_deactivate_session();
1821         lst_wait_until(cfs_atomic_read(&sfw_data.fw_nzombies) == 0,
1822                        sfw_data.fw_lock,
1823                        "waiting for %d zombie sessions to die.\n",
1824                        cfs_atomic_read(&sfw_data.fw_nzombies));
1825
1826         spin_unlock(&sfw_data.fw_lock);
1827
1828         for (i = 0; ; i++) {
1829                 sv = &sfw_services[i];
1830                 if (sv->sv_name == NULL)
1831                         break;
1832
1833                 srpc_shutdown_service(sv);
1834                 srpc_remove_service(sv);
1835         }
1836
1837         cfs_list_for_each_entry_typed (tsc, &sfw_data.fw_tests,
1838                                        sfw_test_case_t, tsc_list) {
1839                 sv = tsc->tsc_srv_service;
1840                 srpc_shutdown_service(sv);
1841                 srpc_remove_service(sv);
1842         }
1843
1844         while (!cfs_list_empty(&sfw_data.fw_zombie_rpcs)) {
1845                 srpc_client_rpc_t *rpc;
1846
1847                 rpc = cfs_list_entry(sfw_data.fw_zombie_rpcs.next,
1848                                      srpc_client_rpc_t, crpc_list);
1849                 cfs_list_del(&rpc->crpc_list);
1850
1851                 LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
1852         }
1853
1854         for (i = 0; ; i++) {
1855                 sv = &sfw_services[i];
1856                 if (sv->sv_name == NULL)
1857                         break;
1858
1859                 srpc_wait_service_shutdown(sv);
1860         }
1861
1862         while (!cfs_list_empty(&sfw_data.fw_tests)) {
1863                 tsc = cfs_list_entry(sfw_data.fw_tests.next,
1864                                      sfw_test_case_t, tsc_list);
1865
1866                 srpc_wait_service_shutdown(tsc->tsc_srv_service);
1867
1868                 cfs_list_del(&tsc->tsc_list);
1869                 LIBCFS_FREE(tsc, sizeof(*tsc));
1870         }
1871
1872         return;
1873 }