32 u32 thread_index,
void *fp,
void *rpc_args)
35 session_fifo_event_t evt = { {0}, };
38 evt.event_type = evt_type;
42 evt.rpc_args.arg = rpc_args;
74 void (*fnp) (
void *) = fp;
100 memset (s, 0,
sizeof (*s));
102 s->thread_index = thread_index;
111 memset (s, 0xFA,
sizeof (*s));
117 svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
118 u32 fifo_segment_index;
123 &fifo_segment_index)))
126 server_rx_fifo->master_session_index = s->session_index;
127 server_rx_fifo->master_thread_index = s->thread_index;
129 server_tx_fifo->master_session_index = s->session_index;
130 server_tx_fifo->master_thread_index = s->thread_index;
132 s->server_rx_fifo = server_rx_fifo;
133 s->server_tx_fifo = server_tx_fifo;
134 s->svm_segment_index = fifo_segment_index;
142 u32 thread_index = tc->thread_index;
149 s->enqueue_epoch = ~0;
152 s->connection_index = tc->c_index;
153 tc->s_index = s->session_index;
190 u32 to_drop = n_bytes_to_drop;
220 u32 chain_bi, len, diff;
226 if (is_in_order && offset)
254 return (rv > 0) ? (written + rv) : written;
306 u8 queue_event,
u8 is_in_order)
309 int enqueued = 0, rv, in_order_off;
344 u32 thread_index = s->thread_index;
345 u32 enqueue_epoch = smm->current_enqueue_epoch[tc->proto][thread_index];
347 if (s->enqueue_epoch != enqueue_epoch)
349 s->enqueue_epoch = enqueue_epoch;
350 vec_add1 (smm->session_to_enqueue[tc->proto][thread_index],
351 s - smm->sessions[thread_index]);
360 u8 proto,
u8 queue_event)
362 int enqueued = 0, rv, in_order_off;
380 u32 thread_index = s->thread_index;
381 u32 enqueue_epoch = smm->current_enqueue_epoch[proto][thread_index];
383 if (s->enqueue_epoch != enqueue_epoch)
385 s->enqueue_epoch = enqueue_epoch;
386 vec_add1 (smm->session_to_enqueue[proto][thread_index],
387 s - smm->sessions[thread_index]);
413 if (!s->server_tx_fifo)
423 return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
445 session_fifo_event_t evt;
462 clib_warning (
"invalid s->app_index = %d", s->app_index);
467 if (app->cb_fns.builtin_server_rx_callback)
468 return app->cb_fns.builtin_server_rx_callback (s);
474 evt.fifo = s->server_rx_fifo;
478 q = app->event_queue;
493 ed->data[0] = evt.event_type;
517 indices = smm->session_to_enqueue[transport_proto][thread_index];
519 for (i = 0; i <
vec_len (indices); i++)
527 smm->session_to_enqueue[transport_proto][thread_index] = indices;
528 smm->current_enqueue_epoch[transport_proto][thread_index]++;
540 u32 rx_pointer,
u32 tx_pointer)
576 opaque = tc->s_index;
591 new_s->app_index = app->index;
597 if (app->cb_fns.session_connected_callback (app->index, opaque, new_s,
613 typedef struct _session_switch_pool_args
617 u32 new_thread_index;
618 u32 new_session_index;
628 s =
session_get (args->session_index, args->thread_index);
629 s->server_tx_fifo->master_session_index = args->new_session_index;
630 s->server_tx_fifo->master_thread_index = args->new_thread_index;
632 tp_vfts[tp].cleanup (s->connection_index, s->thread_index);
642 u32 old_thread_index,
652 new_s->connection_index = tc->c_index;
653 new_s->server_rx_fifo->master_session_index = new_s->session_index;
654 new_s->server_rx_fifo->master_thread_index = new_s->thread_index;
663 rpc_args->new_session_index = new_s->session_index;
664 rpc_args->new_thread_index = new_s->thread_index;
665 rpc_args->session_index = tc->s_index;
666 rpc_args->thread_index = old_thread_index;
670 tc->s_index = new_s->session_index;
671 new_s->connection_index = tc->c_index;
672 *new_session = new_s;
684 server->cb_fns.session_accept_callback (s);
702 server->cb_fns.session_disconnect_callback (s);
754 app->cb_fns.session_reset_callback (s);
780 s->app_index = server->index;
781 s->listener_index = listener_index;
787 server->cb_fns.session_accept_callback (s);
818 rv =
tp_vfts[rmt->transport_proto].open (tep);
821 SESSION_DBG (
"Transport failed to open connection.");
822 return VNET_API_ERROR_SESSION_CONNECT;
825 tc =
tp_vfts[rmt->transport_proto].get_half_open ((
u32) rv);
837 handle = (((
u64) app_index) << 32) | (
u64) tc->c_index;
843 tc->s_index = opaque;
854 s->app_index = app->index;
858 app->cb_fns.session_connected_callback (app->index, opaque, s, 0);
878 tci =
tp_vfts[sep->transport_proto].bind (s->session_index,
882 if (tci == (
u32) ~ 0)
886 s->connection_index = tci;
887 tc =
tp_vfts[sep->transport_proto].get_listener (tci);
914 tc =
tp_vfts[tp].get_listener (s->connection_index);
918 return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
922 tp_vfts[tp].unbind (s->connection_index);
972 u32 event_queue_length = 2048;
974 if (smm->vpp_event_queues[thread_index] == 0)
979 if (smm->configured_event_queue_length)
980 event_queue_length = smm->configured_event_queue_length;
982 smm->vpp_event_queues[thread_index] =
985 sizeof (session_fifo_event_t), 0 ,
1005 u32 next_index = ~0;
1009 vec_validate (smm->session_type_to_next, session_type);
1020 smm->session_type_to_next[session_type] = next_index;
1022 vft->tx_fifo_offset != 0);
1032 return tp_vfts[tp].get_connection (s->connection_index,
1042 return tp_vfts[tp].get_listener (s->connection_index);
1051 tc =
tp_vfts[tp].get_listener (listener->connection_index);
1059 sep->port = tc->lcl_port;
1060 sep->transport_proto = tc->proto;
1061 sep->is_ip4 = tc->is_ip4;
1071 u32 preallocated_sessions_per_worker;
1076 if (num_threads < 1)
1086 vec_validate (smm->pending_event_vector, num_threads - 1);
1087 vec_validate (smm->pending_disconnects, num_threads - 1);
1088 vec_validate (smm->free_event_vector, num_threads - 1);
1091 vec_validate (smm->peekers_readers_locks, num_threads - 1);
1092 vec_validate (smm->peekers_write_locks, num_threads - 1);
1095 for (j = 0; j < num_threads; j++)
1097 vec_validate (smm->session_to_enqueue[i], num_threads - 1);
1098 vec_validate (smm->current_enqueue_epoch[i], num_threads - 1);
1101 for (i = 0; i < num_threads; i++)
1104 _vec_len (smm->free_event_vector[i]) = 0;
1106 _vec_len (smm->pending_event_vector[i]) = 0;
1108 _vec_len (smm->pending_disconnects[i]) = 0;
1109 if (num_threads > 1)
1117 vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1121 for (i = 0; i <
vec_len (smm->vpp_event_queues); i++)
1125 if (smm->preallocated_sessions)
1127 if (num_threads == 1)
1134 preallocated_sessions_per_worker =
1135 (1.1 * (
f64) smm->preallocated_sessions /
1136 (
f64) (num_threads - 1));
1138 for (j = 1; j < num_threads; j++)
1141 preallocated_sessions_per_worker);
1150 smm->is_enabled = 1;
1161 u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1195 smm->is_enabled = 0;
1210 if (
unformat (input,
"event-queue-length %d", &nitems))
1213 smm->configured_event_queue_length = nitems;
1215 clib_warning (
"event queue length %d too small, ignored", nitems);
1217 else if (
unformat (input,
"preallocated-sessions %d",
1218 &smm->preallocated_sessions))
1220 else if (
unformat (input,
"v4-session-table-buckets %d",
1221 &smm->configured_v4_session_table_buckets))
1223 else if (
unformat (input,
"v4-halfopen-table-buckets %d",
1224 &smm->configured_v4_halfopen_table_buckets))
1226 else if (
unformat (input,
"v6-session-table-buckets %d",
1227 &smm->configured_v6_session_table_buckets))
1229 else if (
unformat (input,
"v6-halfopen-table-buckets %d",
1230 &smm->configured_v6_halfopen_table_buckets))
1232 else if (
unformat (input,
"v4-session-table-memory %U",
1235 if (tmp >= 0x100000000)
1238 smm->configured_v4_session_table_memory = tmp;
1240 else if (
unformat (input,
"v4-halfopen-table-memory %U",
1243 if (tmp >= 0x100000000)
1246 smm->configured_v4_halfopen_table_memory = tmp;
1248 else if (
unformat (input,
"v6-session-table-memory %U",
1251 if (tmp >= 0x100000000)
1254 smm->configured_v6_session_table_memory = tmp;
1256 else if (
unformat (input,
"v6-halfopen-table-memory %U",
1259 if (tmp >= 0x100000000)
1262 smm->configured_v6_halfopen_table_memory = tmp;
1264 else if (
unformat (input,
"local-endpoints-table-memory %U",
1267 if (tmp >= 0x100000000)
1270 smm->local_endpoints_table_memory = tmp;
1272 else if (
unformat (input,
"local-endpoints-table-buckets %d",
1273 &smm->local_endpoints_table_buckets))
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
int session_lookup_del_connection(transport_connection_t *tc)
Delete transport connection from session table.
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.
transport_proto_vft_t * tp_vfts
Per-type vector of transport protocol virtual function tables.
struct _transport_connection transport_connection_t
session_manager_main_t session_manager_main
static_always_inline void clib_spinlock_unlock_if_init(clib_spinlock_t *p)
static int session_enqueue_notify(stream_session_t *s, u8 block)
Notify session peer that new data has been enqueued.
#define pool_get_aligned_will_expand(P, YESNO, A)
See if pool_get will expand the pool or not.
struct _transport_proto_vft transport_proto_vft_t
#define vec_add1(V, E)
Add 1 element to end of vector (unspecified alignment).
void session_send_rpc_evt_to_thread(u32 thread_index, void *fp, void *rpc_args)
transport_connection_t * session_get_transport(stream_session_t *s)
static u32 svm_fifo_max_enqueue(svm_fifo_t *f)
int session_lookup_del_session(stream_session_t *s)
int session_enqueue_stream_connection(transport_connection_t *tc, vlib_buffer_t *b, u32 offset, u8 queue_event, u8 is_in_order)
u64 session_lookup_half_open_handle(transport_connection_t *tc)
void session_node_enable_disable(u8 is_en)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
static uword vlib_node_add_next(vlib_main_t *vm, uword node, uword next_node)
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
static void session_send_evt_to_thread(u64 session_handle, fifo_event_type_t evt_type, u32 thread_index, void *fp, void *rpc_args)
struct _svm_fifo svm_fifo_t
segment_manager_t * application_get_listen_segment_manager(application_t *app, stream_session_t *s)
#define VLIB_BUFFER_NEXT_PRESENT
void app_namespaces_init(void)
static clib_error_t * session_config_fn(vlib_main_t *vm, unformat_input_t *input)
#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 void session_free(stream_session_t *s)
static stream_session_t * session_alloc_for_connection(transport_connection_t *tc)
#define clib_error_return(e, args...)
svm_region_t * vlib_rp
Current binary api segment descriptor.
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 lookup table.
struct _stream_session_t stream_session_t
void session_send_session_evt_to_thread(u64 session_handle, fifo_event_type_t evt_type, u32 thread_index)
int session_lookup_del_half_open(transport_connection_t *tc)
static u32 vlib_get_buffer_index(vlib_main_t *vm, void *p)
Translate buffer pointer into buffer index.
static clib_error_t * session_manager_main_enable(vlib_main_t *vm)
static void clib_spinlock_init(clib_spinlock_t *p)
#define session_endpoint_to_transport(_sep)
int unix_shared_memory_queue_add(unix_shared_memory_queue_t *q, u8 *elem, int nowait)
static transport_proto_t session_get_transport_proto(stream_session_t *s)
u32 stream_session_dequeue_drop(transport_connection_t *tc, u32 max_bytes)
int session_open(u32 app_index, session_endpoint_t *rmt, u32 opaque)
Ask transport to open connection to remote transport endpoint.
int session_dgram_connect_notify(transport_connection_t *tc, u32 old_thread_index, stream_session_t **new_session)
Move dgram session to the right thread.
static session_manager_main_t * vnet_get_session_manager_main()
u16 current_length
Nbytes between current data and the end of this buffer.
struct _session_endpoint session_endpoint_t
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 * session_manager_main_init(vlib_main_t *vm)
static stream_session_t * session_get_if_valid(u64 si, u32 thread_index)
struct _session_manager_main session_manager_main_t
#define foreach_vlib_main(body)
int stream_session_accept(transport_connection_t *tc, u32 listener_index, u8 notify)
Accept a stream session.
clib_error_t * vnet_session_enable_disable(vlib_main_t *vm, u8 is_en)
static stream_session_t * session_get(u32 si, u32 thread_index)
struct _session_switch_pool_args session_switch_pool_args_t
API main structure, used by both vpp and binary API clients.
#define pool_get_aligned(P, E, A)
Allocate an object E from a pool P (general version).
#define SESSION_DBG(_fmt, _args...)
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)
void transport_init(void)
static_always_inline uword vlib_get_thread_index(void)
static session_type_t session_type_from_proto_and_ip(transport_proto_t proto, u8 is_ip4)
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
static void session_manager_set_transport_rx_fn(session_type_t type, u8 is_peek)
Set peek or dequeue function for given session type.
int svm_fifo_enqueue_nowait(svm_fifo_t *f, u32 max_bytes, u8 *copy_from_here)
void transport_enable_disable(vlib_main_t *vm, u8 is_en)
#define pool_init_fixed(pool, max_elts)
initialize a fixed-size, preallocated pool
static u8 transport_is_stream(u8 proto)
int session_enqueue_dgram_connection(stream_session_t *s, vlib_buffer_t *b, u8 proto, u8 queue_event)
void stream_session_disconnect_notify(transport_connection_t *tc)
Notification from transport that connection is being closed.
static int session_alloc_fifos(segment_manager_t *sm, stream_session_t *s)
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 clib_mem_free(void *p)
int listen_session_get_local_session_endpoint(stream_session_t *listener, session_endpoint_t *sep)
void stream_session_disconnect(stream_session_t *s)
Disconnect session and propagate to transport.
static void vlib_buffer_advance(vlib_buffer_t *b, word l)
Advance current data pointer by the supplied (signed!) amount.
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.
static void * clib_mem_alloc(uword size)
static vlib_main_t * vlib_get_main(void)
#define HALF_OPEN_LOOKUP_INVALID_VALUE
static void session_enqueue_discard_chain_bytes(vlib_main_t *vm, vlib_buffer_t *b, vlib_buffer_t **chain_b, u32 n_bytes_to_drop)
Discards bytes from buffer chain.
u32 total_length_not_including_first_buffer
Only valid for first buffer in chain.
template key/value backing page structure
static u64 session_handle(stream_session_t *s)
static void session_switch_pool(void *cb_args)
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.
enum _transport_proto transport_proto_t
transport_connection_t * listen_session_get_transport(stream_session_t *s)
void session_lookup_init(void)
#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)
static stream_session_t * session_clone_safe(u32 session_index, u32 thread_index)
application_t * application_get(u32 index)
struct _transport_endpoint transport_endpoint_t
struct _segment_manager segment_manager_t
int session_lookup_add_connection(transport_connection_t *tc, u64 value)
Add transport connection to a session table.
int session_manager_flush_enqueue_events(u8 transport_proto, u32 thread_index)
Flushes queue of sessions that are to be notified of new data enqueued events.
int session_stream_connect_notify(transport_connection_t *tc, u8 is_fail)
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.
int application_is_builtin_proxy(application_t *app)
stream_session_t * session_alloc(u32 thread_index)
#define CLIB_CACHE_LINE_BYTES
u32 flags
buffer flags: VLIB_BUFFER_FREE_LIST_INDEX_MASK: bits used to store free list index, VLIB_BUFFER_IS_TRACED: trace this buffer.
static int session_alloc_and_init(segment_manager_t *sm, transport_connection_t *tc, u8 alloc_fifos, stream_session_t **ret_s)
int svm_fifo_peek(svm_fifo_t *f, u32 relative_offset, u32 max_bytes, u8 *copy_here)
void session_register_transport(transport_proto_t transport_proto, const transport_proto_vft_t *vft, u8 is_ip4, u32 output_node)
Initialize session layer for given transport proto and ip version.
int session_lookup_add_half_open(transport_connection_t *tc, u64 value)
int stream_session_peek_bytes(transport_connection_t *tc, u8 *buffer, u32 offset, u32 max_bytes)
static_always_inline void clib_spinlock_lock_if_init(clib_spinlock_t *p)
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
application_t * application_get_if_valid(u32 index)
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
int stream_session_listen(stream_session_t *s, session_endpoint_t *sep)
Ask transport to listen on local transport endpoint.