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