37 u32 tries = 0, max_tries;
43 if (tries++ == max_tries)
61 evt->event_type = evt_type;
65 evt->rpc_args.fp = data;
66 evt->rpc_args.arg = args;
116 void (*fnp) (
void *) = fp;
142 memset (s, 0,
sizeof (*s));
144 s->thread_index = thread_index;
153 memset (s, 0xFA,
sizeof (*s));
167 svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
168 u32 fifo_segment_index;
173 &fifo_segment_index)))
176 server_rx_fifo->master_session_index = s->session_index;
177 server_rx_fifo->master_thread_index = s->thread_index;
179 server_tx_fifo->master_session_index = s->session_index;
180 server_tx_fifo->master_thread_index = s->thread_index;
182 s->server_rx_fifo = server_rx_fifo;
183 s->server_tx_fifo = server_tx_fifo;
184 s->svm_segment_index = fifo_segment_index;
192 u32 thread_index = tc->thread_index;
200 s->enqueue_epoch = (
u64) ~ 0;
203 s->connection_index = tc->c_index;
204 tc->s_index = s->session_index;
241 u32 to_drop = n_bytes_to_drop;
243 while (to_drop && (next->
flags & VLIB_BUFFER_NEXT_PRESENT))
271 u32 chain_bi, len, diff;
277 if (is_in_order && offset)
305 return (rv > 0) ? (written + rv) : written;
331 while ((chain_bi = (chain_b->
flags & VLIB_BUFFER_NEXT_PRESENT)
357 u8 queue_event,
u8 is_in_order)
360 int enqueued = 0, rv, in_order_off;
395 u32 thread_index = s->thread_index;
396 u64 enqueue_epoch = smm->current_enqueue_epoch[tc->proto][thread_index];
398 if (s->enqueue_epoch != enqueue_epoch)
400 s->enqueue_epoch = enqueue_epoch;
401 vec_add1 (smm->session_to_enqueue[tc->proto][thread_index],
412 session_dgram_hdr_t * hdr,
415 int enqueued = 0, rv, in_order_off;
436 u32 thread_index = s->thread_index;
437 u64 enqueue_epoch = smm->current_enqueue_epoch[proto][thread_index];
439 if (s->enqueue_epoch != enqueue_epoch)
441 s->enqueue_epoch = enqueue_epoch;
442 vec_add1 (smm->session_to_enqueue[proto][thread_index],
469 if (!s->server_tx_fifo)
479 return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
505 SESSION_DBG (
"invalid s->app_index = %d", s->app_wrk_index);
547 indices = smm->session_to_enqueue[transport_proto][thread_index];
549 for (i = 0; i <
vec_len (indices); i++)
562 smm->session_to_enqueue[transport_proto][thread_index] = indices;
563 smm->current_enqueue_epoch[transport_proto][thread_index]++;
585 u32 rx_pointer,
u32 tx_pointer)
596 u32 opaque = 0, new_ti, new_si;
622 opaque = tc->s_index;
639 new_s->app_wrk_index = app_wrk->
wrk_index;
640 new_si = new_s->session_index;
641 new_ti = new_s->thread_index;
648 if (app->
cb_fns.session_connected_callback (app_wrk->
wrk_index, opaque,
670 typedef struct _session_switch_pool_args
674 u32 new_thread_index;
675 u32 new_session_index;
685 s =
session_get (args->session_index, args->thread_index);
686 s->server_tx_fifo->master_session_index = args->new_session_index;
687 s->server_tx_fifo->master_thread_index = args->new_thread_index;
689 tp_vfts[tp].cleanup (s->connection_index, s->thread_index);
699 u32 old_thread_index,
709 new_s->connection_index = tc->c_index;
710 new_s->server_rx_fifo->master_session_index = new_s->session_index;
711 new_s->server_rx_fifo->master_thread_index = new_s->thread_index;
720 rpc_args->new_session_index = new_s->session_index;
721 rpc_args->new_thread_index = new_s->thread_index;
722 rpc_args->session_index = tc->s_index;
723 rpc_args->thread_index = old_thread_index;
727 tc->s_index = new_s->session_index;
728 new_s->connection_index = tc->c_index;
729 *new_session = new_s;
745 app->
cb_fns.session_accept_callback (s);
770 app->
cb_fns.session_disconnect_callback (s);
810 switch (s->session_state)
848 app->
cb_fns.session_reset_callback (s);
872 s->listener_index = listener_index;
879 app->
cb_fns.session_accept_callback (s);
897 rv =
tp_vfts[rmt->transport_proto].open (tep);
900 SESSION_DBG (
"Transport failed to open connection.");
901 return VNET_API_ERROR_SESSION_CONNECT;
904 tc =
tp_vfts[rmt->transport_proto].get_half_open ((
u32) rv);
918 app->
cb_fns.session_connected_callback (app_wrk->
wrk_index, opaque, s, 0);
932 rv =
tp_vfts[rmt->transport_proto].open (tep);
935 SESSION_DBG (
"Transport failed to open connection.");
936 return VNET_API_ERROR_SESSION_CONNECT;
939 tc =
tp_vfts[rmt->transport_proto].get_half_open ((
u32) rv);
948 handle = (((
u64) app_wrk_index) << 32) | (
u64) tc->c_index;
954 tc->s_index = opaque;
962 sep->app_wrk_index = app_wrk_index;
963 sep->opaque = opaque;
1011 u32 tc_index, s_index;
1015 s_index = ls->session_index;
1016 tc_index =
tp_vfts[sep->transport_proto].bind (s_index, tep);
1018 if (tc_index == (
u32) ~ 0)
1023 ls->connection_index = tc_index;
1026 tc =
tp_vfts[sep->transport_proto].get_listener (tc_index);
1047 tc =
tp_vfts[tp].get_listener (s->connection_index);
1051 return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
1055 tp_vfts[tp].unbind (s->connection_index);
1070 session_event_t *evt;
1089 vec_add2 (smm->pending_disconnects[s->thread_index], evt, 1);
1090 memset (evt, 0,
sizeof (*evt));
1174 u32 evt_q_length = 2048, evt_size =
sizeof (session_event_t);
1177 u64 eqs_size = 64 << 20;
1178 pid_t vpp_pid = getpid ();
1182 if (smm->configured_event_queue_length)
1183 evt_q_length = smm->configured_event_queue_length;
1185 if (smm->evt_qs_use_memfd_seg)
1187 if (smm->evt_qs_segment_size)
1188 eqs_size = smm->evt_qs_segment_size;
1193 eqs->
name =
format (0,
"%s%c",
"evt-qs-segment", 0);
1203 if (smm->evt_qs_use_memfd_seg)
1208 for (i = 0; i <
vec_len (smm->vpp_event_queues); i++)
1211 u32 notif_q_size =
clib_max (16, evt_q_length >> 4);
1213 {evt_q_length, evt_size, 0}
1215 {notif_q_size, 256, 0}
1222 if (smm->evt_qs_use_memfd_seg)
1229 if (smm->evt_qs_use_memfd_seg)
1239 if (smm->evt_qs_use_memfd_seg)
1240 return &smm->evt_qs_segment;
1249 session_tx_fifo_dequeue_and_snd
1266 u32 next_index = ~0;
1270 vec_validate (smm->session_type_to_next, session_type);
1274 if (output_node != ~0)
1284 smm->session_type_to_next[session_type] = next_index;
1285 smm->session_tx_fns[session_type] =
session_tx_fns[vft->tx_type];
1295 return tp_vfts[tp].get_connection (s->connection_index,
1305 return tp_vfts[tp].get_listener (s->connection_index);
1314 tc =
tp_vfts[tp].get_listener (listener->connection_index);
1322 sep->port = tc->lcl_port;
1323 sep->transport_proto = tc->proto;
1324 sep->is_ip4 = tc->is_ip4;
1342 u32 num_threads, preallocated_sessions_per_worker;
1347 if (num_threads < 1)
1353 vec_validate (smm->pending_event_vector, num_threads - 1);
1354 vec_validate (smm->pending_disconnects, num_threads - 1);
1355 vec_validate (smm->free_event_vector, num_threads - 1);
1362 vec_validate (smm->current_enqueue_epoch[i], num_threads - 1);
1363 vec_validate (smm->session_to_enqueue[i], num_threads - 1);
1364 for (j = 0; j < num_threads; j++)
1365 smm->current_enqueue_epoch[i][j] = 1;
1368 for (i = 0; i < num_threads; i++)
1371 _vec_len (smm->free_event_vector[i]) = 0;
1373 _vec_len (smm->pending_event_vector[i]) = 0;
1375 _vec_len (smm->pending_disconnects[i]) = 0;
1376 if (num_threads > 1)
1381 vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1388 sm_args->baseva = smm->session_baseva + smm->evt_qs_segment_size;
1389 sm_args->size = smm->session_va_space_size;
1393 if (smm->preallocated_sessions)
1395 if (num_threads == 1)
1402 preallocated_sessions_per_worker =
1403 (1.1 * (
f64) smm->preallocated_sessions /
1404 (
f64) (num_threads - 1));
1406 for (j = 1; j < num_threads; j++)
1409 preallocated_sessions_per_worker);
1418 smm->is_enabled = 1;
1429 u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1435 if (have_workers && ii == 0)
1485 smm->session_baseva = 0x200000000ULL;
1486 smm->session_va_space_size = (
u64) 128 << 30;
1487 smm->evt_qs_segment_size = 64 << 20;
1488 smm->is_enabled = 0;
1503 if (
unformat (input,
"event-queue-length %d", &nitems))
1506 smm->configured_event_queue_length = nitems;
1508 clib_warning (
"event queue length %d too small, ignored", nitems);
1510 else if (
unformat (input,
"preallocated-sessions %d",
1511 &smm->preallocated_sessions))
1513 else if (
unformat (input,
"v4-session-table-buckets %d",
1514 &smm->configured_v4_session_table_buckets))
1516 else if (
unformat (input,
"v4-halfopen-table-buckets %d",
1517 &smm->configured_v4_halfopen_table_buckets))
1519 else if (
unformat (input,
"v6-session-table-buckets %d",
1520 &smm->configured_v6_session_table_buckets))
1522 else if (
unformat (input,
"v6-halfopen-table-buckets %d",
1523 &smm->configured_v6_halfopen_table_buckets))
1525 else if (
unformat (input,
"v4-session-table-memory %U",
1528 if (tmp >= 0x100000000)
1531 smm->configured_v4_session_table_memory = tmp;
1533 else if (
unformat (input,
"v4-halfopen-table-memory %U",
1536 if (tmp >= 0x100000000)
1539 smm->configured_v4_halfopen_table_memory = tmp;
1541 else if (
unformat (input,
"v6-session-table-memory %U",
1544 if (tmp >= 0x100000000)
1547 smm->configured_v6_session_table_memory = tmp;
1549 else if (
unformat (input,
"v6-halfopen-table-memory %U",
1552 if (tmp >= 0x100000000)
1555 smm->configured_v6_halfopen_table_memory = tmp;
1557 else if (
unformat (input,
"local-endpoints-table-memory %U",
1560 if (tmp >= 0x100000000)
1563 smm->local_endpoints_table_memory = tmp;
1565 else if (
unformat (input,
"local-endpoints-table-buckets %d",
1566 &smm->local_endpoints_table_buckets))
1568 else if (
unformat (input,
"evt_qs_memfd_seg"))
1569 smm->evt_qs_use_memfd_seg = 1;
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
static session_open_service_fn session_open_srv_fns[TRANSPORT_N_SERVICES]
app_worker_t * app_worker_get(u32 wrk_index)
void session_flush_frames_main_thread(vlib_main_t *vm)
static u8 svm_msg_q_msg_is_invalid(svm_msg_q_msg_t *msg)
Check if message is invalid.
int session_lookup_del_connection(transport_connection_t *tc)
Delete transport connection from session table.
void * svm_msg_q_msg_data(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Get data for message in queue.
static u8 svm_msg_q_ring_is_full(svm_msg_q_t *mq, u32 ring_index)
int session_open(u32 app_wrk_index, session_endpoint_t *rmt, u32 opaque)
Ask transport to open connection to remote transport endpoint.
static void svm_pop_heap(void *oldheap)
void svm_fifo_init_pointers(svm_fifo_t *f, u32 pointer)
Set fifo pointers to requested offset.
#define SESSION_Q_PROCESS_FLUSH_FRAMES
static void clib_rwlock_writer_lock(clib_rwlock_t *p)
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
app_worker_t * application_listener_select_worker(stream_session_t *ls, u8 is_local)
#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)
#define vec_add2(V, P, N)
Add N elements to end of vector V, return pointer to new elements in P.
transport_tx_fn_type_t transport_protocol_tx_fn_type(transport_proto_t tp)
int session_open_vc(u32 app_wrk_index, session_endpoint_t *rmt, u32 opaque)
void svm_fifo_dequeue_drop_all(svm_fifo_t *f)
transport_connection_t * session_get_transport(stream_session_t *s)
static u32 svm_fifo_max_enqueue(svm_fifo_t *f)
ssvm_shared_header_t * sh
int session_lookup_del_session(stream_session_t *s)
void segment_manager_dealloc_fifos(u32 segment_index, svm_fifo_t *rx_fifo, svm_fifo_t *tx_fifo)
#define vec_validate_aligned(V, I, A)
Make sure vector is long enough for given index (no header, specified alignment)
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)
enum transport_service_type_ transport_service_type_t
static uword vlib_node_add_next(vlib_main_t *vm, uword node, uword next_node)
session_fifo_rx_fn session_tx_fifo_peek_and_snd
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
static session_fifo_rx_fn * session_tx_fns[TRANSPORT_TX_N_FNS]
struct _svm_fifo svm_fifo_t
void session_free(stream_session_t *s)
segment_manager_t * app_worker_get_listen_segment_manager(app_worker_t *app, stream_session_t *listener)
int session_stop_listen(stream_session_t *s)
Ask transport to stop listening on local transport endpoint.
int session_open_cl(u32 app_wrk_index, session_endpoint_t *rmt, u32 opaque)
memset(h->entries, 0, sizeof(h->entries[0])*entries)
static svm_msg_q_t * session_manager_get_vpp_event_queue(u32 thread_index)
void app_namespaces_init(void)
static clib_error_t * session_config_fn(vlib_main_t *vm, unformat_input_t *input)
svm_msg_q_t * svm_msg_q_alloc(svm_msg_q_cfg_t *cfg)
Allocate message queue.
#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)
int session_send_ctrl_evt_to_thread(stream_session_t *s, session_evt_type_t evt_type)
static stream_session_t * session_alloc_for_connection(transport_connection_t *tc)
int session_dequeue_notify(stream_session_t *s)
static stream_session_t * listen_session_get(u32 index)
static void * ssvm_push_heap(ssvm_shared_header_t *sh)
#define clib_error_return(e, args...)
svm_region_t * vlib_rp
Current binary api segment descriptor.
int svm_fifo_enqueue_nowait(svm_fifo_t *f, u32 max_bytes, const u8 *copy_from_here)
void stream_session_delete_notify(transport_connection_t *tc)
Notification from transport that connection is being deleted.
vhost_vring_state_t state
int session_enqueue_dgram_connection(stream_session_t *s, session_dgram_hdr_t *hdr, vlib_buffer_t *b, u8 proto, u8 queue_event)
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
int session_send_io_evt_to_thread(svm_fifo_t *f, session_evt_type_t evt_type)
void session_vpp_event_queues_allocate(session_manager_main_t *smm)
Allocate event queues in the shared-memory segment.
int ssvm_master_init(ssvm_private_t *ssvm, ssvm_segment_type_t type)
enum transport_dequeue_type_ transport_tx_fn_type_t
static void ssvm_pop_heap(void *oldheap)
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)
#define session_endpoint_to_transport(_sep)
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 segment_manager_alloc_session_fifos(segment_manager_t *sm, svm_fifo_t **rx_fifo, svm_fifo_t **tx_fifo, u32 *fifo_segment_index)
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()
static void clib_rwlock_init(clib_rwlock_t *p)
int svm_msg_q_alloc_consumer_eventfd(svm_msg_q_t *mq)
Allocate event fd for queue consumer.
transport_service_type_t transport_protocol_service_type(transport_proto_t tp)
u16 current_length
Nbytes between current data and the end of this buffer.
struct _session_endpoint session_endpoint_t
app_worker_t * app_worker_get_if_valid(u32 wrk_index)
static session_handle_t session_handle(stream_session_t *s)
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.
#define VLIB_CONFIG_FUNCTION(x, n,...)
u32 wrk_index
Worker index in global worker pool.
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)
void session_free_w_fifos(stream_session_t *s)
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.
int session_alloc_fifos(segment_manager_t *sm, stream_session_t *s)
#define pool_get_aligned(P, E, A)
Allocate an object E from a pool P (general version).
#define SESSION_DBG(_fmt, _args...)
static void clib_rwlock_writer_unlock(clib_rwlock_t *p)
u32 n_rings
number of msg rings
#define SESSION_EVT_DBG(_evt, _args...)
void transport_init(void)
session_fifo_rx_fn session_tx_fifo_dequeue_and_snd
static int session_enqueue_notify(stream_session_t *s)
Notify session peer that new data has been enqueued.
static_always_inline uword vlib_get_thread_index(void)
static vlib_process_t * vlib_get_current_process(vlib_main_t *vm)
static session_type_t session_type_from_proto_and_ip(transport_proto_t proto, u8 is_ip4)
static void svm_msg_q_unlock(svm_msg_q_t *mq)
Unlock message queue.
int session_listen(stream_session_t *ls, session_endpoint_extended_t *sep)
Ask transport to listen on session endpoint.
static void vlib_process_signal_event_mt(vlib_main_t *vm, uword node_index, uword type_opaque, uword data)
Signal event to process from any thread.
transport_service_type_t session_transport_service_type(stream_session_t *s)
#define clib_warning(format, args...)
static int session_send_evt_to_thread(void *data, void *args, u32 thread_index, session_evt_type_t evt_type)
void stream_session_init_fifos_pointers(transport_connection_t *tc, u32 rx_pointer, u32 tx_pointer)
Init fifo tail and head pointers.
void transport_enable_disable(vlib_main_t *vm, u8 is_en)
static u8 vlib_thread_is_main_w_barrier(void)
static int svm_msg_q_try_lock(svm_msg_q_t *mq)
Try locking message queue.
#define pool_init_fixed(pool, max_elts)
initialize a fixed-size, preallocated pool
application_t * application_get(u32 app_index)
void stream_session_disconnect_notify(transport_connection_t *tc)
Notification from transport that connection is being closed.
int app_worker_lock_and_send_event(app_worker_t *app, stream_session_t *s, u8 evt_type)
Send event to application.
#define SESSION_Q_PROCESS_STOP
ssvm_private_t * session_manager_get_evt_q_segment(void)
void svm_msg_q_add_and_unlock(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Producer enqueue one message to queue with mutex held.
u32 session_tx_fifo_max_dequeue(transport_connection_t *tc)
u32 next_buffer
Next buffer for this linked-list of buffers.
void segment_manager_main_init(segment_manager_main_init_args_t *a)
static void clib_mem_free(void *p)
session_cb_vft_t cb_fns
Callbacks: shoulder-taps for the server/client.
int listen_session_get_local_session_endpoint(stream_session_t *listener, session_endpoint_t *sep)
void stream_session_disconnect(stream_session_t *s)
Initialize session disconnect.
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)
static void * clib_mem_alloc(uword size)
svm_msg_q_ring_cfg_t * ring_cfgs
array of ring cfgs
static vlib_main_t * vlib_get_main(void)
stream_session_t * session_alloc(u32 thread_index)
#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 void session_switch_pool(void *cb_args)
int svm_fifo_dequeue_drop(svm_fifo_t *f, u32 max_bytes)
u32 q_nitems
msg queue size (not rings)
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)
session_fifo_rx_fn session_tx_fifo_dequeue_internal
int( session_fifo_rx_fn)(vlib_main_t *vm, vlib_node_runtime_t *node, session_event_t *e0, stream_session_t *s0, int *n_tx_pkts)
static stream_session_t * session_clone_safe(u32 session_index, u32 thread_index)
struct _transport_endpoint transport_endpoint_t
struct _segment_manager segment_manager_t
int session_send_io_evt_to_thread_custom(void *data, u32 thread_index, session_evt_type_t evt_type)
int session_manager_flush_all_enqueue_events(u8 transport_proto)
u32 app_index
Index of owning app.
int session_lookup_add_connection(transport_connection_t *tc, u64 value)
Add transport connection to a session table.
transport_tx_fn_type_t session_transport_tx_fn_type(stream_session_t *s)
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()
vlib_node_registration_t session_queue_process_node
(constructor) VLIB_REGISTER_NODE (session_queue_process_node)
static u32 vlib_num_workers()
static vlib_node_t * vlib_get_node(vlib_main_t *vm, u32 i)
Get vlib node by index.
static int session_enqueue_chain_tail(stream_session_t *s, vlib_buffer_t *b, u32 offset, u8 is_in_order)
Enqueue buffer chain tail.
u8 session_tx_is_dgram(stream_session_t *s)
int(* session_open_service_fn)(u32, session_endpoint_t *, u32)
int application_is_builtin_proxy(application_t *app)
segment_manager_t * app_worker_get_connect_segment_manager(app_worker_t *app)
int consumer_pid
pid of msg consumer
#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.
int session_open_app(u32 app_wrk_index, session_endpoint_t *rmt, u32 opaque)
static int session_alloc_and_init(segment_manager_t *sm, transport_connection_t *tc, u8 alloc_fifos, stream_session_t **ret_s)
void vlib_start_process(vlib_main_t *vm, uword process_index)
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)
u8 transport_protocol_is_cl(transport_proto_t tp)
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
svm_msg_q_msg_t svm_msg_q_alloc_msg_w_ring(svm_msg_q_t *mq, u32 ring_index)
Allocate message buffer on ring.
void stream_session_disconnect_transport(stream_session_t *s)
Notify transport the session can be disconnected.
struct _session_endpoint_extended session_endpoint_extended_t
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.