vppinfra: add os_get_online_cpu_core() and os_get_online_cpu_node()
[vpp.git] / src / vlib / threads.c
index b02271b..b96cbf9 100644 (file)
 #include <vppinfra/time_range.h>
 #include <vppinfra/interrupt.h>
 #include <vppinfra/linux/sysfs.h>
+#include <vppinfra/bitmap.h>
+#include <vppinfra/unix.h>
 #include <vlib/vlib.h>
 
 #include <vlib/threads.h>
 
-#include <vlib/stat_weak_inlines.h>
+#include <vlib/stats/stats.h>
 
 u32
 vl (void *p)
@@ -179,12 +181,15 @@ vlib_thread_init (vlib_main_t * vm)
   u32 first_index = 1;
   u32 i;
   uword *avail_cpu;
+  u32 stats_num_worker_threads_dir_index;
+
+  stats_num_worker_threads_dir_index =
+    vlib_stats_add_gauge ("/sys/num_worker_threads");
+  ASSERT (stats_num_worker_threads_dir_index != ~0);
 
   /* get bitmaps of active cpu cores and sockets */
-  tm->cpu_core_bitmap =
-    clib_sysfs_list_to_bitmap ("/sys/devices/system/cpu/online");
-  tm->cpu_socket_bitmap =
-    clib_sysfs_list_to_bitmap ("/sys/devices/system/node/online");
+  tm->cpu_core_bitmap = os_get_online_cpu_core_bitmap ();
+  tm->cpu_socket_bitmap = os_get_online_cpu_node_bitmap ();
 
   avail_cpu = clib_bitmap_dup (tm->cpu_core_bitmap);
 
@@ -199,45 +204,35 @@ vlib_thread_init (vlib_main_t * vm)
     }
 
   /* grab cpu for main thread */
-  if (tm->main_lcore == ~0)
-    {
-      /* if main-lcore is not set, we try to use lcore 1 */
-      if (clib_bitmap_get (avail_cpu, 1))
-       tm->main_lcore = 1;
-      else
-       tm->main_lcore = clib_bitmap_first_set (avail_cpu);
-      if (tm->main_lcore == (u8) ~ 0)
-       return clib_error_return (0, "no available cpus to be used for the"
-                                 " main thread");
-    }
-  else
+  if (tm->main_lcore != ~0)
     {
       if (clib_bitmap_get (avail_cpu, tm->main_lcore) == 0)
        return clib_error_return (0, "cpu %u is not available to be used"
                                  " for the main thread", tm->main_lcore);
+      avail_cpu = clib_bitmap_set (avail_cpu, tm->main_lcore, 0);
     }
-  avail_cpu = clib_bitmap_set (avail_cpu, tm->main_lcore, 0);
 
   /* assume that there is socket 0 only if there is no data from sysfs */
   if (!tm->cpu_socket_bitmap)
     tm->cpu_socket_bitmap = clib_bitmap_set (0, 0, 1);
 
   /* pin main thread to main_lcore  */
-  if (tm->cb.vlib_thread_set_lcore_cb)
-    {
-      tm->cb.vlib_thread_set_lcore_cb (0, tm->main_lcore);
-    }
-  else
+  if (tm->main_lcore != ~0)
     {
       cpu_set_t cpuset;
       CPU_ZERO (&cpuset);
       CPU_SET (tm->main_lcore, &cpuset);
-      pthread_setaffinity_np (pthread_self (), sizeof (cpu_set_t), &cpuset);
+      if (pthread_setaffinity_np (pthread_self (), sizeof (cpu_set_t),
+                                 &cpuset))
+       {
+         return clib_error_return (0, "could not pin main thread to cpu %u",
+                                   tm->main_lcore);
+       }
     }
 
   /* Set up thread 0 */
   vec_validate_aligned (vlib_worker_threads, 0, CLIB_CACHE_LINE_BYTES);
-  _vec_len (vlib_worker_threads) = 1;
+  vec_set_len (vlib_worker_threads, 1);
   w = vlib_worker_threads;
   w->thread_mheap = clib_mem_get_heap ();
   w->thread_stack = vlib_thread_stacks[0];
