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