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         strscpy(&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 = strscpy(&reply->mksn_name[0], &sn->sn_name[0],
 442                                         sizeof(reply->mksn_name));
 443                         if (cplen < 0)
 444                                 return cplen;
 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         int cplen;
 524
 525         if (sn == NULL) {
 526                 reply->dbg_status = ESRCH;
 527                 reply->dbg_sid = get_old_sid(NULL);
 528                 return 0;
 529         }
 530
 531         reply->dbg_status  = 0;
 532         reply->dbg_sid = get_old_sid(sn);
 533         reply->dbg_timeout = sn->sn_timeout;
 534         cplen = strscpy(reply->dbg_name, &sn->sn_name[0],
 535                     sizeof(reply->dbg_name));
 536         if (cplen < 0)
 537                 return cplen;
 538
 539         return 0;
 540 }
 541
 542 static void
 543 sfw_test_rpc_fini(struct srpc_client_rpc *rpc)
 544 {
 545         struct sfw_test_unit *tsu = rpc->crpc_priv;
 546         struct sfw_test_instance *tsi = tsu->tsu_instance;
 547
 548         /* Called with hold of tsi->tsi_lock */
 549         LASSERT(list_empty(&rpc->crpc_list));
 550         rpc->crpc_wi.swi_state = SWI_STATE_DONE;
 551         list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
 552 }
 553
 554 static inline int
 555 sfw_test_buffers(struct sfw_test_instance *tsi)
 556 {
 557         struct sfw_test_case    *tsc;
 558         struct srpc_service     *svc;
 559         int                     nbuf;
 560
 561         LASSERT(tsi != NULL);
 562         tsc = sfw_find_test_case(tsi->tsi_service);
 563         LASSERT(tsc != NULL);
 564         svc = tsc->tsc_srv_service;
 565         LASSERT(svc != NULL);
 566
 567         nbuf = min(svc->sv_wi_total, tsi->tsi_loop) / svc->sv_ncpts;
 568         return max(SFW_TEST_WI_MIN, nbuf + SFW_TEST_WI_EXTRA);
 569 }
 570
 571 static int
 572 sfw_load_test(struct sfw_test_instance *tsi)
 573 {
 574         struct sfw_test_case    *tsc;
 575         struct srpc_service     *svc;
 576         int                     nbuf;
 577         int                     rc;
 578
 579         LASSERT(tsi != NULL);
 580         tsc = sfw_find_test_case(tsi->tsi_service);
 581         nbuf = sfw_test_buffers(tsi);
 582         LASSERT(tsc != NULL);
 583         svc = tsc->tsc_srv_service;
 584
 585         if (tsi->tsi_is_client) {
 586                 tsi->tsi_ops = tsc->tsc_cli_ops;
 587                 return 0;
 588         }
 589
 590         rc = srpc_service_add_buffers(svc, nbuf);
 591         if (rc != 0) {
 592                 CWARN("Failed to reserve enough buffers: "
 593                       "service %s, %d needed: %d\n", svc->sv_name, nbuf, rc);
 594                 /* NB: this error handler is not strictly correct, because
 595                  * it may release more buffers than already allocated,
 596                  * but it doesn't matter because request portal should
 597                  * be lazy portal and will grow buffers if necessary. */
 598                 srpc_service_remove_buffers(svc, nbuf);
 599                 return -ENOMEM;
 600         }
 601
 602         CDEBUG(D_NET, "Reserved %d buffers for test %s\n",
 603                nbuf * (srpc_serv_is_framework(svc) ?
 604                        1 : cfs_cpt_number(cfs_cpt_tab)), svc->sv_name);
 605         return 0;
 606 }
 607
 608 static void
 609 sfw_unload_test(struct sfw_test_instance *tsi)
 610 {
 611         struct sfw_test_case *tsc;
 612
 613         LASSERT(tsi != NULL);
 614         tsc = sfw_find_test_case(tsi->tsi_service);
 615         LASSERT(tsc != NULL);
 616
 617         if (tsi->tsi_is_client)
 618                 return;
 619
 620         /* shrink buffers, because request portal is lazy portal
 621          * which can grow buffers at runtime so we may leave
 622          * some buffers behind, but never mind... */
 623         srpc_service_remove_buffers(tsc->tsc_srv_service,
 624                                     sfw_test_buffers(tsi));
 625 }
 626
 627 static void
 628 sfw_destroy_test_instance(struct sfw_test_instance *tsi)
 629 {
 630         struct srpc_client_rpc *rpc;
 631         struct sfw_test_unit *tsu;
 632
 633         if (!tsi->tsi_is_client)
 634                 goto clean;
 635
 636         tsi->tsi_ops->tso_fini(tsi);
 637
 638         LASSERT(!tsi->tsi_stopping);
 639         LASSERT(list_empty(&tsi->tsi_active_rpcs));
 640         LASSERT(!sfw_test_active(tsi));
 641
 642         while (!list_empty(&tsi->tsi_units)) {
 643                 tsu = list_first_entry(&tsi->tsi_units,
 644                                        struct sfw_test_unit, tsu_list);
 645                 list_del(&tsu->tsu_list);
 646                 LIBCFS_FREE(tsu, sizeof(*tsu));
 647         }
 648
 649         while (!list_empty(&tsi->tsi_free_rpcs)) {
 650                 rpc = list_first_entry(&tsi->tsi_free_rpcs,
 651                                        struct srpc_client_rpc, crpc_list);
 652                 list_del(&rpc->crpc_list);
 653                 swi_cancel_workitem(&rpc->crpc_wi);
 654                 LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
 655         }
 656
 657 clean:
 658         sfw_unload_test(tsi);
 659         LIBCFS_FREE(tsi, sizeof(*tsi));
 660 }
 661
 662 static void
 663 sfw_destroy_batch(struct sfw_batch *tsb)
 664 {
 665         struct sfw_test_instance *tsi;
 666
 667         LASSERT(!sfw_batch_active(tsb));
 668         LASSERT(list_empty(&tsb->bat_list));
 669
 670         while (!list_empty(&tsb->bat_tests)) {
 671                 tsi = list_first_entry(&tsb->bat_tests,
 672                                        struct sfw_test_instance, tsi_list);
 673                 list_del_init(&tsi->tsi_list);
 674                 sfw_destroy_test_instance(tsi);
 675         }
 676
 677         LIBCFS_FREE(tsb, sizeof(*tsb));
 678 }
 679
 680 static void
 681 sfw_destroy_session(struct sfw_session *sn)
 682 {
 683         struct sfw_batch *batch;
 684
 685         LASSERT(list_empty(&sn->sn_list));
 686         LASSERT(sn != sfw_data.fw_session);
 687
 688         while (!list_empty(&sn->sn_batches)) {
 689                 batch = list_first_entry(&sn->sn_batches,
 690                                          struct sfw_batch, bat_list);
 691                 list_del_init(&batch->bat_list);
 692                 sfw_destroy_batch(batch);
 693         }
 694
 695         LIBCFS_FREE(sn, sizeof(*sn));
 696         atomic_dec(&sfw_data.fw_nzombies);
 697 }
 698
 699 static void
 700 sfw_unpack_addtest_req(struct srpc_msg *msg)
 701 {
 702         struct srpc_test_reqst *req = &msg->msg_body.tes_reqst;
 703
 704         LASSERT(msg->msg_type == SRPC_MSG_TEST_REQST);
 705         LASSERT(req->tsr_is_client);
 706
 707         if (msg->msg_magic == SRPC_MSG_MAGIC)
 708                 return; /* no flipping needed */
 709
 710         LASSERT(msg->msg_magic == __swab32(SRPC_MSG_MAGIC));
 711
 712         if (req->tsr_service == SRPC_SERVICE_BRW) {
 713                 if ((msg->msg_ses_feats & LST_FEAT_BULK_LEN) == 0) {
 714                         struct test_bulk_req *bulk = &req->tsr_u.bulk_v0;
 715
 716                         __swab32s(&bulk->blk_opc);
 717                         __swab32s(&bulk->blk_npg);
 718                         __swab32s(&bulk->blk_flags);
 719
 720                 } else {
 721                         struct test_bulk_req_v1 *bulk = &req->tsr_u.bulk_v1;
 722
 723                         __swab16s(&bulk->blk_opc);
 724                         __swab16s(&bulk->blk_flags);
 725                         __swab32s(&bulk->blk_offset);
 726                         __swab32s(&bulk->blk_len);
 727                 }
 728
 729                 return;
 730         }
 731
 732         if (req->tsr_service == SRPC_SERVICE_PING) {
 733                 struct test_ping_req *ping = &req->tsr_u.ping;
 734
 735                 __swab32s(&ping->png_size);
 736                 __swab32s(&ping->png_flags);
 737                 return;
 738         }
 739
 740         LBUG();
 741 }
 742
 743 static int
 744 sfw_add_test_instance(struct sfw_batch *tsb, struct srpc_server_rpc *rpc)
 745 {
 746         struct srpc_msg *msg = &rpc->srpc_reqstbuf->buf_msg;
 747         struct srpc_test_reqst *req = &msg->msg_body.tes_reqst;
 748         struct srpc_bulk *bk = rpc->srpc_bulk;
 749         int ndest = req->tsr_ndest;
 750         struct sfw_test_unit *tsu;
 751         struct sfw_test_instance *tsi;
 752         int i;
 753         int rc;
 754
 755         LIBCFS_ALLOC(tsi, sizeof(*tsi));
 756         if (tsi == NULL) {
 757                 CERROR("Can't allocate test instance for batch: %llu: rc = %d\n",
 758                        tsb->bat_id.bat_id, -ENOMEM);
 759                 return -ENOMEM;
 760         }
 761
 762         spin_lock_init(&tsi->tsi_lock);
 763         atomic_set(&tsi->tsi_nactive, 0);
 764         INIT_LIST_HEAD(&tsi->tsi_units);
 765         INIT_LIST_HEAD(&tsi->tsi_free_rpcs);
 766         INIT_LIST_HEAD(&tsi->tsi_active_rpcs);
 767
 768         tsi->tsi_stopping      = 0;
 769         tsi->tsi_batch         = tsb;
 770         tsi->tsi_loop          = req->tsr_loop;
 771         tsi->tsi_concur        = req->tsr_concur;
 772         tsi->tsi_service       = req->tsr_service;
 773         tsi->tsi_is_client     = !!(req->tsr_is_client);
 774         tsi->tsi_stoptsu_onerr = !!(req->tsr_stop_onerr);
 775
 776         rc = sfw_load_test(tsi);
 777         if (rc != 0) {
 778                 LIBCFS_FREE(tsi, sizeof(*tsi));
 779                 return rc;
 780         }
 781
 782         LASSERT(!sfw_batch_active(tsb));
 783
 784         if (!tsi->tsi_is_client) {
 785                 /* it's test server, just add it to tsb */
 786                 list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
 787                 return 0;
 788         }
 789
 790         LASSERT(bk != NULL);
 791         LASSERT(bk->bk_niov * SFW_ID_PER_PAGE >= (unsigned int)ndest);
 792         LASSERT((unsigned int)bk->bk_len >=
 793                 sizeof(struct lnet_process_id_packed) * ndest);
 794
 795         sfw_unpack_addtest_req(msg);
 796         memcpy(&tsi->tsi_u, &req->tsr_u, sizeof(tsi->tsi_u));
 797
 798         for (i = 0; i < ndest; i++) {
 799                 struct lnet_process_id_packed *dests;
 800                 struct lnet_process_id_packed  id;
 801                 int j;
 802
 803                 dests = page_address(bk->bk_iovs[i / SFW_ID_PER_PAGE].bv_page);
 804                 LASSERT(dests != NULL);  /* my pages are within KVM always */
 805                 id = dests[i % SFW_ID_PER_PAGE];
 806                 if (msg->msg_magic != SRPC_MSG_MAGIC)
 807                         sfw_unpack_id(id);
 808
 809                 for (j = 0; j < tsi->tsi_concur; j++) {
 810                         LIBCFS_ALLOC(tsu, sizeof(*tsu));
 811                         if (tsu == NULL) {
 812                                 rc = -ENOMEM;
 813                                 CERROR("Can't allocate tsu for %d: rc = %d\n",
 814                                        tsi->tsi_service, rc);
 815                                 goto error;
 816                         }
 817
 818                         tsu->tsu_dest.nid = id.nid;
 819                         tsu->tsu_dest.pid = id.pid;
 820                         tsu->tsu_instance = tsi;
 821                         tsu->tsu_private  = NULL;
 822                         list_add_tail(&tsu->tsu_list, &tsi->tsi_units);
 823                 }
 824         }
 825
 826         rc = tsi->tsi_ops->tso_init(tsi);
 827         if (rc == 0) {
 828                 list_add_tail(&tsi->tsi_list, &tsb->bat_tests);
 829                 return 0;
 830         }
 831
 832 error:
 833         LASSERT(rc != 0);
 834         sfw_destroy_test_instance(tsi);
 835         return rc;
 836 }
 837
 838 static void
 839 sfw_test_unit_done(struct sfw_test_unit *tsu)
 840 {
 841         struct sfw_test_instance *tsi = tsu->tsu_instance;
 842         struct sfw_batch *tsb = tsi->tsi_batch;
 843         struct sfw_session *sn = tsb->bat_session;
 844
 845         LASSERT(sfw_test_active(tsi));
 846
 847         if (!atomic_dec_and_test(&tsi->tsi_nactive))
 848                 return;
 849
 850         /* the test instance is done */
 851         spin_lock(&tsi->tsi_lock);
 852
 853         tsi->tsi_stopping = 0;
 854
 855         spin_unlock(&tsi->tsi_lock);
 856
 857         spin_lock(&sfw_data.fw_lock);
 858
 859         if (!atomic_dec_and_test(&tsb->bat_nactive) ||/* tsb still active */
 860             sn == sfw_data.fw_session) {                  /* sn also active */
 861                 spin_unlock(&sfw_data.fw_lock);
 862                 return;
 863         }
 864
 865         LASSERT(!list_empty(&sn->sn_list)); /* I'm a zombie! */
 866
 867         list_for_each_entry(tsb, &sn->sn_batches, bat_list) {
 868                 if (sfw_batch_active(tsb)) {
 869                         spin_unlock(&sfw_data.fw_lock);
 870                         return;
 871                 }
 872         }
 873
 874         list_del_init(&sn->sn_list);
 875         spin_unlock(&sfw_data.fw_lock);
 876
 877         sfw_destroy_session(sn);
 878 }
 879
 880 static void
 881 sfw_test_rpc_done(struct srpc_client_rpc *rpc)
 882 {
 883         struct sfw_test_unit *tsu = rpc->crpc_priv;
 884         struct sfw_test_instance *tsi = tsu->tsu_instance;
 885         int done = 0;
 886
 887         tsi->tsi_ops->tso_done_rpc(tsu, rpc);
 888
 889         spin_lock(&tsi->tsi_lock);
 890
 891         LASSERT(sfw_test_active(tsi));
 892         LASSERT(!list_empty(&rpc->crpc_list));
 893
 894         list_del_init(&rpc->crpc_list);
 895
 896         /* batch is stopping or loop is done or get error */
 897         if (tsi->tsi_stopping || tsu->tsu_loop == 0 ||
 898             (rpc->crpc_status != 0 && tsi->tsi_stoptsu_onerr))
 899                 done = 1;
 900
 901         /* dec ref for poster */
 902         srpc_client_rpc_decref(rpc);
 903
 904         spin_unlock(&tsi->tsi_lock);
 905
 906         if (!done) {
 907                 swi_schedule_workitem(&tsu->tsu_worker);
 908                 return;
 909         }
 910
 911         sfw_test_unit_done(tsu);
 912 }
 913
 914 int
 915 sfw_create_test_rpc(struct sfw_test_unit *tsu, struct lnet_process_id peer,
 916                     unsigned features, int nblk, int blklen,
 917                     struct srpc_client_rpc **rpcpp)
 918 {
 919         struct srpc_client_rpc *rpc = NULL;
 920         struct sfw_test_instance *tsi = tsu->tsu_instance;
 921
 922         spin_lock(&tsi->tsi_lock);
 923
 924         LASSERT(sfw_test_active(tsi));
 925
 926         if (!list_empty(&tsi->tsi_free_rpcs)) {
 927                 /* pick request from buffer */
 928                 rpc = list_first_entry(&tsi->tsi_free_rpcs,
 929                                        struct srpc_client_rpc, crpc_list);
 930                 LASSERT(nblk == rpc->crpc_bulk.bk_niov);
 931                 list_del_init(&rpc->crpc_list);
 932         }
 933
 934         spin_unlock(&tsi->tsi_lock);
 935
 936         if (rpc == NULL) {
 937                 rpc = srpc_create_client_rpc(peer, tsi->tsi_service, nblk,
 938                                              blklen, sfw_test_rpc_done,
 939                                              sfw_test_rpc_fini, tsu);
 940         } else {
 941                 swi_cancel_workitem(&rpc->crpc_wi);
 942                 srpc_init_client_rpc(rpc, peer, tsi->tsi_service, nblk,
 943                                      blklen, sfw_test_rpc_done,
 944                                      sfw_test_rpc_fini, tsu);
 945         }
 946
 947         if (rpc == NULL) {
 948                 CERROR("Can't create rpc for test %d: rc = %d\n",
 949                        tsi->tsi_service, -ENOMEM);
 950                 return -ENOMEM;
 951         }
 952
 953         rpc->crpc_reqstmsg.msg_ses_feats = features;
 954         *rpcpp = rpc;
 955
 956         return 0;
 957 }
 958
 959 static void
 960 sfw_run_test(struct swi_workitem *wi)
 961 {
 962         struct sfw_test_unit *tsu = container_of(wi, struct sfw_test_unit,
 963                                                  tsu_worker);
 964         struct sfw_test_instance *tsi = tsu->tsu_instance;
 965         struct srpc_client_rpc *rpc = NULL;
 966
 967         if (tsi->tsi_ops->tso_prep_rpc(tsu, tsu->tsu_dest, &rpc) != 0) {
 968                 LASSERT(rpc == NULL);
 969                 wi->swi_state = SWI_STATE_DONE;
 970                 goto test_done;
 971         }
 972
 973         LASSERT(rpc != NULL);
 974
 975         spin_lock(&tsi->tsi_lock);
 976         if (wi->swi_state == SWI_STATE_DONE) {
 977                 spin_unlock(&tsi->tsi_lock);
 978                 return;
 979         }
 980
 981         if (tsi->tsi_stopping) {
 982                 wi->swi_state = SWI_STATE_DONE;
 983                 list_add(&rpc->crpc_list, &tsi->tsi_free_rpcs);
 984                 spin_unlock(&tsi->tsi_lock);
 985                 goto test_done;
 986         }
 987
 988         if (tsu->tsu_loop > 0)
 989                 tsu->tsu_loop--;
 990
 991         list_add_tail(&rpc->crpc_list, &tsi->tsi_active_rpcs);
 992         wi->swi_state = SWI_STATE_RUNNING;
 993         spin_unlock(&tsi->tsi_lock);
 994
 995         spin_lock(&rpc->crpc_lock);
 996         rpc->crpc_timeout = rpc_timeout;
 997         srpc_post_rpc(rpc);
 998         spin_unlock(&rpc->crpc_lock);
 999         return;
1000
1001 test_done:
1002         /*
1003          * No one can schedule me now since:
1004          * - previous RPC, if any, has done and
1005          * - no new RPC is initiated.
1006          * - my batch is still active; no one can run it again now.
1007          * Cancel pending schedules and prevent future schedule attempts:
1008          */
1009         sfw_test_unit_done(tsu);
1010 }
1011
1012 static int
1013 sfw_run_batch(struct sfw_batch *tsb)
1014 {
1015         struct swi_workitem *wi;
1016         struct sfw_test_unit *tsu;
1017         struct sfw_test_instance *tsi;
1018
1019         if (sfw_batch_active(tsb)) {
1020                 CDEBUG(D_NET, "Batch already active: %llu (%d)\n",
1021                        tsb->bat_id.bat_id, atomic_read(&tsb->bat_nactive));
1022                 return 0;
1023         }
1024
1025         list_for_each_entry(tsi, &tsb->bat_tests, tsi_list) {
1026                 if (!tsi->tsi_is_client)        /* skip server instances */
1027                         continue;
1028
1029                 LASSERT(!tsi->tsi_stopping);
1030                 LASSERT(!sfw_test_active(tsi));
1031
1032                 atomic_inc(&tsb->bat_nactive);
1033
1034                 list_for_each_entry(tsu, &tsi->tsi_units, tsu_list) {
1035                         atomic_inc(&tsi->tsi_nactive);
1036                         tsu->tsu_loop = tsi->tsi_loop;
1037                         wi = &tsu->tsu_worker;
1038                         swi_init_workitem(wi, sfw_run_test,
1039                                           lst_test_wq[lnet_cpt_of_nid(
1040                                                               tsu->tsu_dest.nid,
1041                                                               NULL)]);
1042                         swi_schedule_workitem(wi);
1043                 }
1044         }
1045
1046         return 0;
1047 }
1048
1049 static int
1050 sfw_stop_batch(struct sfw_batch *tsb, int force)
1051 {
1052         struct sfw_test_instance *tsi;
1053         struct srpc_client_rpc *rpc;
1054
1055         if (!sfw_batch_active(tsb)) {
1056                 CDEBUG(D_NET, "Batch %llu inactive\n", tsb->bat_id.bat_id);
1057                 return 0;
1058         }
1059
1060         list_for_each_entry(tsi, &tsb->bat_tests, tsi_list) {
1061                 spin_lock(&tsi->tsi_lock);
1062
1063                 if (!tsi->tsi_is_client ||
1064                     !sfw_test_active(tsi) || tsi->tsi_stopping) {
1065                         spin_unlock(&tsi->tsi_lock);
1066                         continue;
1067                 }
1068
1069                 tsi->tsi_stopping = 1;
1070
1071                 if (!force) {
1072                         spin_unlock(&tsi->tsi_lock);
1073                         continue;
1074                 }
1075
1076                 /* abort launched rpcs in the test */
1077                 list_for_each_entry(rpc, &tsi->tsi_active_rpcs, crpc_list) {
1078                         spin_lock(&rpc->crpc_lock);
1079
1080                         srpc_abort_rpc(rpc, -EINTR);
1081
1082                         spin_unlock(&rpc->crpc_lock);
1083                 }
1084
1085                 spin_unlock(&tsi->tsi_lock);
1086         }
1087
1088         return 0;
1089 }
1090
1091 static int
1092 sfw_query_batch(struct sfw_batch *tsb, int testidx,
1093                 struct srpc_batch_reply *reply)
1094 {
1095         struct sfw_test_instance *tsi;
1096
1097         if (testidx < 0)
1098                 return -EINVAL;
1099
1100         if (testidx == 0) {
1101                 reply->bar_active = atomic_read(&tsb->bat_nactive);
1102                 return 0;
1103         }
1104
1105         list_for_each_entry(tsi, &tsb->bat_tests, tsi_list) {
1106                 if (testidx-- > 1)
1107                         continue;
1108
1109                 reply->bar_active = atomic_read(&tsi->tsi_nactive);
1110                 return 0;
1111         }
1112
1113         return -ENOENT;
1114 }
1115
1116 int
1117 sfw_alloc_pages(struct srpc_server_rpc *rpc, int cpt, int len,
1118                 int sink)
1119 {
1120         LASSERT(rpc->srpc_bulk == NULL);
1121         LASSERT(len > 0 && len <= LNET_MTU);
1122
1123         rpc->srpc_bulk = srpc_alloc_bulk(cpt, len);
1124         if (rpc->srpc_bulk == NULL)
1125                 return -ENOMEM;
1126
1127         srpc_init_bulk(rpc->srpc_bulk, 0, len, sink);
1128
1129         return 0;
1130 }
1131
1132 static int
1133 sfw_add_test(struct srpc_server_rpc *rpc)
1134 {
1135         struct sfw_session *sn = sfw_data.fw_session;
1136         struct srpc_test_reply *reply = &rpc->srpc_replymsg.msg_body.tes_reply;
1137         struct srpc_test_reqst *request;
1138         int rc;
1139         struct sfw_batch *bat;
1140
1141         request = &rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst;
1142         reply->tsr_sid = get_old_sid(sn);
1143
1144         if (request->tsr_loop == 0 || request->tsr_concur == 0 ||
1145             request->tsr_sid.ses_nid == LNET_NID_ANY ||
1146             request->tsr_ndest > SFW_MAX_NDESTS ||
1147             (request->tsr_is_client && request->tsr_ndest == 0) ||
1148             request->tsr_concur > SFW_MAX_CONCUR ||
1149             request->tsr_service > SRPC_SERVICE_MAX_ID ||
1150             request->tsr_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID) {
1151                 reply->tsr_status = EINVAL;
1152                 return 0;
1153         }
1154
1155         if (sn == NULL || !sfw_sid_equal(request->tsr_sid, sn->sn_id) ||
1156             sfw_find_test_case(request->tsr_service) == NULL) {
1157                 reply->tsr_status = ENOENT;
1158                 return 0;
1159         }
1160
1161         bat = sfw_bid2batch(request->tsr_bid);
1162         if (bat == NULL) {
1163                 CERROR("dropping RPC %s from %s under memory pressure: rc = %d\n",
1164                         rpc->srpc_scd->scd_svc->sv_name,
1165                         libcfs_id2str(rpc->srpc_peer), -ENOMEM);
1166                 return -ENOMEM;
1167         }
1168
1169         if (sfw_batch_active(bat)) {
1170                 reply->tsr_status = EBUSY;
1171                 return 0;
1172         }
1173
1174         if (request->tsr_is_client && rpc->srpc_bulk == NULL) {
1175                 /* rpc will be resumed later in sfw_bulk_ready */
1176                 int     len;
1177
1178                 len = sizeof(struct lnet_process_id_packed) *
1179                               request->tsr_ndest;
1180
1181                 return sfw_alloc_pages(rpc, CFS_CPT_ANY, len, 1);
1182         }
1183
1184         rc = sfw_add_test_instance(bat, rpc);
1185         CDEBUG(rc == 0 ? D_NET : D_WARNING,
1186                "%s test: sv %d %s, loop %d, concur %d, ndest %d\n",
1187                rc == 0 ? "Added" : "Failed to add", request->tsr_service,
1188                request->tsr_is_client ? "client" : "server",
1189                request->tsr_loop, request->tsr_concur, request->tsr_ndest);
1190
1191         reply->tsr_status = (rc < 0) ? -rc : rc;
1192         return 0;
1193 }
1194
1195 static int
1196 sfw_control_batch(struct srpc_batch_reqst *request,
1197                   struct srpc_batch_reply *reply)
1198 {
1199         struct sfw_session *sn = sfw_data.fw_session;
1200         struct sfw_batch *bat;
1201         int rc = 0;
1202
1203         reply->bar_sid = get_old_sid(sn);
1204
1205         if (sn == NULL || !sfw_sid_equal(request->bar_sid, sn->sn_id)) {
1206                 reply->bar_status = ESRCH;
1207                 return 0;
1208         }
1209
1210         bat = sfw_find_batch(request->bar_bid);
1211         if (bat == NULL) {
1212                 reply->bar_status = ENOENT;
1213                 return 0;
1214         }
1215
1216         switch (request->bar_opc) {
1217         case SRPC_BATCH_OPC_RUN:
1218                 rc = sfw_run_batch(bat);
1219                 break;
1220
1221         case SRPC_BATCH_OPC_STOP:
1222                 rc = sfw_stop_batch(bat, request->bar_arg);
1223                 break;
1224
1225         case SRPC_BATCH_OPC_QUERY:
1226                 rc = sfw_query_batch(bat, request->bar_testidx, reply);
1227                 break;
1228
1229         default:
1230                 return -EINVAL; /* drop it */
1231         }
1232
1233         reply->bar_status = (rc < 0) ? -rc : rc;
1234         return 0;
1235 }
1236
1237 static int
1238 sfw_handle_server_rpc(struct srpc_server_rpc *rpc)
1239 {
1240         struct srpc_service     *sv = rpc->srpc_scd->scd_svc;
1241         struct srpc_msg     *reply      = &rpc->srpc_replymsg;
1242         struct srpc_msg     *request    = &rpc->srpc_reqstbuf->buf_msg;
1243         unsigned        features = LST_FEATS_MASK;
1244         int             rc = 0;
1245
1246         LASSERT(sfw_data.fw_active_srpc == NULL);
1247         LASSERT(sv->sv_id <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1248
1249         spin_lock(&sfw_data.fw_lock);
1250
1251         if (sfw_data.fw_shuttingdown) {
1252                 spin_unlock(&sfw_data.fw_lock);
1253                 return -ESHUTDOWN;
1254         }
1255
1256         /* Remove timer to avoid racing with it or expiring active session */
1257         if (sfw_del_session_timer() != 0) {
1258                 CERROR("dropping RPC %s from %s: racing with expiry timer: rc = %d\n",
1259                        sv->sv_name, libcfs_id2str(rpc->srpc_peer), -EAGAIN);
1260                 spin_unlock(&sfw_data.fw_lock);
1261                 return -EAGAIN;
1262         }
1263
1264         sfw_data.fw_active_srpc = rpc;
1265         spin_unlock(&sfw_data.fw_lock);
1266
1267         sfw_unpack_message(request);
1268         LASSERT(request->msg_type == srpc_service2request(sv->sv_id));
1269
1270         /* rpc module should have checked this */
1271         LASSERT(request->msg_version == SRPC_MSG_VERSION);
1272
1273         if (sv->sv_id != SRPC_SERVICE_MAKE_SESSION &&
1274             sv->sv_id != SRPC_SERVICE_DEBUG) {
1275                 struct sfw_session *sn = sfw_data.fw_session;
1276
1277                 if (sn != NULL &&
1278                     sn->sn_features != request->msg_ses_feats) {
1279                         CNETERR("Features of framework RPC don't match features of current session: %x/%x\n",
1280                                 request->msg_ses_feats, sn->sn_features);
1281                         reply->msg_body.reply.status = EPROTO;
1282                         reply->msg_body.reply.sid.ses_stamp = sn->sn_id.ses_stamp;
1283                         reply->msg_body.reply.sid.ses_nid =
1284                                 lnet_nid_to_nid4(&sn->sn_id.ses_nid);
1285                         goto out;
1286                 }
1287
1288         } else if ((request->msg_ses_feats & ~LST_FEATS_MASK) != 0) {
1289                 /* NB: at this point, old version will ignore features and
1290                  * create new session anyway, so console should be able
1291                  * to handle this */
1292                 reply->msg_body.reply.status = EPROTO;
1293                 goto out;
1294         }
1295
1296         switch (sv->sv_id) {
1297         default:
1298                 LBUG();
1299         case SRPC_SERVICE_TEST:
1300                 rc = sfw_add_test(rpc);
1301                 break;
1302
1303         case SRPC_SERVICE_BATCH:
1304                 rc = sfw_control_batch(&request->msg_body.bat_reqst,
1305                                        &reply->msg_body.bat_reply);
1306                 break;
1307
1308         case SRPC_SERVICE_QUERY_STAT:
1309                 rc = sfw_get_stats(&request->msg_body.stat_reqst,
1310                                    &reply->msg_body.stat_reply);
1311                 break;
1312
1313         case SRPC_SERVICE_DEBUG:
1314                 rc = sfw_debug_session(&request->msg_body.dbg_reqst,
1315                                        &reply->msg_body.dbg_reply);
1316                 break;
1317
1318         case SRPC_SERVICE_MAKE_SESSION:
1319                 rc = sfw_make_session(&request->msg_body.mksn_reqst,
1320                                       &reply->msg_body.mksn_reply);
1321                 break;
1322
1323         case SRPC_SERVICE_REMOVE_SESSION:
1324                 rc = sfw_remove_session(&request->msg_body.rmsn_reqst,
1325                                         &reply->msg_body.rmsn_reply);
1326                 break;
1327         }
1328
1329         if (sfw_data.fw_session != NULL)
1330                 features = sfw_data.fw_session->sn_features;
1331  out:
1332         reply->msg_ses_feats = features;
1333         rpc->srpc_done = sfw_server_rpc_done;
1334         spin_lock(&sfw_data.fw_lock);
1335
1336         if (!sfw_data.fw_shuttingdown)
1337                 sfw_add_session_timer();
1338
1339         sfw_data.fw_active_srpc = NULL;
1340         spin_unlock(&sfw_data.fw_lock);
1341         return rc;
1342 }
1343
1344 static int
1345 sfw_bulk_ready(struct srpc_server_rpc *rpc, int status)
1346 {
1347         struct srpc_service     *sv = rpc->srpc_scd->scd_svc;
1348         int                     rc;
1349
1350         LASSERT(rpc->srpc_bulk != NULL);
1351         LASSERT(sv->sv_id == SRPC_SERVICE_TEST);
1352         LASSERT(sfw_data.fw_active_srpc == NULL);
1353         LASSERT(rpc->srpc_reqstbuf->buf_msg.msg_body.tes_reqst.tsr_is_client);
1354
1355         spin_lock(&sfw_data.fw_lock);
1356
1357         if (status != 0) {
1358                 CERROR("Bulk transfer failed for RPC: service %s, peer %s, status %d: rc = %d\n",
1359                        sv->sv_name, libcfs_id2str(rpc->srpc_peer), status, -EIO);
1360                 spin_unlock(&sfw_data.fw_lock);
1361                 return -EIO;
1362         }
1363
1364         if (sfw_data.fw_shuttingdown) {
1365                 spin_unlock(&sfw_data.fw_lock);
1366                 return -ESHUTDOWN;
1367         }
1368
1369         if (sfw_del_session_timer() != 0) {
1370                 CERROR("dropping RPC %s from %s: racing with expiry timer: rc = %d\n",
1371                        sv->sv_name, libcfs_id2str(rpc->srpc_peer), -EAGAIN);
1372                 spin_unlock(&sfw_data.fw_lock);
1373                 return -EAGAIN;
1374         }
1375
1376         sfw_data.fw_active_srpc = rpc;
1377         spin_unlock(&sfw_data.fw_lock);
1378
1379         rc = sfw_add_test(rpc);
1380
1381         spin_lock(&sfw_data.fw_lock);
1382
1383         if (!sfw_data.fw_shuttingdown)
1384                 sfw_add_session_timer();
1385
1386         sfw_data.fw_active_srpc = NULL;
1387         spin_unlock(&sfw_data.fw_lock);
1388         return rc;
1389 }
1390
1391 struct srpc_client_rpc *
1392 sfw_create_rpc(struct lnet_process_id peer, int service,
1393                unsigned features, int nbulkiov, int bulklen,
1394                void (*done)(struct srpc_client_rpc *), void *priv)
1395 {
1396         struct srpc_client_rpc *rpc = NULL;
1397
1398         spin_lock(&sfw_data.fw_lock);
1399
1400         LASSERT(!sfw_data.fw_shuttingdown);
1401         LASSERT(service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1402
1403         if (nbulkiov == 0 && !list_empty(&sfw_data.fw_zombie_rpcs)) {
1404                 rpc = list_first_entry(&sfw_data.fw_zombie_rpcs,
1405                                        struct srpc_client_rpc, crpc_list);
1406                 list_del(&rpc->crpc_list);
1407         }
1408         spin_unlock(&sfw_data.fw_lock);
1409
1410         if (rpc) {
1411                 /* Ensure that rpc is done */
1412                 swi_cancel_workitem(&rpc->crpc_wi);
1413                 srpc_init_client_rpc(rpc, peer, service, 0, 0,
1414                                      done, sfw_client_rpc_fini, priv);
1415         } else {
1416                 rpc = srpc_create_client_rpc(peer, service,
1417                                              nbulkiov, bulklen, done,
1418                                              nbulkiov != 0 ?  NULL :
1419                                              sfw_client_rpc_fini,
1420                                              priv);
1421         }
1422
1423         if (rpc != NULL) /* "session" is concept in framework */
1424                 rpc->crpc_reqstmsg.msg_ses_feats = features;
1425
1426         return rpc;
1427 }
1428
1429 void
1430 sfw_unpack_message(struct srpc_msg *msg)
1431 {
1432         if (msg->msg_magic == SRPC_MSG_MAGIC)
1433                 return; /* no flipping needed */
1434
1435         /* srpc module should guarantee I wouldn't get crap */
1436         LASSERT(msg->msg_magic == __swab32(SRPC_MSG_MAGIC));
1437
1438         if (msg->msg_type == SRPC_MSG_STAT_REQST) {
1439                 struct srpc_stat_reqst *req = &msg->msg_body.stat_reqst;
1440
1441                 __swab32s(&req->str_type);
1442                 __swab64s(&req->str_rpyid);
1443                 sfw_unpack_sid(req->str_sid);
1444                 return;
1445         }
1446
1447         if (msg->msg_type == SRPC_MSG_STAT_REPLY) {
1448                 struct srpc_stat_reply *rep = &msg->msg_body.stat_reply;
1449
1450                 __swab32s(&rep->str_status);
1451                 sfw_unpack_sid(rep->str_sid);
1452                 sfw_unpack_fw_counters(rep->str_fw);
1453                 sfw_unpack_rpc_counters(rep->str_rpc);
1454                 sfw_unpack_lnet_counters(rep->str_lnet);
1455                 return;
1456         }
1457
1458         if (msg->msg_type == SRPC_MSG_MKSN_REQST) {
1459                 struct srpc_mksn_reqst *req = &msg->msg_body.mksn_reqst;
1460
1461                 __swab64s(&req->mksn_rpyid);
1462                 __swab32s(&req->mksn_force);
1463                 sfw_unpack_sid(req->mksn_sid);
1464                 return;
1465         }
1466
1467         if (msg->msg_type == SRPC_MSG_MKSN_REPLY) {
1468                 struct srpc_mksn_reply *rep = &msg->msg_body.mksn_reply;
1469
1470                 __swab32s(&rep->mksn_status);
1471                 __swab32s(&rep->mksn_timeout);
1472                 sfw_unpack_sid(rep->mksn_sid);
1473                 return;
1474         }
1475
1476         if (msg->msg_type == SRPC_MSG_RMSN_REQST) {
1477                 struct srpc_rmsn_reqst *req = &msg->msg_body.rmsn_reqst;
1478
1479                 __swab64s(&req->rmsn_rpyid);
1480                 sfw_unpack_sid(req->rmsn_sid);
1481                 return;
1482         }
1483
1484         if (msg->msg_type == SRPC_MSG_RMSN_REPLY) {
1485                 struct srpc_rmsn_reply *rep = &msg->msg_body.rmsn_reply;
1486
1487                 __swab32s(&rep->rmsn_status);
1488                 sfw_unpack_sid(rep->rmsn_sid);
1489                 return;
1490         }
1491
1492         if (msg->msg_type == SRPC_MSG_DEBUG_REQST) {
1493                 struct srpc_debug_reqst *req = &msg->msg_body.dbg_reqst;
1494
1495                 __swab64s(&req->dbg_rpyid);
1496                 __swab32s(&req->dbg_flags);
1497                 sfw_unpack_sid(req->dbg_sid);
1498                 return;
1499         }
1500
1501         if (msg->msg_type == SRPC_MSG_DEBUG_REPLY) {
1502                 struct srpc_debug_reply *rep = &msg->msg_body.dbg_reply;
1503
1504                 __swab32s(&rep->dbg_nbatch);
1505                 __swab32s(&rep->dbg_timeout);
1506                 sfw_unpack_sid(rep->dbg_sid);
1507                 return;
1508         }
1509
1510         if (msg->msg_type == SRPC_MSG_BATCH_REQST) {
1511                 struct srpc_batch_reqst *req = &msg->msg_body.bat_reqst;
1512
1513                 __swab32s(&req->bar_opc);
1514                 __swab64s(&req->bar_rpyid);
1515                 __swab32s(&req->bar_testidx);
1516                 __swab32s(&req->bar_arg);
1517                 sfw_unpack_sid(req->bar_sid);
1518                 __swab64s(&req->bar_bid.bat_id);
1519                 return;
1520         }
1521
1522         if (msg->msg_type == SRPC_MSG_BATCH_REPLY) {
1523                 struct srpc_batch_reply *rep = &msg->msg_body.bat_reply;
1524
1525                 __swab32s(&rep->bar_status);
1526                 sfw_unpack_sid(rep->bar_sid);
1527                 return;
1528         }
1529
1530         if (msg->msg_type == SRPC_MSG_TEST_REQST) {
1531                 struct srpc_test_reqst *req = &msg->msg_body.tes_reqst;
1532
1533                 __swab64s(&req->tsr_rpyid);
1534                 __swab64s(&req->tsr_bulkid);
1535                 __swab32s(&req->tsr_loop);
1536                 __swab32s(&req->tsr_ndest);
1537                 __swab32s(&req->tsr_concur);
1538                 __swab32s(&req->tsr_service);
1539                 sfw_unpack_sid(req->tsr_sid);
1540                 __swab64s(&req->tsr_bid.bat_id);
1541                 return;
1542         }
1543
1544         if (msg->msg_type == SRPC_MSG_TEST_REPLY) {
1545                 struct srpc_test_reply *rep = &msg->msg_body.tes_reply;
1546
1547                 __swab32s(&rep->tsr_status);
1548                 sfw_unpack_sid(rep->tsr_sid);
1549                 return;
1550         }
1551
1552         if (msg->msg_type == SRPC_MSG_JOIN_REQST) {
1553                 struct srpc_join_reqst *req = &msg->msg_body.join_reqst;
1554
1555                 __swab64s(&req->join_rpyid);
1556                 sfw_unpack_sid(req->join_sid);
1557                 return;
1558         }
1559
1560         if (msg->msg_type == SRPC_MSG_JOIN_REPLY) {
1561                 struct srpc_join_reply *rep = &msg->msg_body.join_reply;
1562
1563                 __swab32s(&rep->join_status);
1564                 __swab32s(&rep->join_timeout);
1565                 sfw_unpack_sid(rep->join_sid);
1566                 return;
1567         }
1568
1569         LBUG();
1570 }
1571
1572 void
1573 sfw_abort_rpc(struct srpc_client_rpc *rpc)
1574 {
1575         LASSERT(atomic_read(&rpc->crpc_refcount) > 0);
1576         LASSERT(rpc->crpc_service <= SRPC_FRAMEWORK_SERVICE_MAX_ID);
1577
1578         spin_lock(&rpc->crpc_lock);
1579         srpc_abort_rpc(rpc, -EINTR);
1580         spin_unlock(&rpc->crpc_lock);
1581 }
1582
1583 void
1584 sfw_post_rpc(struct srpc_client_rpc *rpc)
1585 {
1586         spin_lock(&rpc->crpc_lock);
1587
1588         LASSERT(!rpc->crpc_closed);
1589         LASSERT(!rpc->crpc_aborted);
1590         LASSERT(list_empty(&rpc->crpc_list));
1591         LASSERT(!sfw_data.fw_shuttingdown);
1592
1593         rpc->crpc_timeout = rpc_timeout;
1594         srpc_post_rpc(rpc);
1595
1596         spin_unlock(&rpc->crpc_lock);
1597 }
1598
1599 #if 0
1600 static struct srpc_service sfw_services[] = {
1601         { .sv_id = SRPC_SERVICE_DEBUG,          .sv_name = "debug", },
1602         { .sv_id = SRPC_SERVICE_QUERY_STAT,     .sv_name = "query stats", },
1603         { .sv_id = SRPC_SERVICE_MAKE_SESSION,   .sv_name = "make session", },
1604         { .sv_id = SRPC_SERVICE_REMOVE_SESSION, .sv_name = "remove session", },
1605         { .sv_id = SRPC_SERVICE_BATCH,          .sv_name = "batch service", },
1606         { .sv_id = SRPC_SERVICE_TEST,           .sv_name = "test service", },
1607         { .sv_id = 0, } };
1608 #endif
1609
1610 int
1611 sfw_startup(void)
1612 {
1613         struct srpc_service *sv;
1614         struct sfw_test_case *tsc;
1615         int i, rc, error;
1616
1617         if (session_timeout < 0) {
1618                 CERROR("Session timeout must be non-negative: %d: rc = %d\n",
1619                        session_timeout, -EINVAL);
1620                 return -EINVAL;
1621         }
1622
1623         if (rpc_timeout < 0) {
1624                 CERROR("RPC timeout must be non-negative: %d: rc = %d\n",
1625                        rpc_timeout, -EINVAL);
1626                 return -EINVAL;
1627         }
1628
1629         if (session_timeout == 0)
1630                 CWARN("Zero session_timeout specified - test sessions never expire.\n");
1631
1632         if (rpc_timeout == 0)
1633                 CWARN("Zero rpc_timeout specified - test RPC never expire.\n");
1634
1635         memset(&sfw_data, 0, sizeof(struct smoketest_framework));
1636
1637         sfw_data.fw_session     = NULL;
1638         sfw_data.fw_active_srpc = NULL;
1639         spin_lock_init(&sfw_data.fw_lock);
1640         atomic_set(&sfw_data.fw_nzombies, 0);
1641         INIT_LIST_HEAD(&sfw_data.fw_tests);
1642         INIT_LIST_HEAD(&sfw_data.fw_zombie_rpcs);
1643         INIT_LIST_HEAD(&sfw_data.fw_zombie_sessions);
1644
1645         brw_init_test_service();
1646         rc = sfw_register_test(&brw_test_service, &brw_test_client);
1647         LASSERT(rc == 0);
1648
1649         ping_init_test_client();
1650         ping_init_test_service();
1651         rc = sfw_register_test(&ping_test_service, &ping_test_client);
1652         LASSERT(rc == 0);
1653
1654         error = 0;
1655         list_for_each_entry(tsc, &sfw_data.fw_tests, tsc_list) {
1656                 sv = tsc->tsc_srv_service;
1657
1658                 rc = srpc_add_service(sv);
1659                 LASSERT(rc != -EBUSY);
1660                 if (rc != 0) {
1661                         CWARN("Failed to add %s service: %d\n",
1662                               sv->sv_name, rc);
1663                         error = rc;
1664                 }
1665         }
1666
1667         for (i = 0; ; i++) {
1668                 sv = &sfw_services[i];
1669                 if (sv->sv_name == NULL)
1670                         break;
1671
1672                 sv->sv_bulk_ready = NULL;
1673                 sv->sv_handler    = sfw_handle_server_rpc;
1674                 sv->sv_wi_total   = SFW_FRWK_WI_MAX;
1675                 if (sv->sv_id == SRPC_SERVICE_TEST)
1676                         sv->sv_bulk_ready = sfw_bulk_ready;
1677
1678                 rc = srpc_add_service(sv);
1679                 LASSERT(rc != -EBUSY);
1680                 if (rc != 0) {
1681                         CWARN("Failed to add %s service: %d\n", sv->sv_name,
1682                               rc);
1683                         error = rc;
1684                 }
1685
1686                 /* about to sfw_shutdown, no need to add buffer */
1687                 if (error)
1688                         continue;
1689
1690                 rc = srpc_service_add_buffers(sv, sv->sv_wi_total);
1691                 if (rc != 0) {
1692                         CWARN("Failed to reserve enough buffers: service %s, %d needed: %d\n",
1693                               sv->sv_name, sv->sv_wi_total, rc);
1694                         error = -ENOMEM;
1695                 }
1696         }
1697
1698         if (error != 0)
1699                 sfw_shutdown();
1700         return error;
1701 }
1702
1703 void
1704 sfw_shutdown(void)
1705 {
1706         struct srpc_service *sv;
1707         struct sfw_test_case *tsc;
1708         int i;
1709
1710         spin_lock(&sfw_data.fw_lock);
1711
1712         sfw_data.fw_shuttingdown = 1;
1713         lst_wait_until(sfw_data.fw_active_srpc == NULL, sfw_data.fw_lock,
1714                        "waiting for active RPC to finish.\n");
1715
1716         if (sfw_del_session_timer() != 0)
1717                 lst_wait_until(sfw_data.fw_session == NULL, sfw_data.fw_lock,
1718                                "waiting for session timer to explode.\n");
1719
1720         sfw_deactivate_session();
1721         lst_wait_until(atomic_read(&sfw_data.fw_nzombies) == 0,
1722                        sfw_data.fw_lock,
1723                        "waiting for %d zombie sessions to die.\n",
1724                        atomic_read(&sfw_data.fw_nzombies));
1725
1726         spin_unlock(&sfw_data.fw_lock);
1727
1728         for (i = 0; ; i++) {
1729                 sv = &sfw_services[i];
1730                 if (sv->sv_name == NULL)
1731                         break;
1732
1733                 srpc_shutdown_service(sv);
1734                 srpc_remove_service(sv);
1735         }
1736
1737         list_for_each_entry(tsc, &sfw_data.fw_tests, tsc_list) {
1738                 sv = tsc->tsc_srv_service;
1739                 srpc_shutdown_service(sv);
1740                 srpc_remove_service(sv);
1741         }
1742
1743         while (!list_empty(&sfw_data.fw_zombie_rpcs)) {
1744                 struct srpc_client_rpc *rpc;
1745
1746                 rpc = list_first_entry(&sfw_data.fw_zombie_rpcs,
1747                                        struct srpc_client_rpc, crpc_list);
1748                 list_del(&rpc->crpc_list);
1749
1750                 LIBCFS_FREE(rpc, srpc_client_rpc_size(rpc));
1751         }
1752
1753         for (i = 0; ; i++) {
1754                 sv = &sfw_services[i];
1755                 if (sv->sv_name == NULL)
1756                         break;
1757
1758                 srpc_wait_service_shutdown(sv);
1759         }
1760
1761         while (!list_empty(&sfw_data.fw_tests)) {
1762                 tsc = list_first_entry(&sfw_data.fw_tests,
1763                                        struct sfw_test_case, tsc_list);
1764
1765                 srpc_wait_service_shutdown(tsc->tsc_srv_service);
1766
1767                 list_del(&tsc->tsc_list);
1768                 LIBCFS_FREE(tsc, sizeof(*tsc));
1769         }
1770 }