vlib/vlib/threads.c

   1 /*
   2  * Copyright (c) 2015 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15 #define _GNU_SOURCE
  16 #include <sched.h>
  17
  18 #include <signal.h>
  19 #include <math.h>
  20 #include <vppinfra/format.h>
  21 #include <vlib/vlib.h>
  22
  23 #include <vlib/threads.h>
  24 #include <vlib/unix/cj.h>
  25
  26
  27 #if DPDK==1
  28 #include <rte_config.h>
  29 #include <rte_common.h>
  30 #include <rte_eal.h>
  31 #include <rte_launch.h>
  32 #include <rte_lcore.h>
  33 #endif
  34 DECLARE_CJ_GLOBAL_LOG;
  35
  36 #define FRAME_QUEUE_NELTS 32
  37
  38
  39 #if DPDK==1
  40 /*
  41  *  Weak definitions of DPDK symbols used in this file.
  42  *  Needed for linking test programs without DPDK libs.
  43  */
  44 unsigned __thread __attribute__ ((weak)) RTE_PER_LCORE (_lcore_id);
  45 struct lcore_config __attribute__ ((weak)) lcore_config[];
  46 unsigned __attribute__ ((weak)) rte_socket_id ();
  47 int __attribute__ ((weak)) rte_eal_remote_launch ();
  48 #endif
  49 u32
  50 vl (void *p)
  51 {
  52   return vec_len (p);
  53 }
  54
  55 vlib_thread_main_t vlib_thread_main;
  56
  57 uword
  58 os_get_cpu_number (void)
  59 {
  60   void *sp;
  61   uword n;
  62   u32 len;
  63
  64   len = vec_len (vlib_thread_stacks);
  65   if (len == 0)
  66     return 0;
  67
  68   /* Get any old stack address. */
  69   sp = &sp;
  70
  71   n = ((uword) sp - (uword) vlib_thread_stacks[0])
  72     >> VLIB_LOG2_THREAD_STACK_SIZE;
  73
  74   /* "processes" have their own stacks, and they always run in thread 0 */
  75   n = n >= len ? 0 : n;
  76
  77   return n;
  78 }
  79
  80 uword
  81 os_get_ncpus (void)
  82 {
  83   u32 len;
  84
  85   len = vec_len (vlib_thread_stacks);
  86   if (len == 0)
  87     return 1;
  88   else
  89     return len;
  90 }
  91
  92 void
  93 vlib_set_thread_name (char *name)
  94 {
  95   int pthread_setname_np (pthread_t __target_thread, const char *__name);
  96   int rv;
  97   pthread_t thread = pthread_self ();
  98
  99   if (thread)
 100     {
 101       rv = pthread_setname_np (thread, name);
 102       if (rv)
 103         clib_warning ("pthread_setname_np returned %d", rv);
 104     }
 105 }
 106
 107 static int
 108 sort_registrations_by_no_clone (void *a0, void *a1)
 109 {
 110   vlib_thread_registration_t **tr0 = a0;
 111   vlib_thread_registration_t **tr1 = a1;
 112
 113   return ((i32) ((*tr0)->no_data_structure_clone)
 114           - ((i32) ((*tr1)->no_data_structure_clone)));
 115 }
 116
 117 static uword *
 118 vlib_sysfs_list_to_bitmap (char *filename)
 119 {
 120   FILE *fp;
 121   uword *r = 0;
 122
 123   fp = fopen (filename, "r");
 124
 125   if (fp != NULL)
 126     {
 127       u8 *buffer = 0;
 128       vec_validate (buffer, 256 - 1);
 129       if (fgets ((char *) buffer, 256, fp))
 130         {
 131           unformat_input_t in;
 132           unformat_init_string (&in, (char *) buffer,
 133                                 strlen ((char *) buffer));
 134           if (unformat (&in, "%U", unformat_bitmap_list, &r) != 1)
 135             clib_warning ("unformat_bitmap_list failed");
 136           unformat_free (&in);
 137         }
 138       vec_free (buffer);
 139       fclose (fp);
 140     }
 141   return r;
 142 }
 143
 144
 145 /* Called early in the init sequence */
 146
 147 clib_error_t *
 148 vlib_thread_init (vlib_main_t * vm)
 149 {
 150   vlib_thread_main_t *tm = &vlib_thread_main;
 151   vlib_worker_thread_t *w;
 152   vlib_thread_registration_t *tr;
 153   u32 n_vlib_mains = 1;
 154   u32 first_index = 1;
 155   u32 i;
 156   uword *avail_cpu;
 157
 158   /* get bitmaps of active cpu cores and sockets */
 159   tm->cpu_core_bitmap =
 160     vlib_sysfs_list_to_bitmap ("/sys/devices/system/cpu/online");
 161   tm->cpu_socket_bitmap =
 162     vlib_sysfs_list_to_bitmap ("/sys/devices/system/node/online");
 163
 164   avail_cpu = clib_bitmap_dup (tm->cpu_core_bitmap);
 165
 166   /* skip cores */
 167   for (i = 0; i < tm->skip_cores; i++)
 168     {
 169       uword c = clib_bitmap_first_set (avail_cpu);
 170       if (c == ~0)
 171         return clib_error_return (0, "no available cpus to skip");
 172
 173       avail_cpu = clib_bitmap_set (avail_cpu, c, 0);
 174     }
 175
 176   /* grab cpu for main thread */
 177   if (!tm->main_lcore)
 178     {
 179       tm->main_lcore = clib_bitmap_first_set (avail_cpu);
 180       if (tm->main_lcore == (u8) ~ 0)
 181         return clib_error_return (0, "no available cpus to be used for the"
 182                                   " main thread");
 183     }
 184   else
 185     {
 186       if (clib_bitmap_get (avail_cpu, tm->main_lcore) == 0)
 187         return clib_error_return (0, "cpu %u is not available to be used"
 188                                   " for the main thread", tm->main_lcore);
 189     }
 190   avail_cpu = clib_bitmap_set (avail_cpu, tm->main_lcore, 0);
 191
 192   /* assume that there is socket 0 only if there is no data from sysfs */
 193   if (!tm->cpu_socket_bitmap)
 194     tm->cpu_socket_bitmap = clib_bitmap_set (0, 0, 1);
 195
 196   /* pin main thread to main_lcore  */
 197 #if DPDK==0
 198   {
 199     cpu_set_t cpuset;
 200     CPU_ZERO (&cpuset);
 201     CPU_SET (tm->main_lcore, &cpuset);
 202     pthread_setaffinity_np (pthread_self (), sizeof (cpu_set_t), &cpuset);
 203   }
 204 #endif
 205
 206   /* as many threads as stacks... */
 207   vec_validate_aligned (vlib_worker_threads, vec_len (vlib_thread_stacks) - 1,
 208                         CLIB_CACHE_LINE_BYTES);
 209
 210   /* Preallocate thread 0 */
 211   _vec_len (vlib_worker_threads) = 1;
 212   w = vlib_worker_threads;
 213   w->thread_mheap = clib_mem_get_heap ();
 214   w->thread_stack = vlib_thread_stacks[0];
 215   w->dpdk_lcore_id = -1;
 216   w->lwp = syscall (SYS_gettid);
 217   tm->n_vlib_mains = 1;
 218
 219   /* assign threads to cores and set n_vlib_mains */
 220   tr = tm->next;
 221
 222   while (tr)
 223     {
 224       vec_add1 (tm->registrations, tr);
 225       tr = tr->next;
 226     }
 227
 228   vec_sort_with_function (tm->registrations, sort_registrations_by_no_clone);
 229
 230   for (i = 0; i < vec_len (tm->registrations); i++)
 231     {
 232       int j;
 233       tr = tm->registrations[i];
 234       tr->first_index = first_index;
 235       first_index += tr->count;
 236       n_vlib_mains += (tr->no_data_structure_clone == 0) ? tr->count : 0;
 237
 238       /* construct coremask */
 239       if (tr->use_pthreads || !tr->count)
 240         continue;
 241
 242       if (tr->coremask)
 243         {
 244           uword c;
 245           /* *INDENT-OFF* */
 246           clib_bitmap_foreach (c, tr->coremask, ({
 247             if (clib_bitmap_get(avail_cpu, c) == 0)
 248               return clib_error_return (0, "cpu %u is not available to be used"
 249                                         " for the '%s' thread",c, tr->name);
 250
 251             avail_cpu = clib_bitmap_set(avail_cpu, c, 0);
 252           }));
 253 /* *INDENT-ON* */
 254
 255         }
 256       else
 257         {
 258           for (j = 0; j < tr->count; j++)
 259             {
 260               uword c = clib_bitmap_first_set (avail_cpu);
 261               if (c == ~0)
 262                 return clib_error_return (0,
 263                                           "no available cpus to be used for"
 264                                           " the '%s' thread", tr->name);
 265
 266               avail_cpu = clib_bitmap_set (avail_cpu, c, 0);
 267               tr->coremask = clib_bitmap_set (tr->coremask, c, 1);
 268             }
 269         }
 270     }
 271
 272   clib_bitmap_free (avail_cpu);
 273
 274   tm->n_vlib_mains = n_vlib_mains;
 275
 276   vec_validate_aligned (vlib_worker_threads, first_index - 1,
 277                         CLIB_CACHE_LINE_BYTES);
 278
 279
 280   tm->efd.enabled = VLIB_EFD_DISABLED;
 281   tm->efd.queue_hi_thresh = ((VLIB_EFD_DEF_WORKER_HI_THRESH_PCT *
 282                               FRAME_QUEUE_NELTS) / 100);
 283   return 0;
 284 }
 285
 286 vlib_worker_thread_t *
 287 vlib_alloc_thread (vlib_main_t * vm)
 288 {
 289   vlib_worker_thread_t *w;
 290
 291   if (vec_len (vlib_worker_threads) >= vec_len (vlib_thread_stacks))
 292     {
 293       clib_warning ("out of worker threads... Quitting...");
 294       exit (1);
 295     }
 296   vec_add2 (vlib_worker_threads, w, 1);
 297   w->thread_stack = vlib_thread_stacks[w - vlib_worker_threads];
 298   return w;
 299 }
 300
 301 vlib_frame_queue_t *
 302 vlib_frame_queue_alloc (int nelts)
 303 {
 304   vlib_frame_queue_t *fq;
 305
 306   fq = clib_mem_alloc_aligned (sizeof (*fq), CLIB_CACHE_LINE_BYTES);
 307   memset (fq, 0, sizeof (*fq));
 308   fq->nelts = nelts;
 309   fq->vector_threshold = 128;   // packets
 310   vec_validate_aligned (fq->elts, nelts - 1, CLIB_CACHE_LINE_BYTES);
 311
 312   if (1)
 313     {
 314       if (((uword) & fq->tail) & (CLIB_CACHE_LINE_BYTES - 1))
 315         fformat (stderr, "WARNING: fq->tail unaligned\n");
 316       if (((uword) & fq->head) & (CLIB_CACHE_LINE_BYTES - 1))
 317         fformat (stderr, "WARNING: fq->head unaligned\n");
 318       if (((uword) fq->elts) & (CLIB_CACHE_LINE_BYTES - 1))
 319         fformat (stderr, "WARNING: fq->elts unaligned\n");
 320
 321       if (sizeof (fq->elts[0]) % CLIB_CACHE_LINE_BYTES)
 322         fformat (stderr, "WARNING: fq->elts[0] size %d\n",
 323                  sizeof (fq->elts[0]));
 324       if (nelts & (nelts - 1))
 325         {
 326           fformat (stderr, "FATAL: nelts MUST be a power of 2\n");
 327           abort ();
 328         }
 329     }
 330
 331   return (fq);
 332 }
 333
 334 void vl_msg_api_handler_no_free (void *) __attribute__ ((weak));
 335 void
 336 vl_msg_api_handler_no_free (void *v)
 337 {
 338 }
 339
 340 /* Turned off, save as reference material... */
 341 #if 0
 342 static inline int
 343 vlib_frame_queue_dequeue_internal (int thread_id,
 344                                    vlib_main_t * vm, vlib_node_main_t * nm)
 345 {
 346   vlib_frame_queue_t *fq = vlib_frame_queues[thread_id];
 347   vlib_frame_queue_elt_t *elt;
 348   vlib_frame_t *f;
 349   vlib_pending_frame_t *p;
 350   vlib_node_runtime_t *r;
 351   u32 node_runtime_index;
 352   int msg_type;
 353   u64 before;
 354   int processed = 0;
 355
 356   ASSERT (vm == vlib_mains[thread_id]);
 357
 358   while (1)
 359     {
 360       if (fq->head == fq->tail)
 361         return processed;
 362
 363       elt = fq->elts + ((fq->head + 1) & (fq->nelts - 1));
 364
 365       if (!elt->valid)
 366         return processed;
 367
 368       before = clib_cpu_time_now ();
 369
 370       f = elt->frame;
 371       node_runtime_index = elt->node_runtime_index;
 372       msg_type = elt->msg_type;
 373
 374       switch (msg_type)
 375         {
 376         case VLIB_FRAME_QUEUE_ELT_FREE_BUFFERS:
 377           vlib_buffer_free (vm, vlib_frame_vector_args (f), f->n_vectors);
 378           /* note fallthrough... */
 379         case VLIB_FRAME_QUEUE_ELT_FREE_FRAME:
 380           r = vec_elt_at_index (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL],
 381                                 node_runtime_index);
 382           vlib_frame_free (vm, r, f);
 383           break;
 384         case VLIB_FRAME_QUEUE_ELT_DISPATCH_FRAME:
 385           vec_add2 (vm->node_main.pending_frames, p, 1);
 386           f->flags |= (VLIB_FRAME_PENDING | VLIB_FRAME_FREE_AFTER_DISPATCH);
 387           p->node_runtime_index = elt->node_runtime_index;
 388           p->frame_index = vlib_frame_index (vm, f);
 389           p->next_frame_index = VLIB_PENDING_FRAME_NO_NEXT_FRAME;
 390           fq->dequeue_vectors += (u64) f->n_vectors;
 391           break;
 392         case VLIB_FRAME_QUEUE_ELT_API_MSG:
 393           vl_msg_api_handler_no_free (f);
 394           break;
 395         default:
 396           clib_warning ("bogus frame queue message, type %d", msg_type);
 397           break;
 398         }
 399       elt->valid = 0;
 400       fq->dequeues++;
 401       fq->dequeue_ticks += clib_cpu_time_now () - before;
 402       CLIB_MEMORY_BARRIER ();
 403       fq->head++;
 404       processed++;
 405     }
 406   ASSERT (0);
 407   return processed;
 408 }
 409
 410 int
 411 vlib_frame_queue_dequeue (int thread_id,
 412                           vlib_main_t * vm, vlib_node_main_t * nm)
 413 {
 414   return vlib_frame_queue_dequeue_internal (thread_id, vm, nm);
 415 }
 416
 417 int
 418 vlib_frame_queue_enqueue (vlib_main_t * vm, u32 node_runtime_index,
 419                           u32 frame_queue_index, vlib_frame_t * frame,
 420                           vlib_frame_queue_msg_type_t type)
 421 {
 422   vlib_frame_queue_t *fq = vlib_frame_queues[frame_queue_index];
 423   vlib_frame_queue_elt_t *elt;
 424   u32 save_count;
 425   u64 new_tail;
 426   u64 before = clib_cpu_time_now ();
 427
 428   ASSERT (fq);
 429
 430   new_tail = __sync_add_and_fetch (&fq->tail, 1);
 431
 432   /* Wait until a ring slot is available */
 433   while (new_tail >= fq->head + fq->nelts)
 434     {
 435       f64 b4 = vlib_time_now_ticks (vm, before);
 436       vlib_worker_thread_barrier_check (vm, b4);
 437       /* Bad idea. Dequeue -> enqueue -> dequeue -> trouble */
 438       // vlib_frame_queue_dequeue (vm->cpu_index, vm, nm);
 439     }
 440
 441   elt = fq->elts + (new_tail & (fq->nelts - 1));
 442
 443   /* this would be very bad... */
 444   while (elt->valid)
 445     {
 446     }
 447
 448   /* Once we enqueue the frame, frame->n_vectors is owned elsewhere... */
 449   save_count = frame->n_vectors;
 450
 451   elt->frame = frame;
 452   elt->node_runtime_index = node_runtime_index;
 453   elt->msg_type = type;
 454   CLIB_MEMORY_BARRIER ();
 455   elt->valid = 1;
 456
 457   return save_count;
 458 }
 459 #endif /* 0 */
 460
 461 /* To be called by vlib worker threads upon startup */
 462 void
 463 vlib_worker_thread_init (vlib_worker_thread_t * w)
 464 {
 465   vlib_thread_main_t *tm = vlib_get_thread_main ();
 466
 467   /* worker threads wants no signals. */
 468   {
 469     sigset_t s;
 470     sigfillset (&s);
 471     pthread_sigmask (SIG_SETMASK, &s, 0);
 472   }
 473
 474   clib_mem_set_heap (w->thread_mheap);
 475
 476   if (vec_len (tm->thread_prefix) && w->registration->short_name)
 477     {
 478       w->name = format (0, "%v_%s_%d%c", tm->thread_prefix,
 479                         w->registration->short_name, w->instance_id, '\0');
 480       vlib_set_thread_name ((char *) w->name);
 481     }
 482
 483   if (!w->registration->use_pthreads)
 484     {
 485
 486       /* Initial barrier sync, for both worker and i/o threads */
 487       clib_smp_atomic_add (vlib_worker_threads->workers_at_barrier, 1);
 488
 489       while (*vlib_worker_threads->wait_at_barrier)
 490         ;
 491
 492       clib_smp_atomic_add (vlib_worker_threads->workers_at_barrier, -1);
 493     }
 494 }
 495
 496 void *
 497 vlib_worker_thread_bootstrap_fn (void *arg)
 498 {
 499   void *rv;
 500   vlib_worker_thread_t *w = arg;
 501
 502   w->lwp = syscall (SYS_gettid);
 503   w->dpdk_lcore_id = -1;
 504 #if DPDK==1
 505   if (w->registration && !w->registration->use_pthreads && rte_socket_id)       /* do we really have dpdk linked */
 506     {
 507       unsigned lcore = rte_lcore_id ();
 508       lcore = lcore < RTE_MAX_LCORE ? lcore : -1;
 509       w->dpdk_lcore_id = lcore;
 510     }
 511 #endif
 512
 513   rv = (void *) clib_calljmp
 514     ((uword (*)(uword)) w->thread_function,
 515      (uword) arg, w->thread_stack + VLIB_THREAD_STACK_SIZE);
 516   /* NOTREACHED, we hope */
 517   return rv;
 518 }
 519
 520 static int
 521 vlib_launch_thread (void *fp, vlib_worker_thread_t * w, unsigned lcore_id)
 522 {
 523   void *(*fp_arg) (void *) = fp;
 524
 525 #if DPDK==1
 526   if (!w->registration->use_pthreads)
 527     if (rte_eal_remote_launch)  /* do we have dpdk linked */
 528       return rte_eal_remote_launch (fp, (void *) w, lcore_id);
 529     else
 530       return -1;
 531   else
 532 #endif
 533     {
 534       int ret;
 535       pthread_t worker;
 536       cpu_set_t cpuset;
 537       CPU_ZERO (&cpuset);
 538       CPU_SET (lcore_id, &cpuset);
 539
 540       ret = pthread_create (&worker, NULL /* attr */ , fp_arg, (void *) w);
 541       if (ret == 0)
 542         return pthread_setaffinity_np (worker, sizeof (cpu_set_t), &cpuset);
 543       else
 544         return ret;
 545     }
 546 }
 547
 548 static clib_error_t *
 549 start_workers (vlib_main_t * vm)
 550 {
 551   int i, j;
 552   vlib_worker_thread_t *w;
 553   vlib_main_t *vm_clone;
 554   void *oldheap;
 555   vlib_frame_queue_t *fq;
 556   vlib_thread_main_t *tm = &vlib_thread_main;
 557   vlib_thread_registration_t *tr;
 558   vlib_node_runtime_t *rt;
 559   u32 n_vlib_mains = tm->n_vlib_mains;
 560   u32 worker_thread_index;
 561   u8 *main_heap = clib_mem_get_per_cpu_heap ();
 562   mheap_t *main_heap_header = mheap_header (main_heap);
 563
 564   vec_reset_length (vlib_worker_threads);
 565
 566   /* Set up the main thread */
 567   vec_add2_aligned (vlib_worker_threads, w, 1, CLIB_CACHE_LINE_BYTES);
 568   w->elog_track.name = "main thread";
 569   elog_track_register (&vm->elog_main, &w->elog_track);
 570
 571   if (vec_len (tm->thread_prefix))
 572     {
 573       w->name = format (0, "%v_main%c", tm->thread_prefix, '\0');
 574       vlib_set_thread_name ((char *) w->name);
 575     }
 576
 577 #if DPDK==1
 578   w->dpdk_lcore_id = -1;
 579   if (rte_socket_id)            /* do we really have dpdk linked */
 580     {
 581       unsigned lcore = rte_lcore_id ();
 582       w->dpdk_lcore_id = lcore < RTE_MAX_LCORE ? lcore : -1;;
 583     }
 584 #endif
 585
 586   /*
 587    * Truth of the matter: we always use at least two
 588    * threads. So, make the main heap thread-safe
 589    * and make the event log thread-safe.
 590    */
 591   main_heap_header->flags |= MHEAP_FLAG_THREAD_SAFE;
 592   vm->elog_main.lock =
 593     clib_mem_alloc_aligned (CLIB_CACHE_LINE_BYTES, CLIB_CACHE_LINE_BYTES);
 594   vm->elog_main.lock[0] = 0;
 595
 596   if (n_vlib_mains > 1)
 597     {
 598       vec_validate (vlib_mains, tm->n_vlib_mains - 1);
 599       _vec_len (vlib_mains) = 0;
 600       vec_add1 (vlib_mains, vm);
 601
 602       vec_validate (vlib_frame_queues, tm->n_vlib_mains - 1);
 603       _vec_len (vlib_frame_queues) = 0;
 604       fq = vlib_frame_queue_alloc (FRAME_QUEUE_NELTS);
 605       vec_add1 (vlib_frame_queues, fq);
 606
 607       vlib_worker_threads->wait_at_barrier =
 608         clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES);
 609       vlib_worker_threads->workers_at_barrier =
 610         clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES);
 611
 612       /* Ask for an initial barrier sync */
 613       *vlib_worker_threads->workers_at_barrier = 0;
 614       *vlib_worker_threads->wait_at_barrier = 1;
 615
 616       worker_thread_index = 1;
 617
 618       for (i = 0; i < vec_len (tm->registrations); i++)
 619         {
 620           vlib_node_main_t *nm, *nm_clone;
 621           vlib_buffer_main_t *bm_clone;
 622           vlib_buffer_free_list_t *fl_clone, *fl_orig;
 623           vlib_buffer_free_list_t *orig_freelist_pool;
 624           int k;
 625
 626           tr = tm->registrations[i];
 627
 628           if (tr->count == 0)
 629             continue;
 630
 631           for (k = 0; k < tr->count; k++)
 632             {
 633               vec_add2 (vlib_worker_threads, w, 1);
 634               if (tr->mheap_size)
 635                 w->thread_mheap =
 636                   mheap_alloc (0 /* use VM */ , tr->mheap_size);
 637               else
 638                 w->thread_mheap = main_heap;
 639               w->thread_stack = vlib_thread_stacks[w - vlib_worker_threads];
 640               w->thread_function = tr->function;
 641               w->thread_function_arg = w;
 642               w->instance_id = k;
 643               w->registration = tr;
 644
 645               w->elog_track.name =
 646                 (char *) format (0, "%s %d", tr->name, k + 1);
 647               vec_add1 (w->elog_track.name, 0);
 648               elog_track_register (&vm->elog_main, &w->elog_track);
 649
 650               if (tr->no_data_structure_clone)
 651                 continue;
 652
 653               /* Allocate "to-worker-N" frame queue */
 654               if (tr->frame_queue_nelts)
 655                 {
 656                   fq = vlib_frame_queue_alloc (tr->frame_queue_nelts);
 657                 }
 658               else
 659                 {
 660                   fq = vlib_frame_queue_alloc (FRAME_QUEUE_NELTS);
 661                 }
 662
 663               vec_validate (vlib_frame_queues, worker_thread_index);
 664               vlib_frame_queues[worker_thread_index] = fq;
 665
 666               /* Fork vlib_global_main et al. Look for bugs here */
 667               oldheap = clib_mem_set_heap (w->thread_mheap);
 668
 669               vm_clone = clib_mem_alloc (sizeof (*vm_clone));
 670               clib_memcpy (vm_clone, vlib_mains[0], sizeof (*vm_clone));
 671
 672               vm_clone->cpu_index = worker_thread_index;
 673               vm_clone->heap_base = w->thread_mheap;
 674               vm_clone->mbuf_alloc_list = 0;
 675               memset (&vm_clone->random_buffer, 0,
 676                       sizeof (vm_clone->random_buffer));
 677
 678               nm = &vlib_mains[0]->node_main;
 679               nm_clone = &vm_clone->node_main;
 680               /* fork next frames array, preserving node runtime indices */
 681               nm_clone->next_frames = vec_dup (nm->next_frames);
 682               for (j = 0; j < vec_len (nm_clone->next_frames); j++)
 683                 {
 684                   vlib_next_frame_t *nf = &nm_clone->next_frames[j];
 685                   u32 save_node_runtime_index;
 686                   u32 save_flags;
 687
 688                   save_node_runtime_index = nf->node_runtime_index;
 689                   save_flags = nf->flags & VLIB_FRAME_NO_FREE_AFTER_DISPATCH;
 690                   vlib_next_frame_init (nf);
 691                   nf->node_runtime_index = save_node_runtime_index;
 692                   nf->flags = save_flags;
 693                 }
 694
 695               /* fork the frame dispatch queue */
 696               nm_clone->pending_frames = 0;
 697               vec_validate (nm_clone->pending_frames, 10);      /* $$$$$?????? */
 698               _vec_len (nm_clone->pending_frames) = 0;
 699
 700               /* fork nodes */
 701               nm_clone->nodes = 0;
 702               for (j = 0; j < vec_len (nm->nodes); j++)
 703                 {
 704                   vlib_node_t *n;
 705                   n = clib_mem_alloc_no_fail (sizeof (*n));
 706                   clib_memcpy (n, nm->nodes[j], sizeof (*n));
 707                   /* none of the copied nodes have enqueue rights given out */
 708                   n->owner_node_index = VLIB_INVALID_NODE_INDEX;
 709                   memset (&n->stats_total, 0, sizeof (n->stats_total));
 710                   memset (&n->stats_last_clear, 0,
 711                           sizeof (n->stats_last_clear));
 712                   vec_add1 (nm_clone->nodes, n);
 713                 }
 714               nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL] =
 715                 vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]);
 716
 717               nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT] =
 718                 vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]);
 719               vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
 720                 rt->cpu_index = vm_clone->cpu_index;
 721
 722               nm_clone->processes = vec_dup (nm->processes);
 723
 724               /* zap the (per worker) frame freelists, etc */
 725               nm_clone->frame_sizes = 0;
 726               nm_clone->frame_size_hash = 0;
 727
 728               /* Packet trace buffers are guaranteed to be empty, nothing to do here */
 729
 730               clib_mem_set_heap (oldheap);
 731               vec_add1 (vlib_mains, vm_clone);
 732
 733               vm_clone->error_main.counters =
 734                 vec_dup (vlib_mains[0]->error_main.counters);
 735               vm_clone->error_main.counters_last_clear =
 736                 vec_dup (vlib_mains[0]->error_main.counters_last_clear);
 737
 738               /* Fork the vlib_buffer_main_t free lists, etc. */
 739               bm_clone = vec_dup (vm_clone->buffer_main);
 740               vm_clone->buffer_main = bm_clone;
 741
 742               orig_freelist_pool = bm_clone->buffer_free_list_pool;
 743               bm_clone->buffer_free_list_pool = 0;
 744
 745             /* *INDENT-OFF* */
 746             pool_foreach (fl_orig, orig_freelist_pool,
 747                           ({
 748                             pool_get_aligned (bm_clone->buffer_free_list_pool,
 749                                               fl_clone, CLIB_CACHE_LINE_BYTES);
 750                             ASSERT (fl_orig - orig_freelist_pool
 751                                     == fl_clone - bm_clone->buffer_free_list_pool);
 752
 753                             fl_clone[0] = fl_orig[0];
 754                             fl_clone->aligned_buffers = 0;
 755                             fl_clone->unaligned_buffers = 0;
 756                             fl_clone->n_alloc = 0;
 757                           }));
 758 /* *INDENT-ON* */
 759
 760               worker_thread_index++;
 761             }
 762         }
 763     }
 764   else
 765     {
 766       /* only have non-data-structure copy threads to create... */
 767       for (i = 0; i < vec_len (tm->registrations); i++)
 768         {
 769           tr = tm->registrations[i];
 770
 771           for (j = 0; j < tr->count; j++)
 772             {
 773               vec_add2 (vlib_worker_threads, w, 1);
 774               if (tr->mheap_size)
 775                 w->thread_mheap =
 776                   mheap_alloc (0 /* use VM */ , tr->mheap_size);
 777               else
 778                 w->thread_mheap = main_heap;
 779               w->thread_stack = vlib_thread_stacks[w - vlib_worker_threads];
 780               w->thread_function = tr->function;
 781               w->thread_function_arg = w;
 782               w->instance_id = j;
 783               w->elog_track.name =
 784                 (char *) format (0, "%s %d", tr->name, j + 1);
 785               w->registration = tr;
 786               vec_add1 (w->elog_track.name, 0);
 787               elog_track_register (&vm->elog_main, &w->elog_track);
 788             }
 789         }
 790     }
 791
 792   worker_thread_index = 1;
 793
 794   for (i = 0; i < vec_len (tm->registrations); i++)
 795     {
 796       int j;
 797
 798       tr = tm->registrations[i];
 799
 800       if (tr->use_pthreads || tm->use_pthreads)
 801         {
 802           for (j = 0; j < tr->count; j++)
 803             {
 804               w = vlib_worker_threads + worker_thread_index++;
 805               if (vlib_launch_thread (vlib_worker_thread_bootstrap_fn, w, 0) <
 806                   0)
 807                 clib_warning ("Couldn't start '%s' pthread ", tr->name);
 808             }
 809         }
 810       else
 811         {
 812           uword c;
 813             /* *INDENT-OFF* */
 814             clib_bitmap_foreach (c, tr->coremask, ({
 815               w = vlib_worker_threads + worker_thread_index++;
 816               if (vlib_launch_thread (vlib_worker_thread_bootstrap_fn, w, c) < 0)
 817                 clib_warning ("Couldn't start DPDK lcore %d", c);
 818
 819             }));
 820 /* *INDENT-ON* */
 821         }
 822     }
 823   vlib_worker_thread_barrier_sync (vm);
 824   vlib_worker_thread_barrier_release (vm);
 825   return 0;
 826 }
 827
 828 VLIB_MAIN_LOOP_ENTER_FUNCTION (start_workers);
 829
 830 void
 831 vlib_worker_thread_node_runtime_update (void)
 832 {
 833   int i, j;
 834   vlib_worker_thread_t *w;
 835   vlib_main_t *vm;
 836   vlib_node_main_t *nm, *nm_clone;
 837   vlib_node_t **old_nodes_clone;
 838   vlib_main_t *vm_clone;
 839   vlib_node_runtime_t *rt, *old_rt;
 840   void *oldheap;
 841   never_inline void
 842     vlib_node_runtime_sync_stats (vlib_main_t * vm,
 843                                   vlib_node_runtime_t * r,
 844                                   uword n_calls,
 845                                   uword n_vectors, uword n_clocks);
 846
 847   ASSERT (os_get_cpu_number () == 0);
 848
 849   if (vec_len (vlib_mains) == 0)
 850     return;
 851
 852   vm = vlib_mains[0];
 853   nm = &vm->node_main;
 854
 855   ASSERT (os_get_cpu_number () == 0);
 856   ASSERT (*vlib_worker_threads->wait_at_barrier == 1);
 857
 858   /*
 859    * Scrape all runtime stats, so we don't lose node runtime(s) with
 860    * pending counts, or throw away worker / io thread counts.
 861    */
 862   for (j = 0; j < vec_len (nm->nodes); j++)
 863     {
 864       vlib_node_t *n;
 865       n = nm->nodes[j];
 866       vlib_node_sync_stats (vm, n);
 867     }
 868
 869   for (i = 1; i < vec_len (vlib_mains); i++)
 870     {
 871       vlib_node_t *n;
 872
 873       vm_clone = vlib_mains[i];
 874       nm_clone = &vm_clone->node_main;
 875
 876       for (j = 0; j < vec_len (nm_clone->nodes); j++)
 877         {
 878           n = nm_clone->nodes[j];
 879
 880           rt = vlib_node_get_runtime (vm_clone, n->index);
 881           vlib_node_runtime_sync_stats (vm_clone, rt, 0, 0, 0);
 882         }
 883     }
 884
 885   for (i = 1; i < vec_len (vlib_mains); i++)
 886     {
 887       vlib_node_runtime_t *rt;
 888       w = vlib_worker_threads + i;
 889       oldheap = clib_mem_set_heap (w->thread_mheap);
 890
 891       vm_clone = vlib_mains[i];
 892
 893       /* Re-clone error heap */
 894       u64 *old_counters = vm_clone->error_main.counters;
 895       u64 *old_counters_all_clear = vm_clone->error_main.counters_last_clear;
 896       clib_memcpy (&vm_clone->error_main, &vm->error_main,
 897                    sizeof (vm->error_main));
 898       j = vec_len (vm->error_main.counters) - 1;
 899       vec_validate_aligned (old_counters, j, CLIB_CACHE_LINE_BYTES);
 900       vec_validate_aligned (old_counters_all_clear, j, CLIB_CACHE_LINE_BYTES);
 901       vm_clone->error_main.counters = old_counters;
 902       vm_clone->error_main.counters_last_clear = old_counters_all_clear;
 903
 904       nm_clone = &vm_clone->node_main;
 905       vec_free (nm_clone->next_frames);
 906       nm_clone->next_frames = vec_dup (nm->next_frames);
 907
 908       for (j = 0; j < vec_len (nm_clone->next_frames); j++)
 909         {
 910           vlib_next_frame_t *nf = &nm_clone->next_frames[j];
 911           u32 save_node_runtime_index;
 912           u32 save_flags;
 913
 914           save_node_runtime_index = nf->node_runtime_index;
 915           save_flags = nf->flags & VLIB_FRAME_NO_FREE_AFTER_DISPATCH;
 916           vlib_next_frame_init (nf);
 917           nf->node_runtime_index = save_node_runtime_index;
 918           nf->flags = save_flags;
 919         }
 920
 921       old_nodes_clone = nm_clone->nodes;
 922       nm_clone->nodes = 0;
 923
 924       /* re-fork nodes */
 925       for (j = 0; j < vec_len (nm->nodes); j++)
 926         {
 927           vlib_node_t *old_n_clone;
 928           vlib_node_t *new_n, *new_n_clone;
 929
 930           new_n = nm->nodes[j];
 931           old_n_clone = old_nodes_clone[j];
 932
 933           new_n_clone = clib_mem_alloc_no_fail (sizeof (*new_n_clone));
 934           clib_memcpy (new_n_clone, new_n, sizeof (*new_n));
 935           /* none of the copied nodes have enqueue rights given out */
 936           new_n_clone->owner_node_index = VLIB_INVALID_NODE_INDEX;
 937
 938           if (j >= vec_len (old_nodes_clone))
 939             {
 940               /* new node, set to zero */
 941               memset (&new_n_clone->stats_total, 0,
 942                       sizeof (new_n_clone->stats_total));
 943               memset (&new_n_clone->stats_last_clear, 0,
 944                       sizeof (new_n_clone->stats_last_clear));
 945             }
 946           else
 947             {
 948               /* Copy stats if the old data is valid */
 949               clib_memcpy (&new_n_clone->stats_total,
 950                            &old_n_clone->stats_total,
 951                            sizeof (new_n_clone->stats_total));
 952               clib_memcpy (&new_n_clone->stats_last_clear,
 953                            &old_n_clone->stats_last_clear,
 954                            sizeof (new_n_clone->stats_last_clear));
 955
 956               /* keep previous node state */
 957               new_n_clone->state = old_n_clone->state;
 958             }
 959           vec_add1 (nm_clone->nodes, new_n_clone);
 960         }
 961       /* Free the old node clone */
 962       for (j = 0; j < vec_len (old_nodes_clone); j++)
 963         clib_mem_free (old_nodes_clone[j]);
 964       vec_free (old_nodes_clone);
 965
 966       vec_free (nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]);
 967
 968       nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL] =
 969         vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]);
 970
 971       /* clone input node runtime */
 972       old_rt = nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT];
 973
 974       nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT] =
 975         vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]);
 976
 977       vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
 978       {
 979         rt->cpu_index = vm_clone->cpu_index;
 980       }
 981
 982       for (j = 0; j < vec_len (old_rt); j++)
 983         {
 984           rt = vlib_node_get_runtime (vm_clone, old_rt[j].node_index);
 985           rt->state = old_rt[j].state;
 986         }
 987
 988       vec_free (old_rt);
 989
 990       nm_clone->processes = vec_dup (nm->processes);
 991
 992       clib_mem_set_heap (oldheap);
 993
 994       // vnet_main_fork_fixup (i);
 995     }
 996 }
 997
 998 static clib_error_t *
 999 cpu_config (vlib_main_t * vm, unformat_input_t * input)