@@ -314,7 +309,8 @@ vlib_thread_init (vlib_main_t * vm)
              if (c == ~0)
                return clib_error_return (0,
                                          "no available cpus to be used for"
-                                         " the '%s' thread", tr->name);
+                                         " the '%s' thread #%u",
+                                         tr->name, tr->count);
 
              avail_cpu = clib_bitmap_set (avail_cpu, 0, avail_c0);
              avail_cpu = clib_bitmap_set (avail_cpu, c, 0);
@@ -326,6 +322,7 @@ vlib_thread_init (vlib_main_t * vm)
   clib_bitmap_free (avail_cpu);
 
   tm->n_vlib_mains = n_vlib_mains;
+  vlib_stats_set_gauge (stats_num_worker_threads_dir_index, n_vlib_mains - 1);
 
   /*
    * Allocate the remaining worker threads, and thread stack vector slots
@@ -346,26 +343,13 @@ vlib_frame_queue_alloc (int nelts)
   fq = clib_mem_alloc_aligned (sizeof (*fq), CLIB_CACHE_LINE_BYTES);
   clib_memset (fq, 0, sizeof (*fq));
   fq->nelts = nelts;
-  fq->vector_threshold = 128;  // packets
+  fq->vector_threshold = 2 * VLIB_FRAME_SIZE;
   vec_validate_aligned (fq->elts, nelts - 1, CLIB_CACHE_LINE_BYTES);
 
-  if (1)
+  if (nelts & (nelts - 1))
     {
-      if (((uword) & fq->tail) & (CLIB_CACHE_LINE_BYTES - 1))
-       fformat (stderr, "WARNING: fq->tail unaligned\n");
-      if (((uword) & fq->head) & (CLIB_CACHE_LINE_BYTES - 1))
-       fformat (stderr, "WARNING: fq->head unaligned\n");
-      if (((uword) fq->elts) & (CLIB_CACHE_LINE_BYTES - 1))
-       fformat (stderr, "WARNING: fq->elts unaligned\n");
-
-      if (sizeof (fq->elts[0]) % CLIB_CACHE_LINE_BYTES)
-       fformat (stderr, "WARNING: fq->elts[0] size %d\n",
-                sizeof (fq->elts[0]));
-      if (nelts & (nelts - 1))
-       {
-         fformat (stderr, "FATAL: nelts MUST be a power of 2\n");
-         abort ();
-       }
+      fformat (stderr, "FATAL: nelts MUST be a power of 2\n");
+      abort ();
     }
 
   return (fq);
@@ -418,7 +402,6 @@ vlib_worker_thread_init (vlib_worker_thread_t * w)
 void *
 vlib_worker_thread_bootstrap_fn (void *arg)
 {
-  void *rv;
   vlib_worker_thread_t *w = arg;
 
   w->lwp = syscall (SYS_gettid);
@@ -426,13 +409,21 @@ vlib_worker_thread_bootstrap_fn (void *arg)
 
   __os_thread_index = w - vlib_worker_threads;
 
-  vlib_process_start_switch_stack (vlib_get_main_by_index (__os_thread_index),
-                                  0);
-  rv = (void *) clib_calljmp
-    ((uword (*)(uword)) w->thread_function,
-     (uword) arg, w->thread_stack + VLIB_THREAD_STACK_SIZE);
-  /* NOTREACHED, we hope */
-  return rv;
+  if (CLIB_DEBUG > 0)
+    {
+      void *frame_addr = __builtin_frame_address (0);
+      if (frame_addr < (void *) w->thread_stack ||
+         frame_addr > (void *) w->thread_stack + VLIB_THREAD_STACK_SIZE)
+       {
+         /* heap is not set yet */
+         fprintf (stderr, "thread stack is not set properly\n");
+         exit (1);
+       }
+    }
+
+  w->thread_function (arg);
+
+  return 0;
 }
 
 void
@@ -449,9 +440,7 @@ vlib_get_thread_core_numa (vlib_worker_thread_t * w, unsigned cpu_id)
   clib_sysfs_read ((char *) p, "%d", &core_id);
   vec_reset_length (p);
 
