vlib: don't leak node frames on refork
[vpp.git] / src / vlib / threads.c
index 3cd8c01..adf225b 100644 (file)
@@ -175,7 +175,6 @@ vlib_thread_init (vlib_main_t * vm)
   vlib_thread_main_t *tm = &vlib_thread_main;
   vlib_worker_thread_t *w;
   vlib_thread_registration_t *tr;
-  cpu_set_t cpuset;
   u32 n_vlib_mains = 1;
   u32 first_index = 1;
   u32 i;
@@ -205,37 +204,30 @@ 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  */
-  CPU_ZERO (&cpuset);
-  CPU_SET (tm->main_lcore, &cpuset);
-  pthread_setaffinity_np (pthread_self (), sizeof (cpu_set_t), &cpuset);
+  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);
+    }
 
   /* 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];
@@ -404,23 +396,28 @@ 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;
-  vlib_main_t *vm = 0;
 
   w->lwp = syscall (SYS_gettid);
   w->thread_id = pthread_self ();
 
   __os_thread_index = w - vlib_worker_threads;
 
-  vm = vlib_global_main.vlib_mains[__os_thread_index];
+  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);
+       }
+    }
 
-  vlib_process_start_switch_stack (vm, 0);
-  rv = (void *) clib_calljmp
-    ((uword (*)(uword)) w->thread_function,
-     (uword) arg, w->thread_stack + VLIB_THREAD_STACK_SIZE);
-  /* NOTREACHED, we hope */
-  return rv;
+  w->thread_function (arg);
+
+  return 0;
 }
 
 void
@@ -463,6 +460,7 @@ 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;
@@ -493,12 +491,22 @@ vlib_launch_thread_int (void *fp, vlib_worker_thread_t * w, unsigned cpu_id)
       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;
 }
 
@@ -506,6 +514,7 @@ 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;
@@ -515,6 +524,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);
 
@@ -540,7 +550,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)
@@ -584,6 +594,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 */
@@ -619,13 +630,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) &
-                         ~(CLIB_CACHE_LINE_BYTES - 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
@@ -655,7 +662,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;
@@ -732,13 +739,10 @@ start_workers (vlib_main_t * vm)
                                CLIB_CACHE_LINE_BYTES);
 
              /* Switch to the stats segment ... */
-             void *oldheap = vlib_stats_set_heap ();
-             vm_clone->error_main.counters =
-               vec_dup_aligned (vlib_get_first_main ()->error_main.counters,
-                                CLIB_CACHE_LINE_BYTES);
-             clib_mem_set_heap (oldheap);
-             vlib_stats_update_error_vector (vm_clone->error_main.counters,
-                                             worker_thread_index, 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,
@@ -816,6 +820,14 @@ start_workers (vlib_main_t * vm)
        }
     }
   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;
 }
@@ -877,6 +889,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;
 
@@ -888,25 +901,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_set_heap ();
-  vec_validate_aligned (old_counters, j, CLIB_CACHE_LINE_BYTES);
-  clib_mem_set_heap (oldheap);
-  vm_clone->error_main.counters = old_counters;
-  vlib_stats_update_error_vector (vm_clone->error_main.counters,
-                                 vm_clone->thread_index, 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);
@@ -1470,6 +1487,56 @@ 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));
+    }
+
+  /* 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
  */
@@ -1514,8 +1581,6 @@ vlib_worker_thread_fn (void *arg)
   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);
@@ -1541,12 +1606,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;
 
@@ -1558,11 +1629,24 @@ 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;
 
   vec_validate (fqm->vlib_frame_queues, 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++)
     {
       fq = vlib_frame_queue_alloc (frame_queue_nelts);
@@ -1598,12 +1682,17 @@ 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)