- Data structure preallocation.
- Input state machine fixes for mid-stream 3-way handshake retries.
- Batch connections in the builtin_client
- Multiple private fifo segment support
- Fix elog simultaneous event type registration
- Fix sacks when segment hole is added after highest sacked
- Add "accepting" session state for sessions pending accept
- Add ssvm non-recursive locking
- Estimate RTT for syn-ack
- Don't init fifo pointers. We're using relative offsets for ooo
segments
- CLI to dump individual session
Change-Id: Ie0598563fd246537bafba4feed7985478ea1d415
Signed-off-by: Dave Barach <dbarach@cisco.com>
Signed-off-by: Florin Coras <fcoras@cisco.com>
h->tag = tag;
}
+always_inline void
+ssvm_lock_non_recursive (ssvm_shared_header_t * h, u32 tag)
+{
+ while (__sync_lock_test_and_set (&h->lock, 1))
+ ;
+
+ h->tag = tag;
+}
+
always_inline void
ssvm_unlock (ssvm_shared_header_t * h)
{
}
}
+always_inline void
+ssvm_unlock_non_recursive (ssvm_shared_header_t * h)
+{
+ h->tag = 0;
+ CLIB_MEMORY_BARRIER ();
+ h->lock = 0;
+}
+
static inline void *
ssvm_push_heap (ssvm_shared_header_t * sh)
{
static inline u8
position_lt (svm_fifo_t * f, u32 a, u32 b)
{
- return (ooo_segment_distance_to_tail (f, a)
- < ooo_segment_distance_to_tail (f, b));
+ return (ooo_segment_distance_from_tail (f, a)
+ < ooo_segment_distance_from_tail (f, b));
}
static inline u8
position_leq (svm_fifo_t * f, u32 a, u32 b)
{
- return (ooo_segment_distance_to_tail (f, a)
- <= ooo_segment_distance_to_tail (f, b));
+ return (ooo_segment_distance_from_tail (f, a)
+ <= ooo_segment_distance_from_tail (f, b));
}
static inline u8
position_gt (svm_fifo_t * f, u32 a, u32 b)
{
- return (ooo_segment_distance_to_tail (f, a)
- > ooo_segment_distance_to_tail (f, b));
+ return (ooo_segment_distance_from_tail (f, a)
+ > ooo_segment_distance_from_tail (f, b));
}
static inline u32
position_diff (svm_fifo_t * f, u32 posa, u32 posb)
{
- return ooo_segment_distance_to_tail (f, posa)
- - ooo_segment_distance_to_tail (f, posb);
+ return ooo_segment_distance_from_tail (f, posa)
+ - ooo_segment_distance_from_tail (f, posb);
}
static inline u32
if (f == 0)
return 0;
- memset (f, 0, sizeof (*f) + data_size_in_bytes);
+ memset (f, 0, sizeof (*f));
f->nitems = data_size_in_bytes;
f->ooos_list_head = OOO_SEGMENT_INVALID_INDEX;
{
s = prev;
s_end_pos = ooo_segment_end_pos (f, s);
- goto merge;
+
+ /* Check head and tail now since segment may be wider at both ends so
+ * merge tests lower won't work */
+ if (position_lt (f, normalized_position, s->start))
+ {
+ s->start = normalized_position;
+ s->length = position_diff (f, s_end_pos, s->start);
+ }
+ if (position_gt (f, normalized_end_position, s_end_pos))
+ {
+ s->length = position_diff (f, normalized_end_position, s->start);
+ }
+ goto check_tail;
}
s_index = s - f->ooo_segments;
* Merge needed
*/
-merge:
-
/* Merge at head */
if (position_lt (f, normalized_position, s->start))
{
goto done;
}
+check_tail:
/* The new segment's tail may cover multiple smaller ones */
if (position_gt (f, normalized_end_position, s_end_pos))
{
/* If partial overlap with last, merge */
if (it && position_leq (f, it->start, normalized_end_position))
{
- s->length = ooo_segment_end_pos (f, it) - s->start;
+ s->length =
+ position_diff (f, ooo_segment_end_pos (f, it), s->start);
ooo_segment_del (f, it - f->ooo_segments);
}
}
i32 diff;
s = pool_elt_at_index (f->ooo_segments, f->ooos_list_head);
+ diff = ooo_segment_distance_to_tail (f, s->start);
- diff = (f->tail >= s->start) ?
- f->tail - s->start : f->nitems + f->tail - s->start;
+ ASSERT (diff != n_bytes_enqueued);
if (diff > n_bytes_enqueued)
return 0;
if (s->next != OOO_SEGMENT_INVALID_INDEX)
{
s = pool_elt_at_index (f->ooo_segments, s->next);
- diff = (f->tail >= s->start) ?
- f->tail - s->start : f->nitems + f->tail - s->start;
+ diff = ooo_segment_distance_to_tail (f, s->start);
ooo_segment_del (f, index);
}
/* End of search */
}
}
+ ASSERT (bytes >= 0 && bytes <= f->nitems);
return bytes;
}
}
else
{
+ ASSERT (0);
+
/* Account for a zero-copy enqueue done elsewhere */
ASSERT (max_bytes <= (nitems - cursize));
f->tail += max_bytes;
total_copy_bytes += ooo_segment_try_collect (f, total_copy_bytes);
/* Atomically increase the queue length */
+ ASSERT (cursize + total_copy_bytes <= nitems);
__sync_fetch_and_add (&f->cursize, total_copy_bytes);
return (total_copy_bytes);
cursize = svm_fifo_max_dequeue (f);
nitems = f->nitems;
+ ASSERT (required_bytes < nitems);
+
normalized_offset = (f->tail + offset) % nitems;
/* Will this request fit? */
}
else
{
+ ASSERT (0);
/* Account for a zero-copy dequeue done elsewhere */
ASSERT (max_bytes <= cursize);
f->head += max_bytes;
total_copy_bytes = max_bytes;
}
+ ASSERT (f->head <= nitems);
+ ASSERT (cursize >= total_copy_bytes);
__sync_fetch_and_sub (&f->cursize, total_copy_bytes);
return (total_copy_bytes);
f->head = (f->head == nitems) ? 0 : f->head;
}
+ ASSERT (f->head <= nitems);
+ ASSERT (cursize >= total_drop_bytes);
__sync_fetch_and_sub (&f->cursize, total_drop_bytes);
return total_drop_bytes;
}
always_inline u32
-ooo_segment_distance_to_tail (svm_fifo_t * f, u32 a)
+ooo_segment_distance_from_tail (svm_fifo_t * f, u32 pos)
{
/* Ambiguous. Assumption is that ooo segments don't touch tail */
- if (a == f->tail && f->tail == f->head)
+ if (PREDICT_FALSE (pos == f->tail && f->tail == f->head))
return f->nitems;
- return ((f->nitems + a - f->tail) % f->nitems);
+ return (((f->nitems + pos) - f->tail) % f->nitems);
+}
+
+always_inline u32
+ooo_segment_distance_to_tail (svm_fifo_t * f, u32 pos)
+{
+ return (((f->nitems + f->tail) - pos) % f->nitems);
}
always_inline u32
ooo_segment_offset (svm_fifo_t * f, ooo_segment_t * s)
{
- return ooo_segment_distance_to_tail (f, s->start);
+ return ooo_segment_distance_from_tail (f, s->start);
}
always_inline u32
ooo_segment_end_offset (svm_fifo_t * f, ooo_segment_t * s)
{
- return ooo_segment_distance_to_tail (f, s->start) + s->length;
+ return ooo_segment_distance_from_tail (f, s->start) + s->length;
}
always_inline u32
rx_fifo_size = (sizeof (*f) + a->rx_fifo_size) * a->preallocated_fifo_pairs;
tx_fifo_size = (sizeof (*f) + a->tx_fifo_size) * a->preallocated_fifo_pairs;
+ if (0)
+ clib_warning ("rx_fifo_size %u (%d mb), tx_fifo_size %u (%d mb)",
+ rx_fifo_size, rx_fifo_size >> 20,
+ tx_fifo_size, tx_fifo_size >> 20);
+
/* Allocate rx fifo space. May fail. */
rx_fifo_space = clib_mem_alloc_aligned_at_offset
(rx_fifo_size, CLIB_CACHE_LINE_BYTES, 0 /* align_offset */ ,
ssvm_pop_heap (oldheap);
sh->ready = 1;
- a->new_segment_index = s - sm->segments;
+ vec_add1 (a->new_segment_indices, s - sm->segments);
return (0);
}
svm_fifo_segment_main_t *sm = &svm_fifo_segment_main;
ssvm_shared_header_t *sh;
svm_fifo_segment_header_t *fsh;
+ void *oldheap;
+ u8 **heaps = 0;
+ mheap_t *heap_header;
+ int segment_count = 1;
+ int i;
- /* Allocate a fresh segment */
- pool_get (sm->segments, s);
- memset (s, 0, sizeof (*s));
-
- s->ssvm.ssvm_size = ~0;
- s->ssvm.i_am_master = 1;
- s->ssvm.my_pid = getpid ();
- s->ssvm.name = (u8 *) a->segment_name;
- s->ssvm.requested_va = ~0;
-
- /* Allocate a [sic] shared memory header, in process memory... */
- sh = clib_mem_alloc_aligned (sizeof (*sh), CLIB_CACHE_LINE_BYTES);
- s->ssvm.sh = sh;
+ if (a->private_segment_count && a->private_segment_size)
+ {
+ void *mem;
+ u8 *heap;
+ u32 pagesize = clib_mem_get_page_size ();
+ u32 rnd_size;
- memset (sh, 0, sizeof (*sh));
- sh->heap = clib_mem_get_heap ();
+ for (i = 0; i < a->private_segment_count; i++)
+ {
+ rnd_size = (a->private_segment_size + (pagesize - 1)) & ~pagesize;
+
+ mem = mmap (0, rnd_size, PROT_READ | PROT_WRITE,
+ MAP_PRIVATE | MAP_ANONYMOUS,
+ -1 /* fd */ , 0 /* offset */ );
+
+ if (mem == MAP_FAILED)
+ {
+ clib_unix_warning ("mmap");
+ return -1;
+ }
+ heap = mheap_alloc (mem, rnd_size);
+ heap_header = mheap_header (heap);
+ heap_header->flags |= MHEAP_FLAG_THREAD_SAFE;
+ vec_add1 (heaps, heap);
+ }
+ segment_count = a->private_segment_count;
+ }
- /* Set up svm_fifo_segment shared header */
- fsh = clib_mem_alloc (sizeof (*fsh));
- memset (fsh, 0, sizeof (*fsh));
- sh->opaque[0] = fsh;
- s->h = fsh;
- fsh->segment_name = format (0, "%s%c", a->segment_name, 0);
+ /* Spread preallocated fifo pairs across segments */
+ a->preallocated_fifo_pairs /= segment_count;
- preallocate_fifo_pairs (fsh, a);
+ /* Allocate segments */
+ for (i = 0; i < segment_count; i++)
+ {
+ pool_get (sm->segments, s);
+ memset (s, 0, sizeof (*s));
+
+ s->ssvm.ssvm_size = ~0;
+ s->ssvm.i_am_master = 1;
+ s->ssvm.my_pid = getpid ();
+ s->ssvm.name = (u8 *) a->segment_name;
+ s->ssvm.requested_va = ~0;
+
+ /* Allocate a [sic] shared memory header, in process memory... */
+ sh = clib_mem_alloc_aligned (sizeof (*sh), CLIB_CACHE_LINE_BYTES);
+ s->ssvm.sh = sh;
+
+ memset (sh, 0, sizeof (*sh));
+ sh->heap = a->private_segment_count ? heaps[i] : clib_mem_get_heap ();
+
+ /* Set up svm_fifo_segment shared header */
+ fsh = clib_mem_alloc (sizeof (*fsh));
+ memset (fsh, 0, sizeof (*fsh));
+ sh->opaque[0] = fsh;
+ s->h = fsh;
+ fsh->segment_name = format (0, "%s%c", a->segment_name, 0);
+
+ if (a->private_segment_count)
+ {
+ oldheap = clib_mem_get_heap ();
+ clib_mem_set_heap (sh->heap);
+ preallocate_fifo_pairs (fsh, a);
+ clib_mem_set_heap (oldheap);
+ }
- sh->ready = 1;
- a->new_segment_index = s - sm->segments;
+ sh->ready = 1;
+ vec_add1 (a->new_segment_indices, s - sm->segments);
+ }
+ vec_free (heaps);
return (0);
}
fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
s->h = fsh;
- a->new_segment_index = s - sm->segments;
+ vec_add1 (a->new_segment_indices, s - sm->segments);
return (0);
}
sh = s->ssvm.sh;
fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
- ssvm_lock (sh, 1, 0);
+ ssvm_lock_non_recursive (sh, 1);
oldheap = ssvm_push_heap (sh);
switch (list_index)
if (PREDICT_FALSE (f == 0))
{
ssvm_pop_heap (oldheap);
- ssvm_unlock (sh);
+ ssvm_unlock_non_recursive (sh);
return (0);
}
}
ssvm_pop_heap (oldheap);
- ssvm_unlock (sh);
+ ssvm_unlock_non_recursive (sh);
return (f);
}
svm_fifo_segment_header_t *fsh;
void *oldheap;
+
sh = s->ssvm.sh;
fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
- ssvm_lock (sh, 1, 0);
+ ssvm_lock_non_recursive (sh, 2);
oldheap = ssvm_push_heap (sh);
switch (list_index)
}
ssvm_pop_heap (oldheap);
- ssvm_unlock (sh);
+ ssvm_unlock_non_recursive (sh);
}
void
{
char *segment_name;
u32 segment_size;
- u32 new_segment_index;
+ u32 *new_segment_indices;
u32 rx_fifo_size;
u32 tx_fifo_size;
u32 preallocated_fifo_pairs;
+ u32 private_segment_count;
+ u32 private_segment_size;
} svm_fifo_segment_create_args_t;
static inline svm_fifo_segment_private_t *
if (rv)
return clib_error_return (0, "svm_fifo_segment_create returned %d", rv);
- sp = svm_fifo_get_segment (a->new_segment_index);
+ sp = svm_fifo_get_segment (a->new_segment_indices[0]);
f = svm_fifo_segment_alloc_fifo (sp, 4096, FIFO_SEGMENT_RX_FREELIST);
if (rv)
return clib_error_return (0, "svm_fifo_segment_create returned %d", rv);
- sp = svm_fifo_get_segment (a->new_segment_index);
+ sp = svm_fifo_get_segment (a->new_segment_indices[0]);
f = svm_fifo_segment_alloc_fifo (sp, 4096, FIFO_SEGMENT_RX_FREELIST);
if (rv)
return clib_error_return (0, "svm_fifo_segment_create returned %d", rv);
- sp = svm_fifo_get_segment (a->new_segment_index);
+ sp = svm_fifo_get_segment (a->new_segment_indices[0]);
for (i = 0; i < 1000; i++)
{
if (rv)
return clib_error_return (0, "svm_fifo_segment_create returned %d", rv);
- sp = svm_fifo_get_segment (a->new_segment_index);
+ sp = svm_fifo_get_segment (a->new_segment_indices[0]);
f = svm_fifo_segment_alloc_fifo (sp, 200 << 10, FIFO_SEGMENT_RX_FREELIST);
if (rv)
return clib_error_return (0, "svm_fifo_segment_attach returned %d", rv);
- sp = svm_fifo_get_segment (a->new_segment_index);
+ sp = svm_fifo_get_segment (a->new_segment_indices[0]);
sh = sp->ssvm.sh;
fsh = (svm_fifo_segment_header_t *) sh->opaque[0];
return;
}
- segment_index = a->new_segment_index;
+ segment_index = a->new_segment_indices[0];
vec_add2 (utm->seg, seg, 1);
memcpy (seg, sm->segments + segment_index, sizeof (*seg));
sleep (1);
props->preallocated_fifo_pairs = options[APP_OPTIONS_PREALLOC_FIFO_PAIRS];
props->use_private_segment = options[APP_OPTIONS_FLAGS]
& APP_OPTIONS_FLAGS_BUILTIN_APP;
+ props->private_segment_count = options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT];
+ props->private_segment_size = options[APP_OPTIONS_PRIVATE_SEGMENT_SIZE];
first_seg_size = options[SESSION_OPTIONS_SEGMENT_SIZE];
if ((rv = segment_manager_init (sm, props, first_seg_size)))
return 0;
}
-session_type_t
-session_type_from_proto_and_ip (session_api_proto_t proto, u8 is_ip4)
-{
- if (proto == SESSION_PROTO_TCP)
- {
- if (is_ip4)
- return SESSION_TYPE_IP4_TCP;
- else
- return SESSION_TYPE_IP6_TCP;
- }
- else
- {
- if (is_ip4)
- return SESSION_TYPE_IP4_UDP;
- else
- return SESSION_TYPE_IP6_UDP;
- }
-
- return SESSION_N_TYPES;
-}
-
int
vnet_bind_uri (vnet_bind_args_t * a)
{
#include <vnet/session/application.h>
#include <vnet/session/transport.h>
-typedef enum _session_api_proto
-{
- SESSION_PROTO_TCP,
- SESSION_PROTO_UDP
-} session_api_proto_t;
-
typedef struct _vnet_app_attach_args_t
{
/** Binary API client index */
struct
{
transport_endpoint_t tep;
- session_api_proto_t proto;
+ transport_proto_t proto;
};
};
struct
{
transport_endpoint_t tep;
- session_api_proto_t proto;
+ transport_proto_t proto;
};
};
u32 app_index;
APP_EVT_QUEUE_SIZE,
APP_OPTIONS_FLAGS,
APP_OPTIONS_PREALLOC_FIFO_PAIRS,
+ APP_OPTIONS_PRIVATE_SEGMENT_COUNT,
+ APP_OPTIONS_PRIVATE_SEGMENT_SIZE,
SESSION_OPTIONS_SEGMENT_SIZE,
SESSION_OPTIONS_ADD_SEGMENT_SIZE,
SESSION_OPTIONS_RX_FIFO_SIZE,
n_tx_pkts, 0);
}
-stream_session_t *
-session_event_get_session (session_fifo_event_t * e0, u8 thread_index)
+always_inline stream_session_t *
+session_event_get_session (session_fifo_event_t * e, u8 thread_index)
{
- svm_fifo_t *f0;
- stream_session_t *s0;
- u32 session_index0;
-
- f0 = e0->fifo;
- session_index0 = f0->master_session_index;
-
- /* $$$ add multiple event queues, per vpp worker thread */
- ASSERT (f0->master_thread_index == thread_index);
-
- s0 = stream_session_get_if_valid (session_index0, thread_index);
-
- ASSERT (s0 == 0 || s0->thread_index == thread_index);
-
- return s0;
+ ASSERT (e->fifo->master_thread_index == thread_index);
+ return stream_session_get_if_valid (e->fifo->master_session_index,
+ thread_index);
}
void
case FIFO_EVENT_BUILTIN_RX:
s0 = session_event_get_session (e0, my_thread_index);
svm_fifo_unset_event (s0->server_rx_fifo);
- /* Get session's server */
app = application_get (s0->app_index);
app->cb_fns.builtin_server_rx_callback (s0);
break;
/**
* Process private segment index
*/
-u32 private_segment_index = ~0;
+u32 *private_segment_indices;
/**
* Default fifo and segment size. TODO config.
return VNET_API_ERROR_SVM_SEGMENT_CREATE_FAIL;
}
- vec_add1 (sm->segment_indices, ca->new_segment_index);
+ vec_append (sm->segment_indices, ca->new_segment_indices);
+ vec_free (ca->new_segment_indices);
return 0;
}
{
svm_fifo_segment_create_args_t _a, *a = &_a;
- if (private_segment_index != ~0)
+ if (private_segment_indices)
return;
memset (a, 0, sizeof (*a));
a->segment_name = "process-private-segment";
a->segment_size = ~0;
- a->new_segment_index = ~0;
a->rx_fifo_size = props->rx_fifo_size;
a->tx_fifo_size = props->tx_fifo_size;
a->preallocated_fifo_pairs = props->preallocated_fifo_pairs;
+ a->private_segment_count = props->private_segment_count;
+ a->private_segment_size = props->private_segment_size;
if (svm_fifo_segment_create_process_private (a))
clib_warning ("Failed to create process private segment");
- private_segment_index = a->new_segment_index;
- ASSERT (private_segment_index != ~0);
+ private_segment_indices = a->new_segment_indices;
+ ASSERT (vec_len (private_segment_indices));
}
/**
}
else
{
- if (private_segment_index == ~0)
+ if (vec_len (private_segment_indices) == 0)
segment_manager_alloc_process_private_segment (properties);
- ASSERT (private_segment_index != ~0);
- vec_add1 (sm->segment_indices, private_segment_index);
+ ASSERT (vec_len (private_segment_indices));
+ vec_append (sm->segment_indices, private_segment_indices);
}
clib_spinlock_init (&sm->lockp);
/* See if we're supposed to create another segment */
if (*server_rx_fifo == 0)
{
- if (sm->properties->add_segment)
+ if (sm->properties->add_segment && !sm->properties->use_private_segment)
{
if (added_a_segment)
{
svm_fifo_segment_free_fifo (fifo_segment, tx_fifo,
FIFO_SEGMENT_TX_FREELIST);
+ /* Don't try to delete process-private segments */
+ if (sm->properties->private_segment_count > 0)
+ return;
+
/* Remove segment only if it holds no fifos and not the first */
if (sm->segment_indices[0] != svm_segment_index
&& !svm_fifo_segment_has_fifos (fifo_segment))
/** Use private memory segment instead of shared memory */
u8 use_private_segment;
+
+ /** Use one or more private mheaps, instead of the global heap */
+ u32 private_segment_count;
+ u32 private_segment_size;
} segment_manager_properties_t;
typedef struct _segment_manager
*/
stream_session_t *
stream_session_lookup4 (ip4_address_t * lcl, ip4_address_t * rmt,
- u16 lcl_port, u16 rmt_port, u8 proto,
- u32 my_thread_index)
+ u16 lcl_port, u16 rmt_port, u8 proto)
{
session_manager_main_t *smm = &session_manager_main;
session_kv4_t kv4;
+ stream_session_t *s;
int rv;
/* Lookup session amongst established ones */
make_v4_ss_kv (&kv4, lcl, rmt, lcl_port, rmt_port, proto);
rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
if (rv == 0)
- return stream_session_get_tsi (kv4.value, my_thread_index);
+ return stream_session_get_from_handle (kv4.value);
/* If nothing is found, check if any listener is available */
- return stream_session_lookup_listener4 (lcl, lcl_port, proto);
+ if ((s = stream_session_lookup_listener4 (lcl, lcl_port, proto)))
+ return s;
+
+ /* Finally, try half-open connections */
+ rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
+ if (rv == 0)
+ return stream_session_get_from_handle (kv4.value);
+ return 0;
}
stream_session_t *
* wildcarded local source (listener bound to all interfaces) */
stream_session_t *
stream_session_lookup6 (ip6_address_t * lcl, ip6_address_t * rmt,
- u16 lcl_port, u16 rmt_port, u8 proto,
- u32 my_thread_index)
+ u16 lcl_port, u16 rmt_port, u8 proto)
{
session_manager_main_t *smm = vnet_get_session_manager_main ();
session_kv6_t kv6;
+ stream_session_t *s;
int rv;
make_v6_ss_kv (&kv6, lcl, rmt, lcl_port, rmt_port, proto);
rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
if (rv == 0)
- return stream_session_get_tsi (kv6.value, my_thread_index);
+ return stream_session_get_from_handle (kv6.value);
/* If nothing is found, check if any listener is available */
- return stream_session_lookup_listener6 (lcl, lcl_port, proto);
+ if ((s = stream_session_lookup_listener6 (lcl, lcl_port, proto)))
+ return s;
+
+ /* Finally, try half-open connections */
+ rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
+ if (rv == 0)
+ return stream_session_get_from_handle (kv6.value);
+ return 0;
}
stream_session_t *
rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
if (rv == 0)
return tp_vfts[proto].get_half_open (kv4.value & 0xFFFFFFFF);
-
return 0;
}
u32 thread_index = tc->thread_index;
int rv;
+ ASSERT (thread_index == vlib_get_thread_index ());
+
if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo,
&server_tx_fifo,
&fifo_segment_index)))
s->app_index = server->index;
s->listener_index = listener_index;
+ s->session_state = SESSION_STATE_ACCEPTING;
/* Shoulder-tap the server */
if (notify)
}
}
+session_type_t
+session_type_from_proto_and_ip (transport_proto_t proto, u8 is_ip4)
+{
+ if (proto == TRANSPORT_PROTO_TCP)
+ {
+ if (is_ip4)
+ return SESSION_TYPE_IP4_TCP;
+ else
+ return SESSION_TYPE_IP6_TCP;
+ }
+ else
+ {
+ if (is_ip4)
+ return SESSION_TYPE_IP4_UDP;
+ else
+ return SESSION_TYPE_IP6_UDP;
+ }
+
+ return SESSION_N_TYPES;
+}
+
static clib_error_t *
session_manager_main_enable (vlib_main_t * vm)
{
session_vpp_event_queue_allocate (smm, i);
/* $$$$ preallocate hack config parameter */
- for (i = 0; i < 200000; i++)
+ for (i = 0; i < smm->preallocated_sessions; i++)
{
- stream_session_t *ss;
+ stream_session_t *ss __attribute__ ((unused));
pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES);
- memset (ss, 0, sizeof (*ss));
}
- for (i = 0; i < 200000; i++)
+ for (i = 0; i < smm->preallocated_sessions; i++)
pool_put_index (smm->sessions[0], i);
clib_bihash_init_16_8 (&smm->v4_session_hash, "v4 session table",
return 0;
}
-VLIB_INIT_FUNCTION (session_manager_main_init)
- static clib_error_t *session_config_fn (vlib_main_t * vm,
- unformat_input_t * input)
+VLIB_INIT_FUNCTION (session_manager_main_init);
+
+static clib_error_t *
+session_config_fn (vlib_main_t * vm, unformat_input_t * input)
{
session_manager_main_t *smm = &session_manager_main;
u32 nitems;
else
clib_warning ("event queue length %d too small, ignored", nitems);
}
+ if (unformat (input, "preallocated-sessions %d",
+ &smm->preallocated_sessions))
+ ;
else
return clib_error_return (0, "unknown input `%U'",
format_unformat_error, input);
SESSION_N_TYPES,
} session_type_t;
+
+session_type_t
+session_type_from_proto_and_ip (transport_proto_t proto, u8 is_ip4);
+
/*
* Application session state
*/
{
SESSION_STATE_LISTENING,
SESSION_STATE_CONNECTING,
+ SESSION_STATE_ACCEPTING,
SESSION_STATE_READY,
SESSION_STATE_CLOSED,
SESSION_STATE_N_STATES,
/** Per transport rx function that can either dequeue or peek */
session_fifo_rx_fn *session_tx_fns[SESSION_N_TYPES];
+ /** Session manager is enabled */
u8 is_enabled;
+ /** Preallocate session config parameter */
+ u32 preallocated_sessions;
+
/* Convenience */
vlib_main_t *vlib_main;
vnet_main_t *vnet_main;
u16 lcl_port, u8 proto);
stream_session_t *stream_session_lookup4 (ip4_address_t * lcl,
ip4_address_t * rmt, u16 lcl_port,
- u16 rmt_port, u8 proto,
- u32 thread_index);
+ u16 rmt_port, u8 proto);
stream_session_t *stream_session_lookup_listener6 (ip6_address_t * lcl,
u16 lcl_port, u8 proto);
stream_session_t *stream_session_lookup6 (ip6_address_t * lcl,
ip6_address_t * rmt, u16 lcl_port,
- u16 rmt_port, u8, u32 thread_index);
+ u16 rmt_port, u8 proto);
transport_connection_t
* stream_session_lookup_transport4 (ip4_address_t * lcl,
ip4_address_t * rmt, u16 lcl_port,
ti_and_si & 0xFFFFFFFFULL);
}
+always_inline u8
+stream_session_is_valid (u32 si, u8 thread_index)
+{
+ stream_session_t *s;
+ s = pool_elt_at_index (session_manager_main.sessions[thread_index], si);
+ if (s->thread_index != thread_index || s->session_index != si
+ || s->server_rx_fifo->master_session_index != si
+ || s->server_tx_fifo->master_session_index != si
+ || s->server_rx_fifo->master_thread_index != thread_index
+ || s->server_tx_fifo->master_thread_index != thread_index)
+ return 0;
+ return 1;
+}
+
always_inline stream_session_t *
stream_session_get (u32 si, u32 thread_index)
{
+ ASSERT (stream_session_is_valid (si, thread_index));
return pool_elt_at_index (session_manager_main.sessions[thread_index], si);
}
if (pool_is_free_index (session_manager_main.sessions[thread_index], si))
return 0;
+ ASSERT (stream_session_is_valid (si, thread_index));
return pool_elt_at_index (session_manager_main.sessions[thread_index], si);
}
svm_fifo_max_enqueue (ss->server_tx_fifo),
stream_session_get_index (ss));
- if (ss->session_state == SESSION_STATE_READY)
+ if (ss->session_state == SESSION_STATE_READY
+ || ss->session_state == SESSION_STATE_ACCEPTING)
{
s = format (s, "%U", tp_vft->format_connection, ss->connection_index,
ss->thread_index, verbose);
}
else if (ss->session_state == SESSION_STATE_CLOSED)
{
- s = format (s, "[CL] %-40U", tp_vft->format_connection,
- ss->connection_index, ss->thread_index, verbose);
+ s =
+ format (s, "[CL] %U", tp_vft->format_connection, ss->connection_index,
+ ss->thread_index, verbose);
if (verbose == 1)
s = format (s, "%v", str);
if (verbose > 1)
int verbose = 0, i;
stream_session_t *pool;
stream_session_t *s;
- u8 *str = 0;
+ u8 *str = 0, one_session = 0, proto_set = 0, proto = 0;
+ u8 is_ip4 = 0, s_type = 0;
+ ip4_address_t lcl_ip4, rmt_ip4;
+ u32 lcl_port = 0, rmt_port = 0;
+
+ memset (&lcl_ip4, 0, sizeof (lcl_ip4));
+ memset (&rmt_ip4, 0, sizeof (rmt_ip4));
if (!smm->is_enabled)
{
;
else if (unformat (input, "verbose"))
verbose = 1;
+ else if (unformat (input, "tcp"))
+ {
+ proto_set = 1;
+ proto = TRANSPORT_PROTO_TCP;
+ }
+ else if (unformat (input, "%U:%d->%U:%d",
+ unformat_ip4_address, &lcl_ip4, &lcl_port,
+ unformat_ip4_address, &rmt_ip4, &rmt_port))
+ {
+ one_session = 1;
+ is_ip4 = 1;
+ }
+
else
break;
}
+ if (one_session)
+ {
+ if (!proto_set)
+ {
+ vlib_cli_output (vm, "proto not set");
+ return clib_error_return (0, "proto not set");
+ }
+
+ s_type = session_type_from_proto_and_ip (proto, is_ip4);
+ s = stream_session_lookup4 (&lcl_ip4, &rmt_ip4,
+ clib_host_to_net_u16 (lcl_port),
+ clib_host_to_net_u16 (rmt_port), s_type);
+ if (s)
+ vlib_cli_output (vm, "%U", format_stream_session, s, 2);
+ else
+ vlib_cli_output (vm, "session does not exist");
+
+ return 0;
+ }
+
for (i = 0; i < vec_len (smm->sessions); i++)
{
u32 once_per_pool;
}
else
vlib_cli_output (vm, "Thread %d: no active sessions", i);
+ vec_reset_length (str);
}
vec_free (str);
};
/* *INDENT-ON* */
+static int
+clear_session (stream_session_t * s)
+{
+ application_t *server = application_get (s->app_index);
+ server->cb_fns.session_disconnect_callback (s);
+ return 0;
+}
+
static clib_error_t *
clear_session_command_fn (vlib_main_t * vm, unformat_input_t * input,
vlib_cli_command_t * cmd)
{
session_manager_main_t *smm = &session_manager_main;
- u32 thread_index = 0;
+ u32 thread_index = 0, clear_all = 0;
u32 session_index = ~0;
- stream_session_t *pool, *session;
- application_t *server;
+ stream_session_t **pool, *session;
if (!smm->is_enabled)
{
;
else if (unformat (input, "session %d", &session_index))
;
+ else if (unformat (input, "all"))
+ clear_all = 1;
else
return clib_error_return (0, "unknown input `%U'",
format_unformat_error, input);
}
- if (session_index == ~0)
+ if (!clear_all && session_index == ~0)
return clib_error_return (0, "session <nn> required, but not set.");
- if (thread_index > vec_len (smm->sessions))
- return clib_error_return (0, "thread %d out of range [0-%d]",
- thread_index, vec_len (smm->sessions));
-
- pool = smm->sessions[thread_index];
-
- if (pool_is_free_index (pool, session_index))
- return clib_error_return (0, "session %d not active", session_index);
-
- session = pool_elt_at_index (pool, session_index);
- server = application_get (session->app_index);
+ if (session_index != ~0)
+ {
+ session = stream_session_get_if_valid (session_index, thread_index);
+ if (!session)
+ return clib_error_return (0, "no session %d on thread %d",
+ session_index, thread_index);
+ clear_session (session);
+ }
- /* Disconnect both app and transport */
- server->cb_fns.session_disconnect_callback (session);
+ if (clear_all)
+ {
+ /* *INDENT-OFF* */
+ vec_foreach (pool, smm->sessions)
+ {
+ pool_foreach(session, *pool, ({
+ clear_session (session);
+ }));
+ };
+ /* *INDENT-ON* */
+ }
return 0;
}
t->rmt_port, t->proto);
}
+typedef enum _transport_proto
+{
+ TRANSPORT_PROTO_TCP,
+ TRANSPORT_PROTO_UDP
+} transport_proto_t;
+
typedef struct _transport_endpoint
{
ip46_address_t ip; /** ip address */
{
tclient_main_t *tm = &tclient_main;
int my_thread_index = vlib_get_thread_index ();
- vl_api_disconnect_session_t *dmp;
session_t *sp;
int i;
int delete_session;
u32 *connection_indices;
- u32 tx_quota = 0;
- u32 delta, prev_bytes_received_this_session;
+ u32 *connections_this_batch;
+ u32 nconnections_this_batch;
connection_indices = tm->connection_index_by_thread[my_thread_index];
+ connections_this_batch =
+ tm->connections_this_batch_by_thread[my_thread_index];
- if (tm->run_test == 0 || vec_len (connection_indices) == 0)
+ if ((tm->run_test == 0) ||
+ ((vec_len (connection_indices) == 0)
+ && vec_len (connections_this_batch) == 0))
return 0;
- for (i = 0; i < vec_len (connection_indices); i++)
+ /* Grab another pile of connections */
+ if (PREDICT_FALSE (vec_len (connections_this_batch) == 0))
+ {
+ nconnections_this_batch =
+ clib_min (tm->connections_per_batch, vec_len (connection_indices));
+
+ ASSERT (nconnections_this_batch > 0);
+ vec_validate (connections_this_batch, nconnections_this_batch - 1);
+ clib_memcpy (connections_this_batch,
+ connection_indices + vec_len (connection_indices)
+ - nconnections_this_batch,
+ nconnections_this_batch * sizeof (u32));
+ _vec_len (connection_indices) -= nconnections_this_batch;
+ }
+
+ if (PREDICT_FALSE (tm->prev_conns != tm->connections_per_batch
+ && tm->prev_conns == vec_len (connections_this_batch)))
+ {
+ tm->repeats++;
+ tm->prev_conns = vec_len (connections_this_batch);
+ if (tm->repeats == 500000)
+ {
+ clib_warning ("stuck clients");
+ }
+ }
+ else
+ {
+ tm->prev_conns = vec_len (connections_this_batch);
+ tm->repeats = 0;
+ }
+
+ for (i = 0; i < vec_len (connections_this_batch); i++)
{
delete_session = 1;
- sp = pool_elt_at_index (tm->sessions, connection_indices[i]);
+ sp = pool_elt_at_index (tm->sessions, connections_this_batch[i]);
- if ((tm->no_return || tx_quota < 60) && sp->bytes_to_send > 0)
+ if (sp->bytes_to_send > 0)
{
send_test_chunk (tm, sp);
delete_session = 0;
- tx_quota++;
}
- if (!tm->no_return && sp->bytes_to_receive > 0)
+ if (sp->bytes_to_receive > 0)
{
- prev_bytes_received_this_session = sp->bytes_received;
receive_test_chunk (tm, sp);
- delta = sp->bytes_received - prev_bytes_received_this_session;
- if (delta > 0)
- tx_quota--;
delete_session = 0;
}
if (PREDICT_FALSE (delete_session == 1))
{
- __sync_fetch_and_add (&tm->tx_total, tm->bytes_to_send);
+ u32 index, thread_index;
+ stream_session_t *s;
+
+ __sync_fetch_and_add (&tm->tx_total, sp->bytes_sent);
__sync_fetch_and_add (&tm->rx_total, sp->bytes_received);
- dmp = vl_msg_api_alloc_as_if_client (sizeof (*dmp));
- memset (dmp, 0, sizeof (*dmp));
- dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
- dmp->client_index = tm->my_client_index;
- dmp->handle = sp->vpp_session_handle;
- if (!unix_shared_memory_queue_add (tm->vl_input_queue, (u8 *) & dmp,
- 1))
+ stream_session_parse_handle (sp->vpp_session_handle,
+ &index, &thread_index);
+ s = stream_session_get_if_valid (index, thread_index);
+
+ if (s)
{
- vec_delete (connection_indices, 1, i);
- tm->connection_index_by_thread[my_thread_index] =
- connection_indices;
+ stream_session_disconnect (s);
+ vec_delete (connections_this_batch, 1, i);
+ i--;
__sync_fetch_and_add (&tm->ready_connections, -1);
}
else
- {
- vl_msg_api_free (dmp);
- }
+ clib_warning ("session AWOL?");
/* Kick the debug CLI process */
if (tm->ready_connections == 0)
}
}
}
+
+ tm->connection_index_by_thread[my_thread_index] = connection_indices;
+ tm->connections_this_batch_by_thread[my_thread_index] =
+ connections_this_batch;
return 0;
}
tm->vlib_main = vm;
vec_validate (tm->connection_index_by_thread, thread_main->n_vlib_mains);
+ vec_validate (tm->connections_this_batch_by_thread,
+ thread_main->n_vlib_mains);
return 0;
}
pool_get (tm->sessions, session);
memset (session, 0, sizeof (*session));
session_index = session - tm->sessions;
- session->bytes_to_receive = session->bytes_to_send = tm->bytes_to_send;
+ session->bytes_to_send = tm->bytes_to_send;
+ session->bytes_to_receive = tm->no_return ? 0ULL : tm->bytes_to_send;
session->server_rx_fifo = s->server_rx_fifo;
session->server_rx_fifo->client_session_index = session_index;
session->server_tx_fifo = s->server_tx_fifo;
options[SESSION_OPTIONS_SEGMENT_SIZE] = (2ULL << 32);
options[SESSION_OPTIONS_RX_FIFO_SIZE] = tm->fifo_size;
options[SESSION_OPTIONS_TX_FIFO_SIZE] = tm->fifo_size / 2;
+ options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = tm->private_segment_count;
+ options[APP_OPTIONS_PRIVATE_SEGMENT_SIZE] = tm->private_segment_size;
options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] = prealloc_fifos;
options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
tm->bytes_to_send = 8192;
tm->no_return = 0;
tm->fifo_size = 64 << 10;
+ tm->connections_per_batch = 1000;
+ tm->private_segment_count = 0;
+ tm->private_segment_size = 0;
vec_free (tm->connect_uri);
tm->no_return = 1;
else if (unformat (input, "fifo-size %d", &tm->fifo_size))
tm->fifo_size <<= 10;
+ else if (unformat (input, "private-segment-count %d",
+ &tm->private_segment_count))
+ ;
+ else if (unformat (input, "private-segment-size %dm", &tmp))
+ tm->private_segment_size = tmp << 20;
+ else if (unformat (input, "private-segment-size %dg", &tmp))
+ tm->private_segment_size = tmp << 30;
+ else if (unformat (input, "private-segment-size %d", &tmp))
+ tm->private_segment_size = tmp;
+ else if (unformat (input, "preallocate-fifos"))
+ tm->prealloc_fifos = 1;
+ else
+ if (unformat (input, "client-batch %d", &tm->connections_per_batch))
+ ;
else
return clib_error_return (0, "unknown input `%U'",
format_unformat_error, input);
vlib_cli_output (vm, "zero delta-t?");
cleanup:
- pool_free (tm->sessions);
+ tm->run_test = 0;
for (i = 0; i < vec_len (tm->connection_index_by_thread); i++)
- vec_reset_length (tm->connection_index_by_thread[i]);
+ {
+ vec_reset_length (tm->connection_index_by_thread[i]);
+ vec_reset_length (tm->connections_this_batch_by_thread[i]);
+ }
+ pool_free (tm->sessions);
return 0;
}
u32 configured_segment_size;
u32 fifo_size;
u32 expected_connections; /**< Number of clients/connections */
+ u32 connections_per_batch; /**< Connections to rx/tx at once */
+ u32 private_segment_count; /**< Number of private fifo segs */
+ u32 private_segment_size; /**< size of private fifo segs */
/*
* Test state variables
uword *session_index_by_vpp_handles; /**< Hash table for disconnecting */
u8 *connect_test_data; /**< Pre-computed test data */
u32 **connection_index_by_thread;
+ u32 **connections_this_batch_by_thread; /**< active connection batch */
pthread_t client_thread_handle;
volatile u32 ready_connections;
f64 test_start_time;
f64 test_end_time;
-
+ u32 prev_conns;
+ u32 repeats;
/*
* Flags
*/
u32 fifo_size; /**< Fifo size */
u32 rcv_buffer_size; /**< Rcv buffer size */
u32 prealloc_fifos; /**< Preallocate fifos */
+ u32 private_segment_count; /**< Number of private segments */
+ u32 private_segment_size; /**< Size of private segments */
/*
* Test state
*/
u8 **rx_buf; /**< Per-thread RX buffer */
u64 byte_index;
+ u32 **rx_retries;
vlib_main_t *vlib_main;
} builtin_server_main_t;
session_manager_get_vpp_event_queue (s->thread_index);
s->session_state = SESSION_STATE_READY;
bsm->byte_index = 0;
+ vec_validate (bsm->rx_retries[s->thread_index], s->session_index);
+ bsm->rx_retries[s->thread_index][s->session_index] = 0;
return 0;
}
builtin_server_main_t *bsm = &builtin_server_main;
session_fifo_event_t evt;
static int serial_number = 0;
- u32 my_thread_id = vlib_get_thread_index ();
+ u32 thread_index = vlib_get_thread_index ();
+
+ ASSERT (s->thread_index == thread_index);
rx_fifo = s->server_rx_fifo;
tx_fifo = s->server_tx_fifo;
+ ASSERT (rx_fifo->master_thread_index == thread_index);
+ ASSERT (tx_fifo->master_thread_index == thread_index);
+
max_dequeue = svm_fifo_max_dequeue (s->server_rx_fifo);
max_enqueue = svm_fifo_max_enqueue (s->server_tx_fifo);
evt.event_type = FIFO_EVENT_BUILTIN_RX;
evt.event_id = 0;
- q = bsm->vpp_queue[s->thread_index];
+ q = bsm->vpp_queue[thread_index];
if (PREDICT_FALSE (q->cursize == q->maxsize))
clib_warning ("out of event queue space");
- else
- unix_shared_memory_queue_add (q, (u8 *) & evt,
- 0 /* don't wait for mutex */ );
+ else if (unix_shared_memory_queue_add (q, (u8 *) & evt, 0 /* don't wait for mutex */
+ ))
+ clib_warning ("failed to enqueue self-tap");
+
+ bsm->rx_retries[thread_index][s->session_index]++;
+ if (bsm->rx_retries[thread_index][s->session_index] == 500000)
+ {
+ clib_warning ("session stuck: %U", format_stream_session, s, 2);
+ }
+ }
+ else
+ {
+ bsm->rx_retries[thread_index][s->session_index] = 0;
}
return 0;
}
- _vec_len (bsm->rx_buf[my_thread_id]) = max_transfer;
+ _vec_len (bsm->rx_buf[thread_index]) = max_transfer;
actual_transfer = svm_fifo_dequeue_nowait (rx_fifo, max_transfer,
- bsm->rx_buf[my_thread_id]);
+ bsm->rx_buf[thread_index]);
ASSERT (actual_transfer == max_transfer);
// test_bytes (bsm, actual_transfer);
*/
n_written = svm_fifo_enqueue_nowait (tx_fifo, actual_transfer,
- bsm->rx_buf[my_thread_id]);
+ bsm->rx_buf[thread_index]);
if (n_written != max_transfer)
clib_warning ("short trout!");
evt.event_type = FIFO_EVENT_APP_TX;
evt.event_id = serial_number++;
- unix_shared_memory_queue_add (bsm->vpp_queue[s->thread_index],
- (u8 *) & evt, 0 /* do wait for mutex */ );
+ if (unix_shared_memory_queue_add (bsm->vpp_queue[s->thread_index],
+ (u8 *) & evt,
+ 0 /* do wait for mutex */ ))
+ clib_warning ("failed to enqueue tx evt");
}
- if (PREDICT_FALSE (max_enqueue < max_dequeue))
+ if (PREDICT_FALSE (n_written < max_dequeue))
goto rx_event;
return 0;
a->options[SESSION_OPTIONS_SEGMENT_SIZE] = 512 << 20;
a->options[SESSION_OPTIONS_RX_FIFO_SIZE] = bsm->fifo_size;
a->options[SESSION_OPTIONS_TX_FIFO_SIZE] = bsm->fifo_size;
- a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
+ a->options[APP_OPTIONS_PRIVATE_SEGMENT_COUNT] = bsm->private_segment_count;
+ a->options[APP_OPTIONS_PRIVATE_SEGMENT_SIZE] = bsm->private_segment_size;
a->options[APP_OPTIONS_PREALLOC_FIFO_PAIRS] =
bsm->prealloc_fifos ? bsm->prealloc_fifos : 1;
+
+ a->options[APP_OPTIONS_FLAGS] = APP_OPTIONS_FLAGS_BUILTIN_APP;
+
a->segment_name = segment_name;
a->segment_name_length = ARRAY_LEN (segment_name);
num_threads = 1 /* main thread */ + vtm->n_threads;
vec_validate (builtin_server_main.vpp_queue, num_threads - 1);
vec_validate (bsm->rx_buf, num_threads - 1);
+ vec_validate (bsm->rx_retries, num_threads - 1);
+
for (i = 0; i < num_threads; i++)
vec_validate (bsm->rx_buf[i], bsm->rcv_buffer_size);
{
builtin_server_main_t *bsm = &builtin_server_main;
int rv;
+ u32 tmp;
bsm->no_echo = 0;
bsm->fifo_size = 64 << 10;
bsm->rcv_buffer_size = 128 << 10;
bsm->prealloc_fifos = 0;
+ bsm->private_segment_count = 0;
+ bsm->private_segment_size = 0;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
bsm->fifo_size <<= 10;
else if (unformat (input, "rcv-buf-size %d", &bsm->rcv_buffer_size))
;
- else if (unformat (input, "prealloc-fifos", &bsm->prealloc_fifos))
+ else if (unformat (input, "prealloc-fifos %d", &bsm->prealloc_fifos))
+ ;
+ else if (unformat (input, "private-segment-count %d",
+ &bsm->private_segment_count))
;
+ else if (unformat (input, "private-segment-size %dm", &tmp))
+ bsm->private_segment_size = tmp << 20;
+ else if (unformat (input, "private-segment-size %dg", &tmp))
+ bsm->private_segment_size = tmp << 30;
+ else if (unformat (input, "private-segment-size %d", &tmp))
+ bsm->private_segment_size = tmp;
else
return clib_error_return (0, "unknown input `%U'",
format_unformat_error, input);
tcp_connection_unbind (u32 listener_index)
{
tcp_main_t *tm = vnet_get_tcp_main ();
- TCP_EVT_DBG (TCP_EVT_UNBIND,
- pool_elt_at_index (tm->listener_pool, listener_index));
+ tcp_connection_t *tc;
+
+ tc = pool_elt_at_index (tm->listener_pool, listener_index);
+
+ TCP_EVT_DBG (TCP_EVT_UNBIND, tc);
+
+ /* Poison the entry */
+ if (CLIB_DEBUG > 0)
+ memset (tc, 0xFA, sizeof (*tc));
+
pool_put_index (tm->listener_pool, listener_index);
}
/* Check if half-open */
if (tc->state == TCP_STATE_SYN_SENT)
- pool_put (tm->half_open_connections, tc);
+ {
+ /* Poison the entry */
+ if (CLIB_DEBUG > 0)
+ memset (tc, 0xFA, sizeof (*tc));
+ pool_put (tm->half_open_connections, tc);
+ }
else
- pool_put (tm->connections[tc->c_thread_index], tc);
+ {
+ int thread_index = tc->c_thread_index;
+ /* Poison the entry */
+ if (CLIB_DEBUG > 0)
+ memset (tc, 0xFA, sizeof (*tc));
+ pool_put (tm->connections[thread_index], tc);
+ }
}
/**
/* Make sure all timers are cleared */
tcp_connection_timers_reset (tc);
-
stream_session_reset_notify (&tc->connection);
+
+ /* Wait for cleanup from session layer but not forever */
+ tcp_timer_set (tc, TCP_TIMER_WAITCLOSE, TCP_CLEANUP_TIME);
break;
case TCP_STATE_CLOSED:
return;
}
-
}
/**
tries = max - min;
time_now = tcp_time_now ();
+ /* Only support active opens from thread 0 */
+ ASSERT (vlib_get_thread_index () == 0);
+
/* Start at random point or max */
pool_get (tm->local_endpoints, tep);
clib_memcpy (&tep->ip, ip, sizeof (*ip));
}
}
+#if 0
typedef struct ip4_tcp_hdr
{
ip4_header_t ip;
tcp_connection_stack_on_fib_entry (tc);
}
+#endif /* 0 */
/** Initialize tcp connection variables
*
tcp_init_mss (tc);
scoreboard_init (&tc->sack_sb);
tcp_cc_init (tc);
- tcp_connection_fib_attach (tc);
+ // tcp_connection_fib_attach (tc);
}
int
if (is_ip4)
{
ip4_address_t *ip4;
- ip4 = ip_interface_get_first_ip (sw_if_index, 1);
- lcl_addr.ip4.as_u32 = ip4->as_u32;
+ int index;
+ if (vec_len (tm->ip4_src_addresses))
+ {
+ index = tm->last_v4_address_rotor++;
+ if (tm->last_v4_address_rotor >= vec_len (tm->ip4_src_addresses))
+ tm->last_v4_address_rotor = 0;
+ lcl_addr.ip4.as_u32 = tm->ip4_src_addresses[index].as_u32;
+ }
+ else
+ {
+ ip4 = ip_interface_get_first_ip (sw_if_index, 1);
+ lcl_addr.ip4.as_u32 = ip4->as_u32;
+ }
}
else
{
ip6_address_t *ip6;
- ip6 = ip_interface_get_first_ip (sw_if_index, 0);
- clib_memcpy (&lcl_addr.ip6, ip6, sizeof (*ip6));
+ int index;
+
+ if (vec_len (tm->ip6_src_addresses))
+ {
+ index = tm->last_v6_address_rotor++;
+ if (tm->last_v6_address_rotor >= vec_len (tm->ip6_src_addresses))
+ tm->last_v6_address_rotor = 0;
+ clib_memcpy (&lcl_addr.ip6, &tm->ip6_src_addresses[index],
+ sizeof (*ip6));
+ }
+ else
+ {
+ ip6 = ip_interface_get_first_ip (sw_if_index, 0);
+ clib_memcpy (&lcl_addr.ip6, ip6, sizeof (*ip6));
+ }
}
/* Allocate source port */
format_tcp_vars (u8 * s, va_list * args)
{
tcp_connection_t *tc = va_arg (*args, tcp_connection_t *);
- s = format (s, " snd_una %u snd_nxt %u snd_una_max %u\n",
+ s = format (s, " snd_una %u snd_nxt %u snd_una_max %u",
tc->snd_una - tc->iss, tc->snd_nxt - tc->iss,
tc->snd_una_max - tc->iss);
s = format (s, " rcv_nxt %u rcv_las %u\n",
s = format (s, " cong %U ", format_tcp_congestion_status, tc);
s = format (s, "cwnd %u ssthresh %u rtx_bytes %u bytes_acked %u\n",
tc->cwnd, tc->ssthresh, tc->snd_rxt_bytes, tc->bytes_acked);
- s = format (s, " prev_ssthresh %u snd_congestion %u dupack %u\n",
+ s = format (s, " prev_ssthresh %u snd_congestion %u dupack %u",
tc->prev_ssthresh, tc->snd_congestion - tc->iss,
tc->rcv_dupacks);
+ s = format (s, " limited_transmit %u\n", tc->limited_transmit - tc->iss);
+ s = format (s, " tsecr %u tsecr_last_ack %u\n", tc->rcv_opts.tsecr,
+ tc->tsecr_last_ack);
s = format (s, " rto %u rto_boff %u srtt %u rttvar %u rtt_ts %u ", tc->rto,
tc->rto_boff, tc->srtt, tc->rttvar, tc->rtt_ts);
s = format (s, "rtt_seq %u\n", tc->rtt_seq);
+ s = format (s, " tsval_recent %u tsval_recent_age %u\n", tc->tsval_recent,
+ tcp_time_now () - tc->tsval_recent_age);
s = format (s, " scoreboard: %U\n", format_tcp_scoreboard, &tc->sack_sb);
if (vec_len (tc->snd_sacks))
s = format (s, " sacks tx: %U\n", format_tcp_sacks, tc);
tcp_connection_t *tc = va_arg (*args, tcp_connection_t *);
sack_block_t *sacks = tc->snd_sacks;
sack_block_t *block;
- vec_foreach (block, sacks)
- {
- s = format (s, " start %u end %u\n", block->start - tc->irs,
- block->end - tc->irs);
- }
+ int i, len = 0;
+
+ len = vec_len (sacks);
+ for (i = 0; i < len - 1; i++)
+ {
+ block = &sacks[i];
+ s = format (s, " start %u end %u\n", block->start - tc->irs,
+ block->end - tc->irs);
+ }
+ if (len)
+ {
+ block = &sacks[len - 1];
+ s = format (s, " start %u end %u", block->start - tc->irs,
+ block->end - tc->irs);
+ }
return s;
}
always_inline u32
tcp_round_snd_space (tcp_connection_t * tc, u32 snd_space)
{
- if (tc->snd_wnd < tc->snd_mss)
+ if (PREDICT_FALSE (tc->snd_wnd < tc->snd_mss))
{
return tc->snd_wnd <= snd_space ? tc->snd_wnd : 0;
}
/* If we can't write at least a segment, don't try at all */
- if (snd_space < tc->snd_mss)
- return 0;
+ if (PREDICT_FALSE (snd_space < tc->snd_mss))
+ {
+ if (snd_space > clib_min (tc->mss, tc->rcv_opts.mss) - TCP_HDR_LEN_MAX)
+ return snd_space;
+ return 0;
+ }
/* round down to mss multiple */
return snd_space - (snd_space % tc->snd_mss);
vlib_thread_main_t *vtm = vlib_get_thread_main ();
clib_error_t *error = 0;
u32 num_threads;
+ int thread, i;
+ tcp_connection_t *tc __attribute__ ((unused));
if ((error = vlib_call_init_function (vm, ip_main_init)))
return error;
num_threads = 1 /* main thread */ + vtm->n_threads;
vec_validate (tm->connections, num_threads - 1);
+ /*
+ * Preallocate connections
+ */
+ for (thread = 0; thread < num_threads; thread++)
+ {
+ for (i = 0; i < tm->preallocated_connections; i++)
+ pool_get (tm->connections[thread], tc);
+
+ for (i = 0; i < tm->preallocated_connections; i++)
+ pool_put_index (tm->connections[thread], i);
+ }
+
+ /*
+ * Preallocate half-open connections
+ */
+ for (i = 0; i < tm->preallocated_half_open_connections; i++)
+ pool_get (tm->half_open_connections, tc);
+
+ for (i = 0; i < tm->preallocated_half_open_connections; i++)
+ pool_put_index (tm->half_open_connections, i);
+
/* Initialize per worker thread tx buffers (used for control messages) */
vec_validate (tm->tx_buffers, num_threads - 1);
{
tcp_main_t *tm = vnet_get_tcp_main ();
- tm->vlib_main = vm;
tm->vnet_main = vnet_get_main ();
tm->is_enabled = 0;
VLIB_INIT_FUNCTION (tcp_init);
+
+static clib_error_t *
+tcp_config_fn (vlib_main_t * vm, unformat_input_t * input)
+{
+ tcp_main_t *tm = vnet_get_tcp_main ();
+
+ while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
+ {
+ if (unformat
+ (input, "preallocated-connections %d",
+ &tm->preallocated_connections))
+ ;
+ else if (unformat (input, "preallocated-half-open-connections %d",
+ &tm->preallocated_half_open_connections))
+ ;
+ else
+ return clib_error_return (0, "unknown input `%U'",
+ format_unformat_error, input);
+ }
+ return 0;
+}
+
+VLIB_CONFIG_FUNCTION (tcp_config_fn, "tcp");
+
+static clib_error_t *
+tcp_src_address (vlib_main_t * vm,
+ unformat_input_t * input, vlib_cli_command_t * cmd_arg)
+{
+ tcp_main_t *tm = vnet_get_tcp_main ();
+ ip4_address_t v4start, v4end;
+ ip6_address_t v6start, v6end;
+ int v4set = 0;
+ int v6set = 0;
+
+ while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
+ {
+ if (unformat (input, "%U - %U", unformat_ip4_address, &v4start,
+ unformat_ip4_address, &v4end))
+ v4set = 1;
+ else if (unformat (input, "%U", unformat_ip4_address, &v4start))
+ {
+ memcpy (&v4end, &v4start, sizeof (v4start));
+ v4set = 1;
+ }
+ else if (unformat (input, "%U - %U", unformat_ip6_address, &v6start,
+ unformat_ip4_address, &v6end))
+ v6set = 1;
+ else if (unformat (input, "%U", unformat_ip6_address, &v6start))
+ {
+ memcpy (&v6end, &v6start, sizeof (v4start));
+ v6set = 1;
+ }
+ else
+ break;
+ }
+
+ if (!v4set && !v6set)
+ return clib_error_return (0, "at least one v4 or v6 address required");
+
+ if (v4set)
+ {
+ u32 tmp;
+
+ do
+ {
+ vec_add1 (tm->ip4_src_addresses, v4start);
+ tmp = clib_net_to_host_u32 (v4start.as_u32);
+ tmp++;
+ v4start.as_u32 = clib_host_to_net_u32 (tmp);
+ }
+ while (clib_host_to_net_u32 (v4start.as_u32) <=
+ clib_host_to_net_u32 (v4end.as_u32));
+ }
+ if (v6set)
+ {
+ clib_warning ("v6 src address list unimplemented...");
+ }
+ return 0;
+}
+
+/* *INDENT-OFF* */
+VLIB_CLI_COMMAND (tcp_src_address_command, static) =
+{
+ .path = "tcp src-address",
+ .short_help = "tcp src-address <ip-addr> [- <ip-addr>] add src address range",
+ .function = tcp_src_address,
+};
+/* *INDENT-ON* */
+
+
+
/*
* fd.io coding-style-patch-verification: ON
*
/* Flag that indicates if stack is on or off */
u8 is_enabled;
+ /** Number of preallocated connections */
+ u32 preallocated_connections;
+ u32 preallocated_half_open_connections;
+
+ /** Vectors of src addresses. Optional unless one needs > 63K active-opens */
+ ip4_address_t *ip4_src_addresses;
+ u32 last_v4_address_rotor;
+ u32 last_v6_address_rotor;
+ ip6_address_t *ip6_src_addresses;
+
/* convenience */
vlib_main_t *vlib_main;
vnet_main_t *vnet_main;
always_inline void
tcp_timer_set (tcp_connection_t * tc, u8 timer_id, u32 interval)
{
+ ASSERT (tc->c_thread_index == vlib_get_thread_index ());
tc->timers[timer_id]
= tw_timer_start_16t_2w_512sl (&tcp_main.timer_wheels[tc->c_thread_index],
tc->c_c_index, timer_id, interval);
always_inline void
tcp_timer_reset (tcp_connection_t * tc, u8 timer_id)
{
+ ASSERT (tc->c_thread_index == vlib_get_thread_index ());
if (tc->timers[timer_id] == TCP_TIMER_HANDLE_INVALID)
return;
always_inline void
tcp_timer_update (tcp_connection_t * tc, u8 timer_id, u32 interval)
{
+ ASSERT (tc->c_thread_index == vlib_get_thread_index ());
if (tc->timers[timer_id] != TCP_TIMER_HANDLE_INVALID)
tw_timer_stop_16t_2w_512sl (&tcp_main.timer_wheels[tc->c_thread_index],
tc->timers[timer_id]);
"establish", \
}, \
}; \
- DECLARE_ETD(_tc, _e, 2); \
- ed->data[0] = _timer_id; \
- ed->data[1] = _timer_id; \
+ if (_tc) \
+ { \
+ DECLARE_ETD(_tc, _e, 2); \
+ ed->data[0] = _timer_id; \
+ ed->data[1] = _timer_id; \
+ } \
+ else \
+ { \
+ clib_warning ("pop for unexisting connection %d", _tc_index); \
+ } \
}
#define TCP_EVT_SEG_INVALID_HANDLER(_tc, _seq, _end, ...) \
if (tcp_opts_tstamp (&tc->rcv_opts) && tc->tsval_recent
&& seq_leq (seq, tc->rcv_las) && seq_leq (tc->rcv_las, seq_end))
{
+ ASSERT (timestamp_leq (tc->tsval_recent, tc->rcv_opts.tsval));
tc->tsval_recent = tc->rcv_opts.tsval;
tc->tsval_recent_age = tcp_time_now ();
}
if (tc->srtt != 0)
{
err = mrtt - tc->srtt;
-// tc->srtt += err >> 3;
/* XXX Drop in RTT results in RTTVAR increase and bigger RTO.
* The increase should be bound */
-// tc->rttvar += ((int) clib_abs (err) - (int) tc->rttvar) >> 2;
-
tc->srtt = clib_max ((int) tc->srtt + (err >> 3), 1);
diff = (clib_abs (err) - (int) tc->rttvar) >> 2;
tc->rttvar = clib_max ((int) tc->rttvar + diff, 1);
&& (prev_snd_wnd == tc->snd_wnd));
}
+static u8
+tcp_is_lost_fin (tcp_connection_t * tc)
+{
+ if ((tc->flags & TCP_CONN_FINSNT) && tc->snd_una_max - tc->snd_una == 1)
+ return 1;
+ return 0;
+}
+
/**
* Checks if ack is a congestion control event.
*/
*is_dack = tc->sack_sb.last_sacked_bytes
|| tcp_ack_is_dupack (tc, b, prev_snd_wnd, prev_snd_una);
- return (*is_dack || tcp_in_cong_recovery (tc));
+ return ((*is_dack || tcp_in_cong_recovery (tc)) && !tcp_is_lost_fin (tc));
}
void
* last hole end */
tmp = tc->rcv_opts.sacks[vec_len (tc->rcv_opts.sacks) - 1];
last_hole = scoreboard_last_hole (sb);
- if (seq_gt (tc->snd_una_max, sb->high_sacked)
- && seq_gt (tc->snd_una_max, last_hole->end))
- last_hole->end = tc->snd_una_max;
- /* keep track of max byte sacked for when the last hole
+ if (seq_gt (tc->snd_una_max, last_hole->end))
+ {
+ if (seq_geq (last_hole->start, sb->high_sacked))
+ {
+ last_hole->end = tc->snd_una_max;
+ }
+ /* New hole after high sacked block */
+ else if (seq_lt (sb->high_sacked, tc->snd_una_max))
+ {
+ scoreboard_insert_hole (sb, sb->tail, sb->high_sacked,
+ tc->snd_una_max);
+ }
+ }
+ /* Keep track of max byte sacked for when the last hole
* is acked */
if (seq_gt (tmp.end, sb->high_sacked))
sb->high_sacked = tmp.end;
while (hole && blk_index < vec_len (tc->rcv_opts.sacks))
{
blk = &tc->rcv_opts.sacks[blk_index];
-
if (seq_leq (blk->start, hole->start))
{
/* Block covers hole. Remove hole */
}
else if (!next_hole)
{
+ ASSERT (seq_geq (sb->high_sacked, ack));
sb->snd_una_adv = sb->high_sacked - ack;
sb->last_bytes_delivered += sb->high_sacked - hole->end;
}
{
hole->end = blk->start;
}
-
hole = scoreboard_next_hole (sb, hole);
}
}
scoreboard_update_bytes (tc, sb);
sb->last_sacked_bytes = sb->sacked_bytes
- (old_sacked_bytes - sb->last_bytes_delivered);
+ ASSERT (sb->last_sacked_bytes <= sb->sacked_bytes);
ASSERT (sb->sacked_bytes == 0
|| sb->sacked_bytes < tc->snd_una_max - seq_max (tc->snd_una, ack));
ASSERT (sb->last_sacked_bytes + sb->lost_bytes <= tc->snd_una_max
- seq_max (tc->snd_una, ack));
+ ASSERT (sb->head == TCP_INVALID_SACK_HOLE_INDEX || tcp_in_recovery (tc)
+ || sb->holes[sb->head].start == ack + sb->snd_una_adv);
}
/**
static u8
tcp_cc_is_spurious_retransmit (tcp_connection_t * tc)
{
- return (tc->snd_rxt_ts
+ return (tcp_in_recovery (tc)
+ && tc->snd_rxt_ts
&& tcp_opts_tstamp (&tc->rcv_opts)
&& timestamp_lt (tc->rcv_opts.tsecr, tc->snd_rxt_ts));
}
{
ASSERT (tc->snd_una != tc->snd_una_max
|| tc->sack_sb.last_sacked_bytes);
+
tc->rcv_dupacks++;
if (tc->rcv_dupacks > TCP_DUPACK_THRESHOLD && !tc->bytes_acked)
goto partial_ack_test;
}
- /* If of of the two conditions lower hold, reset dupacks
- * 1) Cumulative ack does not cover more than congestion threshold,
- * and the following doesn't hold: the congestion window is
- * greater than SMSS bytes and the difference between highest_ack
- * and prev_highest_ack is at most 4*SMSS bytes (XXX)
- * 2) RFC6582 heuristic to avoid multiple fast retransmits
+ /* If of of the two conditions lower hold, reset dupacks because
+ * we're probably after timeout (RFC6582 heuristics).
+ * If Cumulative ack does not cover more than congestion threshold,
+ * and:
+ * 1) The following doesn't hold: The congestion window is greater
+ * than SMSS bytes and the difference between highest_ack
+ * and prev_highest_ack is at most 4*SMSS bytes
+ * 2) Echoed timestamp in the last non-dup ack does not equal the
+ * stored timestamp
*/
- if ((seq_gt (tc->snd_una, tc->snd_congestion)
- || !(tc->cwnd > tc->snd_mss
- && tc->bytes_acked <= 4 * tc->snd_mss))
- || tc->rcv_opts.tsecr != tc->tsecr_last_ack)
+ if (seq_leq (tc->snd_una, tc->snd_congestion)
+ && ((!(tc->cwnd > tc->snd_mss
+ && tc->bytes_acked <= 4 * tc->snd_mss))
+ || (tc->rcv_opts.tsecr != tc->tsecr_last_ack)))
{
tc->rcv_dupacks = 0;
return;
* three segments that have left the network and should've been
* buffered at the receiver XXX */
tc->cwnd = tc->ssthresh + tc->rcv_dupacks * tc->snd_mss;
+ ASSERT (tc->cwnd >= tc->snd_mss);
/* If cwnd allows, send more data */
if (tcp_opts_sack_permitted (&tc->rcv_opts)
>= tc->sack_sb.last_bytes_delivered);
rxt_delivered = tc->bytes_acked + tc->sack_sb.snd_una_adv
- tc->sack_sb.last_bytes_delivered;
- if (rxt_delivered && seq_gt (tc->sack_sb.high_rxt, tc->snd_una))
+ if (0 && rxt_delivered && seq_gt (tc->sack_sb.high_rxt, tc->snd_una))
{
/* If we have sacks and we haven't gotten an ack beyond high_rxt,
* remove sacked bytes delivered */
{
int written;
+ ASSERT (seq_geq (vnet_buffer (b)->tcp.seq_number, tc->rcv_nxt));
+
/* Pure ACK. Update rcv_nxt and be done. */
if (PREDICT_FALSE (data_len == 0))
{
/* Chop off the bytes in the past */
n_bytes_to_drop = tc->rcv_nxt - vnet_buffer (b)->tcp.seq_number;
n_data_bytes -= n_bytes_to_drop;
+ vnet_buffer (b)->tcp.seq_number = tc->rcv_nxt;
vlib_buffer_advance (b, n_bytes_to_drop);
goto in_order;
goto drop;
}
- stream_session_init_fifos_pointers (&new_tc0->connection,
- new_tc0->irs + 1,
- new_tc0->iss + 1);
/* Make sure after data segment processing ACK is sent */
new_tc0->flags |= TCP_CONN_SNDACK;
+
+ /* Update rtt with the syn-ack sample */
+ new_tc0->bytes_acked = 1;
+ tcp_update_rtt (new_tc0, vnet_buffer (b0)->tcp.ack_number);
}
/* SYN: Simultaneous open. Change state to SYN-RCVD and send SYN-ACK */
else
goto drop;
}
- stream_session_init_fifos_pointers (&new_tc0->connection,
- new_tc0->irs + 1,
- new_tc0->iss + 1);
+ tc0->rtt_ts = 0;
+
tcp_make_synack (new_tc0, b0);
next0 = tcp_next_output (is_ip4);
<< tc0->rcv_opts.wscale;
tc0->snd_wl1 = vnet_buffer (b0)->tcp.seq_number;
tc0->snd_wl2 = vnet_buffer (b0)->tcp.ack_number;
-
- /* Shoulder tap the server */
stream_session_accept_notify (&tc0->connection);
/* Reset SYN-ACK retransmit timer */
/* If FIN is ACKed */
if (tc0->snd_una == tc0->snd_una_max)
{
+ ASSERT (tcp_fin (tcp0));
tc0->state = TCP_STATE_FIN_WAIT_2;
/* Stop all timers, 2MSL will be set lower */
tcp_connection_timers_reset (tc0);
tcp_make_synack (child0, b0);
next0 = tcp_next_output (is_ip4);
- /* Init fifo pointers after we have iss */
- stream_session_init_fifos_pointers (&child0->connection,
- child0->irs + 1,
- child0->iss + 1);
drop:
if (PREDICT_FALSE (b0->flags & VLIB_BUFFER_IS_TRACED))
{
_(LISTEN, TCP_FLAG_SYN, TCP_INPUT_NEXT_LISTEN, TCP_ERROR_NONE);
_(LISTEN, TCP_FLAG_ACK, TCP_INPUT_NEXT_RESET, TCP_ERROR_NONE);
_(LISTEN, TCP_FLAG_RST, TCP_INPUT_NEXT_DROP, TCP_ERROR_NONE);
+ _(LISTEN, TCP_FLAG_FIN | TCP_FLAG_ACK, TCP_INPUT_NEXT_RESET,
+ TCP_ERROR_NONE);
/* ACK for for a SYN-ACK -> tcp-rcv-process. */
_(SYN_RCVD, TCP_FLAG_ACK, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
_(SYN_RCVD, TCP_FLAG_RST, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
+ _(SYN_RCVD, TCP_FLAG_SYN, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
/* SYN-ACK for a SYN */
_(SYN_SENT, TCP_FLAG_SYN | TCP_FLAG_ACK, TCP_INPUT_NEXT_SYN_SENT,
TCP_ERROR_NONE);
_(ESTABLISHED, TCP_FLAG_RST, TCP_INPUT_NEXT_ESTABLISHED, TCP_ERROR_NONE);
_(ESTABLISHED, TCP_FLAG_RST | TCP_FLAG_ACK, TCP_INPUT_NEXT_ESTABLISHED,
TCP_ERROR_NONE);
+ _(ESTABLISHED, TCP_FLAG_SYN, TCP_INPUT_NEXT_ESTABLISHED, TCP_ERROR_NONE);
/* ACK or FIN-ACK to our FIN */
_(FIN_WAIT_1, TCP_FLAG_ACK, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
_(FIN_WAIT_1, TCP_FLAG_ACK | TCP_FLAG_FIN, TCP_INPUT_NEXT_RCV_PROCESS,
TCP_ERROR_NONE);
/* FIN in reply to our FIN from the other side */
_(FIN_WAIT_1, TCP_FLAG_FIN, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
+ _(FIN_WAIT_1, TCP_FLAG_RST, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
/* FIN confirming that the peer (app) has closed */
_(FIN_WAIT_2, TCP_FLAG_FIN, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
_(FIN_WAIT_2, TCP_FLAG_ACK, TCP_INPUT_NEXT_RCV_PROCESS, TCP_ERROR_NONE);
TCP_ERROR_NONE);
_(CLOSED, TCP_FLAG_ACK, TCP_INPUT_NEXT_RESET, TCP_ERROR_CONNECTION_CLOSED);
_(CLOSED, TCP_FLAG_RST, TCP_INPUT_NEXT_DROP, TCP_ERROR_CONNECTION_CLOSED);
+ _(CLOSED, TCP_FLAG_FIN | TCP_FLAG_ACK, TCP_INPUT_NEXT_RESET,
+ TCP_ERROR_CONNECTION_CLOSED);
#undef _
}
* window deflation" attempts to ensure that, when fast recovery
* eventually ends, approximately ssthresh amount of data will be
* outstanding in the network.*/
- tc->cwnd = (tc->cwnd > tc->bytes_acked) ?
- tc->cwnd - tc->bytes_acked : 0;
+ tc->cwnd = (tc->cwnd > tc->bytes_acked + tc->snd_mss) ?
+ tc->cwnd - tc->bytes_acked : tc->snd_mss;
if (tc->bytes_acked > tc->snd_mss)
tc->cwnd += tc->snd_mss;
}
vlib_node_registration_t tcp4_output_node;
vlib_node_registration_t tcp6_output_node;
-typedef enum _tcp_output_nect
+typedef enum _tcp_output_next
{
TCP_OUTPUT_NEXT_DROP,
+ TCP_OUTPUT_NEXT_IP_LOOKUP,
TCP_OUTPUT_N_NEXT
} tcp_output_next_t;
#define foreach_tcp4_output_next \
_ (DROP, "error-drop") \
+ _ (IP_LOOKUP, "ip4-lookup")
#define foreach_tcp6_output_next \
_ (DROP, "error-drop") \
+ _ (IP_LOOKUP, "ip6-lookup")
static char *tcp_error_strings[] = {
#define tcp_error(n,s) s,
#define tcp_get_free_buffer_index(tm, bidx) \
do { \
u32 *my_tx_buffers, n_free_buffers; \
- u32 thread_index = vlib_get_thread_index(); \
- my_tx_buffers = tm->tx_buffers[thread_index]; \
+ u32 thread_index = vlib_get_thread_index(); \
+ my_tx_buffers = tm->tx_buffers[thread_index]; \
if (PREDICT_FALSE(vec_len (my_tx_buffers) == 0)) \
{ \
n_free_buffers = 32; /* TODO config or macro */ \
vec_validate (my_tx_buffers, n_free_buffers - 1); \
_vec_len(my_tx_buffers) = vlib_buffer_alloc_from_free_list ( \
- tm->vlib_main, my_tx_buffers, n_free_buffers, \
+ vlib_get_main(), my_tx_buffers, n_free_buffers, \
VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX); \
- tm->tx_buffers[thread_index] = my_tx_buffers; \
+ tm->tx_buffers[thread_index] = my_tx_buffers; \
} \
/* buffer shortage */ \
if (PREDICT_FALSE (vec_len (my_tx_buffers) == 0)) \
_vec_len (my_tx_buffers) -= 1; \
} while (0)
-#define tcp_return_buffer(tm) \
-do { \
- u32 *my_tx_buffers; \
- u32 thread_index = vlib_get_thread_index(); \
- my_tx_buffers = tm->tx_buffers[thread_index]; \
- _vec_len (my_tx_buffers) +=1; \
+#define tcp_return_buffer(tm) \
+do { \
+ u32 *my_tx_buffers; \
+ u32 thread_index = vlib_get_thread_index(); \
+ my_tx_buffers = tm->tx_buffers[thread_index]; \
+ _vec_len (my_tx_buffers) +=1; \
} while (0)
always_inline void
tcp_push_ip_hdr (tcp_main_t * tm, tcp_connection_t * tc, vlib_buffer_t * b)
{
tcp_header_t *th = vlib_buffer_get_current (b);
-
+ vlib_main_t *vm = vlib_get_main ();
if (tc->c_is_ip4)
{
ip4_header_t *ih;
- ih = vlib_buffer_push_ip4 (tm->vlib_main, b, &tc->c_lcl_ip4,
+ ih = vlib_buffer_push_ip4 (vm, b, &tc->c_lcl_ip4,
&tc->c_rmt_ip4, IP_PROTOCOL_TCP);
- th->checksum = ip4_tcp_udp_compute_checksum (tm->vlib_main, b, ih);
+ th->checksum = ip4_tcp_udp_compute_checksum (vm, b, ih);
}
else
{
ip6_header_t *ih;
int bogus = ~0;
- ih = vlib_buffer_push_ip6 (tm->vlib_main, b, &tc->c_lcl_ip6,
+ ih = vlib_buffer_push_ip6 (vm, b, &tc->c_lcl_ip6,
&tc->c_rmt_ip6, IP_PROTOCOL_TCP);
- th->checksum = ip6_tcp_udp_icmp_compute_checksum (tm->vlib_main, b, ih,
- &bogus);
+ th->checksum = ip6_tcp_udp_icmp_compute_checksum (vm, b, ih, &bogus);
ASSERT (!bogus);
}
}
/* Decide where to send the packet */
next_index = is_ip4 ? tcp4_output_node.index : tcp6_output_node.index;
+ /* Initialize the trajectory trace, if configured */
+ if (VLIB_BUFFER_TRACE_TRAJECTORY > 0)
+ {
+ b->pre_data[0] = 1;
+ b->pre_data[1] = next_index;
+ }
+
/* Enqueue the packet */
f = vlib_get_frame_to_node (vm, next_index);
to_next = vlib_frame_vector_args (f);
/* Account for the SYN */
tc->snd_nxt += 1;
+ tc->rtt_ts = 0;
}
else
{
/* Nothing to send */
if (n_bytes <= 0)
{
- clib_warning ("persist found nothing to send");
+ // clib_warning ("persist found nothing to send");
tcp_return_buffer (tm);
return;
}
tcp_connection_t *tc0;
tcp_tx_trace_t *t0;
tcp_header_t *th0 = 0;
- u32 error0 = TCP_ERROR_PKTS_SENT, next0 = TCP_OUTPUT_NEXT_DROP;
+ u32 error0 = TCP_ERROR_PKTS_SENT, next0 = TCP_OUTPUT_NEXT_IP_LOOKUP;
bi0 = from[0];
to_next[0] = bi0;
tc0->rto_boff = 0;
}
+#if 0
/* Make sure we haven't lost route to our peer */
if (PREDICT_FALSE (tc0->last_fib_check
< tc0->snd_opts.tsval + TCP_FIB_RECHECK_PERIOD))
/* Use pre-computed dpo to set next node */
next0 = tc0->c_rmt_dpo.dpoi_next_node;
vnet_buffer (b0)->ip.adj_index[VLIB_TX] = tc0->c_rmt_dpo.dpoi_index;
+#endif
+
+ vnet_buffer (b0)->sw_if_index[VLIB_RX] = 0;
+ vnet_buffer (b0)->sw_if_index[VLIB_TX] = ~0;
b0->flags |= VNET_BUFFER_LOCALLY_ORIGINATED;
done:
#define TCP_OPTION_LEN_TIMESTAMP 10
#define TCP_OPTION_LEN_SACK_BLOCK 8
+#define TCP_HDR_LEN_MAX 60
#define TCP_WND_MAX 65535U
#define TCP_MAX_WND_SCALE 14 /* See RFC 1323 */
#define TCP_OPTS_ALIGN 4
{
tcp_connection_t _tc, *tc = &_tc;
sack_block_t *sacks;
- int i, verbose = 0;
+ int i, verbose = 0, expected;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
sacks = vec_dup (tc->snd_sacks);
tcp_update_sack_list (tc, 1100, 1200);
- TCP_TEST ((vec_len (tc->snd_sacks) == 5), "sack blocks %d expected %d",
- vec_len (tc->snd_sacks), 5);
+ if (verbose)
+ vlib_cli_output (vm, "add new segment [1100, 1200]\n%U",
+ format_tcp_sacks, tc);
+ expected = 5 < TCP_MAX_SACK_BLOCKS ? 6 : 5;
+ TCP_TEST ((vec_len (tc->snd_sacks) == expected),
+ "sack blocks %d expected %d", vec_len (tc->snd_sacks), expected);
TCP_TEST ((tc->snd_sacks[0].start == 1100),
"first sack block start %u expected %u", tc->snd_sacks[0].start,
1100);
/* lookup session */
s0 = stream_session_lookup4 (&ip0->dst_address, &ip0->src_address,
udp0->dst_port, udp0->src_port,
- SESSION_TYPE_IP4_UDP, my_thread_index);
+ SESSION_TYPE_IP4_UDP);
/* no listener */
if (PREDICT_FALSE (s0 == 0))
Code Review / vpp.git / commitdiff