-  /* *INDENT-OFF* */
-  clib_sysfs_read ("/sys/devices/system/node/online", "%U",
-        unformat_bitmap_list, &nbmp);
+  nbmp = os_get_online_cpu_node_bitmap ();
   clib_bitmap_foreach (node, nbmp)  {
     p = format (p, "%s%u/cpulist%c", sys_node_path, node, 0);
     clib_sysfs_read ((char *) p, "%U", unformat_bitmap_list, &cbmp);
@@ -460,7 +449,7 @@ vlib_get_thread_core_numa (vlib_worker_thread_t * w, unsigned cpu_id)
     vec_reset_length (cbmp);
     vec_reset_length (p);
   }
-  /* *INDENT-ON* */
+
   vec_free (nbmp);
   vec_free (cbmp);
   vec_free (p);
@@ -474,6 +463,9 @@ vlib_launch_thread_int (void *fp, vlib_worker_thread_t * w, unsigned cpu_id)
 {
   clib_mem_main_t *mm = &clib_mem_main;
   vlib_thread_main_t *tm = &vlib_thread_main;
+  pthread_t worker;
+  pthread_attr_t attr;
+  cpu_set_t cpuset;
   void *(*fp_arg) (void *) = fp;
   void *numa_heap;
 
@@ -500,29 +492,33 @@ vlib_launch_thread_int (void *fp, vlib_worker_thread_t * w, unsigned cpu_id)
        }
     }
 
-  if (tm->cb.vlib_launch_thread_cb && !w->registration->use_pthreads)
-    return tm->cb.vlib_launch_thread_cb (fp, (void *) w, cpu_id);
-  else
-    {
-      pthread_t worker;
-      cpu_set_t cpuset;
       CPU_ZERO (&cpuset);
       CPU_SET (cpu_id, &cpuset);
 
-      if (pthread_create (&worker, NULL /* attr */ , fp_arg, (void *) w))
+      if (pthread_attr_init (&attr))
+       return clib_error_return_unix (0, "pthread_attr_init");
+
+      if (pthread_attr_setstack (&attr, w->thread_stack,
+                                VLIB_THREAD_STACK_SIZE))
+       return clib_error_return_unix (0, "pthread_attr_setstack");
+
+      if (pthread_create (&worker, &attr, fp_arg, (void *) w))
        return clib_error_return_unix (0, "pthread_create");
 
       if (pthread_setaffinity_np (worker, sizeof (cpu_set_t), &cpuset))
        return clib_error_return_unix (0, "pthread_setaffinity_np");
 
+      if (pthread_attr_destroy (&attr))
+       return clib_error_return_unix (0, "pthread_attr_destroy");
+
       return 0;
-    }
 }
 
 static clib_error_t *
 start_workers (vlib_main_t * vm)
 {
   vlib_global_main_t *vgm = vlib_get_global_main ();
+  vlib_main_t *fvm = vlib_get_first_main ();
   int i, j;
   vlib_worker_thread_t *w;
   vlib_main_t *vm_clone;
@@ -532,6 +528,7 @@ start_workers (vlib_main_t * vm)
   vlib_node_runtime_t *rt;
   u32 n_vlib_mains = tm->n_vlib_mains;
   u32 worker_thread_index;
+  u32 stats_err_entry_index = fvm->error_main.stats_err_entry_index;
   clib_mem_heap_t *main_heap = clib_mem_get_per_cpu_heap ();
   vlib_stats_register_mem_heap (main_heap);
 
@@ -557,7 +554,7 @@ start_workers (vlib_main_t * vm)
 
   vec_validate_aligned (vgm->vlib_mains, n_vlib_mains - 1,
                        CLIB_CACHE_LINE_BYTES);
-  _vec_len (vgm->vlib_mains) = 0;
+  vec_set_len (vgm->vlib_mains, 0);
   vec_add1_aligned (vgm->vlib_mains, vm, CLIB_CACHE_LINE_BYTES);
 
   if (n_vlib_mains > 1)
@@ -601,6 +598,7 @@ start_workers (vlib_main_t * vm)
          for (k = 0; k < tr->count; k++)
            {
              vlib_node_t *n;
+             u64 **c;
 
              vec_add2 (vlib_worker_threads, w, 1);
              /* Currently unused, may not really work */
@@ -636,12 +634,9 @@ start_workers (vlib_main_t * vm)
                           sizeof (*vm_clone));
 
              vm_clone->thread_index = worker_thread_index;
-             vm_clone->heap_base = w->thread_mheap;
-             vm_clone->heap_aligned_base = (void *)
-               (((uword) w->thread_mheap) & ~(VLIB_FRAME_ALIGN - 1));
              vm_clone->pending_rpc_requests = 0;
              vec_validate (vm_clone->pending_rpc_requests, 0);
-             _vec_len (vm_clone->pending_rpc_requests) = 0;
+             vec_set_len (vm_clone->pending_rpc_requests, 0);
              clib_memset (&vm_clone->random_buffer, 0,
                           sizeof (vm_clone->random_buffer));
              clib_spinlock_init
@@ -671,7 +666,7 @@ start_workers (vlib_main_t * vm)
              /* fork the frame dispatch queue */
              nm_clone->pending_frames = 0;
              vec_validate (nm_clone->pending_frames, 10);
-             _vec_len (nm_clone->pending_frames) = 0;
+             vec_set_len (nm_clone->pending_frames, 0);
 
              /* fork nodes */
              nm_clone->nodes = 0;
@@ -697,7 +692,6 @@ start_workers (vlib_main_t * vm)
                           nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL])
              {
                vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-               rt->thread_index = vm_clone->thread_index;
                /* copy initial runtime_data from node */
                if (n->runtime_data && n->runtime_data_bytes > 0)
                  clib_memcpy (rt->runtime_data, n->runtime_data,
@@ -709,12 +703,14 @@ start_workers (vlib_main_t * vm)
                vec_dup_aligned (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT],
                                 CLIB_CACHE_LINE_BYTES);
              clib_interrupt_init (
-               &nm_clone->interrupts,
+               &nm_clone->input_node_interrupts,
                vec_len (nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT]));
