session: cleanup segment manager and fifo segment
[vpp.git] / src / vnet / session / application_worker.c
1 /*
2  * Copyright (c) 2019 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
16 #include <vnet/session/application.h>
17 #include <vnet/session/application_interface.h>
18 #include <vnet/session/session.h>
19
20 /**
21  * Pool of workers associated to apps
22  */
23 static app_worker_t *app_workers;
24
25 app_worker_t *
26 app_worker_alloc (application_t * app)
27 {
28   app_worker_t *app_wrk;
29   pool_get (app_workers, app_wrk);
30   clib_memset (app_wrk, 0, sizeof (*app_wrk));
31   app_wrk->wrk_index = app_wrk - app_workers;
32   app_wrk->app_index = app->app_index;
33   app_wrk->wrk_map_index = ~0;
34   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
35   app_wrk->first_segment_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
36   APP_DBG ("New app %v worker %u", app_get_name (app), app_wrk->wrk_index);
37   return app_wrk;
38 }
39
40 app_worker_t *
41 app_worker_get (u32 wrk_index)
42 {
43   return pool_elt_at_index (app_workers, wrk_index);
44 }
45
46 app_worker_t *
47 app_worker_get_if_valid (u32 wrk_index)
48 {
49   if (pool_is_free_index (app_workers, wrk_index))
50     return 0;
51   return pool_elt_at_index (app_workers, wrk_index);
52 }
53
54 void
55 app_worker_free (app_worker_t * app_wrk)
56 {
57   application_t *app = application_get (app_wrk->app_index);
58   vnet_unlisten_args_t _a, *a = &_a;
59   u64 handle, *handles = 0;
60   segment_manager_t *sm;
61   u32 sm_index;
62   int i;
63   app_listener_t *al;
64   session_t *ls;
65
66   /*
67    *  Listener cleanup
68    */
69
70   /* *INDENT-OFF* */
71   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
72     ls = listen_session_get_from_handle (handle);
73     al = app_listener_get (app, ls->al_index);
74     vec_add1 (handles, app_listener_handle (al));
75     sm = segment_manager_get (sm_index);
76     sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
77   }));
78   /* *INDENT-ON* */
79
80   for (i = 0; i < vec_len (handles); i++)
81     {
82       a->app_index = app->app_index;
83       a->wrk_map_index = app_wrk->wrk_map_index;
84       a->handle = handles[i];
85       /* seg manager is removed when unbind completes */
86       (void) vnet_unlisten (a);
87     }
88
89   /*
90    * Connects segment manager cleanup
91    */
92
93   if (app_wrk->connects_seg_manager != APP_INVALID_SEGMENT_MANAGER_INDEX)
94     {
95       sm = segment_manager_get (app_wrk->connects_seg_manager);
96       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
97       sm->first_is_protected = 0;
98       segment_manager_init_free (sm);
99     }
100
101   /* If first segment manager is used by a listener */
102   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
103       && app_wrk->first_segment_manager != app_wrk->connects_seg_manager)
104     {
105       sm = segment_manager_get (app_wrk->first_segment_manager);
106       sm->first_is_protected = 0;
107       sm->app_wrk_index = SEGMENT_MANAGER_INVALID_APP_INDEX;
108       /* .. and has no fifos, e.g. it might be used for redirected sessions,
109        * remove it */
110       if (!segment_manager_has_fifos (sm))
111         segment_manager_free (sm);
112     }
113
114   pool_put (app_workers, app_wrk);
115   if (CLIB_DEBUG)
116     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
117 }
118
119 application_t *
120 app_worker_get_app (u32 wrk_index)
121 {
122   app_worker_t *app_wrk;
123   app_wrk = app_worker_get_if_valid (wrk_index);
124   if (!app_wrk)
125     return 0;
126   return application_get_if_valid (app_wrk->app_index);
127 }
128
129 static segment_manager_t *
130 app_worker_alloc_segment_manager (app_worker_t * app_wrk)
131 {
132   segment_manager_t *sm = 0;
133
134   /* If the first segment manager is not in use, don't allocate a new one */
135   if (app_wrk->first_segment_manager != APP_INVALID_SEGMENT_MANAGER_INDEX
136       && app_wrk->first_segment_manager_in_use == 0)
137     {
138       sm = segment_manager_get (app_wrk->first_segment_manager);
139       app_wrk->first_segment_manager_in_use = 1;
140       return sm;
141     }
142
143   sm = segment_manager_alloc ();
144   sm->app_wrk_index = app_wrk->wrk_index;
145
146   return sm;
147 }
148
149 static int
150 app_worker_alloc_session_fifos (segment_manager_t * sm, session_t * s)
151 {
152   svm_fifo_t *rx_fifo = 0, *tx_fifo = 0;
153   int rv;
154
155   if ((rv = segment_manager_alloc_session_fifos (sm, &rx_fifo, &tx_fifo)))
156     return rv;
157
158   rx_fifo->master_session_index = s->session_index;
159   rx_fifo->master_thread_index = s->thread_index;
160
161   tx_fifo->master_session_index = s->session_index;
162   tx_fifo->master_thread_index = s->thread_index;
163
164   s->rx_fifo = rx_fifo;
165   s->tx_fifo = tx_fifo;
166   return 0;
167 }
168
169 int
170 app_worker_init_listener (app_worker_t * app_wrk, session_t * ls)
171 {
172   segment_manager_t *sm;
173
174   /* Allocate segment manager. All sessions derived out of a listen session
175    * have fifos allocated by the same segment manager. */
176   if (!(sm = app_worker_alloc_segment_manager (app_wrk)))
177     return -1;
178
179   /* Keep track of the segment manager for the listener or this worker */
180   hash_set (app_wrk->listeners_table, listen_session_get_handle (ls),
181             segment_manager_index (sm));
182
183   if (session_transport_service_type (ls) == TRANSPORT_SERVICE_CL)
184     {
185       if (!ls->rx_fifo && app_worker_alloc_session_fifos (sm, ls))
186         return -1;
187     }
188   return 0;
189 }
190
191 int
192 app_worker_start_listen (app_worker_t * app_wrk,
193                          app_listener_t * app_listener)
194 {
195   session_t *ls;
196
197   if (clib_bitmap_get (app_listener->workers, app_wrk->wrk_map_index))
198     return VNET_API_ERROR_ADDRESS_IN_USE;
199
200   app_listener->workers = clib_bitmap_set (app_listener->workers,
201                                            app_wrk->wrk_map_index, 1);
202
203   if (app_listener->session_index != SESSION_INVALID_INDEX)
204     {
205       ls = session_get (app_listener->session_index, 0);
206       if (app_worker_init_listener (app_wrk, ls))
207         return -1;
208     }
209
210   if (app_listener->local_index != SESSION_INVALID_INDEX)
211     {
212       ls = session_get (app_listener->local_index, 0);
213       if (app_worker_init_listener (app_wrk, ls))
214         return -1;
215     }
216
217   return 0;
218 }
219
220 static void
221 app_worker_stop_listen_session (app_worker_t * app_wrk, session_t * ls)
222 {
223   session_handle_t handle;
224   segment_manager_t *sm;
225   uword *sm_indexp;
226
227   handle = listen_session_get_handle (ls);
228   sm_indexp = hash_get (app_wrk->listeners_table, handle);
229   if (PREDICT_FALSE (!sm_indexp))
230     return;
231
232   sm = segment_manager_get (*sm_indexp);
233   if (app_wrk->first_segment_manager == *sm_indexp)
234     {
235       /* Delete sessions but don't remove segment manager */
236       app_wrk->first_segment_manager_in_use = 0;
237       segment_manager_del_sessions (sm);
238     }
239   else
240     {
241       segment_manager_init_free (sm);
242     }
243   hash_unset (app_wrk->listeners_table, handle);
244 }
245
246 int
247 app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al)
248 {
249   session_t *ls;
250
251   if (!clib_bitmap_get (al->workers, app_wrk->wrk_map_index))
252     return 0;
253
254   if (al->session_index != SESSION_INVALID_INDEX)
255     {
256       ls = listen_session_get (al->session_index);
257       app_worker_stop_listen_session (app_wrk, ls);
258     }
259
260   if (al->local_index != SESSION_INVALID_INDEX)
261     {
262       ls = listen_session_get (al->local_index);
263       app_worker_stop_listen_session (app_wrk, ls);
264     }
265
266   clib_bitmap_set_no_check (al->workers, app_wrk->wrk_map_index, 0);
267   if (clib_bitmap_is_zero (al->workers))
268     app_listener_cleanup (al);
269
270   return 0;
271 }
272
273 int
274 app_worker_init_accepted (session_t * s)
275 {
276   app_worker_t *app_wrk;
277   segment_manager_t *sm;
278   session_t *listener;
279
280   listener = listen_session_get (s->listener_index);
281   app_wrk = application_listener_select_worker (listener);
282   s->app_wrk_index = app_wrk->wrk_index;
283
284   sm = app_worker_get_listen_segment_manager (app_wrk, listener);
285   if (app_worker_alloc_session_fifos (sm, s))
286     return -1;
287
288   return 0;
289 }
290
291 int
292 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
293 {
294   application_t *app = application_get (app_wrk->app_index);
295   return app->cb_fns.session_accept_callback (s);
296 }
297
298 int
299 app_worker_init_connected (app_worker_t * app_wrk, session_t * s)
300 {
301   application_t *app = application_get (app_wrk->app_index);
302   segment_manager_t *sm;
303
304   /* Allocate fifos for session, unless the app is a builtin proxy */
305   if (!application_is_builtin_proxy (app))
306     {
307       sm = app_worker_get_connect_segment_manager (app_wrk);
308       if (app_worker_alloc_session_fifos (sm, s))
309         return -1;
310     }
311   return 0;
312 }
313
314 int
315 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s, u32 opaque)
316 {
317   application_t *app = application_get (app_wrk->app_index);
318   return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
319                                                  s, s == 0 /* is_fail */ );
320 }
321
322 int
323 app_worker_close_notify (app_worker_t * app_wrk, session_t * s)
324 {
325   application_t *app = application_get (app_wrk->app_index);
326   app->cb_fns.session_disconnect_callback (s);
327   return 0;
328 }
329
330 int
331 app_worker_reset_notify (app_worker_t * app_wrk, session_t * s)
332 {
333   application_t *app = application_get (app_wrk->app_index);
334   app->cb_fns.session_reset_callback (s);
335   return 0;
336 }
337
338 int
339 app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s)
340 {
341   application_t *app = application_get (app_wrk->app_index);
342   app->cb_fns.builtin_app_rx_callback (s);
343   return 0;
344 }
345
346 int
347 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
348 {
349   segment_manager_t *sm;
350   svm_fifo_t *rxf, *txf;
351
352   if (s->session_state == SESSION_STATE_LISTENING)
353     return application_change_listener_owner (s, app_wrk);
354
355   s->app_wrk_index = app_wrk->wrk_index;
356
357   rxf = s->rx_fifo;
358   txf = s->tx_fifo;
359
360   if (!rxf || !txf)
361     return 0;
362
363   s->rx_fifo = 0;
364   s->tx_fifo = 0;
365
366   sm = app_worker_get_or_alloc_connect_segment_manager (app_wrk);
367   if (app_worker_alloc_session_fifos (sm, s))
368     return -1;
369
370   if (!svm_fifo_is_empty_cons (rxf))
371     svm_fifo_clone (s->rx_fifo, rxf);
372
373   if (!svm_fifo_is_empty_cons (txf))
374     svm_fifo_clone (s->tx_fifo, txf);
375
376   segment_manager_dealloc_fifos (rxf, txf);
377
378   return 0;
379 }
380
381 int
382 app_worker_connect_session (app_worker_t * app, session_endpoint_t * sep,
383                             u32 api_context)
384 {
385   int rv;
386
387   /* Make sure we have a segment manager for connects */
388   app_worker_alloc_connects_segment_manager (app);
389
390   if ((rv = session_open (app->wrk_index, sep, api_context)))
391     return rv;
392
393   return 0;
394 }
395
396 int
397 app_worker_alloc_connects_segment_manager (app_worker_t * app_wrk)
398 {
399   segment_manager_t *sm;
400
401   if (app_wrk->connects_seg_manager == APP_INVALID_SEGMENT_MANAGER_INDEX)
402     {
403       sm = app_worker_alloc_segment_manager (app_wrk);
404       if (sm == 0)
405         return -1;
406       app_wrk->connects_seg_manager = segment_manager_index (sm);
407     }
408   return 0;
409 }
410
411 segment_manager_t *
412 app_worker_get_connect_segment_manager (app_worker_t * app)
413 {
414   ASSERT (app->connects_seg_manager != (u32) ~ 0);
415   return segment_manager_get (app->connects_seg_manager);
416 }
417
418 segment_manager_t *
419 app_worker_get_or_alloc_connect_segment_manager (app_worker_t * app_wrk)
420 {
421   if (app_wrk->connects_seg_manager == (u32) ~ 0)
422     app_worker_alloc_connects_segment_manager (app_wrk);
423   return segment_manager_get (app_wrk->connects_seg_manager);
424 }
425
426 segment_manager_t *
427 app_worker_get_listen_segment_manager (app_worker_t * app,
428                                        session_t * listener)
429 {
430   uword *smp;
431   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
432   ASSERT (smp != 0);
433   return segment_manager_get (*smp);
434 }
435
436 session_t *
437 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
438                            u8 transport_proto)
439 {
440   session_t *listener;
441   u64 handle;
442   u32 sm_index;
443   u8 sst;
444
445   sst = session_type_from_proto_and_ip (transport_proto,
446                                         fib_proto == FIB_PROTOCOL_IP4);
447
448   /* *INDENT-OFF* */
449    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
450      listener = listen_session_get_from_handle (handle);
451      if (listener->session_type == sst
452          && !(listener->flags & SESSION_F_PROXY))
453        return listener;
454    }));
455   /* *INDENT-ON* */
456
457   return 0;
458 }
459
460 session_t *
461 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
462                            u8 transport_proto)
463 {
464   session_t *listener;
465   u64 handle;
466   u32 sm_index;
467   u8 sst;
468
469   sst = session_type_from_proto_and_ip (transport_proto,
470                                         fib_proto == FIB_PROTOCOL_IP4);
471
472   /* *INDENT-OFF* */
473    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
474      listener = listen_session_get_from_handle (handle);
475      if (listener->session_type == sst && (listener->flags & SESSION_F_PROXY))
476        return listener;
477    }));
478   /* *INDENT-ON* */
479
480   return 0;
481 }
482
483 /**
484  * Send an API message to the external app, to map new segment
485  */
486 int
487 app_worker_add_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
488 {
489   application_t *app = application_get (app_wrk->app_index);
490   return app->cb_fns.add_segment_callback (app_wrk->api_client_index,
491                                            segment_handle);
492 }
493
494 int
495 app_worker_del_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
496 {
497   application_t *app = application_get (app_wrk->app_index);
498   return app->cb_fns.del_segment_callback (app_wrk->api_client_index,
499                                            segment_handle);
500 }
501
502 static inline u8
503 app_worker_application_is_builtin (app_worker_t * app_wrk)
504 {
505   return app_wrk->app_is_builtin;
506 }
507
508 static inline int
509 app_enqueue_evt (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, u8 lock)
510 {
511   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
512     {
513       clib_warning ("evt q full");
514       svm_msg_q_free_msg (mq, msg);
515       if (lock)
516         svm_msg_q_unlock (mq);
517       return -1;
518     }
519
520   if (lock)
521     {
522       svm_msg_q_add_and_unlock (mq, msg);
523       return 0;
524     }
525
526   /* Even when not locking the ring, we must wait for queue mutex */
527   if (svm_msg_q_add (mq, msg, SVM_Q_WAIT))
528     {
529       clib_warning ("msg q add returned");
530       return -1;
531     }
532   return 0;
533 }
534
535 static inline int
536 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s, u8 lock)
537 {
538   session_event_t *evt;
539   svm_msg_q_msg_t msg;
540   svm_msg_q_t *mq;
541
542   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY
543                      && s->session_state != SESSION_STATE_LISTENING))
544     return 0;
545
546   if (app_worker_application_is_builtin (app_wrk))
547     return app_worker_builtin_rx (app_wrk, s);
548
549   if (svm_fifo_has_event (s->rx_fifo))
550     return 0;
551
552   mq = app_wrk->event_queue;
553   if (lock)
554     svm_msg_q_lock (mq);
555
556   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
557     {
558       clib_warning ("evt q rings full");
559       if (lock)
560         svm_msg_q_unlock (mq);
561       return -1;
562     }
563
564   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
565   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
566
567   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
568   evt->session_index = s->rx_fifo->client_session_index;
569   evt->event_type = SESSION_IO_EVT_RX;
570
571   (void) svm_fifo_set_event (s->rx_fifo);
572
573   if (app_enqueue_evt (mq, &msg, lock))
574     return -1;
575   return 0;
576 }
577
578 static inline int
579 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s, u8 lock)
580 {
581   svm_msg_q_t *mq;
582   session_event_t *evt;
583   svm_msg_q_msg_t msg;
584
585   if (app_worker_application_is_builtin (app_wrk))
586     return 0;
587
588   mq = app_wrk->event_queue;
589   if (lock)
590     svm_msg_q_lock (mq);
591
592   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
593     {
594       clib_warning ("evt q rings full");
595       if (lock)
596         svm_msg_q_unlock (mq);
597       return -1;
598     }
599
600   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
601   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
602
603   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
604   evt->event_type = SESSION_IO_EVT_TX;
605   evt->session_index = s->tx_fifo->client_session_index;
606
607   return app_enqueue_evt (mq, &msg, lock);
608 }
609
610 /* *INDENT-OFF* */
611 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
612                                        session_t *s,
613                                        u8 lock);
614 static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
615     app_send_io_evt_rx,
616     app_send_io_evt_tx,
617 };
618 /* *INDENT-ON* */
619
620 /**
621  * Send event to application
622  *
623  * Logic from queue perspective is non-blocking. If there's
624  * not enough space to enqueue a message, we return.
625  */
626 int
627 app_worker_send_event (app_worker_t * app, session_t * s, u8 evt_type)
628 {
629   ASSERT (app && evt_type <= SESSION_IO_EVT_TX);
630   return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
631 }
632
633 /**
634  * Send event to application
635  *
636  * Logic from queue perspective is blocking. However, if queue is full,
637  * we return.
638  */
639 int
640 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
641                                 u8 evt_type)
642 {
643   return app_send_evt_handler_fns[evt_type] (app, s, 1 /* lock */ );
644 }
645
646 u8 *
647 format_app_worker_listener (u8 * s, va_list * args)
648 {
649   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
650   u64 handle = va_arg (*args, u64);
651   u32 sm_index = va_arg (*args, u32);
652   int verbose = va_arg (*args, int);
653   session_t *listener;
654   const u8 *app_name;
655   u8 *str;
656
657   if (!app_wrk)
658     {
659       if (verbose)
660         s = format (s, "%-40s%-25s%=10s%-15s%-15s%-10s", "Connection", "App",
661                     "Wrk", "API Client", "ListenerID", "SegManager");
662       else
663         s = format (s, "%-40s%-25s%=10s", "Connection", "App", "Wrk");
664
665       return s;
666     }
667
668   app_name = application_name_from_index (app_wrk->app_index);
669   listener = listen_session_get_from_handle (handle);
670   str = format (0, "%U", format_session, listener, verbose);
671
672   if (verbose)
673     {
674       char buf[32];
675       sprintf (buf, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
676       s = format (s, "%-40s%-25s%=10s%-15u%-15u%-10u", str, app_name,
677                   buf, app_wrk->api_client_index, handle, sm_index);
678     }
679   else
680     s = format (s, "%-40s%-25s%=10u", str, app_name, app_wrk->wrk_map_index);
681
682   return s;
683 }
684
685 u8 *
686 format_app_worker (u8 * s, va_list * args)
687 {
688   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
689   u32 indent = 1;
690
691   s = format (s, "%U wrk-index %u app-index %u map-index %u "
692               "api-client-index %d\n", format_white_space, indent,
693               app_wrk->wrk_index, app_wrk->app_index, app_wrk->wrk_map_index,
694               app_wrk->api_client_index);
695   return s;
696 }
697
698 void
699 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
700 {
701   segment_manager_t *sm;
702
703   /* Header */
704   if (!app_wrk)
705     {
706       segment_manager_format_sessions (0, verbose);
707       return;
708     }
709
710   if (app_wrk->connects_seg_manager == (u32) ~ 0)
711     return;
712
713   sm = segment_manager_get (app_wrk->connects_seg_manager);
714   segment_manager_format_sessions (sm, verbose);
715 }
716
717 /*
718  * fd.io coding-style-patch-verification: ON
719  *
720  * Local Variables:
721  * eval: (c-set-style "gnu")
722  * End:
723  */