Add transport_opts to connect_sock bapi
[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->name, 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_builtin_tx (app_worker_t * app_wrk, session_t * s)
348 {
349   application_t *app = application_get (app_wrk->app_index);
350
351   if (!app->cb_fns.builtin_app_tx_callback)
352     return 0;
353
354   app->cb_fns.builtin_app_tx_callback (s);
355   return 0;
356 }
357
358 int
359 app_worker_own_session (app_worker_t * app_wrk, session_t * s)
360 {
361   segment_manager_t *sm;
362   svm_fifo_t *rxf, *txf;
363
364   if (s->session_state == SESSION_STATE_LISTENING)
365     return application_change_listener_owner (s, app_wrk);
366
367   s->app_wrk_index = app_wrk->wrk_index;
368
369   rxf = s->rx_fifo;
370   txf = s->tx_fifo;
371
372   if (!rxf || !txf)
373     return 0;
374
375   s->rx_fifo = 0;
376   s->tx_fifo = 0;
377
378   sm = app_worker_get_or_alloc_connect_segment_manager (app_wrk);
379   if (app_worker_alloc_session_fifos (sm, s))
380     return -1;
381
382   if (!svm_fifo_is_empty_cons (rxf))
383     svm_fifo_clone (s->rx_fifo, rxf);
384
385   if (!svm_fifo_is_empty_cons (txf))
386     svm_fifo_clone (s->tx_fifo, txf);
387
388   segment_manager_dealloc_fifos (rxf, txf);
389
390   return 0;
391 }
392
393 int
394 app_worker_connect_session (app_worker_t * app, session_endpoint_t * sep,
395                             u32 api_context)
396 {
397   int rv;
398
399   /* Make sure we have a segment manager for connects */
400   app_worker_alloc_connects_segment_manager (app);
401
402   if ((rv = session_open (app->wrk_index, sep, api_context)))
403     return rv;
404
405   return 0;
406 }
407
408 int
409 app_worker_alloc_connects_segment_manager (app_worker_t * app_wrk)
410 {
411   segment_manager_t *sm;
412
413   if (app_wrk->connects_seg_manager == APP_INVALID_SEGMENT_MANAGER_INDEX)
414     {
415       sm = app_worker_alloc_segment_manager (app_wrk);
416       if (sm == 0)
417         return -1;
418       app_wrk->connects_seg_manager = segment_manager_index (sm);
419     }
420   return 0;
421 }
422
423 segment_manager_t *
424 app_worker_get_connect_segment_manager (app_worker_t * app)
425 {
426   ASSERT (app->connects_seg_manager != (u32) ~ 0);
427   return segment_manager_get (app->connects_seg_manager);
428 }
429
430 segment_manager_t *
431 app_worker_get_or_alloc_connect_segment_manager (app_worker_t * app_wrk)
432 {
433   if (app_wrk->connects_seg_manager == (u32) ~ 0)
434     app_worker_alloc_connects_segment_manager (app_wrk);
435   return segment_manager_get (app_wrk->connects_seg_manager);
436 }
437
438 segment_manager_t *
439 app_worker_get_listen_segment_manager (app_worker_t * app,
440                                        session_t * listener)
441 {
442   uword *smp;
443   smp = hash_get (app->listeners_table, listen_session_get_handle (listener));
444   ASSERT (smp != 0);
445   return segment_manager_get (*smp);
446 }
447
448 session_t *
449 app_worker_first_listener (app_worker_t * app_wrk, u8 fib_proto,
450                            u8 transport_proto)
451 {
452   session_t *listener;
453   u64 handle;
454   u32 sm_index;
455   u8 sst;
456
457   sst = session_type_from_proto_and_ip (transport_proto,
458                                         fib_proto == FIB_PROTOCOL_IP4);
459
460   /* *INDENT-OFF* */
461    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
462      listener = listen_session_get_from_handle (handle);
463      if (listener->session_type == sst
464          && !(listener->flags & SESSION_F_PROXY))
465        return listener;
466    }));
467   /* *INDENT-ON* */
468
469   return 0;
470 }
471
472 session_t *
473 app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
474                            u8 transport_proto)
475 {
476   session_t *listener;
477   u64 handle;
478   u32 sm_index;
479   u8 sst;
480
481   sst = session_type_from_proto_and_ip (transport_proto,
482                                         fib_proto == FIB_PROTOCOL_IP4);
483
484   /* *INDENT-OFF* */
485    hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
486      listener = listen_session_get_from_handle (handle);
487      if (listener->session_type == sst && (listener->flags & SESSION_F_PROXY))
488        return listener;
489    }));
490   /* *INDENT-ON* */
491
492   return 0;
493 }
494
495 /**
496  * Send an API message to the external app, to map new segment
497  */
498 int
499 app_worker_add_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
500 {
501   application_t *app = application_get (app_wrk->app_index);
502   return app->cb_fns.add_segment_callback (app_wrk->api_client_index,
503                                            segment_handle);
504 }
505
506 int
507 app_worker_del_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
508 {
509   application_t *app = application_get (app_wrk->app_index);
510   return app->cb_fns.del_segment_callback (app_wrk->api_client_index,
511                                            segment_handle);
512 }
513
514 static inline u8
515 app_worker_application_is_builtin (app_worker_t * app_wrk)
516 {
517   return app_wrk->app_is_builtin;
518 }
519
520 static inline int
521 app_enqueue_evt (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, u8 lock)
522 {
523   if (PREDICT_FALSE (svm_msg_q_is_full (mq)))
524     {
525       clib_warning ("evt q full");
526       svm_msg_q_free_msg (mq, msg);
527       if (lock)
528         svm_msg_q_unlock (mq);
529       return -1;
530     }
531
532   if (lock)
533     {
534       svm_msg_q_add_and_unlock (mq, msg);
535       return 0;
536     }
537
538   /* Even when not locking the ring, we must wait for queue mutex */
539   if (svm_msg_q_add (mq, msg, SVM_Q_WAIT))
540     {
541       clib_warning ("msg q add returned");
542       return -1;
543     }
544   return 0;
545 }
546
547 static inline int
548 app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s, u8 lock)
549 {
550   session_event_t *evt;
551   svm_msg_q_msg_t msg;
552   svm_msg_q_t *mq;
553
554   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY
555                      && s->session_state != SESSION_STATE_LISTENING))
556     return 0;
557
558   if (app_worker_application_is_builtin (app_wrk))
559     return app_worker_builtin_rx (app_wrk, s);
560
561   if (svm_fifo_has_event (s->rx_fifo))
562     return 0;
563
564   mq = app_wrk->event_queue;
565   if (lock)
566     svm_msg_q_lock (mq);
567
568   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
569     {
570       clib_warning ("evt q rings full");
571       if (lock)
572         svm_msg_q_unlock (mq);
573       return -1;
574     }
575
576   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
577   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
578
579   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
580   evt->session_index = s->rx_fifo->client_session_index;
581   evt->event_type = SESSION_IO_EVT_RX;
582
583   (void) svm_fifo_set_event (s->rx_fifo);
584
585   if (app_enqueue_evt (mq, &msg, lock))
586     return -1;
587   return 0;
588 }
589
590 static inline int
591 app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s, u8 lock)
592 {
593   svm_msg_q_t *mq;
594   session_event_t *evt;
595   svm_msg_q_msg_t msg;
596
597   if (app_worker_application_is_builtin (app_wrk))
598     return app_worker_builtin_tx (app_wrk, s);
599
600   mq = app_wrk->event_queue;
601   if (lock)
602     svm_msg_q_lock (mq);
603
604   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
605     {
606       clib_warning ("evt q rings full");
607       if (lock)
608         svm_msg_q_unlock (mq);
609       return -1;
610     }
611
612   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
613   ASSERT (!svm_msg_q_msg_is_invalid (&msg));
614
615   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
616   evt->event_type = SESSION_IO_EVT_TX;
617   evt->session_index = s->tx_fifo->client_session_index;
618
619   return app_enqueue_evt (mq, &msg, lock);
620 }
621
622 /* *INDENT-OFF* */
623 typedef int (app_send_evt_handler_fn) (app_worker_t *app,
624                                        session_t *s,
625                                        u8 lock);
626 static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
627     app_send_io_evt_rx,
628     app_send_io_evt_tx,
629 };
630 /* *INDENT-ON* */
631
632 /**
633  * Send event to application
634  *
635  * Logic from queue perspective is non-blocking. If there's
636  * not enough space to enqueue a message, we return.
637  */
638 int
639 app_worker_send_event (app_worker_t * app, session_t * s, u8 evt_type)
640 {
641   ASSERT (app && evt_type <= SESSION_IO_EVT_TX);
642   return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
643 }
644
645 /**
646  * Send event to application
647  *
648  * Logic from queue perspective is blocking. However, if queue is full,
649  * we return.
650  */
651 int
652 app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
653                                 u8 evt_type)
654 {
655   return app_send_evt_handler_fns[evt_type] (app, s, 1 /* lock */ );
656 }
657
658 u8 *
659 format_app_worker_listener (u8 * s, va_list * args)
660 {
661   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
662   u64 handle = va_arg (*args, u64);
663   u32 sm_index = va_arg (*args, u32);
664   int verbose = va_arg (*args, int);
665   session_t *listener;
666   const u8 *app_name;
667   u8 *str;
668
669   if (!app_wrk)
670     {
671       if (verbose)
672         s = format (s, "%-40s%-25s%=10s%-15s%-15s%-10s", "Connection", "App",
673                     "Wrk", "API Client", "ListenerID", "SegManager");
674       else
675         s = format (s, "%-40s%-25s%=10s", "Connection", "App", "Wrk");
676
677       return s;
678     }
679
680   app_name = application_name_from_index (app_wrk->app_index);
681   listener = listen_session_get_from_handle (handle);
682   str = format (0, "%U", format_session, listener, verbose);
683
684   if (verbose)
685     {
686       char buf[32];
687       sprintf (buf, "%u(%u)", app_wrk->wrk_map_index, app_wrk->wrk_index);
688       s = format (s, "%-40s%-25s%=10s%-15u%-15u%-10u", str, app_name,
689                   buf, app_wrk->api_client_index, handle, sm_index);
690     }
691   else
692     s = format (s, "%-40s%-25s%=10u", str, app_name, app_wrk->wrk_map_index);
693
694   return s;
695 }
696
697 u8 *
698 format_app_worker (u8 * s, va_list * args)
699 {
700   app_worker_t *app_wrk = va_arg (*args, app_worker_t *);
701   u32 indent = 1;
702
703   s = format (s, "%U wrk-index %u app-index %u map-index %u "
704               "api-client-index %d\n", format_white_space, indent,
705               app_wrk->wrk_index, app_wrk->app_index, app_wrk->wrk_map_index,
706               app_wrk->api_client_index);
707   return s;
708 }
709
710 void
711 app_worker_format_connects (app_worker_t * app_wrk, int verbose)
712 {
713   segment_manager_t *sm;
714
715   /* Header */
716   if (!app_wrk)
717     {
718       segment_manager_format_sessions (0, verbose);
719       return;
720     }
721
722   if (app_wrk->connects_seg_manager == (u32) ~ 0)
723     return;
724
725   sm = segment_manager_get (app_wrk->connects_seg_manager);
726   segment_manager_format_sessions (sm, verbose);
727 }
728
729 /*
730  * fd.io coding-style-patch-verification: ON
731  *
732  * Local Variables:
733  * eval: (c-set-style "gnu")
734  * End:
735  */