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