X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvnet%2Fsession%2Fsession.c;h=ee22ccbe0adeba5f6a672d578669ec0a3c90ff3c;hb=c87c91d8b0e85997debaf575f2e30cc2702edf25;hp=c5aaf2e2b15a1385a3021c789c947a0b45276c99;hpb=acd2a6a32cebab70a054e9335e446798538d0524;p=vpp.git diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c index c5aaf2e2b15..ee22ccbe0ad 100644 --- a/src/vnet/session/session.c +++ b/src/vnet/session/session.c @@ -18,368 +18,19 @@ */ #include +#include +#include #include #include #include -#include #include -#include - -/** - * Per-type vector of transport protocol virtual function tables - */ -static transport_proto_vft_t *tp_vfts; session_manager_main_t session_manager_main; - -/* - * Session lookup key; (src-ip, dst-ip, src-port, dst-port, session-type) - * Value: (owner thread index << 32 | session_index); - */ -void -stream_session_table_add_for_tc (transport_connection_t * tc, u64 value) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv4_t kv4; - session_kv6_t kv6; - - switch (tc->proto) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - make_v4_ss_kv_from_tc (&kv4, tc); - kv4.value = value; - clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4, 1 /* is_add */ ); - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - make_v6_ss_kv_from_tc (&kv6, tc); - kv6.value = value; - clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6, 1 /* is_add */ ); - break; - default: - clib_warning ("Session type not supported"); - ASSERT (0); - } -} - -void -stream_session_table_add (session_manager_main_t * smm, stream_session_t * s, - u64 value) -{ - transport_connection_t *tc; - - tc = tp_vfts[s->session_type].get_connection (s->connection_index, - s->thread_index); - stream_session_table_add_for_tc (tc, value); -} - -static void -stream_session_half_open_table_add (session_type_t sst, - transport_connection_t * tc, u64 value) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv4_t kv4; - session_kv6_t kv6; - - switch (sst) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - make_v4_ss_kv_from_tc (&kv4, tc); - kv4.value = value; - clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4, - 1 /* is_add */ ); - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - make_v6_ss_kv_from_tc (&kv6, tc); - kv6.value = value; - clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6, - 1 /* is_add */ ); - break; - default: - clib_warning ("Session type not supported"); - ASSERT (0); - } -} - -int -stream_session_table_del_for_tc (transport_connection_t * tc) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv4_t kv4; - session_kv6_t kv6; - switch (tc->proto) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - make_v4_ss_kv_from_tc (&kv4, tc); - return clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4, - 0 /* is_add */ ); - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - make_v6_ss_kv_from_tc (&kv6, tc); - return clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6, - 0 /* is_add */ ); - break; - default: - clib_warning ("Session type not supported"); - ASSERT (0); - } - - return 0; -} - -static int -stream_session_table_del (session_manager_main_t * smm, stream_session_t * s) -{ - transport_connection_t *ts; - - ts = tp_vfts[s->session_type].get_connection (s->connection_index, - s->thread_index); - return stream_session_table_del_for_tc (ts); -} - -static void -stream_session_half_open_table_del (session_manager_main_t * smm, u8 sst, - transport_connection_t * tc) -{ - session_kv4_t kv4; - session_kv6_t kv6; - - switch (sst) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - make_v4_ss_kv_from_tc (&kv4, tc); - clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4, - 0 /* is_add */ ); - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - make_v6_ss_kv_from_tc (&kv6, tc); - clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6, - 0 /* is_add */ ); - break; - default: - clib_warning ("Session type not supported"); - ASSERT (0); - } -} - -stream_session_t * -stream_session_lookup_listener4 (ip4_address_t * lcl, u16 lcl_port, u8 proto) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv4_t kv4; - int rv; - - make_v4_listener_kv (&kv4, lcl, lcl_port, proto); - rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4); - if (rv == 0) - return pool_elt_at_index (smm->listen_sessions[proto], (u32) kv4.value); - - /* Zero out the lcl ip */ - kv4.key[0] = 0; - rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4); - if (rv == 0) - return pool_elt_at_index (smm->listen_sessions[proto], kv4.value); - - return 0; -} - -/** Looks up a session based on the 5-tuple passed as argument. - * - * First it tries to find an established session, if this fails, it tries - * finding a listener session if this fails, it tries a lookup with a - * wildcarded local source (listener bound to all interfaces) - */ -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) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv4_t kv4; - 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); - - /* If nothing is found, check if any listener is available */ - return stream_session_lookup_listener4 (lcl, lcl_port, proto); -} - -stream_session_t * -stream_session_lookup_listener6 (ip6_address_t * lcl, u16 lcl_port, u8 proto) -{ - session_manager_main_t *smm = &session_manager_main; - session_kv6_t kv6; - int rv; - - make_v6_listener_kv (&kv6, lcl, lcl_port, proto); - rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6); - if (rv == 0) - return pool_elt_at_index (smm->listen_sessions[proto], kv6.value); - - /* Zero out the lcl ip */ - kv6.key[0] = kv6.key[1] = 0; - rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6); - if (rv == 0) - return pool_elt_at_index (smm->listen_sessions[proto], kv6.value); - - return 0; -} - -/* Looks up a session based on the 5-tuple passed as argument. - * First it tries to find an established session, if this fails, it tries - * finding a listener session if this fails, it tries a lookup with a - * 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) -{ - session_manager_main_t *smm = vnet_get_session_manager_main (); - session_kv6_t kv6; - 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); - - /* If nothing is found, check if any listener is available */ - return stream_session_lookup_listener6 (lcl, lcl_port, proto); -} - -stream_session_t * -stream_session_lookup_listener (ip46_address_t * lcl, u16 lcl_port, u8 proto) -{ - switch (proto) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - return stream_session_lookup_listener4 (&lcl->ip4, lcl_port, proto); - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - return stream_session_lookup_listener6 (&lcl->ip6, lcl_port, proto); - break; - } - return 0; -} - -static u64 -stream_session_half_open_lookup (session_manager_main_t * smm, - ip46_address_t * lcl, ip46_address_t * rmt, - u16 lcl_port, u16 rmt_port, u8 proto) -{ - session_kv4_t kv4; - session_kv6_t kv6; - int rv; - - switch (proto) - { - case SESSION_TYPE_IP4_UDP: - case SESSION_TYPE_IP4_TCP: - make_v4_ss_kv (&kv4, &lcl->ip4, &rmt->ip4, lcl_port, rmt_port, proto); - rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4); - - if (rv == 0) - return kv4.value; - - return (u64) ~ 0; - break; - case SESSION_TYPE_IP6_UDP: - case SESSION_TYPE_IP6_TCP: - make_v6_ss_kv (&kv6, &lcl->ip6, &rmt->ip6, lcl_port, rmt_port, proto); - rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6); - - if (rv == 0) - return kv6.value; - - return (u64) ~ 0; - break; - } - return 0; -} - -transport_connection_t * -stream_session_lookup_transport4 (ip4_address_t * lcl, ip4_address_t * rmt, - u16 lcl_port, u16 rmt_port, u8 proto, - u32 my_thread_index) -{ - 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) - { - s = stream_session_get_tsi (kv4.value, my_thread_index); - - return tp_vfts[s->session_type].get_connection (s->connection_index, - my_thread_index); - } - - /* If nothing is found, check if any listener is available */ - s = stream_session_lookup_listener4 (lcl, lcl_port, proto); - if (s) - return tp_vfts[s->session_type].get_listener (s->connection_index); - - /* Finally, try half-open connections */ - 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; -} - -transport_connection_t * -stream_session_lookup_transport6 (ip6_address_t * lcl, ip6_address_t * rmt, - u16 lcl_port, u16 rmt_port, u8 proto, - u32 my_thread_index) -{ - session_manager_main_t *smm = &session_manager_main; - stream_session_t *s; - session_kv6_t kv6; - 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) - { - s = stream_session_get_tsi (kv6.value, my_thread_index); - - return tp_vfts[s->session_type].get_connection (s->connection_index, - my_thread_index); - } - - /* If nothing is found, check if any listener is available */ - s = stream_session_lookup_listener6 (lcl, lcl_port, proto); - if (s) - return tp_vfts[s->session_type].get_listener (s->connection_index); - - /* Finally, try half-open connections */ - rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6); - if (rv == 0) - return tp_vfts[proto].get_half_open (kv6.value & 0xFFFFFFFF); - - return 0; -} +extern transport_proto_vft_t *tp_vfts; int stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc, - stream_session_t ** ret_s) + u8 alloc_fifos, stream_session_t ** ret_s) { session_manager_main_t *smm = &session_manager_main; svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0; @@ -390,30 +41,39 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc, u32 thread_index = tc->thread_index; int rv; - if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo, - &server_tx_fifo, - &fifo_segment_index))) - return rv; + ASSERT (thread_index == vlib_get_thread_index ()); /* Create the session */ pool_get_aligned (smm->sessions[thread_index], s, CLIB_CACHE_LINE_BYTES); memset (s, 0, sizeof (*s)); - - /* Initialize backpointers */ pool_index = s - smm->sessions[thread_index]; - server_rx_fifo->master_session_index = pool_index; - server_rx_fifo->master_thread_index = thread_index; - server_tx_fifo->master_session_index = pool_index; - server_tx_fifo->master_thread_index = thread_index; + /* Allocate fifos */ + if (alloc_fifos) + { + if ((rv = segment_manager_alloc_session_fifos (sm, &server_rx_fifo, + &server_tx_fifo, + &fifo_segment_index))) + { + pool_put (smm->sessions[thread_index], s); + return rv; + } + /* Initialize backpointers */ + server_rx_fifo->master_session_index = pool_index; + server_rx_fifo->master_thread_index = thread_index; - s->server_rx_fifo = server_rx_fifo; - s->server_tx_fifo = server_tx_fifo; + server_tx_fifo->master_session_index = pool_index; + server_tx_fifo->master_thread_index = thread_index; + + s->server_rx_fifo = server_rx_fifo; + s->server_tx_fifo = server_tx_fifo; + s->svm_segment_index = fifo_segment_index; + } /* Initialize state machine, such as it is... */ - s->session_type = tc->proto; + s->session_type = session_type_from_proto_and_ip (tc->transport_proto, + tc->is_ip4); s->session_state = SESSION_STATE_CONNECTING; - s->svm_segment_index = fifo_segment_index; s->thread_index = thread_index; s->session_index = pool_index; @@ -424,7 +84,7 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc, tc->s_index = s->session_index; /* Add to the main lookup table */ - value = (((u64) thread_index) << 32) | (u64) s->session_index; + value = stream_session_handle (s); stream_session_table_add_for_tc (tc, value); *ret_s = s; @@ -432,38 +92,104 @@ stream_session_create_i (segment_manager_t * sm, transport_connection_t * tc, return 0; } -/** Enqueue buffer chain tail */ +/** + * Discards bytes from buffer chain + * + * It discards n_bytes_to_drop starting at first buffer after chain_b + */ +always_inline void +session_enqueue_discard_chain_bytes (vlib_main_t * vm, vlib_buffer_t * b, + vlib_buffer_t ** chain_b, + u32 n_bytes_to_drop) +{ + vlib_buffer_t *next = *chain_b; + u32 to_drop = n_bytes_to_drop; + ASSERT (b->flags & VLIB_BUFFER_NEXT_PRESENT); + while (to_drop && (next->flags & VLIB_BUFFER_NEXT_PRESENT)) + { + next = vlib_get_buffer (vm, next->next_buffer); + if (next->current_length > to_drop) + { + vlib_buffer_advance (next, to_drop); + to_drop = 0; + } + else + { + to_drop -= next->current_length; + next->current_length = 0; + } + } + *chain_b = next; + + if (to_drop == 0) + b->total_length_not_including_first_buffer -= n_bytes_to_drop; +} + +/** + * Enqueue buffer chain tail + */ always_inline int session_enqueue_chain_tail (stream_session_t * s, vlib_buffer_t * b, u32 offset, u8 is_in_order) { vlib_buffer_t *chain_b; - u32 chain_bi = b->next_buffer; + u32 chain_bi, len, diff; vlib_main_t *vm = vlib_get_main (); - u8 *data, len; - u16 written = 0; + u8 *data; + u32 written = 0; int rv = 0; + if (is_in_order && offset) + { + diff = offset - b->current_length; + if (diff > b->total_length_not_including_first_buffer) + return 0; + chain_b = b; + session_enqueue_discard_chain_bytes (vm, b, &chain_b, diff); + chain_bi = vlib_get_buffer_index (vm, chain_b); + } + else + chain_bi = b->next_buffer; + do { chain_b = vlib_get_buffer (vm, chain_bi); data = vlib_buffer_get_current (chain_b); len = chain_b->current_length; + if (!len) + continue; if (is_in_order) { rv = svm_fifo_enqueue_nowait (s->server_rx_fifo, len, data); - if (rv < len) + if (rv == len) + { + written += rv; + } + else if (rv < len) { return (rv > 0) ? (written + rv) : written; } - written += rv; + else if (rv > len) + { + written += rv; + + /* written more than what was left in chain */ + if (written > b->total_length_not_including_first_buffer) + return written; + + /* drop the bytes that have already been delivered */ + session_enqueue_discard_chain_bytes (vm, b, &chain_b, rv - len); + } } else { rv = svm_fifo_enqueue_with_offset (s->server_rx_fifo, offset, len, data); if (rv) - return -1; + { + clib_warning ("failed to enqueue multi-buffer seg"); + return -1; + } offset += len; } } @@ -495,22 +221,22 @@ stream_session_enqueue_data (transport_connection_t * tc, vlib_buffer_t * b, u32 offset, u8 queue_event, u8 is_in_order) { stream_session_t *s; - int enqueued = 0, rv; + int enqueued = 0, rv, in_order_off; s = stream_session_get (tc->s_index, tc->thread_index); if (is_in_order) { - enqueued = - svm_fifo_enqueue_nowait (s->server_rx_fifo, b->current_length, - vlib_buffer_get_current (b)); - if (PREDICT_FALSE - ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && enqueued > 0)) + enqueued = svm_fifo_enqueue_nowait (s->server_rx_fifo, + b->current_length, + vlib_buffer_get_current (b)); + if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) + && enqueued >= 0)) { - rv = session_enqueue_chain_tail (s, b, 0, 1); - if (rv <= 0) - return enqueued; - enqueued += rv; + in_order_off = enqueued > b->current_length ? enqueued : 0; + rv = session_enqueue_chain_tail (s, b, in_order_off, 1); + if (rv > 0) + enqueued += rv; } } else @@ -519,9 +245,10 @@ stream_session_enqueue_data (transport_connection_t * tc, vlib_buffer_t * b, b->current_length, vlib_buffer_get_current (b)); if (PREDICT_FALSE ((b->flags & VLIB_BUFFER_NEXT_PRESENT) && !rv)) - rv = session_enqueue_chain_tail (s, b, offset + b->current_length, 0); - if (rv) - return -1; + session_enqueue_chain_tail (s, b, offset + b->current_length, 0); + /* if something was enqueued, report even this as success for ooo + * segment handling */ + return rv; } if (queue_event) @@ -551,7 +278,7 @@ u8 stream_session_no_space (transport_connection_t * tc, u32 thread_index, u16 data_len) { - stream_session_t *s = stream_session_get (tc->c_index, thread_index); + stream_session_t *s = stream_session_get (tc->s_index, thread_index); if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY)) return 1; @@ -563,6 +290,15 @@ stream_session_no_space (transport_connection_t * tc, u32 thread_index, } u32 +stream_session_tx_fifo_max_dequeue (transport_connection_t * tc) +{ + stream_session_t *s = stream_session_get (tc->s_index, tc->thread_index); + if (!s->server_tx_fifo) + return 0; + return svm_fifo_max_dequeue (s->server_tx_fifo); +} + +int stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer, u32 offset, u32 max_bytes) { @@ -597,7 +333,13 @@ stream_session_enqueue_notify (stream_session_t * s, u8 block) return 0; /* Get session's server */ - app = application_get (s->app_index); + app = application_get_if_valid (s->app_index); + + if (PREDICT_FALSE (app == 0)) + { + clib_warning ("invalid s->app_index = %d", s->app_index); + return 0; + } /* Built-in server? Hand event to the callback... */ if (app->cb_fns.builtin_server_rx_callback) @@ -658,8 +400,9 @@ session_manager_flush_enqueue_events (u32 thread_index) stream_session_t *s0; /* Get session */ - s0 = stream_session_get (session_indices_to_enqueue[i], thread_index); - if (stream_session_enqueue_notify (s0, 0 /* don't block */ )) + s0 = stream_session_get_if_valid (session_indices_to_enqueue[i], + thread_index); + if (s0 == 0 || stream_session_enqueue_notify (s0, 0 /* don't block */ )) { errors++; } @@ -691,47 +434,69 @@ stream_session_init_fifos_pointers (transport_connection_t * tc, svm_fifo_init_pointers (s->server_tx_fifo, tx_pointer); } -void -stream_session_connect_notify (transport_connection_t * tc, u8 sst, - u8 is_fail) +int +stream_session_connect_notify (transport_connection_t * tc, u8 is_fail) { - session_manager_main_t *smm = &session_manager_main; application_t *app; stream_session_t *new_s = 0; u64 handle; - u32 api_context = 0; - - handle = stream_session_half_open_lookup (smm, &tc->lcl_ip, &tc->rmt_ip, - tc->lcl_port, tc->rmt_port, - tc->proto); + u32 opaque = 0; + int error = 0; + u8 st; + + st = session_type_from_proto_and_ip (tc->transport_proto, tc->is_ip4); + handle = stream_session_half_open_lookup_handle (&tc->lcl_ip, &tc->rmt_ip, + tc->lcl_port, tc->rmt_port, + st); if (handle == HALF_OPEN_LOOKUP_INVALID_VALUE) { clib_warning ("This can't be good!"); - return; + return -1; } - /* Get the app's index from the handle we stored when opening connection */ - app = application_get (handle >> 32); - api_context = tc->s_index; + /* Get the app's index from the handle we stored when opening connection + * and the opaque (api_context for external apps) from transport session + * index*/ + app = application_get_if_valid (handle >> 32); + if (!app) + return -1; + + opaque = tc->s_index; if (!is_fail) { segment_manager_t *sm; + u8 alloc_fifos; sm = application_get_connect_segment_manager (app); - + alloc_fifos = application_is_proxy (app); /* Create new session (svm segments are allocated if needed) */ - if (stream_session_create_i (sm, tc, &new_s)) - return; - - new_s->app_index = app->index; + if (stream_session_create_i (sm, tc, alloc_fifos, &new_s)) + { + is_fail = 1; + error = -1; + } + else + new_s->app_index = app->index; } - /* Notify client */ - app->cb_fns.session_connected_callback (app->index, api_context, new_s, - is_fail); + /* Notify client application */ + if (app->cb_fns.session_connected_callback (app->index, opaque, new_s, + is_fail)) + { + clib_warning ("failed to notify app"); + if (!is_fail) + stream_session_disconnect (new_s); + } + else + { + if (!is_fail) + new_s->session_state = SESSION_STATE_READY; + } /* Cleanup session lookup */ - stream_session_half_open_table_del (smm, sst, tc); + stream_session_half_open_table_del (tc); + + return error; } void @@ -770,15 +535,19 @@ void stream_session_delete (stream_session_t * s) { session_manager_main_t *smm = vnet_get_session_manager_main (); + int rv; /* Delete from the main lookup table. */ - stream_session_table_del (smm, s); + if ((rv = stream_session_table_del (s))) + clib_warning ("hash delete error, rv %d", rv); /* Cleanup fifo segments */ segment_manager_dealloc_fifos (s->svm_segment_index, s->server_rx_fifo, s->server_tx_fifo); pool_put (smm->sessions[s->thread_index], s); + if (CLIB_DEBUG) + memset (s, 0xFA, sizeof (*s)); } /** @@ -796,9 +565,7 @@ stream_session_delete_notify (transport_connection_t * tc) /* App might've been removed already */ s = stream_session_get_if_valid (tc->s_index, tc->thread_index); if (!s) - { - return; - } + return; stream_session_delete (s); } @@ -834,11 +601,12 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index, server = application_get (listener->app_index); sm = application_get_listen_segment_manager (server, listener); - if ((rv = stream_session_create_i (sm, tc, &s))) + if ((rv = stream_session_create_i (sm, tc, 1, &s))) return rv; s->app_index = server->index; s->listener_index = listener_index; + s->session_state = SESSION_STATE_ACCEPTING; /* Shoulder-tap the server */ if (notify) @@ -863,14 +631,14 @@ stream_session_accept (transport_connection_t * tc, u32 listener_index, */ int stream_session_open (u32 app_index, session_type_t st, - transport_endpoint_t * tep, + transport_endpoint_t * rmt, transport_connection_t ** res) { transport_connection_t *tc; int rv; u64 handle; - rv = tp_vfts[st].open (&tep->ip, tep->port); + rv = tp_vfts[st].open (rmt); if (rv < 0) { clib_warning ("Transport failed to open connection."); @@ -885,7 +653,7 @@ stream_session_open (u32 app_index, session_type_t st, handle = (((u64) app_index) << 32) | (u64) tc->c_index; /* Add to the half-open lookup table */ - stream_session_half_open_table_add (st, tc, handle); + stream_session_half_open_table_add (tc, handle); *res = tc; @@ -907,7 +675,7 @@ stream_session_listen (stream_session_t * s, transport_endpoint_t * tep) u32 tci; /* Transport bind/listen */ - tci = tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port); + tci = tp_vfts[s->session_type].bind (s->session_index, tep); if (tci == (u32) ~ 0) return -1; @@ -972,8 +740,13 @@ session_send_session_evt_to_thread (u64 session_handle, /* Based on request block (or not) for lack of space */ if (PREDICT_TRUE (q->cursize < q->maxsize)) - unix_shared_memory_queue_add (q, (u8 *) & evt, - 0 /* do wait for mutex */ ); + { + if (unix_shared_memory_queue_add (q, (u8 *) & evt, + 1 /* do wait for mutex */ )) + { + clib_warning ("failed to enqueue evt"); + } + } else { clib_warning ("queue full"); @@ -1004,41 +777,18 @@ stream_session_disconnect (stream_session_t * s) void stream_session_cleanup (stream_session_t * s) { - session_manager_main_t *smm = &session_manager_main; int rv; s->session_state = SESSION_STATE_CLOSED; /* Delete from the main lookup table to avoid more enqueues */ - rv = stream_session_table_del (smm, s); + rv = stream_session_table_del (s); if (rv) clib_warning ("hash delete error, rv %d", rv); tp_vfts[s->session_type].cleanup (s->connection_index, s->thread_index); } -void -session_register_transport (u8 type, const transport_proto_vft_t * vft) -{ - session_manager_main_t *smm = vnet_get_session_manager_main (); - - vec_validate (tp_vfts, type); - tp_vfts[type] = *vft; - - /* If an offset function is provided, then peek instead of dequeue */ - smm->session_tx_fns[type] = - (vft->tx_fifo_offset) ? session_tx_fifo_peek_and_snd : - session_tx_fifo_dequeue_and_snd; -} - -transport_proto_vft_t * -session_get_transport_vft (u8 type) -{ - if (type >= vec_len (tp_vfts)) - return 0; - return &tp_vfts[type]; -} - /** * Allocate vpp event queue (once) per worker thread */ @@ -1048,30 +798,54 @@ session_vpp_event_queue_allocate (session_manager_main_t * smm, { api_main_t *am = &api_main; void *oldheap; + u32 event_queue_length = 2048; if (smm->vpp_event_queues[thread_index] == 0) { /* Allocate event fifo in the /vpe-api shared-memory segment */ oldheap = svm_push_data_heap (am->vlib_rp); + if (smm->configured_event_queue_length) + event_queue_length = smm->configured_event_queue_length; + smm->vpp_event_queues[thread_index] = - unix_shared_memory_queue_init (2048 /* nels $$$$ config */ , - sizeof (session_fifo_event_t), - 0 /* consumer pid */ , - 0 - /* (do not) send signal when queue non-empty */ - ); + unix_shared_memory_queue_init + (event_queue_length, + sizeof (session_fifo_event_t), 0 /* consumer pid */ , + 0 /* (do not) send signal when queue non-empty */ ); svm_pop_heap (oldheap); } } +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_manager_main_t *smm = &session_manager_main; vlib_thread_main_t *vtm = vlib_get_thread_main (); u32 num_threads; + u32 preallocated_sessions_per_worker; int i; num_threads = 1 /* main thread */ + vtm->n_threads; @@ -1108,30 +882,37 @@ session_manager_main_enable (vlib_main_t * vm) for (i = 0; i < vec_len (smm->vpp_event_queues); i++) session_vpp_event_queue_allocate (smm, i); - /* $$$$ preallocate hack config parameter */ - for (i = 0; i < 200000; i++) + /* Preallocate sessions */ + if (num_threads == 1) { - stream_session_t *ss; - pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES); - memset (ss, 0, sizeof (*ss)); - } + for (i = 0; i < smm->preallocated_sessions; i++) + { + stream_session_t *ss __attribute__ ((unused)); + pool_get_aligned (smm->sessions[0], ss, CLIB_CACHE_LINE_BYTES); + } - for (i = 0; i < 200000; i++) - pool_put_index (smm->sessions[0], i); + for (i = 0; i < smm->preallocated_sessions; i++) + pool_put_index (smm->sessions[0], i); + } + else + { + int j; + preallocated_sessions_per_worker = smm->preallocated_sessions / + (num_threads - 1); - clib_bihash_init_16_8 (&smm->v4_session_hash, "v4 session table", - 200000 /* $$$$ config parameter nbuckets */ , - (64 << 20) /*$$$ config parameter table size */ ); - clib_bihash_init_48_8 (&smm->v6_session_hash, "v6 session table", - 200000 /* $$$$ config parameter nbuckets */ , - (64 << 20) /*$$$ config parameter table size */ ); + for (j = 1; j < num_threads; j++) + { + for (i = 0; i < preallocated_sessions_per_worker; i++) + { + stream_session_t *ss __attribute__ ((unused)); + pool_get_aligned (smm->sessions[j], ss, CLIB_CACHE_LINE_BYTES); + } + for (i = 0; i < preallocated_sessions_per_worker; i++) + pool_put_index (smm->sessions[j], i); + } + } - clib_bihash_init_16_8 (&smm->v4_half_open_hash, "v4 half-open table", - 200000 /* $$$$ config parameter nbuckets */ , - (64 << 20) /*$$$ config parameter table size */ ); - clib_bihash_init_48_8 (&smm->v6_half_open_hash, "v6 half-open table", - 200000 /* $$$$ config parameter nbuckets */ , - (64 << 20) /*$$$ config parameter table size */ ); + session_lookup_init (); smm->is_enabled = 1; @@ -1178,15 +959,84 @@ clib_error_t * session_manager_main_init (vlib_main_t * vm) { session_manager_main_t *smm = &session_manager_main; - - smm->vlib_main = vm; - smm->vnet_main = vnet_get_main (); smm->is_enabled = 0; + return 0; +} + +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; + uword tmp; + while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT) + { + if (unformat (input, "event-queue-length %d", &nitems)) + { + if (nitems >= 2048) + smm->configured_event_queue_length = nitems; + else + clib_warning ("event queue length %d too small, ignored", nitems); + } + else if (unformat (input, "preallocated-sessions %d", + &smm->preallocated_sessions)) + ; + else if (unformat (input, "v4-session-table-buckets %d", + &smm->configured_v4_session_table_buckets)) + ; + else if (unformat (input, "v4-halfopen-table-buckets %d", + &smm->configured_v4_halfopen_table_buckets)) + ; + else if (unformat (input, "v6-session-table-buckets %d", + &smm->configured_v6_session_table_buckets)) + ; + else if (unformat (input, "v6-halfopen-table-buckets %d", + &smm->configured_v6_halfopen_table_buckets)) + ; + else if (unformat (input, "v4-session-table-memory %U", + unformat_memory_size, &tmp)) + { + if (tmp >= 0x100000000) + return clib_error_return (0, "memory size %llx (%lld) too large", + tmp, tmp); + smm->configured_v4_session_table_memory = tmp; + } + else if (unformat (input, "v4-halfopen-table-memory %U", + unformat_memory_size, &tmp)) + { + if (tmp >= 0x100000000) + return clib_error_return (0, "memory size %llx (%lld) too large", + tmp, tmp); + smm->configured_v4_halfopen_table_memory = tmp; + } + else if (unformat (input, "v6-session-table-memory %U", + unformat_memory_size, &tmp)) + { + if (tmp >= 0x100000000) + return clib_error_return (0, "memory size %llx (%lld) too large", + tmp, tmp); + smm->configured_v6_session_table_memory = tmp; + } + else if (unformat (input, "v6-halfopen-table-memory %U", + unformat_memory_size, &tmp)) + { + if (tmp >= 0x100000000) + return clib_error_return (0, "memory size %llx (%lld) too large", + tmp, tmp); + smm->configured_v6_halfopen_table_memory = tmp; + } + else + return clib_error_return (0, "unknown input `%U'", + format_unformat_error, input); + } return 0; } -VLIB_INIT_FUNCTION (session_manager_main_init) +VLIB_CONFIG_FUNCTION (session_config_fn, "session"); + /* * fd.io coding-style-patch-verification: ON *