lnet/selftest/framework.c

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