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