+             clib_interrupt_init (
+               &nm_clone->pre_input_node_interrupts,
+               vec_len (nm_clone->nodes_by_type[VLIB_NODE_TYPE_PRE_INPUT]));
              vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
              {
                vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-               rt->thread_index = vm_clone->thread_index;
                /* copy initial runtime_data from node */
                if (n->runtime_data && n->runtime_data_bytes > 0)
                  clib_memcpy (rt->runtime_data, n->runtime_data,
@@ -729,7 +725,6 @@ start_workers (vlib_main_t * vm)
                           nm_clone->nodes_by_type[VLIB_NODE_TYPE_PRE_INPUT])
              {
                vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-               rt->thread_index = vm_clone->thread_index;
                /* copy initial runtime_data from node */
                if (n->runtime_data && n->runtime_data_bytes > 0)
                  clib_memcpy (rt->runtime_data, n->runtime_data,
@@ -741,10 +736,7 @@ start_workers (vlib_main_t * vm)
                                                     CLIB_CACHE_LINE_BYTES);
 
              /* Create per-thread frame freelist */
-             nm_clone->frame_sizes = vec_new (vlib_frame_size_t, 1);
-#ifdef VLIB_SUPPORTS_ARBITRARY_SCALAR_SIZES
-             nm_clone->frame_size_hash = hash_create (0, sizeof (uword));
-#endif
+             nm_clone->frame_sizes = 0;
              nm_clone->node_by_error = nm->node_by_error;
 
              /* Packet trace buffers are guaranteed to be empty, nothing to do here */
@@ -754,12 +746,10 @@ start_workers (vlib_main_t * vm)
                                CLIB_CACHE_LINE_BYTES);
 
              /* Switch to the stats segment ... */
-             void *oldheap = vlib_stats_push_heap (0);
-             vm_clone->error_main.counters =
-               vec_dup_aligned (vlib_get_first_main ()->error_main.counters,
-                                CLIB_CACHE_LINE_BYTES);
-             vlib_stats_pop_heap2 (vm_clone->error_main.counters,
-                                   worker_thread_index, oldheap, 1);
+             vlib_stats_validate (stats_err_entry_index, worker_thread_index,
+                                  vec_len (fvm->error_main.counters) - 1);
+             c = vlib_stats_get_entry_data_pointer (stats_err_entry_index);
+             vm_clone->error_main.counters = c[worker_thread_index];
 
              vm_clone->error_main.counters_last_clear = vec_dup_aligned (
                vlib_get_first_main ()->error_main.counters_last_clear,
@@ -815,28 +805,37 @@ start_workers (vlib_main_t * vm)
        {
          for (j = 0; j < tr->count; j++)
            {
+
              w = vlib_worker_threads + worker_thread_index++;
              err = vlib_launch_thread_int (vlib_worker_thread_bootstrap_fn,
                                            w, 0);
              if (err)
-               clib_error_report (err);
+               clib_unix_error ("%U, thread %s init on cpu %d failed",
+                                format_clib_error, err, tr->name, 0);
            }
        }
       else
        {
          uword c;
-          /* *INDENT-OFF* */
           clib_bitmap_foreach (c, tr->coremask)  {
             w = vlib_worker_threads + worker_thread_index++;
            err = vlib_launch_thread_int (vlib_worker_thread_bootstrap_fn,
                                          w, c);
            if (err)
-             clib_error_report (err);
-          }
-          /* *INDENT-ON* */
+             clib_unix_error ("%U, thread %s init on cpu %d failed",
+                              format_clib_error, err, tr->name, c);
+           }
        }
     }
   vlib_worker_thread_barrier_sync (vm);
