aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authorFlorin Coras <fcoras@cisco.com>2018-05-04 15:46:57 -0700
committerDave Barach <openvpp@barachs.net>2018-05-09 11:34:06 +0000
commit8e43d04ca4f4496aaefc4f5e2b6e1c0951624099 (patch)
tree3addc5766002d5224dde2c7fa4efe018480830e4 /src
parentee7f0bd9e7ce4106d3b9511b0efede4326bded51 (diff)
session: cleanup session tx function
- rework the function to declutter and avoid building more than one tx frame - add dual loop although benefits in my tests seem to be minimal - improve tcp/udp echo external apps. They have slightly better throughput than internal echo apps. - udp bugfixes Change-Id: Iea4a245b1b1bb407a7f403dedcce2664a49f774b Signed-off-by: Florin Coras <fcoras@cisco.com>
Diffstat (limited to 'src')
-rw-r--r--src/svm/svm_fifo.c6
-rw-r--r--src/tests/vnet/session/tcp_echo.c59
-rw-r--r--src/tests/vnet/session/udp_echo.c69
-rw-r--r--src/vnet/session-apps/echo_client.c58
-rw-r--r--src/vnet/session-apps/echo_client.h2
-rw-r--r--src/vnet/session/application_interface.h6
-rw-r--r--src/vnet/session/session.c1
-rw-r--r--src/vnet/session/session.h76
-rw-r--r--src/vnet/session/session_node.c631
9 files changed, 532 insertions, 376 deletions
diff --git a/src/svm/svm_fifo.c b/src/svm/svm_fifo.c
index 3552192a768..dbdb813a7df 100644
--- a/src/svm/svm_fifo.c
+++ b/src/svm/svm_fifo.c
@@ -622,14 +622,14 @@ void
svm_fifo_overwrite_head (svm_fifo_t * f, u8 * data, u32 len)
{
u32 first_chunk;
+ first_chunk = f->nitems - f->head;
ASSERT (len <= f->nitems);
- if (len < f->nitems - f->head)
+ if (len <= first_chunk)
clib_memcpy (&f->data[f->head], data, len);
else
{
- first_chunk = len - (f->nitems - f->head);
clib_memcpy (&f->data[f->head], data, first_chunk);
- clib_memcpy (f->data, data + first_chunk, len - first_chunk);
+ clib_memcpy (&f->data[0], data + first_chunk, len - first_chunk);
}
}
diff --git a/src/tests/vnet/session/tcp_echo.c b/src/tests/vnet/session/tcp_echo.c
index 0108b16ac7b..d3d4c0ad0a6 100644
--- a/src/tests/vnet/session/tcp_echo.c
+++ b/src/tests/vnet/session/tcp_echo.c
@@ -85,7 +85,7 @@ typedef struct
int i_am_master;
/* drop all packets */
- int drop_packets;
+ int no_return;
/* Our event queue */
svm_queue_t *our_event_queue;
@@ -117,6 +117,7 @@ typedef struct
u32 client_bytes_received;
u8 test_return_packets;
u64 bytes_to_send;
+ u32 fifo_size;
/** Flag that decides if socket, instead of svm, api is used to connect to
* vpp. If sock api is used, shm binary api is subsequently bootstrapped
@@ -250,7 +251,6 @@ application_send_attach (echo_main_t * em)
vl_api_application_tls_cert_add_t *cert_mp;
vl_api_application_tls_key_add_t *key_mp;
- u32 fifo_size = 4 << 20;
bmp = vl_msg_api_alloc (sizeof (*bmp));
memset (bmp, 0, sizeof (*bmp));
@@ -260,8 +260,8 @@ application_send_attach (echo_main_t * em)
bmp->options[APP_OPTIONS_FLAGS] =
APP_OPTIONS_FLAGS_ACCEPT_REDIRECT | APP_OPTIONS_FLAGS_ADD_SEGMENT;
bmp->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = 16;
- bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = fifo_size;
- bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = fifo_size;
+ bmp->options[APP_OPTIONS_RX_FIFO_SIZE] = em->fifo_size;
+ bmp->options[APP_OPTIONS_TX_FIFO_SIZE] = em->fifo_size;
bmp->options[APP_OPTIONS_ADD_SEGMENT_SIZE] = 128 << 20;
bmp->options[APP_OPTIONS_SEGMENT_SIZE] = 256 << 20;
vl_msg_api_send_shmem (em->vl_input_queue, (u8 *) & bmp);
@@ -757,11 +757,10 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
static void
send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
{
+ u32 bytes_to_snd, enq_space, min_chunk = 16 << 10;
u8 *test_data = em->connect_test_data;
u64 bytes_sent = 0;
int test_buf_offset = 0;
- u32 bytes_to_snd;
- u32 queue_max_chunk = 128 << 10, actual_write;
session_fifo_event_t evt;
int rv;
@@ -771,11 +770,12 @@ send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
while (bytes_to_snd > 0 && !em->time_to_stop)
{
- actual_write = (bytes_to_snd > queue_max_chunk) ?
- queue_max_chunk : bytes_to_snd;
- rv = svm_fifo_enqueue_nowait (tx_fifo, actual_write,
+ enq_space = svm_fifo_max_enqueue (tx_fifo);
+ if (enq_space < clib_min (bytes_to_snd, min_chunk))
+ continue;
+ rv = svm_fifo_enqueue_nowait (tx_fifo,
+ clib_min (bytes_to_snd, enq_space),
test_data + test_buf_offset);
-
if (rv > 0)
{
bytes_to_snd -= rv;
@@ -787,9 +787,8 @@ send_test_chunk (echo_main_t * em, svm_fifo_t * tx_fifo, int mypid, u32 bytes)
/* Fabricate TX event, send to vpp */
evt.fifo = tx_fifo;
evt.event_type = FIFO_EVENT_APP_TX;
-
- svm_queue_add (em->vpp_event_queue,
- (u8 *) & evt, 0 /* do wait for mutex */ );
+ svm_queue_add (em->vpp_event_queue, (u8 *) & evt,
+ 0 /* do wait for mutex */ );
}
}
}
@@ -824,7 +823,7 @@ client_send_data (echo_main_t * em)
if (leftover)
send_test_chunk (em, tx_fifo, mypid, leftover);
- if (!em->drop_packets)
+ if (!em->no_return)
{
f64 timeout = clib_time_now (&em->clib_time) + 10;
@@ -839,6 +838,7 @@ client_send_data (echo_main_t * em)
}
}
}
+
em->time_to_stop = 1;
}
@@ -911,7 +911,7 @@ client_run (echo_main_t * em)
}
/* Init test data */
- vec_validate (em->connect_test_data, 128 * 1024 - 1);
+ vec_validate (em->connect_test_data, 1024 * 1024 - 1);
for (i = 0; i < vec_len (em->connect_test_data); i++)
em->connect_test_data[i] = i & 0xff;
@@ -1130,7 +1130,7 @@ server_handle_fifo_event_rx (echo_main_t * em, session_fifo_event_t * e)
}
/* Reflect if a non-drop session */
- if (!em->drop_packets && n_read > 0)
+ if (!em->no_return && n_read > 0)
{
offset = 0;
do
@@ -1342,13 +1342,13 @@ tcp_echo_api_hookup (echo_main_t * em)
int
main (int argc, char **argv)
{
- int i_am_master = 1, drop_packets = 0, test_return_packets = 0;
+ int i_am_server = 1, test_return_packets = 0;
echo_main_t *em = &echo_main;
unformat_input_t _argv, *a = &_argv;
u8 *chroot_prefix;
u8 *heap, *uri = 0;
u8 *bind_uri = (u8 *) "tcp://0.0.0.0/1234";
- u8 *connect_uri = (u8 *) "tcp://6.0.1.2/1234";
+ u8 *connect_uri = (u8 *) "tcp://6.0.1.1/1234";
u64 bytes_to_send = 64 << 10, mbytes;
char *app_name;
u32 tmp;
@@ -1365,11 +1365,11 @@ main (int argc, char **argv)
vec_validate (em->rx_buf, 128 << 10);
em->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
-
em->my_pid = getpid ();
em->configured_segment_size = 1 << 20;
em->socket_name = 0;
em->use_sock_api = 1;
+ em->fifo_size = 64 << 10;
clib_time_init (&em->clib_time);
init_error_string_table (em);
@@ -1388,12 +1388,12 @@ main (int argc, char **argv)
em->configured_segment_size = tmp << 20;
else if (unformat (a, "segment-size %dG", &tmp))
em->configured_segment_size = tmp << 30;
- else if (unformat (a, "master"))
- i_am_master = 1;
- else if (unformat (a, "slave"))
- i_am_master = 0;
- else if (unformat (a, "drop"))
- drop_packets = 1;
+ else if (unformat (a, "server"))
+ i_am_server = 1;
+ else if (unformat (a, "client"))
+ i_am_server = 0;
+ else if (unformat (a, "no-return"))
+ em->no_return = 1;
else if (unformat (a, "test"))
test_return_packets = 1;
else if (unformat (a, "mbytes %lld", &mbytes))
@@ -1408,6 +1408,8 @@ main (int argc, char **argv)
;
else if (unformat (a, "use-svm-api"))
em->use_sock_api = 0;
+ else if (unformat (a, "fifo-size %d", &tmp))
+ em->fifo_size = tmp << 10;
else
{
fformat (stderr, "%s: usage [master|slave]\n");
@@ -1429,9 +1431,8 @@ main (int argc, char **argv)
em->connect_uri = format (0, "%s%c", connect_uri, 0);
}
- em->i_am_master = i_am_master;
+ em->i_am_master = i_am_server;
em->segment_main = &svm_fifo_segment_main;
- em->drop_packets = drop_packets;
em->test_return_packets = test_return_packets;
em->bytes_to_send = bytes_to_send;
em->time_to_stop = 0;
@@ -1439,7 +1440,7 @@ main (int argc, char **argv)
setup_signal_handlers ();
tcp_echo_api_hookup (em);
- app_name = i_am_master ? "tcp_echo_server" : "tcp_echo_client";
+ app_name = i_am_server ? "tcp_echo_server" : "tcp_echo_client";
if (connect_to_vpp (app_name) < 0)
{
svm_region_exit ();
@@ -1447,7 +1448,7 @@ main (int argc, char **argv)
exit (1);
}
- if (i_am_master == 0)
+ if (i_am_server == 0)
client_run (em);
else
server_run (em);
diff --git a/src/tests/vnet/session/udp_echo.c b/src/tests/vnet/session/udp_echo.c
index fe5461d2c59..54e00b181bd 100644
--- a/src/tests/vnet/session/udp_echo.c
+++ b/src/tests/vnet/session/udp_echo.c
@@ -125,6 +125,8 @@ typedef struct
uword *segments_table;
u8 do_echo;
+ u8 have_return;
+ u64 bytes_to_send;
} udp_echo_main_t;
#if CLIB_DEBUG > 0
@@ -147,7 +149,6 @@ static void
stats_signal (int signum)
{
udp_echo_main_t *um = &udp_echo_main;
-
um->time_to_print_stats = 1;
}
@@ -593,15 +594,20 @@ send_test_chunk (udp_echo_main_t * utm, app_session_t * s, u32 bytes)
u8 *test_data = utm->connect_test_data;
int test_buf_offset = 0;
u64 bytes_sent = 0;
- u32 bytes_to_snd;
+ u32 bytes_to_snd, enq_space, min_chunk;
int rv;
+ min_chunk = clib_min (65536, s->tx_fifo->nitems);
bytes_to_snd = (bytes == 0) ? vec_len (test_data) : bytes;
if (bytes_to_snd > vec_len (test_data))
bytes_to_snd = vec_len (test_data);
while (bytes_to_snd > 0 && !utm->time_to_stop)
{
+ enq_space = svm_fifo_max_enqueue (s->tx_fifo);
+ if (enq_space < clib_min (bytes_to_snd, min_chunk))
+ continue;
+
rv = app_send (s, test_data + test_buf_offset, bytes_to_snd, 0);
if (rv > 0)
{
@@ -621,12 +627,14 @@ recv_test_chunk (udp_echo_main_t * utm, app_session_t * s)
void
client_send_data (udp_echo_main_t * utm)
{
- u8 *test_data;
+ f64 start_time, end_time, delta;
app_session_t *session;
+ char *transfer_type;
u32 n_iterations;
+ u8 *test_data;
int i;
- vec_validate (utm->connect_test_data, 64 * 1024 - 1);
+ vec_validate (utm->connect_test_data, 1024 * 1024 - 1);
for (i = 0; i < vec_len (utm->connect_test_data); i++)
utm->connect_test_data[i] = i & 0xff;
@@ -635,22 +643,38 @@ client_send_data (udp_echo_main_t * utm)
ASSERT (vec_len (test_data) > 0);
vec_validate (utm->rx_buf, vec_len (test_data) - 1);
- n_iterations = NITER;
+ n_iterations = utm->bytes_to_send / vec_len (test_data);
+ if (!n_iterations)
+ n_iterations = 1;
+ start_time = clib_time_now (&utm->clib_time);
for (i = 0; i < n_iterations; i++)
{
send_test_chunk (utm, session, 0);
- recv_test_chunk (utm, session);
+ if (utm->have_return)
+ recv_test_chunk (utm, session);
if (utm->time_to_stop)
break;
}
- f64 timeout = clib_time_now (&utm->clib_time) + 5;
- while (clib_time_now (&utm->clib_time) < timeout)
+ if (utm->have_return)
{
- recv_test_chunk (utm, session);
+ f64 timeout = clib_time_now (&utm->clib_time) + 5;
+ while (clib_time_now (&utm->clib_time) < timeout)
+ recv_test_chunk (utm, session);
}
+ end_time = clib_time_now (&utm->clib_time);
+ delta = end_time - start_time;
+ transfer_type = utm->have_return ? "full-duplex" : "half-duplex";
+ clib_warning ("%lld bytes (%lld mbytes, %lld gbytes) in %.2f seconds",
+ utm->bytes_to_send, utm->bytes_to_send / (1ULL << 20),
+ utm->bytes_to_send / (1ULL << 30), delta);
+ clib_warning ("%.2f bytes/second %s", ((f64) utm->bytes_to_send) / (delta),
+ transfer_type);
+ clib_warning ("%.4f gbit/second %s",
+ (((f64) utm->bytes_to_send * 8.0) / delta / 1e9),
+ transfer_type);
}
static void
@@ -1205,7 +1229,7 @@ int
main (int argc, char **argv)
{
udp_echo_main_t *utm = &udp_echo_main;
- u8 *uri = (u8 *) "udp://0.0.0.0/1234";
+ u8 *uri = (u8 *) "udp://6.0.1.1/1234";
unformat_input_t _argv, *a = &_argv;
int i_am_server = 1;
app_session_t *session;
@@ -1217,27 +1241,27 @@ main (int argc, char **argv)
int i;
clib_mem_init (0, 256 << 20);
-
heap = clib_mem_get_per_cpu_heap ();
h = mheap_header (heap);
-
/* make the main heap thread-safe */
h->flags |= MHEAP_FLAG_THREAD_SAFE;
+ svm_fifo_segment_main_init (0x200000000ULL, 20);
vec_validate (utm->rx_buf, 8192);
-
utm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
utm->my_pid = getpid ();
utm->configured_segment_size = 1 << 20;
utm->segments_table = hash_create_vec (0, sizeof (u8), sizeof (u64));
-
+ utm->have_return = 1;
+ utm->bytes_to_send = 1024;
+ utm->fifo_size = 128 << 10;
+ utm->segment_main = &svm_fifo_segment_main;
+ utm->cut_through_session_index = ~0;
clib_time_init (&utm->clib_time);
+
init_error_string_table (utm);
- svm_fifo_segment_main_init (0x200000000ULL, 20);
unformat_init_command_line (a, argv);
- utm->fifo_size = 128 << 10;
-
while (unformat_check_input (a) != UNFORMAT_END_OF_INPUT)
{
if (unformat (a, "chroot prefix %s", &chroot_prefix))
@@ -1254,6 +1278,12 @@ main (int argc, char **argv)
i_am_server = 1;
else if (unformat (a, "client"))
i_am_server = 0;
+ else if (unformat (a, "no-return"))
+ utm->have_return = 0;
+ else if (unformat (a, "mbytes %d", &tmp))
+ utm->bytes_to_send = (u64) tmp << 20;
+ else if (unformat (a, "fifo-size %d", &tmp))
+ utm->fifo_size = tmp << 10;
else
{
fformat (stderr, "%s: usage [server|client]\n");
@@ -1261,9 +1291,7 @@ main (int argc, char **argv)
}
}
- utm->cut_through_session_index = ~0;
utm->i_am_server = i_am_server;
- utm->segment_main = &svm_fifo_segment_main;
setup_signal_handlers ();
tcp_echo_api_hookup (utm);
@@ -1290,7 +1318,7 @@ main (int argc, char **argv)
if (i_am_server == 0)
{
client_test (utm);
- exit (0);
+ goto done;
}
/* $$$$ hack preallocation */
@@ -1304,6 +1332,7 @@ main (int argc, char **argv)
udp_server_test (utm);
+done:
vl_client_disconnect_from_vlib ();
exit (0);
}
diff --git a/src/vnet/session-apps/echo_client.c b/src/vnet/session-apps/echo_client.c
index d47b0d97337..3c1904cb55e 100644
--- a/src/vnet/session-apps/echo_client.c
+++ b/src/vnet/session-apps/echo_client.c
@@ -56,11 +56,60 @@ send_data_chunk (echo_client_main_t * ecm, session_t * s)
s->bytes_to_send);
if (!ecm->is_dgram)
- rv = app_send_stream (&s->data, test_data + test_buf_offset,
- bytes_this_chunk, 0);
+ {
+ if (ecm->no_copy)
+ {
+ svm_fifo_t *f = s->data.tx_fifo;
+ rv = clib_min (svm_fifo_max_enqueue (f), bytes_this_chunk);
+ svm_fifo_enqueue_nocopy (f, rv);
+ if (svm_fifo_set_event (f))
+ {
+ session_fifo_event_t evt;
+ evt.fifo = f;
+ evt.event_type = FIFO_EVENT_APP_TX;
+ svm_queue_add (s->data.vpp_evt_q, (u8 *) & evt, 0);
+ }
+ }
+ else
+ rv = app_send_stream (&s->data, test_data + test_buf_offset,
+ bytes_this_chunk, 0);
+ }
else
- rv = app_send_dgram (&s->data, test_data + test_buf_offset,
- bytes_this_chunk, 0);
+ {
+ if (ecm->no_copy)
+ {
+ session_dgram_hdr_t hdr;
+ svm_fifo_t *f = s->data.tx_fifo;
+ app_session_transport_t *at = &s->data.transport;
+ u32 max_enqueue = svm_fifo_max_enqueue (f);
+
+ if (max_enqueue <= sizeof (session_dgram_hdr_t))
+ return;
+
+ max_enqueue -= sizeof (session_dgram_hdr_t);
+ rv = clib_min (max_enqueue, bytes_this_chunk);
+
+ hdr.data_length = rv;
+ hdr.data_offset = 0;
+ clib_memcpy (&hdr.rmt_ip, &at->rmt_ip, sizeof (ip46_address_t));
+ hdr.is_ip4 = at->is_ip4;
+ hdr.rmt_port = at->rmt_port;
+ clib_memcpy (&hdr.lcl_ip, &at->lcl_ip, sizeof (ip46_address_t));
+ hdr.lcl_port = at->lcl_port;
+ svm_fifo_enqueue_nowait (f, sizeof (hdr), (u8 *) & hdr);
+ svm_fifo_enqueue_nocopy (f, rv);
+ if (svm_fifo_set_event (f))
+ {
+ session_fifo_event_t evt;
+ evt.fifo = f;
+ evt.event_type = FIFO_EVENT_APP_TX;
+ svm_queue_add (s->data.vpp_evt_q, (u8 *) & evt, 0);
+ }
+ }
+ else
+ rv = app_send_dgram (&s->data, test_data + test_buf_offset,
+ bytes_this_chunk, 0);
+ }
/* If we managed to enqueue data... */
if (rv > 0)
@@ -591,6 +640,7 @@ echo_clients_command_fn (vlib_main_t * vm,
ecm->test_failed = 0;
ecm->vlib_main = vm;
ecm->tls_engine = TLS_ENGINE_OPENSSL;
+ ecm->no_copy = 0;
if (thread_main->n_vlib_mains > 1)
clib_spinlock_init (&ecm->sessions_lock);
diff --git a/src/vnet/session-apps/echo_client.h b/src/vnet/session-apps/echo_client.h
index 344e43865b5..8c07b10fbba 100644
--- a/src/vnet/session-apps/echo_client.h
+++ b/src/vnet/session-apps/echo_client.h
@@ -64,6 +64,8 @@ typedef struct
u32 private_segment_size; /**< size of private fifo segs */
u32 tls_engine; /**< TLS engine mbedtls/openssl */
u8 is_dgram;
+ u32 no_copy; /**< Don't memcpy data to tx fifo */
+
/*
* Test state variables
*/
diff --git a/src/vnet/session/application_interface.h b/src/vnet/session/application_interface.h
index 5fd218533dc..49c2862ac45 100644
--- a/src/vnet/session/application_interface.h
+++ b/src/vnet/session/application_interface.h
@@ -204,7 +204,7 @@ app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at,
int rv;
max_enqueue = svm_fifo_max_enqueue (f);
- if (svm_fifo_max_enqueue (f) <= sizeof (session_dgram_hdr_t))
+ if (max_enqueue <= sizeof (session_dgram_hdr_t))
return 0;
max_enqueue -= sizeof (session_dgram_hdr_t);
@@ -217,9 +217,6 @@ app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at,
clib_memcpy (&hdr.lcl_ip, &at->lcl_ip, sizeof (ip46_address_t));
hdr.lcl_port = at->lcl_port;
rv = svm_fifo_enqueue_nowait (f, sizeof (hdr), (u8 *) & hdr);
- if (rv <= 0)
- return 0;
-
ASSERT (rv == sizeof (hdr));
if ((rv = svm_fifo_enqueue_nowait (f, actual_write, data)) > 0)
@@ -231,6 +228,7 @@ app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at,
svm_queue_add (vpp_evt_q, (u8 *) & evt, noblock);
}
}
+ ASSERT (rv);
return rv;
}
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 568065ebc0d..a8a9c66ac81 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -1272,6 +1272,7 @@ session_manager_main_enable (vlib_main_t * vm)
vec_validate (smm->free_event_vector, num_threads - 1);
vec_validate (smm->vpp_event_queues, num_threads - 1);
vec_validate (smm->peekers_rw_locks, num_threads - 1);
+ vec_validate_aligned (smm->ctx, num_threads - 1, CLIB_CACHE_LINE_BYTES);
for (i = 0; i < TRANSPORT_N_PROTO; i++)
{
diff --git a/src/vnet/session/session.h b/src/vnet/session/session.h
index b54f46189f3..c5779b4ffd1 100644
--- a/src/vnet/session/session.h
+++ b/src/vnet/session/session.h
@@ -103,14 +103,58 @@ typedef CLIB_PACKED (struct {
}) session_fifo_event_t;
/* *INDENT-ON* */
+typedef struct session_dgram_pre_hdr_
+{
+ u32 data_length;
+ u32 data_offset;
+} session_dgram_pre_hdr_t;
+
+/* *INDENT-OFF* */
+typedef CLIB_PACKED (struct session_dgram_header_
+{
+ u32 data_length;
+ u32 data_offset;
+ ip46_address_t rmt_ip;
+ ip46_address_t lcl_ip;
+ u16 rmt_port;
+ u16 lcl_port;
+ u8 is_ip4;
+}) session_dgram_hdr_t;
+/* *INDENT-ON* */
+
+#define SESSION_CONN_ID_LEN 37
+#define SESSION_CONN_HDR_LEN 45
+
+STATIC_ASSERT (sizeof (session_dgram_hdr_t) == (SESSION_CONN_ID_LEN + 8),
+ "session conn id wrong length");
+
+typedef struct session_tx_context_
+{
+ CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
+ stream_session_t *s;
+ transport_proto_vft_t *transport_vft;
+ transport_connection_t *tc;
+ vlib_buffer_t *b;
+ u32 max_dequeue;
+ u32 snd_space;
+ u32 left_to_snd;
+ u32 tx_offset;
+ u32 max_len_to_snd;
+ u16 deq_per_first_buf;
+ u16 deq_per_buf;
+ u16 snd_mss;
+ u8 n_bufs_per_seg;
+ CLIB_CACHE_LINE_ALIGN_MARK (cacheline1);
+ session_dgram_hdr_t hdr;
+} session_tx_context_t;
+
/* Forward definition */
typedef struct _session_manager_main session_manager_main_t;
typedef int
(session_fifo_rx_fn) (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_main_t * smm,
session_fifo_event_t * e0, stream_session_t * s0,
- u32 thread_index, int *n_tx_pkts);
+ int *n_tx_pkts);
extern session_fifo_rx_fn session_tx_fifo_peek_and_snd;
extern session_fifo_rx_fn session_tx_fifo_dequeue_and_snd;
@@ -144,6 +188,9 @@ struct _session_manager_main
/** per-worker postponed disconnects */
session_fifo_event_t **pending_disconnects;
+ /** per-worker session context */
+ session_tx_context_t *ctx;
+
/** vpp fifo event queue */
svm_queue_t **vpp_event_queues;
@@ -204,31 +251,6 @@ struct _session_manager_main
};
-typedef struct session_dgram_pre_hdr_
-{
- u32 data_length;
- u32 data_offset;
-} session_dgram_pre_hdr_t;
-
-/* *INDENT-OFF* */
-typedef CLIB_PACKED (struct session_dgram_header_
-{
- u32 data_length;
- u32 data_offset;
- ip46_address_t rmt_ip;
- ip46_address_t lcl_ip;
- u16 rmt_port;
- u16 lcl_port;
- u8 is_ip4;
-}) session_dgram_hdr_t;
-/* *INDENT-ON* */
-
-#define SESSION_CONN_ID_LEN 37
-#define SESSION_CONN_HDR_LEN 45
-
-STATIC_ASSERT (sizeof (session_dgram_hdr_t) == (SESSION_CONN_ID_LEN + 8),
- "session conn id wrong length");
-
extern session_manager_main_t session_manager_main;
extern vlib_node_registration_t session_queue_node;
diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 14716965547..269e2fb591e 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -18,6 +18,7 @@
#include <vnet/vnet.h>
#include <vppinfra/elog.h>
#include <vnet/session/transport.h>
+#include <vnet/session/session.h>
#include <vnet/session/application.h>
#include <vnet/session/session_debug.h>
#include <svm/queue.h>
@@ -65,420 +66,473 @@ static char *session_queue_error_strings[] = {
};
always_inline void
-session_tx_fifo_chain_tail (session_manager_main_t * smm, vlib_main_t * vm,
- u8 thread_index, svm_fifo_t * fifo,
- vlib_buffer_t * b0, u32 bi0, u8 n_bufs_per_seg,
- u32 left_from_seg, u32 * left_to_snd0,
- u16 * n_bufs, u32 * tx_offset, u16 deq_per_buf,
- u8 peek_data, transport_tx_fn_type_t tx_type)
+session_tx_trace_buffer (vlib_main_t * vm, vlib_node_runtime_t * node,
+ u32 next_index, vlib_buffer_t * b,
+ stream_session_t * s, u32 * n_trace)
{
- vlib_buffer_t *chain_b0, *prev_b0;
- u32 chain_bi0, to_deq;
- u16 len_to_deq0, n_bytes_read;
- u8 *data0, j;
-
- b0->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
- b0->total_length_not_including_first_buffer = 0;
+ session_queue_trace_t *t;
+ vlib_trace_buffer (vm, node, next_index, b, 1 /* follow_chain */ );
+ vlib_set_trace_count (vm, node, --*n_trace);
+ t = vlib_add_trace (vm, node, b, sizeof (*t));
+ t->session_index = s->session_index;
+ t->server_thread_index = s->thread_index;
+}
- chain_bi0 = bi0;
- chain_b0 = b0;
+always_inline 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)
+{
+ session_manager_main_t *smm = &session_manager_main;
+ vlib_buffer_t *chain_b, *prev_b;
+ u32 chain_bi0, to_deq, left_from_seg;
+ u16 len_to_deq, n_bytes_read;
+ u8 *data, j;
+
+ b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
+ b->total_length_not_including_first_buffer = 0;
+
+ chain_b = b;
+ left_from_seg = clib_min (ctx->snd_mss - b->current_length,
+ ctx->left_to_snd);
to_deq = left_from_seg;
- for (j = 1; j < n_bufs_per_seg; j++)
+ for (j = 1; j < ctx->n_bufs_per_seg; j++)
{
- prev_b0 = chain_b0;
- len_to_deq0 = clib_min (to_deq, deq_per_buf);
+ prev_b = chain_b;
+ len_to_deq = clib_min (to_deq, ctx->deq_per_buf);
*n_bufs -= 1;
- chain_bi0 = smm->tx_buffers[thread_index][*n_bufs];
- _vec_len (smm->tx_buffers[thread_index]) = *n_bufs;
+ chain_bi0 = smm->tx_buffers[ctx->s->thread_index][*n_bufs];
+ _vec_len (smm->tx_buffers[ctx->s->thread_index]) = *n_bufs;
- chain_b0 = vlib_get_buffer (vm, chain_bi0);
- chain_b0->current_data = 0;
- data0 = vlib_buffer_get_current (chain_b0);
+ chain_b = vlib_get_buffer (vm, chain_bi0);
+ chain_b->current_data = 0;
+ data = vlib_buffer_get_current (chain_b);
if (peek_data)
{
- n_bytes_read = svm_fifo_peek (fifo, *tx_offset, len_to_deq0, data0);
- *tx_offset += n_bytes_read;
+ n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo,
+ ctx->tx_offset, len_to_deq, data);
+ ctx->tx_offset += n_bytes_read;
}
else
{
- if (tx_type == TRANSPORT_TX_DGRAM)
+ if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
{
- session_dgram_hdr_t *hdr;
+ svm_fifo_t *f = ctx->s->server_tx_fifo;
+ session_dgram_hdr_t *hdr = &ctx->hdr;
u16 deq_now;
- hdr = (session_dgram_hdr_t *) svm_fifo_head (fifo);
deq_now = clib_min (hdr->data_length - hdr->data_offset,
- len_to_deq0);
- n_bytes_read = svm_fifo_peek (fifo, hdr->data_offset, deq_now,
- data0);
+ len_to_deq);
+ n_bytes_read = svm_fifo_peek (f, hdr->data_offset, deq_now,
+ data);
ASSERT (n_bytes_read > 0);
hdr->data_offset += n_bytes_read;
if (hdr->data_offset == hdr->data_length)
- svm_fifo_dequeue_drop (fifo, hdr->data_length);
+ svm_fifo_dequeue_drop (f, hdr->data_length);
}
else
- n_bytes_read = svm_fifo_dequeue_nowait (fifo, len_to_deq0, data0);
+ n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
+ len_to_deq, data);
}
- ASSERT (n_bytes_read == len_to_deq0);
- chain_b0->current_length = n_bytes_read;
- b0->total_length_not_including_first_buffer += chain_b0->current_length;
+ ASSERT (n_bytes_read == len_to_deq);
+ chain_b->current_length = n_bytes_read;
+ b->total_length_not_including_first_buffer += chain_b->current_length;
/* update previous buffer */
- prev_b0->next_buffer = chain_bi0;
- prev_b0->flags |= VLIB_BUFFER_NEXT_PRESENT;
+ prev_b->next_buffer = chain_bi0;
+ prev_b->flags |= VLIB_BUFFER_NEXT_PRESENT;
/* update current buffer */
- chain_b0->next_buffer = 0;
+ chain_b->next_buffer = 0;
to_deq -= n_bytes_read;
if (to_deq == 0)
break;
}
ASSERT (to_deq == 0
- && b0->total_length_not_including_first_buffer == left_from_seg);
- *left_to_snd0 -= left_from_seg;
+ && b->total_length_not_including_first_buffer == left_from_seg);
+ ctx->left_to_snd -= left_from_seg;
}
always_inline int
-session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
+session_output_try_get_buffers (vlib_main_t * vm,
session_manager_main_t * smm,
- session_fifo_event_t * e0,
- stream_session_t * s0, u32 thread_index,
- int *n_tx_packets, u8 peek_data)
+ u32 thread_index, u16 * n_bufs, u32 wanted)
{
- u32 n_trace = vlib_get_trace_count (vm, node);
- u32 left_to_snd0, max_len_to_snd0, len_to_deq0, snd_space0;
- u32 n_bufs_per_evt, n_frames_per_evt, n_bufs_per_frame;
- transport_connection_t *tc0;
- transport_proto_vft_t *transport_vft;
- transport_proto_t tp;
- u32 next_index, next0, *to_next, n_left_to_next, bi0;
- vlib_buffer_t *b0;
- u32 tx_offset = 0, max_dequeue0, n_bytes_per_seg, left_for_seg;
- u16 snd_mss0, n_bufs_per_seg, n_bufs;
+ u32 bufs_alloc = 0, bufs_now;
+ vec_validate_aligned (smm->tx_buffers[thread_index], *n_bufs + wanted - 1,
+ CLIB_CACHE_LINE_BYTES);
+ do
+ {
+ bufs_now =
+ vlib_buffer_alloc (vm,
+ &smm->tx_buffers[thread_index][*n_bufs +
+ bufs_alloc],
+ wanted - bufs_alloc);
+ bufs_alloc += bufs_now;
+ }
+ while (bufs_now > 0 && ((bufs_alloc + *n_bufs < wanted)));
+
+ *n_bufs += bufs_alloc;
+ _vec_len (smm->tx_buffers[thread_index]) = *n_bufs;
+ return bufs_alloc;
+}
+
+always_inline void
+session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
+ vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
+{
+ u32 len_to_deq;
u8 *data0;
- int i, n_bytes_read;
- u32 n_bytes_per_buf, deq_per_buf, deq_per_first_buf;
- u32 bufs_alloc, bufs_now;
- session_dgram_hdr_t hdr;
-
- next_index = next0 = smm->session_type_to_next[s0->session_type];
- tp = session_get_transport_proto (s0);
- transport_vft = transport_protocol_get_vft (tp);
+ int n_bytes_read;
+
+ /*
+ * Start with the first buffer in chain
+ */
+ b->error = 0;
+ b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
+ b->current_data = 0;
+ b->total_length_not_including_first_buffer = 0;
+
+ data0 = vlib_buffer_make_headroom (b, MAX_HDRS_LEN);
+ len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
+
if (peek_data)
{
- if (PREDICT_FALSE (s0->session_state < SESSION_STATE_READY))
- {
- /* Can retransmit for closed sessions but can't send new data if
- * session is not ready or closed */
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
- return 0;
- }
- tc0 =
- transport_vft->get_connection (s0->connection_index, thread_index);
+ n_bytes_read = svm_fifo_peek (ctx->s->server_tx_fifo, ctx->tx_offset,
+ len_to_deq, data0);
+ ASSERT (n_bytes_read > 0);
+ /* Keep track of progress locally, transport is also supposed to
+ * increment it independently when pushing the header */
+ ctx->tx_offset += n_bytes_read;
}
else
{
- if (s0->session_state == SESSION_STATE_LISTENING)
- tc0 = transport_vft->get_listener (s0->connection_index);
+ if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
+ {
+ session_dgram_hdr_t *hdr = &ctx->hdr;
+ svm_fifo_t *f = ctx->s->server_tx_fifo;
+ u16 deq_now;
+ u32 offset;
+
+ ASSERT (hdr->data_length > hdr->data_offset);
+ deq_now = clib_min (hdr->data_length - hdr->data_offset,
+ len_to_deq);
+ offset = hdr->data_offset + SESSION_CONN_HDR_LEN;
+ n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
+ ASSERT (n_bytes_read > 0);
+
+ if (ctx->s->session_state == SESSION_STATE_LISTENING)
+ {
+ ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
+ ctx->tc->rmt_port = hdr->rmt_port;
+ }
+ hdr->data_offset += n_bytes_read;
+ if (hdr->data_offset == hdr->data_length)
+ {
+ offset = hdr->data_length + SESSION_CONN_HDR_LEN;
+ svm_fifo_dequeue_drop (f, offset);
+ }
+ }
else
{
- if (PREDICT_FALSE (s0->session_state == SESSION_STATE_CLOSED))
- return 0;
- tc0 = transport_vft->get_connection (s0->connection_index,
- thread_index);
+ n_bytes_read = svm_fifo_dequeue_nowait (ctx->s->server_tx_fifo,
+ len_to_deq, data0);
+ ASSERT (n_bytes_read > 0);
}
}
+ b->current_length = n_bytes_read;
+ ctx->left_to_snd -= n_bytes_read;
- /* Make sure we have space to send and there's something to dequeue */
- snd_mss0 = transport_vft->send_mss (tc0);
- snd_space0 = transport_vft->send_space (tc0);
+ /*
+ * Fill in the remaining buffers in the chain, if any
+ */
+ if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
+ session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
+
+ /* *INDENT-OFF* */
+ SESSION_EVT_DBG(SESSION_EVT_DEQ, s, ({
+ ed->data[0] = e->event_type;
+ ed->data[1] = max_dequeue;
+ ed->data[2] = len_to_deq;
+ ed->data[3] = left_to_snd;
+ }));
+ /* *INDENT-ON* */
+}
- /* Can't make any progress */
- if (snd_space0 == 0 || snd_mss0 == 0)
+always_inline u8
+session_tx_not_ready (stream_session_t * s, u8 peek_data)
+{
+ if (peek_data)
{
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
- return 0;
+ /* Can retransmit for closed sessions but can't send new data if
+ * session is not ready or closed */
+ if (s->session_state < SESSION_STATE_READY)
+ return 1;
}
+ return 0;
+}
- /* Allow enqueuing of a new event */
- svm_fifo_unset_event (s0->server_tx_fifo);
+always_inline transport_connection_t *
+session_tx_get_transport (session_tx_context_t * ctx, u8 peek_data)
+{
+ if (peek_data)
+ {
+ return ctx->transport_vft->get_connection (ctx->s->connection_index,
+ ctx->s->thread_index);
+ }
+ else
+ {
+ if (ctx->s->session_state == SESSION_STATE_LISTENING)
+ return ctx->transport_vft->get_listener (ctx->s->connection_index);
+ else
+ {
+ return ctx->transport_vft->get_connection (ctx->s->connection_index,
+ ctx->s->thread_index);
+ }
+ }
+}
- /* Check how much we can pull. */
- max_dequeue0 = svm_fifo_max_dequeue (s0->server_tx_fifo);
+always_inline void
+session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
+ u8 peek_data)
+{
+ u32 n_bytes_per_buf, n_bytes_per_seg;
+ ctx->max_dequeue = svm_fifo_max_dequeue (ctx->s->server_tx_fifo);
if (peek_data)
{
/* Offset in rx fifo from where to peek data */
- tx_offset = transport_vft->tx_fifo_offset (tc0);
- if (PREDICT_FALSE (tx_offset >= max_dequeue0))
- return 0;
- max_dequeue0 -= tx_offset;
+ ctx->tx_offset = ctx->transport_vft->tx_fifo_offset (ctx->tc);
+ if (PREDICT_FALSE (ctx->tx_offset >= ctx->max_dequeue))
+ {
+ ctx->max_len_to_snd = 0;
+ return;
+ }
+ ctx->max_dequeue -= ctx->tx_offset;
}
else
{
- if (transport_vft->tx_type == TRANSPORT_TX_DGRAM)
+ if (ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
{
- if (max_dequeue0 < sizeof (hdr))
- return 0;
- svm_fifo_peek (s0->server_tx_fifo, 0, sizeof (hdr), (u8 *) & hdr);
- ASSERT (hdr.data_length > hdr.data_offset);
- max_dequeue0 = hdr.data_length - hdr.data_offset;
+ if (ctx->max_dequeue <= sizeof (ctx->hdr))
+ {
+ ctx->max_len_to_snd = 0;
+ return;
+ }
+ svm_fifo_peek (ctx->s->server_tx_fifo, 0, sizeof (ctx->hdr),
+ (u8 *) & ctx->hdr);
+ ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
+ ctx->max_dequeue = ctx->hdr.data_length - ctx->hdr.data_offset;
}
}
- ASSERT (max_dequeue0 > 0);
+ ASSERT (ctx->max_dequeue > 0);
/* Ensure we're not writing more than transport window allows */
- if (max_dequeue0 < snd_space0)
+ if (ctx->max_dequeue < ctx->snd_space)
{
/* Constrained by tx queue. Try to send only fully formed segments */
- max_len_to_snd0 = (max_dequeue0 > snd_mss0) ?
- max_dequeue0 - max_dequeue0 % snd_mss0 : max_dequeue0;
+ ctx->max_len_to_snd =
+ (ctx->max_dequeue > ctx->snd_mss) ?
+ ctx->max_dequeue - ctx->max_dequeue % ctx->snd_mss : ctx->max_dequeue;
/* TODO Nagle ? */
}
else
{
/* Expectation is that snd_space0 is already a multiple of snd_mss */
- max_len_to_snd0 = snd_space0;
+ ctx->max_len_to_snd = ctx->snd_space;
}
- n_bytes_per_buf = vlib_buffer_free_list_buffer_size
- (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
+ n_bytes_per_buf = vlib_buffer_free_list_buffer_size (vm,
+ VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
ASSERT (n_bytes_per_buf > MAX_HDRS_LEN);
- n_bytes_per_seg = MAX_HDRS_LEN + snd_mss0;
- n_bufs_per_seg = ceil ((double) n_bytes_per_seg / n_bytes_per_buf);
- n_bufs_per_evt = ceil ((double) max_len_to_snd0 / n_bytes_per_seg);
- n_frames_per_evt = ceil ((double) n_bufs_per_evt / VLIB_FRAME_SIZE);
- n_bufs_per_frame = n_bufs_per_seg * VLIB_FRAME_SIZE;
+ n_bytes_per_seg = MAX_HDRS_LEN + ctx->snd_mss;
+ ctx->n_bufs_per_seg = ceil ((double) n_bytes_per_seg / n_bytes_per_buf);
+ ctx->deq_per_buf = clib_min (ctx->snd_mss, n_bytes_per_buf);
+ ctx->deq_per_first_buf = clib_min (ctx->snd_mss,
+ n_bytes_per_buf - MAX_HDRS_LEN);
+}
+
+always_inline int
+session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
+ session_fifo_event_t * e,
+ stream_session_t * s, int *n_tx_packets,
+ u8 peek_data)
+{
+ u32 next_index, next0, next1, next2, next3, *to_next, n_left_to_next;
+ u32 n_trace = vlib_get_trace_count (vm, node), n_packets = 0, pbi;
+ u32 n_bufs_per_frame, thread_index = s->thread_index;
+ session_manager_main_t *smm = &session_manager_main;
+ session_tx_context_t *ctx = &smm->ctx[thread_index];
+ transport_proto_t tp;
+ vlib_buffer_t *pb;
+ u16 n_bufs;
- deq_per_buf = clib_min (snd_mss0, n_bytes_per_buf);
- deq_per_first_buf = clib_min (snd_mss0, n_bytes_per_buf - MAX_HDRS_LEN);
+ if (PREDICT_FALSE (session_tx_not_ready (s, peek_data)))
+ {
+ vec_add1 (smm->pending_event_vector[thread_index], *e);
+ return 0;
+ }
+
+ next_index = smm->session_type_to_next[s->session_type];
+ next0 = next1 = next2 = next3 = next_index;
+
+ tp = session_get_transport_proto (s);
+ ctx->s = s;
+ ctx->transport_vft = transport_protocol_get_vft (tp);
+ ctx->tc = session_tx_get_transport (ctx, peek_data);
+ ctx->snd_mss = ctx->transport_vft->send_mss (ctx->tc);
+ ctx->snd_space = ctx->transport_vft->send_space (ctx->tc);
+ if (ctx->snd_space == 0 || ctx->snd_mss == 0)
+ {
+ vec_add1 (smm->pending_event_vector[thread_index], *e);
+ return 0;
+ }
+
+ /* Allow enqueuing of a new event */
+ svm_fifo_unset_event (s->server_tx_fifo);
+
+ /* Check how much we can pull. */
+ session_tx_set_dequeue_params (vm, ctx, peek_data);
+ if (PREDICT_FALSE (!ctx->max_len_to_snd))
+ return 0;
n_bufs = vec_len (smm->tx_buffers[thread_index]);
- left_to_snd0 = max_len_to_snd0;
- for (i = 0; i < n_frames_per_evt; i++)
+ ctx->left_to_snd = ctx->max_len_to_snd;
+
+ /*
+ * Make sure we have at least one full frame of buffers ready
+ */
+ n_bufs_per_frame = ctx->n_bufs_per_seg * VLIB_FRAME_SIZE;
+ if (n_bufs < n_bufs_per_frame)
{
- /* Make sure we have at least one full frame of buffers ready */
+ session_output_try_get_buffers (vm, smm, thread_index, &n_bufs,
+ n_bufs_per_frame);
if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
{
- vec_validate (smm->tx_buffers[thread_index],
- n_bufs + n_bufs_per_frame - 1);
- bufs_alloc = 0;
- do
- {
- bufs_now =
- vlib_buffer_alloc (vm,
- &smm->tx_buffers[thread_index][n_bufs +
- bufs_alloc],
- n_bufs_per_frame - bufs_alloc);
- bufs_alloc += bufs_now;
- }
- while (bufs_now > 0 && ((bufs_alloc + n_bufs < n_bufs_per_frame)));
-
- n_bufs += bufs_alloc;
- _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
-
- if (PREDICT_FALSE (n_bufs < n_bufs_per_frame))
- {
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
- return -1;
- }
- ASSERT (n_bufs >= n_bufs_per_frame);
+ vec_add1 (smm->pending_event_vector[thread_index], *e);
+ return -1;
}
+ }
- vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
- while (left_to_snd0 && n_left_to_next)
+ /*
+ * Write until we fill up a frame
+ */
+ vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
+ while (ctx->left_to_snd && n_left_to_next)
+ {
+ while (ctx->left_to_snd > 3 * ctx->snd_mss && n_left_to_next >= 4)
{
- /*
- * Handle first buffer in chain separately
- */
+ vlib_buffer_t *b0, *b1;
+ u32 bi0, bi1;
- len_to_deq0 = clib_min (left_to_snd0, deq_per_first_buf);
- if (left_to_snd0 > len_to_deq0 && n_left_to_next > 1)
- {
- vlib_buffer_t *pb;
- u32 pbi = smm->tx_buffers[thread_index][n_bufs - 2];
- pb = vlib_get_buffer (vm, pbi);
- vlib_prefetch_buffer_header (pb, LOAD);
- }
+ pbi = smm->tx_buffers[thread_index][n_bufs - 3];
+ pb = vlib_get_buffer (vm, pbi);
+ vlib_prefetch_buffer_header (pb, STORE);
+ pbi = smm->tx_buffers[thread_index][n_bufs - 4];
+ pb = vlib_get_buffer (vm, pbi);
+ vlib_prefetch_buffer_header (pb, STORE);
- /* Get free buffer */
- ASSERT (n_bufs >= 1);
- bi0 = smm->tx_buffers[thread_index][--n_bufs];
- _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
-
- /* usual speculation, or the enqueue_x1 macro will barf */
- to_next[0] = bi0;
- to_next += 1;
- n_left_to_next -= 1;
+ to_next[0] = bi0 = smm->tx_buffers[thread_index][--n_bufs];
+ to_next[1] = bi1 = smm->tx_buffers[thread_index][--n_bufs];
b0 = vlib_get_buffer (vm, bi0);
- b0->error = 0;
- b0->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
- b0->current_data = 0;
- b0->total_length_not_including_first_buffer = 0;
+ b1 = vlib_get_buffer (vm, bi1);
- data0 = vlib_buffer_make_headroom (b0, MAX_HDRS_LEN);
- if (peek_data)
- {
- n_bytes_read = svm_fifo_peek (s0->server_tx_fifo, tx_offset,
- len_to_deq0, data0);
- if (n_bytes_read <= 0)
- goto dequeue_fail;
- /* Keep track of progress locally, transport is also supposed to
- * increment it independently when pushing the header */
- tx_offset += n_bytes_read;
- }
- else
- {
- if (transport_vft->tx_type == TRANSPORT_TX_DGRAM)
- {
- svm_fifo_t *f = s0->server_tx_fifo;
- u16 deq_now;
- u32 offset;
-
- ASSERT (hdr.data_length > hdr.data_offset);
- deq_now = clib_min (hdr.data_length - hdr.data_offset,
- len_to_deq0);
- offset = hdr.data_offset + SESSION_CONN_HDR_LEN;
- n_bytes_read = svm_fifo_peek (f, offset, deq_now, data0);
- if (PREDICT_FALSE (n_bytes_read <= 0))
- goto dequeue_fail;
-
- if (s0->session_state == SESSION_STATE_LISTENING)
- {
- ip_copy (&tc0->rmt_ip, &hdr.rmt_ip, tc0->is_ip4);
- tc0->rmt_port = hdr.rmt_port;
- }
- hdr.data_offset += n_bytes_read;
- if (hdr.data_offset == hdr.data_length)
- {
- offset = hdr.data_length + SESSION_CONN_HDR_LEN;
- svm_fifo_dequeue_drop (f, offset);
- }
- }
- else
- {
- n_bytes_read = svm_fifo_dequeue_nowait (s0->server_tx_fifo,
- len_to_deq0, data0);
- if (n_bytes_read <= 0)
- goto dequeue_fail;
- }
- }
+ session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
+ session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
- b0->current_length = n_bytes_read;
- left_to_snd0 -= n_bytes_read;
- *n_tx_packets = *n_tx_packets + 1;
+ ctx->transport_vft->push_header (ctx->tc, b0);
+ ctx->transport_vft->push_header (ctx->tc, b1);
- /*
- * Fill in the remaining buffers in the chain, if any
- */
- if (PREDICT_FALSE (n_bufs_per_seg > 1 && left_to_snd0))
+ to_next += 2;
+ n_left_to_next -= 2;
+ n_packets += 2;
+
+ VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
+ VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b1);
+
+ if (PREDICT_FALSE (n_trace > 0))
{
- left_for_seg = clib_min (snd_mss0 - n_bytes_read, left_to_snd0);
- session_tx_fifo_chain_tail (smm, vm, thread_index,
- s0->server_tx_fifo, b0, bi0,
- n_bufs_per_seg, left_for_seg,
- &left_to_snd0, &n_bufs, &tx_offset,
- deq_per_buf, peek_data,
- transport_vft->tx_type);
+ session_tx_trace_buffer (vm, node, next_index, b0, s, &n_trace);
+ if (n_trace)
+ session_tx_trace_buffer (vm, node, next_index, b1, s,
+ &n_trace);
}
+ vlib_validate_buffer_enqueue_x2 (vm, node, next_index, to_next,
+ n_left_to_next, bi0, bi1,
+ next0, next1);
+ }
+ while (ctx->left_to_snd && n_left_to_next)
+ {
+ vlib_buffer_t *b0;
+ u32 bi0;
+
+ ASSERT (n_bufs >= 1);
+ to_next[0] = bi0 = smm->tx_buffers[thread_index][--n_bufs];
+ b0 = vlib_get_buffer (vm, bi0);
+ session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
+
/* Ask transport to push header after current_length and
* total_length_not_including_first_buffer are updated */
- transport_vft->push_header (tc0, b0);
+ ctx->transport_vft->push_header (ctx->tc, b0);
- /* *INDENT-OFF* */
- SESSION_EVT_DBG(SESSION_EVT_DEQ, s0, ({
- ed->data[0] = e0->event_type;
- ed->data[1] = max_dequeue0;
- ed->data[2] = len_to_deq0;
- ed->data[3] = left_to_snd0;
- }));
- /* *INDENT-ON* */
+ to_next += 1;
+ n_left_to_next -= 1;
+ n_packets += 1;
VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b0);
-
if (PREDICT_FALSE (n_trace > 0))
- {
- session_queue_trace_t *t0;
- vlib_trace_buffer (vm, node, next_index, b0,
- 1 /* follow_chain */ );
- vlib_set_trace_count (vm, node, --n_trace);
- t0 = vlib_add_trace (vm, node, b0, sizeof (*t0));
- t0->session_index = s0->session_index;
- t0->server_thread_index = s0->thread_index;
- }
+ session_tx_trace_buffer (vm, node, next_index, b0, s, &n_trace);
- vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
- to_next, n_left_to_next,
- bi0, next0);
+ vlib_validate_buffer_enqueue_x1 (vm, node, next_index, to_next,
+ n_left_to_next, bi0, next0);
}
- vlib_put_next_frame (vm, node, next_index, n_left_to_next);
}
+ _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
+ *n_tx_packets += n_packets;
+ vlib_put_next_frame (vm, node, next_index, n_left_to_next);
/* If we couldn't dequeue all bytes mark as partially read */
- if (max_len_to_snd0 < max_dequeue0)
- if (svm_fifo_set_event (s0->server_tx_fifo))
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
+ if (ctx->max_len_to_snd < ctx->max_dequeue)
+ if (svm_fifo_set_event (s->server_tx_fifo))
+ vec_add1 (smm->pending_event_vector[thread_index], *e);
- if (!peek_data && transport_vft->tx_type == TRANSPORT_TX_DGRAM)
+ if (!peek_data && ctx->transport_vft->tx_type == TRANSPORT_TX_DGRAM)
{
/* Fix dgram pre header */
- if (max_len_to_snd0 < max_dequeue0)
- svm_fifo_overwrite_head (s0->server_tx_fifo, (u8 *) & hdr,
+ if (ctx->max_len_to_snd < ctx->max_dequeue)
+ svm_fifo_overwrite_head (s->server_tx_fifo, (u8 *) & ctx->hdr,
sizeof (session_dgram_pre_hdr_t));
/* More data needs to be read */
- else if (svm_fifo_max_dequeue (s0->server_tx_fifo) > 0)
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
+ else if (svm_fifo_max_dequeue (s->server_tx_fifo) > 0)
+ if (svm_fifo_set_event (s->server_tx_fifo))
+ vec_add1 (smm->pending_event_vector[thread_index], *e);
}
return 0;
-
-dequeue_fail:
- /*
- * Can't read from fifo. If we don't already have an event, save as partially
- * read, return buff to free list and return
- */
- clib_warning ("dequeue fail");
- if (svm_fifo_set_event (s0->server_tx_fifo))
- {
- vec_add1 (smm->pending_event_vector[thread_index], *e0);
- }
- vlib_put_next_frame (vm, node, next_index, n_left_to_next + 1);
- _vec_len (smm->tx_buffers[thread_index]) += 1;
-
- return 0;
}
int
session_tx_fifo_peek_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_main_t * smm,
session_fifo_event_t * e0,
- stream_session_t * s0, u32 thread_index,
- int *n_tx_pkts)
+ stream_session_t * s0, int *n_tx_pkts)
{
- return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
- n_tx_pkts, 1);
+ return session_tx_fifo_read_and_snd_i (vm, node, e0, s0, n_tx_pkts, 1);
}
int
session_tx_fifo_dequeue_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_main_t * smm,
session_fifo_event_t * e0,
- stream_session_t * s0, u32 thread_index,
- int *n_tx_pkts)
+ stream_session_t * s0, int *n_tx_pkts)
{
- return session_tx_fifo_read_and_snd_i (vm, node, smm, e0, s0, thread_index,
- n_tx_pkts, 0);
+ return session_tx_fifo_read_and_snd_i (vm, node, e0, s0, n_tx_pkts, 0);
}
int
session_tx_fifo_dequeue_internal (vlib_main_t * vm,
vlib_node_runtime_t * node,
- session_manager_main_t * smm,
session_fifo_event_t * e0,
- stream_session_t * s0, u32 thread_index,
- int *n_tx_pkts)
+ stream_session_t * s0, int *n_tx_pkts)
{
application_t *app;
app = application_get (s0->opaque);
@@ -719,8 +773,7 @@ skip_dequeue:
}
/* Spray packets in per session type frames, since they go to
* different nodes */
- rv = (smm->session_tx_fns[s0->session_type]) (vm, node, smm, e0, s0,
- my_thread_index,
+ rv = (smm->session_tx_fns[s0->session_type]) (vm, node, e0, s0,
&n_tx_packets);
/* Out of buffers */
if (PREDICT_FALSE (rv < 0))
@@ -732,13 +785,13 @@ skip_dequeue:
break;
case FIFO_EVENT_DISCONNECT:
/* Make sure disconnects run after the pending list is drained */
- if (!e0->postponed)
+ s0 = session_get_from_handle (e0->session_handle);
+ if (!e0->postponed || svm_fifo_max_dequeue (s0->server_tx_fifo))
{
e0->postponed = 1;
vec_add1 (smm->pending_disconnects[my_thread_index], *e0);
continue;
}
- s0 = session_get_from_handle (e0->session_handle);
stream_session_disconnect_transport (s0);
break;
case FIFO_EVENT_BUILTIN_RX: