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