VPP-1102: fix dangling references in RPC handling
[vpp.git] / src / 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 DECLARE_CJ_GLOBAL_LOG;
26
27 #define FRAME_QUEUE_NELTS 32
28
29 u32
30 vl (void *p)
31 {
32   return vec_len (p);
33 }
34
35 vlib_worker_thread_t *vlib_worker_threads;
36 vlib_thread_main_t vlib_thread_main;
37
38 /*
39  * Barrier tracing can be enabled on a normal build to collect information
40  * on barrier use, including timings and call stacks.  Deliberately not
41  * keyed off CLIB_DEBUG, because that can add significant overhead which
42  * imapacts observed timings.
43  */
44
45 #ifdef BARRIER_TRACING
46  /*
47   * Output of barrier tracing can be to syslog or elog as suits
48   */
49 #ifdef BARRIER_TRACING_ELOG
50 static u32
51 elog_id_for_msg_name (const char *msg_name)
52 {
53   uword *p, r;
54   static uword *h;
55   u8 *name_copy;
56
57   if (!h)
58     h = hash_create_string (0, sizeof (uword));
59
60   p = hash_get_mem (h, msg_name);
61   if (p)
62     return p[0];
63   r = elog_string (&vlib_global_main.elog_main, "%s", msg_name);
64
65   name_copy = format (0, "%s%c", msg_name, 0);
66
67   hash_set_mem (h, name_copy, r);
68
69   return r;
70 }
71
72   /*
73    * elog Barrier trace functions, which are nulled out if BARRIER_TRACING isn't
74    * defined
75    */
76
77 static inline void
78 barrier_trace_sync (f64 t_entry, f64 t_open, f64 t_closed)
79 {
80     /* *INDENT-OFF* */
81     ELOG_TYPE_DECLARE (e) =
82       {
83         .format = "barrier <%d#%s(O:%dus:%dus)(%dus)",
84         .format_args = "i4T4i4i4i4",
85       };
86     /* *INDENT-ON* */
87   struct
88   {
89     u32 count, caller, t_entry, t_open, t_closed;
90   } *ed = 0;
91
92   ed = ELOG_DATA (&vlib_global_main.elog_main, e);
93   ed->count = (int) vlib_worker_threads[0].barrier_sync_count;
94   ed->caller = elog_id_for_msg_name (vlib_worker_threads[0].barrier_caller);
95   ed->t_entry = (int) (1000000.0 * t_entry);
96   ed->t_open = (int) (1000000.0 * t_open);
97   ed->t_closed = (int) (1000000.0 * t_closed);
98 }
99
100 static inline void
101 barrier_trace_sync_rec (f64 t_entry)
102 {
103     /* *INDENT-OFF* */
104     ELOG_TYPE_DECLARE (e) =
105       {
106         .format = "barrier    <%d(%dus)%s",
107         .format_args = "i4i4T4",
108       };
109     /* *INDENT-ON* */
110   struct
111   {
112     u32 depth, t_entry, caller;
113   } *ed = 0;
114
115   ed = ELOG_DATA (&vlib_global_main.elog_main, e);
116   ed->depth = (int) vlib_worker_threads[0].recursion_level - 1;
117   ed->t_entry = (int) (1000000.0 * t_entry);
118   ed->caller = elog_id_for_msg_name (vlib_worker_threads[0].barrier_caller);
119 }
120
121 static inline void
122 barrier_trace_release_rec (f64 t_entry)
123 {
124     /* *INDENT-OFF* */
125     ELOG_TYPE_DECLARE (e) =
126       {
127         .format = "barrier      (%dus)%d>",
128         .format_args = "i4i4",
129       };
130     /* *INDENT-ON* */
131   struct
132   {
133     u32 t_entry, depth;
134   } *ed = 0;
135
136   ed = ELOG_DATA (&vlib_global_main.elog_main, e);
137   ed->t_entry = (int) (1000000.0 * t_entry);
138   ed->depth = (int) vlib_worker_threads[0].recursion_level;
139 }
140
141 static inline void
142 barrier_trace_release (f64 t_entry, f64 t_closed_total, f64 t_update_main)
143 {
144     /* *INDENT-OFF* */
145     ELOG_TYPE_DECLARE (e) =
146       {
147         .format = "barrier   (%dus){%d}(C:%dus)#%d>",
148         .format_args = "i4i4i4i4",
149       };
150     /* *INDENT-ON* */
151   struct
152   {
153     u32 t_entry, t_update_main, t_closed_total, count;
154   } *ed = 0;
155
156   ed = ELOG_DATA (&vlib_global_main.elog_main, e);
157   ed->t_entry = (int) (1000000.0 * t_entry);
158   ed->t_update_main = (int) (1000000.0 * t_update_main);
159   ed->t_closed_total = (int) (1000000.0 * t_closed_total);
160   ed->count = (int) vlib_worker_threads[0].barrier_sync_count;
161
162   /* Reset context for next trace */
163   vlib_worker_threads[0].barrier_context = NULL;
164 }
165 #else
166 char barrier_trace[65536];
167 char *btp = barrier_trace;
168
169   /*
170    * syslog Barrier trace functions, which are nulled out if BARRIER_TRACING
171    * isn't defined
172    */
173
174
175 static inline void
176 barrier_trace_sync (f64 t_entry, f64 t_open, f64 t_closed)
177 {
178   btp += sprintf (btp, "<%u#%s",
179                   (unsigned int) vlib_worker_threads[0].barrier_sync_count,
180                   vlib_worker_threads[0].barrier_caller);
181
182   if (vlib_worker_threads[0].barrier_context)
183     {
184       btp += sprintf (btp, "[%s]", vlib_worker_threads[0].barrier_context);
185
186     }
187
188   btp += sprintf (btp, "(O:%dus:%dus)(%dus):",
189                   (int) (1000000.0 * t_entry),
190                   (int) (1000000.0 * t_open), (int) (1000000.0 * t_closed));
191
192 }
193
194 static inline void
195 barrier_trace_sync_rec (f64 t_entry)
196 {
197   btp += sprintf (btp, "<%u(%dus)%s:",
198                   (int) vlib_worker_threads[0].recursion_level - 1,
199                   (int) (1000000.0 * t_entry),
200                   vlib_worker_threads[0].barrier_caller);
201 }
202
203 static inline void
204 barrier_trace_release_rec (f64 t_entry)
205 {
206   btp += sprintf (btp, ":(%dus)%u>", (int) (1000000.0 * t_entry),
207                   (int) vlib_worker_threads[0].recursion_level);
208 }
209
210 static inline void
211 barrier_trace_release (f64 t_entry, f64 t_closed_total, f64 t_update_main)
212 {
213
214   btp += sprintf (btp, ":(%dus)", (int) (1000000.0 * t_entry));
215   if (t_update_main > 0)
216     {
217       btp += sprintf (btp, "{%dus}", (int) (1000000.0 * t_update_main));
218     }
219
220   btp += sprintf (btp, "(C:%dus)#%u>",
221                   (int) (1000000.0 * t_closed_total),
222                   (int) vlib_worker_threads[0].barrier_sync_count);
223
224   /* Dump buffer to syslog, and reset for next trace */
225   fformat (stderr, "BTRC %s\n", barrier_trace);
226   btp = barrier_trace;
227   vlib_worker_threads[0].barrier_context = NULL;
228 }
229 #endif
230 #else
231
232   /* Null functions for default case where barrier tracing isn't used */
233 static inline void
234 barrier_trace_sync (f64 t_entry, f64 t_open, f64 t_closed)
235 {
236 }
237
238 static inline void
239 barrier_trace_sync_rec (f64 t_entry)
240 {
241 }
242
243 static inline void
244 barrier_trace_release_rec (f64 t_entry)
245 {
246 }
247
248 static inline void
249 barrier_trace_release (f64 t_entry, f64 t_closed_total, f64 t_update_main)
250 {
251 }
252 #endif
253
254 uword
255 os_get_nthreads (void)
256 {
257   u32 len;
258
259   len = vec_len (vlib_thread_stacks);
260   if (len == 0)
261     return 1;
262   else
263     return len;
264 }
265
266 void
267 vlib_set_thread_name (char *name)
268 {
269   int pthread_setname_np (pthread_t __target_thread, const char *__name);
270   int rv;
271   pthread_t thread = pthread_self ();
272
273   if (thread)
274     {
275       rv = pthread_setname_np (thread, name);
276       if (rv)
277         clib_warning ("pthread_setname_np returned %d", rv);
278     }
279 }
280
281 static int
282 sort_registrations_by_no_clone (void *a0, void *a1)
283 {
284   vlib_thread_registration_t **tr0 = a0;
285   vlib_thread_registration_t **tr1 = a1;
286
287   return ((i32) ((*tr0)->no_data_structure_clone)
288           - ((i32) ((*tr1)->no_data_structure_clone)));
289 }
290
291 static uword *
292 clib_sysfs_list_to_bitmap (char *filename)
293 {
294   FILE *fp;
295   uword *r = 0;
296
297   fp = fopen (filename, "r");
298
299   if (fp != NULL)
300     {
301       u8 *buffer = 0;
302       vec_validate (buffer, 256 - 1);
303       if (fgets ((char *) buffer, 256, fp))
304         {
305           unformat_input_t in;
306           unformat_init_string (&in, (char *) buffer,
307                                 strlen ((char *) buffer));
308           if (unformat (&in, "%U", unformat_bitmap_list, &r) != 1)
309             clib_warning ("unformat_bitmap_list failed");
310           unformat_free (&in);
311         }
312       vec_free (buffer);
313       fclose (fp);
314     }
315   return r;
316 }
317
318
319 /* Called early in the init sequence */
320
321 clib_error_t *
322 vlib_thread_init (vlib_main_t * vm)
323 {
324   vlib_thread_main_t *tm = &vlib_thread_main;
325   vlib_worker_thread_t *w;
326   vlib_thread_registration_t *tr;
327   u32 n_vlib_mains = 1;
328   u32 first_index = 1;
329   u32 i;
330   uword *avail_cpu;
331
332   /* get bitmaps of active cpu cores and sockets */
333   tm->cpu_core_bitmap =
334     clib_sysfs_list_to_bitmap ("/sys/devices/system/cpu/online");
335   tm->cpu_socket_bitmap =
336     clib_sysfs_list_to_bitmap ("/sys/devices/system/node/online");
337
338   avail_cpu = clib_bitmap_dup (tm->cpu_core_bitmap);
339
340   /* skip cores */
341   for (i = 0; i < tm->skip_cores; i++)
342     {
343       uword c = clib_bitmap_first_set (avail_cpu);
344       if (c == ~0)
345         return clib_error_return (0, "no available cpus to skip");
346
347       avail_cpu = clib_bitmap_set (avail_cpu, c, 0);
348     }
349
350   /* grab cpu for main thread */
351   if (!tm->main_lcore)
352     {
353       tm->main_lcore = clib_bitmap_first_set (avail_cpu);
354       if (tm->main_lcore == (u8) ~ 0)
355         return clib_error_return (0, "no available cpus to be used for the"
356                                   " main thread");
357     }
358   else
359     {
360       if (clib_bitmap_get (avail_cpu, tm->main_lcore) == 0)
361         return clib_error_return (0, "cpu %u is not available to be used"
362                                   " for the main thread", tm->main_lcore);
363     }
364   avail_cpu = clib_bitmap_set (avail_cpu, tm->main_lcore, 0);
365
366   /* assume that there is socket 0 only if there is no data from sysfs */
367   if (!tm->cpu_socket_bitmap)
368     tm->cpu_socket_bitmap = clib_bitmap_set (0, 0, 1);
369
370   /* pin main thread to main_lcore  */
371   if (tm->cb.vlib_thread_set_lcore_cb)
372     {
373       tm->cb.vlib_thread_set_lcore_cb (0, tm->main_lcore);
374     }
375   else
376     {
377       cpu_set_t cpuset;
378       CPU_ZERO (&cpuset);
379       CPU_SET (tm->main_lcore, &cpuset);
380       pthread_setaffinity_np (pthread_self (), sizeof (cpu_set_t), &cpuset);
381     }
382
383   /* as many threads as stacks... */
384   vec_validate_aligned (vlib_worker_threads, vec_len (vlib_thread_stacks) - 1,
385                         CLIB_CACHE_LINE_BYTES);
386
387   /* Preallocate thread 0 */
388   _vec_len (vlib_worker_threads) = 1;
389   w = vlib_worker_threads;
390   w->thread_mheap = clib_mem_get_heap ();
391   w->thread_stack = vlib_thread_stacks[0];
392   w->lcore_id = tm->main_lcore;
393   w->lwp = syscall (SYS_gettid);
394   w->thread_id = pthread_self ();
395   tm->n_vlib_mains = 1;
396
397   if (tm->sched_policy != ~0)
398     {
399       struct sched_param sched_param;
400       if (!sched_getparam (w->lwp, &sched_param))
401         {
402           if (tm->sched_priority != ~0)
403             sched_param.sched_priority = tm->sched_priority;
404           sched_setscheduler (w->lwp, tm->sched_policy, &sched_param);
405         }
406     }
407
408   /* assign threads to cores and set n_vlib_mains */
409   tr = tm->next;
410
411   while (tr)
412     {
413       vec_add1 (tm->registrations, tr);
414       tr = tr->next;
415     }
416
417   vec_sort_with_function (tm->registrations, sort_registrations_by_no_clone);
418
419   for (i = 0; i < vec_len (tm->registrations); i++)
420     {
421       int j;
422       tr = tm->registrations[i];
423       tr->first_index = first_index;
424       first_index += tr->count;
425       n_vlib_mains += (tr->no_data_structure_clone == 0) ? tr->count : 0;
426
427       /* construct coremask */
428       if (tr->use_pthreads || !tr->count)
429         continue;
430
431       if (tr->coremask)
432         {
433           uword c;
434           /* *INDENT-OFF* */
435           clib_bitmap_foreach (c, tr->coremask, ({
436             if (clib_bitmap_get(avail_cpu, c) == 0)
437               return clib_error_return (0, "cpu %u is not available to be used"
438                                         " for the '%s' thread",c, tr->name);
439
440             avail_cpu = clib_bitmap_set(avail_cpu, c, 0);
441           }));
442 /* *INDENT-ON* */
443
444         }
445       else
446         {
447           for (j = 0; j < tr->count; j++)
448             {
449               uword c = clib_bitmap_first_set (avail_cpu);
450               if (c == ~0)
451                 return clib_error_return (0,
452                                           "no available cpus to be used for"
453                                           " the '%s' thread", tr->name);
454
455               avail_cpu = clib_bitmap_set (avail_cpu, c, 0);
456               tr->coremask = clib_bitmap_set (tr->coremask, c, 1);
457             }
458         }
459     }
460
461   clib_bitmap_free (avail_cpu);
462
463   tm->n_vlib_mains = n_vlib_mains;
464
465   vec_validate_aligned (vlib_worker_threads, first_index - 1,
466                         CLIB_CACHE_LINE_BYTES);
467
468   return 0;
469 }
470
471 vlib_frame_queue_t *
472 vlib_frame_queue_alloc (int nelts)
473 {
474   vlib_frame_queue_t *fq;
475
476   fq = clib_mem_alloc_aligned (sizeof (*fq), CLIB_CACHE_LINE_BYTES);
477   memset (fq, 0, sizeof (*fq));
478   fq->nelts = nelts;
479   fq->vector_threshold = 128;   // packets
480   vec_validate_aligned (fq->elts, nelts - 1, CLIB_CACHE_LINE_BYTES);
481
482   if (1)
483     {
484       if (((uword) & fq->tail) & (CLIB_CACHE_LINE_BYTES - 1))
485         fformat (stderr, "WARNING: fq->tail unaligned\n");
486       if (((uword) & fq->head) & (CLIB_CACHE_LINE_BYTES - 1))
487         fformat (stderr, "WARNING: fq->head unaligned\n");
488       if (((uword) fq->elts) & (CLIB_CACHE_LINE_BYTES - 1))
489         fformat (stderr, "WARNING: fq->elts unaligned\n");
490
491       if (sizeof (fq->elts[0]) % CLIB_CACHE_LINE_BYTES)
492         fformat (stderr, "WARNING: fq->elts[0] size %d\n",
493                  sizeof (fq->elts[0]));
494       if (nelts & (nelts - 1))
495         {
496           fformat (stderr, "FATAL: nelts MUST be a power of 2\n");
497           abort ();
498         }
499     }
500
501   return (fq);
502 }
503
504 void vl_msg_api_handler_no_free (void *) __attribute__ ((weak));
505 void
506 vl_msg_api_handler_no_free (void *v)
507 {
508 }
509
510 /* Turned off, save as reference material... */
511 #if 0
512 static inline int
513 vlib_frame_queue_dequeue_internal (int thread_id,
514                                    vlib_main_t * vm, vlib_node_main_t * nm)
515 {
516   vlib_frame_queue_t *fq = vlib_frame_queues[thread_id];
517   vlib_frame_queue_elt_t *elt;
518   vlib_frame_t *f;
519   vlib_pending_frame_t *p;
520   vlib_node_runtime_t *r;
521   u32 node_runtime_index;
522   int msg_type;
523   u64 before;
524   int processed = 0;
525
526   ASSERT (vm == vlib_mains[thread_id]);
527
528   while (1)
529     {
530       if (fq->head == fq->tail)
531         return processed;
532
533       elt = fq->elts + ((fq->head + 1) & (fq->nelts - 1));
534
535       if (!elt->valid)
536         return processed;
537
538       before = clib_cpu_time_now ();
539
540       f = elt->frame;
541       node_runtime_index = elt->node_runtime_index;
542       msg_type = elt->msg_type;
543
544       switch (msg_type)
545         {
546         case VLIB_FRAME_QUEUE_ELT_FREE_BUFFERS:
547           vlib_buffer_free (vm, vlib_frame_vector_args (f), f->n_vectors);
548           /* note fallthrough... */
549         case VLIB_FRAME_QUEUE_ELT_FREE_FRAME:
550           r = vec_elt_at_index (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL],
551                                 node_runtime_index);
552           vlib_frame_free (vm, r, f);
553           break;
554         case VLIB_FRAME_QUEUE_ELT_DISPATCH_FRAME:
555           vec_add2 (vm->node_main.pending_frames, p, 1);
556           f->flags |= (VLIB_FRAME_PENDING | VLIB_FRAME_FREE_AFTER_DISPATCH);
557           p->node_runtime_index = elt->node_runtime_index;
558           p->frame_index = vlib_frame_index (vm, f);
559           p->next_frame_index = VLIB_PENDING_FRAME_NO_NEXT_FRAME;
560           fq->dequeue_vectors += (u64) f->n_vectors;
561           break;
562         case VLIB_FRAME_QUEUE_ELT_API_MSG:
563           vl_msg_api_handler_no_free (f);
564           break;
565         default:
566           clib_warning ("bogus frame queue message, type %d", msg_type);
567           break;
568         }
569       elt->valid = 0;
570       fq->dequeues++;
571       fq->dequeue_ticks += clib_cpu_time_now () - before;
572       CLIB_MEMORY_BARRIER ();
573       fq->head++;
574       processed++;
575     }
576   ASSERT (0);
577   return processed;
578 }
579
580 int
581 vlib_frame_queue_dequeue (int thread_id,
582                           vlib_main_t * vm, vlib_node_main_t * nm)
583 {
584   return vlib_frame_queue_dequeue_internal (thread_id, vm, nm);
585 }
586
587 int
588 vlib_frame_queue_enqueue (vlib_main_t * vm, u32 node_runtime_index,
589                           u32 frame_queue_index, vlib_frame_t * frame,
590                           vlib_frame_queue_msg_type_t type)
591 {
592   vlib_frame_queue_t *fq = vlib_frame_queues[frame_queue_index];
593   vlib_frame_queue_elt_t *elt;
594   u32 save_count;
595   u64 new_tail;
596   u64 before = clib_cpu_time_now ();
597
598   ASSERT (fq);
599
600   new_tail = __sync_add_and_fetch (&fq->tail, 1);
601
602   /* Wait until a ring slot is available */
603   while (new_tail >= fq->head + fq->nelts)
604     {
605       f64 b4 = vlib_time_now_ticks (vm, before);
606       vlib_worker_thread_barrier_check (vm, b4);
607       /* Bad idea. Dequeue -> enqueue -> dequeue -> trouble */
608       // vlib_frame_queue_dequeue (vm->thread_index, vm, nm);
609     }
610
611   elt = fq->elts + (new_tail & (fq->nelts - 1));
612
613   /* this would be very bad... */
614   while (elt->valid)
615     {
616     }
617
618   /* Once we enqueue the frame, frame->n_vectors is owned elsewhere... */
619   save_count = frame->n_vectors;
620
621   elt->frame = frame;
622   elt->node_runtime_index = node_runtime_index;
623   elt->msg_type = type;
624   CLIB_MEMORY_BARRIER ();
625   elt->valid = 1;
626
627   return save_count;
628 }
629 #endif /* 0 */
630
631 /* To be called by vlib worker threads upon startup */
632 void
633 vlib_worker_thread_init (vlib_worker_thread_t * w)
634 {
635   vlib_thread_main_t *tm = vlib_get_thread_main ();
636
637   /*
638    * Note: disabling signals in worker threads as follows
639    * prevents the api post-mortem dump scheme from working
640    * {
641    *    sigset_t s;
642    *    sigfillset (&s);
643    *    pthread_sigmask (SIG_SETMASK, &s, 0);
644    *  }
645    */
646
647   clib_mem_set_heap (w->thread_mheap);
648
649   if (vec_len (tm->thread_prefix) && w->registration->short_name)
650     {
651       w->name = format (0, "%v_%s_%d%c", tm->thread_prefix,
652                         w->registration->short_name, w->instance_id, '\0');
653       vlib_set_thread_name ((char *) w->name);
654     }
655
656   if (!w->registration->use_pthreads)
657     {
658
659       /* Initial barrier sync, for both worker and i/o threads */
660       clib_smp_atomic_add (vlib_worker_threads->workers_at_barrier, 1);
661
662       while (*vlib_worker_threads->wait_at_barrier)
663         ;
664
665       clib_smp_atomic_add (vlib_worker_threads->workers_at_barrier, -1);
666     }
667 }
668
669 void *
670 vlib_worker_thread_bootstrap_fn (void *arg)
671 {
672   void *rv;
673   vlib_worker_thread_t *w = arg;
674
675   w->lwp = syscall (SYS_gettid);
676   w->thread_id = pthread_self ();
677
678   __os_thread_index = w - vlib_worker_threads;
679
680   rv = (void *) clib_calljmp
681     ((uword (*)(uword)) w->thread_function,
682      (uword) arg, w->thread_stack + VLIB_THREAD_STACK_SIZE);
683   /* NOTREACHED, we hope */
684   return rv;
685 }
686
687 static clib_error_t *
688 vlib_launch_thread_int (void *fp, vlib_worker_thread_t * w, unsigned lcore_id)
689 {
690   vlib_thread_main_t *tm = &vlib_thread_main;
691   void *(*fp_arg) (void *) = fp;
692
693   w->lcore_id = lcore_id;
694   if (tm->cb.vlib_launch_thread_cb && !w->registration->use_pthreads)
695     return tm->cb.vlib_launch_thread_cb (fp, (void *) w, lcore_id);
696   else
697     {
698       pthread_t worker;
699       cpu_set_t cpuset;
700       CPU_ZERO (&cpuset);
701       CPU_SET (lcore_id, &cpuset);
702
703       if (pthread_create (&worker, NULL /* attr */ , fp_arg, (void *) w))
704         return clib_error_return_unix (0, "pthread_create");
705
706       if (pthread_setaffinity_np (worker, sizeof (cpu_set_t), &cpuset))
707         return clib_error_return_unix (0, "pthread_setaffinity_np");
708
709       return 0;
710     }
711 }
712
713 static clib_error_t *
714 start_workers (vlib_main_t * vm)
715 {
716   int i, j;
717   vlib_worker_thread_t *w;
718   vlib_main_t *vm_clone;
719   void *oldheap;
720   vlib_thread_main_t *tm = &vlib_thread_main;
721   vlib_thread_registration_t *tr;
722   vlib_node_runtime_t *rt;
723   u32 n_vlib_mains = tm->n_vlib_mains;
724   u32 worker_thread_index;
725   u8 *main_heap = clib_mem_get_per_cpu_heap ();
726   mheap_t *main_heap_header = mheap_header (main_heap);
727
728   vec_reset_length (vlib_worker_threads);
729
730   /* Set up the main thread */
731   vec_add2_aligned (vlib_worker_threads, w, 1, CLIB_CACHE_LINE_BYTES);
732   w->elog_track.name = "main thread";
733   elog_track_register (&vm->elog_main, &w->elog_track);
734
735   if (vec_len (tm->thread_prefix))
736     {
737       w->name = format (0, "%v_main%c", tm->thread_prefix, '\0');
738       vlib_set_thread_name ((char *) w->name);
739     }
740
741   /*
742    * Truth of the matter: we always use at least two
743    * threads. So, make the main heap thread-safe
744    * and make the event log thread-safe.
745    */
746   main_heap_header->flags |= MHEAP_FLAG_THREAD_SAFE;
747   vm->elog_main.lock =
748     clib_mem_alloc_aligned (CLIB_CACHE_LINE_BYTES, CLIB_CACHE_LINE_BYTES);
749   vm->elog_main.lock[0] = 0;
750
751   if (n_vlib_mains > 1)
752     {
753       /* Replace hand-crafted length-1 vector with a real vector */
754       vlib_mains = 0;
755
756       vec_validate_aligned (vlib_mains, tm->n_vlib_mains - 1,
757                             CLIB_CACHE_LINE_BYTES);
758       _vec_len (vlib_mains) = 0;
759       vec_add1_aligned (vlib_mains, vm, CLIB_CACHE_LINE_BYTES);
760
761       vlib_worker_threads->wait_at_barrier =
762         clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES);
763       vlib_worker_threads->workers_at_barrier =
764         clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES);
765
766       vlib_worker_threads->node_reforks_required =
767         clib_mem_alloc_aligned (sizeof (u32), CLIB_CACHE_LINE_BYTES);
768
769       /* Ask for an initial barrier sync */
770       *vlib_worker_threads->workers_at_barrier = 0;
771       *vlib_worker_threads->wait_at_barrier = 1;
772
773       /* Without update or refork */
774       *vlib_worker_threads->node_reforks_required = 0;
775       vm->need_vlib_worker_thread_node_runtime_update = 0;
776
777       /* init timing */
778       vm->barrier_epoch = 0;
779       vm->barrier_no_close_before = 0;
780
781       worker_thread_index = 1;
782
783       for (i = 0; i < vec_len (tm->registrations); i++)
784         {
785           vlib_node_main_t *nm, *nm_clone;
786           vlib_buffer_main_t *bm_clone;
787           vlib_buffer_free_list_t *fl_clone, *fl_orig;
788           vlib_buffer_free_list_t *orig_freelist_pool;
789           int k;
790
791           tr = tm->registrations[i];
792
793           if (tr->count == 0)
794             continue;
795
796           for (k = 0; k < tr->count; k++)
797             {
798               vlib_node_t *n;
799
800               vec_add2 (vlib_worker_threads, w, 1);
801               if (tr->mheap_size)
802                 w->thread_mheap =
803                   mheap_alloc (0 /* use VM */ , tr->mheap_size);
804               else
805                 w->thread_mheap = main_heap;
806
807               w->thread_stack =
808                 vlib_thread_stack_init (w - vlib_worker_threads);
809               w->thread_function = tr->function;
810               w->thread_function_arg = w;
811               w->instance_id = k;
812               w->registration = tr;
813
814               w->elog_track.name =
815                 (char *) format (0, "%s %d", tr->name, k + 1);
816               vec_add1 (w->elog_track.name, 0);
817               elog_track_register (&vm->elog_main, &w->elog_track);
818
819               if (tr->no_data_structure_clone)
820                 continue;
821
822               /* Fork vlib_global_main et al. Look for bugs here */
823               oldheap = clib_mem_set_heap (w->thread_mheap);
824
825               vm_clone = clib_mem_alloc (sizeof (*vm_clone));
826               clib_memcpy (vm_clone, vlib_mains[0], sizeof (*vm_clone));
827
828               vm_clone->thread_index = worker_thread_index;
829               vm_clone->heap_base = w->thread_mheap;
830               vm_clone->mbuf_alloc_list = 0;
831               vm_clone->init_functions_called =
832                 hash_create (0, /* value bytes */ 0);
833               vm_clone->pending_rpc_requests = 0;
834               vec_validate (vm_clone->pending_rpc_requests, 0);
835               _vec_len (vm_clone->pending_rpc_requests) = 0;
836               memset (&vm_clone->random_buffer, 0,
837                       sizeof (vm_clone->random_buffer));
838
839               nm = &vlib_mains[0]->node_main;
840               nm_clone = &vm_clone->node_main;
841               /* fork next frames array, preserving node runtime indices */
842               nm_clone->next_frames = vec_dup (nm->next_frames);
843               for (j = 0; j < vec_len (nm_clone->next_frames); j++)
844                 {
845                   vlib_next_frame_t *nf = &nm_clone->next_frames[j];
846                   u32 save_node_runtime_index;
847                   u32 save_flags;
848
849                   save_node_runtime_index = nf->node_runtime_index;
850                   save_flags = nf->flags & VLIB_FRAME_NO_FREE_AFTER_DISPATCH;
851                   vlib_next_frame_init (nf);
852                   nf->node_runtime_index = save_node_runtime_index;
853                   nf->flags = save_flags;
854                 }
855
856               /* fork the frame dispatch queue */
857               nm_clone->pending_frames = 0;
858               vec_validate (nm_clone->pending_frames, 10);      /* $$$$$?????? */
859               _vec_len (nm_clone->pending_frames) = 0;
860
861               /* fork nodes */
862               nm_clone->nodes = 0;
863
864               /* Allocate all nodes in single block for speed */
865               n = clib_mem_alloc_no_fail (vec_len (nm->nodes) * sizeof (*n));
866
867               for (j = 0; j < vec_len (nm->nodes); j++)
868                 {
869                   clib_memcpy (n, nm->nodes[j], sizeof (*n));
870                   /* none of the copied nodes have enqueue rights given out */
871                   n->owner_node_index = VLIB_INVALID_NODE_INDEX;
872                   memset (&n->stats_total, 0, sizeof (n->stats_total));
873                   memset (&n->stats_last_clear, 0,
874                           sizeof (n->stats_last_clear));
875                   vec_add1 (nm_clone->nodes, n);
876                   n++;
877                 }
878               nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL] =
879                 vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]);
880               vec_foreach (rt,
881                            nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL])
882               {
883                 vlib_node_t *n = vlib_get_node (vm, rt->node_index);
884                 rt->thread_index = vm_clone->thread_index;
885                 /* copy initial runtime_data from node */
886                 if (n->runtime_data && n->runtime_data_bytes > 0)
887                   clib_memcpy (rt->runtime_data, n->runtime_data,
888                                clib_min (VLIB_NODE_RUNTIME_DATA_SIZE,
889                                          n->runtime_data_bytes));
890               }
891
892               nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT] =
893                 vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]);
894               vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
895               {
896                 vlib_node_t *n = vlib_get_node (vm, rt->node_index);
897                 rt->thread_index = vm_clone->thread_index;
898                 /* copy initial runtime_data from node */
899                 if (n->runtime_data && n->runtime_data_bytes > 0)
900                   clib_memcpy (rt->runtime_data, n->runtime_data,
901                                clib_min (VLIB_NODE_RUNTIME_DATA_SIZE,
902                                          n->runtime_data_bytes));
903               }
904
905               nm_clone->processes = vec_dup (nm->processes);
906
907               /* zap the (per worker) frame freelists, etc */
908               nm_clone->frame_sizes = 0;
909               nm_clone->frame_size_hash = hash_create (0, sizeof (uword));
910
911               /* Packet trace buffers are guaranteed to be empty, nothing to do here */
912
913               clib_mem_set_heap (oldheap);
914               vec_add1_aligned (vlib_mains, vm_clone, CLIB_CACHE_LINE_BYTES);
915
916               vm_clone->error_main.counters =
917                 vec_dup (vlib_mains[0]->error_main.counters);
918               vm_clone->error_main.counters_last_clear =
919                 vec_dup (vlib_mains[0]->error_main.counters_last_clear);
920
921               /* Fork the vlib_buffer_main_t free lists, etc. */
922               bm_clone = vec_dup (vm_clone->buffer_main);
923               vm_clone->buffer_main = bm_clone;
924
925               orig_freelist_pool = bm_clone->buffer_free_list_pool;
926               bm_clone->buffer_free_list_pool = 0;
927
928             /* *INDENT-OFF* */
929             pool_foreach (fl_orig, orig_freelist_pool,
930                           ({
931                             pool_get_aligned (bm_clone->buffer_free_list_pool,
932                                               fl_clone, CLIB_CACHE_LINE_BYTES);
933                             ASSERT (fl_orig - orig_freelist_pool
934                                     == fl_clone - bm_clone->buffer_free_list_pool);
935
936                             fl_clone[0] = fl_orig[0];
937                             fl_clone->buffers = 0;
938                             fl_clone->n_alloc = 0;
939                           }));
940 /* *INDENT-ON* */
941
942               worker_thread_index++;
943             }
944         }
945     }
946   else
947     {
948       /* only have non-data-structure copy threads to create... */
949       for (i = 0; i < vec_len (tm->registrations); i++)
950         {
951           tr = tm->registrations[i];
952
953           for (j = 0; j < tr->count; j++)
954             {
955               vec_add2 (vlib_worker_threads, w, 1);
956               if (tr->mheap_size)
957                 w->thread_mheap =
958                   mheap_alloc (0 /* use VM */ , tr->mheap_size);
959               else
960                 w->thread_mheap = main_heap;
961               w->thread_stack =
962                 vlib_thread_stack_init (w - vlib_worker_threads);
963               w->thread_function = tr->function;
964               w->thread_function_arg = w;
965               w->instance_id = j;
966               w->elog_track.name =
967                 (char *) format (0, "%s %d", tr->name, j + 1);
968               w->registration = tr;
969               vec_add1 (w->elog_track.name, 0);
970               elog_track_register (&vm->elog_main, &w->elog_track);
971             }
972         }
973     }
974
975   worker_thread_index = 1;
976
977   for (i = 0; i < vec_len (tm->registrations); i++)
978     {
979       clib_error_t *err;
980       int j;
981
982       tr = tm->registrations[i];
983
984       if (tr->use_pthreads || tm->use_pthreads)
985         {
986           for (j = 0; j < tr->count; j++)
987             {
988               w = vlib_worker_threads + worker_thread_index++;
989               err = vlib_launch_thread_int (vlib_worker_thread_bootstrap_fn,
990                                             w, 0);
991               if (err)
992                 clib_error_report (err);
993             }
994         }
995       else
996         {
997           uword c;
998           /* *INDENT-OFF* */
999           clib_bitmap_foreach (c, tr->coremask, ({
1000             w = vlib_worker_threads + worker_thread_index++;
1001             err = vlib_launch_thread_int (vlib_worker_thread_bootstrap_fn,
1002                                           w, c);
1003             if (err)
1004               clib_error_report (err);
1005           }));
1006           /* *INDENT-ON* */
1007         }
1008     }
1009   vlib_worker_thread_barrier_sync (vm);
1010   vlib_worker_thread_barrier_release (vm);
1011   return 0;
1012 }
1013
1014 VLIB_MAIN_LOOP_ENTER_FUNCTION (start_workers);
1015
1016
1017 static inline void
1018 worker_thread_node_runtime_update_internal (void)
1019 {
1020   int i, j;
1021   vlib_main_t *vm;
1022   vlib_node_main_t *nm, *nm_clone;
1023   vlib_main_t *vm_clone;
1024   vlib_node_runtime_t *rt;
1025   never_inline void
1026     vlib_node_runtime_sync_stats (vlib_main_t * vm,
1027                                   vlib_node_runtime_t * r,
1028                                   uword n_calls,
1029                                   uword n_vectors, uword n_clocks);
1030
1031   ASSERT (vlib_get_thread_index () == 0);
1032
1033   vm = vlib_mains[0];
1034   nm = &vm->node_main;
1035
1036   ASSERT (*vlib_worker_threads->wait_at_barrier == 1);
1037
1038   /*
1039    * Scrape all runtime stats, so we don't lose node runtime(s) with
1040    * pending counts, or throw away worker / io thread counts.
1041    */
1042   for (j = 0; j < vec_len (nm->nodes); j++)
1043     {
1044       vlib_node_t *n;
1045       n = nm->nodes[j];
1046       vlib_node_sync_stats (vm, n);
1047     }
1048
1049   for (i = 1; i < vec_len (vlib_mains); i++)
1050     {
1051       vlib_node_t *n;
1052
1053       vm_clone = vlib_mains[i];
1054       nm_clone = &vm_clone->node_main;
1055
1056       for (j = 0; j < vec_len (nm_clone->nodes); j++)
1057         {
1058           n = nm_clone->nodes[j];
1059
1060           rt = vlib_node_get_runtime (vm_clone, n->index);
1061           vlib_node_runtime_sync_stats (vm_clone, rt, 0, 0, 0);
1062         }
1063     }
1064
1065   /* Per-worker clone rebuilds are now done on each thread */
1066 }
1067
1068
1069 void
1070 vlib_worker_thread_node_refork (void)
1071 {
1072   vlib_main_t *vm, *vm_clone;
1073   vlib_node_main_t *nm, *nm_clone;
1074   vlib_node_t **old_nodes_clone;
1075   vlib_node_runtime_t *rt, *old_rt;
1076
1077   vlib_node_t *new_n_clone;
1078
1079   int j;
1080
1081   vm = vlib_mains[0];
1082   nm = &vm->node_main;
1083   vm_clone = vlib_get_main ();
1084   nm_clone = &vm_clone->node_main;
1085
1086   /* Re-clone error heap */
1087   u64 *old_counters = vm_clone->error_main.counters;
1088   u64 *old_counters_all_clear = vm_clone->error_main.counters_last_clear;
1089
1090   clib_memcpy (&vm_clone->error_main, &vm->error_main,
1091                sizeof (vm->error_main));
1092   j = vec_len (vm->error_main.counters) - 1;
1093   vec_validate_aligned (old_counters, j, CLIB_CACHE_LINE_BYTES);
1094   vec_validate_aligned (old_counters_all_clear, j, CLIB_CACHE_LINE_BYTES);
1095   vm_clone->error_main.counters = old_counters;
1096   vm_clone->error_main.counters_last_clear = old_counters_all_clear;
1097
1098   nm_clone = &vm_clone->node_main;
1099   vec_free (nm_clone->next_frames);
1100   nm_clone->next_frames = vec_dup (nm->next_frames);
1101
1102   for (j = 0; j < vec_len (nm_clone->next_frames); j++)
1103     {
1104       vlib_next_frame_t *nf = &nm_clone->next_frames[j];
1105       u32 save_node_runtime_index;
1106       u32 save_flags;
1107
1108       save_node_runtime_index = nf->node_runtime_index;
1109       save_flags = nf->flags & VLIB_FRAME_NO_FREE_AFTER_DISPATCH;
1110       vlib_next_frame_init (nf);
1111       nf->node_runtime_index = save_node_runtime_index;
1112       nf->flags = save_flags;
1113     }
1114
1115   old_nodes_clone = nm_clone->nodes;
1116   nm_clone->nodes = 0;
1117
1118   /* re-fork nodes */
1119
1120   /* Allocate all nodes in single block for speed */
1121   new_n_clone =
1122     clib_mem_alloc_no_fail (vec_len (nm->nodes) * sizeof (*new_n_clone));
1123   for (j = 0; j < vec_len (nm->nodes); j++)
1124     {
1125       vlib_node_t *old_n_clone;
1126       vlib_node_t *new_n;
1127
1128       new_n = nm->nodes[j];
1129       old_n_clone = old_nodes_clone[j];
1130
1131       clib_memcpy (new_n_clone, new_n, sizeof (*new_n));
1132       /* none of the copied nodes have enqueue rights given out */
1133       new_n_clone->owner_node_index = VLIB_INVALID_NODE_INDEX;
1134
1135       if (j >= vec_len (old_nodes_clone))
1136         {
1137           /* new node, set to zero */
1138           memset (&new_n_clone->stats_total, 0,
1139                   sizeof (new_n_clone->stats_total));
1140           memset (&new_n_clone->stats_last_clear, 0,
1141                   sizeof (new_n_clone->stats_last_clear));
1142         }
1143       else
1144         {
1145           /* Copy stats if the old data is valid */
1146           clib_memcpy (&new_n_clone->stats_total,
1147                        &old_n_clone->stats_total,
1148                        sizeof (new_n_clone->stats_total));
1149           clib_memcpy (&new_n_clone->stats_last_clear,
1150                        &old_n_clone->stats_last_clear,
1151                        sizeof (new_n_clone->stats_last_clear));
1152
1153           /* keep previous node state */
1154           new_n_clone->state = old_n_clone->state;
1155         }
1156       vec_add1 (nm_clone->nodes, new_n_clone);
1157       new_n_clone++;
1158     }
1159   /* Free the old node clones */
1160   clib_mem_free (old_nodes_clone[0]);
1161
1162   vec_free (old_nodes_clone);
1163
1164
1165   /* re-clone internal nodes */
1166   old_rt = nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL];
1167   nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL] =
1168     vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INTERNAL]);
1169
1170   vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INTERNAL])
1171   {
1172     vlib_node_t *n = vlib_get_node (vm, rt->node_index);
1173     rt->thread_index = vm_clone->thread_index;
1174     /* copy runtime_data, will be overwritten later for existing rt */
1175     if (n->runtime_data && n->runtime_data_bytes > 0)
1176       clib_memcpy (rt->runtime_data, n->runtime_data,
1177                    clib_min (VLIB_NODE_RUNTIME_DATA_SIZE,
1178                              n->runtime_data_bytes));
1179   }
1180
1181   for (j = 0; j < vec_len (old_rt); j++)
1182     {
1183       rt = vlib_node_get_runtime (vm_clone, old_rt[j].node_index);
1184       rt->state = old_rt[j].state;
1185       clib_memcpy (rt->runtime_data, old_rt[j].runtime_data,
1186                    VLIB_NODE_RUNTIME_DATA_SIZE);
1187     }
1188
1189   vec_free (old_rt);
1190
1191   /* re-clone input nodes */
1192   old_rt = nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT];
1193   nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT] =
1194     vec_dup (nm->nodes_by_type[VLIB_NODE_TYPE_INPUT]);
1195
1196   vec_foreach (rt, nm_clone->nodes_by_type[VLIB_NODE_TYPE_INPUT])
1197   {
1198     vlib_node_t *n = vlib_get_node (vm, rt->node_index);
1199     rt->thread_index = vm_clone->thread_index;
1200     /* copy runtime_data, will be overwritten later for existing rt */
1201     if (n->runtime_data && n->runtime_data_bytes > 0)
1202       clib_memcpy (rt->runtime_data, n->runtime_data,
1203                    clib_min (VLIB_NODE_RUNTIME_DATA_SIZE,
1204                              n->runtime_data_bytes));
1205   }
1206
1207   for (j = 0; j < vec_len (old_rt); j++)
1208     {
1209       rt = vlib_node_get_runtime (vm_clone, old_rt[j].node_index);
1210       rt->state = old_rt[j].state;
1211       clib_memcpy (rt->runtime_data, old_rt[j].runtime_data,
1212                    VLIB_NODE_RUNTIME_DATA_SIZE);
1213     }
1214
1215   vec_free (old_rt);
1216
1217   nm_clone->processes = vec_dup (nm->processes);
1218 }
1219
1220 void
1221 vlib_worker_thread_node_runtime_update (void)
1222 {
1223   /*
1224    * Make a note that we need to do a node runtime update
1225    * prior to releasing the barrier.
1226    */
1227   vlib_global_main.need_vlib_worker_thread_node_runtime_update = 1;
1228 }
1229
1230 u32
1231 unformat_sched_policy (unformat_input_t * input, va_list * args)
1232 {
1233   u32 *r = va_arg (*args, u32 *);
1234
1235   if (0);
1236 #define _(v,f,s) else if (unformat (input, s)) *r = SCHED_POLICY_##f;
1237   foreach_sched_policy
1238 #undef _
1239     else
1240     return 0;
1241   return 1;
1242 }
1243
1244 static clib_error_t *
1245 cpu_config (vlib_main_t * vm, unformat_input_t * input)
1246 {
1247   vlib_thread_registration_t *tr;
1248   uword *p;
1249   vlib_thread_main_t *tm = &vlib_thread_main;
1250   u8 *name;
1251   u64 coremask;
1252   uword *bitmap;
1253   u32 count;
1254
1255   tm->thread_registrations_by_name = hash_create_string (0, sizeof (uword));
1256
1257   tm->n_thread_stacks = 1;      /* account for main thread */
1258   tm->sched_policy = ~0;
1259   tm->sched_priority = ~0;
1260
1261   tr = tm->next;
1262
1263   while (tr)
1264     {
1265       hash_set_mem (tm->thread_registrations_by_name, tr->name, (uword) tr);
1266       tr = tr->next;
1267     }
1268
1269   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
1270     {
1271       if (unformat (input, "use-pthreads"))
1272         tm->use_pthreads = 1;
1273       else if (unformat (input, "thread-prefix %v", &tm->thread_prefix))
1274         ;
1275       else if (unformat (input, "main-core %u", &tm->main_lcore))
1276         ;
1277       else if (unformat (input, "skip-cores %u", &tm->skip_cores))
1278         ;
1279       else if (unformat (input, "coremask-%s %llx", &name, &coremask))
1280         {
1281           p = hash_get_mem (tm->thread_registrations_by_name, name);
1282           if (p == 0)
1283             return clib_error_return (0, "no such thread type '%s'", name);
1284
1285           tr = (vlib_thread_registration_t *) p[0];
1286
1287           if (tr->use_pthreads)
1288             return clib_error_return (0,
1289                                       "coremask cannot be set for '%s' threads",
1290                                       name);
1291
1292           tr->coremask = clib_bitmap_set_multiple
1293             (tr->coremask, 0, coremask, BITS (coremask));
1294           tr->count = clib_bitmap_count_set_bits (tr->coremask);
1295         }
1296       else if (unformat (input, "corelist-%s %U", &name, unformat_bitmap_list,
1297                          &bitmap))
1298         {
1299           p = hash_get_mem (tm->thread_registrations_by_name, name);
1300           if (p == 0)
1301             return clib_error_return (0, "no such thread type '%s'", name);
1302
1303           tr = (vlib_thread_registration_t *) p[0];
1304
1305           if (tr->use_pthreads)
1306             return clib_error_return (0,
1307                                       "corelist cannot be set for '%s' threads",
1308                                       name);
1309
1310           tr->coremask = bitmap;
1311           tr->count = clib_bitmap_count_set_bits (tr->coremask);
1312         }
1313       else
1314         if (unformat
1315             (input, "scheduler-policy %U", unformat_sched_policy,
1316              &tm->sched_policy))
1317         ;
1318       else if (unformat (input, "scheduler-priority %u", &tm->sched_priority))
1319         ;
1320       else if (unformat (input, "%s %u", &name, &count))
1321         {
1322           p = hash_get_mem (tm->thread_registrations_by_name, name);
1323           if (p == 0)
1324             return clib_error_return (0, "no such thread type 3 '%s'", name);
1325
1326           tr = (vlib_thread_registration_t *) p[0];
1327           if (tr->fixed_count)
1328             return clib_error_return
1329               (0, "number of %s threads not configurable", tr->name);
1330           tr->count = count;
1331         }
1332       else
1333         break;
1334     }
1335
1336   if (tm->sched_priority != ~0)
1337     {
1338       if (tm->sched_policy == SCHED_FIFO || tm->sched_policy == SCHED_RR)
1339         {
1340           u32 prio_max = sched_get_priority_max (tm->sched_policy);
1341           u32 prio_min = sched_get_priority_min (tm->sched_policy);
1342           if (tm->sched_priority > prio_max)
1343             tm->sched_priority = prio_max;
1344           if (tm->sched_priority < prio_min)
1345             tm->sched_priority = prio_min;
1346         }
1347       else
1348         {
1349           return clib_error_return
1350             (0,
1351              "scheduling priority (%d) is not allowed for `normal` scheduling policy",
1352              tm->sched_priority);
1353         }
1354     }
1355   tr = tm->next;
1356
1357   if (!tm->thread_prefix)
1358     tm->thread_prefix = format (0, "vpp");
1359
1360   while (tr)
1361     {
1362       tm->n_thread_stacks += tr->count;
1363       tm->n_pthreads += tr->count * tr->use_pthreads;
1364       tm->n_threads += tr->count * (tr->use_pthreads == 0);
1365       tr = tr->next;
1366     }
1367
1368   return 0;
1369 }
1370
1371 VLIB_EARLY_CONFIG_FUNCTION (cpu_config, "cpu");
1372
1373 #if !defined (__x86_64__) && !defined (__i386__) && !defined (__aarch64__) && !defined (__powerpc64__) && !defined(__arm__)
1374 void
1375 __sync_fetch_and_add_8 (void)
1376 {
1377   fformat (stderr, "%s called\n", __FUNCTION__);
1378   abort ();
1379 }
1380
1381 void
1382 __sync_add_and_fetch_8 (void)
1383 {
1384   fformat (stderr, "%s called\n", __FUNCTION__);
1385   abort ();
1386 }
1387 #endif
1388
1389 void vnet_main_fixup (vlib_fork_fixup_t which) __attribute__ ((weak));
1390 void
1391 vnet_main_fixup (vlib_fork_fixup_t which)
1392 {
1393 }
1394
1395 void
1396 vlib_worker_thread_fork_fixup (vlib_fork_fixup_t which)
1397 {
1398   vlib_main_t *vm = vlib_get_main ();
1399
1400   if (vlib_mains == 0)
1401     return;
1402
1403   ASSERT (vlib_get_thread_index () == 0);
1404   vlib_worker_thread_barrier_sync (vm);
1405
1406   switch (which)
1407     {
1408     case VLIB_WORKER_THREAD_FORK_FIXUP_NEW_SW_IF_INDEX:
1409       vnet_main_fixup (VLIB_WORKER_THREAD_FORK_FIXUP_NEW_SW_IF_INDEX);
1410       break;
1411
1412     default:
1413       ASSERT (0);
1414     }
1415   vlib_worker_thread_barrier_release (vm);
1416 }
1417
1418   /*
1419    * Enforce minimum open time to minimize packet loss due to Rx overflow,
1420    * based on a test based heuristic that barrier should be open for at least
1421    * 3 time as long as it is closed (with an upper bound of 1ms because by that
1422    *  point it is probably too late to make a difference)
1423    */
1424
1425 #ifndef BARRIER_MINIMUM_OPEN_LIMIT
1426 #define BARRIER_MINIMUM_OPEN_LIMIT 0.001
1427 #endif
1428
1429 #ifndef BARRIER_MINIMUM_OPEN_FACTOR
1430 #define BARRIER_MINIMUM_OPEN_FACTOR 3
1431 #endif
1432
1433 void
1434 vlib_worker_thread_barrier_sync_int (vlib_main_t * vm)
1435 {
1436   f64 deadline;
1437   f64 now;
1438   f64 t_entry;
1439   f64 t_open;
1440   f64 t_closed;
1441   u32 count;
1442
1443   if (vec_len (vlib_mains) < 2)
1444     return;
1445
1446   ASSERT (vlib_get_thread_index () == 0);
1447
1448   count = vec_len (vlib_mains) - 1;
1449
1450   /* Record entry relative to last close */
1451   now = vlib_time_now (vm);
1452   t_entry = now - vm->barrier_epoch;
1453
1454   /* Tolerate recursive calls */
1455   if (++vlib_worker_threads[0].recursion_level > 1)
1456     {
1457       barrier_trace_sync_rec (t_entry);
1458       return;
1459     }
1460
1461   vlib_worker_threads[0].barrier_sync_count++;
1462
1463   /* Enforce minimum barrier open time to minimize packet loss */
1464   ASSERT (vm->barrier_no_close_before <= (now + BARRIER_MINIMUM_OPEN_LIMIT));
1465   while ((now = vlib_time_now (vm)) < vm->barrier_no_close_before)
1466     ;
1467
1468   /* Record time of closure */
1469   t_open = now - vm->barrier_epoch;
1470   vm->barrier_epoch = now;
1471
1472   deadline = now + BARRIER_SYNC_TIMEOUT;
1473
1474   *vlib_worker_threads->wait_at_barrier = 1;
1475   while (*vlib_worker_threads->workers_at_barrier != count)
1476     {
1477       if ((now = vlib_time_now (vm)) > deadline)
1478         {
1479           fformat (stderr, "%s: worker thread deadlock\n", __FUNCTION__);
1480           os_panic ();
1481         }
1482     }
1483
1484   t_closed = now - vm->barrier_epoch;
1485
1486   barrier_trace_sync (t_entry, t_open, t_closed);
1487
1488 }
1489
1490 void
1491 vlib_worker_thread_barrier_release (vlib_main_t * vm)
1492 {
1493   f64 deadline;
1494   f64 now;
1495   f64 minimum_open;
1496   f64 t_entry;
1497   f64 t_closed_total;
1498   f64 t_update_main = 0.0;
1499   int refork_needed = 0;
1500
1501   if (vec_len (vlib_mains) < 2)
1502     return;
1503
1504   ASSERT (vlib_get_thread_index () == 0);
1505
1506
1507   now = vlib_time_now (vm);
1508   t_entry = now - vm->barrier_epoch;
1509
1510   if (--vlib_worker_threads[0].recursion_level > 0)
1511     {
1512       barrier_trace_release_rec (t_entry);
1513       return;
1514     }
1515
1516   /* Update (all) node runtimes before releasing the barrier, if needed */
1517   if (vm->need_vlib_worker_thread_node_runtime_update)
1518     {
1519       /* Do stats elements on main thread */
1520       worker_thread_node_runtime_update_internal ();
1521       vm->need_vlib_worker_thread_node_runtime_update = 0;
1522
1523       /* Do per thread rebuilds in parallel */
1524       refork_needed = 1;
1525       clib_smp_atomic_add (vlib_worker_threads->node_reforks_required,
1526                            (vec_len (vlib_mains) - 1));
1527       now = vlib_time_now (vm);
1528       t_update_main = now - vm->barrier_epoch;
1529     }
1530
1531   deadline = now + BARRIER_SYNC_TIMEOUT;
1532
1533   *vlib_worker_threads->wait_at_barrier = 0;
1534
1535   while (*vlib_worker_threads->workers_at_barrier > 0)
1536     {
1537       if ((now = vlib_time_now (vm)) > deadline)
1538         {
1539           fformat (stderr, "%s: worker thread deadlock\n", __FUNCTION__);
1540           os_panic ();
1541         }
1542     }
1543
1544   /* Wait for reforks before continuing */
1545   if (refork_needed)
1546     {
1547       now = vlib_time_now (vm);
1548
1549       deadline = now + BARRIER_SYNC_TIMEOUT;
1550
1551       while (*vlib_worker_threads->node_reforks_required > 0)
1552         {
1553           if ((now = vlib_time_now (vm)) > deadline)
1554             {
1555               fformat (stderr, "%s: worker thread refork deadlock\n",
1556                        __FUNCTION__);
1557               os_panic ();
1558             }
1559         }
1560     }
1561
1562   t_closed_total = now - vm->barrier_epoch;
1563
1564   minimum_open = t_closed_total * BARRIER_MINIMUM_OPEN_FACTOR;
1565
1566   if (minimum_open > BARRIER_MINIMUM_OPEN_LIMIT)
1567     {
1568       minimum_open = BARRIER_MINIMUM_OPEN_LIMIT;
1569     }
1570
1571   vm->barrier_no_close_before = now + minimum_open;
1572
1573   /* Record barrier epoch (used to enforce minimum open time) */
1574   vm->barrier_epoch = now;
1575
1576   barrier_trace_release (t_entry, t_closed_total, t_update_main);
1577
1578 }
1579
1580 /*
1581  * Check the frame queue to see if any frames are available.
1582  * If so, pull the packets off the frames and put them to
1583  * the handoff node.
1584  */
1585 int
1586 vlib_frame_queue_dequeue (vlib_main_t * vm, vlib_frame_queue_main_t * fqm)
1587 {
1588   u32 thread_id = vm->thread_index;
1589   vlib_frame_queue_t *fq = fqm->vlib_frame_queues[thread_id];
1590   vlib_frame_queue_elt_t *elt;
1591   u32 *from, *to;
1592   vlib_frame_t *f;
1593   int msg_type;
1594   int processed = 0;
1595   u32 n_left_to_node;
1596   u32 vectors = 0;
1597
1598   ASSERT (fq);
1599   ASSERT (vm == vlib_mains[thread_id]);
1600
1601   if (PREDICT_FALSE (fqm->node_index == ~0))
1602     return 0;
1603   /*
1604    * Gather trace data for frame queues
1605    */
1606   if (PREDICT_FALSE (fq->trace))
1607     {
1608       frame_queue_trace_t *fqt;
1609       frame_queue_nelt_counter_t *fqh;
1610       u32 elix;
1611
1612       fqt = &fqm->frame_queue_traces[thread_id];
1613
1614       fqt->nelts = fq->nelts;
1615       fqt->head = fq->head;
1616       fqt->head_hint = fq->head_hint;
1617       fqt->tail = fq->tail;
1618       fqt->threshold = fq->vector_threshold;
1619       fqt->n_in_use = fqt->tail - fqt->head;
1620       if (fqt->n_in_use >= fqt->nelts)
1621         {
1622           // if beyond max then use max
1623           fqt->n_in_use = fqt->nelts - 1;
1624         }
1625
1626       /* Record the number of elements in use in the histogram */
1627       fqh = &fqm->frame_queue_histogram[thread_id];
1628       fqh->count[fqt->n_in_use]++;
1629
1630       /* Record a snapshot of the elements in use */
1631       for (elix = 0; elix < fqt->nelts; elix++)
1632         {
1633           elt = fq->elts + ((fq->head + 1 + elix) & (fq->nelts - 1));
1634           if (1 || elt->valid)
1635             {
1636               fqt->n_vectors[elix] = elt->n_vectors;
1637             }
1638         }
1639       fqt->written = 1;
1640     }
1641
1642   while (1)
1643     {
1644       if (fq->head == fq->tail)
1645         {
1646           fq->head_hint = fq->head;
1647           return processed;
1648         }
1649
1650       elt = fq->elts + ((fq->head + 1) & (fq->nelts - 1));
1651
1652       if (!elt->valid)
1653         {
1654           fq->head_hint = fq->head;
1655           return processed;
1656         }
1657
1658       from = elt->buffer_index;
1659       msg_type = elt->msg_type;
1660
1661       ASSERT (msg_type == VLIB_FRAME_QUEUE_ELT_DISPATCH_FRAME);
1662       ASSERT (elt->n_vectors <= VLIB_FRAME_SIZE);
1663
1664       f = vlib_get_frame_to_node (vm, fqm->node_index);
1665
1666       to = vlib_frame_vector_args (f);
1667
1668       n_left_to_node = elt->n_vectors;
1669
1670       while (n_left_to_node >= 4)
1671         {
1672           to[0] = from[0];
1673           to[1] = from[1];
1674           to[2] = from[2];
1675           to[3] = from[3];
1676           to += 4;
1677           from += 4;
1678           n_left_to_node -= 4;
1679         }
1680
1681       while (n_left_to_node > 0)
1682         {
1683           to[0] = from[0];
1684           to++;
1685           from++;
1686           n_left_to_node--;
1687         }
1688
1689       vectors += elt->n_vectors;
1690       f->n_vectors = elt->n_vectors;
1691       vlib_put_frame_to_node (vm, fqm->node_index, f);
1692
1693       elt->valid = 0;
1694       elt->n_vectors = 0;
1695       elt->msg_type = 0xfefefefe;
1696       CLIB_MEMORY_BARRIER ();
1697       fq->head++;
1698       processed++;
1699
1700       /*
1701        * Limit the number of packets pushed into the graph
1702        */
1703       if (vectors >= fq->vector_threshold)
1704         {
1705           fq->head_hint = fq->head;
1706           return processed;
1707         }
1708     }
1709   ASSERT (0);
1710   return processed;
1711 }
1712
1713 void
1714 vlib_worker_thread_fn (void *arg)
1715 {
1716   vlib_worker_thread_t *w = (vlib_worker_thread_t *) arg;
1717   vlib_thread_main_t *tm = vlib_get_thread_main ();
1718   vlib_main_t *vm = vlib_get_main ();
1719   clib_error_t *e;
1720
1721   ASSERT (vm->thread_index == vlib_get_thread_index ());
1722
1723   vlib_worker_thread_init (w);
1724   clib_time_init (&vm->clib_time);
1725   clib_mem_set_heap (w->thread_mheap);
1726
1727   /* Wait until the dpdk init sequence is complete */
1728   while (tm->extern_thread_mgmt && tm->worker_thread_release == 0)
1729     vlib_worker_thread_barrier_check ();
1730
1731   e = vlib_call_init_exit_functions
1732     (vm, vm->worker_init_function_registrations, 1 /* call_once */ );
1733   if (e)
1734     clib_error_report (e);
1735
1736   vlib_worker_loop (vm);
1737 }
1738
1739 /* *INDENT-OFF* */
1740 VLIB_REGISTER_THREAD (worker_thread_reg, static) = {
1741   .name = "workers",
1742   .short_name = "wk",
1743   .function = vlib_worker_thread_fn,
1744 };
1745 /* *INDENT-ON* */
1746
1747 u32
1748 vlib_frame_queue_main_init (u32 node_index, u32 frame_queue_nelts)
1749 {
1750   vlib_thread_main_t *tm = vlib_get_thread_main ();
1751   vlib_frame_queue_main_t *fqm;
1752   vlib_frame_queue_t *fq;
1753   int i;
1754
1755   if (frame_queue_nelts == 0)
1756     frame_queue_nelts = FRAME_QUEUE_NELTS;
1757
1758   vec_add2 (tm->frame_queue_mains, fqm, 1);
1759
1760   fqm->node_index = node_index;
1761
1762   vec_validate (fqm->vlib_frame_queues, tm->n_vlib_mains - 1);
1763   _vec_len (fqm->vlib_frame_queues) = 0;
1764   for (i = 0; i < tm->n_vlib_mains; i++)
1765     {
1766       fq = vlib_frame_queue_alloc (frame_queue_nelts);
1767       vec_add1 (fqm->vlib_frame_queues, fq);
1768     }
1769
1770   return (fqm - tm->frame_queue_mains);
1771 }
1772
1773 int
1774 vlib_thread_cb_register (struct vlib_main_t *vm, vlib_thread_callbacks_t * cb)
1775 {
1776   vlib_thread_main_t *tm = vlib_get_thread_main ();
1777
1778   if (tm->extern_thread_mgmt)
1779     return -1;
1780
1781   tm->cb.vlib_launch_thread_cb = cb->vlib_launch_thread_cb;
1782   tm->extern_thread_mgmt = 1;
1783   return 0;
1784 }
1785
1786 void
1787 vlib_process_signal_event_mt_helper (vlib_process_signal_event_mt_args_t *
1788                                      args)
1789 {
1790   ASSERT (vlib_get_thread_index () == 0);
1791   vlib_process_signal_event (vlib_get_main (), args->node_index,
1792                              args->type_opaque, args->data);
1793 }
1794
1795 void *rpc_call_main_thread_cb_fn;
1796
1797 void
1798 vlib_rpc_call_main_thread (void *callback, u8 * args, u32 arg_size)
1799 {
1800   if (rpc_call_main_thread_cb_fn)
1801     {
1802       void (*fp) (void *, u8 *, u32) = rpc_call_main_thread_cb_fn;
1803       (*fp) (callback, args, arg_size);
1804     }
1805   else
1806     clib_warning ("BUG: rpc_call_main_thread_cb_fn NULL!");
1807 }
1808
1809 clib_error_t *
1810 threads_init (vlib_main_t * vm)
1811 {
1812   return 0;
1813 }
1814
1815 VLIB_INIT_FUNCTION (threads_init);
1816
1817 /*
1818  * fd.io coding-style-patch-verification: ON
1819  *
1820  * Local Variables:
1821  * eval: (c-set-style "gnu")
1822  * End:
1823  */