vlib: introduce DMA infrastructure
[vpp.git] / src / vlib / threads.c
index b470976..5599c5b 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];
@@ -558,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)
@@ -638,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
@@ -674,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;
@@ -832,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;
 }
@@ -1480,6 +1476,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
  */
@@ -1549,12 +1595,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;
 
@@ -1566,11 +1618,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);
@@ -1606,12 +1671,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)