lnet/selftest/framework.c

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