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