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