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