48 case SESSION_TYPE_IP4_UDP:
49 case SESSION_TYPE_IP4_TCP:
52 clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4, 1 );
54 case SESSION_TYPE_IP6_UDP:
55 case SESSION_TYPE_IP6_TCP:
58 clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6, 1 );
72 tc =
tp_vfts[s->session_type].get_connection (s->connection_index,
87 case SESSION_TYPE_IP4_UDP:
88 case SESSION_TYPE_IP4_TCP:
91 clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4,
94 case SESSION_TYPE_IP6_UDP:
95 case SESSION_TYPE_IP6_TCP:
98 clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6,
115 case SESSION_TYPE_IP4_UDP:
116 case SESSION_TYPE_IP4_TCP:
118 return clib_bihash_add_del_16_8 (&smm->v4_session_hash, &kv4,
121 case SESSION_TYPE_IP6_UDP:
122 case SESSION_TYPE_IP6_TCP:
124 return clib_bihash_add_del_48_8 (&smm->v6_session_hash, &kv6,
140 ts =
tp_vfts[s->session_type].get_connection (s->connection_index,
154 case SESSION_TYPE_IP4_UDP:
155 case SESSION_TYPE_IP4_TCP:
157 clib_bihash_add_del_16_8 (&smm->v4_half_open_hash, &kv4,
160 case SESSION_TYPE_IP6_UDP:
161 case SESSION_TYPE_IP6_TCP:
163 clib_bihash_add_del_48_8 (&smm->v6_half_open_hash, &kv6,
180 rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
186 rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
201 u16 lcl_port,
u16 rmt_port,
u8 proto,
210 rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
226 rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
231 kv6.
key[0] = kv6.
key[1] = 0;
232 rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
245 u16 lcl_port,
u16 rmt_port,
u8 proto,
253 rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
266 case SESSION_TYPE_IP4_UDP:
267 case SESSION_TYPE_IP4_TCP:
270 case SESSION_TYPE_IP6_UDP:
271 case SESSION_TYPE_IP6_TCP:
280 ip46_address_t * lcl, ip46_address_t * rmt,
281 u16 lcl_port,
u16 rmt_port,
u8 proto)
289 case SESSION_TYPE_IP4_UDP:
290 case SESSION_TYPE_IP4_TCP:
291 make_v4_ss_kv (&kv4, &lcl->ip4, &rmt->ip4, lcl_port, rmt_port, proto);
292 rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
299 case SESSION_TYPE_IP6_UDP:
300 case SESSION_TYPE_IP6_TCP:
301 make_v6_ss_kv (&kv6, &lcl->ip6, &rmt->ip6, lcl_port, rmt_port, proto);
302 rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
315 u16 lcl_port,
u16 rmt_port,
u8 proto,
325 rv = clib_bihash_search_inline_16_8 (&smm->v4_session_hash, &kv4);
330 return tp_vfts[s->session_type].get_connection (s->connection_index,
337 return tp_vfts[s->session_type].get_listener (s->connection_index);
340 rv = clib_bihash_search_inline_16_8 (&smm->v4_half_open_hash, &kv4);
342 return tp_vfts[proto].get_half_open (kv4.
value & 0xFFFFFFFF);
349 u16 lcl_port,
u16 rmt_port,
u8 proto,
358 rv = clib_bihash_search_inline_48_8 (&smm->v6_session_hash, &kv6);
363 return tp_vfts[s->session_type].get_connection (s->connection_index,
370 return tp_vfts[s->session_type].get_listener (s->connection_index);
373 rv = clib_bihash_search_inline_48_8 (&smm->v6_half_open_hash, &kv6);
375 return tp_vfts[proto].get_half_open (kv6.
value & 0xFFFFFFFF);
385 svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
386 u32 fifo_segment_index;
390 u32 thread_index = tc->thread_index;
395 &fifo_segment_index)))
400 memset (s, 0,
sizeof (*s));
403 pool_index = s - smm->sessions[thread_index];
404 server_rx_fifo->master_session_index = pool_index;
405 server_rx_fifo->master_thread_index = thread_index;
407 server_tx_fifo->master_session_index = pool_index;
408 server_tx_fifo->master_thread_index = thread_index;
410 s->server_rx_fifo = server_rx_fifo;
411 s->server_tx_fifo = server_tx_fifo;
414 s->session_type = tc->proto;
416 s->svm_segment_index = fifo_segment_index;
417 s->thread_index = thread_index;
418 s->session_index = pool_index;
421 s->connection_index = tc->c_index;
424 tc->s_index = s->session_index;
427 value = (((
u64) thread_index) << 32) | (
u64) s->session_index;
457 return (rv > 0) ? (written + rv) : written;
498 int enqueued = 0, rv;
532 u32 thread_index = s->thread_index;
533 u32 my_enqueue_epoch = smm->current_enqueue_epoch[thread_index];
535 if (s->enqueue_epoch != my_enqueue_epoch)
537 s->enqueue_epoch = my_enqueue_epoch;
538 vec_add1 (smm->session_indices_to_enqueue_by_thread[thread_index],
539 s - smm->sessions[thread_index]);
579 return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
601 session_fifo_event_t evt;
603 static u32 serial_number;
612 if (app->cb_fns.builtin_server_rx_callback)
613 return app->cb_fns.builtin_server_rx_callback (s);
619 evt.fifo = s->server_rx_fifo;
621 evt.event_id = serial_number++;
624 q = app->event_queue;
639 ed->data[0] = evt.event_id;
659 u32 *session_indices_to_enqueue;
662 session_indices_to_enqueue =
663 smm->session_indices_to_enqueue_by_thread[thread_index];
665 for (i = 0; i <
vec_len (session_indices_to_enqueue); i++)
679 smm->session_indices_to_enqueue_by_thread[thread_index] =
680 session_indices_to_enqueue;
683 smm->current_enqueue_epoch[thread_index]++;
695 u32 rx_pointer,
u32 tx_pointer)
715 tc->lcl_port, tc->rmt_port,
725 api_context = tc->s_index;
739 new_s->app_index = app->index;
743 app->cb_fns.session_connected_callback (app->index, api_context, new_s,
760 server->cb_fns.session_accept_callback (s);
778 server->cb_fns.session_disconnect_callback (s);
796 pool_put (smm->sessions[s->thread_index], s);
831 app->cb_fns.session_reset_callback (s);
855 s->app_index = server->index;
856 s->listener_index = listener_index;
861 server->cb_fns.session_accept_callback (s);
888 rv =
tp_vfts[st].open (&tep->ip, tep->port);
892 return VNET_API_ERROR_SESSION_CONNECT_FAIL;
900 handle = (((
u64) app_index) << 32) | (
u64) tc->c_index;
925 tci =
tp_vfts[s->session_type].bind (s->session_index, &tep->ip, tep->port);
927 if (tci == (
u32) ~ 0)
931 s->connection_index = tci;
932 tc =
tp_vfts[s->session_type].get_listener (tci);
960 tc =
tp_vfts[s->session_type].get_listener (s->connection_index);
964 return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
968 tp_vfts[s->session_type].unbind (s->connection_index);
977 static u16 serial_number = 0;
978 session_fifo_event_t evt;
982 evt.session_handle = session_handle;
983 evt.event_type = evt_type;
984 evt.event_id = serial_number++;
1015 tp_vfts[s->session_type].close (s->connection_index, s->thread_index);
1037 tp_vfts[s->session_type].cleanup (s->connection_index, s->thread_index);
1049 smm->session_tx_fns[type] =
1071 u32 event_queue_length = 2048;
1073 if (smm->vpp_event_queues[thread_index] == 0)
1078 if (smm->configured_event_queue_length)
1079 event_queue_length = smm->configured_event_queue_length;
1081 smm->vpp_event_queues[thread_index] =
1083 (event_queue_length,
1084 sizeof (session_fifo_event_t), 0 ,
1101 if (num_threads < 1)
1110 vec_validate (smm->session_indices_to_enqueue_by_thread, num_threads - 1);
1112 vec_validate (smm->pending_event_vector, num_threads - 1);
1113 vec_validate (smm->free_event_vector, num_threads - 1);
1114 vec_validate (smm->current_enqueue_epoch, num_threads - 1);
1117 for (i = 0; i < num_threads; i++)
1120 _vec_len (smm->free_event_vector[i]) = 0;
1122 _vec_len (smm->pending_event_vector[i]) = 0;
1126 vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1130 for (i = 0; i <
vec_len (smm->vpp_event_queues); i++)
1134 for (i = 0; i < 200000; i++)
1138 memset (ss, 0,
sizeof (*ss));
1141 for (i = 0; i < 200000; i++)
1144 clib_bihash_init_16_8 (&smm->v4_session_hash,
"v4 session table",
1147 clib_bihash_init_48_8 (&smm->v6_session_hash,
"v6 session table",
1151 clib_bihash_init_16_8 (&smm->v4_half_open_hash,
"v4 half-open table",
1154 clib_bihash_init_48_8 (&smm->v6_half_open_hash,
"v6 half-open table",
1158 smm->is_enabled = 1;
1169 u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1204 smm->vlib_main = vm;
1206 smm->is_enabled = 0;
1220 if (
unformat (input,
"event-queue-length %d", &nitems))
1223 smm->configured_event_queue_length = nitems;
1225 clib_warning (
"event queue length %d too small, ignored", nitems);
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
int session_manager_flush_enqueue_events(u32 thread_index)
Flushes queue of sessions that are to be notified of new data enqueued events.
sll srl srl sll sra u16x4 i
static void svm_pop_heap(void *oldheap)
int segment_manager_alloc_session_fifos(segment_manager_t *sm, svm_fifo_t **server_rx_fifo, svm_fifo_t **server_tx_fifo, u32 *fifo_segment_index)
void svm_fifo_init_pointers(svm_fifo_t *f, u32 pointer)
Set fifo pointers to requested offset.
int stream_session_accept(transport_connection_t *tc, u32 listener_index, u8 sst, u8 notify)
Accept a stream session.
void stream_session_table_add(session_manager_main_t *smm, stream_session_t *s, u64 value)
static transport_proto_vft_t * tp_vfts
Per-type vector of transport protocol virtual function tables.
stream_session_t * stream_session_lookup_listener4(ip4_address_t *lcl, u16 lcl_port, u8 proto)
vnet_main_t * vnet_get_main(void)
struct _transport_connection transport_connection_t
void stream_session_table_add_for_tc(transport_connection_t *tc, u64 value)
static void make_v4_ss_kv_from_tc(session_kv4_t *kv, transport_connection_t *t)
int stream_session_enqueue_data(transport_connection_t *tc, vlib_buffer_t *b, u32 offset, u8 queue_event, u8 is_in_order)
session_manager_main_t session_manager_main
#define vec_add1(V, E)
Add 1 element to end of vector (unspecified alignment).
static void make_v6_ss_kv(session_kv6_t *kv, ip6_address_t *lcl, ip6_address_t *rmt, u16 lcl_port, u16 rmt_port, u8 proto)
int stream_session_table_del_for_tc(transport_connection_t *tc)
static void make_v6_ss_kv_from_tc(session_kv6_t *kv, transport_connection_t *t)
int stream_session_create_i(segment_manager_t *sm, transport_connection_t *tc, stream_session_t **ret_s)
static u32 svm_fifo_max_enqueue(svm_fifo_t *f)
void session_node_enable_disable(u8 is_en)
session_fifo_rx_fn session_tx_fifo_peek_and_snd
static void make_v6_listener_kv(session_kv6_t *kv, ip6_address_t *lcl, u16 lcl_port, u8 proto)
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
struct _stream_session_t stream_session_t
struct _svm_fifo svm_fifo_t
segment_manager_t * application_get_listen_segment_manager(application_t *app, stream_session_t *s)
static void make_v4_listener_kv(session_kv4_t *kv, ip4_address_t *lcl, u16 lcl_port, u8 proto)
#define VLIB_BUFFER_NEXT_PRESENT
static void stream_session_half_open_table_del(session_manager_main_t *smm, u8 sst, transport_connection_t *tc)
#define VLIB_INIT_FUNCTION(x)
static void * svm_push_data_heap(svm_region_t *rp)
void stream_session_accept_notify(transport_connection_t *tc)
static u32 svm_fifo_max_dequeue(svm_fifo_t *f)
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)
stream_session_t * stream_session_lookup_listener6(ip6_address_t *lcl, u16 lcl_port, u8 proto)
#define clib_error_return(e, args...)
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)
Looks up a session based on the 5-tuple passed as argument.
void stream_session_delete_notify(transport_connection_t *tc)
Notification from transport that connection is being deleted.
void stream_session_cleanup(stream_session_t *s)
Cleanup transport and session state.
void stream_session_delete(stream_session_t *s)
Cleans up session and associated app if needed.
int stream_session_connect_notify(transport_connection_t *tc, u8 sst, u8 is_fail)
void session_send_session_evt_to_thread(u64 session_handle, fifo_event_type_t evt_type, u32 thread_index)
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)
static clib_error_t * session_manager_main_enable(vlib_main_t *vm)
struct _transport_proto_vft transport_proto_vft_t
int unix_shared_memory_queue_add(unix_shared_memory_queue_t *q, u8 *elem, int nowait)
u32 stream_session_dequeue_drop(transport_connection_t *tc, u32 max_bytes)
static session_manager_main_t * vnet_get_session_manager_main()
#define pool_elt_at_index(p, i)
Returns pointer to element at given index.
u16 current_length
Nbytes between current data and the end of this buffer.
static void * vlib_buffer_get_current(vlib_buffer_t *b)
Get pointer to current data to process.
#define pool_put(P, E)
Free an object E in pool P.
void svm_fifo_segment_init(u64 baseva, u32 timeout_in_seconds)
#define VLIB_CONFIG_FUNCTION(x, n,...)
clib_error_t * vnet_tcp_enable_disable(vlib_main_t *vm, u8 is_en)
clib_error_t * session_manager_main_init(vlib_main_t *vm)
struct _session_manager_main session_manager_main_t
#define foreach_vlib_main(body)
clib_error_t * vnet_session_enable_disable(vlib_main_t *vm, u8 is_en)
#define pool_get_aligned(P, E, A)
Allocate an object E from a pool P (general version).
static u8 svm_fifo_set_event(svm_fifo_t *f)
Sets fifo event flag.
#define SESSION_EVT_DBG(_evt, _args...)
static unix_shared_memory_queue_t * session_manager_get_vpp_event_queue(u32 thread_index)
session_fifo_rx_fn session_tx_fifo_dequeue_and_snd
int stream_session_open(u32 app_index, session_type_t st, transport_endpoint_t *tep, transport_connection_t **res)
Ask transport to open connection to remote transport endpoint.
u32 stream_session_tx_fifo_max_dequeue(transport_connection_t *tc)
segment_manager_t * application_get_connect_segment_manager(application_t *app)
#define clib_warning(format, args...)
void stream_session_init_fifos_pointers(transport_connection_t *tc, u32 rx_pointer, u32 tx_pointer)
Init fifo tail and head pointers.
struct _application application_t
int svm_fifo_enqueue_nowait(svm_fifo_t *f, u32 max_bytes, u8 *copy_from_here)
static stream_session_t * stream_session_get_if_valid(u64 si, u32 thread_index)
void stream_session_disconnect_notify(transport_connection_t *tc)
Notification from transport that connection is being closed.
static stream_session_t * stream_session_get_tsi(u64 ti_and_si, u32 thread_index)
#define pool_put_index(p, i)
Free pool element with given index.
unix_shared_memory_queue_t * unix_shared_memory_queue_init(int nels, int elsize, int consumer_pid, int signal_when_queue_non_empty)
vhost_vring_state_t state
u32 next_buffer
Next buffer for this linked-list of buffers.
static void make_v4_ss_kv(session_kv4_t *kv, ip4_address_t *lcl, ip4_address_t *rmt, u16 lcl_port, u16 rmt_port, u8 proto)
static void stream_session_half_open_table_add(session_type_t sst, transport_connection_t *tc, u64 value)
void stream_session_disconnect(stream_session_t *s)
Disconnect session and propagate to transport.
void stream_session_reset_notify(transport_connection_t *tc)
Notify application that connection has been reset.
static void vlib_node_set_state(vlib_main_t *vm, u32 node_index, vlib_node_state_t new_state)
Set node dispatch state.
int svm_fifo_enqueue_with_offset(svm_fifo_t *f, u32 offset, u32 required_bytes, u8 *copy_from_here)
int stream_session_stop_listen(stream_session_t *s)
Ask transport to stop listening on local transport endpoint.
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)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
void session_register_transport(u8 type, const transport_proto_vft_t *vft)
static vlib_main_t * vlib_get_main(void)
#define HALF_OPEN_LOOKUP_INVALID_VALUE
transport_proto_vft_t * session_get_transport_vft(u8 type)
template key/value backing page structure
static int stream_session_table_del(session_manager_main_t *smm, stream_session_t *s)
static int stream_session_enqueue_notify(stream_session_t *s, u8 block)
Notify session peer that new data has been enqueued.
int svm_fifo_dequeue_drop(svm_fifo_t *f, u32 max_bytes)
void segment_manager_dealloc_fifos(u32 svm_segment_index, svm_fifo_t *rx_fifo, svm_fifo_t *tx_fifo)
void session_vpp_event_queue_allocate(session_manager_main_t *smm, u32 thread_index)
Allocate vpp event queue (once) per worker thread.
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
static stream_session_t * listen_session_get(session_type_t type, u32 index)
application_t * application_get(u32 index)
struct _transport_endpoint transport_endpoint_t
vlib_init_function_t *static _vlib_init_function_session_manager_main_init clib_error_t * session_config_fn(vlib_main_t *vm, unformat_input_t *input)
(constructor) VLIB_INIT_FUNCTION (session_manager_main_init)
struct _segment_manager segment_manager_t
int stream_session_listen(stream_session_t *s, transport_endpoint_t *tep)
Ask transport to listen on local transport endpoint.
static vlib_thread_main_t * vlib_get_thread_main()
static int session_enqueue_chain_tail(stream_session_t *s, vlib_buffer_t *b, u32 offset, u8 is_in_order)
Enqueue buffer chain tail.
stream_session_t * stream_session_lookup_listener(ip46_address_t *lcl, u16 lcl_port, u8 proto)
#define CLIB_CACHE_LINE_BYTES
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)
u32 flags
buffer flags: VLIB_BUFFER_IS_TRACED: trace this buffer.
int svm_fifo_peek(svm_fifo_t *f, u32 relative_offset, u32 max_bytes, u8 *copy_here)
int stream_session_peek_bytes(transport_connection_t *tc, u8 *buffer, u32 offset, u32 max_bytes)
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
u8 stream_session_no_space(transport_connection_t *tc, u32 thread_index, u16 data_len)
Check if we have space in rx fifo to push more bytes.
struct _unix_shared_memory_queue unix_shared_memory_queue_t
static stream_session_t * stream_session_get(u32 si, u32 thread_index)