+  {
+    clib_error_t *err;
+    err = vlib_call_init_exit_functions (
+      vm, &vgm->num_workers_change_function_registrations, 1 /* call_once */,
+      1 /* is_global */);
+    if (err)
+      clib_error_report (err);
+  }
   vlib_worker_thread_barrier_release (vm);
   return 0;
 }
@@ -898,6 +897,7 @@ vlib_worker_thread_node_refork (void)
   vlib_node_main_t *nm, *nm_clone;
   vlib_node_t **old_nodes_clone;
   vlib_node_runtime_t *rt, *old_rt;
+  u64 **c;
 
   vlib_node_t *new_n_clone;
 
@@ -909,24 +909,29 @@ vlib_worker_thread_node_refork (void)
   nm_clone = &vm_clone->node_main;
 
   /* Re-clone error heap */
-  u64 *old_counters = vm_clone->error_main.counters;
   u64 *old_counters_all_clear = vm_clone->error_main.counters_last_clear;
 
   clib_memcpy_fast (&vm_clone->error_main, &vm->error_main,
                    sizeof (vm->error_main));
   j = vec_len (vm->error_main.counters) - 1;
 
-  /* Switch to the stats segment ... */
-  void *oldheap = vlib_stats_push_heap (0);
-  vec_validate_aligned (old_counters, j, CLIB_CACHE_LINE_BYTES);
-  vm_clone->error_main.counters = old_counters;
-  vlib_stats_pop_heap2 (vm_clone->error_main.counters, vm_clone->thread_index,
-                       oldheap, 0);
+  c = vlib_stats_get_entry_data_pointer (vm->error_main.stats_err_entry_index);
+  vm_clone->error_main.counters = c[vm_clone->thread_index];
 
   vec_validate_aligned (old_counters_all_clear, j, CLIB_CACHE_LINE_BYTES);
   vm_clone->error_main.counters_last_clear = old_counters_all_clear;
 
-  nm_clone = &vm_clone->node_main;
+  for (j = 0; j < vec_len (nm_clone->next_frames); j++)
+    {
+      vlib_next_frame_t *nf = &nm_clone->next_frames[j];
+      if ((nf->flags & VLIB_FRAME_IS_ALLOCATED) && nf->frame != NULL)
+       {
+         vlib_frame_t *f = nf->frame;
+         nf->frame = NULL;
+         vlib_frame_free (vm_clone, f);
+       }
+    }
+
   vec_free (nm_clone->next_frames);
   nm_clone->next_frames = vec_dup_aligned (nm->next_frames,
                                           CLIB_CACHE_LINE_BYTES);
