Using message queue to send commands from cli thread to bgp thread.

13 files changed, 285 insertions, 110 deletions

diff --git a/bgpd/bgp_main.c b/bgpd/bgp_main.c
index cc7319af..668bb9a5 100644
--- a/bgpd/bgp_main.c
+++ b/bgpd/bgp_main.c
@@ -73,6 +73,7 @@ static const struct option longopts[] =
 void sighup (void);
 void sigint (void);
 void sigusr1 (void);
+void sigusr2 (void);
 
 static void bgp_exit (int);
 
@@ -87,6 +88,10 @@ static struct quagga_signal_t bgp_signals[] =
     .handler = &sigusr1,
   },
   {
+    .signal = SIGUSR2,
+    .handler = &sigusr2,
+  },
+  {
     .signal = SIGINT,
     .handler = &sigint,
   },
@@ -209,6 +214,18 @@ sigusr1 (void)
   zlog_rotate (NULL);
 }
 
+/* SIGUSR2 handler. */
+void
+sigusr2 (void)
+{
+  /* Used to signal message queues */
+  if (qpthreads_enabled)
+    return;
+  else
+    exit(1);
+}
+
+
 /*
   Try to free up allocations we know about so that diagnostic tools such as
   valgrind are able to better illuminate leaks.
@@ -450,11 +467,11 @@ main (int argc, char **argv)
 
   if (qpthreads_enabled)
     {
-      cli_nexus = qpn_init_new(cli_nexus, 1); /* main thread */
-      bgp_nexus = qpn_init_new(bgp_nexus, 0);
+      cli_nexus = qpn_init_main(cli_nexus); /* main thread */
+      bgp_nexus = qpn_init_bgp(bgp_nexus);
 
       zprivs_init_r ();
-      vty_init_r(cli_nexus);
+      vty_init_r(cli_nexus, bgp_nexus);
     }
 
   /* Make bgp vty socket. */
@@ -469,8 +486,7 @@ main (int argc, char **argv)
   /* Launch finite state machines */
   if (qpthreads_enabled)
     {
-      /* for now BGP is still using threads */
-      qpn_exec_legacy(bgp_nexus);
+      qpn_exec(bgp_nexus);
       qpn_exec(cli_nexus);      /* must be last to start - on main thraed */
     }
   else
diff --git a/lib/Makefile.am b/lib/Makefile.am
index 3184a9b6..97f61e47 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -13,7 +13,8 @@ libzebra_la_SOURCES = \
 	filter.c routemap.c distribute.c stream.c str.c log.c plist.c \
 	zclient.c sockopt.c smux.c md5.c if_rmap.c keychain.c privs.c \
 	sigevent.c pqueue.c jhash.c memtypes.c workqueue.c symtab.c heap.c \
-	qtime.c qpthreads.c mqueue.c qpselect.c qtimers.c qpnexus.c
+	qtime.c qpthreads.c mqueue.c qpselect.c qtimers.c qpnexus.c \
+	command_queue.c
 
 BUILT_SOURCES = memtypes.h route_types.h
 
@@ -29,7 +30,8 @@ pkginclude_HEADERS = \
 	plist.h zclient.h sockopt.h smux.h md5.h if_rmap.h keychain.h \
 	privs.h sigevent.h pqueue.h jhash.h zassert.h memtypes.h \
 	workqueue.h route_types.h symtab.h heap.h \
-	qtime.h qpthreads.h mqueue.h qpselect.h qtimers.h qpnexus.h
+	qtime.h qpthreads.h mqueue.h qpselect.h qtimers.h qpnexus.h \
+	command_queue.h
 
 EXTRA_DIST = regex.c regex-gnu.h memtypes.awk route_types.awk route_types.txt
 
