drivers/staging/lustre/lnet/selftest/conrpc.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2012, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lnet/selftest/conctl.c
  33  *
  34  * Console framework rpcs
  35  *
  36  * Author: Liang Zhen <liang@whamcloud.com>
  37  */
  38
  39 #include "../../include/linux/libcfs/libcfs.h"
  40 #include "../../include/linux/lnet/lib-lnet.h"
  41 #include "timer.h"
  42 #include "conrpc.h"
  43 #include "console.h"
  44
  45 void lstcon_rpc_stat_reply(struct lstcon_rpc_trans *, struct srpc_msg *,
  46                            struct lstcon_node *, lstcon_trans_stat_t *);
  47
  48 static void
  49 lstcon_rpc_done(struct srpc_client_rpc *rpc)
  50 {
  51         struct lstcon_rpc *crpc = (struct lstcon_rpc *)rpc->crpc_priv;
  52
  53         LASSERT(crpc && rpc == crpc->crp_rpc);
  54         LASSERT(crpc->crp_posted && !crpc->crp_finished);
  55
  56         spin_lock(&rpc->crpc_lock);
  57
  58         if (!crpc->crp_trans) {
  59                 /*
  60                  * Orphan RPC is not in any transaction,
  61                  * I'm just a poor body and nobody loves me
  62                  */
  63                 spin_unlock(&rpc->crpc_lock);
  64
  65                 /* release it */
  66                 lstcon_rpc_put(crpc);
  67                 return;
  68         }
  69
  70         /* not an orphan RPC */
  71         crpc->crp_finished = 1;
  72
  73         if (!crpc->crp_stamp) {
  74                 /* not aborted */
  75                 LASSERT(!crpc->crp_status);
  76
  77                 crpc->crp_stamp = cfs_time_current();
  78                 crpc->crp_status = rpc->crpc_status;
  79         }
  80
  81         /* wakeup (transaction)thread if I'm the last RPC in the transaction */
  82         if (atomic_dec_and_test(&crpc->crp_trans->tas_remaining))
  83                 wake_up(&crpc->crp_trans->tas_waitq);
  84
  85         spin_unlock(&rpc->crpc_lock);
  86 }
  87
  88 static int
  89 lstcon_rpc_init(struct lstcon_node *nd, int service, unsigned feats,
  90                 int bulk_npg, int bulk_len, int embedded, struct lstcon_rpc *crpc)
  91 {
  92         crpc->crp_rpc = sfw_create_rpc(nd->nd_id, service,
  93                                        feats, bulk_npg, bulk_len,
  94                                        lstcon_rpc_done, (void *)crpc);
  95         if (!crpc->crp_rpc)
  96                 return -ENOMEM;
  97
  98         crpc->crp_trans = NULL;
  99         crpc->crp_node = nd;
 100         crpc->crp_posted = 0;
 101         crpc->crp_finished = 0;
 102         crpc->crp_unpacked = 0;
 103         crpc->crp_status = 0;
 104         crpc->crp_stamp = 0;
 105         crpc->crp_embedded = embedded;
 106         INIT_LIST_HEAD(&crpc->crp_link);
 107
 108         atomic_inc(&console_session.ses_rpc_counter);
 109
 110         return 0;
 111 }
 112
 113 static int
 114 lstcon_rpc_prep(struct lstcon_node *nd, int service, unsigned feats,
 115                 int bulk_npg, int bulk_len, struct lstcon_rpc **crpcpp)
 116 {
 117         struct lstcon_rpc *crpc = NULL;
 118         int rc;
 119
 120         spin_lock(&console_session.ses_rpc_lock);
 121
 122         crpc = list_first_entry_or_null(&console_session.ses_rpc_freelist,
 123                                         struct lstcon_rpc, crp_link);
 124         if (crpc)
 125                 list_del_init(&crpc->crp_link);
 126
 127         spin_unlock(&console_session.ses_rpc_lock);
 128
 129         if (!crpc) {
 130                 LIBCFS_ALLOC(crpc, sizeof(*crpc));
 131                 if (!crpc)
 132                         return -ENOMEM;
 133         }
 134
 135         rc = lstcon_rpc_init(nd, service, feats, bulk_npg, bulk_len, 0, crpc);
 136         if (!rc) {
 137                 *crpcpp = crpc;
 138                 return 0;
 139         }
 140
 141         LIBCFS_FREE(crpc, sizeof(*crpc));
 142
 143         return rc;
 144 }
 145
 146 void
 147 lstcon_rpc_put(struct lstcon_rpc *crpc)
 148 {
 149         struct srpc_bulk *bulk = &crpc->crp_rpc->crpc_bulk;
 150         int i;
 151
 152         LASSERT(list_empty(&crpc->crp_link));
 153
 154         for (i = 0; i < bulk->bk_niov; i++) {
 155                 if (!bulk->bk_iovs[i].bv_page)
 156                         continue;
 157
 158                 __free_page(bulk->bk_iovs[i].bv_page);
 159         }
 160
 161         srpc_client_rpc_decref(crpc->crp_rpc);
 162
 163         if (crpc->crp_embedded) {
 164                 /* embedded RPC, don't recycle it */
 165                 memset(crpc, 0, sizeof(*crpc));
 166                 crpc->crp_embedded = 1;
 167
 168         } else {
 169                 spin_lock(&console_session.ses_rpc_lock);
 170
 171                 list_add(&crpc->crp_link,
 172                          &console_session.ses_rpc_freelist);
 173
 174                 spin_unlock(&console_session.ses_rpc_lock);
 175         }
 176
 177         /* RPC is not alive now */
 178         atomic_dec(&console_session.ses_rpc_counter);
 179 }
 180
 181 static void
 182 lstcon_rpc_post(struct lstcon_rpc *crpc)
 183 {
 184         struct lstcon_rpc_trans *trans = crpc->crp_trans;
 185
 186         LASSERT(trans);
 187
 188         atomic_inc(&trans->tas_remaining);
 189         crpc->crp_posted = 1;
 190
 191         sfw_post_rpc(crpc->crp_rpc);
 192 }
 193
 194 static char *
 195 lstcon_rpc_trans_name(int transop)
 196 {
 197         if (transop == LST_TRANS_SESNEW)
 198                 return "SESNEW";
 199
 200         if (transop == LST_TRANS_SESEND)
 201                 return "SESEND";
 202
 203         if (transop == LST_TRANS_SESQRY)
 204                 return "SESQRY";
 205
 206         if (transop == LST_TRANS_SESPING)
 207                 return "SESPING";
 208
 209         if (transop == LST_TRANS_TSBCLIADD)
 210                 return "TSBCLIADD";
 211
 212         if (transop == LST_TRANS_TSBSRVADD)
 213                 return "TSBSRVADD";
 214
 215         if (transop == LST_TRANS_TSBRUN)
 216                 return "TSBRUN";
 217
 218         if (transop == LST_TRANS_TSBSTOP)
 219                 return "TSBSTOP";
 220
 221         if (transop == LST_TRANS_TSBCLIQRY)
 222                 return "TSBCLIQRY";
 223
 224         if (transop == LST_TRANS_TSBSRVQRY)
 225                 return "TSBSRVQRY";
 226
 227         if (transop == LST_TRANS_STATQRY)
 228                 return "STATQRY";
 229
 230         return "Unknown";
 231 }
 232
 233 int
 234 lstcon_rpc_trans_prep(struct list_head *translist, int transop,
 235                       struct lstcon_rpc_trans **transpp)
 236 {
 237         struct lstcon_rpc_trans *trans;
 238
 239         if (translist) {
 240                 list_for_each_entry(trans, translist, tas_link) {
 241                         /*
 242                          * Can't enqueue two private transaction on
 243                          * the same object
 244                          */
 245                         if ((trans->tas_opc & transop) == LST_TRANS_PRIVATE)
 246                                 return -EPERM;
 247                 }
 248         }
 249
 250         /* create a trans group */
 251         LIBCFS_ALLOC(trans, sizeof(*trans));
 252         if (!trans)
 253                 return -ENOMEM;
 254
 255         trans->tas_opc = transop;
 256
 257         if (!translist)
 258                 INIT_LIST_HEAD(&trans->tas_olink);
 259         else
 260                 list_add_tail(&trans->tas_olink, translist);
 261
 262         list_add_tail(&trans->tas_link, &console_session.ses_trans_list);
 263
 264         INIT_LIST_HEAD(&trans->tas_rpcs_list);
 265         atomic_set(&trans->tas_remaining, 0);
 266         init_waitqueue_head(&trans->tas_waitq);
 267
 268         spin_lock(&console_session.ses_rpc_lock);
 269         trans->tas_features = console_session.ses_features;
 270         spin_unlock(&console_session.ses_rpc_lock);
 271
 272         *transpp = trans;
 273         return 0;
 274 }
 275
 276 void
 277 lstcon_rpc_trans_addreq(struct lstcon_rpc_trans *trans, struct lstcon_rpc *crpc)
 278 {
 279         list_add_tail(&crpc->crp_link, &trans->tas_rpcs_list);
 280         crpc->crp_trans = trans;
 281 }
 282
 283 void
 284 lstcon_rpc_trans_abort(struct lstcon_rpc_trans *trans, int error)
 285 {
 286         struct srpc_client_rpc *rpc;
 287         struct lstcon_rpc *crpc;
 288         struct lstcon_node *nd;
 289
 290         list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
 291                 rpc = crpc->crp_rpc;
 292
 293                 spin_lock(&rpc->crpc_lock);
 294
 295                 if (!crpc->crp_posted ||        /* not posted */
 296                     crpc->crp_stamp) {          /* rpc done or aborted already */
 297                         if (!crpc->crp_stamp) {
 298                                 crpc->crp_stamp = cfs_time_current();
 299                                 crpc->crp_status = -EINTR;
 300                         }
 301                         spin_unlock(&rpc->crpc_lock);
 302                         continue;
 303                 }
 304
 305                 crpc->crp_stamp = cfs_time_current();
 306                 crpc->crp_status = error;
 307
 308                 spin_unlock(&rpc->crpc_lock);
 309
 310                 sfw_abort_rpc(rpc);
 311
 312                 if (error != -ETIMEDOUT)
 313                         continue;
 314
 315                 nd = crpc->crp_node;
 316                 if (cfs_time_after(nd->nd_stamp, crpc->crp_stamp))
 317                         continue;
 318
 319                 nd->nd_stamp = crpc->crp_stamp;
 320                 nd->nd_state = LST_NODE_DOWN;
 321         }
 322 }
 323
 324 static int
 325 lstcon_rpc_trans_check(struct lstcon_rpc_trans *trans)
 326 {
 327         if (console_session.ses_shutdown &&
 328             !list_empty(&trans->tas_olink)) /* Not an end session RPC */
 329                 return 1;
 330
 331         return !atomic_read(&trans->tas_remaining) ? 1 : 0;
 332 }
 333
 334 int
 335 lstcon_rpc_trans_postwait(struct lstcon_rpc_trans *trans, int timeout)
 336 {
 337         struct lstcon_rpc *crpc;
 338         int rc;
 339
 340         if (list_empty(&trans->tas_rpcs_list))
 341                 return 0;
 342
 343         if (timeout < LST_TRANS_MIN_TIMEOUT)
 344                 timeout = LST_TRANS_MIN_TIMEOUT;
 345
 346         CDEBUG(D_NET, "Transaction %s started\n",
 347                lstcon_rpc_trans_name(trans->tas_opc));
 348
 349         /* post all requests */
 350         list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
 351                 LASSERT(!crpc->crp_posted);
 352
 353                 lstcon_rpc_post(crpc);
 354         }
 355
 356         mutex_unlock(&console_session.ses_mutex);
 357
 358         rc = wait_event_interruptible_timeout(trans->tas_waitq,
 359                                               lstcon_rpc_trans_check(trans),
 360                                               cfs_time_seconds(timeout));
 361         rc = (rc > 0) ? 0 : ((rc < 0) ? -EINTR : -ETIMEDOUT);
 362
 363         mutex_lock(&console_session.ses_mutex);
 364
 365         if (console_session.ses_shutdown)
 366                 rc = -ESHUTDOWN;
 367
 368         if (rc || atomic_read(&trans->tas_remaining)) {
 369                 /* treat short timeout as canceled */
 370                 if (rc == -ETIMEDOUT && timeout < LST_TRANS_MIN_TIMEOUT * 2)
 371                         rc = -EINTR;
 372
 373                 lstcon_rpc_trans_abort(trans, rc);
 374         }
 375
 376         CDEBUG(D_NET, "Transaction %s stopped: %d\n",
 377                lstcon_rpc_trans_name(trans->tas_opc), rc);
 378
 379         lstcon_rpc_trans_stat(trans, lstcon_trans_stat());
 380
 381         return rc;
 382 }
 383
 384 static int
 385 lstcon_rpc_get_reply(struct lstcon_rpc *crpc, struct srpc_msg **msgpp)
 386 {
 387         struct lstcon_node *nd = crpc->crp_node;
 388         struct srpc_client_rpc *rpc = crpc->crp_rpc;
 389         struct srpc_generic_reply *rep;
 390
 391         LASSERT(nd && rpc);
 392         LASSERT(crpc->crp_stamp);
 393
 394         if (crpc->crp_status) {
 395                 *msgpp = NULL;
 396                 return crpc->crp_status;
 397         }
 398
 399         *msgpp = &rpc->crpc_replymsg;
 400         if (!crpc->crp_unpacked) {
 401                 sfw_unpack_message(*msgpp);
 402                 crpc->crp_unpacked = 1;
 403         }
 404
 405         if (cfs_time_after(nd->nd_stamp, crpc->crp_stamp))
 406                 return 0;
 407
 408         nd->nd_stamp = crpc->crp_stamp;
 409         rep = &(*msgpp)->msg_body.reply;
 410
 411         if (rep->sid.ses_nid == LNET_NID_ANY)
 412                 nd->nd_state = LST_NODE_UNKNOWN;
 413         else if (lstcon_session_match(rep->sid))
 414                 nd->nd_state = LST_NODE_ACTIVE;
 415         else
 416                 nd->nd_state = LST_NODE_BUSY;
 417
 418         return 0;
 419 }
 420
 421 void
 422 lstcon_rpc_trans_stat(struct lstcon_rpc_trans *trans, lstcon_trans_stat_t *stat)
 423 {
 424         struct lstcon_rpc *crpc;
 425         struct srpc_msg *rep;
 426         int error;
 427
 428         LASSERT(stat);
 429
 430         memset(stat, 0, sizeof(*stat));
 431
 432         list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
 433                 lstcon_rpc_stat_total(stat, 1);
 434
 435                 LASSERT(crpc->crp_stamp);
 436
 437                 error = lstcon_rpc_get_reply(crpc, &rep);
 438                 if (error) {
 439                         lstcon_rpc_stat_failure(stat, 1);
 440                         if (!stat->trs_rpc_errno)
 441                                 stat->trs_rpc_errno = -error;
 442
 443                         continue;
 444                 }
 445
 446                 lstcon_rpc_stat_success(stat, 1);
 447
 448                 lstcon_rpc_stat_reply(trans, rep, crpc->crp_node, stat);
 449         }
 450
 451         if (trans->tas_opc == LST_TRANS_SESNEW && !stat->trs_fwk_errno) {
 452                 stat->trs_fwk_errno =
 453                       lstcon_session_feats_check(trans->tas_features);
 454         }
 455
 456         CDEBUG(D_NET, "transaction %s : success %d, failure %d, total %d, RPC error(%d), Framework error(%d)\n",
 457                lstcon_rpc_trans_name(trans->tas_opc),
 458                lstcon_rpc_stat_success(stat, 0),
 459                lstcon_rpc_stat_failure(stat, 0),
 460                lstcon_rpc_stat_total(stat, 0),
 461                stat->trs_rpc_errno, stat->trs_fwk_errno);
 462 }
 463
 464 int
 465 lstcon_rpc_trans_interpreter(struct lstcon_rpc_trans *trans,
 466                              struct list_head __user *head_up,
 467                              lstcon_rpc_readent_func_t readent)
 468 {
 469         struct list_head tmp;
 470         struct list_head __user *next;
 471         lstcon_rpc_ent_t *ent;
 472         struct srpc_generic_reply *rep;
 473         struct lstcon_rpc *crpc;
 474         struct srpc_msg *msg;
 475         struct lstcon_node *nd;
 476         long dur;
 477         struct timeval tv;
 478         int error;
 479
 480         LASSERT(head_up);
 481
 482         next = head_up;
 483
 484         list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
 485                 if (copy_from_user(&tmp, next,
 486                                    sizeof(struct list_head)))
 487                         return -EFAULT;
 488
 489                 if (tmp.next == head_up)
 490                         return 0;
 491
 492                 next = tmp.next;
 493
 494                 ent = list_entry(next, lstcon_rpc_ent_t, rpe_link);
 495
 496                 LASSERT(crpc->crp_stamp);
 497
 498                 error = lstcon_rpc_get_reply(crpc, &msg);
 499
 500                 nd = crpc->crp_node;
 501
 502                 dur = (long)cfs_time_sub(crpc->crp_stamp,
 503                       (unsigned long)console_session.ses_id.ses_stamp);
 504                 jiffies_to_timeval(dur, &tv);
 505
 506                 if (copy_to_user(&ent->rpe_peer, &nd->nd_id,
 507                                  sizeof(lnet_process_id_t)) ||
 508                     copy_to_user(&ent->rpe_stamp, &tv, sizeof(tv)) ||
 509                     copy_to_user(&ent->rpe_state, &nd->nd_state,
 510                                  sizeof(nd->nd_state)) ||
 511                     copy_to_user(&ent->rpe_rpc_errno, &error,
 512                                  sizeof(error)))
 513                         return -EFAULT;
 514
 515                 if (error)
 516                         continue;
 517
 518                 /* RPC is done */
 519                 rep = (struct srpc_generic_reply *)&msg->msg_body.reply;
 520
 521                 if (copy_to_user(&ent->rpe_sid, &rep->sid, sizeof(lst_sid_t)) ||
 522                     copy_to_user(&ent->rpe_fwk_errno, &rep->status,
 523                                  sizeof(rep->status)))
 524                         return -EFAULT;
 525
 526                 if (!readent)
 527                         continue;
 528
 529                 error = readent(trans->tas_opc, msg, ent);
 530                 if (error)
 531                         return error;
 532         }
 533
 534         return 0;
 535 }
 536
 537 void
 538 lstcon_rpc_trans_destroy(struct lstcon_rpc_trans *trans)
 539 {
 540         struct srpc_client_rpc *rpc;
 541         struct lstcon_rpc *crpc;
 542         struct lstcon_rpc *tmp;
 543         int count = 0;
 544
 545         list_for_each_entry_safe(crpc, tmp, &trans->tas_rpcs_list, crp_link) {
 546                 rpc = crpc->crp_rpc;
 547
 548                 spin_lock(&rpc->crpc_lock);
 549
 550                 /* free it if not posted or finished already */
 551                 if (!crpc->crp_posted || crpc->crp_finished) {
 552                         spin_unlock(&rpc->crpc_lock);
 553
 554                         list_del_init(&crpc->crp_link);
 555                         lstcon_rpc_put(crpc);
 556
 557                         continue;
 558                 }
 559
 560                 /*
 561                  * rpcs can be still not callbacked (even LNetMDUnlink is
 562                  * called) because huge timeout for inaccessible network,
 563                  * don't make user wait for them, just abandon them, they
 564                  * will be recycled in callback
 565                  */
 566                 LASSERT(crpc->crp_status);
 567
 568                 crpc->crp_node = NULL;
 569                 crpc->crp_trans = NULL;
 570                 list_del_init(&crpc->crp_link);
 571                 count++;
 572
 573                 spin_unlock(&rpc->crpc_lock);
 574
 575                 atomic_dec(&trans->tas_remaining);
 576         }
 577
 578         LASSERT(!atomic_read(&trans->tas_remaining));
 579
 580         list_del(&trans->tas_link);
 581         if (!list_empty(&trans->tas_olink))
 582                 list_del(&trans->tas_olink);
 583
 584         CDEBUG(D_NET, "Transaction %s destroyed with %d pending RPCs\n",
 585                lstcon_rpc_trans_name(trans->tas_opc), count);
 586
 587         LIBCFS_FREE(trans, sizeof(*trans));
 588 }
 589
 590 int
 591 lstcon_sesrpc_prep(struct lstcon_node *nd, int transop,
 592                    unsigned feats, struct lstcon_rpc **crpc)
 593 {
 594         struct srpc_mksn_reqst *msrq;
 595         struct srpc_rmsn_reqst *rsrq;
 596         int rc;
 597
 598         switch (transop) {
 599         case LST_TRANS_SESNEW:
 600                 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_MAKE_SESSION,
 601                                      feats, 0, 0, crpc);
 602                 if (rc)
 603                         return rc;
 604
 605                 msrq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.mksn_reqst;
 606                 msrq->mksn_sid = console_session.ses_id;
 607                 msrq->mksn_force = console_session.ses_force;
 608                 strlcpy(msrq->mksn_name, console_session.ses_name,
 609                         sizeof(msrq->mksn_name));
 610                 break;
 611
 612         case LST_TRANS_SESEND:
 613                 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_REMOVE_SESSION,
 614                                      feats, 0, 0, crpc);
 615                 if (rc)
 616                         return rc;
 617
 618                 rsrq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.rmsn_reqst;
 619                 rsrq->rmsn_sid = console_session.ses_id;
 620                 break;
 621
 622         default:
 623                 LBUG();
 624         }
 625
 626         return 0;
 627 }
 628
 629 int
 630 lstcon_dbgrpc_prep(struct lstcon_node *nd, unsigned feats, struct lstcon_rpc **crpc)
 631 {
 632         struct srpc_debug_reqst *drq;
 633         int rc;
 634
 635         rc = lstcon_rpc_prep(nd, SRPC_SERVICE_DEBUG, feats, 0, 0, crpc);
 636         if (rc)
 637                 return rc;
 638
 639         drq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.dbg_reqst;
 640
 641         drq->dbg_sid = console_session.ses_id;
 642         drq->dbg_flags = 0;
 643
 644         return rc;
 645 }
 646
 647 int
 648 lstcon_batrpc_prep(struct lstcon_node *nd, int transop, unsigned feats,
 649                    struct lstcon_tsb_hdr *tsb, struct lstcon_rpc **crpc)
 650 {
 651         struct lstcon_batch *batch;
 652         struct srpc_batch_reqst *brq;
 653         int rc;
 654
 655         rc = lstcon_rpc_prep(nd, SRPC_SERVICE_BATCH, feats, 0, 0, crpc);
 656         if (rc)
 657                 return rc;
 658
 659         brq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.bat_reqst;
 660
 661         brq->bar_sid = console_session.ses_id;
 662         brq->bar_bid = tsb->tsb_id;
 663         brq->bar_testidx = tsb->tsb_index;
 664         brq->bar_opc = transop == LST_TRANS_TSBRUN ? SRPC_BATCH_OPC_RUN :
 665                        (transop == LST_TRANS_TSBSTOP ? SRPC_BATCH_OPC_STOP :
 666                        SRPC_BATCH_OPC_QUERY);
 667
 668         if (transop != LST_TRANS_TSBRUN &&
 669             transop != LST_TRANS_TSBSTOP)
 670                 return 0;
 671
 672         LASSERT(!tsb->tsb_index);
 673
 674         batch = (struct lstcon_batch *)tsb;
 675         brq->bar_arg = batch->bat_arg;
 676
 677         return 0;
 678 }
 679
 680 int
 681 lstcon_statrpc_prep(struct lstcon_node *nd, unsigned feats, struct lstcon_rpc **crpc)
 682 {
 683         struct srpc_stat_reqst *srq;
 684         int rc;
 685
 686         rc = lstcon_rpc_prep(nd, SRPC_SERVICE_QUERY_STAT, feats, 0, 0, crpc);
 687         if (rc)
 688                 return rc;
 689
 690         srq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.stat_reqst;
 691
 692         srq->str_sid = console_session.ses_id;
 693         srq->str_type = 0; /* XXX remove it */
 694
 695         return 0;
 696 }
 697
 698 static lnet_process_id_packed_t *
 699 lstcon_next_id(int idx, int nkiov, lnet_kiov_t *kiov)
 700 {
 701         lnet_process_id_packed_t *pid;
 702         int i;
 703
 704         i = idx / SFW_ID_PER_PAGE;
 705
 706         LASSERT(i < nkiov);
 707
 708         pid = (lnet_process_id_packed_t *)page_address(kiov[i].bv_page);
 709
 710         return &pid[idx % SFW_ID_PER_PAGE];
 711 }
 712
 713 static int
 714 lstcon_dstnodes_prep(struct lstcon_group *grp, int idx,
 715                      int dist, int span, int nkiov, lnet_kiov_t *kiov)
 716 {
 717         lnet_process_id_packed_t *pid;
 718         struct lstcon_ndlink *ndl;
 719         struct lstcon_node *nd;
 720         int start;
 721         int end;
 722         int i = 0;
 723
 724         LASSERT(dist >= 1);
 725         LASSERT(span >= 1);
 726         LASSERT(grp->grp_nnode >= 1);
 727
 728         if (span > grp->grp_nnode)
 729                 return -EINVAL;
 730
 731         start = ((idx / dist) * span) % grp->grp_nnode;
 732         end = ((idx / dist) * span + span - 1) % grp->grp_nnode;
 733
 734         list_for_each_entry(ndl, &grp->grp_ndl_list, ndl_link) {
 735                 nd = ndl->ndl_node;
 736                 if (i < start) {
 737                         i++;
 738                         continue;
 739                 }
 740
 741                 if (i > (end >= start ? end : grp->grp_nnode))
 742                         break;
 743
 744                 pid = lstcon_next_id((i - start), nkiov, kiov);
 745                 pid->nid = nd->nd_id.nid;
 746                 pid->pid = nd->nd_id.pid;
 747                 i++;
 748         }
 749
 750         if (start <= end) /* done */
 751                 return 0;
 752
 753         list_for_each_entry(ndl, &grp->grp_ndl_list, ndl_link) {
 754                 if (i > grp->grp_nnode + end)
 755                         break;
 756
 757                 nd = ndl->ndl_node;
 758                 pid = lstcon_next_id((i - start), nkiov, kiov);
 759                 pid->nid = nd->nd_id.nid;
 760                 pid->pid = nd->nd_id.pid;
 761                 i++;
 762         }
 763
 764         return 0;
 765 }
 766
 767 static int
 768 lstcon_pingrpc_prep(lst_test_ping_param_t *param, struct srpc_test_reqst *req)
 769 {
 770         struct test_ping_req *prq = &req->tsr_u.ping;
 771
 772         prq->png_size = param->png_size;
 773         prq->png_flags = param->png_flags;
 774         /* TODO dest */
 775         return 0;
 776 }
 777
 778 static int
 779 lstcon_bulkrpc_v0_prep(lst_test_bulk_param_t *param, struct srpc_test_reqst *req)
 780 {
 781         struct test_bulk_req *brq = &req->tsr_u.bulk_v0;
 782
 783         brq->blk_opc = param->blk_opc;
 784         brq->blk_npg = (param->blk_size + PAGE_SIZE - 1) /
 785                         PAGE_SIZE;
 786         brq->blk_flags = param->blk_flags;
 787
 788         return 0;
 789 }
 790
 791 static int
 792 lstcon_bulkrpc_v1_prep(lst_test_bulk_param_t *param, struct srpc_test_reqst *req)
 793 {
 794         struct test_bulk_req_v1 *brq = &req->tsr_u.bulk_v1;
 795
 796         brq->blk_opc = param->blk_opc;
 797         brq->blk_flags = param->blk_flags;
 798         brq->blk_len = param->blk_size;
 799         brq->blk_offset = 0; /* reserved */
 800
 801         return 0;
 802 }
 803
 804 int
 805 lstcon_testrpc_prep(struct lstcon_node *nd, int transop, unsigned feats,
 806                     struct lstcon_test *test, struct lstcon_rpc **crpc)
 807 {
 808         struct lstcon_group *sgrp = test->tes_src_grp;
 809         struct lstcon_group *dgrp = test->tes_dst_grp;
 810         struct srpc_test_reqst *trq;
 811         struct srpc_bulk *bulk;
 812         int i;
 813         int npg = 0;
 814         int nob = 0;
 815         int rc = 0;
 816
 817         if (transop == LST_TRANS_TSBCLIADD) {
 818                 npg = sfw_id_pages(test->tes_span);
 819                 nob = !(feats & LST_FEAT_BULK_LEN) ?
 820                       npg * PAGE_SIZE :
 821                       sizeof(lnet_process_id_packed_t) * test->tes_span;
 822         }
 823
 824         rc = lstcon_rpc_prep(nd, SRPC_SERVICE_TEST, feats, npg, nob, crpc);
 825         if (rc)
 826                 return rc;
 827
 828         trq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.tes_reqst;
 829
 830         if (transop == LST_TRANS_TSBSRVADD) {
 831                 int ndist = (sgrp->grp_nnode + test->tes_dist - 1) /
 832                             test->tes_dist;
 833                 int nspan = (dgrp->grp_nnode + test->tes_span - 1) /
 834                             test->tes_span;
 835                 int nmax = (ndist + nspan - 1) / nspan;
 836
 837                 trq->tsr_ndest = 0;
 838                 trq->tsr_loop = nmax * test->tes_dist * test->tes_concur;
 839         } else {
 840                 bulk = &(*crpc)->crp_rpc->crpc_bulk;
 841
 842                 for (i = 0; i < npg; i++) {
 843                         int len;
 844
 845                         LASSERT(nob > 0);
 846
 847                         len = !(feats & LST_FEAT_BULK_LEN) ?
 848                               PAGE_SIZE :
 849                               min_t(int, nob, PAGE_SIZE);
 850                         nob -= len;
 851
 852                         bulk->bk_iovs[i].bv_offset = 0;
 853                         bulk->bk_iovs[i].bv_len = len;
 854                         bulk->bk_iovs[i].bv_page = alloc_page(GFP_KERNEL);
 855
 856                         if (!bulk->bk_iovs[i].bv_page) {
 857                                 lstcon_rpc_put(*crpc);
 858                                 return -ENOMEM;
 859                         }
 860                 }
 861
 862                 bulk->bk_sink = 0;
 863
 864                 LASSERT(transop == LST_TRANS_TSBCLIADD);
 865
 866                 rc = lstcon_dstnodes_prep(test->tes_dst_grp,
 867                                           test->tes_cliidx++,
 868                                           test->tes_dist,
 869                                           test->tes_span,
 870                                           npg, &bulk->bk_iovs[0]);
 871                 if (rc) {
 872                         lstcon_rpc_put(*crpc);
 873                         return rc;
 874                 }
 875
 876                 trq->tsr_ndest = test->tes_span;
 877                 trq->tsr_loop = test->tes_loop;
 878         }
 879
 880         trq->tsr_sid = console_session.ses_id;
 881         trq->tsr_bid = test->tes_hdr.tsb_id;
 882         trq->tsr_concur = test->tes_concur;
 883         trq->tsr_is_client = (transop == LST_TRANS_TSBCLIADD) ? 1 : 0;
 884         trq->tsr_stop_onerr = !!test->tes_stop_onerr;
 885
 886         switch (test->tes_type) {
 887         case LST_TEST_PING:
 888                 trq->tsr_service = SRPC_SERVICE_PING;
 889                 rc = lstcon_pingrpc_prep((lst_test_ping_param_t *)
 890                                          &test->tes_param[0], trq);
 891                 break;
 892
 893         case LST_TEST_BULK:
 894                 trq->tsr_service = SRPC_SERVICE_BRW;
 895                 if (!(feats & LST_FEAT_BULK_LEN)) {
 896                         rc = lstcon_bulkrpc_v0_prep((lst_test_bulk_param_t *)
 897                                                     &test->tes_param[0], trq);
 898                 } else {
 899                         rc = lstcon_bulkrpc_v1_prep((lst_test_bulk_param_t *)
 900                                                     &test->tes_param[0], trq);
 901                 }
 902
 903                 break;
 904         default:
 905                 LBUG();
 906                 break;
 907         }
 908
 909         return rc;
 910 }
 911
 912 static int
 913 lstcon_sesnew_stat_reply(struct lstcon_rpc_trans *trans,
 914                          struct lstcon_node *nd, struct srpc_msg *reply)
 915 {
 916         struct srpc_mksn_reply *mksn_rep = &reply->msg_body.mksn_reply;
 917         int status = mksn_rep->mksn_status;
 918
 919         if (!status &&
 920             (reply->msg_ses_feats & ~LST_FEATS_MASK)) {
 921                 mksn_rep->mksn_status = EPROTO;
 922                 status = EPROTO;
 923         }
 924
 925         if (status == EPROTO) {
 926                 CNETERR("session protocol error from %s: %u\n",
 927                         libcfs_nid2str(nd->nd_id.nid),
 928                         reply->msg_ses_feats);
 929         }
 930
 931         if (status)
 932                 return status;
 933
 934         if (!trans->tas_feats_updated) {
 935                 spin_lock(&console_session.ses_rpc_lock);
 936                 if (!trans->tas_feats_updated) {        /* recheck with lock */
 937                         trans->tas_feats_updated = 1;
 938                         trans->tas_features = reply->msg_ses_feats;
 939                 }
 940                 spin_unlock(&console_session.ses_rpc_lock);
 941         }
 942
 943         if (reply->msg_ses_feats != trans->tas_features) {
 944                 CNETERR("Framework features %x from %s is different with features on this transaction: %x\n",
 945                         reply->msg_ses_feats, libcfs_nid2str(nd->nd_id.nid),
 946                         trans->tas_features);
 947                 mksn_rep->mksn_status = EPROTO;
 948                 status = EPROTO;
 949         }
 950
 951         if (!status) {
 952                 /* session timeout on remote node */
 953                 nd->nd_timeout = mksn_rep->mksn_timeout;
 954         }
 955
 956         return status;
 957 }
 958
 959 void
 960 lstcon_rpc_stat_reply(struct lstcon_rpc_trans *trans, struct srpc_msg *msg,
 961                       struct lstcon_node *nd, lstcon_trans_stat_t *stat)
 962 {
 963         struct srpc_rmsn_reply *rmsn_rep;
 964         struct srpc_debug_reply *dbg_rep;
 965         struct srpc_batch_reply *bat_rep;
 966         struct srpc_test_reply *test_rep;
 967         struct srpc_stat_reply *stat_rep;
 968         int rc = 0;
 969
 970         switch (trans->tas_opc) {
 971         case LST_TRANS_SESNEW:
 972                 rc = lstcon_sesnew_stat_reply(trans, nd, msg);
 973                 if (!rc) {
 974                         lstcon_sesop_stat_success(stat, 1);
 975                         return;
 976                 }
 977
 978                 lstcon_sesop_stat_failure(stat, 1);
 979                 break;
 980
 981         case LST_TRANS_SESEND:
 982                 rmsn_rep = &msg->msg_body.rmsn_reply;
 983                 /* ESRCH is not an error for end session */
 984                 if (!rmsn_rep->rmsn_status ||
 985                     rmsn_rep->rmsn_status == ESRCH) {
 986                         lstcon_sesop_stat_success(stat, 1);
 987                         return;
 988                 }
 989
 990                 lstcon_sesop_stat_failure(stat, 1);
 991                 rc = rmsn_rep->rmsn_status;
 992                 break;
 993
 994         case LST_TRANS_SESQRY:
 995         case LST_TRANS_SESPING:
 996                 dbg_rep = &msg->msg_body.dbg_reply;
 997
 998                 if (dbg_rep->dbg_status == ESRCH) {
 999                         lstcon_sesqry_stat_unknown(stat, 1);
1000                         return;
1001                 }
1002
1003                 if (lstcon_session_match(dbg_rep->dbg_sid))
1004                         lstcon_sesqry_stat_active(stat, 1);
1005                 else
1006                         lstcon_sesqry_stat_busy(stat, 1);
1007                 return;
1008
1009         case LST_TRANS_TSBRUN:
1010         case LST_TRANS_TSBSTOP:
1011                 bat_rep = &msg->msg_body.bat_reply;
1012
1013                 if (!bat_rep->bar_status) {
1014                         lstcon_tsbop_stat_success(stat, 1);
1015                         return;
1016                 }
1017
1018                 if (bat_rep->bar_status == EPERM &&
1019                     trans->tas_opc == LST_TRANS_TSBSTOP) {
1020                         lstcon_tsbop_stat_success(stat, 1);
1021                         return;
1022                 }
1023
1024                 lstcon_tsbop_stat_failure(stat, 1);
1025                 rc = bat_rep->bar_status;
1026                 break;
1027
1028         case LST_TRANS_TSBCLIQRY:
1029         case LST_TRANS_TSBSRVQRY:
1030                 bat_rep = &msg->msg_body.bat_reply;
1031
1032                 if (bat_rep->bar_active)
1033                         lstcon_tsbqry_stat_run(stat, 1);
1034                 else
1035                         lstcon_tsbqry_stat_idle(stat, 1);
1036
1037                 if (!bat_rep->bar_status)
1038                         return;
1039
1040                 lstcon_tsbqry_stat_failure(stat, 1);
1041                 rc = bat_rep->bar_status;
1042                 break;
1043
1044         case LST_TRANS_TSBCLIADD:
1045         case LST_TRANS_TSBSRVADD:
1046                 test_rep = &msg->msg_body.tes_reply;
1047
1048                 if (!test_rep->tsr_status) {
1049                         lstcon_tsbop_stat_success(stat, 1);
1050                         return;
1051                 }
1052
1053                 lstcon_tsbop_stat_failure(stat, 1);
1054                 rc = test_rep->tsr_status;
1055                 break;
1056
1057         case LST_TRANS_STATQRY:
1058                 stat_rep = &msg->msg_body.stat_reply;
1059
1060                 if (!stat_rep->str_status) {
1061                         lstcon_statqry_stat_success(stat, 1);
1062                         return;
1063                 }
1064
1065                 lstcon_statqry_stat_failure(stat, 1);
1066                 rc = stat_rep->str_status;
1067                 break;
1068
1069         default:
1070                 LBUG();
1071         }
1072
1073         if (!stat->trs_fwk_errno)
1074                 stat->trs_fwk_errno = rc;
1075 }
1076
1077 int
1078 lstcon_rpc_trans_ndlist(struct list_head *ndlist,
1079                         struct list_head *translist, int transop,
1080                         void *arg, lstcon_rpc_cond_func_t condition,
1081                         struct lstcon_rpc_trans **transpp)
1082 {
1083         struct lstcon_rpc_trans *trans;
1084         struct lstcon_ndlink *ndl;
1085         struct lstcon_node *nd;
1086         struct lstcon_rpc *rpc;
1087         unsigned feats;
1088         int rc;
1089
1090         /* Creating session RPG for list of nodes */
1091
1092         rc = lstcon_rpc_trans_prep(translist, transop, &trans);
1093         if (rc) {
1094                 CERROR("Can't create transaction %d: %d\n", transop, rc);
1095                 return rc;
1096         }
1097
1098         feats = trans->tas_features;
1099         list_for_each_entry(ndl, ndlist, ndl_link) {
1100                 rc = !condition ? 1 :
1101                      condition(transop, ndl->ndl_node, arg);
1102
1103                 if (!rc)
1104                         continue;
1105
1106                 if (rc < 0) {
1107                         CDEBUG(D_NET, "Condition error while creating RPC for transaction %d: %d\n",
1108                                transop, rc);
1109                         break;
1110                 }
1111
1112                 nd = ndl->ndl_node;
1113
1114                 switch (transop) {
1115                 case LST_TRANS_SESNEW:
1116                 case LST_TRANS_SESEND:
1117                         rc = lstcon_sesrpc_prep(nd, transop, feats, &rpc);
1118                         break;
1119                 case LST_TRANS_SESQRY:
1120                 case LST_TRANS_SESPING:
1121                         rc = lstcon_dbgrpc_prep(nd, feats, &rpc);
1122                         break;
1123                 case LST_TRANS_TSBCLIADD:
1124                 case LST_TRANS_TSBSRVADD:
1125                         rc = lstcon_testrpc_prep(nd, transop, feats,
1126                                                  (struct lstcon_test *)arg,
1127                                                  &rpc);
1128                         break;
1129                 case LST_TRANS_TSBRUN:
1130                 case LST_TRANS_TSBSTOP:
1131                 case LST_TRANS_TSBCLIQRY:
1132                 case LST_TRANS_TSBSRVQRY:
1133                         rc = lstcon_batrpc_prep(nd, transop, feats,
1134                                                 (struct lstcon_tsb_hdr *)arg,
1135                                                 &rpc);
1136                         break;
1137                 case LST_TRANS_STATQRY:
1138                         rc = lstcon_statrpc_prep(nd, feats, &rpc);
1139                         break;
1140                 default:
1141                         rc = -EINVAL;
1142                         break;
1143                 }
1144
1145                 if (rc) {
1146                         CERROR("Failed to create RPC for transaction %s: %d\n",
1147                                lstcon_rpc_trans_name(transop), rc);
1148                         break;
1149                 }
1150
1151                 lstcon_rpc_trans_addreq(trans, rpc);
1152         }
1153
1154         if (!rc) {
1155                 *transpp = trans;
1156                 return 0;
1157         }
1158
1159         lstcon_rpc_trans_destroy(trans);
1160
1161         return rc;
1162 }
1163
1164 static void
1165 lstcon_rpc_pinger(void *arg)
1166 {
1167         struct stt_timer *ptimer = (struct stt_timer *)arg;
1168         struct lstcon_rpc_trans *trans;
1169         struct lstcon_rpc *crpc;
1170         struct srpc_msg *rep;
1171         struct srpc_debug_reqst *drq;
1172         struct lstcon_ndlink *ndl;
1173         struct lstcon_node *nd;
1174         int intv;
1175         int count = 0;
1176         int rc;
1177
1178         /*
1179          * RPC pinger is a special case of transaction,
1180          * it's called by timer at 8 seconds interval.
1181          */
1182         mutex_lock(&console_session.ses_mutex);
1183
1184         if (console_session.ses_shutdown || console_session.ses_expired) {
1185                 mutex_unlock(&console_session.ses_mutex);
1186                 return;
1187         }
1188
1189         if (!console_session.ses_expired &&
1190             ktime_get_real_seconds() - console_session.ses_laststamp >
1191             (time64_t)console_session.ses_timeout)
1192                 console_session.ses_expired = 1;
1193
1194         trans = console_session.ses_ping;
1195
1196         LASSERT(trans);
1197
1198         list_for_each_entry(ndl, &console_session.ses_ndl_list, ndl_link) {
1199                 nd = ndl->ndl_node;
1200
1201                 if (console_session.ses_expired) {
1202                         /* idle console, end session on all nodes */
1203                         if (nd->nd_state != LST_NODE_ACTIVE)
1204                                 continue;
1205
1206                         rc = lstcon_sesrpc_prep(nd, LST_TRANS_SESEND,
1207                                                 trans->tas_features, &crpc);
1208                         if (rc) {
1209                                 CERROR("Out of memory\n");
1210                                 break;
1211                         }
1212
1213                         lstcon_rpc_trans_addreq(trans, crpc);
1214                         lstcon_rpc_post(crpc);
1215
1216                         continue;
1217                 }
1218
1219                 crpc = &nd->nd_ping;
1220
1221                 if (crpc->crp_rpc) {
1222                         LASSERT(crpc->crp_trans == trans);
1223                         LASSERT(!list_empty(&crpc->crp_link));
1224
1225                         spin_lock(&crpc->crp_rpc->crpc_lock);
1226
1227                         LASSERT(crpc->crp_posted);
1228
1229                         if (!crpc->crp_finished) {
1230                                 /* in flight */
1231                                 spin_unlock(&crpc->crp_rpc->crpc_lock);
1232                                 continue;
1233                         }
1234
1235                         spin_unlock(&crpc->crp_rpc->crpc_lock);
1236
1237                         lstcon_rpc_get_reply(crpc, &rep);
1238
1239                         list_del_init(&crpc->crp_link);
1240
1241                         lstcon_rpc_put(crpc);
1242                 }
1243
1244                 if (nd->nd_state != LST_NODE_ACTIVE)
1245                         continue;
1246
1247                 intv = (jiffies - nd->nd_stamp) / msecs_to_jiffies(MSEC_PER_SEC);
1248                 if (intv < nd->nd_timeout / 2)
1249                         continue;
1250
1251                 rc = lstcon_rpc_init(nd, SRPC_SERVICE_DEBUG,
1252                                      trans->tas_features, 0, 0, 1, crpc);
1253                 if (rc) {
1254                         CERROR("Out of memory\n");
1255                         break;
1256                 }
1257
1258                 drq = &crpc->crp_rpc->crpc_reqstmsg.msg_body.dbg_reqst;
1259
1260                 drq->dbg_sid = console_session.ses_id;
1261                 drq->dbg_flags = 0;
1262
1263                 lstcon_rpc_trans_addreq(trans, crpc);
1264                 lstcon_rpc_post(crpc);
1265
1266                 count++;
1267         }
1268
1269         if (console_session.ses_expired) {
1270                 mutex_unlock(&console_session.ses_mutex);
1271                 return;
1272         }
1273
1274         CDEBUG(D_NET, "Ping %d nodes in session\n", count);
1275
1276         ptimer->stt_expires = ktime_get_real_seconds() + LST_PING_INTERVAL;
1277         stt_add_timer(ptimer);
1278
1279         mutex_unlock(&console_session.ses_mutex);
1280 }
1281
1282 int
1283 lstcon_rpc_pinger_start(void)
1284 {
1285         struct stt_timer *ptimer;
1286         int rc;
1287
1288         LASSERT(list_empty(&console_session.ses_rpc_freelist));
1289         LASSERT(!atomic_read(&console_session.ses_rpc_counter));
1290
1291         rc = lstcon_rpc_trans_prep(NULL, LST_TRANS_SESPING,
1292                                    &console_session.ses_ping);
1293         if (rc) {
1294                 CERROR("Failed to create console pinger\n");
1295                 return rc;
1296         }
1297
1298         ptimer = &console_session.ses_ping_timer;
1299         ptimer->stt_expires = ktime_get_real_seconds() + LST_PING_INTERVAL;
1300
1301         stt_add_timer(ptimer);
1302
1303         return 0;
1304 }
1305
1306 void
1307 lstcon_rpc_pinger_stop(void)
1308 {
1309         LASSERT(console_session.ses_shutdown);
1310
1311         stt_del_timer(&console_session.ses_ping_timer);
1312
1313         lstcon_rpc_trans_abort(console_session.ses_ping, -ESHUTDOWN);
1314         lstcon_rpc_trans_stat(console_session.ses_ping, lstcon_trans_stat());
1315         lstcon_rpc_trans_destroy(console_session.ses_ping);
1316
1317         memset(lstcon_trans_stat(), 0, sizeof(lstcon_trans_stat_t));
1318
1319         console_session.ses_ping = NULL;
1320 }
1321
1322 void
1323 lstcon_rpc_cleanup_wait(void)
1324 {
1325         struct lstcon_rpc_trans *trans;
1326         struct lstcon_rpc *crpc;
1327         struct lstcon_rpc *temp;
1328         struct list_head *pacer;
1329         struct list_head zlist;
1330
1331         /* Called with hold of global mutex */
1332
1333         LASSERT(console_session.ses_shutdown);
1334
1335         while (!list_empty(&console_session.ses_trans_list)) {
1336                 list_for_each(pacer, &console_session.ses_trans_list) {
1337                         trans = list_entry(pacer, struct lstcon_rpc_trans,
1338                                            tas_link);
1339
1340                         CDEBUG(D_NET, "Session closed, wakeup transaction %s\n",
1341                                lstcon_rpc_trans_name(trans->tas_opc));
1342
1343                         wake_up(&trans->tas_waitq);
1344                 }
1345
1346                 mutex_unlock(&console_session.ses_mutex);
1347
1348                 CWARN("Session is shutting down, waiting for termination of transactions\n");
1349                 set_current_state(TASK_UNINTERRUPTIBLE);
1350                 schedule_timeout(cfs_time_seconds(1));
1351
1352                 mutex_lock(&console_session.ses_mutex);
1353         }
1354
1355         spin_lock(&console_session.ses_rpc_lock);
1356
1357         lst_wait_until(!atomic_read(&console_session.ses_rpc_counter),
1358                        console_session.ses_rpc_lock,
1359                        "Network is not accessible or target is down, waiting for %d console RPCs to being recycled\n",
1360                        atomic_read(&console_session.ses_rpc_counter));
1361
1362         list_add(&zlist, &console_session.ses_rpc_freelist);
1363         list_del_init(&console_session.ses_rpc_freelist);
1364
1365         spin_unlock(&console_session.ses_rpc_lock);
1366
1367         list_for_each_entry_safe(crpc, temp, &zlist, crp_link) {
1368                 list_del(&crpc->crp_link);
1369                 LIBCFS_FREE(crpc, sizeof(struct lstcon_rpc));
1370         }
1371 }
1372
1373 int
1374 lstcon_rpc_module_init(void)
1375 {
1376         INIT_LIST_HEAD(&console_session.ses_ping_timer.stt_list);
1377         console_session.ses_ping_timer.stt_func = lstcon_rpc_pinger;
1378         console_session.ses_ping_timer.stt_data = &console_session.ses_ping_timer;
1379
1380         console_session.ses_ping = NULL;
1381
1382         spin_lock_init(&console_session.ses_rpc_lock);
1383         atomic_set(&console_session.ses_rpc_counter, 0);
1384         INIT_LIST_HEAD(&console_session.ses_rpc_freelist);
1385
1386         return 0;
1387 }
1388
1389 void
1390 lstcon_rpc_module_fini(void)
1391 {
1392         LASSERT(list_empty(&console_session.ses_rpc_freelist));
1393         LASSERT(!atomic_read(&console_session.ses_rpc_counter));
1394 }