@@ -1001,7 +1006,6 @@ vlib_worker_thread_node_refork (void)
   vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL])
   {
     vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-    rt->thread_index = vm_clone->thread_index;
     /* copy runtime_data, will be overwritten later for existing rt */
     if (n->runtime_data && n->runtime_data_bytes > 0)
       clib_memcpy_fast (rt->runtime_data, n->runtime_data,
@@ -1026,13 +1030,15 @@ vlib_worker_thread_node_refork (void)
     vec_dup_aligned (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT],
                     CLIB_CACHE_LINE_BYTES);
   clib_interrupt_resize (
-    &nm_clone->interrupts,
+    &nm_clone->input_node_interrupts,
     vec_len (nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT]));
+  clib_interrupt_resize (
+    &nm_clone->pre_input_node_interrupts,
+    vec_len (nm_clone->nodes_by_type[VLIB_NODE_TYPE_PRE_INPUT]));
 
   vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
   {
     vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-    rt->thread_index = vm_clone->thread_index;
     /* copy runtime_data, will be overwritten later for existing rt */
     if (n->runtime_data && n->runtime_data_bytes > 0)
       clib_memcpy_fast (rt->runtime_data, n->runtime_data,
@@ -1060,7 +1066,6 @@ vlib_worker_thread_node_refork (void)
   vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_PRE_INPUT])
   {
     vlib_node_t *n = vlib_get_node (vm, rt->node_index);
-    rt->thread_index = vm_clone->thread_index;
     /* copy runtime_data, will be overwritten later for existing rt */
     if (n->runtime_data && n->runtime_data_bytes > 0)
       clib_memcpy_fast (rt->runtime_data, n->runtime_data,
@@ -1079,6 +1084,7 @@ vlib_worker_thread_node_refork (void)
 
   vec_free (old_rt);
 
+  vec_free (nm_clone->processes);
   nm_clone->processes = vec_dup_aligned (nm->processes,
                                         CLIB_CACHE_LINE_BYTES);
   nm_clone->node_by_error = nm->node_by_error;
@@ -1417,7 +1423,7 @@ vlib_worker_thread_barrier_release (vlib_main_t * vm)
        * rebuilding the stat segment node clones from the
        * stat thread...
        */
-      vlib_stat_segment_lock ();
+      vlib_stats_segment_lock ();
 
       /* Do stats elements on main thread */
       worker_thread_node_runtime_update_internal ();
@@ -1468,7 +1474,7 @@ vlib_worker_thread_barrier_release (vlib_main_t * vm)
              os_panic ();
            }
        }
-      vlib_stat_segment_unlock ();
+      vlib_stats_segment_unlock ();
     }
 
   t_closed_total = now - vm->barrier_epoch;
@@ -1492,6 +1498,57 @@ vlib_worker_thread_barrier_release (vlib_main_t * vm)
                         vm->clib_time.last_cpu_time, 1 /* leave */ );
 }
 
+static void
+vlib_worker_sync_rpc (void *args)
+{
+  ASSERT (vlib_thread_is_main_w_barrier ());
+  vlib_worker_threads->wait_before_barrier = 0;
+}
+
+void
+vlib_workers_sync (void)
+{
+  if (PREDICT_FALSE (!vlib_num_workers ()))
+    return;
+
+  if (!(*vlib_worker_threads->wait_at_barrier) &&
+      !clib_atomic_swap_rel_n (&vlib_worker_threads->wait_before_barrier, 1))
+    {
+      u32 thread_index = vlib_get_thread_index ();
+      vlib_rpc_call_main_thread (vlib_worker_sync_rpc, (u8 *) &thread_index,
+                                sizeof (thread_index));
+      vlib_worker_flush_pending_rpc_requests (vlib_get_main ());
+    }
+
+  /* Wait until main thread asks for barrier */
+  while (!(*vlib_worker_threads->wait_at_barrier))
+    ;
+
+  /* Stop before barrier and make sure all threads are either
+   * at worker barrier or the barrier before it */
+  clib_atomic_fetch_add (&vlib_worker_threads->workers_before_barrier, 1);
+  while (vlib_num_workers () > (*vlib_worker_threads->workers_at_barrier +
+                               vlib_worker_threads->workers_before_barrier))
+    ;
+}
+
+void
+vlib_workers_continue (void)
+{
+  if (PREDICT_FALSE (!vlib_num_workers ()))
+    return;
+
+  clib_atomic_fetch_add (&vlib_worker_threads->done_work_before_barrier, 1);
+
+  /* Wait until all workers are done with work before barrier */
+  while (vlib_worker_threads->done_work_before_barrier <
+        vlib_worker_threads->workers_before_barrier)
+    ;
+
+  clib_atomic_fetch_add (&vlib_worker_threads->done_work_before_barrier, -1);
+  clib_atomic_fetch_add (&vlib_worker_threads->workers_before_barrier, -1);
+}
+
 /**
  * Wait until each of the workers has been once around the track
  */
