vcl: handle worker process exit
[vpp.git] / src / vcl / vcl_private.c
1 /*
2  * Copyright (c) 2018 Cisco and/or its affiliates.
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this
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
16 #include <vcl/vcl_private.h>
17
18 static pthread_key_t vcl_worker_stop_key;
19
20 static const char *
21 vppcom_app_state_str (app_state_t state)
22 {
23   char *st;
24
25   switch (state)
26     {
27     case STATE_APP_START:
28       st = "STATE_APP_START";
29       break;
30
31     case STATE_APP_CONN_VPP:
32       st = "STATE_APP_CONN_VPP";
33       break;
34
35     case STATE_APP_ENABLED:
36       st = "STATE_APP_ENABLED";
37       break;
38
39     case STATE_APP_ATTACHED:
40       st = "STATE_APP_ATTACHED";
41       break;
42
43     default:
44       st = "UNKNOWN_APP_STATE";
45       break;
46     }
47
48   return st;
49 }
50
51 int
52 vcl_wait_for_app_state_change (app_state_t app_state)
53 {
54   vcl_worker_t *wrk = vcl_worker_get_current ();
55   f64 timeout = clib_time_now (&wrk->clib_time) + vcm->cfg.app_timeout;
56
57   while (clib_time_now (&wrk->clib_time) < timeout)
58     {
59       if (vcm->app_state == app_state)
60         return VPPCOM_OK;
61       if (vcm->app_state == STATE_APP_FAILED)
62         return VPPCOM_ECONNABORTED;
63     }
64   VDBG (0, "VCL<%d>: timeout waiting for state %s (%d)", getpid (),
65         vppcom_app_state_str (app_state), app_state);
66   vcl_evt (VCL_EVT_SESSION_TIMEOUT, vcm, app_state);
67
68   return VPPCOM_ETIMEDOUT;
69 }
70
71 vcl_cut_through_registration_t *
72 vcl_ct_registration_lock_and_alloc (vcl_worker_t * wrk)
73 {
74   vcl_cut_through_registration_t *cr;
75   clib_spinlock_lock (&wrk->ct_registration_lock);
76   pool_get (wrk->cut_through_registrations, cr);
77   memset (cr, 0, sizeof (*cr));
78   cr->epoll_evt_conn_index = -1;
79   return cr;
80 }
81
82 u32
83 vcl_ct_registration_index (vcl_worker_t * wrk,
84                            vcl_cut_through_registration_t * ctr)
85 {
86   return (ctr - wrk->cut_through_registrations);
87 }
88
89 void
90 vcl_ct_registration_lock (vcl_worker_t * wrk)
91 {
92   clib_spinlock_lock (&wrk->ct_registration_lock);
93 }
94
95 void
96 vcl_ct_registration_unlock (vcl_worker_t * wrk)
97 {
98   clib_spinlock_unlock (&wrk->ct_registration_lock);
99 }
100
101 vcl_cut_through_registration_t *
102 vcl_ct_registration_get (vcl_worker_t * wrk, u32 ctr_index)
103 {
104   if (pool_is_free_index (wrk->cut_through_registrations, ctr_index))
105     return 0;
106   return pool_elt_at_index (wrk->cut_through_registrations, ctr_index);
107 }
108
109 vcl_cut_through_registration_t *
110 vcl_ct_registration_lock_and_lookup (vcl_worker_t * wrk, uword mq_addr)
111 {
112   uword *p;
113   clib_spinlock_lock (&wrk->ct_registration_lock);
114   p = hash_get (wrk->ct_registration_by_mq, mq_addr);
115   if (!p)
116     return 0;
117   return vcl_ct_registration_get (wrk, p[0]);
118 }
119
120 void
121 vcl_ct_registration_lookup_add (vcl_worker_t * wrk, uword mq_addr,
122                                 u32 ctr_index)
123 {
124   hash_set (wrk->ct_registration_by_mq, mq_addr, ctr_index);
125 }
126
127 void
128 vcl_ct_registration_lookup_del (vcl_worker_t * wrk, uword mq_addr)
129 {
130   hash_unset (wrk->ct_registration_by_mq, mq_addr);
131 }
132
133 void
134 vcl_ct_registration_del (vcl_worker_t * wrk,
135                          vcl_cut_through_registration_t * ctr)
136 {
137   pool_put (wrk->cut_through_registrations, ctr);
138 }
139
140 vcl_mq_evt_conn_t *
141 vcl_mq_evt_conn_alloc (vcl_worker_t * wrk)
142 {
143   vcl_mq_evt_conn_t *mqc;
144   pool_get (wrk->mq_evt_conns, mqc);
145   memset (mqc, 0, sizeof (*mqc));
146   return mqc;
147 }
148
149 u32
150 vcl_mq_evt_conn_index (vcl_worker_t * wrk, vcl_mq_evt_conn_t * mqc)
151 {
152   return (mqc - wrk->mq_evt_conns);
153 }
154
155 vcl_mq_evt_conn_t *
156 vcl_mq_evt_conn_get (vcl_worker_t * wrk, u32 mq_conn_idx)
157 {
158   return pool_elt_at_index (wrk->mq_evt_conns, mq_conn_idx);
159 }
160
161 int
162 vcl_mq_epoll_add_evfd (vcl_worker_t * wrk, svm_msg_q_t * mq)
163 {
164   struct epoll_event e = { 0 };
165   vcl_mq_evt_conn_t *mqc;
166   u32 mqc_index;
167   int mq_fd;
168
169   mq_fd = svm_msg_q_get_consumer_eventfd (mq);
170
171   if (wrk->mqs_epfd < 0 || mq_fd == -1)
172     return -1;
173
174   mqc = vcl_mq_evt_conn_alloc (wrk);
175   mqc_index = vcl_mq_evt_conn_index (wrk, mqc);
176   mqc->mq_fd = mq_fd;
177   mqc->mq = mq;
178
179   e.events = EPOLLIN;
180   e.data.u32 = mqc_index;
181   if (epoll_ctl (wrk->mqs_epfd, EPOLL_CTL_ADD, mq_fd, &e) < 0)
182     {
183       clib_warning ("failed to add mq eventfd to mq epoll fd");
184       return -1;
185     }
186
187   return mqc_index;
188 }
189
190 int
191 vcl_mq_epoll_del_evfd (vcl_worker_t * wrk, u32 mqc_index)
192 {
193   vcl_mq_evt_conn_t *mqc;
194
195   if (wrk->mqs_epfd || mqc_index == ~0)
196     return -1;
197
198   mqc = vcl_mq_evt_conn_get (wrk, mqc_index);
199   if (epoll_ctl (wrk->mqs_epfd, EPOLL_CTL_DEL, mqc->mq_fd, 0) < 0)
200     {
201       clib_warning ("failed to del mq eventfd to mq epoll fd");
202       return -1;
203     }
204   return 0;
205 }
206
207 static vcl_worker_t *
208 vcl_worker_alloc (void)
209 {
210   vcl_worker_t *wrk;
211   pool_get (vcm->workers, wrk);
212   memset (wrk, 0, sizeof (*wrk));
213   wrk->wrk_index = wrk - vcm->workers;
214   return wrk;
215 }
216
217 static void
218 vcl_worker_free (vcl_worker_t * wrk)
219 {
220   pool_put (vcm->workers, wrk);
221 }
222
223 void
224 vcl_worker_cleanup (u8 notify_vpp)
225 {
226   vcl_worker_t *wrk = vcl_worker_get_current ();
227
228   clib_spinlock_lock (&vcm->workers_lock);
229   if (notify_vpp)
230     vcl_send_app_worker_add_del (0 /* is_add */ );
231   if (wrk->mqs_epfd > 0)
232     close (wrk->mqs_epfd);
233   hash_free (wrk->session_index_by_vpp_handles);
234   hash_free (wrk->ct_registration_by_mq);
235   clib_spinlock_free (&wrk->ct_registration_lock);
236   vec_free (wrk->mq_events);
237   vec_free (wrk->mq_msg_vector);
238   vcl_set_worker_index (~0);
239   vcl_worker_free (wrk);
240   clib_spinlock_unlock (&vcm->workers_lock);
241 }
242
243 static void
244 vcl_worker_cleanup_cb (void *arg)
245 {
246   u32 wrk_index = vcl_get_worker_index ();
247   vcl_worker_cleanup (1 /* notify vpp */ );
248   VDBG (0, "cleaned up worker %u", wrk_index);
249 }
250
251 vcl_worker_t *
252 vcl_worker_alloc_and_init ()
253 {
254   vcl_worker_t *wrk;
255
256   /* This was initialized already */
257   if (vcl_get_worker_index () != ~0)
258     return 0;
259
260   if (pool_elts (vcm->workers) == vcm->cfg.max_workers)
261     {
262       VDBG (0, "max-workers %u limit reached", vcm->cfg.max_workers);
263       return 0;
264     }
265
266   clib_spinlock_lock (&vcm->workers_lock);
267   wrk = vcl_worker_alloc ();
268   vcl_set_worker_index (wrk->wrk_index);
269   wrk->thread_id = pthread_self ();
270   wrk->current_pid = getpid ();
271
272   wrk->mqs_epfd = -1;
273   if (vcm->cfg.use_mq_eventfd)
274     {
275       wrk->mqs_epfd = epoll_create (1);
276       if (wrk->mqs_epfd < 0)
277         {
278           clib_unix_warning ("epoll_create() returned");
279           goto done;
280         }
281     }
282
283   wrk->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
284   wrk->ct_registration_by_mq = hash_create (0, sizeof (uword));
285   clib_spinlock_init (&wrk->ct_registration_lock);
286   clib_time_init (&wrk->clib_time);
287   vec_validate (wrk->mq_events, 64);
288   vec_validate (wrk->mq_msg_vector, 128);
289   vec_reset_length (wrk->mq_msg_vector);
290   vec_validate (wrk->unhandled_evts_vector, 128);
291   vec_reset_length (wrk->unhandled_evts_vector);
292   clib_spinlock_unlock (&vcm->workers_lock);
293
294 done:
295   return wrk;
296 }
297
298 int
299 vcl_worker_register_with_vpp (void)
300 {
301   vcl_worker_t *wrk = vcl_worker_get_current ();
302
303   clib_spinlock_lock (&vcm->workers_lock);
304
305   vcm->app_state = STATE_APP_ADDING_WORKER;
306   vcl_send_app_worker_add_del (1 /* is_add */ );
307   if (vcl_wait_for_app_state_change (STATE_APP_READY))
308     {
309       clib_warning ("failed to add worker to vpp");
310       return -1;
311     }
312
313   if (pthread_key_create (&vcl_worker_stop_key, vcl_worker_cleanup_cb))
314     clib_warning ("failed to add pthread cleanup function");
315   if (pthread_setspecific (vcl_worker_stop_key, &wrk->thread_id))
316     clib_warning ("failed to setup key value");
317
318   clib_spinlock_unlock (&vcm->workers_lock);
319
320   VDBG (0, "added worker %u", wrk->wrk_index);
321   return 0;
322 }
323
324 int
325 vcl_worker_set_bapi (void)
326 {
327   vcl_worker_t *wrk = vcl_worker_get_current ();
328   int i;
329
330   /* Find the first worker with the same pid */
331   for (i = 0; i < vec_len (vcm->workers); i++)
332     {
333       if (i == wrk->wrk_index)
334         continue;
335       if (vcm->workers[i].current_pid == wrk->current_pid)
336         {
337           wrk->vl_input_queue = vcm->workers[i].vl_input_queue;
338           wrk->my_client_index = vcm->workers[i].my_client_index;
339           return 0;
340         }
341     }
342   return -1;
343 }
344
345 vcl_shared_session_t *
346 vcl_shared_session_alloc (void)
347 {
348   vcl_shared_session_t *ss;
349   pool_get (vcm->shared_sessions, ss);
350   memset (ss, 0, sizeof (*ss));
351   ss->ss_index = ss - vcm->shared_sessions;
352   return ss;
353 }
354
355 vcl_shared_session_t *
356 vcl_shared_session_get (u32 ss_index)
357 {
358   if (pool_is_free_index (vcm->shared_sessions, ss_index))
359     return 0;
360   return pool_elt_at_index (vcm->shared_sessions, ss_index);
361 }
362
363 void
364 vcl_shared_session_free (vcl_shared_session_t * ss)
365 {
366   pool_put (vcm->shared_sessions, ss);
367 }
368
369 void
370 vcl_worker_share_session (vcl_worker_t * parent, vcl_worker_t * wrk,
371                           vcl_session_t * new_s)
372 {
373   vcl_shared_session_t *ss;
374   vcl_session_t *s;
375
376   s = vcl_session_get (parent, new_s->session_index);
377   if (s->shared_index == ~0)
378     {
379       ss = vcl_shared_session_alloc ();
380       vec_add1 (ss->workers, parent->wrk_index);
381       s->shared_index = ss->ss_index;
382     }
383   else
384     {
385       ss = vcl_shared_session_get (s->shared_index);
386     }
387   new_s->shared_index = ss->ss_index;
388   vec_add1 (ss->workers, wrk->wrk_index);
389 }
390
391 int
392 vcl_worker_unshare_session (vcl_worker_t * wrk, vcl_session_t * s)
393 {
394   vcl_shared_session_t *ss;
395   int i;
396
397   ss = vcl_shared_session_get (s->shared_index);
398   for (i = 0; i < vec_len (ss->workers); i++)
399     {
400       if (ss->workers[i] == wrk->wrk_index)
401         {
402           vec_del1 (ss->workers, i);
403           break;
404         }
405     }
406
407   if (vec_len (ss->workers) == 0)
408     {
409       vcl_shared_session_free (ss);
410       return 1;
411     }
412
413   return 0;
414 }
415
416 void
417 vcl_worker_share_sessions (u32 parent_wrk_index)
418 {
419   vcl_worker_t *parent_wrk, *wrk;
420   vcl_session_t *new_s;
421
422   parent_wrk = vcl_worker_get (parent_wrk_index);
423   if (!parent_wrk->sessions)
424     return;
425
426   wrk = vcl_worker_get_current ();
427   wrk->sessions = pool_dup (parent_wrk->sessions);
428   wrk->session_index_by_vpp_handles =
429     hash_dup (parent_wrk->session_index_by_vpp_handles);
430
431   /* *INDENT-OFF* */
432   pool_foreach (new_s, wrk->sessions, ({
433     vcl_worker_share_session (parent_wrk, wrk, new_s);
434   }));
435   /* *INDENT-ON* */
436 }
437
438 int
439 vcl_session_get_refcnt (vcl_session_t * s)
440 {
441   vcl_shared_session_t *ss;
442   ss = vcl_shared_session_get (s->shared_index);
443   if (ss)
444     return vec_len (ss->workers);
445   return 0;
446 }
447
448 void
449 vcl_segment_table_add (u64 segment_handle, u32 svm_segment_index)
450 {
451   clib_rwlock_writer_lock (&vcm->segment_table_lock);
452   hash_set (vcm->segment_table, segment_handle, svm_segment_index);
453   clib_rwlock_writer_unlock (&vcm->segment_table_lock);
454 }
455
456 u32
457 vcl_segment_table_lookup (u64 segment_handle)
458 {
459   uword *seg_indexp;
460
461   clib_rwlock_reader_lock (&vcm->segment_table_lock);
462   seg_indexp = hash_get (vcm->segment_table, segment_handle);
463   clib_rwlock_reader_unlock (&vcm->segment_table_lock);
464
465   if (!seg_indexp)
466     return VCL_INVALID_SEGMENT_INDEX;
467   return ((u32) * seg_indexp);
468 }
469
470 void
471 vcl_segment_table_del (u64 segment_handle)
472 {
473   clib_rwlock_writer_lock (&vcm->segment_table_lock);
474   hash_unset (vcm->segment_table, segment_handle);
475   clib_rwlock_writer_unlock (&vcm->segment_table_lock);
476 }
477
478 /*
479  * fd.io coding-style-patch-verification: ON
480  *
481  * Local Variables:
482  * eval: (c-set-style "gnu")
483  * End:
484  */