2 * -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
3 * vim:expandtab:shiftwidth=8:tabstop=8:
7 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
9 * This program is free software; you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License version 2 only,
11 * as published by the Free Software Foundation.
13 * This program is distributed in the hope that it will be useful, but
14 * WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * General Public License version 2 for more details (a copy is included
17 * in the LICENSE file that accompanied this code).
19 * You should have received a copy of the GNU General Public License
20 * version 2 along with this program; If not, see [sun.com URL with a
23 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
24 * CA 95054 USA or visit www.sun.com if you need additional information or
30 * Copyright 2008 Sun Microsystems, Inc. All rights reserved
31 * Use is subject to license terms.
34 * This file is part of Lustre, http://www.lustre.org/
35 * Lustre is a trademark of Sun Microsystems, Inc.
37 * lnet/selftest/conctl.c
39 * Console framework rpcs
41 * Author: Liang Zhen <liangzhen@clusterfs.com>
46 #include <libcfs/libcfs.h>
47 #include <lnet/lib-lnet.h>
52 void lstcon_rpc_stat_reply(int, srpc_msg_t *,
53 lstcon_node_t *, lstcon_trans_stat_t *);
56 lstcon_rpc_done(srpc_client_rpc_t *rpc)
58 lstcon_rpc_t *crpc = (lstcon_rpc_t *)rpc->crpc_priv;
60 LASSERT (crpc != NULL && rpc == crpc->crp_rpc);
61 LASSERT (crpc->crp_posted && !crpc->crp_finished);
63 spin_lock(&rpc->crpc_lock);
65 if (crpc->crp_trans == NULL) {
66 /* Orphan RPC is not in any transaction,
67 * I'm just a poor body and nobody loves me */
68 spin_unlock(&rpc->crpc_lock);
75 /* not an orphan RPC */
76 crpc->crp_finished = 1;
78 if (crpc->crp_stamp == 0) {
80 LASSERT (crpc->crp_status == 0);
82 crpc->crp_stamp = cfs_time_current();
83 crpc->crp_status = rpc->crpc_status;
86 /* wakeup (transaction)thread if I'm the last RPC in the transaction */
87 if (atomic_dec_and_test(&crpc->crp_trans->tas_remaining))
88 cfs_waitq_signal(&crpc->crp_trans->tas_waitq);
90 spin_unlock(&rpc->crpc_lock);
94 lstcon_rpc_init(lstcon_node_t *nd, int service,
95 int npg, int cached, lstcon_rpc_t *crpc)
98 crpc->crp_rpc = sfw_create_rpc(nd->nd_id, service,
99 npg, npg * CFS_PAGE_SIZE,
100 lstcon_rpc_done, (void *)crpc);
101 if (crpc->crp_rpc == NULL)
104 crpc->crp_trans = NULL;
106 crpc->crp_posted = 0;
107 crpc->crp_finished = 0;
108 crpc->crp_unpacked = 0;
109 crpc->crp_status = 0;
111 crpc->crp_static = !cached;
112 CFS_INIT_LIST_HEAD(&crpc->crp_link);
114 atomic_inc(&console_session.ses_rpc_counter);
120 lstcon_rpc_prep(lstcon_node_t *nd, int service,
121 int npg, lstcon_rpc_t **crpcpp)
123 lstcon_rpc_t *crpc = NULL;
126 spin_lock(&console_session.ses_rpc_lock);
128 if (!list_empty(&console_session.ses_rpc_freelist)) {
129 crpc = list_entry(console_session.ses_rpc_freelist.next,
130 lstcon_rpc_t, crp_link);
131 list_del_init(&crpc->crp_link);
134 spin_unlock(&console_session.ses_rpc_lock);
137 LIBCFS_ALLOC(crpc, sizeof(*crpc));
142 rc = lstcon_rpc_init(nd, service, npg, 1, crpc);
148 LIBCFS_FREE(crpc, sizeof(*crpc));
154 lstcon_rpc_put(lstcon_rpc_t *crpc)
156 srpc_bulk_t *bulk = &crpc->crp_rpc->crpc_bulk;
159 LASSERT (list_empty(&crpc->crp_link));
161 for (i = 0; i < bulk->bk_niov; i++) {
162 if (bulk->bk_iovs[i].kiov_page == NULL)
165 cfs_free_page(bulk->bk_iovs[i].kiov_page);
168 srpc_client_rpc_decref(crpc->crp_rpc);
170 if (crpc->crp_static) {
171 /* Static RPC, not allocated */
172 memset(crpc, 0, sizeof(*crpc));
173 crpc->crp_static = 1;
176 spin_lock(&console_session.ses_rpc_lock);
178 list_add(&crpc->crp_link, &console_session.ses_rpc_freelist);
180 spin_unlock(&console_session.ses_rpc_lock);
183 /* RPC is not alive now */
184 atomic_dec(&console_session.ses_rpc_counter);
188 lstcon_rpc_post(lstcon_rpc_t *crpc)
190 lstcon_rpc_trans_t *trans = crpc->crp_trans;
192 LASSERT (trans != NULL);
194 atomic_inc(&trans->tas_remaining);
195 crpc->crp_posted = 1;
197 sfw_post_rpc(crpc->crp_rpc);
201 lstcon_rpc_trans_name(int transop)
203 if (transop == LST_TRANS_SESNEW)
206 if (transop == LST_TRANS_SESEND)
209 if (transop == LST_TRANS_SESQRY)
212 if (transop == LST_TRANS_SESPING)
215 if (transop == LST_TRANS_TSBCLIADD)
218 if (transop == LST_TRANS_TSBSRVADD)
221 if (transop == LST_TRANS_TSBRUN)
224 if (transop == LST_TRANS_TSBSTOP)
227 if (transop == LST_TRANS_TSBCLIQRY)
230 if (transop == LST_TRANS_TSBSRVQRY)
233 if (transop == LST_TRANS_STATQRY)
240 lstcon_rpc_trans_prep(struct list_head *translist,
241 int transop, lstcon_rpc_trans_t **transpp)
243 lstcon_rpc_trans_t *trans;
245 if (translist != NULL) {
246 list_for_each_entry(trans, translist, tas_link) {
247 /* Can't enqueue two private transaction on
249 if ((trans->tas_opc & transop) == LST_TRANS_PRIVATE)
254 /* create a trans group */
255 LIBCFS_ALLOC(trans, sizeof(*trans));
259 trans->tas_opc = transop;
261 if (translist == NULL)
262 CFS_INIT_LIST_HEAD(&trans->tas_olink);
264 list_add_tail(&trans->tas_olink, translist);
266 list_add_tail(&trans->tas_link, &console_session.ses_trans_list);
268 CFS_INIT_LIST_HEAD(&trans->tas_rpcs_list);
269 atomic_set(&trans->tas_remaining, 0);
270 cfs_waitq_init(&trans->tas_waitq);
278 lstcon_rpc_trans_addreq(lstcon_rpc_trans_t *trans, lstcon_rpc_t *crpc)
280 list_add_tail(&crpc->crp_link, &trans->tas_rpcs_list);
281 crpc->crp_trans = trans;
285 lstcon_rpc_trans_abort(lstcon_rpc_trans_t *trans, int error)
287 srpc_client_rpc_t *rpc;
291 list_for_each_entry (crpc, &trans->tas_rpcs_list, crp_link) {
294 spin_lock(&rpc->crpc_lock);
296 if (!crpc->crp_posted || crpc->crp_stamp != 0) {
297 /* rpc done or aborted already */
298 spin_unlock(&rpc->crpc_lock);
302 crpc->crp_stamp = cfs_time_current();
303 crpc->crp_status = error;
305 spin_unlock(&rpc->crpc_lock);
309 if (error != ETIMEDOUT)
313 if (cfs_time_after(nd->nd_stamp, crpc->crp_stamp))
316 nd->nd_stamp = crpc->crp_stamp;
317 nd->nd_state = LST_NODE_DOWN;
322 lstcon_rpc_trans_check(lstcon_rpc_trans_t *trans)
324 if (console_session.ses_shutdown &&
325 !list_empty(&trans->tas_olink)) /* It's not an end session RPC */
328 return (atomic_read(&trans->tas_remaining) == 0) ? 1: 0;
332 lstcon_rpc_trans_postwait(lstcon_rpc_trans_t *trans, int timeout)
337 if (list_empty(&trans->tas_rpcs_list))
340 if (timeout < LST_TRANS_MIN_TIMEOUT)
341 timeout = LST_TRANS_MIN_TIMEOUT;
343 CDEBUG(D_NET, "Transaction %s started\n",
344 lstcon_rpc_trans_name(trans->tas_opc));
346 /* post all requests */
347 list_for_each_entry (crpc, &trans->tas_rpcs_list, crp_link) {
348 LASSERT (!crpc->crp_posted);
350 lstcon_rpc_post(crpc);
353 mutex_up(&console_session.ses_mutex);
355 rc = cfs_waitq_wait_event_interruptible_timeout(trans->tas_waitq,
356 lstcon_rpc_trans_check(trans),
359 rc = (rc > 0)? 0: ((rc < 0)? -EINTR: -ETIMEDOUT);
361 mutex_down(&console_session.ses_mutex);
363 if (console_session.ses_shutdown)
367 /* treat short timeout as canceled */
368 if (rc == -ETIMEDOUT && timeout < LST_TRANS_MIN_TIMEOUT * 2)
371 lstcon_rpc_trans_abort(trans, rc);
374 CDEBUG(D_NET, "Transaction %s stopped: %d\n",
375 lstcon_rpc_trans_name(trans->tas_opc), rc);
377 lstcon_rpc_trans_stat(trans, lstcon_trans_stat());
383 lstcon_rpc_get_reply(lstcon_rpc_t *crpc, srpc_msg_t **msgpp)
385 lstcon_node_t *nd = crpc->crp_node;
386 srpc_client_rpc_t *rpc = crpc->crp_rpc;
387 srpc_generic_reply_t *rep;
389 LASSERT (nd != NULL && rpc != NULL);
390 LASSERT (crpc->crp_stamp != 0);
392 if (crpc->crp_status != 0) {
394 return crpc->crp_status;
397 *msgpp = &rpc->crpc_replymsg;
398 if (!crpc->crp_unpacked) {
399 sfw_unpack_message(*msgpp);
400 crpc->crp_unpacked = 1;
403 if (cfs_time_after(nd->nd_stamp, crpc->crp_stamp))
406 nd->nd_stamp = crpc->crp_stamp;
407 rep = &(*msgpp)->msg_body.reply;
409 if (rep->sid.ses_nid == LNET_NID_ANY)
410 nd->nd_state = LST_NODE_UNKNOWN;
411 else if (lstcon_session_match(rep->sid))
412 nd->nd_state = LST_NODE_ACTIVE;
414 nd->nd_state = LST_NODE_BUSY;
420 lstcon_rpc_trans_stat(lstcon_rpc_trans_t *trans, lstcon_trans_stat_t *stat)
423 srpc_client_rpc_t *rpc;
427 LASSERT (stat != NULL);
429 memset(stat, 0, sizeof(*stat));
431 list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
432 lstcon_rpc_stat_total(stat, 1);
436 LASSERT (crpc->crp_stamp != 0);
438 error = lstcon_rpc_get_reply(crpc, &rep);
440 lstcon_rpc_stat_failure(stat, 1);
441 if (stat->trs_rpc_errno == 0)
442 stat->trs_rpc_errno = -error;
447 lstcon_rpc_stat_success(stat, 1);
449 lstcon_rpc_stat_reply(trans->tas_opc, rep,
450 crpc->crp_node, stat);
453 CDEBUG(D_NET, "transaction %s : success %d, failure %d, total %d, "
454 "RPC error(%d), Framework error(%d)\n",
455 lstcon_rpc_trans_name(trans->tas_opc),
456 lstcon_rpc_stat_success(stat, 0),
457 lstcon_rpc_stat_failure(stat, 0),
458 lstcon_rpc_stat_total(stat, 0),
459 stat->trs_rpc_errno, stat->trs_fwk_errno);
465 lstcon_rpc_trans_interpreter(lstcon_rpc_trans_t *trans,
466 struct list_head *head_up,
467 lstcon_rpc_readent_func_t readent)
469 struct list_head tmp;
470 struct list_head *next;
471 lstcon_rpc_ent_t *ent;
472 srpc_generic_reply_t *rep;
473 srpc_client_rpc_t *rpc;
481 LASSERT (head_up != NULL);
485 list_for_each_entry(crpc, &trans->tas_rpcs_list, crp_link) {
486 if (copy_from_user(&tmp, next, sizeof(struct list_head)))
489 if (tmp.next == head_up)
494 ent = list_entry(next, lstcon_rpc_ent_t, rpe_link);
498 LASSERT (crpc->crp_stamp != 0);
500 error = lstcon_rpc_get_reply(crpc, &msg);
504 dur = cfs_time_sub(crpc->crp_stamp,
505 console_session.ses_id.ses_stamp);
506 cfs_duration_usec(dur, &tv);
508 if (copy_to_user(&ent->rpe_peer,
509 &nd->nd_id, sizeof(lnet_process_id_t)) ||
510 copy_to_user(&ent->rpe_stamp, &tv, sizeof(tv)) ||
511 copy_to_user(&ent->rpe_state,
512 &nd->nd_state, sizeof(nd->nd_state)) ||
513 copy_to_user(&ent->rpe_rpc_errno, &error, sizeof(error)))
520 rep = (srpc_generic_reply_t *)&msg->msg_body.reply;
522 if (copy_to_user(&ent->rpe_sid,
523 &rep->sid, sizeof(lst_sid_t)) ||
524 copy_to_user(&ent->rpe_fwk_errno,
525 &rep->status, sizeof(rep->status)))
531 if ((error = readent(trans->tas_opc, msg, ent)) != 0)
539 lstcon_rpc_trans_destroy(lstcon_rpc_trans_t *trans)
541 srpc_client_rpc_t *rpc;
546 list_for_each_entry_safe(crpc, tmp,
547 &trans->tas_rpcs_list, crp_link) {
550 spin_lock(&rpc->crpc_lock);
552 /* free it if not posted or finished already */
553 if (!crpc->crp_posted || crpc->crp_finished) {
554 spin_unlock(&rpc->crpc_lock);
556 list_del_init(&crpc->crp_link);
557 lstcon_rpc_put(crpc);
562 /* rpcs can be still not callbacked (even LNetMDUnlink is called)
563 * because huge timeout for inaccessible network, don't make
564 * user wait for them, just abandon them, they will be recycled
567 LASSERT (crpc->crp_status != 0);
569 crpc->crp_node = NULL;
570 crpc->crp_trans = NULL;
571 list_del_init(&crpc->crp_link);
574 spin_unlock(&rpc->crpc_lock);
576 atomic_dec(&trans->tas_remaining);
579 LASSERT (atomic_read(&trans->tas_remaining) == 0);
581 list_del(&trans->tas_link);
582 if (!list_empty(&trans->tas_olink))
583 list_del(&trans->tas_olink);
585 CDEBUG(D_NET, "Transaction %s destroyed with %d pending RPCs\n",
586 lstcon_rpc_trans_name(trans->tas_opc), count);
588 LIBCFS_FREE(trans, sizeof(*trans));
594 lstcon_sesrpc_prep(lstcon_node_t *nd, int transop, lstcon_rpc_t **crpc)
596 srpc_mksn_reqst_t *msrq;
597 srpc_rmsn_reqst_t *rsrq;
601 case LST_TRANS_SESNEW:
602 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_MAKE_SESSION, 0, crpc);
606 msrq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.mksn_reqst;
607 msrq->mksn_sid = console_session.ses_id;
608 msrq->mksn_force = console_session.ses_force;
609 strncpy(msrq->mksn_name, console_session.ses_name,
610 strlen(console_session.ses_name));
613 case LST_TRANS_SESEND:
614 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_REMOVE_SESSION, 0, crpc);
618 rsrq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.rmsn_reqst;
619 rsrq->rmsn_sid = console_session.ses_id;
630 lstcon_dbgrpc_prep(lstcon_node_t *nd, lstcon_rpc_t **crpc)
632 srpc_debug_reqst_t *drq;
635 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_DEBUG, 0, crpc);
639 drq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.dbg_reqst;
641 drq->dbg_sid = console_session.ses_id;
648 lstcon_batrpc_prep(lstcon_node_t *nd, int transop,
649 lstcon_tsb_hdr_t *tsb, lstcon_rpc_t **crpc)
651 lstcon_batch_t *batch;
652 srpc_batch_reqst_t *brq;
655 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_BATCH, 0, crpc);
659 brq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.bat_reqst;
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);
668 if (transop != LST_TRANS_TSBRUN &&
669 transop != LST_TRANS_TSBSTOP)
672 LASSERT (tsb->tsb_index == 0);
674 batch = (lstcon_batch_t *)tsb;
675 brq->bar_arg = batch->bat_arg;
681 lstcon_statrpc_prep(lstcon_node_t *nd, lstcon_rpc_t **crpc)
683 srpc_stat_reqst_t *srq;
686 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_QUERY_STAT, 0, crpc);
690 srq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.stat_reqst;
692 srq->str_sid = console_session.ses_id;
693 srq->str_type = 0; /* XXX remove it */
699 lstcon_next_id(int idx, int nkiov, lnet_kiov_t *kiov)
701 lnet_process_id_t *pid;
704 i = idx / (CFS_PAGE_SIZE / sizeof(lnet_process_id_t));
708 pid = (lnet_process_id_t *)cfs_page_address(kiov[i].kiov_page);
710 return &pid[idx % (CFS_PAGE_SIZE / sizeof(lnet_process_id_t))];
714 lstcon_dstnodes_prep(lstcon_group_t *grp, int idx,
715 int dist, int span, int nkiov, lnet_kiov_t *kiov)
717 lnet_process_id_t *pid;
718 lstcon_ndlink_t *ndl;
726 LASSERT (grp->grp_nnode >= 1);
728 if (span > grp->grp_nnode)
731 start = ((idx / dist) * span) % grp->grp_nnode;
732 end = ((idx / dist) * span + span - 1) % grp->grp_nnode;
734 list_for_each_entry(ndl, &grp->grp_ndl_list, ndl_link) {
741 if (i > (end >= start ? end: grp->grp_nnode))
744 pid = lstcon_next_id((i - start), nkiov, kiov);
749 if (start <= end) /* done */
752 list_for_each_entry(ndl, &grp->grp_ndl_list, ndl_link) {
753 if (i > grp->grp_nnode + end)
757 pid = lstcon_next_id((i - start), nkiov, kiov);
766 lstcon_pingrpc_prep(lst_test_ping_param_t *param, srpc_test_reqst_t *req)
768 test_ping_req_t *prq = &req->tsr_u.ping;
770 prq->png_size = param->png_size;
771 prq->png_flags = param->png_flags;
777 lstcon_bulkrpc_prep(lst_test_bulk_param_t *param, srpc_test_reqst_t *req)
779 test_bulk_req_t *brq = &req->tsr_u.bulk;
781 brq->blk_opc = param->blk_opc;
782 brq->blk_npg = (param->blk_size + CFS_PAGE_SIZE - 1) / CFS_PAGE_SIZE;
783 brq->blk_flags = param->blk_flags;
789 lstcon_testrpc_prep(lstcon_node_t *nd, int transop,
790 lstcon_test_t *test, lstcon_rpc_t **crpc)
792 lstcon_group_t *sgrp = test->tes_src_grp;
793 lstcon_group_t *dgrp = test->tes_dst_grp;
794 srpc_test_reqst_t *trq;
800 if (transop == LST_TRANS_TSBCLIADD)
801 n = sfw_id_pages(test->tes_span);
803 rc = lstcon_rpc_prep(nd, SRPC_SERVICE_TEST, n, crpc);
807 trq = &(*crpc)->crp_rpc->crpc_reqstmsg.msg_body.tes_reqst;
809 if (transop == LST_TRANS_TSBSRVADD) {
810 int ndist = (sgrp->grp_nnode + test->tes_dist - 1) / test->tes_dist;
811 int nspan = (dgrp->grp_nnode + test->tes_span - 1) / test->tes_span;
812 int nmax = (ndist + nspan - 1) / nspan;
815 trq->tsr_loop = nmax * test->tes_dist * test->tes_concur;
818 bulk = &(*crpc)->crp_rpc->crpc_bulk;
820 for (i = 0; i < n; i++) {
821 bulk->bk_iovs[i].kiov_offset = 0;
822 bulk->bk_iovs[i].kiov_len = CFS_PAGE_SIZE;
823 bulk->bk_iovs[i].kiov_page = cfs_alloc_page(CFS_ALLOC_STD);
825 if (bulk->bk_iovs[i].kiov_page != NULL)
828 lstcon_rpc_put(*crpc);
834 LASSERT (transop == LST_TRANS_TSBCLIADD);
836 rc = lstcon_dstnodes_prep(test->tes_dst_grp,
837 test->tes_cliidx++, test->tes_dist,
838 test->tes_span, n, &bulk->bk_iovs[0]);
840 lstcon_rpc_put(*crpc);
844 trq->tsr_ndest = test->tes_span;
845 trq->tsr_loop = test->tes_loop;
848 trq->tsr_sid = console_session.ses_id;
849 trq->tsr_bid = test->tes_hdr.tsb_id;
850 trq->tsr_concur = test->tes_concur;
851 trq->tsr_is_client = (transop == LST_TRANS_TSBCLIADD) ? 1 : 0;
852 trq->tsr_stop_onerr = test->tes_stop_onerr;
854 switch (test->tes_type) {
856 trq->tsr_service = SRPC_SERVICE_PING;
857 rc = lstcon_pingrpc_prep((lst_test_ping_param_t *)&test->tes_param[0], trq);
860 trq->tsr_service = SRPC_SERVICE_BRW;
861 rc = lstcon_bulkrpc_prep((lst_test_bulk_param_t *)&test->tes_param[0], trq);
872 lstcon_rpc_stat_reply(int transop, srpc_msg_t *msg,
873 lstcon_node_t *nd, lstcon_trans_stat_t *stat)
875 srpc_mksn_reply_t *mksn_rep;
876 srpc_rmsn_reply_t *rmsn_rep;
877 srpc_debug_reply_t *dbg_rep;
878 srpc_batch_reply_t *bat_rep;
879 srpc_test_reply_t *test_rep;
880 srpc_stat_reply_t *stat_rep;
884 case LST_TRANS_SESNEW:
885 mksn_rep = &msg->msg_body.mksn_reply;
887 if (mksn_rep->mksn_status == 0) {
888 lstcon_sesop_stat_success(stat, 1);
889 /* session timeout on remote node */
890 nd->nd_timeout = mksn_rep->mksn_timeout;
894 LASSERT (mksn_rep->mksn_status == EBUSY ||
895 mksn_rep->mksn_status == EINVAL);
897 lstcon_sesop_stat_failure(stat, 1);
898 errno = mksn_rep->mksn_status;
901 case LST_TRANS_SESEND:
902 rmsn_rep = &msg->msg_body.rmsn_reply;
903 /* ESRCH is not an error for end session */
904 if (rmsn_rep->rmsn_status == 0 ||
905 rmsn_rep->rmsn_status == ESRCH) {
906 lstcon_sesop_stat_success(stat, 1);
910 LASSERT (rmsn_rep->rmsn_status == EBUSY ||
911 rmsn_rep->rmsn_status == EINVAL);
913 lstcon_sesop_stat_failure(stat, 1);
914 errno = rmsn_rep->rmsn_status;
917 case LST_TRANS_SESQRY:
918 case LST_TRANS_SESPING:
919 dbg_rep = &msg->msg_body.dbg_reply;
921 if (dbg_rep->dbg_status == ESRCH) {
922 lstcon_sesqry_stat_unknown(stat, 1);
926 LASSERT (dbg_rep->dbg_status == 0);
928 if (lstcon_session_match(dbg_rep->dbg_sid))
929 lstcon_sesqry_stat_active(stat, 1);
931 lstcon_sesqry_stat_busy(stat, 1);
934 case LST_TRANS_TSBRUN:
935 case LST_TRANS_TSBSTOP:
936 bat_rep = &msg->msg_body.bat_reply;
938 if (bat_rep->bar_status == 0) {
939 lstcon_tsbop_stat_success(stat, 1);
943 if (bat_rep->bar_status == EPERM &&
944 transop == LST_TRANS_TSBSTOP) {
945 lstcon_tsbop_stat_success(stat, 1);
949 lstcon_tsbop_stat_failure(stat, 1);
950 errno = bat_rep->bar_status;
953 case LST_TRANS_TSBCLIQRY:
954 case LST_TRANS_TSBSRVQRY:
955 bat_rep = &msg->msg_body.bat_reply;
957 if (bat_rep->bar_active != 0)
958 lstcon_tsbqry_stat_run(stat, 1);
960 lstcon_tsbqry_stat_idle(stat, 1);
962 if (bat_rep->bar_status == 0)
965 lstcon_tsbqry_stat_failure(stat, 1);
966 errno = bat_rep->bar_status;
969 case LST_TRANS_TSBCLIADD:
970 case LST_TRANS_TSBSRVADD:
971 test_rep = &msg->msg_body.tes_reply;
973 if (test_rep->tsr_status == 0) {
974 lstcon_tsbop_stat_success(stat, 1);
978 lstcon_tsbop_stat_failure(stat, 1);
979 errno = test_rep->tsr_status;
982 case LST_TRANS_STATQRY:
983 stat_rep = &msg->msg_body.stat_reply;
985 if (stat_rep->str_status == 0) {
986 lstcon_statqry_stat_success(stat, 1);
990 lstcon_statqry_stat_failure(stat, 1);
991 errno = stat_rep->str_status;
998 if (stat->trs_fwk_errno == 0)
999 stat->trs_fwk_errno = errno;
1005 lstcon_rpc_trans_ndlist(struct list_head *ndlist,
1006 struct list_head *translist, int transop,
1007 void *arg, lstcon_rpc_cond_func_t condition,
1008 lstcon_rpc_trans_t **transpp)
1010 lstcon_rpc_trans_t *trans;
1011 lstcon_ndlink_t *ndl;
1016 /* Creating session RPG for list of nodes */
1018 rc = lstcon_rpc_trans_prep(translist, transop, &trans);
1020 CERROR("Can't create transaction %d: %d\n", transop, rc);
1024 list_for_each_entry(ndl, ndlist, ndl_link) {
1025 rc = condition == NULL ? 1 :
1026 condition(transop, ndl->ndl_node, arg);
1032 CDEBUG(D_NET, "Condition error while creating RPC "
1033 " for transaction %d: %d\n", transop, rc);
1040 case LST_TRANS_SESNEW:
1041 case LST_TRANS_SESEND:
1042 rc = lstcon_sesrpc_prep(nd, transop, &rpc);
1044 case LST_TRANS_SESQRY:
1045 case LST_TRANS_SESPING:
1046 rc = lstcon_dbgrpc_prep(nd, &rpc);
1048 case LST_TRANS_TSBCLIADD:
1049 case LST_TRANS_TSBSRVADD:
1050 rc = lstcon_testrpc_prep(nd, transop,
1051 (lstcon_test_t *)arg, &rpc);
1053 case LST_TRANS_TSBRUN:
1054 case LST_TRANS_TSBSTOP:
1055 case LST_TRANS_TSBCLIQRY:
1056 case LST_TRANS_TSBSRVQRY:
1057 rc = lstcon_batrpc_prep(nd, transop,
1058 (lstcon_tsb_hdr_t *)arg, &rpc);
1060 case LST_TRANS_STATQRY:
1061 rc = lstcon_statrpc_prep(nd, &rpc);
1069 CERROR("Failed to create RPC for transaction %s: %d\n",
1070 lstcon_rpc_trans_name(transop), rc);
1074 lstcon_rpc_trans_addreq(trans, rpc);
1082 lstcon_rpc_trans_destroy(trans);
1088 lstcon_rpc_pinger(void *arg)
1090 stt_timer_t *ptimer = (stt_timer_t *)arg;
1091 lstcon_rpc_trans_t *trans;
1094 srpc_debug_reqst_t *drq;
1095 lstcon_ndlink_t *ndl;
1101 /* RPC pinger is a special case of transaction,
1102 * it's called by timer at 8 seconds interval.
1104 mutex_down(&console_session.ses_mutex);
1106 if (console_session.ses_shutdown || console_session.ses_expired) {
1107 mutex_up(&console_session.ses_mutex);
1111 if (!console_session.ses_expired &&
1112 cfs_time_current_sec() - console_session.ses_laststamp >
1113 console_session.ses_timeout)
1114 console_session.ses_expired = 1;
1116 trans = console_session.ses_ping;
1118 LASSERT (trans != NULL);
1120 list_for_each_entry(ndl, &console_session.ses_ndl_list, ndl_link) {
1123 if (console_session.ses_expired) {
1124 /* idle console, end session on all nodes */
1125 if (nd->nd_state != LST_NODE_ACTIVE)
1128 rc = lstcon_sesrpc_prep(nd, LST_TRANS_SESEND, &crpc);
1130 CERROR("Out of memory\n");
1134 lstcon_rpc_trans_addreq(trans, crpc);
1135 lstcon_rpc_post(crpc);
1140 crpc = &nd->nd_ping;
1142 if (crpc->crp_rpc != NULL) {
1143 LASSERT (crpc->crp_trans == trans);
1144 LASSERT (!list_empty(&crpc->crp_link));
1146 spin_lock(&crpc->crp_rpc->crpc_lock);
1148 LASSERT (crpc->crp_posted);
1150 if (!crpc->crp_finished) {
1152 spin_unlock(&crpc->crp_rpc->crpc_lock);
1156 spin_unlock(&crpc->crp_rpc->crpc_lock);
1158 lstcon_rpc_get_reply(crpc, &rep);
1160 list_del_init(&crpc->crp_link);
1162 lstcon_rpc_put(crpc);
1165 if (nd->nd_state != LST_NODE_ACTIVE)
1168 intv = cfs_duration_sec(cfs_time_sub(cfs_time_current(),
1170 if (intv < nd->nd_timeout / 2)
1173 rc = lstcon_rpc_init(nd, SRPC_SERVICE_DEBUG, 0, 0, crpc);
1175 CERROR("Out of memory\n");
1179 drq = &crpc->crp_rpc->crpc_reqstmsg.msg_body.dbg_reqst;
1181 drq->dbg_sid = console_session.ses_id;
1184 lstcon_rpc_trans_addreq(trans, crpc);
1185 lstcon_rpc_post(crpc);
1190 if (console_session.ses_expired) {
1191 mutex_up(&console_session.ses_mutex);
1195 CDEBUG(D_NET, "Ping %d nodes in session\n", count);
1197 ptimer->stt_expires = cfs_time_current_sec() + LST_PING_INTERVAL;
1198 stt_add_timer(ptimer);
1200 mutex_up(&console_session.ses_mutex);
1204 lstcon_rpc_pinger_start(void)
1206 stt_timer_t *ptimer;
1209 LASSERT (list_empty(&console_session.ses_rpc_freelist));
1210 LASSERT (atomic_read(&console_session.ses_rpc_counter) == 0);
1212 rc = lstcon_rpc_trans_prep(NULL, LST_TRANS_SESPING,
1213 &console_session.ses_ping);
1215 CERROR("Failed to create console pinger\n");
1219 ptimer = &console_session.ses_ping_timer;
1220 ptimer->stt_expires = cfs_time_current_sec() + LST_PING_INTERVAL;
1222 stt_add_timer(ptimer);
1228 lstcon_rpc_pinger_stop(void)
1230 LASSERT (console_session.ses_shutdown);
1232 stt_del_timer(&console_session.ses_ping_timer);
1234 lstcon_rpc_trans_abort(console_session.ses_ping, -ESHUTDOWN);
1235 lstcon_rpc_trans_stat(console_session.ses_ping, lstcon_trans_stat());
1236 lstcon_rpc_trans_destroy(console_session.ses_ping);
1238 memset(lstcon_trans_stat(), 0, sizeof(lstcon_trans_stat_t));
1240 console_session.ses_ping = NULL;
1244 lstcon_rpc_cleanup_wait(void)
1246 lstcon_rpc_trans_t *trans;
1248 struct list_head *pacer;
1249 struct list_head zlist;
1251 /* Called with hold of global mutex */
1253 LASSERT (console_session.ses_shutdown);
1255 while (!list_empty(&console_session.ses_trans_list)) {
1256 list_for_each(pacer, &console_session.ses_trans_list) {
1257 trans = list_entry(pacer, lstcon_rpc_trans_t, tas_link);
1259 CDEBUG(D_NET, "Session closed, wakeup transaction %s\n",
1260 lstcon_rpc_trans_name(trans->tas_opc));
1262 cfs_waitq_signal(&trans->tas_waitq);
1265 mutex_up(&console_session.ses_mutex);
1267 CWARN("Session is shutting down, "
1268 "waiting for termination of transactions\n");
1269 cfs_pause(cfs_time_seconds(1));
1271 mutex_down(&console_session.ses_mutex);
1274 spin_lock(&console_session.ses_rpc_lock);
1276 lst_wait_until((atomic_read(&console_session.ses_rpc_counter) == 0),
1277 console_session.ses_rpc_lock,
1278 "Network is not accessable or target is down, "
1279 "waiting for %d console RPCs to being recycled\n",
1280 atomic_read(&console_session.ses_rpc_counter));
1282 list_add(&zlist, &console_session.ses_rpc_freelist);
1283 list_del_init(&console_session.ses_rpc_freelist);
1285 spin_unlock(&console_session.ses_rpc_lock);
1287 while (!list_empty(&zlist)) {
1288 crpc = list_entry(zlist.next, lstcon_rpc_t, crp_link);
1290 list_del(&crpc->crp_link);
1291 LIBCFS_FREE(crpc, sizeof(lstcon_rpc_t));
1296 lstcon_rpc_module_init(void)
1298 CFS_INIT_LIST_HEAD(&console_session.ses_ping_timer.stt_list);
1299 console_session.ses_ping_timer.stt_func = lstcon_rpc_pinger;
1300 console_session.ses_ping_timer.stt_data = &console_session.ses_ping_timer;
1302 console_session.ses_ping = NULL;
1304 spin_lock_init(&console_session.ses_rpc_lock);
1305 atomic_set(&console_session.ses_rpc_counter, 0);
1306 CFS_INIT_LIST_HEAD(&console_session.ses_rpc_freelist);
1312 lstcon_rpc_module_fini(void)
1314 LASSERT (list_empty(&console_session.ses_rpc_freelist));
1315 LASSERT (atomic_read(&console_session.ses_rpc_counter) == 0);