blob: e9cfc7be8aa1908e402f9160af7a3192c9fce9a6 [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 Coras6792ec02017-03-13 03:49:51 -070021#include <vnet/session/application.h>
Florin Corase69f4952017-03-07 10:06:24 -080022#include <vnet/session/session_debug.h>
Florin Coras6792ec02017-03-13 03:49:51 -070023#include <vlibmemory/unix_shared_memory_queue.h>
Dave Barach68b0fb02017-02-28 15:15:56 -050024
25vlib_node_registration_t session_queue_node;
26
27typedef struct
28{
29 u32 session_index;
30 u32 server_thread_index;
31} session_queue_trace_t;
32
33/* packet trace format function */
34static u8 *
35format_session_queue_trace (u8 * s, va_list * args)
36{
37 CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
38 CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
39 session_queue_trace_t *t = va_arg (*args, session_queue_trace_t *);
40
41 s = format (s, "SESSION_QUEUE: session index %d, server thread index %d",
42 t->session_index, t->server_thread_index);
43 return s;
44}
45
46vlib_node_registration_t session_queue_node;
47
Florin Coras6792ec02017-03-13 03:49:51 -070048#define foreach_session_queue_error \
49_(TX, "Packets transmitted") \
Florin Coras93992a92017-05-24 18:03:56 -070050_(TIMER, "Timer events") \
51_(NO_BUFFER, "Out of buffers")
Dave Barach68b0fb02017-02-28 15:15:56 -050052
53typedef enum
54{
55#define _(sym,str) SESSION_QUEUE_ERROR_##sym,
56 foreach_session_queue_error
57#undef _
58 SESSION_QUEUE_N_ERROR,
59} session_queue_error_t;
60
61static char *session_queue_error_strings[] = {
62#define _(sym,string) string,
63 foreach_session_queue_error
64#undef _
65};
66
Florin Corasf6d68ed2017-05-07 19:12:02 -070067always_inline void
68session_tx_fifo_chain_tail (session_manager_main_t * smm, vlib_main_t * vm,
69 u8 thread_index, svm_fifo_t * fifo,
70 vlib_buffer_t * b0, u32 bi0, u8 n_bufs_per_seg,
Florin Corasb2215d62017-08-01 16:56:58 -070071 u32 left_from_seg, u32 * left_to_snd0,
Florin Coras1f152cd2017-08-18 19:28:03 -070072 u16 * n_bufs, u32 * tx_offset, u16 deq_per_buf,
Florin Corasb2215d62017-08-01 16:56:58 -070073 u8 peek_data)
Florin Corasf6d68ed2017-05-07 19:12:02 -070074{
75 vlib_buffer_t *chain_b0, *prev_b0;
Florin Corasb2215d62017-08-01 16:56:58 -070076 u32 chain_bi0, to_deq;
Florin Corasf6d68ed2017-05-07 19:12:02 -070077 u16 len_to_deq0, n_bytes_read;
78 u8 *data0, j;
79
Florin Corasb2215d62017-08-01 16:56:58 -070080 b0->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
81 b0->total_length_not_including_first_buffer = 0;
82
Florin Corasf6d68ed2017-05-07 19:12:02 -070083 chain_bi0 = bi0;
84 chain_b0 = b0;
Florin Corasb2215d62017-08-01 16:56:58 -070085 to_deq = left_from_seg;
Florin Corasf6d68ed2017-05-07 19:12:02 -070086 for (j = 1; j < n_bufs_per_seg; j++)
87 {
88 prev_b0 = chain_b0;
Florin Corasb2215d62017-08-01 16:56:58 -070089 len_to_deq0 = clib_min (to_deq, deq_per_buf);
Florin Corasf6d68ed2017-05-07 19:12:02 -070090
91 *n_bufs -= 1;
92 chain_bi0 = smm->tx_buffers[thread_index][*n_bufs];
93 _vec_len (smm->tx_buffers[thread_index]) = *n_bufs;
94
95 chain_b0 = vlib_get_buffer (vm, chain_bi0);
96 chain_b0->current_data = 0;
97 data0 = vlib_buffer_get_current (chain_b0);
98 if (peek_data)
99 {
Florin Coras1f152cd2017-08-18 19:28:03 -0700100 n_bytes_read = svm_fifo_peek (fifo, *tx_offset, len_to_deq0, data0);
101 *tx_offset += n_bytes_read;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700102 }
103 else
104 {
105 n_bytes_read = svm_fifo_dequeue_nowait (fifo, len_to_deq0, data0);
106 }
107 ASSERT (n_bytes_read == len_to_deq0);
108 chain_b0->current_length = n_bytes_read;
109 b0->total_length_not_including_first_buffer += chain_b0->current_length;
110
111 /* update previous buffer */
112 prev_b0->next_buffer = chain_bi0;
113 prev_b0->flags |= VLIB_BUFFER_NEXT_PRESENT;
114
115 /* update current buffer */
116 chain_b0->next_buffer = 0;
117
Florin Corasb2215d62017-08-01 16:56:58 -0700118 to_deq -= n_bytes_read;
119 if (to_deq == 0)
Florin Corasf6d68ed2017-05-07 19:12:02 -0700120 break;
121 }
Florin Coras1f152cd2017-08-18 19:28:03 -0700122 ASSERT (to_deq == 0
123 && b0->total_length_not_including_first_buffer == left_from_seg);
Florin Corasb2215d62017-08-01 16:56:58 -0700124 *left_to_snd0 -= left_from_seg;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700125}
126
Dave Barach68b0fb02017-02-28 15:15:56 -0500127always_inline int
Florin Corasd79b41e2017-03-04 05:37:52 -0800128session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
129 session_manager_main_t * smm,
130 session_fifo_event_t * e0,
131 stream_session_t * s0, u32 thread_index,
132 int *n_tx_packets, u8 peek_data)
Dave Barach68b0fb02017-02-28 15:15:56 -0500133{
134 u32 n_trace = vlib_get_trace_count (vm, node);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700135 u32 left_to_snd0, max_len_to_snd0, len_to_deq0, snd_space0;
Florin Corase87216f2017-08-17 16:59:22 -0700136 u32 n_bufs_per_evt, n_frames_per_evt, n_bufs_per_frame;
Dave Barach68b0fb02017-02-28 15:15:56 -0500137 transport_connection_t *tc0;
138 transport_proto_vft_t *transport_vft;
Florin Coras561af9b2017-12-09 10:19:43 -0800139 transport_proto_t tp;
Dave Barach68b0fb02017-02-28 15:15:56 -0500140 u32 next_index, next0, *to_next, n_left_to_next, bi0;
141 vlib_buffer_t *b0;
Florin Coras1f152cd2017-08-18 19:28:03 -0700142 u32 tx_offset = 0, max_dequeue0, n_bytes_per_seg, left_for_seg;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700143 u16 snd_mss0, n_bufs_per_seg, n_bufs;
Dave Barach68b0fb02017-02-28 15:15:56 -0500144 u8 *data0;
Florin Coras6792ec02017-03-13 03:49:51 -0700145 int i, n_bytes_read;
Florin Corase87216f2017-08-17 16:59:22 -0700146 u32 n_bytes_per_buf, deq_per_buf, deq_per_first_buf;
Florin Coras93992a92017-05-24 18:03:56 -0700147 u32 buffers_allocated, buffers_allocated_this_call;
Dave Barach68b0fb02017-02-28 15:15:56 -0500148
Florin Coras561af9b2017-12-09 10:19:43 -0800149 next_index = next0 = smm->session_type_to_next[s0->session_type];
Dave Barach68b0fb02017-02-28 15:15:56 -0500150
Florin Coras561af9b2017-12-09 10:19:43 -0800151 tp = session_get_transport_proto (s0);
152 transport_vft = transport_protocol_get_vft (tp);
Dave Barach68b0fb02017-02-28 15:15:56 -0500153 tc0 = transport_vft->get_connection (s0->connection_index, thread_index);
154
155 /* Make sure we have space to send and there's something to dequeue */
Dave Barach68b0fb02017-02-28 15:15:56 -0500156 snd_mss0 = transport_vft->send_mss (tc0);
Florin Corasc8343412017-05-04 14:25:50 -0700157 snd_space0 = transport_vft->send_space (tc0);
Dave Barach68b0fb02017-02-28 15:15:56 -0500158
Florin Corase04c2992017-03-01 08:17:34 -0800159 /* Can't make any progress */
Florin Coras6792ec02017-03-13 03:49:51 -0700160 if (snd_space0 == 0 || snd_mss0 == 0)
Florin Corase04c2992017-03-01 08:17:34 -0800161 {
Dave Barachacd2a6a2017-05-16 17:41:34 -0400162 vec_add1 (smm->pending_event_vector[thread_index], *e0);
Florin Corase04c2992017-03-01 08:17:34 -0800163 return 0;
164 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500165
Florin Coras6a9b68b2017-11-21 04:20:42 -0800166 /* Allow enqueuing of a new event */
167 svm_fifo_unset_event (s0->server_tx_fifo);
168
Florin Coras9d063042017-09-14 03:08:00 -0400169 /* Check how much we can pull. */
170 max_dequeue0 = svm_fifo_max_dequeue (s0->server_tx_fifo);
171
Dave Barach68b0fb02017-02-28 15:15:56 -0500172 if (peek_data)
173 {
Florin Coras9d063042017-09-14 03:08:00 -0400174 /* Offset in rx fifo from where to peek data */
Florin Coras1f152cd2017-08-18 19:28:03 -0700175 tx_offset = transport_vft->tx_fifo_offset (tc0);
Florin Coras9d063042017-09-14 03:08:00 -0400176 if (PREDICT_FALSE (tx_offset >= max_dequeue0))
177 max_dequeue0 = 0;
178 else
179 max_dequeue0 -= tx_offset;
Dave Barach68b0fb02017-02-28 15:15:56 -0500180 }
181
Florin Coras6792ec02017-03-13 03:49:51 -0700182 /* Nothing to read return */
183 if (max_dequeue0 == 0)
Florin Coras6a9b68b2017-11-21 04:20:42 -0800184 return 0;
Florin Coras6792ec02017-03-13 03:49:51 -0700185
186 /* Ensure we're not writing more than transport window allows */
Florin Coras3e350af2017-03-30 02:54:28 -0700187 if (max_dequeue0 < snd_space0)
188 {
189 /* Constrained by tx queue. Try to send only fully formed segments */
190 max_len_to_snd0 = (max_dequeue0 > snd_mss0) ?
191 max_dequeue0 - max_dequeue0 % snd_mss0 : max_dequeue0;
192 /* TODO Nagle ? */
193 }
194 else
195 {
Florin Coras1f152cd2017-08-18 19:28:03 -0700196 /* Expectation is that snd_space0 is already a multiple of snd_mss */
Florin Coras3e350af2017-03-30 02:54:28 -0700197 max_len_to_snd0 = snd_space0;
198 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500199
Florin Coras93992a92017-05-24 18:03:56 -0700200 n_bytes_per_buf = vlib_buffer_free_list_buffer_size
201 (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
Florin Corase87216f2017-08-17 16:59:22 -0700202 ASSERT (n_bytes_per_buf > MAX_HDRS_LEN);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700203 n_bytes_per_seg = MAX_HDRS_LEN + snd_mss0;
204 n_bufs_per_seg = ceil ((double) n_bytes_per_seg / n_bytes_per_buf);
Florin Corase87216f2017-08-17 16:59:22 -0700205 n_bufs_per_evt = ceil ((double) max_len_to_snd0 / n_bytes_per_seg);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700206 n_frames_per_evt = ceil ((double) n_bufs_per_evt / VLIB_FRAME_SIZE);
Florin Corase87216f2017-08-17 16:59:22 -0700207 n_bufs_per_frame = n_bufs_per_seg * VLIB_FRAME_SIZE;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700208
209 deq_per_buf = clib_min (snd_mss0, n_bytes_per_buf);
Florin Corase87216f2017-08-17 16:59:22 -0700210 deq_per_first_buf = clib_min (snd_mss0, n_bytes_per_buf - MAX_HDRS_LEN);
Dave Barach68b0fb02017-02-28 15:15:56 -0500211
212 n_bufs = vec_len (smm->tx_buffers[thread_index]);
213 left_to_snd0 = max_len_to_snd0;
214 for (i = 0; i < n_frames_per_evt; i++)
215 {
216 /* Make sure we have at least one full frame of buffers ready */
Florin Corase87216f2017-08-17 16:59:22 -0700217 if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
Dave Barach68b0fb02017-02-28 15:15:56 -0500218 {
219 vec_validate (smm->tx_buffers[thread_index],
Florin Corase87216f2017-08-17 16:59:22 -0700220 n_bufs + n_bufs_per_frame - 1);
Florin Coras93992a92017-05-24 18:03:56 -0700221 buffers_allocated = 0;
222 do
Dave Barach68b0fb02017-02-28 15:15:56 -0500223 {
Florin Corase87216f2017-08-17 16:59:22 -0700224 buffers_allocated_this_call = vlib_buffer_alloc (vm,
225 &smm->tx_buffers
226 [thread_index]
227 [n_bufs +
228 buffers_allocated],
229 n_bufs_per_frame
230 -
231 buffers_allocated);
Florin Coras93992a92017-05-24 18:03:56 -0700232 buffers_allocated += buffers_allocated_this_call;
Dave Barach68b0fb02017-02-28 15:15:56 -0500233 }
Florin Coras93992a92017-05-24 18:03:56 -0700234 while (buffers_allocated_this_call > 0
Florin Corase87216f2017-08-17 16:59:22 -0700235 && ((buffers_allocated + n_bufs < n_bufs_per_frame)));
Florin Coras93992a92017-05-24 18:03:56 -0700236
237 n_bufs += buffers_allocated;
Dave Barach68b0fb02017-02-28 15:15:56 -0500238 _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
Florin Coras93992a92017-05-24 18:03:56 -0700239
Florin Corase87216f2017-08-17 16:59:22 -0700240 if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
Florin Coras93992a92017-05-24 18:03:56 -0700241 {
242 vec_add1 (smm->pending_event_vector[thread_index], *e0);
243 return -1;
244 }
Florin Corase87216f2017-08-17 16:59:22 -0700245 ASSERT (n_bufs >= n_bufs_per_frame);
Dave Barach68b0fb02017-02-28 15:15:56 -0500246 }
247
248 vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
Florin Corase87216f2017-08-17 16:59:22 -0700249 while (left_to_snd0 && n_left_to_next)
Dave Barach68b0fb02017-02-28 15:15:56 -0500250 {
Florin Corasf6d68ed2017-05-07 19:12:02 -0700251 /*
252 * Handle first buffer in chain separately
253 */
254
Dave Barach68b0fb02017-02-28 15:15:56 -0500255 /* Get free buffer */
Florin Coras93992a92017-05-24 18:03:56 -0700256 ASSERT (n_bufs >= 1);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700257 bi0 = smm->tx_buffers[thread_index][--n_bufs];
Dave Barach68b0fb02017-02-28 15:15:56 -0500258 _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
259
Florin Corasf6359c82017-06-19 12:26:09 -0400260 /* usual speculation, or the enqueue_x1 macro will barf */
261 to_next[0] = bi0;
262 to_next += 1;
263 n_left_to_next -= 1;
264
Dave Barach68b0fb02017-02-28 15:15:56 -0500265 b0 = vlib_get_buffer (vm, bi0);
266 b0->error = 0;
Florin Coras4eeeaaf2017-09-05 14:03:37 -0400267 b0->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
Dave Barach68b0fb02017-02-28 15:15:56 -0500268 b0->current_data = 0;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700269 b0->total_length_not_including_first_buffer = 0;
Dave Barach68b0fb02017-02-28 15:15:56 -0500270
Florin Corase87216f2017-08-17 16:59:22 -0700271 len_to_deq0 = clib_min (left_to_snd0, deq_per_first_buf);
Florin Corasf6d68ed2017-05-07 19:12:02 -0700272 data0 = vlib_buffer_make_headroom (b0, MAX_HDRS_LEN);
273 if (peek_data)
274 {
Florin Coras1f152cd2017-08-18 19:28:03 -0700275 n_bytes_read = svm_fifo_peek (s0->server_tx_fifo, tx_offset,
Florin Corasf6d68ed2017-05-07 19:12:02 -0700276 len_to_deq0, data0);
Florin Coras9d063042017-09-14 03:08:00 -0400277 if (n_bytes_read <= 0)
278 goto dequeue_fail;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700279 /* Keep track of progress locally, transport is also supposed to
280 * increment it independently when pushing the header */
Florin Coras1f152cd2017-08-18 19:28:03 -0700281 tx_offset += n_bytes_read;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700282 }
283 else
284 {
285 n_bytes_read = svm_fifo_dequeue_nowait (s0->server_tx_fifo,
286 len_to_deq0, data0);
Florin Coras9d063042017-09-14 03:08:00 -0400287 if (n_bytes_read <= 0)
288 goto dequeue_fail;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700289 }
290
Florin Corasf6d68ed2017-05-07 19:12:02 -0700291 b0->current_length = n_bytes_read;
292
293 left_to_snd0 -= n_bytes_read;
294 *n_tx_packets = *n_tx_packets + 1;
295
296 /*
297 * Fill in the remaining buffers in the chain, if any
298 */
Florin Corase87216f2017-08-17 16:59:22 -0700299 if (PREDICT_FALSE (n_bufs_per_seg > 1 && left_to_snd0))
Florin Corasb2215d62017-08-01 16:56:58 -0700300 {
Florin Corasb2215d62017-08-01 16:56:58 -0700301 left_for_seg = clib_min (snd_mss0 - n_bytes_read, left_to_snd0);
302 session_tx_fifo_chain_tail (smm, vm, thread_index,
303 s0->server_tx_fifo, b0, bi0,
304 n_bufs_per_seg, left_for_seg,
Florin Coras1f152cd2017-08-18 19:28:03 -0700305 &left_to_snd0, &n_bufs, &tx_offset,
Florin Corasb2215d62017-08-01 16:56:58 -0700306 deq_per_buf, peek_data);
307 }
Florin Corasf6d68ed2017-05-07 19:12:02 -0700308
309 /* Ask transport to push header after current_length and
310 * total_length_not_including_first_buffer are updated */
311 transport_vft->push_header (tc0, b0);
312
313 /* *INDENT-OFF* */
314 SESSION_EVT_DBG(SESSION_EVT_DEQ, s0, ({
Florin Corasec44e342017-10-16 20:47:56 -0700315 ed->data[0] = e0->event_type;
Florin Corasf6d68ed2017-05-07 19:12:02 -0700316 ed->data[1] = max_dequeue0;
317 ed->data[2] = len_to_deq0;
318 ed->data[3] = left_to_snd0;
319 }));
320 /* *INDENT-ON* */
321
Dave Barach68b0fb02017-02-28 15:15:56 -0500322 VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
Florin Coras4eeeaaf2017-09-05 14:03:37 -0400323
Dave Barach68b0fb02017-02-28 15:15:56 -0500324 if (PREDICT_FALSE (n_trace > 0))
325 {
326 session_queue_trace_t *t0;
327 vlib_trace_buffer (vm, node, next_index, b0,
328 1 /* follow_chain */ );
329 vlib_set_trace_count (vm, node, --n_trace);
330 t0 = vlib_add_trace (vm, node, b0, sizeof (*t0));
331 t0->session_index = s0->session_index;
332 t0->server_thread_index = s0->thread_index;
333 }
334
Dave Barach68b0fb02017-02-28 15:15:56 -0500335 vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
336 to_next, n_left_to_next,
337 bi0, next0);
338 }
339 vlib_put_next_frame (vm, node, next_index, n_left_to_next);
340 }
341
Florin Coras6792ec02017-03-13 03:49:51 -0700342 /* If we couldn't dequeue all bytes mark as partially read */
343 if (max_len_to_snd0 < max_dequeue0)
Dave Barach68b0fb02017-02-28 15:15:56 -0500344 {
Florin Coras6792ec02017-03-13 03:49:51 -0700345 /* If we don't already have new event */
346 if (svm_fifo_set_event (s0->server_tx_fifo))
347 {
Dave Barachacd2a6a2017-05-16 17:41:34 -0400348 vec_add1 (smm->pending_event_vector[thread_index], *e0);
Florin Coras6792ec02017-03-13 03:49:51 -0700349 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500350 }
351 return 0;
352
353dequeue_fail:
Florin Coras6792ec02017-03-13 03:49:51 -0700354 /*
355 * Can't read from fifo. If we don't already have an event, save as partially
356 * read, return buff to free list and return
357 */
358 clib_warning ("dequeue fail");
Dave Barach68b0fb02017-02-28 15:15:56 -0500359
Florin Coras6792ec02017-03-13 03:49:51 -0700360 if (svm_fifo_set_event (s0->server_tx_fifo))
361 {
Dave Barachacd2a6a2017-05-16 17:41:34 -0400362 vec_add1 (smm->pending_event_vector[thread_index], *e0);
Florin Coras6792ec02017-03-13 03:49:51 -0700363 }
364 vlib_put_next_frame (vm, node, next_index, n_left_to_next + 1);
Dave Barach68b0fb02017-02-28 15:15:56 -0500365 _vec_len (smm->tx_buffers[thread_index]) += 1;
366
Dave Barach68b0fb02017-02-28 15:15:56 -0500367 return 0;
368}
369
370int
Florin Corasd79b41e2017-03-04 05:37:52 -0800371session_tx_fifo_peek_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
372 session_manager_main_t * smm,
373 session_fifo_event_t * e0,
374 stream_session_t * s0, u32 thread_index,
375 int *n_tx_pkts)
Dave Barach68b0fb02017-02-28 15:15:56 -0500376{
Florin Corasd79b41e2017-03-04 05:37:52 -0800377 return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
378 n_tx_pkts, 1);
Dave Barach68b0fb02017-02-28 15:15:56 -0500379}
380
381int
Florin Corasd79b41e2017-03-04 05:37:52 -0800382session_tx_fifo_dequeue_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
383 session_manager_main_t * smm,
384 session_fifo_event_t * e0,
385 stream_session_t * s0, u32 thread_index,
386 int *n_tx_pkts)
Dave Barach68b0fb02017-02-28 15:15:56 -0500387{
Florin Corasd79b41e2017-03-04 05:37:52 -0800388 return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
389 n_tx_pkts, 0);
Dave Barach68b0fb02017-02-28 15:15:56 -0500390}
391
Dave Barach2c25a622017-06-26 11:35:07 -0400392always_inline stream_session_t *
393session_event_get_session (session_fifo_event_t * e, u8 thread_index)
Florin Corasa5464812017-04-19 13:00:05 -0700394{
Florin Coras3cbc04b2017-10-02 00:18:51 -0700395 return session_get_if_valid (e->fifo->master_session_index, thread_index);
Florin Corasa5464812017-04-19 13:00:05 -0700396}
397
Dave Barachacd2a6a2017-05-16 17:41:34 -0400398void
399dump_thread_0_event_queue (void)
400{
401 session_manager_main_t *smm = vnet_get_session_manager_main ();
402 vlib_main_t *vm = &vlib_global_main;
403 u32 my_thread_index = vm->thread_index;
404 session_fifo_event_t _e, *e = &_e;
405 stream_session_t *s0;
406 int i, index;
407 i8 *headp;
408
409 unix_shared_memory_queue_t *q;
410 q = smm->vpp_event_queues[my_thread_index];
411
412 index = q->head;
413
414 for (i = 0; i < q->cursize; i++)
415 {
416 headp = (i8 *) (&q->data[0] + q->elsize * index);
417 clib_memcpy (e, headp, q->elsize);
418
419 switch (e->event_type)
420 {
421 case FIFO_EVENT_APP_TX:
422 s0 = session_event_get_session (e, my_thread_index);
423 fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
424 break;
425
426 case FIFO_EVENT_DISCONNECT:
Florin Corascea194d2017-10-02 00:18:51 -0700427 s0 = session_get_from_handle (e->session_handle);
Dave Barachacd2a6a2017-05-16 17:41:34 -0400428 fformat (stdout, "[%04d] disconnect session %d\n", i,
429 s0->session_index);
430 break;
431
432 case FIFO_EVENT_BUILTIN_RX:
433 s0 = session_event_get_session (e, my_thread_index);
434 fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
435 break;
436
437 case FIFO_EVENT_RPC:
438 fformat (stdout, "[%04d] RPC call %llx with %llx\n",
439 i, (u64) (e->rpc_args.fp), (u64) (e->rpc_args.arg));
440 break;
441
442 default:
443 fformat (stdout, "[%04d] unhandled event type %d\n",
444 i, e->event_type);
445 break;
446 }
447
448 index++;
449
450 if (index == q->maxsize)
451 index = 0;
452 }
453}
454
Florin Coras6534b7a2017-07-18 05:38:03 -0400455static u8
456session_node_cmp_event (session_fifo_event_t * e, svm_fifo_t * f)
457{
458 stream_session_t *s;
459 switch (e->event_type)
460 {
461 case FIFO_EVENT_APP_RX:
462 case FIFO_EVENT_APP_TX:
463 case FIFO_EVENT_BUILTIN_RX:
464 if (e->fifo == f)
465 return 1;
466 break;
467 case FIFO_EVENT_DISCONNECT:
468 break;
469 case FIFO_EVENT_RPC:
Florin Corascea194d2017-10-02 00:18:51 -0700470 s = session_get_from_handle (e->session_handle);
Florin Coras6534b7a2017-07-18 05:38:03 -0400471 if (!s)
472 {
473 clib_warning ("session has event but doesn't exist!");
474 break;
475 }
476 if (s->server_rx_fifo == f || s->server_tx_fifo == f)
477 return 1;
478 break;
479 default:
480 break;
481 }
482 return 0;
483}
484
485u8
486session_node_lookup_fifo_event (svm_fifo_t * f, session_fifo_event_t * e)
487{
488 session_manager_main_t *smm = vnet_get_session_manager_main ();
489 unix_shared_memory_queue_t *q;
490 session_fifo_event_t *pending_event_vector, *evt;
491 int i, index, found = 0;
492 i8 *headp;
493 u8 thread_index;
494
495 ASSERT (e);
496 thread_index = f->master_thread_index;
497 /*
498 * Search evt queue
499 */
500 q = smm->vpp_event_queues[thread_index];
501 index = q->head;
502 for (i = 0; i < q->cursize; i++)
503 {
504 headp = (i8 *) (&q->data[0] + q->elsize * index);
505 clib_memcpy (e, headp, q->elsize);
506 found = session_node_cmp_event (e, f);
507 if (found)
508 break;
509 if (++index == q->maxsize)
510 index = 0;
511 }
512 /*
513 * Search pending events vector
514 */
515 pending_event_vector = smm->pending_event_vector[thread_index];
516 vec_foreach (evt, pending_event_vector)
517 {
518 found = session_node_cmp_event (evt, f);
519 if (found)
520 {
521 clib_memcpy (e, evt, sizeof (*evt));
522 break;
523 }
524 }
525 return found;
526}
527
Dave Barach68b0fb02017-02-28 15:15:56 -0500528static uword
529session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
530 vlib_frame_t * frame)
531{
532 session_manager_main_t *smm = vnet_get_session_manager_main ();
Florin Coras3cbc04b2017-10-02 00:18:51 -0700533 session_fifo_event_t *my_pending_event_vector, *pending_disconnects, *e;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400534 session_fifo_event_t *my_fifo_events;
Florin Corase04c2992017-03-01 08:17:34 -0800535 u32 n_to_dequeue, n_events;
Dave Barach68b0fb02017-02-28 15:15:56 -0500536 unix_shared_memory_queue_t *q;
Florin Coras6792ec02017-03-13 03:49:51 -0700537 application_t *app;
Dave Barach68b0fb02017-02-28 15:15:56 -0500538 int n_tx_packets = 0;
Damjan Marion586afd72017-04-05 19:18:20 +0200539 u32 my_thread_index = vm->thread_index;
Dave Barach68b0fb02017-02-28 15:15:56 -0500540 int i, rv;
Florin Coras3e350af2017-03-30 02:54:28 -0700541 f64 now = vlib_time_now (vm);
Dave Barache5f1d272017-05-10 13:34:04 -0400542 void (*fp) (void *);
Florin Coras3e350af2017-03-30 02:54:28 -0700543
544 SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, my_thread_index);
Dave Barach68b0fb02017-02-28 15:15:56 -0500545
546 /*
Florin Coras561af9b2017-12-09 10:19:43 -0800547 * Update transport time
Dave Barach68b0fb02017-02-28 15:15:56 -0500548 */
Florin Coras561af9b2017-12-09 10:19:43 -0800549 transport_update_time (now, my_thread_index);
Dave Barach68b0fb02017-02-28 15:15:56 -0500550
551 /*
552 * Get vpp queue events
553 */
554 q = smm->vpp_event_queues[my_thread_index];
555 if (PREDICT_FALSE (q == 0))
556 return 0;
557
Dave Barachacd2a6a2017-05-16 17:41:34 -0400558 my_fifo_events = smm->free_event_vector[my_thread_index];
559
Dave Barach68b0fb02017-02-28 15:15:56 -0500560 /* min number of events we can dequeue without blocking */
561 n_to_dequeue = q->cursize;
Dave Barachacd2a6a2017-05-16 17:41:34 -0400562 my_pending_event_vector = smm->pending_event_vector[my_thread_index];
Florin Coras3cbc04b2017-10-02 00:18:51 -0700563 pending_disconnects = smm->pending_disconnects[my_thread_index];
Dave Barach68b0fb02017-02-28 15:15:56 -0500564
Florin Coras3cbc04b2017-10-02 00:18:51 -0700565 if (!n_to_dequeue && !vec_len (my_pending_event_vector)
566 && !vec_len (pending_disconnects))
Florin Corase04c2992017-03-01 08:17:34 -0800567 return 0;
568
Florin Coras6792ec02017-03-13 03:49:51 -0700569 SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0);
570
Florin Corase04c2992017-03-01 08:17:34 -0800571 /*
572 * If we didn't manage to process previous events try going
Dave Barach68b0fb02017-02-28 15:15:56 -0500573 * over them again without dequeuing new ones.
Florin Corase04c2992017-03-01 08:17:34 -0800574 */
575 /* XXX: Block senders to sessions that can't keep up */
Dave Barachacd2a6a2017-05-16 17:41:34 -0400576 if (0 && vec_len (my_pending_event_vector) >= 100)
Florin Coras6792ec02017-03-13 03:49:51 -0700577 {
578 clib_warning ("too many fifo events unsolved");
579 goto skip_dequeue;
580 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500581
582 /* See you in the next life, don't be late */
583 if (pthread_mutex_trylock (&q->mutex))
584 return 0;
585
586 for (i = 0; i < n_to_dequeue; i++)
587 {
588 vec_add2 (my_fifo_events, e, 1);
589 unix_shared_memory_queue_sub_raw (q, (u8 *) e);
590 }
591
592 /* The other side of the connection is not polling */
593 if (q->cursize < (q->maxsize / 8))
594 (void) pthread_cond_broadcast (&q->condvar);
595 pthread_mutex_unlock (&q->mutex);
596
Dave Barachacd2a6a2017-05-16 17:41:34 -0400597 vec_append (my_fifo_events, my_pending_event_vector);
Florin Coras3cbc04b2017-10-02 00:18:51 -0700598 vec_append (my_fifo_events, smm->pending_disconnects[my_thread_index]);
Dave Barachacd2a6a2017-05-16 17:41:34 -0400599
600 _vec_len (my_pending_event_vector) = 0;
601 smm->pending_event_vector[my_thread_index] = my_pending_event_vector;
Florin Coras3cbc04b2017-10-02 00:18:51 -0700602 _vec_len (smm->pending_disconnects[my_thread_index]) = 0;
Dave Barach68b0fb02017-02-28 15:15:56 -0500603
604skip_dequeue:
Florin Corase04c2992017-03-01 08:17:34 -0800605 n_events = vec_len (my_fifo_events);
606 for (i = 0; i < n_events; i++)
Dave Barach68b0fb02017-02-28 15:15:56 -0500607 {
Florin Corasa5464812017-04-19 13:00:05 -0700608 stream_session_t *s0; /* $$$ prefetch 1 ahead maybe */
Dave Barach68b0fb02017-02-28 15:15:56 -0500609 session_fifo_event_t *e0;
610
611 e0 = &my_fifo_events[i];
Dave Barach68b0fb02017-02-28 15:15:56 -0500612
613 switch (e0->event_type)
614 {
Florin Corasa5464812017-04-19 13:00:05 -0700615 case FIFO_EVENT_APP_TX:
616 s0 = session_event_get_session (e0, my_thread_index);
617
Florin Coras9d063042017-09-14 03:08:00 -0400618 if (PREDICT_FALSE (!s0))
Florin Corasa5464812017-04-19 13:00:05 -0700619 {
620 clib_warning ("It's dead, Jim!");
621 continue;
622 }
Florin Corasb2215d62017-08-01 16:56:58 -0700623 /* Can retransmit for closed sessions but can't do anything if
624 * session is not ready or closed */
625 if (PREDICT_FALSE (s0->session_state < SESSION_STATE_READY))
Florin Coras93e65802017-11-29 00:07:11 -0500626 {
627 vec_add1 (smm->pending_event_vector[my_thread_index], *e0);
628 continue;
629 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500630 /* Spray packets in per session type frames, since they go to
631 * different nodes */
Florin Corase69f4952017-03-07 10:06:24 -0800632 rv = (smm->session_tx_fns[s0->session_type]) (vm, node, smm, e0, s0,
Dave Barach68b0fb02017-02-28 15:15:56 -0500633 my_thread_index,
634 &n_tx_packets);
Florin Coras6792ec02017-03-13 03:49:51 -0700635 /* Out of buffers */
Florin Coras93992a92017-05-24 18:03:56 -0700636 if (PREDICT_FALSE (rv < 0))
Dave Barachacd2a6a2017-05-16 17:41:34 -0400637 {
Florin Coras93992a92017-05-24 18:03:56 -0700638 vlib_node_increment_counter (vm, node->node_index,
639 SESSION_QUEUE_ERROR_NO_BUFFER, 1);
Dave Barachacd2a6a2017-05-16 17:41:34 -0400640 continue;
641 }
Dave Barach68b0fb02017-02-28 15:15:56 -0500642 break;
Florin Corasa5464812017-04-19 13:00:05 -0700643 case FIFO_EVENT_DISCONNECT:
Florin Coras3cbc04b2017-10-02 00:18:51 -0700644 /* Make sure disconnects run after the pending list is drained */
645 if (!e0->postponed)
646 {
647 e0->postponed = 1;
648 vec_add1 (smm->pending_disconnects[my_thread_index], *e0);
649 continue;
650 }
Florin Corascea194d2017-10-02 00:18:51 -0700651 s0 = session_get_from_handle (e0->session_handle);
Florin Coras6792ec02017-03-13 03:49:51 -0700652 stream_session_disconnect (s0);
653 break;
654 case FIFO_EVENT_BUILTIN_RX:
Florin Corasa5464812017-04-19 13:00:05 -0700655 s0 = session_event_get_session (e0, my_thread_index);
Florin Corasc87c91d2017-08-16 19:55:49 -0700656 if (PREDICT_FALSE (!s0))
657 continue;
Florin Coras6792ec02017-03-13 03:49:51 -0700658 svm_fifo_unset_event (s0->server_rx_fifo);
Florin Coras6792ec02017-03-13 03:49:51 -0700659 app = application_get (s0->app_index);
660 app->cb_fns.builtin_server_rx_callback (s0);
661 break;
Dave Barache5f1d272017-05-10 13:34:04 -0400662 case FIFO_EVENT_RPC:
663 fp = e0->rpc_args.fp;
664 (*fp) (e0->rpc_args.arg);
665 break;
666
Dave Barach68b0fb02017-02-28 15:15:56 -0500667 default:
668 clib_warning ("unhandled event type %d", e0->event_type);
669 }
670 }
671
Dave Barachacd2a6a2017-05-16 17:41:34 -0400672 _vec_len (my_fifo_events) = 0;
673 smm->free_event_vector[my_thread_index] = my_fifo_events;
Dave Barach68b0fb02017-02-28 15:15:56 -0500674
675 vlib_node_increment_counter (vm, session_queue_node.index,
676 SESSION_QUEUE_ERROR_TX, n_tx_packets);
677
Florin Coras6792ec02017-03-13 03:49:51 -0700678 SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 1);
679
Dave Barach68b0fb02017-02-28 15:15:56 -0500680 return n_tx_packets;
681}
682
683/* *INDENT-OFF* */
684VLIB_REGISTER_NODE (session_queue_node) =
685{
686 .function = session_queue_node_fn,
687 .name = "session-queue",
688 .format_trace = format_session_queue_trace,
689 .type = VLIB_NODE_TYPE_INPUT,
690 .n_errors = ARRAY_LEN (session_queue_error_strings),
691 .error_strings = session_queue_error_strings,
Florin Corase04c2992017-03-01 08:17:34 -0800692 .state = VLIB_NODE_STATE_DISABLED,
Dave Barach68b0fb02017-02-28 15:15:56 -0500693};
694/* *INDENT-ON* */
695
Dave Barach2a863912017-11-28 10:11:42 -0500696static clib_error_t *
697session_queue_exit (vlib_main_t * vm)
698{
699 if (vec_len (vlib_mains) < 2)
700 return 0;
701
702 /*
703 * Shut off (especially) worker-thread session nodes.
704 * Otherwise, vpp can crash as the main thread unmaps the
705 * API segment.
706 */
707 vlib_worker_thread_barrier_sync (vm);
708 session_node_enable_disable (0 /* is_enable */ );
709 vlib_worker_thread_barrier_release (vm);
710 return 0;
711}
712
713VLIB_MAIN_LOOP_EXIT_FUNCTION (session_queue_exit);
714
Dave Barach68b0fb02017-02-28 15:15:56 -0500715/*
716 * fd.io coding-style-patch-verification: ON
717 *
718 * Local Variables:
719 * eval: (c-set-style "gnu")
720 * End:
721 */