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;
115 void (*fnp) (
void *) = fp;
141 memset (s, 0,
sizeof (*s));
143 s->thread_index = thread_index;
152 memset (s, 0xFA,
sizeof (*s));
158 svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
159 u32 fifo_segment_index;
164 &fifo_segment_index)))
167 server_rx_fifo->master_session_index = s->session_index;
168 server_rx_fifo->master_thread_index = s->thread_index;
170 server_tx_fifo->master_session_index = s->session_index;
171 server_tx_fifo->master_thread_index = s->thread_index;
173 s->server_rx_fifo = server_rx_fifo;
174 s->server_tx_fifo = server_tx_fifo;
175 s->svm_segment_index = fifo_segment_index;
183 u32 thread_index = tc->thread_index;
191 s->enqueue_epoch = ~0;
194 s->connection_index = tc->c_index;
195 tc->s_index = s->session_index;
232 u32 to_drop = n_bytes_to_drop;
234 while (to_drop && (next->
flags & VLIB_BUFFER_NEXT_PRESENT))
262 u32 chain_bi, len, diff;
268 if (is_in_order && offset)
296 return (rv > 0) ? (written + rv) : written;
322 while ((chain_bi = (chain_b->
flags & VLIB_BUFFER_NEXT_PRESENT)
348 u8 queue_event,
u8 is_in_order)
351 int enqueued = 0, rv, in_order_off;
386 u32 thread_index = s->thread_index;
387 u32 enqueue_epoch = smm->current_enqueue_epoch[tc->proto][thread_index];
389 if (s->enqueue_epoch != enqueue_epoch)
391 s->enqueue_epoch = enqueue_epoch;
392 vec_add1 (smm->session_to_enqueue[tc->proto][thread_index],
393 s - smm->sessions[thread_index]);
403 session_dgram_hdr_t * hdr,
406 int enqueued = 0, rv, in_order_off;
427 u32 thread_index = s->thread_index;
428 u32 enqueue_epoch = smm->current_enqueue_epoch[proto][thread_index];
430 if (s->enqueue_epoch != enqueue_epoch)
432 s->enqueue_epoch = enqueue_epoch;
433 vec_add1 (smm->session_to_enqueue[proto][thread_index],
434 s - smm->sessions[thread_index]);
460 if (!s->server_tx_fifo)
470 return svm_fifo_peek (s->server_tx_fifo, offset, max_bytes, buffer);
496 clib_warning (
"invalid s->app_index = %d", s->app_index);
541 int i, errors = 0, lock;
545 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;
621 opaque = tc->s_index;
637 new_s->app_index = app->index;
638 new_si = new_s->session_index;
639 new_ti = new_s->thread_index;
646 if (app->cb_fns.session_connected_callback (app->index, opaque, new_s,
668 typedef struct _session_switch_pool_args
672 u32 new_thread_index;
673 u32 new_session_index;
683 s =
session_get (args->session_index, args->thread_index);
684 s->server_tx_fifo->master_session_index = args->new_session_index;
685 s->server_tx_fifo->master_thread_index = args->new_thread_index;
687 tp_vfts[tp].cleanup (s->connection_index, s->thread_index);
697 u32 old_thread_index,
707 new_s->connection_index = tc->c_index;
708 new_s->server_rx_fifo->master_session_index = new_s->session_index;
709 new_s->server_rx_fifo->master_thread_index = new_s->thread_index;
718 rpc_args->new_session_index = new_s->session_index;
719 rpc_args->new_thread_index = new_s->thread_index;
720 rpc_args->session_index = tc->s_index;
721 rpc_args->thread_index = old_thread_index;
725 tc->s_index = new_s->session_index;
726 new_s->connection_index = tc->c_index;
727 *new_session = new_s;
739 server->cb_fns.session_accept_callback (s);
757 server->cb_fns.session_disconnect_callback (s);
812 app->cb_fns.session_reset_callback (s);
835 s->app_index = server->index;
836 s->listener_index = listener_index;
842 server->cb_fns.session_accept_callback (s);
859 rv =
tp_vfts[rmt->transport_proto].open (tep);
862 SESSION_DBG (
"Transport failed to open connection.");
863 return VNET_API_ERROR_SESSION_CONNECT;
866 tc =
tp_vfts[rmt->transport_proto].get_half_open ((
u32) rv);
875 s->app_index = app->index;
879 app->cb_fns.session_connected_callback (app->index, opaque, s, 0);
893 rv =
tp_vfts[rmt->transport_proto].open (tep);
896 SESSION_DBG (
"Transport failed to open connection.");
897 return VNET_API_ERROR_SESSION_CONNECT;
900 tc =
tp_vfts[rmt->transport_proto].get_half_open ((
u32) rv);
909 handle = (((
u64) app_index) << 32) | (
u64) tc->c_index;
915 tc->s_index = opaque;
923 sep->app_index = app_index;
924 sep->opaque = opaque;
966 tci =
tp_vfts[sep->transport_proto].bind (s->session_index,
970 if (tci == (
u32) ~ 0)
974 s->connection_index = tci;
975 tc =
tp_vfts[sep->transport_proto].get_listener (tci);
995 tci =
tp_vfts[sep->transport_proto].bind (s->session_index,
999 if (tci == (
u32) ~ 0)
1003 s->connection_index = tci;
1004 tc =
tp_vfts[sep->transport_proto].get_listener (tci);
1025 esep.app_index = s->app_index;
1027 return tp_vfts[sep->transport_proto].bind (s->session_index,
1074 tc =
tp_vfts[tp].get_listener (s->connection_index);
1078 return VNET_API_ERROR_ADDRESS_NOT_IN_USE;
1082 tp_vfts[tp].unbind (s->connection_index);
1097 session_event_t *evt;
1115 || thread_index == s->thread_index)
1117 ASSERT (s->thread_index == thread_index || thread_index == 0);
1118 vec_add2 (smm->pending_disconnects[s->thread_index], evt, 1);
1119 memset (evt, 0,
sizeof (*evt));
1199 u32 evt_q_length = 2048, evt_size =
sizeof (session_event_t);
1202 u64 eqs_size = 64 << 20;
1203 pid_t vpp_pid = getpid ();
1207 if (smm->configured_event_queue_length)
1208 evt_q_length = smm->configured_event_queue_length;
1210 if (smm->evt_qs_use_memfd_seg)
1212 if (smm->evt_qs_segment_size)
1213 eqs_size = smm->evt_qs_segment_size;
1218 eqs->
name =
format (0,
"%s%c",
"evt-qs-segment", 0);
1228 if (smm->evt_qs_use_memfd_seg)
1233 for (i = 0; i <
vec_len (smm->vpp_event_queues); i++)
1236 u32 notif_q_size =
clib_max (16, evt_q_length >> 4);
1238 {evt_q_length, evt_size, 0}
1240 {notif_q_size, 256, 0}
1249 if (smm->evt_qs_use_memfd_seg)
1259 if (smm->evt_qs_use_memfd_seg)
1260 return &smm->evt_qs_segment;
1269 session_tx_fifo_dequeue_and_snd
1286 u32 next_index = ~0;
1290 vec_validate (smm->session_type_to_next, session_type);
1294 if (output_node != ~0)
1304 smm->session_type_to_next[session_type] = next_index;
1305 smm->session_tx_fns[session_type] =
session_tx_fns[vft->tx_type];
1315 return tp_vfts[tp].get_connection (s->connection_index,
1325 return tp_vfts[tp].get_listener (s->connection_index);
1334 tc =
tp_vfts[tp].get_listener (listener->connection_index);
1342 sep->port = tc->lcl_port;
1343 sep->transport_proto = tc->proto;
1344 sep->is_ip4 = tc->is_ip4;
1362 u32 num_threads, preallocated_sessions_per_worker;
1367 if (num_threads < 1)
1373 vec_validate (smm->pending_event_vector, num_threads - 1);
1374 vec_validate (smm->pending_disconnects, num_threads - 1);
1375 vec_validate (smm->free_event_vector, num_threads - 1);
1382 vec_validate (smm->current_enqueue_epoch[i], num_threads - 1);
1383 vec_validate (smm->session_to_enqueue[i], num_threads - 1);
1384 for (j = 0; j < num_threads; j++)
1385 smm->current_enqueue_epoch[i][j] = 1;
1388 for (i = 0; i < num_threads; i++)
1391 _vec_len (smm->free_event_vector[i]) = 0;
1393 _vec_len (smm->pending_event_vector[i]) = 0;
1395 _vec_len (smm->pending_disconnects[i]) = 0;
1396 if (num_threads > 1)
1401 vec_validate (smm->last_event_poll_by_thread, num_threads - 1);
1408 sm_args->baseva = smm->session_baseva + smm->evt_qs_segment_size;
1409 sm_args->size = smm->session_va_space_size;
1413 if (smm->preallocated_sessions)
1415 if (num_threads == 1)
1422 preallocated_sessions_per_worker =
1423 (1.1 * (
f64) smm->preallocated_sessions /
1424 (
f64) (num_threads - 1));
1426 for (j = 1; j < num_threads; j++)
1429 preallocated_sessions_per_worker);
1438 smm->is_enabled = 1;
1449 u8 state = is_en ? VLIB_NODE_STATE_POLLING : VLIB_NODE_STATE_DISABLED;
1455 if (have_workers && ii == 0)
1505 smm->session_baseva = 0x200000000ULL;
1506 smm->session_va_space_size = (
u64) 128 << 30;
1507 smm->evt_qs_segment_size = 64 << 20;
1508 smm->is_enabled = 0;
1523 if (
unformat (input,
"event-queue-length %d", &nitems))
1526 smm->configured_event_queue_length = nitems;
1528 clib_warning (
"event queue length %d too small, ignored", nitems);
1530 else if (
unformat (input,
"preallocated-sessions %d",
1531 &smm->preallocated_sessions))
1533 else if (
unformat (input,
"v4-session-table-buckets %d",
1534 &smm->configured_v4_session_table_buckets))
1536 else if (
unformat (input,
"v4-halfopen-table-buckets %d",
1537 &smm->configured_v4_halfopen_table_buckets))
1539 else if (
unformat (input,
"v6-session-table-buckets %d",
1540 &smm->configured_v6_session_table_buckets))
1542 else if (
unformat (input,
"v6-halfopen-table-buckets %d",
1543 &smm->configured_v6_halfopen_table_buckets))
1545 else if (
unformat (input,
"v4-session-table-memory %U",
1548 if (tmp >= 0x100000000)
1551 smm->configured_v4_session_table_memory = tmp;
1553 else if (
unformat (input,
"v4-halfopen-table-memory %U",
1556 if (tmp >= 0x100000000)
1559 smm->configured_v4_halfopen_table_memory = tmp;
1561 else if (
unformat (input,
"v6-session-table-memory %U",
1564 if (tmp >= 0x100000000)
1567 smm->configured_v6_session_table_memory = tmp;
1569 else if (
unformat (input,
"v6-halfopen-table-memory %U",
1572 if (tmp >= 0x100000000)
1575 smm->configured_v6_halfopen_table_memory = tmp;
1577 else if (
unformat (input,
"local-endpoints-table-memory %U",
1580 if (tmp >= 0x100000000)
1583 smm->local_endpoints_table_memory = tmp;
1585 else if (
unformat (input,
"local-endpoints-table-buckets %d",
1586 &smm->local_endpoints_table_buckets))
1588 else if (
unformat (input,
"evt_qs_memfd_seg"))
1589 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]
void session_flush_frames_main_thread(vlib_main_t *vm)
int session_open_vc(u32 app_index, session_endpoint_t *rmt, u32 opaque)
static session_listen_service_fn session_listen_srv_fns[TRANSPORT_N_SERVICES]
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)
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
int session_listen_cl(stream_session_t *s, session_endpoint_t *sep)
session_manager_main_t session_manager_main
#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).
int application_send_event(application_t *app, stream_session_t *s, u8 evt_type)
Send event to application.
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)
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)
int session_open_app(u32 app_index, session_endpoint_t *rmt, u32 opaque)
segment_manager_t * application_get_listen_segment_manager(application_t *app, stream_session_t *s)
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_listen_vc(stream_session_t *s, session_endpoint_t *sep)
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 session_open(u32 app_index, session_endpoint_t *rmt, u32 opaque)
Ask transport to open connection to remote transport endpoint.
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)
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
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,...)
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.
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_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.
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)
segment_manager_t * application_get_connect_segment_manager(application_t *app)
#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.
#define clib_memcpy(a, b, c)
struct _application application_t
void transport_enable_disable(vlib_main_t *vm, u8 is_en)
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
void stream_session_disconnect_notify(transport_connection_t *tc)
Notification from transport that connection is being closed.
#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)
static int session_enqueue_notify(stream_session_t *s, u8 lock)
Notify session peer that new data has been enqueued.
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 session_listen_app(stream_session_t *s, session_endpoint_t *sep)
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)
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)
application_t * application_get(u32 index)
struct _transport_endpoint transport_endpoint_t
int application_lock_and_send_event(application_t *app, stream_session_t *s, u8 evt_type)
struct _segment_manager segment_manager_t
int session_manager_flush_all_enqueue_events(u8 transport_proto)
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 session_open_cl(u32 app_index, session_endpoint_t *rmt, u32 opaque)
int session_send_io_evt_to_thread_custom(svm_fifo_t *f, u32 thread_index, session_evt_type_t evt_type)
int application_is_builtin_proxy(application_t *app)
int consumer_pid
pid of msg consumer
int(* session_listen_service_fn)(stream_session_t *, session_endpoint_t *)
#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)
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
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.
int stream_session_listen(stream_session_t *s, session_endpoint_t *sep)
Ask transport to listen on local transport endpoint.