@@ -1528,17 +1585,27 @@ vlib_worker_wait_one_loop (void)
   return;
 }
 
+void
+vlib_worker_flush_pending_rpc_requests (vlib_main_t *vm)
+{
+  vlib_main_t *vm_global = vlib_get_first_main ();
+
+  ASSERT (vm != vm_global);
+
+  clib_spinlock_lock_if_init (&vm_global->pending_rpc_lock);
+  vec_append (vm_global->pending_rpc_requests, vm->pending_rpc_requests);
+  vec_reset_length (vm->pending_rpc_requests);
+  clib_spinlock_unlock_if_init (&vm_global->pending_rpc_lock);
+}
+
 void
 vlib_worker_thread_fn (void *arg)
 {
   vlib_global_main_t *vgm = vlib_get_global_main ();
   vlib_worker_thread_t *w = (vlib_worker_thread_t *) arg;
-  vlib_thread_main_t *tm = vlib_get_thread_main ();
   vlib_main_t *vm = vlib_get_main ();
   clib_error_t *e;
 
-  vlib_process_finish_switch_stack (vm);
-
   ASSERT (vm->thread_index == vlib_get_thread_index ());
 
   vlib_worker_thread_init (w);
@@ -1553,10 +1620,6 @@ vlib_worker_thread_fn (void *arg)
   if (e)
     clib_error_report (e);
 
-  /* Wait until the dpdk init sequence is complete */
-  while (tm->extern_thread_mgmt && tm->worker_thread_release == 0)
-    vlib_worker_thread_barrier_check ();
-
   vlib_worker_loop (vm);
 }
 
@@ -1568,12 +1631,18 @@ VLIB_REGISTER_THREAD (worker_thread_reg, static) = {
 };
 /* *INDENT-ON* */
 
+extern clib_march_fn_registration
+  *vlib_frame_queue_dequeue_with_aux_fn_march_fn_registrations;
+extern clib_march_fn_registration
+  *vlib_frame_queue_dequeue_fn_march_fn_registrations;
 u32
 vlib_frame_queue_main_init (u32 node_index, u32 frame_queue_nelts)
 {
   vlib_thread_main_t *tm = vlib_get_thread_main ();
+  vlib_main_t *vm = vlib_get_main ();
   vlib_frame_queue_main_t *fqm;
   vlib_frame_queue_t *fq;
+  vlib_node_t *node;
   int i;
   u32 num_threads;
 
@@ -1585,43 +1654,33 @@ vlib_frame_queue_main_init (u32 node_index, u32 frame_queue_nelts)
 
   vec_add2 (tm->frame_queue_mains, fqm, 1);
 
+  node = vlib_get_node (vm, fqm->node_index);
+  ASSERT (node);
+  if (node->aux_offset)
+    {
+      fqm->frame_queue_dequeue_fn =
+       CLIB_MARCH_FN_VOID_POINTER (vlib_frame_queue_dequeue_with_aux_fn);
+    }
+  else
+    {
+      fqm->frame_queue_dequeue_fn =
+       CLIB_MARCH_FN_VOID_POINTER (vlib_frame_queue_dequeue_fn);
+    }
+
   fqm->node_index = node_index;
   fqm->frame_queue_nelts = frame_queue_nelts;
-  fqm->queue_hi_thresh = frame_queue_nelts - num_threads;
 
   vec_validate (fqm->vlib_frame_queues, tm->n_vlib_mains - 1);
-  vec_validate (fqm->per_thread_data, tm->n_vlib_mains - 1);
-  _vec_len (fqm->vlib_frame_queues) = 0;
+  vec_set_len (fqm->vlib_frame_queues, 0);
   for (i = 0; i < tm->n_vlib_mains; i++)
     {
-      vlib_frame_queue_per_thread_data_t *ptd;
       fq = vlib_frame_queue_alloc (frame_queue_nelts);
       vec_add1 (fqm->vlib_frame_queues, fq);
-
-      ptd = vec_elt_at_index (fqm->per_thread_data, i);
-      vec_validate (ptd->handoff_queue_elt_by_thread_index,
-                   tm->n_vlib_mains - 1);
-      vec_validate_init_empty (ptd->congested_handoff_queue_by_thread_index,
-                              tm->n_vlib_mains - 1,
-                              (vlib_frame_queue_t *) (~0));
     }
 
   return (fqm - tm->frame_queue_mains);
 }
 
