blob: cf2d5766fb0a1e66895e3fde29619702ff7ce6a8 [file] [log] [blame]
Dave Barach68b0fb02017-02-28 15:15:56 -05001/*
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
Florin Coras6792ec02017-03-13 03:49:51 -070016#include <math.h>
Dave Barach68b0fb02017-02-28 15:15:56 -050017#include <vlib/vlib.h>
18#include <vnet/vnet.h>
Dave Barach68b0fb02017-02-28 15:15:56 -050019#include <vppinfra/elog.h>
Florin Coras561af9b2017-12-09 10:19:43 -080020#include <vnet/session/transport.h>
Florin Coras8e43d042018-05-04 15:46:57 -070021#include <vnet/session/session.h>
Florin Coras6792ec02017-03-13 03:49:51 -070022#include <vnet/session/application.h>
Florin Coras52207f12018-07-12 14:48:06 -070023#include <vnet/session/application_interface.h>
Florin Corase69f4952017-03-07 10:06:24 -080024#include <vnet/session/session_debug.h>
Florin Corase86a8ed2018-01-05 03:20:25 -080025#include <svm/queue.h>
Dave Barach68b0fb02017-02-28 15:15:56 -050026
Florin Coras52207f12018-07-12 14:48:06 -070027static void
28session_mq_accepted_reply_handler (void *data)
29{
30 session_accepted_reply_msg_t *mp = (session_accepted_reply_msg_t *) data;
31 vnet_disconnect_args_t _a = { 0 }, *a = &_a;
Florin Coras21795132018-09-09 09:40:51 -070032 app_worker_t *app_wrk;
Florin Coras52207f12018-07-12 14:48:06 -070033 local_session_t *ls;
34 stream_session_t *s;
35
36 /* Server isn't interested, kill the session */
37 if (mp->retval)
38 {
39 a->app_index = mp->context;
40 a->handle = mp->handle;
41 vnet_disconnect_session (a);
42 return;
43 }
44
45 if (session_handle_is_local (mp->handle))
46 {
47 ls = application_get_local_session_from_handle (mp->handle);
Florin Corasab2f6db2018-08-31 14:31:41 -070048 if (!ls)
Florin Coras52207f12018-07-12 14:48:06 -070049 {
Florin Corasab2f6db2018-08-31 14:31:41 -070050 clib_warning ("unknown local handle 0x%lx", mp->handle);
51 return;
52 }
53 app_wrk = app_worker_get (ls->app_wrk_index);
Florin Coras21795132018-09-09 09:40:51 -070054 if (app_wrk->app_index != mp->context)
Florin Corasab2f6db2018-08-31 14:31:41 -070055 {
56 clib_warning ("server %u doesn't own local handle 0x%lx",
Florin Coras52207f12018-07-12 14:48:06 -070057 mp->context, mp->handle);
58 return;
59 }
60 if (application_local_session_connect_notify (ls))
61 return;
62 ls->session_state = SESSION_STATE_READY;
63 }
64 else
65 {
66 s = session_get_from_handle_if_valid (mp->handle);
67 if (!s)
68 {
69 clib_warning ("session doesn't exist");
70 return;
71 }
Florin Coras21795132018-09-09 09:40:51 -070072 app_wrk = app_worker_get (s->app_wrk_index);
73 if (app_wrk->app_index != mp->context)
Florin Coras52207f12018-07-12 14:48:06 -070074 {
75 clib_warning ("app doesn't own session");
76 return;
77 }
78 s->session_state = SESSION_STATE_READY;
Florin Coras84099552018-07-22 12:59:30 -070079 if (!svm_fifo_is_empty (s->server_rx_fifo))
Florin Coras21795132018-09-09 09:40:51 -070080 app_worker_lock_and_send_event (app_wrk, s, FIFO_EVENT_APP_RX);
Florin Coras52207f12018-07-12 14:48:06 -070081 }
82}
83
84static void
85session_mq_reset_reply_handler (void *data)
86{
87 session_reset_reply_msg_t *mp;
Florin Coras15531972018-08-12 23:50:53 -070088 app_worker_t *app_wrk;
Florin Coras52207f12018-07-12 14:48:06 -070089 stream_session_t *s;
Florin Coras15531972018-08-12 23:50:53 -070090 application_t *app;
Florin Coras52207f12018-07-12 14:48:06 -070091 u32 index, thread_index;
92
93 mp = (session_reset_reply_msg_t *) data;
94 app = application_lookup (mp->client_index);
95 if (!app)
96 return;
97
98 session_parse_handle (mp->handle, &index, &thread_index);
99 s = session_get_if_valid (index, thread_index);
Florin Coras15531972018-08-12 23:50:53 -0700100 if (!s)
Florin Coras52207f12018-07-12 14:48:06 -0700101 {
Florin Corasefefc6b2018-11-07 12:49:19 -0800102 SESSION_DBG ("Invalid session!");
Florin Coras52207f12018-07-12 14:48:06 -0700103 return;
104 }
Florin Coras15531972018-08-12 23:50:53 -0700105 app_wrk = app_worker_get (s->app_wrk_index);
106 if (!app_wrk || app_wrk->app_index != app->app_index)
107 {
108 clib_warning ("App % does not own handle 0x%lx!", app->app_index,
109 mp->handle);
110 return;
111 }
Florin Coras52207f12018-07-12 14:48:06 -0700112
113 /* Client objected to resetting the session, log and continue */
114 if (mp->retval)
115 {
116 clib_warning ("client retval %d", mp->retval);
117 return;
118 }
119
120 /* This comes as a response to a reset, transport only waiting for
121 * confirmation to remove connection state, no need to disconnect */
122 stream_session_cleanup (s);
123}
124
125static void
126session_mq_disconnected_handler (void *data)
127{
128 session_disconnected_reply_msg_t *rmp;
129 vnet_disconnect_args_t _a, *a = &_a;
130 svm_msg_q_msg_t _msg, *msg = &_msg;
131 session_disconnected_msg_t *mp;
Florin Coras15531972018-08-12 23:50:53 -0700132 app_worker_t *app_wrk;
Florin Coras52207f12018-07-12 14:48:06 -0700133 session_event_t *evt;
Florin Coras3d0fadc2018-07-17 05:35:47 -0700134 stream_session_t *s;
Florin Coras52207f12018-07-12 14:48:06 -0700135 application_t *app;
136 int rv = 0;
137
138 mp = (session_disconnected_msg_t *) data;
Florin Corasc3638fe2018-08-24 13:58:49 -0700139 if (!(s = session_get_from_handle_if_valid (mp->handle)))
140 {
141 clib_warning ("could not disconnect handle %llu", mp->handle);
142 return;
143 }
Florin Coras15531972018-08-12 23:50:53 -0700144 app_wrk = app_worker_get (s->app_wrk_index);
145 app = application_lookup (mp->client_index);
Florin Corasc3638fe2018-08-24 13:58:49 -0700146 if (!(app_wrk && app && app->app_index == app_wrk->app_index))
Florin Coras52207f12018-07-12 14:48:06 -0700147 {
Florin Corasc3638fe2018-08-24 13:58:49 -0700148 clib_warning ("could not disconnect session: %llu app: %u",
Florin Coras15531972018-08-12 23:50:53 -0700149 mp->handle, mp->client_index);
Florin Coras3d0fadc2018-07-17 05:35:47 -0700150 return;
Florin Coras52207f12018-07-12 14:48:06 -0700151 }
Florin Coras3d0fadc2018-07-17 05:35:47 -0700152
153 a->handle = mp->handle;
Florin Coras15531972018-08-12 23:50:53 -0700154 a->app_index = app_wrk->wrk_index;
Florin Coras3d0fadc2018-07-17 05:35:47 -0700155 rv = vnet_disconnect_session (a);
Florin Coras52207f12018-07-12 14:48:06 -0700156
Florin Coras15531972018-08-12 23:50:53 -0700157 svm_msg_q_lock_and_alloc_msg_w_ring (app_wrk->event_queue,
Florin Coras52207f12018-07-12 14:48:06 -0700158 SESSION_MQ_CTRL_EVT_RING,
159 SVM_Q_WAIT, msg);
Florin Coras15531972018-08-12 23:50:53 -0700160 svm_msg_q_unlock (app_wrk->event_queue);
161 evt = svm_msg_q_msg_data (app_wrk->event_queue, msg);
Dave Barachb7b92992018-10-17 10:38:51 -0400162 clib_memset (evt, 0, sizeof (*evt));
Florin Coras52207f12018-07-12 14:48:06 -0700163 evt->event_type = SESSION_CTRL_EVT_DISCONNECTED;
164 rmp = (session_disconnected_reply_msg_t *) evt->data;
165 rmp->handle = mp->handle;
166 rmp->context = mp->context;
167 rmp->retval = rv;
Florin Coras15531972018-08-12 23:50:53 -0700168 svm_msg_q_add (app_wrk->event_queue, msg, SVM_Q_WAIT);
Florin Coras52207f12018-07-12 14:48:06 -0700169}
170
171static void
172session_mq_disconnected_reply_handler (void *data)
173{
174 session_disconnected_reply_msg_t *mp;
175 vnet_disconnect_args_t _a, *a = &_a;
176 application_t *app;
177
178 mp = (session_disconnected_reply_msg_t *) data;
179
180 /* Client objected to disconnecting the session, log and continue */
181 if (mp->retval)
182 {
183 clib_warning ("client retval %d", mp->retval);
184 return;
185 }
186
187 /* Disconnect has been confirmed. Confirm close to transport */
188 app = application_lookup (mp->context);
189 if (app)
190 {
191 a->handle = mp->handle;
Florin Coras15531972018-08-12 23:50:53 -0700192 a->app_index = app->app_index;
Florin Coras52207f12018-07-12 14:48:06 -0700193 vnet_disconnect_session (a);
194 }
195}
196
Dave Barach68b0fb02017-02-28 15:15:56 -0500197vlib_node_registration_t session_queue_node;
198
199typedef struct
200{
201 u32 session_index;
202 u32 server_thread_index;
203} session_queue_trace_t;
204
205/* packet trace format function */
206static u8 *
207format_session_queue_trace (u8 * s, va_list * args)
208{
209 CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
210 CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
211 session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
212
213 s = format (s, "SESSION_QUEUE: session index %d, server thread index %d",
214 t->session_index, t->server_thread_index);
215 return s;
216}
217
Florin Coras6792ec02017-03-13 03:49:51 -0700218#define foreach_session_queue_error \
219_(TX, "Packets transmitted") \
Florin Coras93992a92017-05-24 18:03:56 -0700220_(TIMER, "Timer events") \
221_(NO_BUFFER, "Out of buffers")
Dave Barach68b0fb02017-02-28 15:15:56 -0500222
223typedef enum
224{
225#define _(sym,str) SESSION_QUEUE_ERROR_##sym,
226 foreach_session_queue_error
227#undef _
228 SESSION_QUEUE_N_ERROR,
229} session_queue_error_t;
230
231static char *session_queue_error_strings[] = {
232#define _(sym,string) string,
233 foreach_session_queue_error
234#undef _
235};
236
Florin Coras0d60a0f2018-06-29 02:02:08 -0700237enum
238{
239 SESSION_TX_NO_BUFFERS = -2,
240 SESSION_TX_NO_DATA,
241 SESSION_TX_OK
242};
243
Florin Coras66cf5e02018-06-08 09:17:39 -0700244static void
245session_tx_trace_frame (vlib_main_t * vm, vlib_node_runtime_t * node,
246 u32 next_index, u32 * to_next, u16 n_segs,
247 stream_session_t * s, u32 n_trace)
Florin Corasf6d68ed2017-05-07 19:12:02 -0700248{
Florin Coras8e43d042018-05-04 15:46:57 -0700249 session_queue_trace_t *t;
Florin Coras66cf5e02018-06-08 09:17:39 -0700250 vlib_buffer_t *b;
251 int i;
252
253 for (i = 0; i < clib_min (n_trace, n_segs); i++)
254 {
255 b = vlib_get_buffer (vm, to_next[i - n_segs]);
256 vlib_trace_buffer (vm, node, next_index, b, 1 /* follow_chain */ );
Florin Coras66cf5e02018-06-08 09:17:39 -0700257 t = vlib_add_trace (vm, node, b, sizeof (*t));
258 t->session_index = s->session_index;
259 t->server_thread_index = s->thread_index;
260 }
Florin Coras4df38712018-06-20 12:44:16 -0700261 vlib_set_trace_count (vm, node, n_trace - i);
Florin Coras8e43d042018-05-04 15:46:57 -0700262}
Florin Corasf6d68ed2017-05-07 19:12:02 -0700263
Florin Coras8e43d042018-05-04 15:46:57 -0700264always_inline void
265session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
266 vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
267{
Florin Coras8e43d042018-05-04 15:46:57 -0700268 vlib_buffer_t *chain_b, *prev_b;
269 u32 chain_bi0, to_deq, left_from_seg;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700270 session_manager_worker_t *wrk;
Florin Coras8e43d042018-05-04 15:46:57 -0700271 u16 len_to_deq, n_bytes_read;
272 u8 *data, j;
Florin Corasb2215d62017-08-01 16:56:58 -0700273
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700274 wrk = session_manager_get_worker (ctx->s->thread_index);
Florin Coras8e43d042018-05-04 15:46:57 -0700275 b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
276 b->total_length_not_including_first_buffer = 0;
277
278 chain_b = b;
279 left_from_seg = clib_min (ctx->snd_mss - b->current_length,
280 ctx->left_to_snd);
Florin Corasb2215d62017-08-01 16:56:58 -0700281 to_deq = left_from_seg;
Florin Coras8e43d042018-05-04 15:46:57 -0700282 for (j = 1; j < ctx->n_bufs_per_seg; j++)
Florin Corasf6d68ed2017-05-07 19:12:02 -0700283 {
Florin Coras8e43d042018-05-04 15:46:57 -0700284 prev_b = chain_b;
285 len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700286
287 *n_bufs -= 1;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700288 chain_bi0 = wrk->tx_buffers[*n_bufs];
289 _vec_len (wrk->tx_buffers) = *n_bufs;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700290
Florin Coras8e43d042018-05-04 15:46:57 -0700291 chain_b = vlib_get_buffer (vm, chain_bi0);
292 chain_b->current_data = 0;
293 data = vlib_buffer_get_current (chain_b);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700294 if (peek_data)
295 {
Florin Coras8e43d042018-05-04 15:46:57 -0700296 n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo,
297 ctx->tx_offset, len_to_deq, data);
298 ctx->tx_offset += n_bytes_read;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700299 }
300 else
301 {
Florin Coras8e43d042018-05-04 15:46:57 -0700302 if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
Florin Coras7fb0fe12018-04-09 09:24:52 -0700303 {
Florin Coras8e43d042018-05-04 15:46:57 -0700304 svm_fifo_t *f = ctx->s->server_tx_fifo;
305 session_dgram_hdr_t *hdr = &ctx->hdr;
Florin Coras7fb0fe12018-04-09 09:24:52 -0700306 u16 deq_now;
Florin Coras7fb0fe12018-04-09 09:24:52 -0700307 deq_now = clib_min (hdr->data_length - hdr->data_offset,
Florin Coras8e43d042018-05-04 15:46:57 -0700308 len_to_deq);
309 n_bytes_read = svm_fifo_peek (f, hdr->data_offset, deq_now,
310 data);
Florin Coras7fb0fe12018-04-09 09:24:52 -0700311 ASSERT (n_bytes_read > 0);
312
313 hdr->data_offset += n_bytes_read;
314 if (hdr->data_offset == hdr->data_length)
shubing guoaea5f392018-08-30 13:29:49 +0800315 {
316 u32 offset = hdr->data_length + SESSION_CONN_HDR_LEN;
317 svm_fifo_dequeue_drop (f, offset);
318 }
Florin Coras7fb0fe12018-04-09 09:24:52 -0700319 }
320 else
Florin Coras8e43d042018-05-04 15:46:57 -0700321 n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
322 len_to_deq, data);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700323 }
Florin Coras8e43d042018-05-04 15:46:57 -0700324 ASSERT (n_bytes_read == len_to_deq);
325 chain_b->current_length = n_bytes_read;
326 b->total_length_not_including_first_buffer += chain_b->current_length;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700327
328 /* update previous buffer */
Florin Coras8e43d042018-05-04 15:46:57 -0700329 prev_b->next_buffer = chain_bi0;
330 prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700331
332 /* update current buffer */
Florin Coras8e43d042018-05-04 15:46:57 -0700333 chain_b->next_buffer = 0;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700334
Florin Corasb2215d62017-08-01 16:56:58 -0700335 to_deq -= n_bytes_read;
336 if (to_deq == 0)
Florin Corasf6d68ed2017-05-07 19:12:02 -0700337 break;
338 }
Florin Coras1f152cd2017-08-18 19:28:03 -0700339 ASSERT (to_deq == 0
Florin Coras8e43d042018-05-04 15:46:57 -0700340 && b->total_length_not_including_first_buffer == left_from_seg);
341 ctx->left_to_snd -= left_from_seg;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700342}
343
Dave Barach68b0fb02017-02-28 15:15:56 -0500344always_inline int
Florin Coras8e43d042018-05-04 15:46:57 -0700345session_output_try_get_buffers (vlib_main_t * vm,
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700346 session_manager_worker_t * wrk,
Florin Coras8e43d042018-05-04 15:46:57 -0700347 u32 thread_index, u16 * n_bufs, u32 wanted)
Dave Barach68b0fb02017-02-28 15:15:56 -0500348{
Florin Corasf08f26d2018-05-10 13:20:47 -0700349 u32 n_alloc;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700350 vec_validate_aligned (wrk->tx_buffers, wanted - 1, CLIB_CACHE_LINE_BYTES);
351 n_alloc = vlib_buffer_alloc (vm, &wrk->tx_buffers[*n_bufs],
Florin Corasf08f26d2018-05-10 13:20:47 -0700352 wanted - *n_bufs);
353 *n_bufs += n_alloc;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700354 _vec_len (wrk->tx_buffers) = *n_bufs;
Florin Corasf08f26d2018-05-10 13:20:47 -0700355 return n_alloc;
Florin Coras8e43d042018-05-04 15:46:57 -0700356}
357
358always_inline void
359session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
360 vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
361{
362 u32 len_to_deq;
363 u8 *data0;
364 int n_bytes_read;
365
366 /*
367 * Start with the first buffer in chain
368 */
369 b->error = 0;
370 b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
371 b->current_data = 0;
Florin Coras8e43d042018-05-04 15:46:57 -0700372
373 data0 = vlib_buffer_make_headroom (b, MAX_HDRS_LEN);
374 len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
375
Florin Coras7fb0fe12018-04-09 09:24:52 -0700376 if (peek_data)
377 {
Florin Coras8e43d042018-05-04 15:46:57 -0700378 n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo, ctx->tx_offset,
379 len_to_deq, data0);
380 ASSERT (n_bytes_read > 0);
381 /* Keep track of progress locally, transport is also supposed to
382 * increment it independently when pushing the header */
383 ctx->tx_offset += n_bytes_read;
Florin Coras7fb0fe12018-04-09 09:24:52 -0700384 }
385 else
386 {
Florin Coras8e43d042018-05-04 15:46:57 -0700387 if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
388 {
389 session_dgram_hdr_t *hdr = &ctx->hdr;
390 svm_fifo_t *f = ctx->s->server_tx_fifo;
391 u16 deq_now;
392 u32 offset;
393
394 ASSERT (hdr->data_length > hdr->data_offset);
395 deq_now = clib_min (hdr->data_length - hdr->data_offset,
396 len_to_deq);
397 offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
398 n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
399 ASSERT (n_bytes_read > 0);
400
401 if (ctx->s->session_state == SESSION_STATE_LISTENING)
402 {
403 ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
404 ctx->tc->rmt_port = hdr->rmt_port;
405 }
406 hdr->data_offset += n_bytes_read;
407 if (hdr->data_offset == hdr->data_length)
408 {
409 offset = hdr->data_length + SESSION_CONN_HDR_LEN;
410 svm_fifo_dequeue_drop (f, offset);
411 }
412 }
Florin Coras7fb0fe12018-04-09 09:24:52 -0700413 else
414 {
Florin Coras8e43d042018-05-04 15:46:57 -0700415 n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
416 len_to_deq, data0);
417 ASSERT (n_bytes_read > 0);
Florin Coras7fb0fe12018-04-09 09:24:52 -0700418 }
419 }
Florin Coras8e43d042018-05-04 15:46:57 -0700420 b->current_length = n_bytes_read;
421 ctx->left_to_snd -= n_bytes_read;
Dave Barach68b0fb02017-02-28 15:15:56 -0500422
Florin Coras8e43d042018-05-04 15:46:57 -0700423 /*
424 * Fill in the remaining buffers in the chain, if any
425 */
426 if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
427 session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
Dave Barach68b0fb02017-02-28 15:15:56 -0500428
Florin Coras8e43d042018-05-04 15:46:57 -0700429 /* *INDENT-OFF* */
Florin Coras8b20bf52018-06-14 14:55:50 -0700430 SESSION_EVT_DBG(SESSION_EVT_DEQ, ctx->s, ({
431 ed->data[0] = FIFO_EVENT_APP_TX;
432 ed->data[1] = ctx->max_dequeue;
Florin Coras8e43d042018-05-04 15:46:57 -0700433 ed->data[2] = len_to_deq;
Florin Coras8b20bf52018-06-14 14:55:50 -0700434 ed->data[3] = ctx->left_to_snd;
Florin Coras8e43d042018-05-04 15:46:57 -0700435 }));
436 /* *INDENT-ON* */
437}
438
439always_inline u8
440session_tx_not_ready (stream_session_t * s, u8 peek_data)
441{
442 if (peek_data)
Florin Corase04c2992017-03-01 08:17:34 -0800443 {
Florin Coras8e43d042018-05-04 15:46:57 -0700444 /* Can retransmit for closed sessions but can't send new data if
445 * session is not ready or closed */
446 if (s->session_state < SESSION_STATE_READY)
447 return 1;
Florin Corasca1c8f32018-05-23 21:01:30 -0700448 if (s->session_state == SESSION_STATE_CLOSED)
449 return 2;
Florin Corase04c2992017-03-01 08:17:34 -0800450 }
Florin Coras8e43d042018-05-04 15:46:57 -0700451 return 0;
452}
Dave Barach68b0fb02017-02-28 15:15:56 -0500453
Florin Coras8e43d042018-05-04 15:46:57 -0700454always_inline transport_connection_t *
455session_tx_get_transport (session_tx_context_t * ctx, u8 peek_data)
456{
457 if (peek_data)
458 {
459 return ctx->transport_vft->get_connection (ctx->s->connection_index,
460 ctx->s->thread_index);
461 }
462 else
463 {
464 if (ctx->s->session_state == SESSION_STATE_LISTENING)
465 return ctx->transport_vft->get_listener (ctx->s->connection_index);
466 else
467 {
468 return ctx->transport_vft->get_connection (ctx->s->connection_index,
469 ctx->s->thread_index);
470 }
471 }
472}
Florin Coras6a9b68b2017-11-21 04:20:42 -0800473
Florin Coras8e43d042018-05-04 15:46:57 -0700474always_inline void
475session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
Florin Corasf08f26d2018-05-10 13:20:47 -0700476 u32 max_segs, u8 peek_data)
Florin Coras8e43d042018-05-04 15:46:57 -0700477{
478 u32 n_bytes_per_buf, n_bytes_per_seg;
479 ctx->max_dequeue = svm_fifo_max_dequeue (ctx->s->server_tx_fifo);
Dave Barach68b0fb02017-02-28 15:15:56 -0500480 if (peek_data)
481 {
Florin Coras9d063042017-09-14 03:08:00 -0400482 /* Offset in rx fifo from where to peek data */
Florin Coras8e43d042018-05-04 15:46:57 -0700483 ctx->tx_offset = ctx->transport_vft->tx_fifo_offset (ctx->tc);
484 if (PREDICT_FALSE (ctx->tx_offset >= ctx->max_dequeue))
485 {
486 ctx->max_len_to_snd = 0;
487 return;
488 }
489 ctx->max_dequeue -= ctx->tx_offset;
Dave Barach68b0fb02017-02-28 15:15:56 -0500490 }
Florin Coras7fb0fe12018-04-09 09:24:52 -0700491 else
492 {
Florin Coras8e43d042018-05-04 15:46:57 -0700493 if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
Florin Coras7fb0fe12018-04-09 09:24:52 -0700494 {
Florin Coras8e43d042018-05-04 15:46:57 -0700495 if (ctx->max_dequeue <= sizeof (ctx->hdr))
496 {
497 ctx->max_len_to_snd = 0;
498 return;
499 }
500 svm_fifo_peek (ctx->s->server_tx_fifo, 0, sizeof (ctx->hdr),
501 (u8 *) & ctx->hdr);
502 ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
503 ctx->max_dequeue = ctx->hdr.data_length - ctx->hdr.data_offset;
Florin Coras7fb0fe12018-04-09 09:24:52 -0700504 }
505 }
Florin Coras8e43d042018-05-04 15:46:57 -0700506 ASSERT (ctx->max_dequeue > 0);
Florin Coras6792ec02017-03-13 03:49:51 -0700507
508 /* Ensure we're not writing more than transport window allows */
Florin Coras8e43d042018-05-04 15:46:57 -0700509 if (ctx->max_dequeue < ctx->snd_space)
Florin Coras3e350af2017-03-30 02:54:28 -0700510 {
511 /* Constrained by tx queue. Try to send only fully formed segments */
Florin Coras8e43d042018-05-04 15:46:57 -0700512 ctx->max_len_to_snd =
513 (ctx->max_dequeue > ctx->snd_mss) ?
514 ctx->max_dequeue - ctx->max_dequeue % ctx->snd_mss : ctx->max_dequeue;
Florin Coras3e350af2017-03-30 02:54:28 -0700515 /* TODO Nagle ? */
516 }
517 else
518 {
Florin Coras1f152cd2017-08-18 19:28:03 -0700519 /* Expectation is that snd_space0 is already a multiple of snd_mss */
Florin Coras8e43d042018-05-04 15:46:57 -0700520 ctx->max_len_to_snd = ctx->snd_space;
Florin Coras3e350af2017-03-30 02:54:28 -0700521 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500522
Florin Corasf08f26d2018-05-10 13:20:47 -0700523 /* Check if we're tx constrained by the node */
524 ctx->n_segs_per_evt = ceil ((f64) ctx->max_len_to_snd / ctx->snd_mss);
525 if (ctx->n_segs_per_evt > max_segs)
526 {
527 ctx->n_segs_per_evt = max_segs;
528 ctx->max_len_to_snd = max_segs * ctx->snd_mss;
529 }
530
Florin Coras8e43d042018-05-04 15:46:57 -0700531 n_bytes_per_buf = vlib_buffer_free_list_buffer_size (vm,
532 VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
Florin Corase87216f2017-08-17 16:59:22 -0700533 ASSERT (n_bytes_per_buf > MAX_HDRS_LEN);
Florin Coras8e43d042018-05-04 15:46:57 -0700534 n_bytes_per_seg = MAX_HDRS_LEN + ctx->snd_mss;
Florin Corasf08f26d2018-05-10 13:20:47 -0700535 ctx->n_bufs_per_seg = ceil ((f64) n_bytes_per_seg / n_bytes_per_buf);
Florin Coras8e43d042018-05-04 15:46:57 -0700536 ctx->deq_per_buf = clib_min (ctx->snd_mss, n_bytes_per_buf);
537 ctx->deq_per_first_buf = clib_min (ctx->snd_mss,
538 n_bytes_per_buf - MAX_HDRS_LEN);
539}
Florin Corasf6d68ed2017-05-07 19:12:02 -0700540
Florin Coras8e43d042018-05-04 15:46:57 -0700541always_inline int
542session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
Florin Coras52207f12018-07-12 14:48:06 -0700543 session_event_t * e,
Florin Coras8e43d042018-05-04 15:46:57 -0700544 stream_session_t * s, int *n_tx_packets,
545 u8 peek_data)
546{
Florin Coras66cf5e02018-06-08 09:17:39 -0700547 u32 next_index, next0, next1, *to_next, n_left_to_next;
Florin Corasf08f26d2018-05-10 13:20:47 -0700548 u32 n_trace = vlib_get_trace_count (vm, node), n_bufs_needed = 0;
549 u32 thread_index = s->thread_index, n_left, pbi;
Florin Coras8e43d042018-05-04 15:46:57 -0700550 session_manager_main_t *smm = &session_manager_main;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700551 session_manager_worker_t *wrk = &smm->wrk[thread_index];
552 session_tx_context_t *ctx = &wrk->ctx;
Florin Coras8e43d042018-05-04 15:46:57 -0700553 transport_proto_t tp;
554 vlib_buffer_t *pb;
Florin Corasca1c8f32018-05-23 21:01:30 -0700555 u16 n_bufs, rv;
Florin Coras8e43d042018-05-04 15:46:57 -0700556
Florin Corasca1c8f32018-05-23 21:01:30 -0700557 if (PREDICT_FALSE ((rv = session_tx_not_ready (s, peek_data))))
Florin Coras8e43d042018-05-04 15:46:57 -0700558 {
Florin Corasca1c8f32018-05-23 21:01:30 -0700559 if (rv < 2)
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700560 vec_add1 (wrk->pending_event_vector, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700561 return SESSION_TX_NO_DATA;
Florin Coras8e43d042018-05-04 15:46:57 -0700562 }
563
564 next_index = smm->session_type_to_next[s->session_type];
Florin Coras66cf5e02018-06-08 09:17:39 -0700565 next0 = next1 = next_index;
Florin Coras8e43d042018-05-04 15:46:57 -0700566
567 tp = session_get_transport_proto (s);
568 ctx->s = s;
569 ctx->transport_vft = transport_protocol_get_vft (tp);
570 ctx->tc = session_tx_get_transport (ctx, peek_data);
571 ctx->snd_mss = ctx->transport_vft->send_mss (ctx->tc);
Florin Corasd67f1122018-05-21 17:47:40 -0700572 ctx->snd_space =
Florin Corase55a6d72018-10-31 23:09:22 -0700573 transport_connection_snd_space (ctx->tc, vm->clib_time.last_cpu_time,
574 ctx->snd_mss);
Florin Coras8e43d042018-05-04 15:46:57 -0700575 if (ctx->snd_space == 0 || ctx->snd_mss == 0)
576 {
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700577 vec_add1 (wrk->pending_event_vector, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700578 return SESSION_TX_NO_DATA;
Florin Coras8e43d042018-05-04 15:46:57 -0700579 }
580
581 /* Allow enqueuing of a new event */
582 svm_fifo_unset_event (s->server_tx_fifo);
583
584 /* Check how much we can pull. */
Florin Corasf08f26d2018-05-10 13:20:47 -0700585 session_tx_set_dequeue_params (vm, ctx, VLIB_FRAME_SIZE - *n_tx_packets,
586 peek_data);
587
Florin Coras8e43d042018-05-04 15:46:57 -0700588 if (PREDICT_FALSE (!ctx->max_len_to_snd))
Florin Coras0d60a0f2018-06-29 02:02:08 -0700589 return SESSION_TX_NO_DATA;
Dave Barach68b0fb02017-02-28 15:15:56 -0500590
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700591 n_bufs = vec_len (wrk->tx_buffers);
Florin Corasf08f26d2018-05-10 13:20:47 -0700592 n_bufs_needed = ctx->n_segs_per_evt * ctx->n_bufs_per_seg;
Florin Coras8e43d042018-05-04 15:46:57 -0700593
594 /*
595 * Make sure we have at least one full frame of buffers ready
596 */
Florin Corasf08f26d2018-05-10 13:20:47 -0700597 if (n_bufs < n_bufs_needed)
Dave Barach68b0fb02017-02-28 15:15:56 -0500598 {
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700599 session_output_try_get_buffers (vm, wrk, thread_index, &n_bufs,
Florin Corasf08f26d2018-05-10 13:20:47 -0700600 ctx->n_bufs_per_seg * VLIB_FRAME_SIZE);
601 if (PREDICT_FALSE (n_bufs < n_bufs_needed))
Dave Barach68b0fb02017-02-28 15:15:56 -0500602 {
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700603 vec_add1 (wrk->pending_event_vector, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700604 return SESSION_TX_NO_BUFFERS;
Dave Barach68b0fb02017-02-28 15:15:56 -0500605 }
Florin Coras8e43d042018-05-04 15:46:57 -0700606 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500607
Florin Coras8e43d042018-05-04 15:46:57 -0700608 /*
609 * Write until we fill up a frame
610 */
611 vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
Florin Corasf08f26d2018-05-10 13:20:47 -0700612 if (PREDICT_FALSE (ctx->n_segs_per_evt > n_left_to_next))
Florin Coras8e43d042018-05-04 15:46:57 -0700613 {
Florin Corasf08f26d2018-05-10 13:20:47 -0700614 ctx->n_segs_per_evt = n_left_to_next;
615 ctx->max_len_to_snd = ctx->snd_mss * n_left_to_next;
616 }
617 ctx->left_to_snd = ctx->max_len_to_snd;
618 n_left = ctx->n_segs_per_evt;
Florin Coras66cf5e02018-06-08 09:17:39 -0700619
620 while (n_left >= 4)
621 {
622 vlib_buffer_t *b0, *b1;
623 u32 bi0, bi1;
624
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700625 pbi = wrk->tx_buffers[n_bufs - 3];
Florin Coras66cf5e02018-06-08 09:17:39 -0700626 pb = vlib_get_buffer (vm, pbi);
627 vlib_prefetch_buffer_header (pb, STORE);
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700628 pbi = wrk->tx_buffers[n_bufs - 4];
Florin Coras66cf5e02018-06-08 09:17:39 -0700629 pb = vlib_get_buffer (vm, pbi);
630 vlib_prefetch_buffer_header (pb, STORE);
631
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700632 to_next[0] = bi0 = wrk->tx_buffers[--n_bufs];
633 to_next[1] = bi1 = wrk->tx_buffers[--n_bufs];
Florin Coras66cf5e02018-06-08 09:17:39 -0700634
635 b0 = vlib_get_buffer (vm, bi0);
636 b1 = vlib_get_buffer (vm, bi1);
637
638 session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
639 session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
640
641 ctx->transport_vft->push_header (ctx->tc, b0);
642 ctx->transport_vft->push_header (ctx->tc, b1);
643
644 to_next += 2;
645 n_left_to_next -= 2;
646 n_left -= 2;
647
648 VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
649 VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b1);
650
651 vlib_validate_buffer_enqueue_x2 (vm, node, next_index, to_next,
652 n_left_to_next, bi0, bi1, next0,
653 next1);
654 }
Florin Corasf08f26d2018-05-10 13:20:47 -0700655 while (n_left)
656 {
Florin Coras66cf5e02018-06-08 09:17:39 -0700657 vlib_buffer_t *b0;
658 u32 bi0;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700659
Florin Coras91ca4622018-06-30 11:27:59 -0700660 if (n_left > 1)
661 {
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700662 pbi = wrk->tx_buffers[n_bufs - 2];
Florin Coras91ca4622018-06-30 11:27:59 -0700663 pb = vlib_get_buffer (vm, pbi);
664 vlib_prefetch_buffer_header (pb, STORE);
665 }
666
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700667 to_next[0] = bi0 = wrk->tx_buffers[--n_bufs];
Florin Coras66cf5e02018-06-08 09:17:39 -0700668 b0 = vlib_get_buffer (vm, bi0);
669 session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
Florin Coras81a13db2018-03-16 08:48:31 -0700670
Florin Coras66cf5e02018-06-08 09:17:39 -0700671 /* Ask transport to push header after current_length and
672 * total_length_not_including_first_buffer are updated */
673 ctx->transport_vft->push_header (ctx->tc, b0);
Florin Corasf6359c82017-06-19 12:26:09 -0400674
Florin Coras66cf5e02018-06-08 09:17:39 -0700675 to_next += 1;
676 n_left_to_next -= 1;
677 n_left -= 1;
Dave Barach68b0fb02017-02-28 15:15:56 -0500678
Florin Coras66cf5e02018-06-08 09:17:39 -0700679 VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
Florin Coras7fb0fe12018-04-09 09:24:52 -0700680
Florin Coras66cf5e02018-06-08 09:17:39 -0700681 vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next,
682 n_left_to_next, bi0, next0);
Dave Barach68b0fb02017-02-28 15:15:56 -0500683 }
Florin Coras66cf5e02018-06-08 09:17:39 -0700684
685 if (PREDICT_FALSE (n_trace > 0))
686 session_tx_trace_frame (vm, node, next_index, to_next,
687 ctx->n_segs_per_evt, s, n_trace);
688
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700689 _vec_len (wrk->tx_buffers) = n_bufs;
Florin Corasf08f26d2018-05-10 13:20:47 -0700690 *n_tx_packets += ctx->n_segs_per_evt;
Florin Corasd67f1122018-05-21 17:47:40 -0700691 transport_connection_update_tx_stats (ctx->tc, ctx->max_len_to_snd);
Florin Coras8e43d042018-05-04 15:46:57 -0700692 vlib_put_next_frame (vm, node, next_index, n_left_to_next);
Dave Barach68b0fb02017-02-28 15:15:56 -0500693
Florin Coras6792ec02017-03-13 03:49:51 -0700694 /* If we couldn't dequeue all bytes mark as partially read */
Florin Corasf08f26d2018-05-10 13:20:47 -0700695 ASSERT (ctx->left_to_snd == 0);
Florin Coras8e43d042018-05-04 15:46:57 -0700696 if (ctx->max_len_to_snd < ctx->max_dequeue)
697 if (svm_fifo_set_event (s->server_tx_fifo))
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700698 vec_add1 (wrk->pending_event_vector, *e);
Florin Coras7fb0fe12018-04-09 09:24:52 -0700699
Florin Coras8e43d042018-05-04 15:46:57 -0700700 if (!peek_data && ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
Dave Barach68b0fb02017-02-28 15:15:56 -0500701 {
Florin Coras7fb0fe12018-04-09 09:24:52 -0700702 /* Fix dgram pre header */
Florin Coras8e43d042018-05-04 15:46:57 -0700703 if (ctx->max_len_to_snd < ctx->max_dequeue)
704 svm_fifo_overwrite_head (s->server_tx_fifo, (u8 *) & ctx->hdr,
Florin Coras7fb0fe12018-04-09 09:24:52 -0700705 sizeof (session_dgram_pre_hdr_t));
706 /* More data needs to be read */
Florin Coras8e43d042018-05-04 15:46:57 -0700707 else if (svm_fifo_max_dequeue (s->server_tx_fifo) > 0)
708 if (svm_fifo_set_event (s->server_tx_fifo))
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700709 vec_add1 (wrk->pending_event_vector, *e);
Dave Barach68b0fb02017-02-28 15:15:56 -0500710 }
Florin Coras0d60a0f2018-06-29 02:02:08 -0700711 return SESSION_TX_OK;
Dave Barach68b0fb02017-02-28 15:15:56 -0500712}
713
714int
Florin Corasd79b41e2017-03-04 05:37:52 -0800715session_tx_fifo_peek_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
Florin Coras52207f12018-07-12 14:48:06 -0700716 session_event_t * e,
Florin Coras0d60a0f2018-06-29 02:02:08 -0700717 stream_session_t * s, int *n_tx_pkts)
Dave Barach68b0fb02017-02-28 15:15:56 -0500718{
Florin Coras0d60a0f2018-06-29 02:02:08 -0700719 return session_tx_fifo_read_and_snd_i (vm, node, e, s, n_tx_pkts, 1);
Dave Barach68b0fb02017-02-28 15:15:56 -0500720}
721
722int
Florin Corasd79b41e2017-03-04 05:37:52 -0800723session_tx_fifo_dequeue_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
Florin Coras52207f12018-07-12 14:48:06 -0700724 session_event_t * e,
Florin Coras0d60a0f2018-06-29 02:02:08 -0700725 stream_session_t * s, int *n_tx_pkts)
Dave Barach68b0fb02017-02-28 15:15:56 -0500726{
Florin Coras0d60a0f2018-06-29 02:02:08 -0700727 return session_tx_fifo_read_and_snd_i (vm, node, e, s, n_tx_pkts, 0);
Dave Barach68b0fb02017-02-28 15:15:56 -0500728}
729
Florin Coras371ca502018-02-21 12:07:41 -0800730int
731session_tx_fifo_dequeue_internal (vlib_main_t * vm,
732 vlib_node_runtime_t * node,
Florin Coras52207f12018-07-12 14:48:06 -0700733 session_event_t * e,
Florin Coras0d60a0f2018-06-29 02:02:08 -0700734 stream_session_t * s, int *n_tx_pkts)
Florin Coras371ca502018-02-21 12:07:41 -0800735{
736 application_t *app;
Florin Corasef915342018-09-29 10:23:06 -0700737 if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED))
738 return 0;
Florin Corasab2f6db2018-08-31 14:31:41 -0700739 app = application_get (s->t_app_index);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700740 svm_fifo_unset_event (s->server_tx_fifo);
741 return app->cb_fns.builtin_app_tx_callback (s);
Florin Coras371ca502018-02-21 12:07:41 -0800742}
743
Dave Barach2c25a622017-06-26 11:35:07 -0400744always_inline stream_session_t *
Florin Coras52207f12018-07-12 14:48:06 -0700745session_event_get_session (session_event_t * e, u8 thread_index)
Florin Corasa5464812017-04-19 13:00:05 -0700746{
Florin Coras3cbc04b2017-10-02 00:18:51 -0700747 return session_get_if_valid (e->fifo->master_session_index, thread_index);
Florin Corasa5464812017-04-19 13:00:05 -0700748}
749
Florin Corasd67f1122018-05-21 17:47:40 -0700750static void
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700751session_update_dispatch_period (session_manager_worker_t * wrk, f64 now,
Florin Corasd67f1122018-05-21 17:47:40 -0700752 u32 thread_index)
753{
Florin Corasefefc6b2018-11-07 12:49:19 -0800754 if (wrk->last_tx_packets)
Florin Corasc44a5582018-11-01 16:30:54 -0700755 {
756 f64 sample = now - wrk->last_vlib_time;
Florin Corasc44a5582018-11-01 16:30:54 -0700757 wrk->dispatch_period = (wrk->dispatch_period + sample) * 0.5;
758 }
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700759 wrk->last_vlib_time = now;
Florin Corasd67f1122018-05-21 17:47:40 -0700760}
761
Florin Coras0d60a0f2018-06-29 02:02:08 -0700762static uword
763session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
764 vlib_frame_t * frame)
765{
766 session_manager_main_t *smm = vnet_get_session_manager_main ();
Florin Coras3c2fed52018-07-04 04:15:05 -0700767 u32 thread_index = vm->thread_index, n_to_dequeue, n_events;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700768 session_manager_worker_t *wrk = &smm->wrk[thread_index];
Florin Coras5f56d732018-10-30 10:21:59 -0700769 session_event_t *e, *fifo_events;
Florin Coras3c2fed52018-07-04 04:15:05 -0700770 svm_msg_q_msg_t _msg, *msg = &_msg;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700771 f64 now = vlib_time_now (vm);
Florin Coras3c2fed52018-07-04 04:15:05 -0700772 int n_tx_packets = 0, i, rv;
Florin Coras15531972018-08-12 23:50:53 -0700773 app_worker_t *app_wrk;
Florin Coras3c2fed52018-07-04 04:15:05 -0700774 application_t *app;
775 svm_msg_q_t *mq;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700776 void (*fp) (void *);
777
778 SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, thread_index);
779
780 /*
781 * Update transport time
782 */
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700783 session_update_dispatch_period (wrk, now, thread_index);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700784 transport_update_time (now, thread_index);
785
Florin Coras0d60a0f2018-06-29 02:02:08 -0700786 SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0);
787
Florin Coras5f56d732018-10-30 10:21:59 -0700788 /* Make sure postponed events are handled first */
789 fifo_events = wrk->free_event_vector;
790 vec_append (fifo_events, wrk->postponed_event_vector);
Florin Coras36ee9f12018-11-02 12:52:10 -0700791 _vec_len (wrk->postponed_event_vector) = 0;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700792
Florin Coras5f56d732018-10-30 10:21:59 -0700793 /* Try to dequeue what is available. Don't wait for lock.
794 * XXX: we may need priorities here */
795 mq = wrk->vpp_event_queue;
796 n_to_dequeue = svm_msg_q_size (mq);
797 if (n_to_dequeue && svm_msg_q_try_lock (mq) == 0)
798 {
799 for (i = 0; i < n_to_dequeue; i++)
800 {
801 vec_add2 (fifo_events, e, 1);
802 svm_msg_q_sub_w_lock (mq, msg);
Dave Barach178cf492018-11-13 16:34:13 -0500803 clib_memcpy_fast (e, svm_msg_q_msg_data (mq, msg), sizeof (*e));
Florin Coras5f56d732018-10-30 10:21:59 -0700804 svm_msg_q_free_msg (mq, msg);
805 }
806 svm_msg_q_unlock (mq);
807 }
808
809 vec_append (fifo_events, wrk->pending_event_vector);
810 vec_append (fifo_events, wrk->pending_disconnects);
811
Florin Coras5f56d732018-10-30 10:21:59 -0700812 _vec_len (wrk->pending_event_vector) = 0;
Florin Coras36ee9f12018-11-02 12:52:10 -0700813 _vec_len (wrk->pending_disconnects) = 0;
Florin Coras5f56d732018-10-30 10:21:59 -0700814
Florin Coras0d60a0f2018-06-29 02:02:08 -0700815 n_events = vec_len (fifo_events);
Florin Coras5f56d732018-10-30 10:21:59 -0700816 if (PREDICT_FALSE (!n_events))
817 return 0;
818
Florin Coras0d60a0f2018-06-29 02:02:08 -0700819 for (i = 0; i < n_events; i++)
820 {
821 stream_session_t *s; /* $$$ prefetch 1 ahead maybe */
Florin Coras52207f12018-07-12 14:48:06 -0700822 session_event_t *e;
Florin Coras0e88e852018-09-17 22:09:02 -0700823 u8 want_tx_evt;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700824
825 e = &fifo_events[i];
826 switch (e->event_type)
827 {
828 case FIFO_EVENT_APP_TX:
829 /* Don't try to send more that one frame per dispatch cycle */
830 if (n_tx_packets == VLIB_FRAME_SIZE)
831 {
Florin Coras5f56d732018-10-30 10:21:59 -0700832 vec_add1 (wrk->postponed_event_vector, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700833 break;
834 }
835
836 s = session_event_get_session (e, thread_index);
837 if (PREDICT_FALSE (!s))
838 {
Florin Coras326b81e2018-10-04 19:03:05 -0700839 clib_warning ("session was freed!");
Florin Coras0d60a0f2018-06-29 02:02:08 -0700840 continue;
841 }
842
Florin Coras0e88e852018-09-17 22:09:02 -0700843 want_tx_evt = svm_fifo_want_tx_evt (s->server_tx_fifo);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700844 /* Spray packets in per session type frames, since they go to
845 * different nodes */
846 rv = (smm->session_tx_fns[s->session_type]) (vm, node, e, s,
847 &n_tx_packets);
848 if (PREDICT_TRUE (rv == SESSION_TX_OK))
849 {
Florin Coras0e88e852018-09-17 22:09:02 -0700850 if (PREDICT_FALSE (want_tx_evt))
851 {
852 svm_fifo_set_want_tx_evt (s->server_tx_fifo, 0);
853 session_dequeue_notify (s);
854 }
Florin Coras0d60a0f2018-06-29 02:02:08 -0700855 }
856 else if (PREDICT_FALSE (rv == SESSION_TX_NO_BUFFERS))
857 {
858 vlib_node_increment_counter (vm, node->node_index,
859 SESSION_QUEUE_ERROR_NO_BUFFER, 1);
860 continue;
861 }
862 break;
863 case FIFO_EVENT_DISCONNECT:
864 /* Make sure stream disconnects run after the pending list is
865 * drained */
866 s = session_get_from_handle (e->session_handle);
867 if (!e->postponed)
868 {
869 e->postponed = 1;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700870 vec_add1 (wrk->pending_disconnects, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700871 continue;
872 }
873 /* If tx queue is still not empty, wait */
874 if (svm_fifo_max_dequeue (s->server_tx_fifo))
875 {
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700876 vec_add1 (wrk->pending_disconnects, *e);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700877 continue;
878 }
879
880 stream_session_disconnect_transport (s);
881 break;
882 case FIFO_EVENT_BUILTIN_RX:
883 s = session_event_get_session (e, thread_index);
Florin Corasef915342018-09-29 10:23:06 -0700884 if (PREDICT_FALSE (!s || s->session_state >= SESSION_STATE_CLOSING))
Florin Coras0d60a0f2018-06-29 02:02:08 -0700885 continue;
886 svm_fifo_unset_event (s->server_rx_fifo);
Florin Coras15531972018-08-12 23:50:53 -0700887 app_wrk = app_worker_get (s->app_wrk_index);
888 app = application_get (app_wrk->app_index);
Florin Coras0d60a0f2018-06-29 02:02:08 -0700889 app->cb_fns.builtin_app_rx_callback (s);
890 break;
Florin Corasef915342018-09-29 10:23:06 -0700891 case FIFO_EVENT_BUILTIN_TX:
892 s = session_get_from_handle_if_valid (e->session_handle);
893 if (PREDICT_TRUE (s != 0))
894 session_tx_fifo_dequeue_internal (vm, node, e, s, &n_tx_packets);
895 break;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700896 case FIFO_EVENT_RPC:
897 fp = e->rpc_args.fp;
898 (*fp) (e->rpc_args.arg);
899 break;
Florin Coras52207f12018-07-12 14:48:06 -0700900 case SESSION_CTRL_EVT_DISCONNECTED:
901 session_mq_disconnected_handler (e->data);
902 break;
903 case SESSION_CTRL_EVT_ACCEPTED_REPLY:
904 session_mq_accepted_reply_handler (e->data);
905 break;
906 case SESSION_CTRL_EVT_CONNECTED_REPLY:
907 break;
908 case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
909 session_mq_disconnected_reply_handler (e->data);
910 break;
911 case SESSION_CTRL_EVT_RESET_REPLY:
912 session_mq_reset_reply_handler (e->data);
913 break;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700914 default:
915 clib_warning ("unhandled event type %d", e->event_type);
916 }
917 }
918
919 _vec_len (fifo_events) = 0;
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700920 wrk->free_event_vector = fifo_events;
Florin Corasc44a5582018-11-01 16:30:54 -0700921 wrk->last_tx_packets = n_tx_packets;
Florin Coras0d60a0f2018-06-29 02:02:08 -0700922
923 vlib_node_increment_counter (vm, session_queue_node.index,
924 SESSION_QUEUE_ERROR_TX, n_tx_packets);
925
926 SESSION_EVT_DBG (SESSION_EVT_DISPATCH_END, smm, thread_index);
927
928 return n_tx_packets;
929}
930
931/* *INDENT-OFF* */
932VLIB_REGISTER_NODE (session_queue_node) =
933{
934 .function = session_queue_node_fn,
935 .name = "session-queue",
936 .format_trace = format_session_queue_trace,
937 .type = VLIB_NODE_TYPE_INPUT,
938 .n_errors = ARRAY_LEN (session_queue_error_strings),
939 .error_strings = session_queue_error_strings,
940 .state = VLIB_NODE_STATE_DISABLED,
941};
942/* *INDENT-ON* */
943
Dave Barachacd2a6a2017-05-16 17:41:34 -0400944void
945dump_thread_0_event_queue (void)
946{
947 session_manager_main_t *smm = vnet_get_session_manager_main ();
948 vlib_main_t *vm = &vlib_global_main;
949 u32 my_thread_index = vm->thread_index;
Florin Coras52207f12018-07-12 14:48:06 -0700950 session_event_t _e, *e = &_e;
Florin Coras3c2fed52018-07-04 04:15:05 -0700951 svm_msg_q_ring_t *ring;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400952 stream_session_t *s0;
Florin Coras3c2fed52018-07-04 04:15:05 -0700953 svm_msg_q_msg_t *msg;
954 svm_msg_q_t *mq;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400955 int i, index;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400956
Florin Coras5a7ca7b2018-10-30 12:01:48 -0700957 mq = smm->wrk[my_thread_index].vpp_event_queue;
Florin Coras3c2fed52018-07-04 04:15:05 -0700958 index = mq->q->head;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400959
Florin Coras3c2fed52018-07-04 04:15:05 -0700960 for (i = 0; i < mq->q->cursize; i++)
Dave Barachacd2a6a2017-05-16 17:41:34 -0400961 {
Florin Coras3c2fed52018-07-04 04:15:05 -0700962 msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
963 ring = svm_msg_q_ring (mq, msg->ring_index);
Dave Barach178cf492018-11-13 16:34:13 -0500964 clib_memcpy_fast (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
Dave Barachacd2a6a2017-05-16 17:41:34 -0400965
966 switch (e->event_type)
967 {
968 case FIFO_EVENT_APP_TX:
969 s0 = session_event_get_session (e, my_thread_index);
970 fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
971 break;
972
973 case FIFO_EVENT_DISCONNECT:
Florin Corascea194d2017-10-02 00:18:51 -0700974 s0 = session_get_from_handle (e->session_handle);
Dave Barachacd2a6a2017-05-16 17:41:34 -0400975 fformat (stdout, "[%04d] disconnect session %d\n", i,
976 s0->session_index);
977 break;
978
979 case FIFO_EVENT_BUILTIN_RX:
980 s0 = session_event_get_session (e, my_thread_index);
981 fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
982 break;
983
984 case FIFO_EVENT_RPC:
985 fformat (stdout, "[%04d] RPC call %llx with %llx\n",
986 i, (u64) (e->rpc_args.fp), (u64) (e->rpc_args.arg));
987 break;
988
989 default:
990 fformat (stdout, "[%04d] unhandled event type %d\n",
991 i, e->event_type);
992 break;
993 }
994
995 index++;
996
Florin Coras3c2fed52018-07-04 04:15:05 -0700997 if (index == mq->q->maxsize)
Dave Barachacd2a6a2017-05-16 17:41:34 -0400998 index = 0;
999 }
1000}
1001
Florin Coras6534b7a2017-07-18 05:38:03 -04001002static u8
Florin Coras52207f12018-07-12 14:48:06 -07001003session_node_cmp_event (session_event_t * e, svm_fifo_t * f)
Florin Coras6534b7a2017-07-18 05:38:03 -04001004{
1005 stream_session_t *s;
1006 switch (e->event_type)
1007 {
1008 case FIFO_EVENT_APP_RX:
1009 case FIFO_EVENT_APP_TX:
1010 case FIFO_EVENT_BUILTIN_RX:
1011 if (e->fifo == f)
1012 return 1;
1013 break;
1014 case FIFO_EVENT_DISCONNECT:
1015 break;
1016 case FIFO_EVENT_RPC:
Florin Corascea194d2017-10-02 00:18:51 -07001017 s = session_get_from_handle (e->session_handle);
Florin Coras6534b7a2017-07-18 05:38:03 -04001018 if (!s)
1019 {
1020 clib_warning ("session has event but doesn't exist!");
1021 break;
1022 }
1023 if (s->server_rx_fifo == f || s->server_tx_fifo == f)
1024 return 1;
1025 break;
1026 default:
1027 break;
1028 }
1029 return 0;
1030}
1031
1032u8
Florin Coras52207f12018-07-12 14:48:06 -07001033session_node_lookup_fifo_event (svm_fifo_t * f, session_event_t * e)
Florin Coras6534b7a2017-07-18 05:38:03 -04001034{
Florin Coras52207f12018-07-12 14:48:06 -07001035 session_event_t *pending_event_vector, *evt;
Florin Coras5a7ca7b2018-10-30 12:01:48 -07001036 session_manager_worker_t *wrk;
Florin Coras6534b7a2017-07-18 05:38:03 -04001037 int i, index, found = 0;
Florin Coras3c2fed52018-07-04 04:15:05 -07001038 svm_msg_q_msg_t *msg;
1039 svm_msg_q_ring_t *ring;
Florin Coras5a7ca7b2018-10-30 12:01:48 -07001040 svm_msg_q_t *mq;
Florin Coras6534b7a2017-07-18 05:38:03 -04001041 u8 thread_index;
1042
1043 ASSERT (e);
1044 thread_index = f->master_thread_index;
Florin Coras5a7ca7b2018-10-30 12:01:48 -07001045 wrk = session_manager_get_worker (thread_index);
1046
Florin Coras6534b7a2017-07-18 05:38:03 -04001047 /*
1048 * Search evt queue
1049 */
Florin Coras5a7ca7b2018-10-30 12:01:48 -07001050 mq = wrk->vpp_event_queue;
Florin Coras3c2fed52018-07-04 04:15:05 -07001051 index = mq->q->head;
1052 for (i = 0; i < mq->q->cursize; i++)
Florin Coras6534b7a2017-07-18 05:38:03 -04001053 {
Florin Coras3c2fed52018-07-04 04:15:05 -07001054 msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
1055 ring = svm_msg_q_ring (mq, msg->ring_index);
Dave Barach178cf492018-11-13 16:34:13 -05001056 clib_memcpy_fast (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
Florin Coras6534b7a2017-07-18 05:38:03 -04001057 found = session_node_cmp_event (e, f);
1058 if (found)
Florin Coras371ca502018-02-21 12:07:41 -08001059 return 1;
Florin Coras3c2fed52018-07-04 04:15:05 -07001060 if (++index == mq->q->maxsize)
Florin Coras6534b7a2017-07-18 05:38:03 -04001061 index = 0;
1062 }
1063 /*
1064 * Search pending events vector
1065 */
Florin Coras5a7ca7b2018-10-30 12:01:48 -07001066 pending_event_vector = wrk->pending_event_vector;
Florin Coras6534b7a2017-07-18 05:38:03 -04001067 vec_foreach (evt, pending_event_vector)
1068 {
1069 found = session_node_cmp_event (evt, f);
1070 if (found)
1071 {
Dave Barach178cf492018-11-13 16:34:13 -05001072 clib_memcpy_fast (e, evt, sizeof (*evt));
Florin Coras6534b7a2017-07-18 05:38:03 -04001073 break;
1074 }
1075 }
1076 return found;
1077}
1078
Dave Barach2a863912017-11-28 10:11:42 -05001079static clib_error_t *
1080session_queue_exit (vlib_main_t * vm)
1081{
1082 if (vec_len (vlib_mains) < 2)
1083 return 0;
1084
1085 /*
1086 * Shut off (especially) worker-thread session nodes.
1087 * Otherwise, vpp can crash as the main thread unmaps the
1088 * API segment.
1089 */
1090 vlib_worker_thread_barrier_sync (vm);
1091 session_node_enable_disable (0 /* is_enable */ );
1092 vlib_worker_thread_barrier_release (vm);
1093 return 0;
1094}
1095
1096VLIB_MAIN_LOOP_EXIT_FUNCTION (session_queue_exit);
1097
Florin Corasfd542f12018-05-16 19:28:24 -07001098static uword
1099session_queue_process (vlib_main_t * vm, vlib_node_runtime_t * rt,
1100 vlib_frame_t * f)
1101{
1102 f64 now, timeout = 1.0;
1103 uword *event_data = 0;
1104 uword event_type;
1105
1106 while (1)
1107 {
1108 vlib_process_wait_for_event_or_clock (vm, timeout);
1109 now = vlib_time_now (vm);
1110 event_type = vlib_process_get_events (vm, (uword **) & event_data);
1111
1112 switch (event_type)
1113 {
1114 case SESSION_Q_PROCESS_FLUSH_FRAMES:
1115 /* Flush the frames by updating all transports times */
1116 transport_update_time (now, 0);
1117 break;
1118 case SESSION_Q_PROCESS_STOP:
1119 timeout = 100000.0;
1120 break;
1121 case ~0:
1122 /* Timed out. Update time for all transports to trigger all
1123 * outstanding retransmits. */
1124 transport_update_time (now, 0);
1125 break;
1126 }
1127 vec_reset_length (event_data);
1128 }
1129 return 0;
1130}
1131
1132/* *INDENT-OFF* */
1133VLIB_REGISTER_NODE (session_queue_process_node) =
1134{
1135 .function = session_queue_process,
1136 .type = VLIB_NODE_TYPE_PROCESS,
1137 .name = "session-queue-process",
1138 .state = VLIB_NODE_STATE_DISABLED,
1139};
1140/* *INDENT-ON* */
1141
1142
Dave Barach68b0fb02017-02-28 15:15:56 -05001143/*
1144 * fd.io coding-style-patch-verification: ON
1145 *
1146 * Local Variables:
1147 * eval: (c-set-style "gnu")
1148 * End:
1149 */