FD.io VPP  v20.05.1-6-gf53edbc3b
Vector Packet Processing
session_node.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2017-2019 Cisco and/or its affiliates.
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  * http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15 
16 #include <math.h>
17 #include <vlib/vlib.h>
18 #include <vnet/vnet.h>
19 #include <vppinfra/elog.h>
20 #include <vnet/session/transport.h>
21 #include <vnet/session/session.h>
26 #include <svm/queue.h>
27 
28 #define app_check_thread_and_barrier(_fn, _arg) \
29  if (!vlib_thread_is_main_w_barrier ()) \
30  { \
31  vlib_rpc_call_main_thread (_fn, (u8 *) _arg, sizeof(*_arg)); \
32  return; \
33  }
34 
35 static void
37 {
39  vnet_listen_args_t _a, *a = &_a;
40  app_worker_t *app_wrk;
41  application_t *app;
42  int rv;
43 
45 
46  app = application_lookup (mp->client_index);
47  if (!app)
48  return;
49 
50  clib_memset (a, 0, sizeof (*a));
51  a->sep.is_ip4 = mp->is_ip4;
52  clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
53  if (mp->is_ip4)
54  ip46_address_mask_ip4 (&a->sep.ip);
55  a->sep.port = mp->port;
56  a->sep.fib_index = mp->vrf;
57  a->sep.sw_if_index = ENDPOINT_INVALID_INDEX;
58  a->sep.transport_proto = mp->proto;
59  a->sep_ext.ckpair_index = mp->ckpair_index;
60  a->sep_ext.crypto_engine = mp->crypto_engine;
61  a->app_index = app->app_index;
62  a->wrk_map_index = mp->wrk_index;
63  a->sep_ext.transport_flags = mp->flags;
64 
65  if ((rv = vnet_listen (a)))
66  clib_warning ("listen returned: %d", rv);
67 
68  app_wrk = application_get_worker (app, mp->wrk_index);
69  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
70  return;
71 }
72 
73 static void
75 {
77  vnet_listen_args_t _a, *a = &_a;
78  app_worker_t *app_wrk;
79  application_t *app;
80  int rv;
81 
83 
84  app = application_lookup (mp->client_index);
85  if (!app)
86  return;
87 
88  clib_memset (a, 0, sizeof (*a));
89  a->uri = (char *) mp->uri;
90  a->app_index = app->app_index;
91  rv = vnet_bind_uri (a);
92 
93  app_wrk = application_get_worker (app, 0);
94  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
95 }
96 
97 static void
99 {
101  vnet_connect_args_t _a, *a = &_a;
102  app_worker_t *app_wrk;
103  application_t *app;
104  int rv;
105 
107 
108  app = application_lookup (mp->client_index);
109  if (!app)
110  return;
111 
112  clib_memset (a, 0, sizeof (*a));
113  a->sep.is_ip4 = mp->is_ip4;
114  clib_memcpy_fast (&a->sep.ip, &mp->ip, sizeof (mp->ip));
115  a->sep.port = mp->port;
116  a->sep.transport_proto = mp->proto;
117  a->sep.peer.fib_index = mp->vrf;
118  clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
119  if (mp->is_ip4)
120  {
121  ip46_address_mask_ip4 (&a->sep.ip);
122  ip46_address_mask_ip4 (&a->sep.peer.ip);
123  }
124  a->sep.peer.port = mp->lcl_port;
125  a->sep.peer.sw_if_index = ENDPOINT_INVALID_INDEX;
126  a->sep_ext.parent_handle = mp->parent_handle;
127  a->sep_ext.ckpair_index = mp->ckpair_index;
128  a->sep_ext.crypto_engine = mp->crypto_engine;
129  a->sep_ext.transport_flags = mp->flags;
130  if (mp->hostname_len)
131  {
132  vec_validate (a->sep_ext.hostname, mp->hostname_len - 1);
133  clib_memcpy_fast (a->sep_ext.hostname, mp->hostname, mp->hostname_len);
134  }
135  a->api_context = mp->context;
136  a->app_index = app->app_index;
137  a->wrk_map_index = mp->wrk_index;
138 
139  if ((rv = vnet_connect (a)))
140  {
141  clib_warning ("connect returned: %U", format_session_error, rv);
142  app_wrk = application_get_worker (app, mp->wrk_index);
143  mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
144  }
145 
146  vec_free (a->sep_ext.hostname);
147 }
148 
149 static void
151 {
153  vnet_connect_args_t _a, *a = &_a;
154  app_worker_t *app_wrk;
155  application_t *app;
156  int rv;
157 
159 
160  app = application_lookup (mp->client_index);
161  if (!app)
162  return;
163 
164  clib_memset (a, 0, sizeof (*a));
165  a->uri = (char *) mp->uri;
166  a->api_context = mp->context;
167  a->app_index = app->app_index;
168  if ((rv = vnet_connect_uri (a)))
169  {
170  clib_warning ("connect_uri returned: %d", rv);
171  app_wrk = application_get_worker (app, 0 /* default wrk only */ );
172  mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
173  }
174 }
175 
176 static void
178 {
180  vnet_disconnect_args_t _a, *a = &_a;
181  application_t *app;
182 
183  app = application_lookup (mp->client_index);
184  if (!app)
185  return;
186 
187  a->app_index = app->app_index;
188  a->handle = mp->handle;
190 }
191 
192 static void
194 {
196  vnet_app_detach_args_t _a, *a = &_a;
197  application_t *app;
198 
200 
201  app = application_lookup (mp->client_index);
202  if (!app)
203  return;
204 
205  a->app_index = app->app_index;
206  a->api_client_index = mp->client_index;
208 }
209 
210 static void
212 {
214  vnet_unlisten_args_t _a, *a = &_a;
215  app_worker_t *app_wrk;
216  application_t *app;
217  int rv;
218 
220 
221  app = application_lookup (mp->client_index);
222  if (!app)
223  return;
224 
225  clib_memset (a, 0, sizeof (*a));
226  a->app_index = app->app_index;
227  a->handle = mp->handle;
228  a->wrk_map_index = mp->wrk_index;
229  if ((rv = vnet_unlisten (a)))
230  clib_warning ("unlisten returned: %d", rv);
231 
232  app_wrk = application_get_worker (app, a->wrk_map_index);
233  if (!app_wrk)
234  return;
235 
236  mq_send_unlisten_reply (app_wrk, mp->handle, mp->context, rv);
237 }
238 
239 static void
241 {
243  vnet_disconnect_args_t _a = { 0 }, *a = &_a;
244  session_state_t old_state;
245  app_worker_t *app_wrk;
246  session_t *s;
247 
248  /* Server isn't interested, kill the session */
249  if (mp->retval)
250  {
251  a->app_index = mp->context;
252  a->handle = mp->handle;
254  return;
255  }
256 
257  /* Mail this back from the main thread. We're not polling in main
258  * thread so we're using other workers for notifications. */
259  if (vlib_num_workers () && vlib_get_thread_index () != 0
260  && session_thread_from_handle (mp->handle) == 0)
261  {
263  (u8 *) mp, sizeof (*mp));
264  return;
265  }
266 
268  if (!s)
269  return;
270 
271  app_wrk = app_worker_get (s->app_wrk_index);
272  if (app_wrk->app_index != mp->context)
273  {
274  clib_warning ("app doesn't own session");
275  return;
276  }
277 
278  if (!session_has_transport (s))
279  {
280  s->session_state = SESSION_STATE_READY;
282  return;
283  }
284  else
285  {
286  old_state = s->session_state;
287  s->session_state = SESSION_STATE_READY;
288 
291 
292  /* Closed while waiting for app to reply. Resend disconnect */
293  if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
294  {
295  app_worker_close_notify (app_wrk, s);
296  s->session_state = old_state;
297  return;
298  }
299  }
300 }
301 
302 static void
304 {
305  vnet_disconnect_args_t _a = { 0 }, *a = &_a;
307  app_worker_t *app_wrk;
308  session_t *s;
309  application_t *app;
310  u32 index, thread_index;
311 
312  mp = (session_reset_reply_msg_t *) data;
313  app = application_lookup (mp->context);
314  if (!app)
315  return;
316 
317  session_parse_handle (mp->handle, &index, &thread_index);
318  s = session_get_if_valid (index, thread_index);
319 
320  /* No session or not the right session */
321  if (!s || s->session_state < SESSION_STATE_TRANSPORT_CLOSING)
322  return;
323 
324  app_wrk = app_worker_get (s->app_wrk_index);
325  if (!app_wrk || app_wrk->app_index != app->app_index)
326  {
327  clib_warning ("App %u does not own handle 0x%lx!", app->app_index,
328  mp->handle);
329  return;
330  }
331 
332  /* Client objected to resetting the session, log and continue */
333  if (mp->retval)
334  {
335  clib_warning ("client retval %d", mp->retval);
336  return;
337  }
338 
339  /* This comes as a response to a reset, transport only waiting for
340  * confirmation to remove connection state, no need to disconnect */
341  a->handle = mp->handle;
342  a->app_index = app->app_index;
344 }
345 
346 static void
348 {
350  vnet_disconnect_args_t _a, *a = &_a;
351  svm_msg_q_msg_t _msg, *msg = &_msg;
353  app_worker_t *app_wrk;
354  session_event_t *evt;
355  session_t *s;
356  application_t *app;
357  int rv = 0;
358 
359  mp = (session_disconnected_msg_t *) data;
360  if (!(s = session_get_from_handle_if_valid (mp->handle)))
361  {
362  clib_warning ("could not disconnect handle %llu", mp->handle);
363  return;
364  }
365  app_wrk = app_worker_get (s->app_wrk_index);
366  app = application_lookup (mp->client_index);
367  if (!(app_wrk && app && app->app_index == app_wrk->app_index))
368  {
369  clib_warning ("could not disconnect session: %llu app: %u",
370  mp->handle, mp->client_index);
371  return;
372  }
373 
374  a->handle = mp->handle;
375  a->app_index = app_wrk->wrk_index;
376  rv = vnet_disconnect_session (a);
377 
380  SVM_Q_WAIT, msg);
381  evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
382  clib_memset (evt, 0, sizeof (*evt));
383  evt->event_type = SESSION_CTRL_EVT_DISCONNECTED_REPLY;
384  rmp = (session_disconnected_reply_msg_t *) evt->data;
385  rmp->handle = mp->handle;
386  rmp->context = mp->context;
387  rmp->retval = rv;
388  svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
389 }
390 
391 static void
393 {
395  vnet_disconnect_args_t _a, *a = &_a;
396  application_t *app;
397 
398  mp = (session_disconnected_reply_msg_t *) data;
399 
400  /* Client objected to disconnecting the session, log and continue */
401  if (mp->retval)
402  {
403  clib_warning ("client retval %d", mp->retval);
404  return;
405  }
406 
407  /* Disconnect has been confirmed. Confirm close to transport */
408  app = application_lookup (mp->context);
409  if (app)
410  {
411  a->handle = mp->handle;
412  a->app_index = app->app_index;
414  }
415 }
416 
417 static void
419 {
422  svm_msg_q_msg_t _msg, *msg = &_msg;
423  app_worker_t *app_wrk;
424  u32 owner_app_wrk_map;
425  session_event_t *evt;
426  session_t *s;
427  application_t *app;
428 
429  app = application_lookup (mp->client_index);
430  if (!app)
431  return;
432  if (!(s = session_get_from_handle_if_valid (mp->handle)))
433  {
434  clib_warning ("invalid handle %llu", mp->handle);
435  return;
436  }
437  app_wrk = app_worker_get (s->app_wrk_index);
438  if (app_wrk->app_index != app->app_index)
439  {
440  clib_warning ("app %u does not own session %llu", app->app_index,
441  mp->handle);
442  return;
443  }
444  owner_app_wrk_map = app_wrk->wrk_map_index;
445  app_wrk = application_get_worker (app, mp->wrk_index);
446 
447  /* This needs to come from the new owner */
448  if (mp->req_wrk_index == owner_app_wrk_map)
449  {
451 
454  SVM_Q_WAIT, msg);
455  evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
456  clib_memset (evt, 0, sizeof (*evt));
457  evt->event_type = SESSION_CTRL_EVT_REQ_WORKER_UPDATE;
458  wump = (session_req_worker_update_msg_t *) evt->data;
459  wump->session_handle = mp->handle;
460  svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
461  return;
462  }
463 
464  app_worker_own_session (app_wrk, s);
465 
466  /*
467  * Send reply
468  */
471  SVM_Q_WAIT, msg);
472  evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
473  clib_memset (evt, 0, sizeof (*evt));
474  evt->event_type = SESSION_CTRL_EVT_WORKER_UPDATE_REPLY;
475  rmp = (session_worker_update_reply_msg_t *) evt->data;
476  rmp->handle = mp->handle;
477  rmp->rx_fifo = pointer_to_uword (s->rx_fifo);
478  rmp->tx_fifo = pointer_to_uword (s->tx_fifo);
480  svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
481 
482  /*
483  * Retransmit messages that may have been lost
484  */
485  if (s->tx_fifo && !svm_fifo_is_empty (s->tx_fifo))
487 
488  if (s->rx_fifo && !svm_fifo_is_empty (s->rx_fifo))
490 
491  if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
492  app_worker_close_notify (app_wrk, s);
493 }
494 
496 
497 typedef struct
498 {
502 
503 /* packet trace format function */
504 static u8 *
505 format_session_queue_trace (u8 * s, va_list * args)
506 {
507  CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
508  CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
509  session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
510 
511  s = format (s, "session index %d thread index %d",
513  return s;
514 }
515 
516 #define foreach_session_queue_error \
517 _(TX, "Packets transmitted") \
518 _(TIMER, "Timer events") \
519 _(NO_BUFFER, "Out of buffers")
520 
521 typedef enum
522 {
523 #define _(sym,str) SESSION_QUEUE_ERROR_##sym,
525 #undef _
528 
529 static char *session_queue_error_strings[] = {
530 #define _(sym,string) string,
532 #undef _
533 };
534 
535 enum
536 {
540 };
541 
542 static void
544  u32 next_index, u32 * to_next, u16 n_segs,
545  session_t * s, u32 n_trace)
546 {
548  vlib_buffer_t *b;
549  int i;
550 
551  for (i = 0; i < clib_min (n_trace, n_segs); i++)
552  {
553  b = vlib_get_buffer (vm, to_next[i]);
554  vlib_trace_buffer (vm, node, next_index, b, 1 /* follow_chain */ );
555  t = vlib_add_trace (vm, node, b, sizeof (*t));
558  }
559  vlib_set_trace_count (vm, node, n_trace - i);
560 }
561 
562 always_inline void
564  vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
565 {
566  vlib_buffer_t *chain_b, *prev_b;
567  u32 chain_bi0, to_deq, left_from_seg;
568  session_worker_t *wrk;
569  u16 len_to_deq, n_bytes_read;
570  u8 *data, j;
571 
572  wrk = session_main_get_worker (ctx->s->thread_index);
573  b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
575 
576  chain_b = b;
577  left_from_seg = clib_min (ctx->sp.snd_mss - b->current_length,
578  ctx->left_to_snd);
579  to_deq = left_from_seg;
580  for (j = 1; j < ctx->n_bufs_per_seg; j++)
581  {
582  prev_b = chain_b;
583  len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
584 
585  *n_bufs -= 1;
586  chain_bi0 = wrk->tx_buffers[*n_bufs];
587  chain_b = vlib_get_buffer (vm, chain_bi0);
588  chain_b->current_data = 0;
589  data = vlib_buffer_get_current (chain_b);
590  if (peek_data)
591  {
592  n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo,
593  ctx->sp.tx_offset, len_to_deq, data);
594  ctx->sp.tx_offset += n_bytes_read;
595  }
596  else
597  {
598  if (ctx->transport_vft->transport_options.tx_type ==
600  {
601  svm_fifo_t *f = ctx->s->tx_fifo;
602  session_dgram_hdr_t *hdr = &ctx->hdr;
603  u16 deq_now;
604  deq_now = clib_min (hdr->data_length - hdr->data_offset,
605  len_to_deq);
606  n_bytes_read = svm_fifo_peek (f, hdr->data_offset, deq_now,
607  data);
608  ASSERT (n_bytes_read > 0);
609 
610  hdr->data_offset += n_bytes_read;
611  if (hdr->data_offset == hdr->data_length)
612  {
614  svm_fifo_dequeue_drop (f, offset);
615  if (ctx->left_to_snd > n_bytes_read)
616  svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
617  (u8 *) & ctx->hdr);
618  }
619  else if (ctx->left_to_snd == n_bytes_read)
620  svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
621  sizeof (session_dgram_pre_hdr_t));
622  }
623  else
624  n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
625  len_to_deq, data);
626  }
627  ASSERT (n_bytes_read == len_to_deq);
628  chain_b->current_length = n_bytes_read;
630 
631  /* update previous buffer */
632  prev_b->next_buffer = chain_bi0;
633  prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
634 
635  /* update current buffer */
636  chain_b->next_buffer = 0;
637 
638  to_deq -= n_bytes_read;
639  if (to_deq == 0)
640  break;
641  }
642  ASSERT (to_deq == 0
643  && b->total_length_not_including_first_buffer == left_from_seg);
644  ctx->left_to_snd -= left_from_seg;
645 }
646 
647 always_inline void
649  vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
650 {
651  u32 len_to_deq;
652  u8 *data0;
653  int n_bytes_read;
654 
655  /*
656  * Start with the first buffer in chain
657  */
658  b->error = 0;
659  b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
660  b->current_data = 0;
661 
663  len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
664 
665  if (peek_data)
666  {
667  n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset,
668  len_to_deq, data0);
669  ASSERT (n_bytes_read > 0);
670  /* Keep track of progress locally, transport is also supposed to
671  * increment it independently when pushing the header */
672  ctx->sp.tx_offset += n_bytes_read;
673  }
674  else
675  {
676  if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
677  {
678  session_dgram_hdr_t *hdr = &ctx->hdr;
679  svm_fifo_t *f = ctx->s->tx_fifo;
680  u16 deq_now;
681  u32 offset;
682 
683  ASSERT (hdr->data_length > hdr->data_offset);
684  deq_now = clib_min (hdr->data_length - hdr->data_offset,
685  len_to_deq);
686  offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
687  n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
688  ASSERT (n_bytes_read > 0);
689 
690  if (ctx->s->session_state == SESSION_STATE_LISTENING)
691  {
692  ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
693  ctx->tc->rmt_port = hdr->rmt_port;
694  }
695  hdr->data_offset += n_bytes_read;
696  if (hdr->data_offset == hdr->data_length)
697  {
698  offset = hdr->data_length + SESSION_CONN_HDR_LEN;
699  svm_fifo_dequeue_drop (f, offset);
700  if (ctx->left_to_snd > n_bytes_read)
701  svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
702  (u8 *) & ctx->hdr);
703  }
704  else if (ctx->left_to_snd == n_bytes_read)
705  svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
706  sizeof (session_dgram_pre_hdr_t));
707  }
708  else
709  {
710  n_bytes_read = svm_fifo_dequeue (ctx->s->tx_fifo,
711  len_to_deq, data0);
712  ASSERT (n_bytes_read > 0);
713  }
714  }
715  b->current_length = n_bytes_read;
716  ctx->left_to_snd -= n_bytes_read;
717 
718  /*
719  * Fill in the remaining buffers in the chain, if any
720  */
721  if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
722  session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
723 }
724 
727 {
728  if (peek_data)
729  {
730  if (PREDICT_TRUE (s->session_state == SESSION_STATE_READY))
731  return 0;
732  /* Can retransmit for closed sessions but can't send new data if
733  * session is not ready or closed */
734  else if (s->session_state < SESSION_STATE_READY)
735  return 1;
736  else if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
737  {
738  /* Allow closed transports to still send custom packets.
739  * For instance, tcp may want to send acks in time-wait. */
740  if (s->session_state != SESSION_STATE_TRANSPORT_DELETED
741  && (s->flags & SESSION_F_CUSTOM_TX))
742  return 0;
743  return 2;
744  }
745  }
746  return 0;
747 }
748 
751 {
752  if (peek_data)
753  {
754  return ctx->transport_vft->get_connection (ctx->s->connection_index,
755  ctx->s->thread_index);
756  }
757  else
758  {
759  if (ctx->s->session_state == SESSION_STATE_LISTENING)
760  return ctx->transport_vft->get_listener (ctx->s->connection_index);
761  else
762  {
763  return ctx->transport_vft->get_connection (ctx->s->connection_index,
764  ctx->s->thread_index);
765  }
766  }
767 }
768 
769 always_inline void
771  u32 max_segs, u8 peek_data)
772 {
773  u32 n_bytes_per_buf, n_bytes_per_seg;
774 
775  n_bytes_per_buf = vlib_buffer_get_default_data_size (vm);
777 
778  if (peek_data)
779  {
780  /* Offset in rx fifo from where to peek data */
781  if (PREDICT_FALSE (ctx->sp.tx_offset >= ctx->max_dequeue))
782  {
783  ctx->max_len_to_snd = 0;
784  return;
785  }
786  ctx->max_dequeue -= ctx->sp.tx_offset;
787  }
788  else
789  {
790  if (ctx->transport_vft->transport_options.tx_type == TRANSPORT_TX_DGRAM)
791  {
792  u32 len, chain_limit;
793 
794  if (ctx->max_dequeue <= sizeof (ctx->hdr))
795  {
796  ctx->max_len_to_snd = 0;
797  return;
798  }
799 
800  svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
801  (u8 *) & ctx->hdr);
802  ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
803  len = ctx->hdr.data_length - ctx->hdr.data_offset;
804 
805  /* Process multiple dgrams if smaller than min (buf_space, mss).
806  * This avoids handling multiple dgrams if they require buffer
807  * chains */
808  chain_limit = clib_min (n_bytes_per_buf - TRANSPORT_MAX_HDRS_LEN,
809  ctx->sp.snd_mss);
810  if (ctx->hdr.data_length <= chain_limit)
811  {
812  u32 first_dgram_len, dgram_len, offset, max_offset;
814 
815  ctx->sp.snd_mss = clib_min (ctx->sp.snd_mss, len);
816  offset = ctx->hdr.data_length + sizeof (session_dgram_hdr_t);
817  first_dgram_len = len;
818  max_offset = clib_min (ctx->max_dequeue, 16 << 10);
819 
820  while (offset < max_offset)
821  {
822  svm_fifo_peek (ctx->s->tx_fifo, offset, sizeof (ctx->hdr),
823  (u8 *) & hdr);
824  ASSERT (hdr.data_length > hdr.data_offset);
825  dgram_len = hdr.data_length - hdr.data_offset;
826  if (len + dgram_len > ctx->max_dequeue
827  || first_dgram_len != dgram_len)
828  break;
829  len += dgram_len;
830  offset += sizeof (hdr) + hdr.data_length;
831  }
832  }
833 
834  ctx->max_dequeue = len;
835  }
836  }
837  ASSERT (ctx->max_dequeue > 0);
838 
839  /* Ensure we're not writing more than transport window allows */
840  if (ctx->max_dequeue < ctx->sp.snd_space)
841  {
842  /* Constrained by tx queue. Try to send only fully formed segments */
843  ctx->max_len_to_snd = (ctx->max_dequeue > ctx->sp.snd_mss) ?
844  (ctx->max_dequeue - (ctx->max_dequeue % ctx->sp.snd_mss)) :
845  ctx->max_dequeue;
846  /* TODO Nagle ? */
847  }
848  else
849  {
850  /* Expectation is that snd_space0 is already a multiple of snd_mss */
851  ctx->max_len_to_snd = ctx->sp.snd_space;
852  }
853 
854  /* Check if we're tx constrained by the node */
855  ctx->n_segs_per_evt = ceil ((f64) ctx->max_len_to_snd / ctx->sp.snd_mss);
856  if (ctx->n_segs_per_evt > max_segs)
857  {
858  ctx->n_segs_per_evt = max_segs;
859  ctx->max_len_to_snd = max_segs * ctx->sp.snd_mss;
860  }
861 
862  ASSERT (n_bytes_per_buf > TRANSPORT_MAX_HDRS_LEN);
863  if (ctx->n_segs_per_evt > 1)
864  {
865  u32 n_bytes_last_seg, n_bufs_last_seg;
866 
867  n_bytes_per_seg = TRANSPORT_MAX_HDRS_LEN + ctx->sp.snd_mss;
868  n_bytes_last_seg = TRANSPORT_MAX_HDRS_LEN + ctx->max_len_to_snd
869  - ((ctx->n_segs_per_evt - 1) * ctx->sp.snd_mss);
870  ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
871  n_bufs_last_seg = ceil ((f64) n_bytes_last_seg / n_bytes_per_buf);
872  ctx->n_bufs_needed = ((ctx->n_segs_per_evt - 1) * ctx->n_bufs_per_seg)
873  + n_bufs_last_seg;
874  }
875  else
876  {
877  n_bytes_per_seg = TRANSPORT_MAX_HDRS_LEN + ctx->max_len_to_snd;
878  ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
879  ctx->n_bufs_needed = ctx->n_bufs_per_seg;
880  }
881 
882  ctx->deq_per_buf = clib_min (ctx->sp.snd_mss, n_bytes_per_buf);
883  ctx->deq_per_first_buf = clib_min (ctx->sp.snd_mss,
884  n_bytes_per_buf -
886 }
887 
888 always_inline void
891  session_evt_elt_t * elt)
892 {
893  session_t *s = ctx->s;
894 
897  if (svm_fifo_set_event (s->tx_fifo))
898  session_evt_add_head_old (wrk, elt);
899 }
900 
901 always_inline int
904  session_evt_elt_t * elt,
905  int *n_tx_packets, u8 peek_data)
906 {
907  u32 n_trace, n_left, pbi, next_index, max_burst;
908  session_tx_context_t *ctx = &wrk->ctx;
910  session_event_t *e = &elt->evt;
911  vlib_main_t *vm = wrk->vm;
913  vlib_buffer_t *pb;
914  u16 n_bufs, rv;
915 
916  if (PREDICT_FALSE ((rv = session_tx_not_ready (ctx->s, peek_data))))
917  {
918  if (rv < 2)
919  session_evt_add_old (wrk, elt);
920  return SESSION_TX_NO_DATA;
921  }
922 
923  next_index = smm->session_type_to_next[ctx->s->session_type];
924  max_burst = VLIB_FRAME_SIZE - *n_tx_packets;
925 
926  tp = session_get_transport_proto (ctx->s);
928  ctx->tc = session_tx_get_transport (ctx, peek_data);
929 
930  if (PREDICT_FALSE (e->event_type == SESSION_IO_EVT_TX_FLUSH))
931  {
932  if (ctx->transport_vft->flush_data)
933  ctx->transport_vft->flush_data (ctx->tc);
934  e->event_type = SESSION_IO_EVT_TX;
935  }
936 
937  if (ctx->s->flags & SESSION_F_CUSTOM_TX)
938  {
939  u32 n_custom_tx;
940  ctx->s->flags &= ~SESSION_F_CUSTOM_TX;
941  ctx->sp.max_burst_size = max_burst;
942  n_custom_tx = ctx->transport_vft->custom_tx (ctx->tc, &ctx->sp);
943  *n_tx_packets += n_custom_tx;
944  if (PREDICT_FALSE
945  (ctx->s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
946  return SESSION_TX_OK;
947  max_burst -= n_custom_tx;
948  if (!max_burst || (ctx->s->flags & SESSION_F_CUSTOM_TX))
949  {
950  session_evt_add_old (wrk, elt);
951  return SESSION_TX_OK;
952  }
953  }
954 
955  transport_connection_snd_params (ctx->tc, &ctx->sp);
956 
957  if (!ctx->sp.snd_space)
958  {
959  /* If the deschedule flag was set, remove session from scheduler.
960  * Transport is responsible for rescheduling this session. */
961  if (ctx->sp.flags & TRANSPORT_SND_F_DESCHED)
963  /* Request to postpone the session, e.g., zero-wnd and transport
964  * is not currently probing */
965  else if (ctx->sp.flags & TRANSPORT_SND_F_POSTPONE)
966  session_evt_add_old (wrk, elt);
967  /* This flow queue is "empty" so it should be re-evaluated before
968  * the ones that have data to send. */
969  else
970  session_evt_add_head_old (wrk, elt);
971 
972  return SESSION_TX_NO_DATA;
973  }
974 
976  {
977  u32 snd_space = transport_connection_tx_pacer_burst (ctx->tc);
978  if (snd_space < TRANSPORT_PACER_MIN_BURST)
979  {
980  session_evt_add_head_old (wrk, elt);
981  return SESSION_TX_NO_DATA;
982  }
983  snd_space = clib_min (ctx->sp.snd_space, snd_space);
984  ctx->sp.snd_space = snd_space >= ctx->sp.snd_mss ?
985  snd_space - snd_space % ctx->sp.snd_mss : snd_space;
986  }
987 
988  /* Check how much we can pull. */
989  session_tx_set_dequeue_params (vm, ctx, max_burst, peek_data);
990 
991  if (PREDICT_FALSE (!ctx->max_len_to_snd))
992  {
994  session_tx_maybe_reschedule (wrk, ctx, elt);
995  return SESSION_TX_NO_DATA;
996  }
997 
1000  n_bufs = vlib_buffer_alloc (vm, wrk->tx_buffers, ctx->n_bufs_needed);
1001  if (PREDICT_FALSE (n_bufs < ctx->n_bufs_needed))
1002  {
1003  if (n_bufs)
1004  vlib_buffer_free (vm, wrk->tx_buffers, n_bufs);
1005  session_evt_add_head_old (wrk, elt);
1007  SESSION_QUEUE_ERROR_NO_BUFFER, 1);
1008  return SESSION_TX_NO_BUFFERS;
1009  }
1010 
1012 
1013  ctx->left_to_snd = ctx->max_len_to_snd;
1014  n_left = ctx->n_segs_per_evt;
1015 
1016  while (n_left >= 4)
1017  {
1018  vlib_buffer_t *b0, *b1;
1019  u32 bi0, bi1;
1020 
1021  pbi = wrk->tx_buffers[n_bufs - 3];
1022  pb = vlib_get_buffer (vm, pbi);
1023  vlib_prefetch_buffer_header (pb, STORE);
1024  pbi = wrk->tx_buffers[n_bufs - 4];
1025  pb = vlib_get_buffer (vm, pbi);
1026  vlib_prefetch_buffer_header (pb, STORE);
1027 
1028  bi0 = wrk->tx_buffers[--n_bufs];
1029  bi1 = wrk->tx_buffers[--n_bufs];
1030 
1031  b0 = vlib_get_buffer (vm, bi0);
1032  b1 = vlib_get_buffer (vm, bi1);
1033 
1034  session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
1035  session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
1036 
1037  ctx->transport_vft->push_header (ctx->tc, b0);
1038  ctx->transport_vft->push_header (ctx->tc, b1);
1039 
1040  n_left -= 2;
1041 
1044 
1045  vec_add1 (wrk->pending_tx_buffers, bi0);
1046  vec_add1 (wrk->pending_tx_buffers, bi1);
1047  vec_add1 (wrk->pending_tx_nexts, next_index);
1048  vec_add1 (wrk->pending_tx_nexts, next_index);
1049  }
1050  while (n_left)
1051  {
1052  vlib_buffer_t *b0;
1053  u32 bi0;
1054 
1055  if (n_left > 1)
1056  {
1057  pbi = wrk->tx_buffers[n_bufs - 2];
1058  pb = vlib_get_buffer (vm, pbi);
1059  vlib_prefetch_buffer_header (pb, STORE);
1060  }
1061 
1062  bi0 = wrk->tx_buffers[--n_bufs];
1063  b0 = vlib_get_buffer (vm, bi0);
1064  session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
1065 
1066  /* Ask transport to push header after current_length and
1067  * total_length_not_including_first_buffer are updated */
1068  ctx->transport_vft->push_header (ctx->tc, b0);
1069 
1070  n_left -= 1;
1071 
1073 
1074  vec_add1 (wrk->pending_tx_buffers, bi0);
1075  vec_add1 (wrk->pending_tx_nexts, next_index);
1076  }
1077 
1078  if (PREDICT_FALSE ((n_trace = vlib_get_trace_count (vm, node)) > 0))
1079  session_tx_trace_frame (vm, node, next_index, wrk->pending_tx_buffers,
1080  ctx->n_segs_per_evt, ctx->s, n_trace);
1081 
1082  if (PREDICT_FALSE (n_bufs))
1083  vlib_buffer_free (vm, wrk->tx_buffers, n_bufs);
1084 
1085  *n_tx_packets += ctx->n_segs_per_evt;
1086 
1087  SESSION_EVT (SESSION_EVT_DEQ, ctx->s, ctx->max_len_to_snd, ctx->max_dequeue,
1088  ctx->s->tx_fifo->has_event, wrk->last_vlib_time);
1089 
1090  ASSERT (ctx->left_to_snd == 0);
1091 
1092  /* If we couldn't dequeue all bytes reschedule as old flow. Otherwise,
1093  * check if application enqueued more data and reschedule accordingly */
1094  if (ctx->max_len_to_snd < ctx->max_dequeue)
1095  session_evt_add_old (wrk, elt);
1096  else
1097  session_tx_maybe_reschedule (wrk, ctx, elt);
1098 
1099  if (!peek_data)
1100  {
1101  if (svm_fifo_needs_deq_ntf (ctx->s->tx_fifo, ctx->max_len_to_snd))
1102  session_dequeue_notify (ctx->s);
1103  }
1104  return SESSION_TX_OK;
1105 }
1106 
1107 int
1110  session_evt_elt_t * e, int *n_tx_packets)
1111 {
1112  return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 1);
1113 }
1114 
1115 int
1118  session_evt_elt_t * e, int *n_tx_packets)
1119 {
1120  return session_tx_fifo_read_and_snd_i (wrk, node, e, n_tx_packets, 0);
1121 }
1122 
1123 int
1126  session_evt_elt_t * elt, int *n_tx_packets)
1127 {
1128  transport_send_params_t *sp = &wrk->ctx.sp;
1129  session_t *s = wrk->ctx.s;
1130  u32 n_packets;
1131 
1132  if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
1133  return 0;
1134 
1135  /* Clear custom-tx flag used to request reschedule for tx */
1136  s->flags &= ~SESSION_F_CUSTOM_TX;
1137 
1138  sp->max_burst_size = clib_min (VLIB_FRAME_SIZE - *n_tx_packets,
1140 
1141  n_packets = transport_custom_tx (session_get_transport_proto (s), s, sp);
1142  *n_tx_packets += n_packets;
1143 
1144  if (s->flags & SESSION_F_CUSTOM_TX)
1145  {
1146  session_evt_add_old (wrk, elt);
1147  }
1148  else if (!(sp->flags & TRANSPORT_SND_F_DESCHED))
1149  {
1152  if (svm_fifo_set_event (s->tx_fifo))
1153  session_evt_add_head_old (wrk, elt);
1154  }
1155 
1156  return n_packets;
1157 }
1158 
1160 session_event_get_session (session_event_t * e, u8 thread_index)
1161 {
1162  return session_get_if_valid (e->session_index, thread_index);
1163 }
1164 
1165 always_inline void
1167 {
1168  clib_llist_index_t ei;
1169  void (*fp) (void *);
1170  session_event_t *e;
1171  session_t *s;
1172 
1173  ei = clib_llist_entry_index (wrk->event_elts, elt);
1174  e = &elt->evt;
1175 
1176  switch (e->event_type)
1177  {
1178  case SESSION_CTRL_EVT_RPC:
1179  fp = e->rpc_args.fp;
1180  (*fp) (e->rpc_args.arg);
1181  break;
1183  s = session_get_from_handle_if_valid (e->session_handle);
1184  if (PREDICT_FALSE (!s))
1185  break;
1187  break;
1189  s = session_get_from_handle_if_valid (e->session_handle);
1190  if (PREDICT_FALSE (!s))
1191  break;
1193  break;
1196  break;
1199  break;
1202  break;
1205  break;
1208  break;
1211  break;
1214  break;
1217  break;
1220  elt));
1221  break;
1224  break;
1227  break;
1230  break;
1231  default:
1232  clib_warning ("unhandled event type %d", e->event_type);
1233  }
1234 
1235  /* Regrab elements in case pool moved */
1236  elt = pool_elt_at_index (wrk->event_elts, ei);
1237  if (!clib_llist_elt_is_linked (elt, evt_list))
1238  {
1239  e = &elt->evt;
1240  if (e->event_type >= SESSION_CTRL_EVT_BOUND)
1241  session_evt_ctrl_data_free (wrk, elt);
1242  session_evt_elt_free (wrk, elt);
1243  }
1244  SESSION_EVT (SESSION_EVT_COUNTS, CNT_CTRL_EVTS, 1, wrk);
1245 }
1246 
1247 always_inline void
1249  session_evt_elt_t * elt, u32 thread_index,
1250  int *n_tx_packets)
1251 {
1252  session_main_t *smm = &session_main;
1253  app_worker_t *app_wrk;
1254  clib_llist_index_t ei;
1255  session_event_t *e;
1256  session_t *s;
1257 
1258  ei = clib_llist_entry_index (wrk->event_elts, elt);
1259  e = &elt->evt;
1260 
1261  switch (e->event_type)
1262  {
1264  case SESSION_IO_EVT_TX:
1265  s = session_event_get_session (e, thread_index);
1266  if (PREDICT_FALSE (!s))
1267  break;
1269  wrk->ctx.s = s;
1270  /* Spray packets in per session type frames, since they go to
1271  * different nodes */
1272  (smm->session_tx_fns[s->session_type]) (wrk, node, elt, n_tx_packets);
1273  break;
1274  case SESSION_IO_EVT_RX:
1275  s = session_event_get_session (e, thread_index);
1276  if (!s)
1277  break;
1280  break;
1282  s = session_event_get_session (e, thread_index);
1283  if (PREDICT_FALSE (!s || s->session_state >= SESSION_STATE_CLOSING))
1284  break;
1286  app_wrk = app_worker_get (s->app_wrk_index);
1287  app_worker_builtin_rx (app_wrk, s);
1288  break;
1290  s = session_get_from_handle_if_valid (e->session_handle);
1291  wrk->ctx.s = s;
1292  if (PREDICT_TRUE (s != 0))
1293  session_tx_fifo_dequeue_internal (wrk, node, elt, n_tx_packets);
1294  break;
1295  default:
1296  clib_warning ("unhandled event type %d", e->event_type);
1297  }
1298 
1299  SESSION_EVT (SESSION_IO_EVT_COUNTS, e->event_type, 1, wrk);
1300 
1301  /* Regrab elements in case pool moved */
1302  elt = pool_elt_at_index (wrk->event_elts, ei);
1303  if (!clib_llist_elt_is_linked (elt, evt_list))
1304  session_evt_elt_free (wrk, elt);
1305 }
1306 
1307 /* *INDENT-OFF* */
1308 static const u32 session_evt_msg_sizes[] = {
1309 #define _(symc, sym) \
1310  [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t),
1312 #undef _
1313 };
1314 /* *INDENT-ON* */
1315 
1316 always_inline void
1317 session_evt_add_to_list (session_worker_t * wrk, session_event_t * evt)
1318 {
1319  session_evt_elt_t *elt;
1320 
1321  if (evt->event_type >= SESSION_CTRL_EVT_RPC)
1322  {
1323  elt = session_evt_alloc_ctrl (wrk);
1324  if (evt->event_type >= SESSION_CTRL_EVT_BOUND)
1325  {
1326  elt->evt.ctrl_data_index = session_evt_ctrl_data_alloc (wrk);
1327  elt->evt.event_type = evt->event_type;
1328  clib_memcpy_fast (session_evt_ctrl_data (wrk, elt), evt->data,
1329  session_evt_msg_sizes[evt->event_type]);
1330  }
1331  else
1332  {
1333  /* Internal control events fit into io events footprint */
1334  clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1335  }
1336  }
1337  else
1338  {
1339  elt = session_evt_alloc_new (wrk);
1340  clib_memcpy_fast (&elt->evt, evt, sizeof (elt->evt));
1341  }
1342 }
1343 
1344 static void
1347 {
1349  wrk->pending_tx_nexts,
1350  vec_len (wrk->pending_tx_nexts));
1353 }
1354 
1355 static uword
1357  vlib_frame_t * frame)
1358 {
1360  u32 thread_index = vm->thread_index, n_to_dequeue;
1361  session_worker_t *wrk = &smm->wrk[thread_index];
1362  session_evt_elt_t *elt, *ctrl_he, *new_he, *old_he;
1363  clib_llist_index_t ei, next_ei, old_ti;
1364  svm_msg_q_msg_t _msg, *msg = &_msg;
1365  int i = 0, n_tx_packets;
1366  session_event_t *evt;
1367  svm_msg_q_t *mq;
1368 
1369  SESSION_EVT (SESSION_EVT_DISPATCH_START, wrk);
1370 
1371  wrk->last_vlib_time = vlib_time_now (vm);
1373 
1374  /*
1375  * Update transport time
1376  */
1377  transport_update_time (wrk->last_vlib_time, thread_index);
1378  n_tx_packets = vec_len (wrk->pending_tx_buffers);
1379  SESSION_EVT (SESSION_EVT_DSP_CNTRS, UPDATE_TIME, wrk);
1380 
1381  /*
1382  * Dequeue and handle new events
1383  */
1384 
1385  /* Try to dequeue what is available. Don't wait for lock.
1386  * XXX: we may need priorities here */
1387  mq = wrk->vpp_event_queue;
1388  n_to_dequeue = svm_msg_q_size (mq);
1389  if (n_to_dequeue && svm_msg_q_try_lock (mq) == 0)
1390  {
1391  for (i = 0; i < n_to_dequeue; i++)
1392  {
1393  svm_msg_q_sub_w_lock (mq, msg);
1394  evt = svm_msg_q_msg_data (mq, msg);
1395  session_evt_add_to_list (wrk, evt);
1396  svm_msg_q_free_msg (mq, msg);
1397  }
1398  svm_msg_q_unlock (mq);
1399  }
1400 
1401  SESSION_EVT (SESSION_EVT_DSP_CNTRS, MQ_DEQ, wrk, n_to_dequeue, !i);
1402 
1403  /*
1404  * Handle control events
1405  */
1406 
1407  ctrl_he = pool_elt_at_index (wrk->event_elts, wrk->ctrl_head);
1408 
1409  /* *INDENT-OFF* */
1410  clib_llist_foreach_safe (wrk->event_elts, evt_list, ctrl_he, elt, ({
1411  clib_llist_remove (wrk->event_elts, evt_list, elt);
1412  session_event_dispatch_ctrl (wrk, elt);
1413  }));
1414  /* *INDENT-ON* */
1415 
1416  SESSION_EVT (SESSION_EVT_DSP_CNTRS, CTRL_EVTS, wrk);
1417 
1418  /*
1419  * Handle the new io events.
1420  */
1421 
1422  new_he = pool_elt_at_index (wrk->event_elts, wrk->new_head);
1423  old_he = pool_elt_at_index (wrk->event_elts, wrk->old_head);
1424  old_ti = clib_llist_prev_index (old_he, evt_list);
1425 
1426  ei = clib_llist_next_index (new_he, evt_list);
1427  while (ei != wrk->new_head && n_tx_packets < VLIB_FRAME_SIZE)
1428  {
1429  elt = pool_elt_at_index (wrk->event_elts, ei);
1430  ei = clib_llist_next_index (elt, evt_list);
1431  clib_llist_remove (wrk->event_elts, evt_list, elt);
1432  session_event_dispatch_io (wrk, node, elt, thread_index, &n_tx_packets);
1433  }
1434 
1435  SESSION_EVT (SESSION_EVT_DSP_CNTRS, NEW_IO_EVTS, wrk);
1436 
1437  /*
1438  * Handle the old io events, if we had any prior to processing the new ones
1439  */
1440 
1441  if (old_ti != wrk->old_head)
1442  {
1443  old_he = pool_elt_at_index (wrk->event_elts, wrk->old_head);
1444  ei = clib_llist_next_index (old_he, evt_list);
1445 
1446  while (n_tx_packets < VLIB_FRAME_SIZE)
1447  {
1448  elt = pool_elt_at_index (wrk->event_elts, ei);
1449  next_ei = clib_llist_next_index (elt, evt_list);
1450  clib_llist_remove (wrk->event_elts, evt_list, elt);
1451 
1452  session_event_dispatch_io (wrk, node, elt, thread_index,
1453  &n_tx_packets);
1454 
1455  if (ei == old_ti)
1456  break;
1457 
1458  ei = next_ei;
1459  };
1460  }
1461 
1462  SESSION_EVT (SESSION_EVT_DSP_CNTRS, OLD_IO_EVTS, wrk);
1463 
1464  if (vec_len (wrk->pending_tx_buffers))
1466 
1468  SESSION_QUEUE_ERROR_TX, n_tx_packets);
1469 
1470  SESSION_EVT (SESSION_EVT_DISPATCH_END, wrk, n_tx_packets);
1471 
1472  return n_tx_packets;
1473 }
1474 
1475 /* *INDENT-OFF* */
1477 {
1478  .function = session_queue_node_fn,
1480  .name = "session-queue",
1481  .format_trace = format_session_queue_trace,
1482  .type = VLIB_NODE_TYPE_INPUT,
1483  .n_errors = ARRAY_LEN (session_queue_error_strings),
1484  .error_strings = session_queue_error_strings,
1485  .state = VLIB_NODE_STATE_DISABLED,
1486 };
1487 /* *INDENT-ON* */
1488 
1489 static clib_error_t *
1491 {
1492  if (vec_len (vlib_mains) < 2)
1493  return 0;
1494 
1495  /*
1496  * Shut off (especially) worker-thread session nodes.
1497  * Otherwise, vpp can crash as the main thread unmaps the
1498  * API segment.
1499  */
1501  session_node_enable_disable (0 /* is_enable */ );
1503  return 0;
1504 }
1505 
1507 
1508 static uword
1510 {
1512 
1513  node = vlib_node_get_runtime (vm, session_queue_node.index);
1514  return session_queue_node_fn (vm, node, 0);
1515 }
1516 
1517 static uword
1519  vlib_frame_t * f)
1520 {
1521  uword *event_data = 0;
1522  f64 timeout = 1.0;
1523  uword event_type;
1524 
1525  while (1)
1526  {
1528  event_type = vlib_process_get_events (vm, (uword **) & event_data);
1529 
1530  switch (event_type)
1531  {
1533  /* Run session queue node on main thread */
1535  break;
1537  timeout = 100000.0;
1538  break;
1539  case ~0:
1540  /* Timed out. Run on main to ensure all events are handled */
1542  break;
1543  }
1544  vec_reset_length (event_data);
1545  }
1546  return 0;
1547 }
1548 
1549 /* *INDENT-OFF* */
1551 {
1552  .function = session_queue_process,
1553  .type = VLIB_NODE_TYPE_PROCESS,
1554  .name = "session-queue-process",
1555  .state = VLIB_NODE_STATE_DISABLED,
1556 };
1557 /* *INDENT-ON* */
1558 
1561  vlib_frame_t * frame)
1562 {
1564  if (!sm->wrk[0].vpp_event_queue)
1565  return 0;
1566  node = vlib_node_get_runtime (vm, session_queue_node.index);
1567  return session_queue_node_fn (vm, node, frame);
1568 }
1569 
1570 /* *INDENT-OFF* */
1572 {
1573  .function = session_queue_pre_input_inline,
1574  .type = VLIB_NODE_TYPE_PRE_INPUT,
1575  .name = "session-queue-main",
1576  .state = VLIB_NODE_STATE_DISABLED,
1577 };
1578 /* *INDENT-ON* */
1579 
1580 /*
1581  * fd.io coding-style-patch-verification: ON
1582  *
1583  * Local Variables:
1584  * eval: (c-set-style "gnu")
1585  * End:
1586  */
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
Definition: vec.h:507
static void session_tx_trace_frame(vlib_main_t *vm, vlib_node_runtime_t *node, u32 next_index, u32 *to_next, u16 n_segs, session_t *s, u32 n_trace)
Definition: session_node.c:543
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:124
int app_worker_lock_and_send_event(app_worker_t *app, session_t *s, u8 evt_type)
Send event to application.
u32 connection_index
Index of the transport connection associated to the session.
#define ENDPOINT_INVALID_INDEX
void session_transport_reset(session_t *s)
Force transport close.
Definition: session.c:1429
void * svm_msg_q_msg_data(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Get data for message in queue.
#define clib_min(x, y)
Definition: clib.h:319
#define CLIB_UNUSED(x)
Definition: clib.h:86
static void app_mq_detach_handler(void *data)
Definition: session_node.c:193
session_type_t session_type
Type built from transport and network protocol types.
static void session_mq_unlisten_handler(void *data)
Definition: session_node.c:211
void ip_copy(ip46_address_t *dst, ip46_address_t *src, u8 is_ip4)
Definition: ip.c:81
int mq_send_session_connected_cb(u32 app_wrk_index, u32 api_context, session_t *s, session_error_t err)
Definition: session_api.c:260
static u32 vlib_get_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt)
Definition: trace_funcs.h:187
svm_msg_q_t * vpp_event_queue
vpp event message queue for worker
Definition: session.h:80
static f64 vlib_process_wait_for_event_or_clock(vlib_main_t *vm, f64 dt)
Suspend a cooperative multi-tasking thread Waits for an event, or for the indicated number of seconds...
Definition: node_funcs.h:673
vlib_node_registration_t session_queue_pre_input_node
(constructor) VLIB_REGISTER_NODE (session_queue_pre_input_node)
a
Definition: bitmap.h:538
static void vlib_buffer_free(vlib_main_t *vm, u32 *buffers, u32 n_buffers)
Free buffers Frees the entire buffer chain for each buffer.
Definition: buffer_funcs.h:937
svm_fifo_t * tx_fifo
session_main_t session_main
Definition: session.c:25
struct _vnet_connect_args vnet_connect_args_t
static void ip46_address_mask_ip4(ip46_address_t *ip46)
Definition: ip46_address.h:61
struct _vnet_unlisten_args_t vnet_unlisten_args_t
#define clib_llist_next_index(E, name)
Definition: llist.h:69
#define PREDICT_TRUE(x)
Definition: clib.h:119
u32 session_index
Index in thread pool where session was allocated.
i16 current_data
signed offset in data[], pre_data[] that we are currently processing.
Definition: buffer.h:110
#define clib_memcpy_fast(a, b, c)
Definition: string.h:81
static int svm_fifo_is_empty_prod(svm_fifo_t *f)
Check if fifo is empty optimized for producer.
Definition: svm_fifo.h:481
session_t * s
Definition: session.h:45
static void session_event_dispatch_ctrl(session_worker_t *wrk, session_evt_elt_t *elt)
clib_memset(h->entries, 0, sizeof(h->entries[0]) *entries)
#define VLIB_NODE_FLAG_TRACE_SUPPORTED
Definition: node.h:308
static f64 vlib_time_now(vlib_main_t *vm)
Definition: main.h:291
static void session_mq_listen_uri_handler(void *data)
Definition: session_node.c:74
svm_fifo_t * rx_fifo
Pointers to rx/tx buffers.
session_worker_t * wrk
Worker contexts.
Definition: session.h:143
static session_t * session_get_if_valid(u64 si, u32 thread_index)
Definition: session.h:308
u32 thread_index
Definition: main.h:218
u16 current_length
Nbytes between current data and the end of this buffer.
Definition: buffer.h:113
#define vec_add1(V, E)
Add 1 element to end of vector (unspecified alignment).
Definition: vec.h:590
static transport_proto_t session_get_transport_proto(session_t *s)
application_t * application_lookup(u32 api_client_index)
Definition: application.c:394
int svm_fifo_peek(svm_fifo_t *f, u32 offset, u32 len, u8 *dst)
Peek data from fifo.
Definition: svm_fifo.c:1038
u32 wrk_map_index
Worker index in app&#39;s map pool.
Definition: application.h:40
static clib_error_t * session_queue_exit(vlib_main_t *vm)
static uword session_queue_node_fn(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
u8 * format(u8 *s, const char *fmt,...)
Definition: format.c:424
int vnet_unlisten(vnet_unlisten_args_t *a)
Definition: application.c:1056
session_evt_elt_t * event_elts
Pool of session event list elements.
Definition: session.h:101
static void * session_evt_ctrl_data(session_worker_t *wrk, session_evt_elt_t *elt)
Definition: session.h:260
void svm_fifo_overwrite_head(svm_fifo_t *f, u8 *src, u32 len)
Overwrite fifo head with new data.
Definition: svm_fifo.c:771
#define vec_validate_aligned(V, I, A)
Make sure vector is long enough for given index (no header, specified alignment)
Definition: vec.h:518
int session_tx_fifo_dequeue_and_snd(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *e, int *n_tx_packets)
u32 flags
Session flags.
static u32 transport_connection_snd_params(transport_connection_t *tc, transport_send_params_t *sp)
Get send parameters for transport connection.
Definition: transport.h:195
void session_node_enable_disable(u8 is_en)
Definition: session.c:1731
vlib_main_t ** vlib_mains
Definition: buffer.c:332
static void session_parse_handle(session_handle_t handle, u32 *index, u32 *thread_index)
unsigned char u8
Definition: types.h:56
app_worker_t * application_get_worker(application_t *app, u32 wrk_map_index)
Definition: application.c:663
transport_send_params_t sp
Definition: session.h:48
u16 * pending_tx_nexts
Vector of nexts for the pending tx buffers.
Definition: session.h:122
static void session_tx_fill_buffer(vlib_main_t *vm, session_tx_context_t *ctx, vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
Definition: session_node.c:648
static u32 session_evt_ctrl_data_alloc(session_worker_t *wrk)
Definition: session.h:242
struct _vnet_bind_args_t vnet_listen_args_t
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
double f64
Definition: types.h:142
#define vlib_worker_thread_barrier_sync(X)
Definition: threads.h:204
void transport_update_time(clib_time_type_t time_now, u8 thread_index)
Definition: transport.c:773
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:130
static void session_mq_disconnect_handler(void *data)
Definition: session_node.c:177
u32 * tx_buffers
Vector of tx buffer free lists.
Definition: session.h:98
int svm_msg_q_lock_and_alloc_msg_w_ring(svm_msg_q_t *mq, u32 ring_index, u8 noblock, svm_msg_q_msg_t *msg)
Lock message queue and allocate message buffer on ring.
static int svm_fifo_is_empty(svm_fifo_t *f)
Check if fifo is empty.
Definition: svm_fifo.h:494
static session_worker_t * session_main_get_worker(u32 thread_index)
Definition: session.h:619
#define static_always_inline
Definition: clib.h:106
vlib_main_t * vm
Convenience pointer to this worker&#39;s vlib_main.
Definition: session.h:89
static uword vlib_process_get_events(vlib_main_t *vm, uword **data_vector)
Return the first event type which has occurred and a vector of per-event data of that type...
Definition: node_funcs.h:516
struct _vnet_disconnect_args_t vnet_disconnect_args_t
static void session_mq_connect_uri_handler(void *data)
Definition: session_node.c:150
clib_llist_index_t new_head
Head of list of elements.
Definition: session.h:110
static u32 svm_fifo_max_dequeue_cons(svm_fifo_t *f)
Fifo max bytes to dequeue optimized for consumer.
Definition: svm_fifo.h:402
u32 * session_type_to_next
Per session type output nodes.
Definition: session.h:157
#define vlib_prefetch_buffer_header(b, type)
Prefetch buffer metadata.
Definition: buffer.h:203
unsigned int u32
Definition: types.h:88
int svm_fifo_dequeue(svm_fifo_t *f, u32 len, u8 *dst)
Dequeue data from fifo.
Definition: svm_fifo.c:1007
int session_send_io_evt_to_thread(svm_fifo_t *f, session_evt_type_t evt_type)
Definition: session.c:79
#define VLIB_FRAME_SIZE
Definition: node.h:380
#define TRANSPORT_PACER_MAX_BURST_PKTS
Definition: transport.h:25
session_state_t
static char * session_queue_error_strings[]
Definition: session_node.c:529
int session_dequeue_notify(session_t *s)
Definition: session.c:685
transport_proto_vft_t * transport_protocol_get_vft(transport_proto_t transport_proto)
Get transport virtual function table.
Definition: transport.c:282
vlib_error_t error
Error code for buffers to be enqueued to error handler.
Definition: buffer.h:136
#define TRANSPORT_MAX_HDRS_LEN
#define pool_elt_at_index(p, i)
Returns pointer to element at given index.
Definition: pool.h:534
#define CLIB_US_TIME_FREQ
Definition: time.h:207
session_event_t evt
Definition: session.h:64
static void * vlib_buffer_make_headroom(vlib_buffer_t *b, u8 size)
Make head room, typically for packet headers.
Definition: buffer.h:350
session_queue_error_t
Definition: session_node.c:521
transport_proto_vft_t * transport_vft
Definition: session.h:46
long ctx[MAX_CONNS]
Definition: main.c:144
unsigned short u16
Definition: types.h:57
static void session_evt_ctrl_data_free(session_worker_t *wrk, session_evt_elt_t *elt)
Definition: session.h:267
static void * vlib_buffer_get_current(vlib_buffer_t *b)
Get pointer to current data to process.
Definition: buffer.h:229
static uword session_queue_process(vlib_main_t *vm, vlib_node_runtime_t *rt, vlib_frame_t *f)
int mq_send_session_bound_cb(u32 app_wrk_index, u32 api_context, session_handle_t handle, int rv)
Definition: session_api.c:340
int session_tx_fifo_peek_and_snd(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *e, int *n_tx_packets)
#define PREDICT_FALSE(x)
Definition: clib.h:118
#define always_inline
Definition: ipsec.h:28
static void svm_fifo_unset_event(svm_fifo_t *f)
Unset fifo event flag.
Definition: svm_fifo.h:712
static u8 transport_connection_is_tx_paced(transport_connection_t *tc)
Check if transport connection is paced.
Definition: transport.h:317
u32 wrk_index
Worker index in global worker pool.
Definition: application.h:37
u32 node_index
Node index.
Definition: node.h:498
session_fifo_rx_fn ** session_tx_fns
Per transport rx function that can either dequeue or peek.
Definition: session.h:152
int session_tx_fifo_dequeue_internal(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, int *n_tx_packets)
u64 session_segment_handle(session_t *s)
Definition: session.c:1547
vlib_main_t * vm
Definition: in2out_ed.c:1599
static void vlib_node_increment_counter(vlib_main_t *vm, u32 node_index, u32 counter_index, u64 increment)
Definition: node_funcs.h:1150
u8 len
Definition: ip_types.api:92
static session_t * session_get_from_handle_if_valid(session_handle_t handle)
Definition: session.h:330
static_always_inline u32 vlib_buffer_get_default_data_size(vlib_main_t *vm)
Definition: buffer_funcs.h:96
static u8 svm_fifo_set_event(svm_fifo_t *f)
Set fifo event flag.
Definition: svm_fifo.h:699
#define foreach_session_ctrl_evt
The fine-grained event logger allows lightweight, thread-safe event logging at minimum cost...
static void session_evt_elt_free(session_worker_t *wrk, session_evt_elt_t *elt)
Definition: session.h:221
#define VLIB_REGISTER_NODE(x,...)
Definition: node.h:169
static u8 svm_fifo_needs_deq_ntf(svm_fifo_t *f, u32 n_last_deq)
Check if fifo needs dequeue notification.
Definition: svm_fifo.h:790
static void session_tx_fifo_chain_tail(vlib_main_t *vm, session_tx_context_t *ctx, vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
Definition: session_node.c:563
static void session_evt_add_to_list(session_worker_t *wrk, session_event_t *evt)
static_always_inline uword vlib_get_thread_index(void)
Definition: threads.h:218
#define CLIB_PREFETCH(addr, size, type)
Definition: cache.h:80
static void svm_msg_q_unlock(svm_msg_q_t *mq)
Unlock message queue.
static void session_event_dispatch_io(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, u32 thread_index, int *n_tx_packets)
static_always_inline void vlib_buffer_enqueue_to_next(vlib_main_t *vm, vlib_node_runtime_t *node, u32 *buffers, u16 *nexts, uword count)
Definition: buffer_node.h:339
sll srl srl sll sra u16x4 i
Definition: vector_sse42.h:317
#define vec_free(V)
Free vector&#39;s memory (no header).
Definition: vec.h:380
#define SESSION_CONN_HDR_LEN
int app_worker_builtin_rx(app_worker_t *app_wrk, session_t *s)
#define VLIB_MAIN_LOOP_EXIT_FUNCTION(x)
Definition: init.h:178
#define clib_warning(format, args...)
Definition: error.h:59
static vlib_node_runtime_t * vlib_node_get_runtime(vlib_main_t *vm, u32 node_index)
Get node runtime by node index.
Definition: node_funcs.h:89
struct _transport_connection transport_connection_t
static session_evt_elt_t * session_evt_alloc_new(session_worker_t *wrk)
Definition: session.h:274
static_always_inline uword session_queue_pre_input_inline(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
#define ARRAY_LEN(x)
Definition: clib.h:66
blocking call - best used in combination with condvars, for eventfds we don&#39;t yield the cpu ...
Definition: queue.h:42
u32 * pending_tx_buffers
Vector of buffers to be sent.
Definition: session.h:119
static int svm_msg_q_try_lock(svm_msg_q_t *mq)
Try locking message queue.
vlib_main_t vlib_node_runtime_t * node
Definition: in2out_ed.c:1599
static u32 session_thread_from_handle(session_handle_t handle)
transport_snd_flags_t flags
Definition: transport.h:61
static int transport_custom_tx(transport_proto_t tp, void *s, transport_send_params_t *sp)
Definition: transport.h:168
#define foreach_session_queue_error
Definition: session_node.c:516
#define ASSERT(truth)
static void session_mq_disconnected_reply_handler(void *data)
Definition: session_node.c:392
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.
static session_t * session_event_get_session(session_event_t *e, u8 thread_index)
void transport_connection_tx_pacer_reset_bucket(transport_connection_t *tc, u32 bucket)
Reset tx pacer bucket.
Definition: transport.c:705
int ct_session_connect_notify(session_t *ss)
u8 data[128]
Definition: ipsec_types.api:89
datagram mode
session_tx_context_t ctx
Context for session tx.
Definition: session.h:95
void transport_connection_update_tx_bytes(transport_connection_t *tc, u32 bytes)
Update tx bytes for paced transport connection.
Definition: transport.c:742
int vnet_listen(vnet_listen_args_t *a)
Definition: application.c:965
static u8 * format_session_error(u8 *s, va_list *args)
int vnet_application_detach(vnet_app_detach_args_t *a)
Detach application from vpp.
Definition: application.c:887
u32 clib_llist_index_t
Definition: llist.h:27
#define TRANSPORT_PACER_MIN_BURST
Definition: transport.h:23
enum _transport_proto transport_proto_t
#define clib_llist_elt_is_linked(E, name)
Check if element is linked in a list.
Definition: llist.h:129
clib_time_type_t last_vlib_time
vlib_time_now last time around the track
Definition: session.h:83
ip46_address_t rmt_ip
struct _vnet_app_detach_args_t vnet_app_detach_args_t
static uword pointer_to_uword(const void *p)
Definition: types.h:131
int vnet_connect(vnet_connect_args_t *a)
Definition: application.c:1019
static u8 * format_session_queue_trace(u8 *s, va_list *args)
Definition: session_node.c:505
static int transport_app_rx_evt(transport_proto_t tp, u32 conn_index, u32 thread_index)
Definition: transport.h:175
session_dgram_hdr_t hdr
Definition: session.h:58
u8 thread_index
Index of the thread that allocated the session.
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
static void session_tx_set_dequeue_params(vlib_main_t *vm, session_tx_context_t *ctx, u32 max_segs, u8 peek_data)
Definition: session_node.c:770
u32 app_index
App index in app pool.
Definition: application.h:92
struct _vlib_node_registration vlib_node_registration_t
template key/value backing page structure
Definition: bihash_doc.h:44
app_worker_t * app_worker_get(u32 wrk_index)
void svm_msg_q_free_msg(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Free message buffer.
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
void vlib_rpc_call_main_thread(void *callback, u8 *args, u32 arg_size)
Definition: threads.c:1919
static u8 session_tx_not_ready(session_t *s, u8 peek_data)
Definition: session_node.c:726
clib_llist_index_t old_head
Head of list of pending events.
Definition: session.h:113
u32 next_buffer
Next buffer for this linked-list of buffers.
Definition: buffer.h:140
volatile u8 session_state
State in session layer state machine.
#define VLIB_BUFFER_TRACE_TRAJECTORY_INIT(b)
Definition: buffer.h:492
VLIB buffer representation.
Definition: buffer.h:102
u64 uword
Definition: types.h:112
int vnet_bind_uri(vnet_listen_args_t *a)
static const u32 session_evt_msg_sizes[]
static void session_mq_worker_update_handler(void *data)
Definition: session_node.c:418
static void session_evt_add_old(session_worker_t *wrk, session_evt_elt_t *elt)
Definition: session.h:227
int vnet_disconnect_session(vnet_disconnect_args_t *a)
Definition: application.c:1087
static void session_mq_accepted_reply_handler(void *data)
Definition: session_node.c:240
static void session_evt_add_head_old(session_worker_t *wrk, session_evt_elt_t *elt)
Definition: session.h:234
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
Definition: session_node.c:495
static session_evt_elt_t * session_evt_alloc_ctrl(session_worker_t *wrk)
Definition: session.h:250
struct clib_bihash_value offset
template key/value backing page structure
static void session_mq_disconnected_handler(void *data)
Definition: session_node.c:347
static u8 session_has_transport(session_t *s)
#define clib_llist_prev_index(E, name)
Definition: llist.h:61
static transport_connection_t * session_tx_get_transport(session_tx_context_t *ctx, u8 peek_data)
Definition: session_node.c:750
u32 app_index
Index of owning app.
Definition: application.h:43
transport_connection_t * tc
Definition: session.h:47
static void session_flush_pending_tx_buffers(session_worker_t *wrk, vlib_node_runtime_t *node)
int app_worker_own_session(app_worker_t *app_wrk, session_t *s)
void vlib_worker_thread_barrier_release(vlib_main_t *vm)
Definition: threads.c:1540
clib_us_time_t last_vlib_us_time
vlib_time_now rounded to us precision and as u64
Definition: session.h:86
static u32 vlib_num_workers()
Definition: threads.h:376
static u32 svm_msg_q_size(svm_msg_q_t *mq)
Check length of message queue.
u32 app_wrk_index
Index of the app worker that owns the session.
vlib_main_t vlib_node_runtime_t vlib_frame_t * frame
Definition: in2out_ed.c:1600
static void session_mq_listen_handler(void *data)
Definition: session_node.c:36
static int session_tx_fifo_read_and_snd_i(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, int *n_tx_packets, u8 peek_data)
Definition: session_node.c:902
#define clib_llist_remove(LP, name, E)
Remove entry from list.
Definition: llist.h:193
static session_main_t * vnet_get_session_main()
Definition: session.h:613
void session_transport_close(session_t *s)
Notify transport the session can be disconnected.
Definition: session.c:1401
clib_llist_index_t ctrl_head
Head of control events list.
Definition: session.h:107
int app_worker_close_notify(app_worker_t *app_wrk, session_t *s)
static void vlib_set_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt, u32 count)
Definition: trace_funcs.h:203
#define CLIB_CACHE_LINE_BYTES
Definition: cache.h:59
u32 total_length_not_including_first_buffer
Only valid for first buffer in chain.
Definition: buffer.h:167
#define clib_llist_foreach_safe(LP, name, H, E, body)
Walk list starting at head safe.
Definition: llist.h:275
int vnet_connect_uri(vnet_connect_args_t *a)
int svm_fifo_dequeue_drop(svm_fifo_t *f, u32 len)
Dequeue and drop bytes from fifo.
Definition: svm_fifo.c:1061
static u32 vlib_buffer_alloc(vlib_main_t *vm, u32 *buffers, u32 n_buffers)
Allocate buffers into supplied array.
Definition: buffer_funcs.h:677
vlib_node_registration_t session_queue_process_node
(constructor) VLIB_REGISTER_NODE (session_queue_process_node)
static uword session_queue_run_on_main(vlib_main_t *vm)
svm_msg_q_t * event_queue
Application listens for events on this svm queue.
Definition: application.h:46
struct _svm_fifo svm_fifo_t
static void transport_connection_deschedule(transport_connection_t *tc)
Definition: transport.h:208
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
Definition: buffer_funcs.h:85
struct session_dgram_header_ session_dgram_hdr_t
#define SESSION_EVT(_evt, _args...)
static void session_mq_reset_reply_handler(void *data)
Definition: session_node.c:303
u32 transport_connection_tx_pacer_burst(transport_connection_t *tc)
Get tx pacer max burst.
Definition: transport.c:729
static void session_mq_connect_handler(void *data)
Definition: session_node.c:98
void mq_send_unlisten_reply(app_worker_t *app_wrk, session_handle_t sh, u32 context, int rv)
Definition: session_api.c:400
#define app_check_thread_and_barrier(_fn, _arg)
Definition: session_node.c:28
static void session_tx_maybe_reschedule(session_worker_t *wrk, session_tx_context_t *ctx, session_evt_elt_t *elt)
Definition: session_node.c:889
#define clib_llist_entry_index(LP, E)
Get list entry index.
Definition: llist.h:53
void svm_msg_q_sub_w_lock(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Consumer dequeue one message from queue with mutex held.