-int
-vlib_thread_cb_register (struct vlib_main_t *vm, vlib_thread_callbacks_t * cb)
-{
-  vlib_thread_main_t *tm = vlib_get_thread_main ();
-
-  if (tm->extern_thread_mgmt)
-    return -1;
-
-  tm->cb.vlib_launch_thread_cb = cb->vlib_launch_thread_cb;
-  tm->extern_thread_mgmt = 1;
-  return 0;
-}
-
 void
 vlib_process_signal_event_mt_helper (vlib_process_signal_event_mt_args_t *
                                     args)
@@ -1648,17 +1707,21 @@ vlib_rpc_call_main_thread (void *callback, u8 * args, u32 arg_size)
 clib_error_t *
 threads_init (vlib_main_t * vm)
 {
+  const vlib_thread_main_t *tm = vlib_get_thread_main ();
+
+  if (tm->main_lcore == ~0 && tm->n_vlib_mains > 1)
+    return clib_error_return (0, "Configuration error, a main core must "
+                                "be specified when using worker threads");
+
   return 0;
 }
 
 VLIB_INIT_FUNCTION (threads_init);
 
-
 static clib_error_t *
 show_clock_command_fn (vlib_main_t * vm,
                       unformat_input_t * input, vlib_cli_command_t * cmd)
 {
-  int i;
   int verbose = 0;
   clib_timebase_t _tb, *tb = &_tb;
 
@@ -1671,24 +1734,19 @@ show_clock_command_fn (vlib_main_t * vm,
                   verbose, format_clib_timebase_time,
                   clib_timebase_now (tb));
 
-  if (vlib_get_n_threads () == 1)
-    return 0;
-
   vlib_cli_output (vm, "Time last barrier release %.9f",
                   vm->time_last_barrier_release);
 
-  for (i = 1; i < vlib_get_n_threads (); i++)
+  foreach_vlib_main ()
     {
-      vlib_main_t *ovm = vlib_get_main_by_index (i);
-      if (ovm == 0)
-       continue;
-
-      vlib_cli_output (vm, "%d: %U", i, format_clib_time, &ovm->clib_time,
-                      verbose);
-
-      vlib_cli_output (
-       vm, "Thread %d offset %.9f error %.9f", i, ovm->time_offset,
-       vm->time_last_barrier_release - ovm->time_last_barrier_release);
+      vlib_cli_output (vm, "%d: %U", this_vlib_main->thread_index,
+                      format_clib_time, &this_vlib_main->clib_time, verbose);
+
+      vlib_cli_output (vm, "Thread %d offset %.9f error %.9f",
+                      this_vlib_main->thread_index,
+                      this_vlib_main->time_offset,
+                      vm->time_last_barrier_release -
+                        this_vlib_main->time_last_barrier_release);
     }
   return 0;
 }