diff --git a/lib/command.c b/lib/command.c
index 80b113e8..c057c0a7 100644
--- a/lib/command.c
+++ b/lib/command.c
@@ -1996,7 +1996,7 @@ node_parent ( enum node_type node )
 /* Execute command by argument vline vector. */
 static int
 cmd_execute_command_real (vector vline, struct vty *vty,
-			  struct cmd_element **cmd)
+			  struct cmd_element **cmd, qpn_nexus bgp_nexus)
 {
   unsigned int i;
   unsigned int index;
@@ -2112,12 +2112,21 @@ cmd_execute_command_real (vector vline, struct vty *vty,
     return CMD_SUCCESS_DAEMON;
 
   /* Execute matched command. */
-  return (*matched_element->func) (matched_element, vty, argc, argv);
+  if (qpthreads_enabled)
+    {
+      /* Don't do it now, but send to bgp qpthread */
+      cq_enqueue(matched_element, vty, argc, argv, bgp_nexus);
+      return CMD_SUCCESS;
+    }
+  else
+    {
+      return (*matched_element->func) (matched_element, vty, argc, argv);
+    }
 }
 
 int
 cmd_execute_command (vector vline, struct vty *vty, struct cmd_element **cmd,
-		     int vtysh) {
+		     qpn_nexus bgp_nexus, int vtysh) {
   int ret, saved_ret, tried = 0;
   enum node_type onode, try_node;
 
@@ -2138,7 +2147,7 @@ cmd_execute_command (vector vline, struct vty *vty, struct cmd_element **cmd,
 	  vector_set_index (shifted_vline, index-1, vector_lookup(vline, index));
 	}
 
-      ret = cmd_execute_command_real (shifted_vline, vty, cmd);
+      ret = cmd_execute_command_real (shifted_vline, vty, cmd, bgp_nexus);
 
       vector_free(shifted_vline);
       vty->node = onode;
@@ -2146,7 +2155,7 @@ cmd_execute_command (vector vline, struct vty *vty, struct cmd_element **cmd,
   }
 
 
-  saved_ret = ret = cmd_execute_command_real (vline, vty, cmd);
+  saved_ret = ret = cmd_execute_command_real (vline, vty, cmd, bgp_nexus);
 
   if (vtysh)
     return saved_ret;
@@ -2157,7 +2166,7 @@ cmd_execute_command (vector vline, struct vty *vty, struct cmd_element **cmd,
     {
       try_node = node_parent(try_node);
       vty->node = try_node;
-      ret = cmd_execute_command_real (vline, vty, cmd);
+      ret = cmd_execute_command_real (vline, vty, cmd, bgp_nexus);
       tried = 1;
       if (ret == CMD_SUCCESS || ret == CMD_WARNING)
 	{
diff --git a/lib/command.h b/lib/command.h
index 1275efee..e4d51ee9 100644
--- a/lib/command.h
+++ b/lib/command.h
@@ -343,7 +343,8 @@ extern char **cmd_complete_command (vector, struct vty *, int *status);
 extern const char *cmd_prompt (enum node_type);
 extern int config_from_file (struct vty *, FILE *);
 extern enum node_type node_parent (enum node_type);
-extern int cmd_execute_command (vector, struct vty *, struct cmd_element **, int);
+extern int cmd_execute_command (vector, struct vty *, struct cmd_element **,
+    qpn_nexus, int);
 extern int cmd_execute_command_strict (vector, struct vty *, struct cmd_element **);
 extern void config_replace_string (struct cmd_element *, char *, ...);
 extern void cmd_init (int);
diff --git a/lib/command_queue.c b/lib/command_queue.c
new file mode 100644
index 00000000..a74398c3
--- /dev/null
+++ b/lib/command_queue.c
@@ -0,0 +1,83 @@
+/* Command Message Queue
+ * Copyright (C) 2009 Chris Hall (GMCH), Highwayman
+ *
+ * This file is part of GNU Zebra.
+ *
+ * GNU Zebra is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published
+ * by the Free Software Foundation; either version 2, or (at your
+ * option) any later version.
+ *
+ * GNU Zebra 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 for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Zebra; see the file COPYING.  If not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#include <zebra.h>
+
+#include "mqueue.h"
+#include "qpnexus.h"
+#include "memory.h"
+#include "command_queue.h"
+
+/* Prototypes */
+static void cq_action(mqueue_block mqb);
+
+/* We have too many parameters for a message queue block so have to marshal */
+struct marshal
+{
+  struct cmd_element *matched_element;
+  struct vty *vty;
+  int argc;
+  char **argv;
+};
+
+void
+cq_enqueue(struct cmd_element *matched_element, struct vty *vty,
+    int argc, const char *argv[], qpn_nexus bgp_nexus)
+{
+  struct marshal *wyatt = XCALLOC(MTYPE_MARSHAL, sizeof(struct marshal)) ;
+  int i;
+  mqueue_block mqb = NULL;
+
+  wyatt->matched_element = matched_element;
+  wyatt->vty = vty;
+  wyatt->argc = argc;
+  wyatt->argv = XCALLOC(MTYPE_MARSHAL, sizeof (char*) * argc);
+  for (i = 0; i < argc; ++i)
+    {
+      wyatt->argv[i] = XSTRDUP(MTYPE_MARSHAL, argv[i]);
+    }
+
+  mqb = mqb_init_new(mqb, cq_action, 0) ;
+  mqb_set_arg0_p(mqb, wyatt);
+  mqueue_enqueue(bgp_nexus->queue, mqb, 0) ;
+}
+
+/* dispatch a command from the message queue block */
+static void
+cq_action(mqueue_block mqb)
+{
+  int result;
+  int i;
+  struct marshal *wyatt = mqb_get_arg0_p(mqb);
+
+  /* Execute matched command. */
+  result = (*wyatt->matched_element->func)
+       (wyatt->matched_element, wyatt->vty, wyatt->argc, wyatt->argv);
+
+  /* clean up */
+  for (i = 0; i< wyatt->argc; ++i)
+    {
+      XFREE(MTYPE_MARSHAL, wyatt->argv[i]);
+    }
+  XFREE(MTYPE_MARSHAL, wyatt->argv);
+  XFREE(MTYPE_MARSHAL, wyatt);
+  mqb_free(mqb);
+}
diff --git a/lib/command_queue.h b/lib/command_queue.h
new file mode 100644
index 00000000..90a4f7b6
--- /dev/null
+++ b/lib/command_queue.h
@@ -0,0 +1,31 @@
+/* Command Message Queue -- header
+ * Copyright (C) 2009 Chris Hall (GMCH), Highwayman
+ *
+ * This file is part of GNU Zebra.
+ *
+ * GNU Zebra is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published
+ * by the Free Software Foundation; either version 2, or (at your
+ * option) any later version.
+ *
+ * GNU Zebra 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 for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNU Zebra; see the file COPYING.  If not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifndef COMMAND_QUEUE_H_
+#define COMMAND_QUEUE_H_
+
+#include "command.h"
+#include "qpnexus.h"
+
+extern void cq_enqueue(struct cmd_element *matched_element, struct vty *vty,
+    int argc, const char *argv[], qpn_nexus);
+
+#endif /* COMMAND_QUEUE_H_ */
diff --git a/lib/memtypes.c b/lib/memtypes.c
index 8146f21f..81acb205 100644
--- a/lib/memtypes.c
+++ b/lib/memtypes.c
@@ -40,6 +40,7 @@ struct memory_list memory_list_lib[] =
   { MTYPE_QTIMER_PILE,          "qtimer pile structure"         },
   { MTYPE_QTIMER,               "qtimer timer"                  },
   { MTYPE_QPN_NEXUS,            "qtn nexus"                     },
+  { MTYPE_MARSHAL,              "marshalled commands"            },
   { MTYPE_VTY,			"VTY"				},
   { MTYPE_VTY_OUT_BUF,		"VTY output buffer"		},
   { MTYPE_VTY_HIST,		"VTY history"			},
diff --git a/lib/qpnexus.c b/lib/qpnexus.c
index 1e71dc35..0f949443 100644
--- a/lib/qpnexus.c
+++ b/lib/qpnexus.c
@@ -24,12 +24,11 @@
 #include "qpnexus.h"
 #include "memory.h"
 #include "thread.h"
+#include "sigevent.h"
 
 /* prototypes */
-static void qpn_dispatch_queue(qpn_nexus qpn);
-static void* qpn_start(void* arg);
-static void* qpn_start_legacy(void* arg);
-static void qpn_thread_prep(qpn_nexus qpn);
+static void* qpn_start_main(void* arg);
+static void* qpn_start_bgp(void* arg);
 
 /* Master of the threads. */
 extern struct thread_master *master;
@@ -50,18 +49,37 @@ extern struct thread_master *master;
  * Returns the qtn_nexus.
  */
 qpn_nexus
-qpn_init_new(qpn_nexus qpn, int main)
+qpn_init_new(qpn_nexus qpn)
 {
   if (qpn == NULL)
     qpn = XCALLOC(MTYPE_QPN_NEXUS, sizeof(struct qpn_nexus)) ;
   else
     memset(qpn, 0,  sizeof(struct qpn_nexus)) ;
 
-   /* will change if we start new thread */
+  return qpn;
+}
+
+/* Initialize main qpthread, no queue */
+qpn_nexus
+qpn_init_main(qpn_nexus qpn)
+{
+  qpn = qpn_init_new(qpn);
+
   qpn->selection = qps_selection_init_new(qpn->selection);
   qpn->pile = qtimer_pile_init_new(qpn->pile);
+  qpn->main_thread = 1;
+  qpn->start = qpn_start_main;
+
+  return qpn;
+}
+
+/* Initialize bgp qpthread */
+qpn_nexus
+qpn_init_bgp(qpn_nexus qpn)
+{
+  qpn = qpn_init_new(qpn);
   qpn->queue = mqueue_init_new(qpn->queue, mqt_signal_unicast);
-  qpn->main_thread = main;
+  qpn->start = qpn_start_bgp;
 
   return qpn;
 }
@@ -74,15 +92,21 @@ qpn_free(qpn_nexus qpn)
   qtimer qtr;
 
   /* timers and the pile */
-  while ((qtr = qtimer_pile_ream(qpn->pile, 1)))
+  if (qpn->pile != NULL)
     {
-      qtimer_free(qtr);
+      while ((qtr = qtimer_pile_ream(qpn->pile, 1)))
+        {
+          qtimer_free(qtr);
+        }
     }
 
   /* files and selection */
-  while ((qf = qps_selection_ream(qpn->selection, 1)))
+  if (qpn->selection != NULL)
     {
-      qps_file_free(qf);
+      while ((qf = qps_selection_ream(qpn->selection, 1)))
+        {
+          qps_file_free(qf);
+        }
     }
 
   /* TODO: free qtn->queue */
@@ -99,34 +123,37 @@ qpn_exec(qpn_nexus qpn)
     {
       /* Run the state machine in calling thread */
       qpn->thread_id = qpt_thread_self();
-      qpn_start(qpn);
+      qpn->start(qpn);
     }
   else
     {
       /* create a qpthread and run the state machine in it */
-      qpn->thread_id = qpt_thread_create(qpn_start, qpn, NULL) ;
+      qpn->thread_id = qpt_thread_create(qpn->start, qpn, NULL) ;
     }
 }
 
-/* Prep thread and signals, then run finite state machine
+/* Main qpthread, prep signals, then run finite state machine
  * using qps_selection and qtimer
 */
 static void*
-qpn_start(void* arg)
+qpn_start_main(void* arg)
 {
   qpn_nexus qpn = arg;
   int actions;
+  qtime_mono_t now;
+  sigset_t newmask;
 
-  qpn_thread_prep(qpn);
+  /* Main thread, block the message queue's signal */
+  sigemptyset (&newmask);
+  sigaddset (&newmask, SIGMQUEUE);
+  qpt_thread_sigmask(SIG_BLOCK, &newmask, NULL);
+  qps_set_signal(qpn->selection, SIGMQUEUE, newmask);
 
   while (!qpn->terminate)
     {
-      qtime_mono_t now;
-
       /* Signals are highest priority.
        * only execute on the main thread */
-      if (qpn->main_thread)
-        quagga_sigevent_process ();
+      quagga_sigevent_process ();
 
       /* process timers */
       now = qt_get_monotonic();
@@ -143,9 +170,6 @@ qpn_start(void* arg)
         {
           actions = qps_dispatch_next(qpn->selection) ;
         }
-
-      /* process message queue */
-      qpn_dispatch_queue(qpn);
     }
 
   qpn_free(qpn);
@@ -153,84 +177,51 @@ qpn_start(void* arg)
   return NULL;
 }
 
-/* Create new qpthread and execute the thread state machine in it */
-void
-qpn_exec_legacy(qpn_nexus qpn)
-{
-  qpn->thread_id = qpt_thread_create(qpn_start_legacy, qpn, NULL) ;
-}
-
-/* Prep thread and signals, then run finite state machine
+/* Bgp prep signals, then run finite state machine
  * using legacy threads
 */
 static void*
-qpn_start_legacy(void* arg)
+qpn_start_bgp(void* arg)
 {
   qpn_nexus qpn = arg;
   struct thread thread;
+  mqueue_block mqb;
+  sigset_t newmask;
 
-  qpn_thread_prep(qpn);
+  /*
+   * Not main thread.  Block most signals, but be careful not to
+   * defer SIGTRAP because doing so breaks gdb, at least on
+   * NetBSD 2.0.  Avoid asking to block SIGKILL, just because
+   * we shouldn't be able to do so.
+   */
+  sigfillset (&newmask);
+  sigdelset (&newmask, SIGTRAP);
+  sigdelset (&newmask, SIGKILL);
+
+  qpt_thread_sigmask(SIG_BLOCK, &newmask, NULL);
+  qpn->mts = mqueue_thread_signal_init(qpn->mts, qpn->thread_id, SIGMQUEUE);
 
   while (!qpn->terminate)
     {
+
+      /* drain the message queue, will be waiting when it's empty */
+      for (;;)
+        {
+          mqb = mqueue_dequeue(qpn->queue, 1, qpn->mts) ;
+          if (mqb == NULL)
+            break;
+
+          mqb_dispatch(mqb);
+        }
+
+      /* TODO: use qpselect stuff */
       if (thread_fetch (master, &thread))
         thread_call (&thread);
 
-      /* process message queue, if any */
-      qpn_dispatch_queue(qpn);
+      mqueue_done_waiting(qpn->queue, qpn->mts);
     }
 
   qpn_free(qpn);
 
   return NULL;
 }
-
-/* dispatch any messages on our message queue */
-static void
-qpn_dispatch_queue(qpn_nexus qpn)
-{
-  mqueue_block mqb;
-
-  for (;;)
-    {
-      mqb = mqueue_dequeue(qpn->queue, 1, qpn->mts) ;
-      if (mqb == NULL)
-        return;
-
-      mqb_dispatch(mqb);
-    }
-}
-
-/* Init code to be run within the thread */
-static void
-qpn_thread_prep(qpn_nexus qpn)
-{
-  sigset_t newmask;
-
-  if (qpn->main_thread)
-    {
-      /* Main thread, block the message queue's signal */
-      sigemptyset (&newmask);
-      sigaddset (&newmask, SIGMQUEUE);
-    }
-  else
-    {
-      /*
-       * Not main thread.  Block most signals, but be careful not to
-       * defer SIGTRAP because doing so breaks gdb, at least on
-       * NetBSD 2.0.  Avoid asking to block SIGKILL, just because
-       * we shouldn't be able to do so.
-        */
-      sigfillset (&newmask);
-      sigdelset (&newmask, SIGTRAP);
-      sigdelset (&newmask, SIGKILL);
-  }
-
-  qpt_thread_sigmask(SIG_BLOCK, &newmask, NULL);
-  qps_set_signal(qpn->selection, SIGMQUEUE, newmask);
-
-  /* init mqueue_thread_signal here now we know our thread-id */
-  qpn->mts = mqueue_thread_signal_init(qpn->mts, qpn->thread_id, SIGMQUEUE);
-}
-
-
diff --git a/lib/qpnexus.h b/lib/qpnexus.h
index 339c9bc5..bda5234f 100644
--- a/lib/qpnexus.h
+++ b/lib/qpnexus.h
@@ -80,14 +80,18 @@ struct qpn_nexus
   mqueue_queue queue;
   mqueue_thread_signal mts;
 
+  /* qpthread routine */
+  void* (*start)(void*);
+
 };
 
 /*==============================================================================
  * Functions
  */
 
-extern qpn_nexus qpn_init_new(qpn_nexus qtn, int main);
+extern qpn_nexus qpn_init_new(qpn_nexus qtn);
+extern qpn_nexus qpn_init_main(qpn_nexus qtn);
+extern qpn_nexus qpn_init_bgp(qpn_nexus qtn);
 extern void qpn_exec(qpn_nexus qtn);
-extern void qpn_exec_legacy(qpn_nexus qtn);
 
 #endif /* _ZEBRA_QPNEXUS_H */
diff --git a/lib/thread.c b/lib/thread.c
index 589443a0..9fa1fd9f 100644
--- a/lib/thread.c
+++ b/lib/thread.c
@@ -41,6 +41,13 @@ static struct timeval relative_time_base;
 static unsigned short timers_inited;
 
 static struct hash *cpu_record = NULL;
+
+/* TODO: remove this */
+#define USE_MQUEUE
+#ifdef USE_MQUEUE
+#include "qpnexus.h"
+static sigset_t newmask;
+#endif
 
 /* Struct timeval's tv_usec one second value.  */
 #define TIMER_SECOND_MICRO 1000000L
@@ -417,6 +424,11 @@ thread_master_debug (struct thread_master *m)
 struct thread_master *
 thread_master_create ()
 {
+#ifdef USE_MQUEUE
+  sigfillset (&newmask);
+  sigdelset (&newmask, SIGMQUEUE);
+#endif
+
   if (cpu_record == NULL) 
     cpu_record 
       = hash_create_size (1011, (unsigned int (*) (void *))cpu_record_hash_key, 
@@ -949,13 +961,27 @@ thread_fetch (struct thread_master *m, struct thread *fetch)
 	  (!timer_wait || (timeval_cmp (*timer_wait, *timer_wait_bg) > 0)))
 	timer_wait = timer_wait_bg;
       
+      /* TODO: remove this */
+#ifdef USE_MQUEUE
+      num = pselect (FD_SETSIZE, &readfd, &writefd, &exceptfd, timer_wait, &newmask);
+#else
       num = select (FD_SETSIZE, &readfd, &writefd, &exceptfd, timer_wait);
-      
+#endif
+
       /* Signals should get quick treatment */
       if (num < 0)
         {
           if (errno == EINTR)
-            continue; /* signal received - process it */
+#ifdef USE_MQUEUE
+            {
+            if (qpthreads_enabled)
+              return NULL;
+            else
+              continue; /* signal received - process it */
+            }
+#else
+          continue; /* signal received - process it */
+#endif
           zlog_warn ("select() error: %s", safe_strerror (errno));
             return NULL;
         }
@@ -1128,3 +1154,4 @@ funcname_thread_execute (struct thread_master *m,
 
   return NULL;
 }
+#undef USE_MQUEUE
diff --git a/lib/vty.c b/lib/vty.c
index 64f99395..23f9bc9c 100644
--- a/lib/vty.c
+++ b/lib/vty.c
@@ -139,6 +139,7 @@ char integrate_default[] = SYSCONFDIR INTEGRATE_DEFAULT_CONFIG;
 /* Master of the threads. */
 static struct thread_master *master = NULL;
 static qpn_nexus cli_nexus = NULL;
+static qpn_nexus bgp_nexus = NULL;
 
 /* VTY standard output function.   vty == NULL or VTY_SHELL => stdout	*/
 int
@@ -226,6 +227,13 @@ uty_vout(struct vty *vty, const char *format, va_list args)
 	XFREE (MTYPE_VTY_OUT_BUF, p);
     }
 
+  if (cli_nexus != NULL && qpt_thread_self() != cli_nexus->thread_id)
+    {
+      /* Wake up */
+      vty_event (VTY_WRITE, vty->fd, vty);
+      qpt_thread_signal(cli_nexus->thread_id, SIGMQUEUE);
+    }
+
   return len;
 }
 
@@ -546,7 +554,7 @@ vty_command (struct vty *vty, char *buf)
 #endif /* CONSUMED_TIME_CHECK */
 
   UNLOCK
-  ret = cmd_execute_command (vline, vty, NULL, 0);
+  ret = cmd_execute_command (vline, vty, NULL, bgp_nexus, 0);
   LOCK
 
   /* Get the name of the protocol if any */
@@ -3465,9 +3473,10 @@ vty_init_vtysh ()
 
 /* qpthreads: Install vty's own commands like `who' command. */
 void
-vty_init_r (qpn_nexus qpn)
+vty_init_r (qpn_nexus cli_n, qpn_nexus bgp_n)
 {
-  cli_nexus = qpn;
+  cli_nexus = cli_n;
+  bgp_nexus = bgp_n;
   vty_mutex = qpt_mutex_init(vty_mutex, qpt_mutex_quagga);
 }
 
diff --git a/lib/vty.h b/lib/vty.h
index 2de8065c..753b8fac 100644
--- a/lib/vty.h
+++ b/lib/vty.h
@@ -26,6 +26,7 @@ Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 #include "qpthreads.h"
 #include "qpselect.h"
 #include "qtimers.h"
+#include "qpnexus.h"
 
 #define VTY_BUFSIZ 512
 #define VTY_MAXHIST 20
@@ -218,7 +219,7 @@ extern int vty_lock_asserted;
 #endif
 
 /* Prototypes. */
-extern void vty_init_r (qpn_nexus);
+extern void vty_init_r (qpn_nexus, qpn_nexus);
 extern void vty_exec_r(void);
 extern void vty_init (struct thread_master *);
 extern void vty_init_vtysh (void);
diff --git a/vtysh/vtysh.c b/vtysh/vtysh.c
index 935c4933..275e037c 100644
--- a/vtysh/vtysh.c
+++ b/vtysh/vtysh.c
@@ -290,7 +290,7 @@ vtysh_execute_func (const char *line, int pager)
   if (vline == NULL)
     return CMD_SUCCESS;
 
-  saved_ret = ret = cmd_execute_command (vline, vty, &cmd, 1);
+  saved_ret = ret = cmd_execute_command (vline, vty, &cmd, NULL, 1);
   saved_node = vty->node;
 
   /* If command doesn't succeeded in current node, try to walk up in node tree.
@@ -300,7 +300,7 @@ vtysh_execute_func (const char *line, int pager)
 	 && vty->node > CONFIG_NODE)
     {
       vty->node = node_parent(vty->node);
-      ret = cmd_execute_command (vline, vty, &cmd, 1);
+      ret = cmd_execute_command (vline, vty, &cmd, NULL, 1);
       tried++;
     }
 
@@ -398,7 +398,7 @@ vtysh_execute_func (const char *line, int pager)
 		    return CMD_SUCCESS;
 		  }
 
-		ret = cmd_execute_command (vline, vty, &cmd, 1);
+		ret = cmd_execute_command (vline, vty, &cmd, , NULL, 1);
 		cmd_free_strvec (vline);
 		if (ret != CMD_SUCCESS_DAEMON)
 		  break;