1000 {
1001   vlib_thread_registration_t *tr;
1002   uword *p;
1003   vlib_thread_main_t *tm = &vlib_thread_main;
1004   u8 *name;
1005   u64 coremask;
1006   uword *bitmap;
1007   u32 count;
1008
1009   tm->thread_registrations_by_name = hash_create_string (0, sizeof (uword));
1010   tm->n_thread_stacks = 1;      /* account for main thread */
1011
1012   tr = tm->next;
1013
1014   while (tr)
1015     {
1016       hash_set_mem (tm->thread_registrations_by_name, tr->name, (uword) tr);
1017       tr = tr->next;
1018     }
1019
1020   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
1021     {
1022       if (unformat (input, "use-pthreads"))
1023         tm->use_pthreads = 1;
1024       else if (unformat (input, "thread-prefix %v", &tm->thread_prefix))
1025         ;
1026       else if (unformat (input, "main-core %u", &tm->main_lcore))
1027         ;
1028       else if (unformat (input, "skip-cores %u", &tm->skip_cores))
1029         ;
1030       else if (unformat (input, "coremask-%s %llx", &name, &coremask))
1031         {
1032           p = hash_get_mem (tm->thread_registrations_by_name, name);
1033           if (p == 0)
1034             return clib_error_return (0, "no such thread type '%s'", name);
1035
1036           tr = (vlib_thread_registration_t *) p[0];
1037
1038           if (tr->use_pthreads)
1039             return clib_error_return (0,
1040                                       "coremask cannot be set for '%s' threads",
1041                                       name);
1042
1043           tr->coremask = clib_bitmap_set_multiple
1044             (tr->coremask, 0, coremask, BITS (coremask));
1045           tr->count = clib_bitmap_count_set_bits (tr->coremask);
1046         }
1047       else if (unformat (input, "corelist-%s %U", &name, unformat_bitmap_list,
1048                          &bitmap))
1049         {
1050           p = hash_get_mem (tm->thread_registrations_by_name, name);
1051           if (p == 0)
1052             return clib_error_return (0, "no such thread type '%s'", name);
1053
1054           tr = (vlib_thread_registration_t *) p[0];
1055
1056           if (tr->use_pthreads)
1057             return clib_error_return (0,
1058                                       "corelist cannot be set for '%s' threads",
1059                                       name);
1060
1061           tr->coremask = bitmap;
1062           tr->count = clib_bitmap_count_set_bits (tr->coremask);
1063         }
1064       else if (unformat (input, "%s %u", &name, &count))
1065         {
1066           p = hash_get_mem (tm->thread_registrations_by_name, name);
1067           if (p == 0)
1068             return clib_error_return (0, "no such thread type '%s'", name);
1069
1070           tr = (vlib_thread_registration_t *) p[0];
1071           if (tr->fixed_count)
1072             return clib_error_return
1073               (0, "number of %s threads not configurable", tr->name);
1074           tr->count = count;
1075         }
1076       else
1077         break;
1078     }
1079
1080   tr = tm->next;
1081
1082   if (!tm->thread_prefix)
1083     tm->thread_prefix = format (0, "vpp");
1084
1085   while (tr)
1086     {
1087       tm->n_thread_stacks += tr->count;
1088       tm->n_pthreads += tr->count * tr->use_pthreads;
1089       tm->n_eal_threads += tr->count * (tr->use_pthreads == 0);
1090       tr = tr->next;
1091     }
1092
1093   return 0;
1094 }
1095
1096 VLIB_EARLY_CONFIG_FUNCTION (cpu_config, "cpu");
1097
1098 #if !defined (__x86_64__) && !defined (__aarch64__) && !defined (__powerpc64__) && !defined(__arm__)
1099 void
1100 __sync_fetch_and_add_8 (void)
1101 {
1102   fformat (stderr, "%s called\n", __FUNCTION__);
1103   abort ();
1104 }
1105
1106 void
1107 __sync_add_and_fetch_8 (void)
1108 {
1109   fformat (stderr, "%s called\n", __FUNCTION__);
1110   abort ();
1111 }
1112 #endif
1113
1114 void vnet_main_fixup (vlib_fork_fixup_t which) __attribute__ ((weak));
1115 void
1116 vnet_main_fixup (vlib_fork_fixup_t which)
1117 {
1118 }
1119
1120 void
1121 vlib_worker_thread_fork_fixup (vlib_fork_fixup_t which)
1122 {
1123   vlib_main_t *vm = vlib_get_main ();
1124
1125   if (vlib_mains == 0)
1126     return;
1127
1128   ASSERT (os_get_cpu_number () == 0);
1129   vlib_worker_thread_barrier_sync (vm);
1130
1131   switch (which)
1132     {
1133     case VLIB_WORKER_THREAD_FORK_FIXUP_NEW_SW_IF_INDEX:
1134       vnet_main_fixup (VLIB_WORKER_THREAD_FORK_FIXUP_NEW_SW_IF_INDEX);
1135       break;
1136
1137     default:
1138       ASSERT (0);
1139     }
1140   vlib_worker_thread_barrier_release (vm);
1141 }
1142
1143 void
1144 vlib_worker_thread_barrier_sync (vlib_main_t * vm)
1145 {
1146   f64 deadline;
1147   u32 count;
1148
1149   if (!vlib_mains)
1150     return;
1151
1152   count = vec_len (vlib_mains) - 1;
1153
1154   /* Tolerate recursive calls */
1155   if (++vlib_worker_threads[0].recursion_level > 1)
1156     return;
1157
1158   vlib_worker_threads[0].barrier_sync_count++;
1159
1160   ASSERT (os_get_cpu_number () == 0);
1161
1162   deadline = vlib_time_now (vm) + BARRIER_SYNC_TIMEOUT;
1163
1164   *vlib_worker_threads->wait_at_barrier = 1;
1165   while (*vlib_worker_threads->workers_at_barrier != count)
1166     {
1167       if (vlib_time_now (vm) > deadline)
1168         {
1169           fformat (stderr, "%s: worker thread deadlock\n", __FUNCTION__);
1170           os_panic ();
1171         }
1172     }
1173 }
1174
1175 void
1176 vlib_worker_thread_barrier_release (vlib_main_t * vm)
1177 {
1178   f64 deadline;
1179
1180   if (!vlib_mains)
1181     return;
1182
1183   if (--vlib_worker_threads[0].recursion_level > 0)
1184     return;
1185
1186   deadline = vlib_time_now (vm) + BARRIER_SYNC_TIMEOUT;
1187
1188   *vlib_worker_threads->wait_at_barrier = 0;
1189
1190   while (*vlib_worker_threads->workers_at_barrier > 0)
1191     {
1192       if (vlib_time_now (vm) > deadline)
1193         {
1194           fformat (stderr, "%s: worker thread deadlock\n", __FUNCTION__);
1195           os_panic ();
1196         }
1197     }
1198 }
1199
1200 /*
1201  * Check the frame queue to see if any frames are available.
1202  * If so, pull the packets off the frames and put them to
1203  * the handoff node.
1204  */
1205 static inline int
1206 vlib_frame_queue_dequeue_internal (vlib_main_t * vm)
1207 {
1208   u32 thread_id = vm->cpu_index;
1209   vlib_frame_queue_t *fq = vlib_frame_queues[thread_id];
1210   vlib_frame_queue_elt_t *elt;
1211   u32 *from, *to;
1212   vlib_frame_t *f;
1213   int msg_type;
1214   int processed = 0;
1215   u32 n_left_to_node;
1216   u32 vectors = 0;
1217   vlib_thread_main_t *tm = vlib_get_thread_main ();
1218
1219   ASSERT (fq);
1220   ASSERT (vm == vlib_mains[thread_id]);
1221
1222   if (PREDICT_FALSE (tm->handoff_dispatch_node_index == ~0))
1223     return 0;
1224   /*
1225    * Gather trace data for frame queues
1226    */
1227   if (PREDICT_FALSE (fq->trace))
1228     {
1229       frame_queue_trace_t *fqt;
1230       frame_queue_nelt_counter_t *fqh;
1231       u32 elix;
1232
1233       fqt = &tm->frame_queue_traces[thread_id];
1234
1235       fqt->nelts = fq->nelts;
1236       fqt->head = fq->head;
1237       fqt->head_hint = fq->head_hint;
1238       fqt->tail = fq->tail;
1239       fqt->threshold = fq->vector_threshold;
1240       fqt->n_in_use = fqt->tail - fqt->head;
1241       if (fqt->n_in_use >= fqt->nelts)
1242         {
1243           // if beyond max then use max
1244           fqt->n_in_use = fqt->nelts - 1;
1245         }
1246
1247       /* Record the number of elements in use in the histogram */
1248       fqh = &tm->frame_queue_histogram[thread_id];
1249       fqh->count[fqt->n_in_use]++;
1250
1251       /* Record a snapshot of the elements in use */
1252       for (elix = 0; elix < fqt->nelts; elix++)
1253         {
1254           elt = fq->elts + ((fq->head + 1 + elix) & (fq->nelts - 1));
1255           if (1 || elt->valid)
1256             {
1257               fqt->n_vectors[elix] = elt->n_vectors;
1258             }
1259         }
1260       fqt->written = 1;
1261     }
1262
1263   while (1)
1264     {
1265       if (fq->head == fq->tail)
1266         {
1267           fq->head_hint = fq->head;
1268           return processed;
1269         }
1270
1271       elt = fq->elts + ((fq->head + 1) & (fq->nelts - 1));
1272
1273       if (!elt->valid)
1274         {
1275           fq->head_hint = fq->head;
1276           return processed;
1277         }
1278
1279       from = elt->buffer_index;
1280       msg_type = elt->msg_type;
1281
1282       ASSERT (msg_type == VLIB_FRAME_QUEUE_ELT_DISPATCH_FRAME);
1283       ASSERT (elt->n_vectors <= VLIB_FRAME_SIZE);
1284
1285       f = vlib_get_frame_to_node (vm, tm->handoff_dispatch_node_index);
1286
1287       to = vlib_frame_vector_args (f);
1288
1289       n_left_to_node = elt->n_vectors;
1290
1291       while (n_left_to_node >= 4)
1292         {
1293           to[0] = from[0];
1294           to[1] = from[1];
1295           to[2] = from[2];
1296           to[3] = from[3];
1297           to += 4;
1298           from += 4;
1299           n_left_to_node -= 4;
1300         }
1301
1302       while (n_left_to_node > 0)
1303         {
1304           to[0] = from[0];
1305           to++;
1306           from++;
1307           n_left_to_node--;
1308         }
1309
1310       vectors += elt->n_vectors;
1311       f->n_vectors = elt->n_vectors;
1312       vlib_put_frame_to_node (vm, tm->handoff_dispatch_node_index, f);
1313
1314       elt->valid = 0;
1315       elt->n_vectors = 0;
1316       elt->msg_type = 0xfefefefe;
1317       CLIB_MEMORY_BARRIER ();
1318       fq->head++;
1319       processed++;
1320
1321       /*
1322        * Limit the number of packets pushed into the graph
1323        */
1324       if (vectors >= fq->vector_threshold)
1325         {
1326           fq->head_hint = fq->head;
1327           return processed;
1328         }
1329     }
1330   ASSERT (0);
1331   return processed;
1332 }
1333
1334 static_always_inline void
1335 vlib_worker_thread_internal (vlib_main_t * vm)
1336 {
1337   vlib_node_main_t *nm = &vm->node_main;
1338   u64 cpu_time_now = clib_cpu_time_now ();
1339
1340   vec_alloc (nm->pending_interrupt_node_runtime_indices, 32);
1341
1342   while (1)
1343     {
1344       vlib_worker_thread_barrier_check ();
1345
1346       vlib_frame_queue_dequeue_internal (vm);
1347
1348       vlib_node_runtime_t *n;
1349       vec_foreach (n, nm->nodes_by_type[VLIB_NODE_TYPE_INPUT])
1350       {
1351         cpu_time_now = dispatch_node (vm, n, VLIB_NODE_TYPE_INPUT,
1352                                       VLIB_NODE_STATE_POLLING, /* frame */ 0,
1353                                       cpu_time_now);
1354       }
1355
1356       /* Next handle interrupts. */
1357       {
1358         uword l = _vec_len (nm->pending_interrupt_node_runtime_indices);
1359         uword i;
1360         if (l > 0)
1361           {
1362             _vec_len (nm->pending_interrupt_node_runtime_indices) = 0;
1363             for (i = 0; i < l; i++)
1364               {
1365                 n = vec_elt_at_index (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT],
1366                                       nm->
1367                                       pending_interrupt_node_runtime_indices
1368                                       [i]);
1369                 cpu_time_now =
1370                   dispatch_node (vm, n, VLIB_NODE_TYPE_INPUT,
1371                                  VLIB_NODE_STATE_INTERRUPT,
1372                                  /* frame */ 0,
1373                                  cpu_time_now);
1374               }
1375           }
1376       }
1377
1378       if (_vec_len (nm->pending_frames))
1379         {
1380           int i;
1381           cpu_time_now = clib_cpu_time_now ();
1382           for (i = 0; i < _vec_len (nm->pending_frames); i++)
1383             {
1384               vlib_pending_frame_t *p;
1385
1386               p = nm->pending_frames + i;
1387
1388               cpu_time_now = dispatch_pending_node (vm, p, cpu_time_now);
1389             }
1390           _vec_len (nm->pending_frames) = 0;
1391         }
1392       vlib_increment_main_loop_counter (vm);
1393
1394       /* Record time stamp in case there are no enabled nodes and above
1395          calls do not update time stamp. */
1396       cpu_time_now = clib_cpu_time_now ();
1397     }
1398 }
1399
1400 void
1401 vlib_worker_thread_fn (void *arg)
1402 {
1403   vlib_worker_thread_t *w = (vlib_worker_thread_t *) arg;
1404   vlib_thread_main_t *tm = vlib_get_thread_main ();
1405   vlib_main_t *vm = vlib_get_main ();
1406
1407   ASSERT (vm->cpu_index == os_get_cpu_number ());
1408
1409   vlib_worker_thread_init (w);
1410   clib_time_init (&vm->clib_time);
1411   clib_mem_set_heap (w->thread_mheap);
1412
1413   /* Wait until the dpdk init sequence is complete */
1414   while (tm->worker_thread_release == 0)
1415     vlib_worker_thread_barrier_check ();
1416
1417   vlib_worker_thread_internal (vm);
1418 }
1419
1420 /* *INDENT-OFF* */
1421 VLIB_REGISTER_THREAD (worker_thread_reg, static) = {
1422   .name = "workers",
1423   .short_name = "wk",
1424   .function = vlib_worker_thread_fn,
1425 };
1426 /* *INDENT-ON* */
1427
1428 clib_error_t *
1429 threads_init (vlib_main_t * vm)
1430 {
1431   vlib_thread_main_t *tm = vlib_get_thread_main ();
1432
1433   tm->handoff_dispatch_node_index = ~0;
1434
1435   return 0;
1436 }
1437
1438 VLIB_INIT_FUNCTION (threads_init);
1439
1440 /*
1441  * fd.io coding-style-patch-verification: ON
1442  *
1443  * Local Variables:
1444  * eval: (c-set-style "gnu")
1445  * End:
1446  */