session: cleanup session event types
[vpp.git] / src / vnet / session / session.c
1 /*
2  * Copyright (c) 2017-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  * @file
17  * @brief Session and session manager
18  */
19
20 #include <vnet/session/session.h>
21 #include <vnet/session/session_debug.h>
22 #include <vnet/session/application.h>
23 #include <vnet/dpo/load_balance.h>
24 #include <vnet/fib/ip4_fib.h>
25
26 session_manager_main_t session_manager_main;
27
28 static inline int
29 session_send_evt_to_thread (void *data, void *args, u32 thread_index,
30                             session_evt_type_t evt_type)
31 {
32   session_event_t *evt;
33   svm_msg_q_msg_t msg;
34   svm_msg_q_t *mq;
35   u32 tries = 0, max_tries;
36
37   mq = session_manager_get_vpp_event_queue (thread_index);
38   while (svm_msg_q_try_lock (mq))
39     {
40       max_tries = vlib_get_current_process (vlib_get_main ())? 1e6 : 3;
41       if (tries++ == max_tries)
42         {
43           SESSION_DBG ("failed to enqueue evt");
44           return -1;
45         }
46     }
47   if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
48     {
49       svm_msg_q_unlock (mq);
50       return -2;
51     }
52   msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
53   if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (&msg)))
54     {
55       svm_msg_q_unlock (mq);
56       return -2;
57     }
58   evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
59   evt->event_type = evt_type;
60   switch (evt_type)
61     {
62     case SESSION_CTRL_EVT_RPC:
63       evt->rpc_args.fp = data;
64       evt->rpc_args.arg = args;
65       break;
66     case SESSION_IO_EVT_TX:
67     case SESSION_IO_EVT_TX_FLUSH:
68     case SESSION_IO_EVT_BUILTIN_RX:
69       evt->fifo = data;
70       break;
71     case SESSION_IO_EVT_BUILTIN_TX:
72     case SESSION_CTRL_EVT_CLOSE:
73       evt->session_handle = session_handle ((session_t *) data);
74       break;
75     default:
76       clib_warning ("evt unhandled!");
77       svm_msg_q_unlock (mq);
78       return -1;
79     }
80
81   svm_msg_q_add_and_unlock (mq, &msg);
82   return 0;
83 }
84
85 int
86 session_send_io_evt_to_thread (svm_fifo_t * f, session_evt_type_t evt_type)
87 {
88   return session_send_evt_to_thread (f, 0, f->master_thread_index, evt_type);
89 }
90
91 int
92 session_send_io_evt_to_thread_custom (void *data, u32 thread_index,
93                                       session_evt_type_t evt_type)
94 {
95   return session_send_evt_to_thread (data, 0, thread_index, evt_type);
96 }
97
98 int
99 session_send_ctrl_evt_to_thread (session_t * s, session_evt_type_t evt_type)
100 {
101   /* only event supported for now is disconnect */
102   ASSERT (evt_type == SESSION_CTRL_EVT_CLOSE);
103   return session_send_evt_to_thread (s, 0, s->thread_index,
104                                      SESSION_CTRL_EVT_CLOSE);
105 }
106
107 void
108 session_send_rpc_evt_to_thread (u32 thread_index, void *fp, void *rpc_args)
109 {
110   if (thread_index != vlib_get_thread_index ())
111     session_send_evt_to_thread (fp, rpc_args, thread_index,
112                                 SESSION_CTRL_EVT_RPC);
113   else
114     {
115       void (*fnp) (void *) = fp;
116       fnp (rpc_args);
117     }
118 }
119
120 static void
121 session_program_transport_close (session_t * s)
122 {
123   u32 thread_index = vlib_get_thread_index ();
124   session_manager_worker_t *wrk;
125   session_event_t *evt;
126
127   if (!session_has_transport (s))
128     {
129       /* Polling may not be enabled on main thread so close now */
130       session_transport_close (s);
131       return;
132     }
133
134   /* If we are in the handler thread, or being called with the worker barrier
135    * held, just append a new event to pending disconnects vector. */
136   if (vlib_thread_is_main_w_barrier () || thread_index == s->thread_index)
137     {
138       wrk = session_manager_get_worker (s->thread_index);
139       vec_add2 (wrk->pending_disconnects, evt, 1);
140       clib_memset (evt, 0, sizeof (*evt));
141       evt->session_handle = session_handle (s);
142       evt->event_type = SESSION_CTRL_EVT_CLOSE;
143     }
144   else
145     session_send_ctrl_evt_to_thread (s, SESSION_CTRL_EVT_CLOSE);
146 }
147
148 session_t *
149 session_alloc (u32 thread_index)
150 {
151   session_manager_worker_t *wrk = &session_manager_main.wrk[thread_index];
152   session_t *s;
153   u8 will_expand = 0;
154   pool_get_aligned_will_expand (wrk->sessions, will_expand,
155                                 CLIB_CACHE_LINE_BYTES);
156   /* If we have peekers, let them finish */
157   if (PREDICT_FALSE (will_expand && vlib_num_workers ()))
158     {
159       clib_rwlock_writer_lock (&wrk->peekers_rw_locks);
160       pool_get_aligned (wrk->sessions, s, CLIB_CACHE_LINE_BYTES);
161       clib_rwlock_writer_unlock (&wrk->peekers_rw_locks);
162     }
163   else
164     {
165       pool_get_aligned (wrk->sessions, s, CLIB_CACHE_LINE_BYTES);
166     }
167   clib_memset (s, 0, sizeof (*s));
168   s->session_index = s - wrk->sessions;
169   s->thread_index = thread_index;
170   return s;
171 }
172
173 void
174 session_free (session_t * s)
175 {
176   pool_put (session_manager_main.wrk[s->thread_index].sessions, s);
177   if (CLIB_DEBUG)
178     clib_memset (s, 0xFA, sizeof (*s));
179 }
180
181 void
182 session_free_w_fifos (session_t * s)
183 {
184   segment_manager_dealloc_fifos (s->svm_segment_index, s->rx_fifo,
185                                  s->tx_fifo);
186   session_free (s);
187 }
188
189 /**
190  * Cleans up session and lookup table.
191  *
192  * Transport connection must still be valid.
193  */
194 static void
195 session_delete (session_t * s)
196 {
197   int rv;
198
199   /* Delete from the main lookup table. */
200   if ((rv = session_lookup_del_session (s)))
201     clib_warning ("hash delete error, rv %d", rv);
202
203   session_free_w_fifos (s);
204 }
205
206 static session_t *
207 session_alloc_for_connection (transport_connection_t * tc)
208 {
209   session_t *s;
210   u32 thread_index = tc->thread_index;
211
212   ASSERT (thread_index == vlib_get_thread_index ()
213           || transport_protocol_is_cl (tc->proto));
214
215   s = session_alloc (thread_index);
216   s->session_type = session_type_from_proto_and_ip (tc->proto, tc->is_ip4);
217   s->enqueue_epoch = (u64) ~ 0;
218   s->session_state = SESSION_STATE_CLOSED;
219
220   /* Attach transport to session and vice versa */
221   s->connection_index = tc->c_index;
222   tc->s_index = s->session_index;
223   return s;
224 }
225
226 /**
227  * Discards bytes from buffer chain
228  *
229  * It discards n_bytes_to_drop starting at first buffer after chain_b
230  */
231 always_inline void
232 session_enqueue_discard_chain_bytes (vlib_main_t * vm, vlib_buffer_t * b,
233                                      vlib_buffer_t ** chain_b,
234                                      u32 n_bytes_to_drop)
235 {
236   vlib_buffer_t *next = *chain_b;
237   u32 to_drop = n_bytes_to_drop;
238   ASSERT (b->flags & VLIB_BUFFER_NEXT_PRESENT);
239   while (to_drop && (next->flags & VLIB_BUFFER_NEXT_PRESENT))
240     {
241       next = vlib_get_buffer (vm, next->next_buffer);
242       if (next->current_length > to_drop)
243         {
244           vlib_buffer_advance (next, to_drop);
245           to_drop = 0;
246         }
247       else
248         {
249           to_drop -= next->current_length;
250           next->current_length = 0;
251         }
252     }
253   *chain_b = next;
254
255   if (to_drop == 0)
256     b->total_length_not_including_first_buffer -= n_bytes_to_drop;
257 }
258
259 /**
260  * Enqueue buffer chain tail
261  */
262 always_inline int
263 session_enqueue_chain_tail (session_t * s, vlib_buffer_t * b,
264                             u32 offset, u8 is_in_order)
265 {
266   vlib_buffer_t *chain_b;
267   u32 chain_bi, len, diff;
268   vlib_main_t *vm = vlib_get_main ();
269   u8 *data;
270   u32 written = 0;
271   int rv = 0;
272
273   if (is_in_order && offset)
274     {
275       diff = offset - b->current_length;
276       if (diff > b->total_length_not_including_first_buffer)
277         return 0;
278       chain_b = b;
279       session_enqueue_discard_chain_bytes (vm, b, &chain_b, diff);
280       chain_bi = vlib_get_buffer_index (vm, chain_b);
281     }
282   else
283     chain_bi = b->next_buffer;
284
285   do
286     {
287       chain_b = vlib_get_buffer (vm, chain_bi);
288       data = vlib_buffer_get_current (chain_b);
289       len = chain_b->current_length;
290       if (!len)
291         continue;
292       if (is_in_order)
293         {
294           rv = svm_fifo_enqueue_nowait (s->rx_fifo, len, data);
295           if (rv == len)
296             {
297               written += rv;
298             }
299           else if (rv < len)
300             {
301               return (rv > 0) ? (written + rv) : written;
302             }
303           else if (rv > len)
304             {
305               written += rv;
306
307               /* written more than what was left in chain */
308               if (written > b->total_length_not_including_first_buffer)
309                 return written;
310
311               /* drop the bytes that have already been delivered */
312               session_enqueue_discard_chain_bytes (vm, b, &chain_b, rv - len);
313             }
314         }
315       else
316         {
317           rv = svm_fifo_enqueue_with_offset (s->rx_fifo, offset, len, data);
318           if (rv)
319             {
320               clib_warning ("failed to enqueue multi-buffer seg");
321               return -1;
322             }
323           offset += len;
324         }
325     }
326   while ((chain_bi = (chain_b->flags & VLIB_BUFFER_NEXT_PRESENT)
327           ? chain_b->next_buffer : 0));
328
329   if (is_in_order)
330     return written;
331
332   return 0;
333 }
334
335 /*
336  * Enqueue data for delivery to session peer. Does not notify peer of enqueue
337  * event but on request can queue notification events for later delivery by
338  * calling stream_server_flush_enqueue_events().
339  *
340  * @param tc Transport connection which is to be enqueued data
341  * @param b Buffer to be enqueued
342  * @param offset Offset at which to start enqueueing if out-of-order
343  * @param queue_event Flag to indicate if peer is to be notified or if event
344  *                    is to be queued. The former is useful when more data is
345  *                    enqueued and only one event is to be generated.
346  * @param is_in_order Flag to indicate if data is in order
347  * @return Number of bytes enqueued or a negative value if enqueueing failed.
348  */
349 int
350 session_enqueue_stream_connection (transport_connection_t * tc,
351                                    vlib_buffer_t * b, u32 offset,
352                                    u8 queue_event, u8 is_in_order)
353 {
354   session_t *s;
355   int enqueued = 0, rv, in_order_off;
356
357   s = session_get (tc->s_index, tc->thread_index);
358
359   if (is_in_order)
360     {
361       enqueued = svm_fifo_enqueue_nowait (s->rx_fifo,
362                                           b->current_length,
363                                           vlib_buffer_get_current (b));
364       if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT)
365                          && enqueued >= 0))
366         {
367           in_order_off = enqueued > b->current_length ? enqueued : 0;
368           rv = session_enqueue_chain_tail (s, b, in_order_off, 1);
369           if (rv > 0)
370             enqueued += rv;
371         }
372     }
373   else
374     {
375       rv = svm_fifo_enqueue_with_offset (s->rx_fifo, offset,
376                                          b->current_length,
377                                          vlib_buffer_get_current (b));
378       if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && !rv))
379         session_enqueue_chain_tail (s, b, offset + b->current_length, 0);
380       /* if something was enqueued, report even this as success for ooo
381        * segment handling */
382       return rv;
383     }
384
385   if (queue_event)
386     {
387       /* Queue RX event on this fifo. Eventually these will need to be flushed
388        * by calling stream_server_flush_enqueue_events () */
389       session_manager_worker_t *wrk;
390
391       wrk = session_manager_get_worker (s->thread_index);
392       if (s->enqueue_epoch != wrk->current_enqueue_epoch[tc->proto])
393         {
394           s->enqueue_epoch = wrk->current_enqueue_epoch[tc->proto];
395           vec_add1 (wrk->session_to_enqueue[tc->proto], s->session_index);
396         }
397     }
398
399   return enqueued;
400 }
401
402 int
403 session_enqueue_dgram_connection (session_t * s,
404                                   session_dgram_hdr_t * hdr,
405                                   vlib_buffer_t * b, u8 proto, u8 queue_event)
406 {
407   int enqueued = 0, rv, in_order_off;
408
409   ASSERT (svm_fifo_max_enqueue (s->rx_fifo)
410           >= b->current_length + sizeof (*hdr));
411
412   svm_fifo_enqueue_nowait (s->rx_fifo, sizeof (session_dgram_hdr_t),
413                            (u8 *) hdr);
414   enqueued = svm_fifo_enqueue_nowait (s->rx_fifo, b->current_length,
415                                       vlib_buffer_get_current (b));
416   if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && enqueued >= 0))
417     {
418       in_order_off = enqueued > b->current_length ? enqueued : 0;
419       rv = session_enqueue_chain_tail (s, b, in_order_off, 1);
420       if (rv > 0)
421         enqueued += rv;
422     }
423   if (queue_event)
424     {
425       /* Queue RX event on this fifo. Eventually these will need to be flushed
426        * by calling stream_server_flush_enqueue_events () */
427       session_manager_worker_t *wrk;
428
429       wrk = session_manager_get_worker (s->thread_index);
430       if (s->enqueue_epoch != wrk->current_enqueue_epoch[proto])
431         {
432           s->enqueue_epoch = wrk->current_enqueue_epoch[proto];
433           vec_add1 (wrk->session_to_enqueue[proto], s->session_index);
434         }
435     }
436   return enqueued;
437 }
438
439 /** Check if we have space in rx fifo to push more bytes */
440 u8
441 stream_session_no_space (transport_connection_t * tc, u32 thread_index,
442                          u16 data_len)
443 {
444   session_t *s = session_get (tc->s_index, thread_index);
445
446   if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY))
447     return 1;
448
449   if (data_len > svm_fifo_max_enqueue (s->rx_fifo))
450     return 1;
451
452   return 0;
453 }
454
455 u32
456 session_tx_fifo_max_dequeue (transport_connection_t * tc)
457 {
458   session_t *s = session_get (tc->s_index, tc->thread_index);
459   if (!s->tx_fifo)
460     return 0;
461   return svm_fifo_max_dequeue (s->tx_fifo);
462 }
463
464 int
465 stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
466                            u32 offset, u32 max_bytes)
467 {
468   session_t *s = session_get (tc->s_index, tc->thread_index);
469   return svm_fifo_peek (s->tx_fifo, offset, max_bytes, buffer);
470 }
471
472 u32
473 session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
474 {
475   session_t *s = session_get (tc->s_index, tc->thread_index);
476   return svm_fifo_dequeue_drop (s->tx_fifo, max_bytes);
477 }
478
479 static inline int
480 session_notify_subscribers (u32 app_index, session_t * s,
481                             svm_fifo_t * f, session_evt_type_t evt_type)
482 {
483   app_worker_t *app_wrk;
484   application_t *app;
485   int i;
486
487   app = application_get (app_index);
488   if (!app)
489     return -1;
490
491   for (i = 0; i < f->n_subscribers; i++)
492     {
493       app_wrk = application_get_worker (app, f->subscribers[i]);
494       if (!app_wrk)
495         continue;
496       if (app_worker_lock_and_send_event (app_wrk, s, evt_type))
497         return -1;
498     }
499
500   return 0;
501 }
502
503 /**
504  * Notify session peer that new data has been enqueued.
505  *
506  * @param s     Stream session for which the event is to be generated.
507  * @param lock  Flag to indicate if call should lock message queue.
508  *
509  * @return 0 on success or negative number if failed to send notification.
510  */
511 static inline int
512 session_enqueue_notify (session_t * s)
513 {
514   app_worker_t *app_wrk;
515
516   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
517   if (PREDICT_FALSE (!app_wrk))
518     {
519       SESSION_DBG ("invalid s->app_index = %d", s->app_wrk_index);
520       return 0;
521     }
522
523   /* *INDENT-OFF* */
524   SESSION_EVT_DBG(SESSION_EVT_ENQ, s, ({
525       ed->data[0] = SESSION_IO_EVT_RX;
526       ed->data[1] = svm_fifo_max_dequeue (s->rx_fifo);
527   }));
528   /* *INDENT-ON* */
529
530   if (PREDICT_FALSE (app_worker_lock_and_send_event (app_wrk, s,
531                                                      SESSION_IO_EVT_RX)))
532     return -1;
533
534   if (PREDICT_FALSE (svm_fifo_n_subscribers (s->rx_fifo)))
535     return session_notify_subscribers (app_wrk->app_index, s,
536                                        s->rx_fifo, SESSION_IO_EVT_RX);
537
538   return 0;
539 }
540
541 int
542 session_dequeue_notify (session_t * s)
543 {
544   app_worker_t *app_wrk;
545
546   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
547   if (PREDICT_FALSE (!app_wrk))
548     return -1;
549
550   if (PREDICT_FALSE (app_worker_lock_and_send_event (app_wrk, s,
551                                                      SESSION_IO_EVT_TX)))
552     return -1;
553
554   if (PREDICT_FALSE (s->tx_fifo->n_subscribers))
555     return session_notify_subscribers (app_wrk->app_index, s,
556                                        s->tx_fifo, SESSION_IO_EVT_TX);
557
558   svm_fifo_clear_tx_ntf (s->tx_fifo);
559
560   return 0;
561 }
562
563 /**
564  * Flushes queue of sessions that are to be notified of new data
565  * enqueued events.
566  *
567  * @param thread_index Thread index for which the flush is to be performed.
568  * @return 0 on success or a positive number indicating the number of
569  *         failures due to API queue being full.
570  */
571 int
572 session_manager_flush_enqueue_events (u8 transport_proto, u32 thread_index)
573 {
574   session_manager_worker_t *wrk = session_manager_get_worker (thread_index);
575   session_t *s;
576   int i, errors = 0;
577   u32 *indices;
578
579   indices = wrk->session_to_enqueue[transport_proto];
580
581   for (i = 0; i < vec_len (indices); i++)
582     {
583       s = session_get_if_valid (indices[i], thread_index);
584       if (PREDICT_FALSE (!s))
585         {
586           errors++;
587           continue;
588         }
589       if (PREDICT_FALSE (session_enqueue_notify (s)))
590         errors++;
591     }
592
593   vec_reset_length (indices);
594   wrk->session_to_enqueue[transport_proto] = indices;
595   wrk->current_enqueue_epoch[transport_proto]++;
596
597   return errors;
598 }
599
600 int
601 session_manager_flush_all_enqueue_events (u8 transport_proto)
602 {
603   vlib_thread_main_t *vtm = vlib_get_thread_main ();
604   int i, errors = 0;
605   for (i = 0; i < 1 + vtm->n_threads; i++)
606     errors += session_manager_flush_enqueue_events (transport_proto, i);
607   return errors;
608 }
609
610 /**
611  * Init fifo tail and head pointers
612  *
613  * Useful if transport uses absolute offsets for tracking ooo segments.
614  */
615 void
616 stream_session_init_fifos_pointers (transport_connection_t * tc,
617                                     u32 rx_pointer, u32 tx_pointer)
618 {
619   session_t *s;
620   s = session_get (tc->s_index, tc->thread_index);
621   svm_fifo_init_pointers (s->rx_fifo, rx_pointer);
622   svm_fifo_init_pointers (s->tx_fifo, tx_pointer);
623 }
624
625 int
626 session_stream_connect_notify (transport_connection_t * tc, u8 is_fail)
627 {
628   u32 opaque = 0, new_ti, new_si;
629   app_worker_t *app_wrk;
630   session_t *s = 0;
631   u64 ho_handle;
632
633   /*
634    * Find connection handle and cleanup half-open table
635    */
636   ho_handle = session_lookup_half_open_handle (tc);
637   if (ho_handle == HALF_OPEN_LOOKUP_INVALID_VALUE)
638     {
639       SESSION_DBG ("half-open was removed!");
640       return -1;
641     }
642   session_lookup_del_half_open (tc);
643
644   /* Get the app's index from the handle we stored when opening connection
645    * and the opaque (api_context for external apps) from transport session
646    * index */
647   app_wrk = app_worker_get_if_valid (ho_handle >> 32);
648   if (!app_wrk)
649     return -1;
650
651   opaque = tc->s_index;
652
653   if (is_fail)
654     return app_worker_connect_notify (app_wrk, s, opaque);
655
656   s = session_alloc_for_connection (tc);
657   s->session_state = SESSION_STATE_CONNECTING;
658   s->app_wrk_index = app_wrk->wrk_index;
659   new_si = s->session_index;
660   new_ti = s->thread_index;
661
662   if (app_worker_init_connected (app_wrk, s))
663     {
664       session_free (s);
665       app_worker_connect_notify (app_wrk, 0, opaque);
666       return -1;
667     }
668
669   if (app_worker_connect_notify (app_wrk, s, opaque))
670     {
671       s = session_get (new_si, new_ti);
672       session_free_w_fifos (s);
673       return -1;
674     }
675
676   s = session_get (new_si, new_ti);
677   s->session_state = SESSION_STATE_READY;
678   session_lookup_add_connection (tc, session_handle (s));
679
680   return 0;
681 }
682
683 typedef struct _session_switch_pool_args
684 {
685   u32 session_index;
686   u32 thread_index;
687   u32 new_thread_index;
688   u32 new_session_index;
689 } session_switch_pool_args_t;
690
691 static void
692 session_switch_pool (void *cb_args)
693 {
694   session_switch_pool_args_t *args = (session_switch_pool_args_t *) cb_args;
695   session_t *s;
696   ASSERT (args->thread_index == vlib_get_thread_index ());
697   s = session_get (args->session_index, args->thread_index);
698   s->tx_fifo->master_session_index = args->new_session_index;
699   s->tx_fifo->master_thread_index = args->new_thread_index;
700   transport_cleanup (session_get_transport_proto (s), s->connection_index,
701                      s->thread_index);
702   session_free (s);
703   clib_mem_free (cb_args);
704 }
705
706 /**
707  * Move dgram session to the right thread
708  */
709 int
710 session_dgram_connect_notify (transport_connection_t * tc,
711                               u32 old_thread_index, session_t ** new_session)
712 {
713   session_t *new_s;
714   session_switch_pool_args_t *rpc_args;
715
716   /*
717    * Clone half-open session to the right thread.
718    */
719   new_s = session_clone_safe (tc->s_index, old_thread_index);
720   new_s->connection_index = tc->c_index;
721   new_s->rx_fifo->master_session_index = new_s->session_index;
722   new_s->rx_fifo->master_thread_index = new_s->thread_index;
723   new_s->session_state = SESSION_STATE_READY;
724   session_lookup_add_connection (tc, session_handle (new_s));
725
726   /*
727    * Ask thread owning the old session to clean it up and make us the tx
728    * fifo owner
729    */
730   rpc_args = clib_mem_alloc (sizeof (*rpc_args));
731   rpc_args->new_session_index = new_s->session_index;
732   rpc_args->new_thread_index = new_s->thread_index;
733   rpc_args->session_index = tc->s_index;
734   rpc_args->thread_index = old_thread_index;
735   session_send_rpc_evt_to_thread (rpc_args->thread_index, session_switch_pool,
736                                   rpc_args);
737
738   tc->s_index = new_s->session_index;
739   new_s->connection_index = tc->c_index;
740   *new_session = new_s;
741   return 0;
742 }
743
744 /**
745  * Notification from transport that connection is being closed.
746  *
747  * A disconnect is sent to application but state is not removed. Once
748  * disconnect is acknowledged by application, session disconnect is called.
749  * Ultimately this leads to close being called on transport (passive close).
750  */
751 void
752 session_transport_closing_notify (transport_connection_t * tc)
753 {
754   app_worker_t *app_wrk;
755   application_t *app;
756   session_t *s;
757
758   s = session_get (tc->s_index, tc->thread_index);
759   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
760     return;
761   s->session_state = SESSION_STATE_TRANSPORT_CLOSING;
762   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
763   if (!app_wrk)
764     return;
765   app = application_get (app_wrk->app_index);
766   app->cb_fns.session_disconnect_callback (s);
767 }
768
769 /**
770  * Notification from transport that connection is being deleted
771  *
772  * This removes the session if it is still valid. It should be called only on
773  * previously fully established sessions. For instance failed connects should
774  * call stream_session_connect_notify and indicate that the connect has
775  * failed.
776  */
777 void
778 session_transport_delete_notify (transport_connection_t * tc)
779 {
780   session_t *s;
781
782   /* App might've been removed already */
783   if (!(s = session_get_if_valid (tc->s_index, tc->thread_index)))
784     return;
785
786   /* Make sure we don't try to send anything more */
787   svm_fifo_dequeue_drop_all (s->tx_fifo);
788
789   switch (s->session_state)
790     {
791     case SESSION_STATE_CREATED:
792       /* Session was created but accept notification was not yet sent to the
793        * app. Cleanup everything. */
794       session_lookup_del_session (s);
795       session_free_w_fifos (s);
796       break;
797     case SESSION_STATE_ACCEPTING:
798     case SESSION_STATE_TRANSPORT_CLOSING:
799       /* If transport finishes or times out before we get a reply
800        * from the app, mark transport as closed and wait for reply
801        * before removing the session. Cleanup session table in advance
802        * because transport will soon be closed and closed sessions
803        * are assumed to have been removed from the lookup table */
804       session_lookup_del_session (s);
805       s->session_state = SESSION_STATE_TRANSPORT_CLOSED;
806       break;
807     case SESSION_STATE_CLOSING:
808     case SESSION_STATE_CLOSED_WAITING:
809       /* Cleanup lookup table as transport needs to still be valid.
810        * Program transport close to ensure that all session events
811        * have been cleaned up. Once transport close is called, the
812        * session is just removed because both transport and app have
813        * confirmed the close*/
814       session_lookup_del_session (s);
815       s->session_state = SESSION_STATE_TRANSPORT_CLOSED;
816       session_program_transport_close (s);
817       break;
818     case SESSION_STATE_TRANSPORT_CLOSED:
819       break;
820     case SESSION_STATE_CLOSED:
821       session_delete (s);
822       break;
823     default:
824       clib_warning ("session state %u", s->session_state);
825       session_delete (s);
826       break;
827     }
828 }
829
830 /**
831  * Notification from transport that session can be closed
832  *
833  * Should be called by transport only if it was closed with non-empty
834  * tx fifo and once it decides to begin the closing procedure prior to
835  * issuing a delete notify. This gives the chance to the session layer
836  * to cleanup any outstanding events.
837  */
838 void
839 session_transport_closed_notify (transport_connection_t * tc)
840 {
841   session_t *s;
842
843   if (!(s = session_get_if_valid (tc->s_index, tc->thread_index)))
844     return;
845
846   /* If app close has not been received or has not yet resulted in
847    * a transport close, only mark the session transport as closed */
848   if (s->session_state <= SESSION_STATE_CLOSING)
849     {
850       session_lookup_del_session (s);
851       s->session_state = SESSION_STATE_TRANSPORT_CLOSED;
852     }
853   else
854     s->session_state = SESSION_STATE_CLOSED;
855 }
856
857 /**
858  * Notify application that connection has been reset.
859  */
860 void
861 session_transport_reset_notify (transport_connection_t * tc)
862 {
863   session_t *s;
864   app_worker_t *app_wrk;
865   application_t *app;
866   s = session_get (tc->s_index, tc->thread_index);
867   svm_fifo_dequeue_drop_all (s->tx_fifo);
868   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
869     return;
870   s->session_state = SESSION_STATE_TRANSPORT_CLOSING;
871   app_wrk = app_worker_get (s->app_wrk_index);
872   app = application_get (app_wrk->app_index);
873   app->cb_fns.session_reset_callback (s);
874 }
875
876 int
877 session_stream_accept_notify (transport_connection_t * tc)
878 {
879   app_worker_t *app_wrk;
880   session_t *s;
881
882   s = session_get (tc->s_index, tc->thread_index);
883   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
884   if (!app_wrk)
885     return -1;
886   s->session_state = SESSION_STATE_ACCEPTING;
887   return app_worker_accept_notify (app_wrk, s);
888 }
889
890 /**
891  * Accept a stream session. Optionally ping the server by callback.
892  */
893 int
894 session_stream_accept (transport_connection_t * tc, u32 listener_index,
895                        u8 notify)
896 {
897   session_t *s;
898   int rv;
899
900   s = session_alloc_for_connection (tc);
901   s->listener_index = listener_index;
902   s->session_state = SESSION_STATE_CREATED;
903
904   if ((rv = app_worker_init_accepted (s)))
905     return rv;
906
907   session_lookup_add_connection (tc, session_handle (s));
908
909   /* Shoulder-tap the server */
910   if (notify)
911     {
912       app_worker_t *app_wrk = app_worker_get (s->app_wrk_index);
913       return app_worker_accept_notify (app_wrk, s);
914     }
915
916   return 0;
917 }
918
919 int
920 session_open_cl (u32 app_wrk_index, session_endpoint_t * rmt, u32 opaque)
921 {
922   transport_connection_t *tc;
923   transport_endpoint_cfg_t *tep;
924   app_worker_t *app_wrk;
925   session_t *s;
926   int rv;
927
928   tep = session_endpoint_to_transport_cfg (rmt);
929   rv = transport_connect (rmt->transport_proto, tep);
930   if (rv < 0)
931     {
932       SESSION_DBG ("Transport failed to open connection.");
933       return VNET_API_ERROR_SESSION_CONNECT;
934     }
935
936   tc = transport_get_half_open (rmt->transport_proto, (u32) rv);
937
938   /* For dgram type of service, allocate session and fifos now */
939   app_wrk = app_worker_get (app_wrk_index);
940   s = session_alloc_for_connection (tc);
941   s->app_wrk_index = app_wrk->wrk_index;
942   s->session_state = SESSION_STATE_OPENED;
943   if (app_worker_init_connected (app_wrk, s))
944     {
945       session_free (s);
946       return -1;
947     }
948
949   return app_worker_connect_notify (app_wrk, s, opaque);
950 }
951
952 int
953 session_open_vc (u32 app_wrk_index, session_endpoint_t * rmt, u32 opaque)
954 {
955   transport_connection_t *tc;
956   transport_endpoint_cfg_t *tep;
957   u64 handle;
958   int rv;
959
960   tep = session_endpoint_to_transport_cfg (rmt);
961   rv = transport_connect (rmt->transport_proto, tep);
962   if (rv < 0)
963     {
964       SESSION_DBG ("Transport failed to open connection.");
965       return VNET_API_ERROR_SESSION_CONNECT;
966     }
967
968   tc = transport_get_half_open (rmt->transport_proto, (u32) rv);
969
970   /* If transport offers a stream service, only allocate session once the
971    * connection has been established.
972    * Add connection to half-open table and save app and tc index. The
973    * latter is needed to help establish the connection while the former
974    * is needed when the connect notify comes and we have to notify the
975    * external app
976    */
977   handle = (((u64) app_wrk_index) << 32) | (u64) tc->c_index;
978   session_lookup_add_half_open (tc, handle);
979
980   /* Store api_context (opaque) for when the reply comes. Not the nicest
981    * thing but better than allocating a separate half-open pool.
982    */
983   tc->s_index = opaque;
984   return 0;
985 }
986
987 int
988 session_open_app (u32 app_wrk_index, session_endpoint_t * rmt, u32 opaque)
989 {
990   session_endpoint_cfg_t *sep = (session_endpoint_cfg_t *) rmt;
991   transport_endpoint_cfg_t *tep_cfg = session_endpoint_to_transport_cfg (sep);
992
993   sep->app_wrk_index = app_wrk_index;
994   sep->opaque = opaque;
995
996   return transport_connect (rmt->transport_proto, tep_cfg);
997 }
998
999 typedef int (*session_open_service_fn) (u32, session_endpoint_t *, u32);
1000
1001 /* *INDENT-OFF* */
1002 static session_open_service_fn session_open_srv_fns[TRANSPORT_N_SERVICES] = {
1003   session_open_vc,
1004   session_open_cl,
1005   session_open_app,
1006 };
1007 /* *INDENT-ON* */
1008
1009 /**
1010  * Ask transport to open connection to remote transport endpoint.
1011  *
1012  * Stores handle for matching request with reply since the call can be
1013  * asynchronous. For instance, for TCP the 3-way handshake must complete
1014  * before reply comes. Session is only created once connection is established.
1015  *
1016  * @param app_index Index of the application requesting the connect
1017  * @param st Session type requested.
1018  * @param tep Remote transport endpoint
1019  * @param opaque Opaque data (typically, api_context) the application expects
1020  *               on open completion.
1021  */
1022 int
1023 session_open (u32 app_wrk_index, session_endpoint_t * rmt, u32 opaque)
1024 {
1025   transport_service_type_t tst;
1026   tst = transport_protocol_service_type (rmt->transport_proto);
1027   return session_open_srv_fns[tst] (app_wrk_index, rmt, opaque);
1028 }
1029
1030 /**
1031  * Ask transport to listen on session endpoint.
1032  *
1033  * @param s Session for which listen will be called. Note that unlike
1034  *          established sessions, listen sessions are not associated to a
1035  *          thread.
1036  * @param sep Local endpoint to be listened on.
1037  */
1038 int
1039 session_listen (session_t * ls, session_endpoint_cfg_t * sep)
1040 {
1041   transport_endpoint_t *tep;
1042   u32 tc_index, s_index;
1043
1044   /* Transport bind/listen */
1045   tep = session_endpoint_to_transport (sep);
1046   s_index = ls->session_index;
1047   tc_index = transport_start_listen (session_get_transport_proto (ls),
1048                                      s_index, tep);
1049
1050   if (tc_index == (u32) ~ 0)
1051     return -1;
1052
1053   /* Attach transport to session. Lookup tables are populated by the app
1054    * worker because local tables (for ct sessions) are not backed by a fib */
1055   ls = listen_session_get (s_index);
1056   ls->connection_index = tc_index;
1057
1058   return 0;
1059 }
1060
1061 /**
1062  * Ask transport to stop listening on local transport endpoint.
1063  *
1064  * @param s Session to stop listening on. It must be in state LISTENING.
1065  */
1066 int
1067 session_stop_listen (session_t * s)
1068 {
1069   transport_proto_t tp = session_get_transport_proto (s);
1070   transport_connection_t *tc;
1071
1072   if (s->session_state != SESSION_STATE_LISTENING)
1073     return -1;
1074
1075   tc = transport_get_listener (tp, s->connection_index);
1076   if (!tc)
1077     return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
1078
1079   session_lookup_del_connection (tc);
1080   transport_stop_listen (tp, s->connection_index);
1081   return 0;
1082 }
1083
1084 /**
1085  * Initialize session closing procedure.
1086  *
1087  * Request is always sent to session node to ensure that all outstanding
1088  * requests are served before transport is notified.
1089  */
1090 void
1091 session_close (session_t * s)
1092 {
1093   if (!s)
1094     return;
1095
1096   if (s->session_state >= SESSION_STATE_CLOSING)
1097     {
1098       /* Session will only be removed once both app and transport
1099        * acknowledge the close */
1100       if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED)
1101         session_program_transport_close (s);
1102
1103       /* Session already closed. Clear the tx fifo */
1104       if (s->session_state == SESSION_STATE_CLOSED)
1105         svm_fifo_dequeue_drop_all (s->tx_fifo);
1106       return;
1107     }
1108
1109   s->session_state = SESSION_STATE_CLOSING;
1110   session_program_transport_close (s);
1111 }
1112
1113 /**
1114  * Notify transport the session can be disconnected. This should eventually
1115  * result in a delete notification that allows us to cleanup session state.
1116  * Called for both active/passive disconnects.
1117  *
1118  * Must be called from the session's thread.
1119  */
1120 void
1121 session_transport_close (session_t * s)
1122 {
1123   /* If transport is already closed, just free the session */
1124   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
1125     {
1126       session_free_w_fifos (s);
1127       return;
1128     }
1129
1130   /* If tx queue wasn't drained, change state to closed waiting for transport.
1131    * This way, the transport, if it so wishes, can continue to try sending the
1132    * outstanding data (in closed state it cannot). It MUST however at one
1133    * point, either after sending everything or after a timeout, call delete
1134    * notify. This will finally lead to the complete cleanup of the session.
1135    */
1136   if (svm_fifo_max_dequeue (s->tx_fifo))
1137     s->session_state = SESSION_STATE_CLOSED_WAITING;
1138   else
1139     s->session_state = SESSION_STATE_CLOSED;
1140
1141   transport_close (session_get_transport_proto (s), s->connection_index,
1142                    s->thread_index);
1143 }
1144
1145 /**
1146  * Cleanup transport and session state.
1147  *
1148  * Notify transport of the cleanup and free the session. This should
1149  * be called only if transport reported some error and is already
1150  * closed.
1151  */
1152 void
1153 session_transport_cleanup (session_t * s)
1154 {
1155   s->session_state = SESSION_STATE_CLOSED;
1156
1157   /* Delete from main lookup table before we axe the the transport */
1158   session_lookup_del_session (s);
1159   transport_cleanup (session_get_transport_proto (s), s->connection_index,
1160                      s->thread_index);
1161   /* Since we called cleanup, no delete notification will come. So, make
1162    * sure the session is properly freed. */
1163   session_free_w_fifos (s);
1164 }
1165
1166 /**
1167  * Allocate event queues in the shared-memory segment
1168  *
1169  * That can either be a newly created memfd segment, that will need to be
1170  * mapped by all stack users, or the binary api's svm region. The latter is
1171  * assumed to be already mapped. NOTE that this assumption DOES NOT hold if
1172  * api clients bootstrap shm api over sockets (i.e. use memfd segments) and
1173  * vpp uses api svm region for event queues.
1174  */
1175 void
1176 session_vpp_event_queues_allocate (session_manager_main_t * smm)
1177 {
1178   u32 evt_q_length = 2048, evt_size = sizeof (session_event_t);
1179   ssvm_private_t *eqs = &smm->evt_qs_segment;
1180   api_main_t *am = &api_main;
1181   uword eqs_size = 64 << 20;
1182   pid_t vpp_pid = getpid ();
1183   void *oldheap;
1184   int i;
1185
1186   if (smm->configured_event_queue_length)
1187     evt_q_length = smm->configured_event_queue_length;
1188
1189   if (smm->evt_qs_use_memfd_seg)
1190     {
1191       if (smm->evt_qs_segment_size)
1192         eqs_size = smm->evt_qs_segment_size;
1193
1194       eqs->ssvm_size = eqs_size;
1195       eqs->i_am_master = 1;
1196       eqs->my_pid = vpp_pid;
1197       eqs->name = format (0, "%s%c", "evt-qs-segment", 0);
1198       eqs->requested_va = smm->session_baseva;
1199
1200       if (ssvm_master_init (eqs, SSVM_SEGMENT_MEMFD))
1201         {
1202           clib_warning ("failed to initialize queue segment");
1203           return;
1204         }
1205     }
1206
1207   if (smm->evt_qs_use_memfd_seg)
1208     oldheap = ssvm_push_heap (eqs->sh);
1209   else
1210     oldheap = svm_push_data_heap (am->vlib_rp);
1211
1212   for (i = 0; i < vec_len (smm->wrk); i++)
1213     {
1214       svm_msg_q_cfg_t _cfg, *cfg = &_cfg;
1215       svm_msg_q_ring_cfg_t rc[SESSION_MQ_N_RINGS] = {
1216         {evt_q_length, evt_size, 0}
1217         ,
1218         {evt_q_length >> 1, 256, 0}
1219       };
1220       cfg->consumer_pid = 0;
1221       cfg->n_rings = 2;
1222       cfg->q_nitems = evt_q_length;
1223       cfg->ring_cfgs = rc;
1224       smm->wrk[i].vpp_event_queue = svm_msg_q_alloc (cfg);
1225       if (smm->evt_qs_use_memfd_seg)
1226         {
1227           if (svm_msg_q_alloc_consumer_eventfd (smm->wrk[i].vpp_event_queue))
1228             clib_warning ("eventfd returned");
1229         }
1230     }
1231
1232   if (smm->evt_qs_use_memfd_seg)
1233     ssvm_pop_heap (oldheap);
1234   else
1235     svm_pop_heap (oldheap);
1236 }
1237
1238 ssvm_private_t *
1239 session_manager_get_evt_q_segment (void)
1240 {
1241   session_manager_main_t *smm = &session_manager_main;
1242   if (smm->evt_qs_use_memfd_seg)
1243     return &smm->evt_qs_segment;
1244   return 0;
1245 }
1246
1247 /* *INDENT-OFF* */
1248 static session_fifo_rx_fn *session_tx_fns[TRANSPORT_TX_N_FNS] = {
1249     session_tx_fifo_peek_and_snd,
1250     session_tx_fifo_dequeue_and_snd,
1251     session_tx_fifo_dequeue_internal,
1252     session_tx_fifo_dequeue_and_snd
1253 };
1254 /* *INDENT-ON* */
1255
1256 /**
1257  * Initialize session layer for given transport proto and ip version
1258  *
1259  * Allocates per session type (transport proto + ip version) data structures
1260  * and adds arc from session queue node to session type output node.
1261  */
1262 void
1263 session_register_transport (transport_proto_t transport_proto,
1264                             const transport_proto_vft_t * vft, u8 is_ip4,
1265                             u32 output_node)
1266 {
1267   session_manager_main_t *smm = &session_manager_main;
1268   session_type_t session_type;
1269   u32 next_index = ~0;
1270
1271   session_type = session_type_from_proto_and_ip (transport_proto, is_ip4);
1272
1273   vec_validate (smm->session_type_to_next, session_type);
1274   vec_validate (smm->session_tx_fns, session_type);
1275
1276   /* *INDENT-OFF* */
1277   if (output_node != ~0)
1278     {
1279       foreach_vlib_main (({
1280           next_index = vlib_node_add_next (this_vlib_main,
1281                                            session_queue_node.index,
1282                                            output_node);
1283       }));
1284     }
1285   /* *INDENT-ON* */
1286
1287   smm->session_type_to_next[session_type] = next_index;
1288   smm->session_tx_fns[session_type] = session_tx_fns[vft->tx_type];
1289 }
1290
1291 transport_connection_t *
1292 session_get_transport (session_t * s)
1293 {
1294   if (s->session_state != SESSION_STATE_LISTENING)
1295     return transport_get_connection (session_get_transport_proto (s),
1296                                      s->connection_index, s->thread_index);
1297   else
1298     return transport_get_listener (session_get_transport_proto (s),
1299                                    s->connection_index);
1300 }
1301
1302 transport_connection_t *
1303 listen_session_get_transport (session_t * s)
1304 {
1305   return transport_get_listener (session_get_transport_proto (s),
1306                                  s->connection_index);
1307 }
1308
1309 int
1310 listen_session_get_local_session_endpoint (session_t * listener,
1311                                            session_endpoint_t * sep)
1312 {
1313   transport_connection_t *tc;
1314   tc = listen_session_get_transport (listener);
1315   if (!tc)
1316     {
1317       clib_warning ("no transport");
1318       return -1;
1319     }
1320
1321   /* N.B. The ip should not be copied because this is the local endpoint */
1322   sep->port = tc->lcl_port;
1323   sep->transport_proto = tc->proto;
1324   sep->is_ip4 = tc->is_ip4;
1325   return 0;
1326 }
1327
1328 void
1329 session_flush_frames_main_thread (vlib_main_t * vm)
1330 {
1331   ASSERT (vlib_get_thread_index () == 0);
1332   vlib_process_signal_event_mt (vm, session_queue_process_node.index,
1333                                 SESSION_Q_PROCESS_FLUSH_FRAMES, 0);
1334 }
1335
1336 static clib_error_t *
1337 session_manager_main_enable (vlib_main_t * vm)
1338 {
1339   segment_manager_main_init_args_t _sm_args = { 0 }, *sm_args = &_sm_args;
1340   session_manager_main_t *smm = &session_manager_main;
1341   vlib_thread_main_t *vtm = vlib_get_thread_main ();
1342   u32 num_threads, preallocated_sessions_per_worker;
1343   session_manager_worker_t *wrk;
1344   int i, j;
1345
1346   num_threads = 1 /* main thread */  + vtm->n_threads;
1347
1348   if (num_threads < 1)
1349     return clib_error_return (0, "n_thread_stacks not set");
1350
1351   /* Allocate cache line aligned worker contexts */
1352   vec_validate_aligned (smm->wrk, num_threads - 1, CLIB_CACHE_LINE_BYTES);
1353
1354   for (i = 0; i < TRANSPORT_N_PROTO; i++)
1355     {
1356       for (j = 0; j < num_threads; j++)
1357         smm->wrk[j].current_enqueue_epoch[i] = 1;
1358     }
1359
1360   for (i = 0; i < num_threads; i++)
1361     {
1362       wrk = &smm->wrk[i];
1363       vec_validate (wrk->free_event_vector, 128);
1364       _vec_len (wrk->free_event_vector) = 0;
1365       vec_validate (wrk->pending_event_vector, 128);
1366       _vec_len (wrk->pending_event_vector) = 0;
1367       vec_validate (wrk->pending_disconnects, 128);
1368       _vec_len (wrk->pending_disconnects) = 0;
1369       vec_validate (wrk->postponed_event_vector, 128);
1370       _vec_len (wrk->postponed_event_vector) = 0;
1371
1372       wrk->last_vlib_time = vlib_time_now (vlib_mains[i]);
1373       wrk->dispatch_period = 500e-6;
1374
1375       if (num_threads > 1)
1376         clib_rwlock_init (&smm->wrk[i].peekers_rw_locks);
1377     }
1378
1379 #if SESSION_DEBUG
1380   vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1381 #endif
1382
1383   /* Allocate vpp event queues segment and queue */
1384   session_vpp_event_queues_allocate (smm);
1385
1386   /* Initialize fifo segment main baseva and timeout */
1387   sm_args->baseva = smm->session_baseva + smm->evt_qs_segment_size;
1388   sm_args->size = smm->session_va_space_size;
1389   segment_manager_main_init (sm_args);
1390
1391   /* Preallocate sessions */
1392   if (smm->preallocated_sessions)
1393     {
1394       if (num_threads == 1)
1395         {
1396           pool_init_fixed (smm->wrk[0].sessions, smm->preallocated_sessions);
1397         }
1398       else
1399         {
1400           int j;
1401           preallocated_sessions_per_worker =
1402             (1.1 * (f64) smm->preallocated_sessions /
1403              (f64) (num_threads - 1));
1404
1405           for (j = 1; j < num_threads; j++)
1406             {
1407               pool_init_fixed (smm->wrk[j].sessions,
1408                                preallocated_sessions_per_worker);
1409             }
1410         }
1411     }
1412
1413   session_lookup_init ();
1414   app_namespaces_init ();
1415   transport_init ();
1416
1417   smm->is_enabled = 1;
1418
1419   /* Enable transports */
1420   transport_enable_disable (vm, 1);
1421   transport_init_tx_pacers_period ();
1422   return 0;
1423 }
1424
1425 void
1426 session_node_enable_disable (u8 is_en)
1427 {
1428   u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1429   vlib_thread_main_t *vtm = vlib_get_thread_main ();
1430   u8 have_workers = vtm->n_threads != 0;
1431
1432   /* *INDENT-OFF* */
1433   foreach_vlib_main (({
1434     if (have_workers && ii == 0)
1435       {
1436         vlib_node_set_state (this_vlib_main, session_queue_process_node.index,
1437                              state);
1438         if (is_en)
1439           {
1440             vlib_node_t *n = vlib_get_node (this_vlib_main,
1441                                             session_queue_process_node.index);
1442             vlib_start_process (this_vlib_main, n->runtime_index);
1443           }
1444         else
1445           {
1446             vlib_process_signal_event_mt (this_vlib_main,
1447                                           session_queue_process_node.index,
1448                                           SESSION_Q_PROCESS_STOP, 0);
1449           }
1450
1451         continue;
1452       }
1453     vlib_node_set_state (this_vlib_main, session_queue_node.index,
1454                          state);
1455   }));
1456   /* *INDENT-ON* */
1457 }
1458
1459 clib_error_t *
1460 vnet_session_enable_disable (vlib_main_t * vm, u8 is_en)
1461 {
1462   clib_error_t *error = 0;
1463   if (is_en)
1464     {
1465       if (session_manager_main.is_enabled)
1466         return 0;
1467
1468       session_node_enable_disable (is_en);
1469       error = session_manager_main_enable (vm);
1470     }
1471   else
1472     {
1473       session_manager_main.is_enabled = 0;
1474       session_node_enable_disable (is_en);
1475     }
1476
1477   return error;
1478 }
1479
1480 clib_error_t *
1481 session_manager_main_init (vlib_main_t * vm)
1482 {
1483   session_manager_main_t *smm = &session_manager_main;
1484   smm->session_baseva = HIGH_SEGMENT_BASEVA;
1485 #if (HIGH_SEGMENT_BASEVA > (4ULL << 30))
1486   smm->session_va_space_size = 128ULL << 30;
1487   smm->evt_qs_segment_size = 64 << 20;
1488 #else
1489   smm->session_va_space_size = 128 << 20;
1490   smm->evt_qs_segment_size = 1 << 20;
1491 #endif
1492   smm->is_enabled = 0;
1493   return 0;
1494 }
1495
1496 VLIB_INIT_FUNCTION (session_manager_main_init);
1497
1498 static clib_error_t *
1499 session_config_fn (vlib_main_t * vm, unformat_input_t * input)
1500 {
1501   session_manager_main_t *smm = &session_manager_main;
1502   u32 nitems;
1503   uword tmp;
1504
1505   while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
1506     {
1507       if (unformat (input, "event-queue-length %d", &nitems))
1508         {
1509           if (nitems >= 2048)
1510             smm->configured_event_queue_length = nitems;
1511           else
1512             clib_warning ("event queue length %d too small, ignored", nitems);
1513         }
1514       else if (unformat (input, "preallocated-sessions %d",
1515                          &smm->preallocated_sessions))
1516         ;
1517       else if (unformat (input, "v4-session-table-buckets %d",
1518                          &smm->configured_v4_session_table_buckets))
1519         ;
1520       else if (unformat (input, "v4-halfopen-table-buckets %d",
1521                          &smm->configured_v4_halfopen_table_buckets))
1522         ;
1523       else if (unformat (input, "v6-session-table-buckets %d",
1524                          &smm->configured_v6_session_table_buckets))
1525         ;
1526       else if (unformat (input, "v6-halfopen-table-buckets %d",
1527                          &smm->configured_v6_halfopen_table_buckets))
1528         ;
1529       else if (unformat (input, "v4-session-table-memory %U",
1530                          unformat_memory_size, &tmp))
1531         {
1532           if (tmp >= 0x100000000)
1533             return clib_error_return (0, "memory size %llx (%lld) too large",
1534                                       tmp, tmp);
1535           smm->configured_v4_session_table_memory = tmp;
1536         }
1537       else if (unformat (input, "v4-halfopen-table-memory %U",
1538                          unformat_memory_size, &tmp))
1539         {
1540           if (tmp >= 0x100000000)
1541             return clib_error_return (0, "memory size %llx (%lld) too large",
1542                                       tmp, tmp);
1543           smm->configured_v4_halfopen_table_memory = tmp;
1544         }
1545       else if (unformat (input, "v6-session-table-memory %U",
1546                          unformat_memory_size, &tmp))
1547         {
1548           if (tmp >= 0x100000000)
1549             return clib_error_return (0, "memory size %llx (%lld) too large",
1550                                       tmp, tmp);
1551           smm->configured_v6_session_table_memory = tmp;
1552         }
1553       else if (unformat (input, "v6-halfopen-table-memory %U",
1554                          unformat_memory_size, &tmp))
1555         {
1556           if (tmp >= 0x100000000)
1557             return clib_error_return (0, "memory size %llx (%lld) too large",
1558                                       tmp, tmp);
1559           smm->configured_v6_halfopen_table_memory = tmp;
1560         }
1561       else if (unformat (input, "local-endpoints-table-memory %U",
1562                          unformat_memory_size, &tmp))
1563         {
1564           if (tmp >= 0x100000000)
1565             return clib_error_return (0, "memory size %llx (%lld) too large",
1566                                       tmp, tmp);
1567           smm->local_endpoints_table_memory = tmp;
1568         }
1569       else if (unformat (input, "local-endpoints-table-buckets %d",
1570                          &smm->local_endpoints_table_buckets))
1571         ;
1572       else if (unformat (input, "evt_qs_memfd_seg"))
1573         smm->evt_qs_use_memfd_seg = 1;
1574       else if (unformat (input, "evt_qs_seg_size %U", unformat_memory_size,
1575                          &smm->evt_qs_segment_size))
1576         ;
1577       else
1578         return clib_error_return (0, "unknown input `%U'",
1579                                   format_unformat_error, input);
1580     }
1581   return 0;
1582 }
1583
1584 VLIB_CONFIG_FUNCTION (session_config_fn, "session");
1585
1586 /*
1587  * fd.io coding-style-patch-verification: ON
1588  *
1589  * Local Variables:
1590  * eval: (c-set-style "gnu")
1591  * End:
1592  */