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