text: /*
   * GPL HEADER START
   *
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   *
   * This program is free software; you can redistribute it and/or modify
   * it under the terms of the GNU General Public License version 2 only,
   * as published by the Free Software Foundation.
   *
   * This program is distributed in the hope that it will be useful, but
   * WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   * General Public License version 2 for more details (a copy is included
   * in the LICENSE file that accompanied this code).
   *
   * You should have received a copy of the GNU General Public License
   * version 2 along with this program; If not, see
   * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
   *
   * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
   * CA 95054 USA or visit www.sun.com if you need additional information or
   * have any questions.
   *
   * GPL HEADER END
   */
  /*
   * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
   * Use is subject to license terms.
   *
   * Copyright (c) 2012, Intel Corporation.
   */
  /*
   * This file is part of Lustre, http://www.lustre.org/
   * Lustre is a trademark of Sun Microsystems, Inc.
   *
   * lnet/selftest/conctl.c
   *
   * Infrastructure of LST console
   *
   * Author: Liang Zhen <liangzhen@clusterfs.com>
   */
  
  
  #include <linux/libcfs/libcfs.h>
  #include <linux/lnet/lib-lnet.h>
  #include "console.h"
  #include "conrpc.h"
  
  #define LST_NODE_STATE_COUNTER(nd, p)		   \
  do {						    \
  	if ((nd)->nd_state == LST_NODE_ACTIVE)	  \
  		(p)->nle_nactive ++;		    \
  	else if ((nd)->nd_state == LST_NODE_BUSY)       \
  		(p)->nle_nbusy ++;		      \
  	else if ((nd)->nd_state == LST_NODE_DOWN)       \
  		(p)->nle_ndown ++;		      \
  	else					    \
  		(p)->nle_nunknown ++;		   \
  	(p)->nle_nnode ++;			      \
  } while (0)
  
  lstcon_session_t	console_session;
  
  static void
  lstcon_node_get(lstcon_node_t *nd)
  {
  	LASSERT (nd->nd_ref >= 1);
  
  	nd->nd_ref++;
  }
  
  static int
  lstcon_node_find(lnet_process_id_t id, lstcon_node_t **ndpp, int create)
  {
  	lstcon_ndlink_t *ndl;
  	unsigned int     idx = LNET_NIDADDR(id.nid) % LST_GLOBAL_HASHSIZE;
  
  	LASSERT (id.nid != LNET_NID_ANY);
  
  	list_for_each_entry(ndl, &console_session.ses_ndl_hash[idx], ndl_hlink) {
  		if (ndl->ndl_node->nd_id.nid != id.nid ||
  		    ndl->ndl_node->nd_id.pid != id.pid)
  			continue;
  
  		lstcon_node_get(ndl->ndl_node);
  		*ndpp = ndl->ndl_node;
  		return 0;
  	}
  
  	if (!create)
  		return -ENOENT;
  
  	LIBCFS_ALLOC(*ndpp, sizeof(lstcon_node_t) + sizeof(lstcon_ndlink_t));
  	if (*ndpp == NULL)
  		return -ENOMEM;
  
  	ndl = (lstcon_ndlink_t *)(*ndpp + 1);
  
  	ndl->ndl_node = *ndpp;
  
  	ndl->ndl_node->nd_ref   = 1;
  	ndl->ndl_node->nd_id    = id;
  	ndl->ndl_node->nd_stamp = cfs_time_current();
  	ndl->ndl_node->nd_state = LST_NODE_UNKNOWN;
  	ndl->ndl_node->nd_timeout = 0;
  	memset(&ndl->ndl_node->nd_ping, 0, sizeof(lstcon_rpc_t));
  
  	/* queued in global hash & list, no refcount is taken by
  	 * global hash & list, if caller release his refcount,
  	 * node will be released */
  	list_add_tail(&ndl->ndl_hlink, &console_session.ses_ndl_hash[idx]);
  	list_add_tail(&ndl->ndl_link, &console_session.ses_ndl_list);
  
  	return 0;
  }
  
  static void
  lstcon_node_put(lstcon_node_t *nd)
  {
  	lstcon_ndlink_t  *ndl;
  
  	LASSERT (nd->nd_ref > 0);
  
  	if (--nd->nd_ref > 0)
  		return;
  
  	ndl = (lstcon_ndlink_t *)(nd + 1);
  
  	LASSERT (!list_empty(&ndl->ndl_link));
  	LASSERT (!list_empty(&ndl->ndl_hlink));
  
  	/* remove from session */
  	list_del(&ndl->ndl_link);
  	list_del(&ndl->ndl_hlink);
  
  	LIBCFS_FREE(nd, sizeof(lstcon_node_t) + sizeof(lstcon_ndlink_t));
  }
  
  static int
  lstcon_ndlink_find(struct list_head *hash,
  		   lnet_process_id_t id, lstcon_ndlink_t **ndlpp, int create)
  {
  	unsigned int     idx = LNET_NIDADDR(id.nid) % LST_NODE_HASHSIZE;
  	lstcon_ndlink_t *ndl;
  	lstcon_node_t   *nd;
  	int	      rc;
  
  	if (id.nid == LNET_NID_ANY)
  		return -EINVAL;
  
  	/* search in hash */
  	list_for_each_entry(ndl, &hash[idx], ndl_hlink) {
  		if (ndl->ndl_node->nd_id.nid != id.nid ||
  		    ndl->ndl_node->nd_id.pid != id.pid)
  			continue;
  
  		*ndlpp = ndl;
  		return 0;
  	}
  
  	if (create == 0)
  		return -ENOENT;
  
  	/* find or create in session hash */
  	rc = lstcon_node_find(id, &nd, (create == 1) ? 1 : 0);
  	if (rc != 0)
  		return rc;
  
  	LIBCFS_ALLOC(ndl, sizeof(lstcon_ndlink_t));
  	if (ndl == NULL) {
  		lstcon_node_put(nd);
  		return -ENOMEM;
  	}
  
  	*ndlpp = ndl;
  
  	ndl->ndl_node = nd;
  	INIT_LIST_HEAD(&ndl->ndl_link);
  	list_add_tail(&ndl->ndl_hlink, &hash[idx]);
  
  	return  0;
  }
  
  static void
  lstcon_ndlink_release(lstcon_ndlink_t *ndl)
  {
  	LASSERT (list_empty(&ndl->ndl_link));
  	LASSERT (!list_empty(&ndl->ndl_hlink));
  
  	list_del(&ndl->ndl_hlink); /* delete from hash */
  	lstcon_node_put(ndl->ndl_node);
  
  	LIBCFS_FREE(ndl, sizeof(*ndl));
  }
  
  static int
  lstcon_group_alloc(char *name, lstcon_group_t **grpp)
  {
  	lstcon_group_t *grp;
  	int	     i;
  
  	LIBCFS_ALLOC(grp, offsetof(lstcon_group_t,
  				   grp_ndl_hash[LST_NODE_HASHSIZE]));
  	if (grp == NULL)
  		return -ENOMEM;
  
  	memset(grp, 0, offsetof(lstcon_group_t,
  				grp_ndl_hash[LST_NODE_HASHSIZE]));
  
  	grp->grp_ref = 1;
  	if (name != NULL)
  		strcpy(grp->grp_name, name);
  
  	INIT_LIST_HEAD(&grp->grp_link);
  	INIT_LIST_HEAD(&grp->grp_ndl_list);
  	INIT_LIST_HEAD(&grp->grp_trans_list);
  
  	for (i = 0; i < LST_NODE_HASHSIZE; i++)
  		INIT_LIST_HEAD(&grp->grp_ndl_hash[i]);
  
  	*grpp = grp;
  
  	return 0;
  }
  
  static void
  lstcon_group_addref(lstcon_group_t *grp)
  {
  	grp->grp_ref ++;
  }
  
  static void lstcon_group_ndlink_release(lstcon_group_t *, lstcon_ndlink_t *);
  
  static void
  lstcon_group_drain(lstcon_group_t *grp, int keep)
  {
  	lstcon_ndlink_t *ndl;
  	lstcon_ndlink_t *tmp;
  
  	list_for_each_entry_safe(ndl, tmp, &grp->grp_ndl_list, ndl_link) {
  		if ((ndl->ndl_node->nd_state & keep) == 0)
  			lstcon_group_ndlink_release(grp, ndl);
  	}
  }
  
  static void
  lstcon_group_decref(lstcon_group_t *grp)
  {
  	int     i;
  
  	if (--grp->grp_ref > 0)
  		return;
  
  	if (!list_empty(&grp->grp_link))
  		list_del(&grp->grp_link);
  
  	lstcon_group_drain(grp, 0);
  
  	for (i = 0; i < LST_NODE_HASHSIZE; i++) {
  		LASSERT (list_empty(&grp->grp_ndl_hash[i]));
  	}
  
  	LIBCFS_FREE(grp, offsetof(lstcon_group_t,
  				  grp_ndl_hash[LST_NODE_HASHSIZE]));
  }
  
  static int
  lstcon_group_find(const char *name, lstcon_group_t **grpp)
  {
  	lstcon_group_t   *grp;
  
  	list_for_each_entry(grp, &console_session.ses_grp_list, grp_link) {
  		if (strncmp(grp->grp_name, name, LST_NAME_SIZE) != 0)
  			continue;
  
  		lstcon_group_addref(grp);  /* +1 ref for caller */
  		*grpp = grp;
  		return 0;
  	}
  
  	return -ENOENT;
  }
  
  static void
  lstcon_group_put(lstcon_group_t *grp)
  {
  	lstcon_group_decref(grp);
  }
  
  static int
  lstcon_group_ndlink_find(lstcon_group_t *grp, lnet_process_id_t id,
  			 lstcon_ndlink_t **ndlpp, int create)
  {
  	int     rc;
  
  	rc = lstcon_ndlink_find(&grp->grp_ndl_hash[0], id, ndlpp, create);
  	if (rc != 0)
  		return rc;
  
  	if (!list_empty(&(*ndlpp)->ndl_link))
  		return 0;
  
  	list_add_tail(&(*ndlpp)->ndl_link, &grp->grp_ndl_list);
  	grp->grp_nnode ++;
  
  	return 0;
  }
  
  static void
  lstcon_group_ndlink_release(lstcon_group_t *grp, lstcon_ndlink_t *ndl)
  {
  	list_del_init(&ndl->ndl_link);
  	lstcon_ndlink_release(ndl);
  	grp->grp_nnode --;
  }
  
  static void
  lstcon_group_ndlink_move(lstcon_group_t *old,
  			 lstcon_group_t *new, lstcon_ndlink_t *ndl)
  {
  	unsigned int idx = LNET_NIDADDR(ndl->ndl_node->nd_id.nid) %
  			   LST_NODE_HASHSIZE;
  
  	list_del(&ndl->ndl_hlink);
  	list_del(&ndl->ndl_link);
  	old->grp_nnode --;
  
  	list_add_tail(&ndl->ndl_hlink, &new->grp_ndl_hash[idx]);
  	list_add_tail(&ndl->ndl_link, &new->grp_ndl_list);
  	new->grp_nnode ++;
  
  	return;
  }
  
  static void
  lstcon_group_move(lstcon_group_t *old, lstcon_group_t *new)
  {
  	lstcon_ndlink_t *ndl;
  
  	while (!list_empty(&old->grp_ndl_list)) {
  		ndl = list_entry(old->grp_ndl_list.next,
  				     lstcon_ndlink_t, ndl_link);
  		lstcon_group_ndlink_move(old, new, ndl);
  	}
  }
  
  static int
  lstcon_sesrpc_condition(int transop, lstcon_node_t *nd, void *arg)
  {
  	lstcon_group_t *grp = (lstcon_group_t *)arg;
  
  	switch (transop) {
  	case LST_TRANS_SESNEW:
  		if (nd->nd_state == LST_NODE_ACTIVE)
  			return 0;
  		break;
  
  	case LST_TRANS_SESEND:
  		if (nd->nd_state != LST_NODE_ACTIVE)
  			return 0;
  
  		if (grp != NULL && nd->nd_ref > 1)
  			return 0;
  		break;
  
  	case LST_TRANS_SESQRY:
  		break;
  
  	default:
  		LBUG();
  	}
  
  	return 1;
  }
  
  static int
  lstcon_sesrpc_readent(int transop, srpc_msg_t *msg,
  		      lstcon_rpc_ent_t *ent_up)
  {
  	srpc_debug_reply_t *rep;
  
  	switch (transop) {
  	case LST_TRANS_SESNEW:
  	case LST_TRANS_SESEND:
  		return 0;
  
  	case LST_TRANS_SESQRY:
  		rep = &msg->msg_body.dbg_reply;
  
  		if (copy_to_user(&ent_up->rpe_priv[0],
  				     &rep->dbg_timeout, sizeof(int)) ||
  		    copy_to_user(&ent_up->rpe_payload[0],
  				     &rep->dbg_name, LST_NAME_SIZE))
  			return -EFAULT;
  
  		return 0;
  
  	default:
  		LBUG();
  	}
  
  	return 0;
  }
  
  static int
  lstcon_group_nodes_add(lstcon_group_t *grp,
  		       int count, lnet_process_id_t *ids_up,
  		       unsigned *featp, struct list_head *result_up)
  {
  	lstcon_rpc_trans_t      *trans;
  	lstcon_ndlink_t	 *ndl;
  	lstcon_group_t	  *tmp;
  	lnet_process_id_t	id;
  	int		      i;
  	int		      rc;
  
  	rc = lstcon_group_alloc(NULL, &tmp);
  	if (rc != 0) {
  		CERROR("Out of memory\n");
  		return -ENOMEM;
  	}
  
  	for (i = 0 ; i < count; i++) {
  		if (copy_from_user(&id, &ids_up[i], sizeof(id))) {
  			rc = -EFAULT;
  			break;
  		}
  
  		/* skip if it's in this group already */
  		rc = lstcon_group_ndlink_find(grp, id, &ndl, 0);
  		if (rc == 0)
  			continue;
  
  		/* add to tmp group */
  		rc = lstcon_group_ndlink_find(tmp, id, &ndl, 1);
  		if (rc != 0) {
  			CERROR("Can't create ndlink, out of memory\n");
  			break;
  		}
  	}
  
  	if (rc != 0) {
  		lstcon_group_put(tmp);
  		return rc;
  	}
  
  	rc = lstcon_rpc_trans_ndlist(&tmp->grp_ndl_list,
  				     &tmp->grp_trans_list, LST_TRANS_SESNEW,
  				     tmp, lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		lstcon_group_put(tmp);
  		return rc;
  	}
  
  	/* post all RPCs */
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up,
  					  lstcon_sesrpc_readent);
  	*featp = trans->tas_features;
  
  	/* destroy all RPGs */
  	lstcon_rpc_trans_destroy(trans);
  
  	lstcon_group_move(tmp, grp);
  	lstcon_group_put(tmp);
  
  	return rc;
  }
  
  static int
  lstcon_group_nodes_remove(lstcon_group_t *grp,
  			  int count, lnet_process_id_t *ids_up,
  			  struct list_head *result_up)
  {
  	lstcon_rpc_trans_t     *trans;
  	lstcon_ndlink_t	*ndl;
  	lstcon_group_t	 *tmp;
  	lnet_process_id_t       id;
  	int		     rc;
  	int		     i;
  
  	/* End session and remove node from the group */
  
  	rc = lstcon_group_alloc(NULL, &tmp);
  	if (rc != 0) {
  		CERROR("Out of memory\n");
  		return -ENOMEM;
  	}
  
  	for (i = 0; i < count; i++) {
  		if (copy_from_user(&id, &ids_up[i], sizeof(id))) {
  			rc = -EFAULT;
  			goto error;
  		}
  
  		/* move node to tmp group */
  		if (lstcon_group_ndlink_find(grp, id, &ndl, 0) == 0)
  			lstcon_group_ndlink_move(grp, tmp, ndl);
  	}
  
  	rc = lstcon_rpc_trans_ndlist(&tmp->grp_ndl_list,
  				     &tmp->grp_trans_list, LST_TRANS_SESEND,
  				     tmp, lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		goto error;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up, NULL);
  
  	lstcon_rpc_trans_destroy(trans);
  	/* release nodes anyway, because we can't rollback status */
  	lstcon_group_put(tmp);
  
  	return rc;
  error:
  	lstcon_group_move(tmp, grp);
  	lstcon_group_put(tmp);
  
  	return rc;
  }
  
  int
  lstcon_group_add(char *name)
  {
  	lstcon_group_t *grp;
  	int	     rc;
  
  	rc = (lstcon_group_find(name, &grp) == 0)? -EEXIST: 0;
  	if (rc != 0) {
  		/* find a group with same name */
  		lstcon_group_put(grp);
  		return rc;
  	}
  
  	rc = lstcon_group_alloc(name, &grp);
  	if (rc != 0) {
  		CERROR("Can't allocate descriptor for group %s\n", name);
  		return -ENOMEM;
  	}
  
  	list_add_tail(&grp->grp_link, &console_session.ses_grp_list);
  
  	return rc;
  }
  
  int
  lstcon_nodes_add(char *name, int count, lnet_process_id_t *ids_up,
  		 unsigned *featp, struct list_head *result_up)
  {
  	lstcon_group_t	 *grp;
  	int		     rc;
  
  	LASSERT (count > 0);
  	LASSERT (ids_up != NULL);
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group %s\n", name);
  		return rc;
  	}
  
  	if (grp->grp_ref > 2) {
  		/* referred by other threads or test */
  		CDEBUG(D_NET, "Group %s is busy\n", name);
  		lstcon_group_put(grp);
  
  		return -EBUSY;
  	}
  
  	rc = lstcon_group_nodes_add(grp, count, ids_up, featp, result_up);
  
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_group_del(char *name)
  {
  	lstcon_rpc_trans_t *trans;
  	lstcon_group_t     *grp;
  	int		 rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group: %s\n", name);
  		return rc;
  	}
  
  	if (grp->grp_ref > 2) {
  		/* referred by others threads or test */
  		CDEBUG(D_NET, "Group %s is busy\n", name);
  		lstcon_group_put(grp);
  		return -EBUSY;
  	}
  
  	rc = lstcon_rpc_trans_ndlist(&grp->grp_ndl_list,
  				     &grp->grp_trans_list, LST_TRANS_SESEND,
  				     grp, lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		lstcon_group_put(grp);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	lstcon_rpc_trans_destroy(trans);
  
  	lstcon_group_put(grp);
  	/* -ref for session, it's destroyed,
  	 * status can't be rolled back, destroy group anyway */
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_group_clean(char *name, int args)
  {
  	lstcon_group_t *grp = NULL;
  	int	     rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group %s\n", name);
  		return rc;
  	}
  
  	if (grp->grp_ref > 2) {
  		/* referred by test */
  		CDEBUG(D_NET, "Group %s is busy\n", name);
  		lstcon_group_put(grp);
  		return -EBUSY;
  	}
  
  	args = (LST_NODE_ACTIVE | LST_NODE_BUSY |
  		LST_NODE_DOWN | LST_NODE_UNKNOWN) & ~args;
  
  	lstcon_group_drain(grp, args);
  
  	lstcon_group_put(grp);
  	/* release empty group */
  	if (list_empty(&grp->grp_ndl_list))
  		lstcon_group_put(grp);
  
  	return 0;
  }
  
  int
  lstcon_nodes_remove(char *name, int count,
  		    lnet_process_id_t *ids_up, struct list_head *result_up)
  {
  	lstcon_group_t *grp = NULL;
  	int	     rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group: %s\n", name);
  		return rc;
  	}
  
  	if (grp->grp_ref > 2) {
  		/* referred by test */
  		CDEBUG(D_NET, "Group %s is busy\n", name);
  		lstcon_group_put(grp);
  		return -EBUSY;
  	}
  
  	rc = lstcon_group_nodes_remove(grp, count, ids_up, result_up);
  
  	lstcon_group_put(grp);
  	/* release empty group */
  	if (list_empty(&grp->grp_ndl_list))
  		lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_group_refresh(char *name, struct list_head *result_up)
  {
  	lstcon_rpc_trans_t      *trans;
  	lstcon_group_t	  *grp;
  	int		      rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group: %s\n", name);
  		return rc;
  	}
  
  	if (grp->grp_ref > 2) {
  		/* referred by test */
  		CDEBUG(D_NET, "Group %s is busy\n", name);
  		lstcon_group_put(grp);
  		return -EBUSY;
  	}
  
  	/* re-invite all inactive nodes int the group */
  	rc = lstcon_rpc_trans_ndlist(&grp->grp_ndl_list,
  				     &grp->grp_trans_list, LST_TRANS_SESNEW,
  				     grp, lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		/* local error, return */
  		CDEBUG(D_NET, "Can't create transaction: %d\n", rc);
  		lstcon_group_put(grp);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up, NULL);
  
  	lstcon_rpc_trans_destroy(trans);
  	/* -ref for me */
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_group_list(int index, int len, char *name_up)
  {
  	lstcon_group_t *grp;
  
  	LASSERT (index >= 0);
  	LASSERT (name_up != NULL);
  
  	list_for_each_entry(grp, &console_session.ses_grp_list, grp_link) {
  		if (index-- == 0) {
  			return copy_to_user(name_up, grp->grp_name, len) ?
  			       -EFAULT : 0;
  		}
  	}
  
  	return -ENOENT;
  }
  
  static int
  lstcon_nodes_getent(struct list_head *head, int *index_p,
  		    int *count_p, lstcon_node_ent_t *dents_up)
  {
  	lstcon_ndlink_t  *ndl;
  	lstcon_node_t    *nd;
  	int	       count = 0;
  	int	       index = 0;
  
  	LASSERT (index_p != NULL && count_p != NULL);
  	LASSERT (dents_up != NULL);
  	LASSERT (*index_p >= 0);
  	LASSERT (*count_p > 0);
  
  	list_for_each_entry(ndl, head, ndl_link) {
  		if (index++ < *index_p)
  			continue;
  
  		if (count >= *count_p)
  			break;
  
  		nd = ndl->ndl_node;
  		if (copy_to_user(&dents_up[count].nde_id,
  				     &nd->nd_id, sizeof(nd->nd_id)) ||
  		    copy_to_user(&dents_up[count].nde_state,
  				     &nd->nd_state, sizeof(nd->nd_state)))
  			return -EFAULT;
  
  		count ++;
  	}
  
  	if (index <= *index_p)
  		return -ENOENT;
  
  	*count_p = count;
  	*index_p = index;
  
  	return 0;
  }
  
  int
  lstcon_group_info(char *name, lstcon_ndlist_ent_t *gents_p,
  		  int *index_p, int *count_p, lstcon_node_ent_t *dents_up)
  {
  	lstcon_ndlist_ent_t *gentp;
  	lstcon_group_t      *grp;
  	lstcon_ndlink_t     *ndl;
  	int		  rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group %s\n", name);
  		return rc;
  	}
  
  	if (dents_up) {
  		/* verbose query */
  		rc = lstcon_nodes_getent(&grp->grp_ndl_list,
  					 index_p, count_p, dents_up);
  		lstcon_group_put(grp);
  
  		return rc;
  	}
  
  	/* non-verbose query */
  	LIBCFS_ALLOC(gentp, sizeof(lstcon_ndlist_ent_t));
  	if (gentp == NULL) {
  		CERROR("Can't allocate ndlist_ent\n");
  		lstcon_group_put(grp);
  
  		return -ENOMEM;
  	}
  
  	memset(gentp, 0, sizeof(lstcon_ndlist_ent_t));
  
  	list_for_each_entry(ndl, &grp->grp_ndl_list, ndl_link)
  		LST_NODE_STATE_COUNTER(ndl->ndl_node, gentp);
  
  	rc = copy_to_user(gents_p, gentp,
  			      sizeof(lstcon_ndlist_ent_t)) ? -EFAULT: 0;
  
  	LIBCFS_FREE(gentp, sizeof(lstcon_ndlist_ent_t));
  
  	lstcon_group_put(grp);
  
  	return 0;
  }
  
  static int
  lstcon_batch_find(const char *name, lstcon_batch_t **batpp)
  {
  	lstcon_batch_t   *bat;
  
  	list_for_each_entry(bat, &console_session.ses_bat_list, bat_link) {
  		if (strncmp(bat->bat_name, name, LST_NAME_SIZE) == 0) {
  			*batpp = bat;
  			return 0;
  		}
  	}
  
  	return -ENOENT;
  }
  
  int
  lstcon_batch_add(char *name)
  {
  	lstcon_batch_t   *bat;
  	int	       i;
  	int	       rc;
  
  	rc = (lstcon_batch_find(name, &bat) == 0)? -EEXIST: 0;
  	if (rc != 0) {
  		CDEBUG(D_NET, "Batch %s already exists\n", name);
  		return rc;
  	}
  
  	LIBCFS_ALLOC(bat, sizeof(lstcon_batch_t));
  	if (bat == NULL) {
  		CERROR("Can't allocate descriptor for batch %s\n", name);
  		return -ENOMEM;
  	}
  
  	LIBCFS_ALLOC(bat->bat_cli_hash,
  		     sizeof(struct list_head) * LST_NODE_HASHSIZE);
  	if (bat->bat_cli_hash == NULL) {
  		CERROR("Can't allocate hash for batch %s\n", name);
  		LIBCFS_FREE(bat, sizeof(lstcon_batch_t));
  
  		return -ENOMEM;
  	}
  
  	LIBCFS_ALLOC(bat->bat_srv_hash,
  		     sizeof(struct list_head) * LST_NODE_HASHSIZE);
  	if (bat->bat_srv_hash == NULL) {
  		CERROR("Can't allocate hash for batch %s\n", name);
  		LIBCFS_FREE(bat->bat_cli_hash, LST_NODE_HASHSIZE);
  		LIBCFS_FREE(bat, sizeof(lstcon_batch_t));
  
  		return -ENOMEM;
  	}
  
  	strcpy(bat->bat_name, name);
  	bat->bat_hdr.tsb_index = 0;
  	bat->bat_hdr.tsb_id.bat_id = ++console_session.ses_id_cookie;
  
  	bat->bat_ntest = 0;
  	bat->bat_state = LST_BATCH_IDLE;
  
  	INIT_LIST_HEAD(&bat->bat_cli_list);
  	INIT_LIST_HEAD(&bat->bat_srv_list);
  	INIT_LIST_HEAD(&bat->bat_test_list);
  	INIT_LIST_HEAD(&bat->bat_trans_list);
  
  	for (i = 0; i < LST_NODE_HASHSIZE; i++) {
  		INIT_LIST_HEAD(&bat->bat_cli_hash[i]);
  		INIT_LIST_HEAD(&bat->bat_srv_hash[i]);
  	}
  
  	list_add_tail(&bat->bat_link, &console_session.ses_bat_list);
  
  	return rc;
  }
  
  int
  lstcon_batch_list(int index, int len, char *name_up)
  {
  	lstcon_batch_t    *bat;
  
  	LASSERT (name_up != NULL);
  	LASSERT (index >= 0);
  
  	list_for_each_entry(bat, &console_session.ses_bat_list, bat_link) {
  		if (index-- == 0) {
  			return copy_to_user(name_up,bat->bat_name, len) ?
  			       -EFAULT: 0;
  		}
  	}
  
  	return -ENOENT;
  }
  
  int
  lstcon_batch_info(char *name, lstcon_test_batch_ent_t *ent_up, int server,
  		  int testidx, int *index_p, int *ndent_p,
  		  lstcon_node_ent_t *dents_up)
  {
  	lstcon_test_batch_ent_t *entp;
  	struct list_head	      *clilst;
  	struct list_head	      *srvlst;
  	lstcon_test_t	   *test = NULL;
  	lstcon_batch_t	  *bat;
  	lstcon_ndlink_t	 *ndl;
  	int		      rc;
  
  	rc = lstcon_batch_find(name, &bat);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find batch %s\n", name);
  		return -ENOENT;
  	}
  
  	if (testidx > 0) {
  		/* query test, test index start from 1 */
  		list_for_each_entry(test, &bat->bat_test_list, tes_link) {
  			if (testidx-- == 1)
  				break;
  		}
  
  		if (testidx > 0) {
  			CDEBUG(D_NET, "Can't find specified test in batch\n");
  			return -ENOENT;
  		}
  	}
  
  	clilst = (test == NULL) ? &bat->bat_cli_list :
  				  &test->tes_src_grp->grp_ndl_list;
  	srvlst = (test == NULL) ? &bat->bat_srv_list :
  				  &test->tes_dst_grp->grp_ndl_list;
  
  	if (dents_up != NULL) {
  		rc = lstcon_nodes_getent((server ? srvlst: clilst),
  					 index_p, ndent_p, dents_up);
  		return rc;
  	}
  
  	/* non-verbose query */
  	LIBCFS_ALLOC(entp, sizeof(lstcon_test_batch_ent_t));
  	if (entp == NULL)
  		return -ENOMEM;
  
  	memset(entp, 0, sizeof(lstcon_test_batch_ent_t));
  
  	if (test == NULL) {
  		entp->u.tbe_batch.bae_ntest = bat->bat_ntest;
  		entp->u.tbe_batch.bae_state = bat->bat_state;
  
  	} else {
  
  		entp->u.tbe_test.tse_type   = test->tes_type;
  		entp->u.tbe_test.tse_loop   = test->tes_loop;
  		entp->u.tbe_test.tse_concur = test->tes_concur;
  	}
  
  	list_for_each_entry(ndl, clilst, ndl_link)
  		LST_NODE_STATE_COUNTER(ndl->ndl_node, &entp->tbe_cli_nle);
  
  	list_for_each_entry(ndl, srvlst, ndl_link)
  		LST_NODE_STATE_COUNTER(ndl->ndl_node, &entp->tbe_srv_nle);
  
  	rc = copy_to_user(ent_up, entp,
  			      sizeof(lstcon_test_batch_ent_t)) ? -EFAULT : 0;
  
  	LIBCFS_FREE(entp, sizeof(lstcon_test_batch_ent_t));
  
  	return rc;
  }
  
  static int
  lstcon_batrpc_condition(int transop, lstcon_node_t *nd, void *arg)
  {
  	switch (transop) {
  	case LST_TRANS_TSBRUN:
  		if (nd->nd_state != LST_NODE_ACTIVE)
  			return -ENETDOWN;
  		break;
  
  	case LST_TRANS_TSBSTOP:
  		if (nd->nd_state != LST_NODE_ACTIVE)
  			return 0;
  		break;
  
  	case LST_TRANS_TSBCLIQRY:
  	case LST_TRANS_TSBSRVQRY:
  		break;
  	}
  
  	return 1;
  }
  
  static int
  lstcon_batch_op(lstcon_batch_t *bat, int transop,
  		struct list_head *result_up)
  {
  	lstcon_rpc_trans_t *trans;
  	int		 rc;
  
  	rc = lstcon_rpc_trans_ndlist(&bat->bat_cli_list,
  				     &bat->bat_trans_list, transop,
  				     bat, lstcon_batrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up, NULL);
  
  	lstcon_rpc_trans_destroy(trans);
  
  	return rc;
  }
  
  int
  lstcon_batch_run(char *name, int timeout, struct list_head *result_up)
  {
  	lstcon_batch_t *bat;
  	int	     rc;
  
  	if (lstcon_batch_find(name, &bat) != 0) {
  		CDEBUG(D_NET, "Can't find batch %s\n", name);
  		return -ENOENT;
  	}
  
  	bat->bat_arg = timeout;
  
  	rc = lstcon_batch_op(bat, LST_TRANS_TSBRUN, result_up);
  
  	/* mark batch as running if it's started in any node */
  	if (lstcon_tsbop_stat_success(lstcon_trans_stat(), 0) != 0)
  		bat->bat_state = LST_BATCH_RUNNING;
  
  	return rc;
  }
  
  int
  lstcon_batch_stop(char *name, int force, struct list_head *result_up)
  {
  	lstcon_batch_t *bat;
  	int	     rc;
  
  	if (lstcon_batch_find(name, &bat) != 0) {
  		CDEBUG(D_NET, "Can't find batch %s\n", name);
  		return -ENOENT;
  	}
  
  	bat->bat_arg = force;
  
  	rc = lstcon_batch_op(bat, LST_TRANS_TSBSTOP, result_up);
  
  	/* mark batch as stopped if all RPCs finished */
  	if (lstcon_tsbop_stat_failure(lstcon_trans_stat(), 0) == 0)
  		bat->bat_state = LST_BATCH_IDLE;
  
  	return rc;
  }
  
  static void
  lstcon_batch_destroy(lstcon_batch_t *bat)
  {
  	lstcon_ndlink_t    *ndl;
  	lstcon_test_t      *test;
  	int		 i;
  
  	list_del(&bat->bat_link);
  
  	while (!list_empty(&bat->bat_test_list)) {
  		test = list_entry(bat->bat_test_list.next,
  				      lstcon_test_t, tes_link);
  		LASSERT (list_empty(&test->tes_trans_list));
  
  		list_del(&test->tes_link);
  
  		lstcon_group_put(test->tes_src_grp);
  		lstcon_group_put(test->tes_dst_grp);
  
  		LIBCFS_FREE(test, offsetof(lstcon_test_t,
  					   tes_param[test->tes_paramlen]));
  	}
  
  	LASSERT (list_empty(&bat->bat_trans_list));
  
  	while (!list_empty(&bat->bat_cli_list)) {
  		ndl = list_entry(bat->bat_cli_list.next,
  				     lstcon_ndlink_t, ndl_link);
  		list_del_init(&ndl->ndl_link);
  
  		lstcon_ndlink_release(ndl);
  	}
  
  	while (!list_empty(&bat->bat_srv_list)) {
  		ndl = list_entry(bat->bat_srv_list.next,
  				     lstcon_ndlink_t, ndl_link);
  		list_del_init(&ndl->ndl_link);
  
  		lstcon_ndlink_release(ndl);
  	}
  
  	for (i = 0; i < LST_NODE_HASHSIZE; i++) {
  		LASSERT (list_empty(&bat->bat_cli_hash[i]));
  		LASSERT (list_empty(&bat->bat_srv_hash[i]));
  	}
  
  	LIBCFS_FREE(bat->bat_cli_hash,
  		    sizeof(struct list_head) * LST_NODE_HASHSIZE);
  	LIBCFS_FREE(bat->bat_srv_hash,
  		    sizeof(struct list_head) * LST_NODE_HASHSIZE);
  	LIBCFS_FREE(bat, sizeof(lstcon_batch_t));
  }
  
  static int
  lstcon_testrpc_condition(int transop, lstcon_node_t *nd, void *arg)
  {
  	lstcon_test_t    *test;
  	lstcon_batch_t   *batch;
  	lstcon_ndlink_t  *ndl;
  	struct list_head       *hash;
  	struct list_head       *head;
  
  	test = (lstcon_test_t *)arg;
  	LASSERT (test != NULL);
  
  	batch = test->tes_batch;
  	LASSERT (batch != NULL);
  
  	if (test->tes_oneside &&
  	    transop == LST_TRANS_TSBSRVADD)
  		return 0;
  
  	if (nd->nd_state != LST_NODE_ACTIVE)
  		return -ENETDOWN;
  
  	if (transop == LST_TRANS_TSBCLIADD) {
  		hash = batch->bat_cli_hash;
  		head = &batch->bat_cli_list;
  
  	} else {
  		LASSERT (transop == LST_TRANS_TSBSRVADD);
  
  		hash = batch->bat_srv_hash;
  		head = &batch->bat_srv_list;
  	}
  
  	LASSERT (nd->nd_id.nid != LNET_NID_ANY);
  
  	if (lstcon_ndlink_find(hash, nd->nd_id, &ndl, 1) != 0)
  		return -ENOMEM;
  
  	if (list_empty(&ndl->ndl_link))
  		list_add_tail(&ndl->ndl_link, head);
  
  	return 1;
  }
  
  static int
  lstcon_test_nodes_add(lstcon_test_t *test, struct list_head *result_up)
  {
  	lstcon_rpc_trans_t     *trans;
  	lstcon_group_t	 *grp;
  	int		     transop;
  	int		     rc;
  
  	LASSERT (test->tes_src_grp != NULL);
  	LASSERT (test->tes_dst_grp != NULL);
  
  	transop = LST_TRANS_TSBSRVADD;
  	grp  = test->tes_dst_grp;
  again:
  	rc = lstcon_rpc_trans_ndlist(&grp->grp_ndl_list,
  				     &test->tes_trans_list, transop,
  				     test, lstcon_testrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	if (lstcon_trans_stat()->trs_rpc_errno != 0 ||
  	    lstcon_trans_stat()->trs_fwk_errno != 0) {
  		lstcon_rpc_trans_interpreter(trans, result_up, NULL);
  
  		lstcon_rpc_trans_destroy(trans);
  		/* return if any error */
  		CDEBUG(D_NET, "Failed to add test %s, "
  			      "RPC error %d, framework error %d\n",
  		       transop == LST_TRANS_TSBCLIADD ? "client" : "server",
  		       lstcon_trans_stat()->trs_rpc_errno,
  		       lstcon_trans_stat()->trs_fwk_errno);
  
  		return rc;
  	}
  
  	lstcon_rpc_trans_destroy(trans);
  
  	if (transop == LST_TRANS_TSBCLIADD)
  		return rc;
  
  	transop = LST_TRANS_TSBCLIADD;
  	grp = test->tes_src_grp;
  	test->tes_cliidx = 0;
  
  	/* requests to test clients */
  	goto again;
  }
  
  static int
  lstcon_verify_batch(const char *name, lstcon_batch_t **batch)
  {
  	int rc;
  
  	rc = lstcon_batch_find(name, batch);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find batch %s\n", name);
  		return rc;
  	}
  
  	if ((*batch)->bat_state != LST_BATCH_IDLE) {
  		CDEBUG(D_NET, "Can't change running batch %s\n", name);
  		return -EINVAL;
  	}
  
  	return 0;
  }
  
  static int
  lstcon_verify_group(const char *name, lstcon_group_t **grp)
  {
  	int			rc;
  	lstcon_ndlink_t		*ndl;
  
  	rc = lstcon_group_find(name, grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "can't find group %s\n", name);
  		return rc;
  	}
  
  	list_for_each_entry(ndl, &(*grp)->grp_ndl_list, ndl_link) {
  		if (ndl->ndl_node->nd_state == LST_NODE_ACTIVE)
  			return 0;
  	}
  
  	CDEBUG(D_NET, "Group %s has no ACTIVE nodes\n", name);
  
  	return -EINVAL;
  }
  
  int
  lstcon_test_add(char *batch_name, int type, int loop,
  		int concur, int dist, int span,
  		char *src_name, char *dst_name,
  		void *param, int paramlen, int *retp,
  		struct list_head *result_up)
  {
  	lstcon_test_t	 *test	 = NULL;
  	int		 rc;
  	lstcon_group_t	 *src_grp = NULL;
  	lstcon_group_t	 *dst_grp = NULL;
  	lstcon_batch_t	 *batch = NULL;
  
  	/*
  	 * verify that a batch of the given name exists, and the groups
  	 * that will be part of the batch exist and have at least one
  	 * active node
  	 */
  	rc = lstcon_verify_batch(batch_name, &batch);
  	if (rc != 0)
  		goto out;
  
  	rc = lstcon_verify_group(src_name, &src_grp);
  	if (rc != 0)
  		goto out;
  
  	rc = lstcon_verify_group(dst_name, &dst_grp);
  	if (rc != 0)
  		goto out;
  
  	if (dst_grp->grp_userland)
  		*retp = 1;
  
  	LIBCFS_ALLOC(test, offsetof(lstcon_test_t, tes_param[paramlen]));
  	if (!test) {
  		CERROR("Can't allocate test descriptor\n");
  		rc = -ENOMEM;
  
  		goto out;
  	}
  
  	memset(test, 0, offsetof(lstcon_test_t, tes_param[paramlen]));
  	test->tes_hdr.tsb_id	= batch->bat_hdr.tsb_id;
  	test->tes_batch		= batch;
  	test->tes_type		= type;
  	test->tes_oneside	= 0; /* TODO */
  	test->tes_loop		= loop;
  	test->tes_concur	= concur;
  	test->tes_stop_onerr	= 1; /* TODO */
  	test->tes_span		= span;
  	test->tes_dist		= dist;
  	test->tes_cliidx	= 0; /* just used for creating RPC */
  	test->tes_src_grp	= src_grp;
  	test->tes_dst_grp	= dst_grp;
  	INIT_LIST_HEAD(&test->tes_trans_list);
  
  	if (param != NULL) {
  		test->tes_paramlen = paramlen;
  		memcpy(&test->tes_param[0], param, paramlen);
  	}
  
  	rc = lstcon_test_nodes_add(test, result_up);
  
  	if (rc != 0)
  		goto out;
  
  	if (lstcon_trans_stat()->trs_rpc_errno != 0 ||
  	    lstcon_trans_stat()->trs_fwk_errno != 0)
  		CDEBUG(D_NET, "Failed to add test %d to batch %s\n", type,
  		       batch_name);
  
  	/* add to test list anyway, so user can check what's going on */
  	list_add_tail(&test->tes_link, &batch->bat_test_list);
  
  	batch->bat_ntest ++;
  	test->tes_hdr.tsb_index = batch->bat_ntest;
  
  	/*  hold groups so nobody can change them */
  	return rc;
  out:
  	if (test != NULL)
  		LIBCFS_FREE(test, offsetof(lstcon_test_t, tes_param[paramlen]));
  
  	if (dst_grp != NULL)
  		lstcon_group_put(dst_grp);
  
  	if (src_grp != NULL)
  		lstcon_group_put(src_grp);
  
  	return rc;
  }
  
  static int
  lstcon_test_find(lstcon_batch_t *batch, int idx, lstcon_test_t **testpp)
  {
  	lstcon_test_t *test;
  
  	list_for_each_entry(test, &batch->bat_test_list, tes_link) {
  		if (idx == test->tes_hdr.tsb_index) {
  			*testpp = test;
  			return 0;
  		}
  	}
  
  	return -ENOENT;
  }
  
  static int
  lstcon_tsbrpc_readent(int transop, srpc_msg_t *msg,
  		      lstcon_rpc_ent_t *ent_up)
  {
  	srpc_batch_reply_t *rep = &msg->msg_body.bat_reply;
  
  	LASSERT (transop == LST_TRANS_TSBCLIQRY ||
  		 transop == LST_TRANS_TSBSRVQRY);
  
  	/* positive errno, framework error code */
  	if (copy_to_user(&ent_up->rpe_priv[0],
  			     &rep->bar_active, sizeof(rep->bar_active)))
  		return -EFAULT;
  
  	return 0;
  }
  
  int
  lstcon_test_batch_query(char *name, int testidx, int client,
  			int timeout, struct list_head *result_up)
  {
  	lstcon_rpc_trans_t *trans;
  	struct list_head	 *translist;
  	struct list_head	 *ndlist;
  	lstcon_tsb_hdr_t   *hdr;
  	lstcon_batch_t     *batch;
  	lstcon_test_t      *test = NULL;
  	int		 transop;
  	int		 rc;
  
  	rc = lstcon_batch_find(name, &batch);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find batch: %s\n", name);
  		return rc;
  	}
  
  	if (testidx == 0) {
  		translist = &batch->bat_trans_list;
  		ndlist    = &batch->bat_cli_list;
  		hdr       = &batch->bat_hdr;
  
  	} else {
  		/* query specified test only */
  		rc = lstcon_test_find(batch, testidx, &test);
  		if (rc != 0) {
  			CDEBUG(D_NET, "Can't find test: %d\n", testidx);
  			return rc;
  		}
  
  		translist = &test->tes_trans_list;
  		ndlist    = &test->tes_src_grp->grp_ndl_list;
  		hdr       = &test->tes_hdr;
  	}
  
  	transop = client ? LST_TRANS_TSBCLIQRY : LST_TRANS_TSBSRVQRY;
  
  	rc = lstcon_rpc_trans_ndlist(ndlist, translist, transop, hdr,
  				     lstcon_batrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, timeout);
  
  	if (testidx == 0 && /* query a batch, not a test */
  	    lstcon_rpc_stat_failure(lstcon_trans_stat(), 0) == 0 &&
  	    lstcon_tsbqry_stat_run(lstcon_trans_stat(), 0) == 0) {
  		/* all RPCs finished, and no active test */
  		batch->bat_state = LST_BATCH_IDLE;
  	}
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up,
  					  lstcon_tsbrpc_readent);
  	lstcon_rpc_trans_destroy(trans);
  
  	return rc;
  }
  
  static int
  lstcon_statrpc_readent(int transop, srpc_msg_t *msg,
  		       lstcon_rpc_ent_t *ent_up)
  {
  	srpc_stat_reply_t *rep = &msg->msg_body.stat_reply;
  	sfw_counters_t    *sfwk_stat;
  	srpc_counters_t   *srpc_stat;
  	lnet_counters_t   *lnet_stat;
  
  	if (rep->str_status != 0)
  		return 0;
  
  	sfwk_stat = (sfw_counters_t *)&ent_up->rpe_payload[0];
  	srpc_stat = (srpc_counters_t *)((char *)sfwk_stat + sizeof(*sfwk_stat));
  	lnet_stat = (lnet_counters_t *)((char *)srpc_stat + sizeof(*srpc_stat));
  
  	if (copy_to_user(sfwk_stat, &rep->str_fw, sizeof(*sfwk_stat)) ||
  	    copy_to_user(srpc_stat, &rep->str_rpc, sizeof(*srpc_stat)) ||
  	    copy_to_user(lnet_stat, &rep->str_lnet, sizeof(*lnet_stat)))
  		return -EFAULT;
  
  	return 0;
  }
  
  static int
  lstcon_ndlist_stat(struct list_head *ndlist,
  		   int timeout, struct list_head *result_up)
  {
  	struct list_head	  head;
  	lstcon_rpc_trans_t *trans;
  	int		 rc;
  
  	INIT_LIST_HEAD(&head);
  
  	rc = lstcon_rpc_trans_ndlist(ndlist, &head,
  				     LST_TRANS_STATQRY, NULL, NULL, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_VALIDATE_TIMEOUT(timeout));
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up,
  					  lstcon_statrpc_readent);
  	lstcon_rpc_trans_destroy(trans);
  
  	return rc;
  }
  
  int
  lstcon_group_stat(char *grp_name, int timeout, struct list_head *result_up)
  {
  	lstcon_group_t     *grp;
  	int		 rc;
  
  	rc = lstcon_group_find(grp_name, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Can't find group %s\n", grp_name);
  		return rc;
  	}
  
  	rc = lstcon_ndlist_stat(&grp->grp_ndl_list, timeout, result_up);
  
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_nodes_stat(int count, lnet_process_id_t *ids_up,
  		  int timeout, struct list_head *result_up)
  {
  	lstcon_ndlink_t	 *ndl;
  	lstcon_group_t	  *tmp;
  	lnet_process_id_t	id;
  	int		      i;
  	int		      rc;
  
  	rc = lstcon_group_alloc(NULL, &tmp);
  	if (rc != 0) {
  		CERROR("Out of memory\n");
  		return -ENOMEM;
  	}
  
  	for (i = 0 ; i < count; i++) {
  		if (copy_from_user(&id, &ids_up[i], sizeof(id))) {
  			rc = -EFAULT;
  			break;
  		}
  
  		/* add to tmp group */
  		rc = lstcon_group_ndlink_find(tmp, id, &ndl, 2);
  		if (rc != 0) {
  			CDEBUG((rc == -ENOMEM) ? D_ERROR : D_NET,
  			       "Failed to find or create %s: %d\n",
  			       libcfs_id2str(id), rc);
  			break;
  		}
  	}
  
  	if (rc != 0) {
  		lstcon_group_put(tmp);
  		return rc;
  	}
  
  	rc = lstcon_ndlist_stat(&tmp->grp_ndl_list, timeout, result_up);
  
  	lstcon_group_put(tmp);
  
  	return rc;
  }
  
  static int
  lstcon_debug_ndlist(struct list_head *ndlist,
  		    struct list_head *translist,
  		    int timeout, struct list_head *result_up)
  {
  	lstcon_rpc_trans_t *trans;
  	int		 rc;
  
  	rc = lstcon_rpc_trans_ndlist(ndlist, translist, LST_TRANS_SESQRY,
  				     NULL, lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	lstcon_rpc_trans_postwait(trans, LST_VALIDATE_TIMEOUT(timeout));
  
  	rc = lstcon_rpc_trans_interpreter(trans, result_up,
  					  lstcon_sesrpc_readent);
  	lstcon_rpc_trans_destroy(trans);
  
  	return rc;
  }
  
  int
  lstcon_session_debug(int timeout, struct list_head *result_up)
  {
  	return lstcon_debug_ndlist(&console_session.ses_ndl_list,
  				   NULL, timeout, result_up);
  }
  
  int
  lstcon_batch_debug(int timeout, char *name,
  		   int client, struct list_head *result_up)
  {
  	lstcon_batch_t *bat;
  	int	     rc;
  
  	rc = lstcon_batch_find(name, &bat);
  	if (rc != 0)
  		return -ENOENT;
  
  	rc = lstcon_debug_ndlist(client ? &bat->bat_cli_list :
  					  &bat->bat_srv_list,
  				 NULL, timeout, result_up);
  
  	return rc;
  }
  
  int
  lstcon_group_debug(int timeout, char *name,
  		   struct list_head *result_up)
  {
  	lstcon_group_t *grp;
  	int	     rc;
  
  	rc = lstcon_group_find(name, &grp);
  	if (rc != 0)
  		return -ENOENT;
  
  	rc = lstcon_debug_ndlist(&grp->grp_ndl_list, NULL,
  				 timeout, result_up);
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_nodes_debug(int timeout,
  		   int count, lnet_process_id_t *ids_up,
  		   struct list_head *result_up)
  {
  	lnet_process_id_t  id;
  	lstcon_ndlink_t   *ndl;
  	lstcon_group_t    *grp;
  	int		i;
  	int		rc;
  
  	rc = lstcon_group_alloc(NULL, &grp);
  	if (rc != 0) {
  		CDEBUG(D_NET, "Out of memory\n");
  		return rc;
  	}
  
  	for (i = 0; i < count; i++) {
  		if (copy_from_user(&id, &ids_up[i], sizeof(id))) {
  			rc = -EFAULT;
  			break;
  		}
  
  		/* node is added to tmp group */
  		rc = lstcon_group_ndlink_find(grp, id, &ndl, 1);
  		if (rc != 0) {
  			CERROR("Can't create node link\n");
  			break;
  		}
  	}
  
  	if (rc != 0) {
  		lstcon_group_put(grp);
  		return rc;
  	}
  
  	rc = lstcon_debug_ndlist(&grp->grp_ndl_list, NULL,
  				 timeout, result_up);
  
  	lstcon_group_put(grp);
  
  	return rc;
  }
  
  int
  lstcon_session_match(lst_sid_t sid)
  {
  	return (console_session.ses_id.ses_nid   == sid.ses_nid &&
  		console_session.ses_id.ses_stamp == sid.ses_stamp) ?  1: 0;
  }
  
  static void
  lstcon_new_session_id(lst_sid_t *sid)
  {
  	lnet_process_id_t      id;
  
  	LASSERT (console_session.ses_state == LST_SESSION_NONE);
  
  	LNetGetId(1, &id);
  	sid->ses_nid   = id.nid;
  	sid->ses_stamp = cfs_time_current();
  }
  
  extern srpc_service_t lstcon_acceptor_service;
  
  int
  lstcon_session_new(char *name, int key, unsigned feats,
  		   int timeout, int force, lst_sid_t *sid_up)
  {
  	int     rc = 0;
  	int     i;
  
  	if (console_session.ses_state != LST_SESSION_NONE) {
  		/* session exists */
  		if (!force) {
  			CNETERR("Session %s already exists\n",
  				console_session.ses_name);
  			return -EEXIST;
  		}
  
  		rc = lstcon_session_end();
  
  		/* lstcon_session_end() only return local error */
  		if  (rc != 0)
  			return rc;
  	}
  
  	if ((feats & ~LST_FEATS_MASK) != 0) {
  		CNETERR("Unknown session features %x\n",
  			(feats & ~LST_FEATS_MASK));
  		return -EINVAL;
  	}
  
  	for (i = 0; i < LST_GLOBAL_HASHSIZE; i++)
  		LASSERT(list_empty(&console_session.ses_ndl_hash[i]));
  
  	lstcon_new_session_id(&console_session.ses_id);
  
  	console_session.ses_key	    = key;
  	console_session.ses_state   = LST_SESSION_ACTIVE;
  	console_session.ses_force   = !!force;
  	console_session.ses_features = feats;
  	console_session.ses_feats_updated = 0;
  	console_session.ses_timeout = (timeout <= 0) ?
  				      LST_CONSOLE_TIMEOUT : timeout;
  	strcpy(console_session.ses_name, name);
  
  	rc = lstcon_batch_add(LST_DEFAULT_BATCH);
  	if (rc != 0)
  		return rc;
  
  	rc = lstcon_rpc_pinger_start();
  	if (rc != 0) {
  		lstcon_batch_t *bat = NULL;
  
  		lstcon_batch_find(LST_DEFAULT_BATCH, &bat);
  		lstcon_batch_destroy(bat);
  
  		return rc;
  	}
  
  	if (copy_to_user(sid_up, &console_session.ses_id,
  			     sizeof(lst_sid_t)) == 0)
  		return rc;
  
  	lstcon_session_end();
  
  	return -EFAULT;
  }
  
  int
  lstcon_session_info(lst_sid_t *sid_up, int *key_up, unsigned *featp,
  		    lstcon_ndlist_ent_t *ndinfo_up, char *name_up, int len)
  {
  	lstcon_ndlist_ent_t *entp;
  	lstcon_ndlink_t     *ndl;
  	int		  rc = 0;
  
  	if (console_session.ses_state != LST_SESSION_ACTIVE)
  		return -ESRCH;
  
  	LIBCFS_ALLOC(entp, sizeof(*entp));
  	if (entp == NULL)
  		return -ENOMEM;
  
  	memset(entp, 0, sizeof(*entp));
  
  	list_for_each_entry(ndl, &console_session.ses_ndl_list, ndl_link)
  		LST_NODE_STATE_COUNTER(ndl->ndl_node, entp);
  
  	if (copy_to_user(sid_up, &console_session.ses_id,
  			     sizeof(lst_sid_t)) ||
  	    copy_to_user(key_up, &console_session.ses_key,
  			     sizeof(*key_up)) ||
  	    copy_to_user(featp, &console_session.ses_features,
  			     sizeof(*featp)) ||
  	    copy_to_user(ndinfo_up, entp, sizeof(*entp)) ||
  	    copy_to_user(name_up, console_session.ses_name, len))
  		rc = -EFAULT;
  
  	LIBCFS_FREE(entp, sizeof(*entp));
  
  	return rc;
  }
  
  int
  lstcon_session_end(void)
  {
  	lstcon_rpc_trans_t *trans;
  	lstcon_group_t     *grp;
  	lstcon_batch_t     *bat;
  	int		 rc = 0;
  
  	LASSERT (console_session.ses_state == LST_SESSION_ACTIVE);
  
  	rc = lstcon_rpc_trans_ndlist(&console_session.ses_ndl_list,
  				     NULL, LST_TRANS_SESEND, NULL,
  				     lstcon_sesrpc_condition, &trans);
  	if (rc != 0) {
  		CERROR("Can't create transaction: %d\n", rc);
  		return rc;
  	}
  
  	console_session.ses_shutdown = 1;
  
  	lstcon_rpc_pinger_stop();
  
  	lstcon_rpc_trans_postwait(trans, LST_TRANS_TIMEOUT);
  
  	lstcon_rpc_trans_destroy(trans);
  	/* User can do nothing even rpc failed, so go on */
  
  	/* waiting for orphan rpcs to die */
  	lstcon_rpc_cleanup_wait();
  
  	console_session.ses_id    = LST_INVALID_SID;
  	console_session.ses_state = LST_SESSION_NONE;
  	console_session.ses_key   = 0;
  	console_session.ses_force = 0;
  	console_session.ses_feats_updated = 0;
  
  	/* destroy all batches */
  	while (!list_empty(&console_session.ses_bat_list)) {
  		bat = list_entry(console_session.ses_bat_list.next,
  				     lstcon_batch_t, bat_link);
  
  		lstcon_batch_destroy(bat);
  	}
  
  	/* destroy all groups */
  	while (!list_empty(&console_session.ses_grp_list)) {
  		grp = list_entry(console_session.ses_grp_list.next,
  				     lstcon_group_t, grp_link);
  		LASSERT (grp->grp_ref == 1);
  
  		lstcon_group_put(grp);
  	}
  
  	/* all nodes should be released */
  	LASSERT (list_empty(&console_session.ses_ndl_list));
  
  	console_session.ses_shutdown = 0;
  	console_session.ses_expired  = 0;
  
  	return rc;
  }
  
  int
  lstcon_session_feats_check(unsigned feats)
  {
  	int rc = 0;
  
  	if ((feats & ~LST_FEATS_MASK) != 0) {
  		CERROR("Can't support these features: %x\n",
  		       (feats & ~LST_FEATS_MASK));
  		return -EPROTO;
  	}
  
  	spin_lock(&console_session.ses_rpc_lock);
  
  	if (!console_session.ses_feats_updated) {
  		console_session.ses_feats_updated = 1;
  		console_session.ses_features = feats;
  	}
  
  	if (console_session.ses_features != feats)
  		rc = -EPROTO;
  
  	spin_unlock(&console_session.ses_rpc_lock);
  
  	if (rc != 0) {
  		CERROR("remote features %x do not match with "
  		       "session features %x of console\n",
  		       feats, console_session.ses_features);
  	}
  
  	return rc;
  }
  
  static int
  lstcon_acceptor_handle (srpc_server_rpc_t *rpc)
  {
  	srpc_msg_t	*rep  = &rpc->srpc_replymsg;
  	srpc_msg_t	*req  = &rpc->srpc_reqstbuf->buf_msg;
  	srpc_join_reqst_t *jreq = &req->msg_body.join_reqst;
  	srpc_join_reply_t *jrep = &rep->msg_body.join_reply;
  	lstcon_group_t    *grp  = NULL;
  	lstcon_ndlink_t   *ndl;
  	int		rc   = 0;
  
  	sfw_unpack_message(req);
  
  	mutex_lock(&console_session.ses_mutex);
  
  	jrep->join_sid = console_session.ses_id;
  
  	if (console_session.ses_id.ses_nid == LNET_NID_ANY) {
  		jrep->join_status = ESRCH;
  		goto out;
  	}
  
  	if (lstcon_session_feats_check(req->msg_ses_feats) != 0) {
  		jrep->join_status = EPROTO;
  		goto out;
  	}
  
  	if (jreq->join_sid.ses_nid != LNET_NID_ANY &&
  	     !lstcon_session_match(jreq->join_sid)) {
  		jrep->join_status = EBUSY;
  		goto out;
  	}
  
  	if (lstcon_group_find(jreq->join_group, &grp) != 0) {
  		rc = lstcon_group_alloc(jreq->join_group, &grp);
  		if (rc != 0) {
  			CERROR("Out of memory\n");
  			goto out;
  		}
  
  		list_add_tail(&grp->grp_link,
  				  &console_session.ses_grp_list);
  		lstcon_group_addref(grp);
  	}
  
  	if (grp->grp_ref > 2) {
  		/* Group in using */
  		jrep->join_status = EBUSY;
  		goto out;
  	}
  
  	rc = lstcon_group_ndlink_find(grp, rpc->srpc_peer, &ndl, 0);
  	if (rc == 0) {
  		jrep->join_status = EEXIST;
  		goto out;
  	}
  
  	rc = lstcon_group_ndlink_find(grp, rpc->srpc_peer, &ndl, 1);
  	if (rc != 0) {
  		CERROR("Out of memory\n");
  		goto out;
  	}
  
  	ndl->ndl_node->nd_state   = LST_NODE_ACTIVE;
  	ndl->ndl_node->nd_timeout = console_session.ses_timeout;
  
  	if (grp->grp_userland == 0)
  		grp->grp_userland = 1;
  
  	strcpy(jrep->join_session, console_session.ses_name);
  	jrep->join_timeout = console_session.ses_timeout;
  	jrep->join_status  = 0;
  
  out:
  	rep->msg_ses_feats = console_session.ses_features;
  	if (grp != NULL)
  		lstcon_group_put(grp);
  
  	mutex_unlock(&console_session.ses_mutex);
  
  	return rc;
  }
  
  srpc_service_t lstcon_acceptor_service;
  void lstcon_init_acceptor_service(void)
  {
  	/* initialize selftest console acceptor service table */
  	lstcon_acceptor_service.sv_name    = "join session";
  	lstcon_acceptor_service.sv_handler = lstcon_acceptor_handle;
  	lstcon_acceptor_service.sv_id      = SRPC_SERVICE_JOIN;
  	lstcon_acceptor_service.sv_wi_total = SFW_FRWK_WI_MAX;
  }
  
  extern int lstcon_ioctl_entry(unsigned int cmd, struct libcfs_ioctl_data *data);
  
  DECLARE_IOCTL_HANDLER(lstcon_ioctl_handler, lstcon_ioctl_entry);
  
  /* initialize console */
  int
  lstcon_console_init(void)
  {
  	int     i;
  	int     rc;
  
  	memset(&console_session, 0, sizeof(lstcon_session_t));
  
  	console_session.ses_id		    = LST_INVALID_SID;
  	console_session.ses_state	    = LST_SESSION_NONE;
  	console_session.ses_timeout	    = 0;
  	console_session.ses_force	    = 0;
  	console_session.ses_expired	    = 0;
  	console_session.ses_feats_updated   = 0;
  	console_session.ses_features	    = LST_FEATS_MASK;
  	console_session.ses_laststamp	    = cfs_time_current_sec();
  
  	mutex_init(&console_session.ses_mutex);
  
  	INIT_LIST_HEAD(&console_session.ses_ndl_list);
  	INIT_LIST_HEAD(&console_session.ses_grp_list);
  	INIT_LIST_HEAD(&console_session.ses_bat_list);
  	INIT_LIST_HEAD(&console_session.ses_trans_list);
  
  	LIBCFS_ALLOC(console_session.ses_ndl_hash,
  		     sizeof(struct list_head) * LST_GLOBAL_HASHSIZE);
  	if (console_session.ses_ndl_hash == NULL)
  		return -ENOMEM;
  
  	for (i = 0; i < LST_GLOBAL_HASHSIZE; i++)
  		INIT_LIST_HEAD(&console_session.ses_ndl_hash[i]);
  
  
  	/* initialize acceptor service table */
  	lstcon_init_acceptor_service();
  
  	rc = srpc_add_service(&lstcon_acceptor_service);
  	LASSERT (rc != -EBUSY);
  	if (rc != 0) {
  		LIBCFS_FREE(console_session.ses_ndl_hash,
  			    sizeof(struct list_head) * LST_GLOBAL_HASHSIZE);
  		return rc;
  	}
  
  	rc = srpc_service_add_buffers(&lstcon_acceptor_service,
  				      lstcon_acceptor_service.sv_wi_total);
  	if (rc != 0) {
  		rc = -ENOMEM;
  		goto out;
  	}
  
  	rc = libcfs_register_ioctl(&lstcon_ioctl_handler);
  
  	if (rc == 0) {
  		lstcon_rpc_module_init();
  		return 0;
  	}
  
  out:
  	srpc_shutdown_service(&lstcon_acceptor_service);
  	srpc_remove_service(&lstcon_acceptor_service);
  
  	LIBCFS_FREE(console_session.ses_ndl_hash,
  		    sizeof(struct list_head) * LST_GLOBAL_HASHSIZE);
  
  	srpc_wait_service_shutdown(&lstcon_acceptor_service);
  
  	return rc;
  }
  
  int
  lstcon_console_fini(void)
  {
  	int     i;
  
  	libcfs_deregister_ioctl(&lstcon_ioctl_handler);
  
  	mutex_lock(&console_session.ses_mutex);
  
  	srpc_shutdown_service(&lstcon_acceptor_service);
  	srpc_remove_service(&lstcon_acceptor_service);
  
  	if (console_session.ses_state != LST_SESSION_NONE)
  		lstcon_session_end();
  
  	lstcon_rpc_module_fini();
  
  	mutex_unlock(&console_session.ses_mutex);
  
  	LASSERT (list_empty(&console_session.ses_ndl_list));
  	LASSERT (list_empty(&console_session.ses_grp_list));
  	LASSERT (list_empty(&console_session.ses_bat_list));
  	LASSERT (list_empty(&console_session.ses_trans_list));
  
  	for (i = 0; i < LST_NODE_HASHSIZE; i++) {
  		LASSERT (list_empty(&console_session.ses_ndl_hash[i]));
  	}
  
  	LIBCFS_FREE(console_session.ses_ndl_hash,
  		    sizeof(struct list_head) * LST_GLOBAL_HASHSIZE);
  
  	srpc_wait_service_shutdown(&lstcon_acceptor_service);
  
  	return 0;
  }
  