FD.io VPP  v18.01.2-1-g9b554f3
Vector Packet Processing
session_node.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 #include <math.h>
17 #include <vlib/vlib.h>
18 #include <vnet/vnet.h>
19 #include <vppinfra/elog.h>
20 #include <vnet/session/transport.h>
24 
26 
27 typedef struct
28 {
32 
33 /* packet trace format function */
34 static u8 *
35 format_session_queue_trace (u8 * s, va_list * args)
36 {
37  CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
38  CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
39  session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
40 
41  s = format (s, "SESSION_QUEUE: session index %d, server thread index %d",
43  return s;
44 }
45 
47 
48 #define foreach_session_queue_error \
49 _(TX, "Packets transmitted") \
50 _(TIMER, "Timer events") \
51 _(NO_BUFFER, "Out of buffers")
52 
53 typedef enum
54 {
55 #define _(sym,str) SESSION_QUEUE_ERROR_##sym,
57 #undef _
60 
61 static char *session_queue_error_strings[] = {
62 #define _(sym,string) string,
64 #undef _
65 };
66 
67 always_inline void
69  u8 thread_index, svm_fifo_t * fifo,
70  vlib_buffer_t * b0, u32 bi0, u8 n_bufs_per_seg,
71  u32 left_from_seg, u32 * left_to_snd0,
72  u16 * n_bufs, u32 * tx_offset, u16 deq_per_buf,
73  u8 peek_data)
74 {
75  vlib_buffer_t *chain_b0, *prev_b0;
76  u32 chain_bi0, to_deq;
77  u16 len_to_deq0, n_bytes_read;
78  u8 *data0, j;
79 
82 
83  chain_bi0 = bi0;
84  chain_b0 = b0;
85  to_deq = left_from_seg;
86  for (j = 1; j < n_bufs_per_seg; j++)
87  {
88  prev_b0 = chain_b0;
89  len_to_deq0 = clib_min (to_deq, deq_per_buf);
90 
91  *n_bufs -= 1;
92  chain_bi0 = smm->tx_buffers[thread_index][*n_bufs];
93  _vec_len (smm->tx_buffers[thread_index]) = *n_bufs;
94 
95  chain_b0 = vlib_get_buffer (vm, chain_bi0);
96  chain_b0->current_data = 0;
97  data0 = vlib_buffer_get_current (chain_b0);
98  if (peek_data)
99  {
100  n_bytes_read = svm_fifo_peek (fifo, *tx_offset, len_to_deq0, data0);
101  *tx_offset += n_bytes_read;
102  }
103  else
104  {
105  n_bytes_read = svm_fifo_dequeue_nowait (fifo, len_to_deq0, data0);
106  }
107  ASSERT (n_bytes_read == len_to_deq0);
108  chain_b0->current_length = n_bytes_read;
110 
111  /* update previous buffer */
112  prev_b0->next_buffer = chain_bi0;
113  prev_b0->flags |= VLIB_BUFFER_NEXT_PRESENT;
114 
115  /* update current buffer */
116  chain_b0->next_buffer = 0;
117 
118  to_deq -= n_bytes_read;
119  if (to_deq == 0)
120  break;
121  }
122  ASSERT (to_deq == 0
123  && b0->total_length_not_including_first_buffer == left_from_seg);
124  *left_to_snd0 -= left_from_seg;
125 }
126 
127 always_inline int
130  session_fifo_event_t * e0,
131  stream_session_t * s0, u32 thread_index,
132  int *n_tx_packets, u8 peek_data)
133 {
134  u32 n_trace = vlib_get_trace_count (vm, node);
135  u32 left_to_snd0, max_len_to_snd0, len_to_deq0, snd_space0;
136  u32 n_bufs_per_evt, n_frames_per_evt, n_bufs_per_frame;
138  transport_proto_vft_t *transport_vft;
140  u32 next_index, next0, *to_next, n_left_to_next, bi0;
141  vlib_buffer_t *b0;
142  u32 tx_offset = 0, max_dequeue0, n_bytes_per_seg, left_for_seg;
143  u16 snd_mss0, n_bufs_per_seg, n_bufs;
144  u8 *data0;
145  int i, n_bytes_read;
146  u32 n_bytes_per_buf, deq_per_buf, deq_per_first_buf;
147  u32 buffers_allocated, buffers_allocated_this_call;
148 
149  next_index = next0 = smm->session_type_to_next[s0->session_type];
150 
151  tp = session_get_transport_proto (s0);
152  transport_vft = transport_protocol_get_vft (tp);
153  tc0 = transport_vft->get_connection (s0->connection_index, thread_index);
154 
155  /* Make sure we have space to send and there's something to dequeue */
156  snd_mss0 = transport_vft->send_mss (tc0);
157  snd_space0 = transport_vft->send_space (tc0);
158 
159  /* Can't make any progress */
160  if (snd_space0 == 0 || snd_mss0 == 0)
161  {
162  vec_add1 (smm->pending_event_vector[thread_index], *e0);
163  return 0;
164  }
165 
166  /* Allow enqueuing of a new event */
167  svm_fifo_unset_event (s0->server_tx_fifo);
168 
169  /* Check how much we can pull. */
170  max_dequeue0 = svm_fifo_max_dequeue (s0->server_tx_fifo);
171 
172  if (peek_data)
173  {
174  /* Offset in rx fifo from where to peek data */
175  tx_offset = transport_vft->tx_fifo_offset (tc0);
176  if (PREDICT_FALSE (tx_offset >= max_dequeue0))
177  max_dequeue0 = 0;
178  else
179  max_dequeue0 -= tx_offset;
180  }
181 
182  /* Nothing to read return */
183  if (max_dequeue0 == 0)
184  return 0;
185 
186  /* Ensure we're not writing more than transport window allows */
187  if (max_dequeue0 < snd_space0)
188  {
189  /* Constrained by tx queue. Try to send only fully formed segments */
190  max_len_to_snd0 = (max_dequeue0 > snd_mss0) ?
191  max_dequeue0 - max_dequeue0 % snd_mss0 : max_dequeue0;
192  /* TODO Nagle ? */
193  }
194  else
195  {
196  /* Expectation is that snd_space0 is already a multiple of snd_mss */
197  max_len_to_snd0 = snd_space0;
198  }
199 
200  n_bytes_per_buf = vlib_buffer_free_list_buffer_size
202  ASSERT (n_bytes_per_buf > MAX_HDRS_LEN);
203  n_bytes_per_seg = MAX_HDRS_LEN + snd_mss0;
204  n_bufs_per_seg = ceil ((double) n_bytes_per_seg / n_bytes_per_buf);
205  n_bufs_per_evt = ceil ((double) max_len_to_snd0 / n_bytes_per_seg);
206  n_frames_per_evt = ceil ((double) n_bufs_per_evt / VLIB_FRAME_SIZE);
207  n_bufs_per_frame = n_bufs_per_seg * VLIB_FRAME_SIZE;
208 
209  deq_per_buf = clib_min (snd_mss0, n_bytes_per_buf);
210  deq_per_first_buf = clib_min (snd_mss0, n_bytes_per_buf - MAX_HDRS_LEN);
211 
212  n_bufs = vec_len (smm->tx_buffers[thread_index]);
213  left_to_snd0 = max_len_to_snd0;
214  for (i = 0; i < n_frames_per_evt; i++)
215  {
216  /* Make sure we have at least one full frame of buffers ready */
217  if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
218  {
219  vec_validate (smm->tx_buffers[thread_index],
220  n_bufs + n_bufs_per_frame - 1);
221  buffers_allocated = 0;
222  do
223  {
224  buffers_allocated_this_call = vlib_buffer_alloc (vm,
225  &smm->tx_buffers
226  [thread_index]
227  [n_bufs +
228  buffers_allocated],
229  n_bufs_per_frame
230  -
231  buffers_allocated);
232  buffers_allocated += buffers_allocated_this_call;
233  }
234  while (buffers_allocated_this_call > 0
235  && ((buffers_allocated + n_bufs < n_bufs_per_frame)));
236 
237  n_bufs += buffers_allocated;
238  _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
239 
240  if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
241  {
242  vec_add1 (smm->pending_event_vector[thread_index], *e0);
243  return -1;
244  }
245  ASSERT (n_bufs >= n_bufs_per_frame);
246  }
247 
248  vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
249  while (left_to_snd0 && n_left_to_next)
250  {
251  /*
252  * Handle first buffer in chain separately
253  */
254 
255  /* Get free buffer */
256  ASSERT (n_bufs >= 1);
257  bi0 = smm->tx_buffers[thread_index][--n_bufs];
258  _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
259 
260  /* usual speculation, or the enqueue_x1 macro will barf */
261  to_next[0] = bi0;
262  to_next += 1;
263  n_left_to_next -= 1;
264 
265  b0 = vlib_get_buffer (vm, bi0);
266  b0->error = 0;
267  b0->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
268  b0->current_data = 0;
270 
271  len_to_deq0 = clib_min (left_to_snd0, deq_per_first_buf);
273  if (peek_data)
274  {
275  n_bytes_read = svm_fifo_peek (s0->server_tx_fifo, tx_offset,
276  len_to_deq0, data0);
277  if (n_bytes_read <= 0)
278  goto dequeue_fail;
279  /* Keep track of progress locally, transport is also supposed to
280  * increment it independently when pushing the header */
281  tx_offset += n_bytes_read;
282  }
283  else
284  {
285  n_bytes_read = svm_fifo_dequeue_nowait (s0->server_tx_fifo,
286  len_to_deq0, data0);
287  if (n_bytes_read <= 0)
288  goto dequeue_fail;
289  }
290 
291  b0->current_length = n_bytes_read;
292 
293  left_to_snd0 -= n_bytes_read;
294  *n_tx_packets = *n_tx_packets + 1;
295 
296  /*
297  * Fill in the remaining buffers in the chain, if any
298  */
299  if (PREDICT_FALSE (n_bufs_per_seg > 1 && left_to_snd0))
300  {
301  left_for_seg = clib_min (snd_mss0 - n_bytes_read, left_to_snd0);
302  session_tx_fifo_chain_tail (smm, vm, thread_index,
303  s0->server_tx_fifo, b0, bi0,
304  n_bufs_per_seg, left_for_seg,
305  &left_to_snd0, &n_bufs, &tx_offset,
306  deq_per_buf, peek_data);
307  }
308 
309  /* Ask transport to push header after current_length and
310  * total_length_not_including_first_buffer are updated */
311  transport_vft->push_header (tc0, b0);
312 
313  /* *INDENT-OFF* */
314  SESSION_EVT_DBG(SESSION_EVT_DEQ, s0, ({
315  ed->data[0] = e0->event_type;
316  ed->data[1] = max_dequeue0;
317  ed->data[2] = len_to_deq0;
318  ed->data[3] = left_to_snd0;
319  }));
320  /* *INDENT-ON* */
321 
323 
324  if (PREDICT_FALSE (n_trace > 0))
325  {
327  vlib_trace_buffer (vm, node, next_index, b0,
328  1 /* follow_chain */ );
329  vlib_set_trace_count (vm, node, --n_trace);
330  t0 = vlib_add_trace (vm, node, b0, sizeof (*t0));
331  t0->session_index = s0->session_index;
332  t0->server_thread_index = s0->thread_index;
333  }
334 
335  vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
336  to_next, n_left_to_next,
337  bi0, next0);
338  }
339  vlib_put_next_frame (vm, node, next_index, n_left_to_next);
340  }
341 
342  /* If we couldn't dequeue all bytes mark as partially read */
343  if (max_len_to_snd0 < max_dequeue0)
344  {
345  /* If we don't already have new event */
346  if (svm_fifo_set_event (s0->server_tx_fifo))
347  {
348  vec_add1 (smm->pending_event_vector[thread_index], *e0);
349  }
350  }
351  return 0;
352 
353 dequeue_fail:
354  /*
355  * Can't read from fifo. If we don't already have an event, save as partially
356  * read, return buff to free list and return
357  */
358  clib_warning ("dequeue fail");
359 
360  if (svm_fifo_set_event (s0->server_tx_fifo))
361  {
362  vec_add1 (smm->pending_event_vector[thread_index], *e0);
363  }
364  vlib_put_next_frame (vm, node, next_index, n_left_to_next + 1);
365  _vec_len (smm->tx_buffers[thread_index]) += 1;
366 
367  return 0;
368 }
369 
370 int
373  session_fifo_event_t * e0,
374  stream_session_t * s0, u32 thread_index,
375  int *n_tx_pkts)
376 {
377  return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
378  n_tx_pkts, 1);
379 }
380 
381 int
384  session_fifo_event_t * e0,
385  stream_session_t * s0, u32 thread_index,
386  int *n_tx_pkts)
387 {
388  return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
389  n_tx_pkts, 0);
390 }
391 
393 session_event_get_session (session_fifo_event_t * e, u8 thread_index)
394 {
395  return session_get_if_valid (e->fifo->master_session_index, thread_index);
396 }
397 
398 void
400 {
403  u32 my_thread_index = vm->thread_index;
404  session_fifo_event_t _e, *e = &_e;
405  stream_session_t *s0;
406  int i, index;
407  i8 *headp;
408 
410  q = smm->vpp_event_queues[my_thread_index];
411 
412  index = q->head;
413 
414  for (i = 0; i < q->cursize; i++)
415  {
416  headp = (i8 *) (&q->data[0] + q->elsize * index);
417  clib_memcpy (e, headp, q->elsize);
418 
419  switch (e->event_type)
420  {
421  case FIFO_EVENT_APP_TX:
422  s0 = session_event_get_session (e, my_thread_index);
423  fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
424  break;
425 
427  s0 = session_get_from_handle (e->session_handle);
428  fformat (stdout, "[%04d] disconnect session %d\n", i,
429  s0->session_index);
430  break;
431 
433  s0 = session_event_get_session (e, my_thread_index);
434  fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
435  break;
436 
437  case FIFO_EVENT_RPC:
438  fformat (stdout, "[%04d] RPC call %llx with %llx\n",
439  i, (u64) (e->rpc_args.fp), (u64) (e->rpc_args.arg));
440  break;
441 
442  default:
443  fformat (stdout, "[%04d] unhandled event type %d\n",
444  i, e->event_type);
445  break;
446  }
447 
448  index++;
449 
450  if (index == q->maxsize)
451  index = 0;
452  }
453 }
454 
455 static u8
456 session_node_cmp_event (session_fifo_event_t * e, svm_fifo_t * f)
457 {
458  stream_session_t *s;
459  switch (e->event_type)
460  {
461  case FIFO_EVENT_APP_RX:
462  case FIFO_EVENT_APP_TX:
464  if (e->fifo == f)
465  return 1;
466  break;
468  break;
469  case FIFO_EVENT_RPC:
470  s = session_get_from_handle (e->session_handle);
471  if (!s)
472  {
473  clib_warning ("session has event but doesn't exist!");
474  break;
475  }
476  if (s->server_rx_fifo == f || s->server_tx_fifo == f)
477  return 1;
478  break;
479  default:
480  break;
481  }
482  return 0;
483 }
484 
485 u8
486 session_node_lookup_fifo_event (svm_fifo_t * f, session_fifo_event_t * e)
487 {
490  session_fifo_event_t *pending_event_vector, *evt;
491  int i, index, found = 0;
492  i8 *headp;
493  u8 thread_index;
494 
495  ASSERT (e);
496  thread_index = f->master_thread_index;
497  /*
498  * Search evt queue
499  */
500  q = smm->vpp_event_queues[thread_index];
501  index = q->head;
502  for (i = 0; i < q->cursize; i++)
503  {
504  headp = (i8 *) (&q->data[0] + q->elsize * index);
505  clib_memcpy (e, headp, q->elsize);
506  found = session_node_cmp_event (e, f);
507  if (found)
508  break;
509  if (++index == q->maxsize)
510  index = 0;
511  }
512  /*
513  * Search pending events vector
514  */
515  pending_event_vector = smm->pending_event_vector[thread_index];
516  vec_foreach (evt, pending_event_vector)
517  {
518  found = session_node_cmp_event (evt, f);
519  if (found)
520  {
521  clib_memcpy (e, evt, sizeof (*evt));
522  break;
523  }
524  }
525  return found;
526 }
527 
528 static uword
530  vlib_frame_t * frame)
531 {
533  session_fifo_event_t *my_pending_event_vector, *pending_disconnects, *e;
534  session_fifo_event_t *my_fifo_events;
535  u32 n_to_dequeue, n_events;
537  application_t *app;
538  int n_tx_packets = 0;
539  u32 my_thread_index = vm->thread_index;
540  int i, rv;
541  f64 now = vlib_time_now (vm);
542  void (*fp) (void *);
543 
544  SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, my_thread_index);
545 
546  /*
547  * Update transport time
548  */
549  transport_update_time (now, my_thread_index);
550 
551  /*
552  * Get vpp queue events
553  */
554  q = smm->vpp_event_queues[my_thread_index];
555  if (PREDICT_FALSE (q == 0))
556  return 0;
557 
558  my_fifo_events = smm->free_event_vector[my_thread_index];
559 
560  /* min number of events we can dequeue without blocking */
561  n_to_dequeue = q->cursize;
562  my_pending_event_vector = smm->pending_event_vector[my_thread_index];
563  pending_disconnects = smm->pending_disconnects[my_thread_index];
564 
565  if (!n_to_dequeue && !vec_len (my_pending_event_vector)
566  && !vec_len (pending_disconnects))
567  return 0;
568 
569  SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0);
570 
571  /*
572  * If we didn't manage to process previous events try going
573  * over them again without dequeuing new ones.
574  */
575  /* XXX: Block senders to sessions that can't keep up */
576  if (0 && vec_len (my_pending_event_vector) >= 100)
577  {
578  clib_warning ("too many fifo events unsolved");
579  goto skip_dequeue;
580  }
581 
582  /* See you in the next life, don't be late */
583  if (pthread_mutex_trylock (&q->mutex))
584  return 0;
585 
586  for (i = 0; i < n_to_dequeue; i++)
587  {
588  vec_add2 (my_fifo_events, e, 1);
590  }
591 
592  /* The other side of the connection is not polling */
593  if (q->cursize < (q->maxsize / 8))
594  (void) pthread_cond_broadcast (&q->condvar);
595  pthread_mutex_unlock (&q->mutex);
596 
597  vec_append (my_fifo_events, my_pending_event_vector);
598  vec_append (my_fifo_events, smm->pending_disconnects[my_thread_index]);
599 
600  _vec_len (my_pending_event_vector) = 0;
601  smm->pending_event_vector[my_thread_index] = my_pending_event_vector;
602  _vec_len (smm->pending_disconnects[my_thread_index]) = 0;
603 
604 skip_dequeue:
605  n_events = vec_len (my_fifo_events);
606  for (i = 0; i < n_events; i++)
607  {
608  stream_session_t *s0; /* $$$ prefetch 1 ahead maybe */
609  session_fifo_event_t *e0;
610 
611  e0 = &my_fifo_events[i];
612 
613  switch (e0->event_type)
614  {
615  case FIFO_EVENT_APP_TX:
616  s0 = session_event_get_session (e0, my_thread_index);
617 
618  if (PREDICT_FALSE (!s0))
619  {
620  clib_warning ("It's dead, Jim!");
621  continue;
622  }
623  /* Can retransmit for closed sessions but can't do anything if
624  * session is not ready or closed */
625  if (PREDICT_FALSE (s0->session_state < SESSION_STATE_READY))
626  {
627  vec_add1 (smm->pending_event_vector[my_thread_index], *e0);
628  continue;
629  }
630  /* Spray packets in per session type frames, since they go to
631  * different nodes */
632  rv = (smm->session_tx_fns[s0->session_type]) (vm, node, smm, e0, s0,
633  my_thread_index,
634  &n_tx_packets);
635  /* Out of buffers */
636  if (PREDICT_FALSE (rv < 0))
637  {
639  SESSION_QUEUE_ERROR_NO_BUFFER, 1);
640  continue;
641  }
642  break;
644  /* Make sure disconnects run after the pending list is drained */
645  if (!e0->postponed)
646  {
647  e0->postponed = 1;
648  vec_add1 (smm->pending_disconnects[my_thread_index], *e0);
649  continue;
650  }
651  s0 = session_get_from_handle (e0->session_handle);
653  break;
655  s0 = session_event_get_session (e0, my_thread_index);
656  if (PREDICT_FALSE (!s0))
657  continue;
658  svm_fifo_unset_event (s0->server_rx_fifo);
659  app = application_get (s0->app_index);
660  app->cb_fns.builtin_server_rx_callback (s0);
661  break;
662  case FIFO_EVENT_RPC:
663  fp = e0->rpc_args.fp;
664  (*fp) (e0->rpc_args.arg);
665  break;
666 
667  default:
668  clib_warning ("unhandled event type %d", e0->event_type);
669  }
670  }
671 
672  _vec_len (my_fifo_events) = 0;
673  smm->free_event_vector[my_thread_index] = my_fifo_events;
674 
676  SESSION_QUEUE_ERROR_TX, n_tx_packets);
677 
678  SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 1);
679 
680  return n_tx_packets;
681 }
682 
683 /* *INDENT-OFF* */
685 {
686  .function = session_queue_node_fn,
687  .name = "session-queue",
688  .format_trace = format_session_queue_trace,
689  .type = VLIB_NODE_TYPE_INPUT,
691  .error_strings = session_queue_error_strings,
692  .state = VLIB_NODE_STATE_DISABLED,
693 };
694 /* *INDENT-ON* */
695 
696 static clib_error_t *
698 {
699  if (vec_len (vlib_mains) < 2)
700  return 0;
701 
702  /*
703  * Shut off (especially) worker-thread session nodes.
704  * Otherwise, vpp can crash as the main thread unmaps the
705  * API segment.
706  */
708  session_node_enable_disable (0 /* is_enable */ );
710  return 0;
711 }
712 
714 
715 /*
716  * fd.io coding-style-patch-verification: ON
717  *
718  * Local Variables:
719  * eval: (c-set-style "gnu")
720  * End:
721  */
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
Definition: vec.h:432
vlib_main_t vlib_global_main
Definition: main.c:1637
sll srl srl sll sra u16x4 i
Definition: vector_sse2.h:337
#define clib_min(x, y)
Definition: clib.h:340
#define CLIB_UNUSED(x)
Definition: clib.h:79
static u32 vlib_get_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt)
Definition: trace_funcs.h:143
struct _transport_connection transport_connection_t
static f64 vlib_time_now(vlib_main_t *vm)
Definition: main.h:224
u32 thread_index
Definition: main.h:173
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:518
struct _vlib_node_registration vlib_node_registration_t
#define vec_add2(V, P, N)
Add N elements to end of vector V, return pointer to new elements in P.
Definition: vec.h:557
static clib_error_t * session_queue_exit(vlib_main_t *vm)
Definition: session_node.c:697
static uword session_queue_node_fn(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
Definition: session_node.c:529
u8 * format(u8 *s, const char *fmt,...)
Definition: format.c:419
void session_node_enable_disable(u8 is_en)
Definition: session.c:1159
vlib_main_t ** vlib_mains
Definition: buffer.c:292
void transport_update_time(f64 time_now, u8 thread_index)
Definition: transport.c:331
#define vlib_worker_thread_barrier_sync(X)
Definition: threads.h:212
struct _svm_fifo svm_fifo_t
static void vlib_trace_buffer(vlib_main_t *vm, vlib_node_runtime_t *r, u32 next_index, vlib_buffer_t *b, int follow_chain)
Definition: trace_funcs.h:104
static int session_tx_fifo_read_and_snd_i(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_main_t *smm, session_fifo_event_t *e0, stream_session_t *s0, u32 thread_index, int *n_tx_packets, u8 peek_data)
Definition: session_node.c:128
void dump_thread_0_event_queue(void)
Definition: session_node.c:399
#define VLIB_BUFFER_NEXT_PRESENT
Definition: buffer.h:95
i16 current_data
signed offset in data[], pre_data[] that we are currently processing.
Definition: buffer.h:68
#define always_inline
Definition: clib.h:92
static u32 svm_fifo_max_dequeue(svm_fifo_t *f)
Definition: svm_fifo.h:100
#define VLIB_BUFFER_TOTAL_LENGTH_VALID
Definition: buffer.h:97
char i8
Definition: types.h:45
unsigned long u64
Definition: types.h:89
struct _stream_session_t stream_session_t
static char * session_queue_error_strings[]
Definition: session_node.c:61
transport_proto_vft_t * transport_protocol_get_vft(transport_proto_t transport_proto)
Get transport virtual function table.
Definition: transport.c:167
static transport_proto_t session_get_transport_proto(stream_session_t *s)
Definition: session.h:283
static session_manager_main_t * vnet_get_session_manager_main()
Definition: session.h:207
u16 current_length
Nbytes between current data and the end of this buffer.
Definition: buffer.h:72
static void * vlib_buffer_make_headroom(vlib_buffer_t *b, u8 size)
Make head room, typically for packet headers.
Definition: buffer.h:307
#define MAX_HDRS_LEN
Definition: session.h:30
session_queue_error_t
Definition: session_node.c:53
static void * vlib_buffer_get_current(vlib_buffer_t *b)
Get pointer to current data to process.
Definition: buffer.h:195
#define VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX
Definition: buffer.h:434
int session_tx_fifo_peek_and_snd(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_main_t *smm, session_fifo_event_t *e0, stream_session_t *s0, u32 thread_index, int *n_tx_pkts)
Definition: session_node.c:371
#define PREDICT_FALSE(x)
Definition: clib.h:105
static u32 vlib_buffer_free_list_buffer_size(vlib_main_t *vm, u32 free_list_index)
Definition: buffer_funcs.h:465
static void svm_fifo_unset_event(svm_fifo_t *f)
Unsets fifo event flag.
Definition: svm_fifo.h:133
int unix_shared_memory_queue_sub_raw(unix_shared_memory_queue_t *q, u8 *elem)
#define VLIB_FRAME_SIZE
Definition: node.h:328
static stream_session_t * session_get_if_valid(u64 si, u32 thread_index)
Definition: session.h:236
u32 node_index
Node index.
Definition: node.h:437
struct _session_manager_main session_manager_main_t
Definition: session.h:103
#define vlib_validate_buffer_enqueue_x1(vm, node, next_index, to_next, n_left_to_next, bi0, next0)
Finish enqueueing one buffer forward in the graph.
Definition: buffer_node.h:218
#define vlib_get_next_frame(vm, node, next_index, vectors, n_vectors_left)
Get pointer to next frame vector data by (vlib_node_runtime_t, next_index).
Definition: node_funcs.h:364
vlib_error_t error
Error code for buffers to be enqueued to error handler.
Definition: buffer.h:113
word fformat(FILE *f, char *fmt,...)
Definition: format.c:453
static void vlib_node_increment_counter(vlib_main_t *vm, u32 node_index, u32 counter_index, u64 increment)
Definition: node_funcs.h:1158
static u8 svm_fifo_set_event(svm_fifo_t *f)
Sets fifo event flag.
Definition: svm_fifo.h:123
The fine-grained event logger allows lightweight, thread-safe event logging at minimum cost...
#define SESSION_EVT_DBG(_evt, _args...)
vlib_main_t * vm
Definition: buffer.c:283
#define VLIB_MAIN_LOOP_EXIT_FUNCTION(x)
Definition: init.h:116
#define clib_warning(format, args...)
Definition: error.h:59
#define clib_memcpy(a, b, c)
Definition: string.h:75
struct _application application_t
#define ARRAY_LEN(x)
Definition: clib.h:59
void vlib_put_next_frame(vlib_main_t *vm, vlib_node_runtime_t *r, u32 next_index, u32 n_vectors_left)
Release pointer to next frame vector data.
Definition: main.c:454
#define foreach_session_queue_error
Definition: session_node.c:48
#define ASSERT(truth)
unsigned int u32
Definition: types.h:88
static u8 session_node_cmp_event(session_fifo_event_t *e, svm_fifo_t *f)
Definition: session_node.c:456
u32 next_buffer
Next buffer for this linked-list of buffers.
Definition: buffer.h:109
#define vec_append(v1, v2)
Append v2 after v1.
Definition: vec.h:815
void stream_session_disconnect(stream_session_t *s)
Disconnect session and propagate to transport.
Definition: session.c:934
static stream_session_t * session_event_get_session(session_fifo_event_t *e, u8 thread_index)
Definition: session_node.c:393
static u8 * format_session_queue_trace(u8 *s, va_list *args)
Definition: session_node.c:35
u64 uword
Definition: types.h:112
static void * vlib_add_trace(vlib_main_t *vm, vlib_node_runtime_t *r, vlib_buffer_t *b, u32 n_data_bytes)
Definition: trace_funcs.h:55
u32 total_length_not_including_first_buffer
Only valid for first buffer in chain.
Definition: buffer.h:142
int session_tx_fifo_dequeue_and_snd(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_main_t *smm, session_fifo_event_t *e0, stream_session_t *s0, u32 thread_index, int *n_tx_pkts)
Definition: session_node.c:382
static stream_session_t * session_get_from_handle(u64 handle)
Definition: session.h:274
unsigned short u16
Definition: types.h:57
enum _transport_proto transport_proto_t
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
double f64
Definition: types.h:142
unsigned char u8
Definition: types.h:56
#define VLIB_BUFFER_TRACE_TRAJECTORY_INIT(b)
Definition: buffer.h:534
application_t * application_get(u32 index)
Definition: application.c:297
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
Definition: session_node.c:25
void vlib_worker_thread_barrier_release(vlib_main_t *vm)
Definition: threads.c:1491
#define VLIB_REGISTER_NODE(x,...)
Definition: node.h:143
#define vec_foreach(var, vec)
Vector iterator.
static void session_tx_fifo_chain_tail(session_manager_main_t *smm, vlib_main_t *vm, u8 thread_index, svm_fifo_t *fifo, vlib_buffer_t *b0, u32 bi0, u8 n_bufs_per_seg, u32 left_from_seg, u32 *left_to_snd0, u16 *n_bufs, u32 *tx_offset, u16 deq_per_buf, u8 peek_data)
Definition: session_node.c:68
static void vlib_set_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt, u32 count)
Definition: trace_funcs.h:159
u8 session_node_lookup_fifo_event(svm_fifo_t *f, session_fifo_event_t *e)
Definition: session_node.c:486
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:75
static u32 vlib_buffer_alloc(vlib_main_t *vm, u32 *buffers, u32 n_buffers)
Allocate buffers into supplied array.
Definition: buffer_funcs.h:341
int svm_fifo_peek(svm_fifo_t *f, u32 relative_offset, u32 max_bytes, u8 *copy_here)
Definition: svm_fifo.c:756
int svm_fifo_dequeue_nowait(svm_fifo_t *f, u32 max_bytes, u8 *copy_here)
Definition: svm_fifo.c:690
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
struct _unix_shared_memory_queue unix_shared_memory_queue_t