28 #define app_check_thread_and_barrier(_fn, _arg) \ 29 if (!vlib_thread_is_main_w_barrier ()) \ 31 vlib_rpc_call_main_thread (_fn, (u8 *) _arg, sizeof(*_arg)); \ 51 a->sep.is_ip4 = mp->
is_ip4;
53 a->sep.port = mp->
port;
54 a->sep.fib_index = mp->
vrf;
56 a->sep.transport_proto = mp->
proto;
86 a->uri = (
char *) mp->
uri;
110 a->sep.is_ip4 = mp->
is_ip4;
112 a->sep.port = mp->
port;
113 a->sep.transport_proto = mp->
proto;
114 a->sep.peer.fib_index = mp->
vrf;
120 a->sep_ext.flags = mp->
flags;
157 a->uri = (
char *) mp->
uri;
256 (
u8 *) mp,
sizeof (*mp));
286 if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
303 u32 index, thread_index;
314 if (!s || s->
session_state < SESSION_STATE_TRANSPORT_CLOSING)
347 session_event_t *evt;
362 clib_warning (
"could not disconnect session: %llu app: %u",
417 u32 owner_app_wrk_map;
418 session_event_t *evt;
504 s =
format (s,
"session index %d thread index %d",
509 #define foreach_session_queue_error \ 510 _(TX, "Packets transmitted") \ 511 _(TIMER, "Timer events") \ 512 _(NO_BUFFER, "Out of buffers") 516 #define _(sym,str) SESSION_QUEUE_ERROR_##sym, 523 #define _(sym,string) string, 537 u32 next_index,
u32 * to_next,
u16 n_segs,
544 for (i = 0; i <
clib_min (n_trace, n_segs); i++)
560 u32 chain_bi0, to_deq, left_from_seg;
562 u16 len_to_deq, n_bytes_read;
566 b->
flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
572 to_deq = left_from_seg;
601 ASSERT (n_bytes_read > 0);
614 ASSERT (n_bytes_read == len_to_deq);
620 prev_b->
flags |= VLIB_BUFFER_NEXT_PRESENT;
625 to_deq -= n_bytes_read;
646 b->
flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
656 ASSERT (n_bytes_read > 0);
675 ASSERT (n_bytes_read > 0);
693 ASSERT (n_bytes_read > 0);
722 && (s->
flags & SESSION_F_CUSTOM_TX))
752 u32 max_segs,
u8 peek_data)
754 u32 n_bytes_per_buf, n_bytes_per_seg;
821 int *n_tx_packets,
u8 peek_data)
823 u32 n_trace, n_bufs_needed = 0, n_left, pbi, next_index, max_burst;
826 session_event_t *e = &elt->
evt;
853 if (ctx->
s->
flags & SESSION_F_CUSTOM_TX)
856 ctx->
s->
flags &= ~SESSION_F_CUSTOM_TX;
858 *n_tx_packets += n_custom_tx;
862 max_burst -= n_custom_tx;
897 snd_space - snd_space % ctx->
snd_mss : snd_space;
923 SESSION_QUEUE_ERROR_NO_BUFFER, 1);
1069 void (*fp) (
void *);
1076 switch (e->event_type)
1079 fp = e->rpc_args.fp;
1080 (*fp) (e->rpc_args.arg);
1132 clib_warning (
"unhandled event type %d", e->event_type);
1160 switch (e->event_type)
1195 clib_warning (
"unhandled event type %d", e->event_type);
1206 #define _(symc, sym) \ 1207 [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t), 1224 elt->
evt.event_type = evt->event_type;
1262 int i, n_tx_packets = 0;
1263 session_event_t *evt;
1286 for (i = 0; i < n_to_dequeue; i++)
1304 clib_llist_remove (wrk->event_elts, evt_list, elt);
1305 session_event_dispatch_ctrl (wrk, elt);
1319 session_evt_type_t et;
1321 et = elt->evt.event_type;
1322 clib_llist_remove (wrk->event_elts, evt_list, elt);
1325 if (n_tx_packets >= VLIB_FRAME_SIZE
1326 && (et == SESSION_IO_EVT_TX || et == SESSION_IO_EVT_TX_FLUSH))
1328 clib_llist_add (wrk->event_elts, evt_list, elt, new_he);
1340 if (old_ti != wrk->old_head)
1363 if (
vec_len (wrk->pending_tx_buffers))
1367 SESSION_QUEUE_ERROR_TX, n_tx_packets);
1369 SESSION_EVT (SESSION_EVT_DISPATCH_END, wrk, n_tx_packets);
1371 return n_tx_packets;
1379 .name =
"session-queue",
1384 .state = VLIB_NODE_STATE_DISABLED,
1394 session_event_t _e, *e = &_e;
1402 index = mq->
q->head;
1404 for (i = 0; i < mq->
q->cursize; i++)
1410 switch (e->event_type)
1419 fformat (stdout,
"[%04d] disconnect session %d\n", i,
1429 fformat (stdout,
"[%04d] RPC call %llx with %llx\n",
1435 fformat (stdout,
"[%04d] unhandled event type %d\n",
1442 if (index == mq->
q->maxsize)
1451 switch (e->event_type)
1458 if (e->session_index == f->master_session_index)
1484 int i, index, found = 0;
1491 thread_index = f->master_thread_index;
1498 index = mq->
q->head;
1499 for (i = 0; i < mq->
q->cursize; i++)
1507 index = (index + 1) % mq->
q->maxsize;
1517 found = session_node_cmp_event (&elt->evt, f);
1520 clib_memcpy_fast (e, &elt->evt, sizeof (*e));
1530 found = session_node_cmp_event (&elt->evt, f);
1533 clib_memcpy_fast (e, &elt->evt, sizeof (*e));
1566 f64 now, timeout = 1.0;
1567 uword *event_data = 0;
1601 .name =
"session-queue-process",
1602 .state = VLIB_NODE_STATE_DISABLED,
1622 .name =
"session-queue-main",
1623 .state = VLIB_NODE_STATE_DISABLED,
#define vec_validate(V, I)
Make sure vector is long enough for given index (no header, unspecified alignment) ...
static void session_tx_trace_frame(vlib_main_t *vm, vlib_node_runtime_t *node, u32 next_index, u32 *to_next, u16 n_segs, session_t *s, u32 n_trace)
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 app_worker_lock_and_send_event(app_worker_t *app, session_t *s, u8 evt_type)
Send event to application.
u32 connection_index
Index of the transport connection associated to the session.
#define ENDPOINT_INVALID_INDEX
vlib_main_t vlib_global_main
void session_transport_reset(session_t *s)
Force transport close.
void * svm_msg_q_msg_data(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Get data for message in queue.
static void app_mq_detach_handler(void *data)
session_type_t session_type
Type built from transport and network protocol types.
static void session_mq_unlisten_handler(void *data)
void ip_copy(ip46_address_t *dst, ip46_address_t *src, u8 is_ip4)
#define SESSION_Q_PROCESS_FLUSH_FRAMES
static u32 vlib_get_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt)
svm_msg_q_t * vpp_event_queue
vpp event message queue for worker
static f64 vlib_process_wait_for_event_or_clock(vlib_main_t *vm, f64 dt)
Suspend a cooperative multi-tasking thread Waits for an event, or for the indicated number of seconds...
vlib_node_registration_t session_queue_pre_input_node
(constructor) VLIB_REGISTER_NODE (session_queue_pre_input_node)
static void vlib_buffer_free(vlib_main_t *vm, u32 *buffers, u32 n_buffers)
Free buffers Frees the entire buffer chain for each buffer.
session_main_t session_main
struct _vnet_connect_args vnet_connect_args_t
struct _vnet_unlisten_args_t vnet_unlisten_args_t
#define clib_llist_next_index(E, name)
u32 session_index
Index in thread pool where session was allocated.
i16 current_data
signed offset in data[], pre_data[] that we are currently processing.
#define clib_memcpy_fast(a, b, c)
static int svm_fifo_is_empty_prod(svm_fifo_t *f)
Check if fifo is empty optimized for producer.
static void session_event_dispatch_ctrl(session_worker_t *wrk, session_evt_elt_t *elt)
clib_memset(h->entries, 0, sizeof(h->entries[0]) *entries)
#define VLIB_NODE_FLAG_TRACE_SUPPORTED
static f64 vlib_time_now(vlib_main_t *vm)
static void session_mq_listen_uri_handler(void *data)
svm_fifo_t * rx_fifo
Pointers to rx/tx buffers.
session_worker_t * wrk
Worker contexts.
static session_t * session_get_if_valid(u64 si, u32 thread_index)
u16 current_length
Nbytes between current data and the end of this buffer.
#define vec_add1(V, E)
Add 1 element to end of vector (unspecified alignment).
static transport_proto_t session_get_transport_proto(session_t *s)
application_t * application_lookup(u32 api_client_index)
int svm_fifo_peek(svm_fifo_t *f, u32 offset, u32 len, u8 *dst)
Peek data from fifo.
u32 wrk_map_index
Worker index in app's map pool.
static clib_error_t * session_queue_exit(vlib_main_t *vm)
static uword session_queue_node_fn(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
int vnet_unlisten(vnet_unlisten_args_t *a)
session_evt_elt_t * event_elts
Pool of session event list elements.
static void * session_evt_ctrl_data(session_worker_t *wrk, session_evt_elt_t *elt)
void svm_fifo_overwrite_head(svm_fifo_t *f, u8 *src, u32 len)
Overwrite fifo head with new data.
#define vec_validate_aligned(V, I, A)
Make sure vector is long enough for given index (no header, specified alignment)
int session_tx_fifo_dequeue_and_snd(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *e, int *n_tx_packets)
void session_node_enable_disable(u8 is_en)
vlib_main_t ** vlib_mains
static void session_parse_handle(session_handle_t handle, u32 *index, u32 *thread_index)
#define clib_llist_foreach(LP, name, H, E, body)
Walk list starting at head.
app_worker_t * application_get_worker(application_t *app, u32 wrk_map_index)
u16 * pending_tx_nexts
Vector of nexts for the pending tx buffers.
static void session_tx_fill_buffer(vlib_main_t *vm, session_tx_context_t *ctx, vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
static u32 session_evt_ctrl_data_alloc(session_worker_t *wrk)
struct _vnet_bind_args_t vnet_listen_args_t
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
#define vlib_worker_thread_barrier_sync(X)
void transport_update_time(clib_time_type_t time_now, u8 thread_index)
struct _svm_fifo svm_fifo_t
static void vlib_trace_buffer(vlib_main_t *vm, vlib_node_runtime_t *r, u32 next_index, vlib_buffer_t *b, int follow_chain)
static void session_mq_disconnect_handler(void *data)
u32 * tx_buffers
Vector of tx buffer free lists.
int svm_msg_q_lock_and_alloc_msg_w_ring(svm_msg_q_t *mq, u32 ring_index, u8 noblock, svm_msg_q_msg_t *msg)
Lock message queue and allocate message buffer on ring.
static int svm_fifo_is_empty(svm_fifo_t *f)
Check if fifo is empty.
static session_worker_t * session_main_get_worker(u32 thread_index)
void dump_thread_0_event_queue(void)
#define static_always_inline
vlib_main_t * vm
Convenience pointer to this worker's vlib_main.
static uword vlib_process_get_events(vlib_main_t *vm, uword **data_vector)
Return the first event type which has occurred and a vector of per-event data of that type...
struct _vnet_disconnect_args_t vnet_disconnect_args_t
u8 session_node_lookup_fifo_event(svm_fifo_t *f, session_event_t *e)
static void session_mq_connect_uri_handler(void *data)
clib_llist_index_t new_head
Head of list of elements.
static u32 svm_fifo_max_dequeue_cons(svm_fifo_t *f)
Fifo max bytes to dequeue optimized for consumer.
u32 * session_type_to_next
Per session type output nodes.
#define vlib_prefetch_buffer_header(b, type)
Prefetch buffer metadata.
int svm_fifo_dequeue(svm_fifo_t *f, u32 len, u8 *dst)
Dequeue data from fifo.
int session_send_io_evt_to_thread(svm_fifo_t *f, session_evt_type_t evt_type)
static char * session_queue_error_strings[]
int session_dequeue_notify(session_t *s)
transport_proto_vft_t * transport_protocol_get_vft(transport_proto_t transport_proto)
Get transport virtual function table.
vlib_error_t error
Error code for buffers to be enqueued to error handler.
#define TRANSPORT_MAX_HDRS_LEN
#define pool_elt_at_index(p, i)
Returns pointer to element at given index.
#define CLIB_US_TIME_FREQ
static void * vlib_buffer_make_headroom(vlib_buffer_t *b, u8 size)
Make head room, typically for packet headers.
static session_t * session_get_from_handle(session_handle_t handle)
transport_proto_vft_t * transport_vft
static void session_evt_ctrl_data_free(session_worker_t *wrk, session_evt_elt_t *elt)
static void * vlib_buffer_get_current(vlib_buffer_t *b)
Get pointer to current data to process.
static uword session_queue_process(vlib_main_t *vm, vlib_node_runtime_t *rt, vlib_frame_t *f)
int mq_send_session_bound_cb(u32 app_wrk_index, u32 api_context, session_handle_t handle, int rv)
int session_tx_fifo_peek_and_snd(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *e, int *n_tx_packets)
static u8 session_node_cmp_event(session_event_t *e, svm_fifo_t *f)
svm_msg_q_ring_t * svm_msg_q_ring(svm_msg_q_t *mq, u32 ring_index)
Get message queue ring.
static void svm_fifo_unset_event(svm_fifo_t *f)
Unset fifo event flag.
static u8 transport_connection_is_tx_paced(transport_connection_t *tc)
Check if transport connection is paced.
u32 wrk_index
Worker index in global worker pool.
u32 node_index
Node index.
static u8 * format_vnet_api_errno(u8 *s, va_list *args)
session_fifo_rx_fn ** session_tx_fns
Per transport rx function that can either dequeue or peek.
u64 session_segment_handle(session_t *s)
static void vlib_node_increment_counter(vlib_main_t *vm, u32 node_index, u32 counter_index, u64 increment)
static session_t * session_get_from_handle_if_valid(session_handle_t handle)
static_always_inline u32 vlib_buffer_get_default_data_size(vlib_main_t *vm)
static u8 svm_fifo_set_event(svm_fifo_t *f)
Set fifo event flag.
#define foreach_session_ctrl_evt
The fine-grained event logger allows lightweight, thread-safe event logging at minimum cost...
int mq_send_session_connected_cb(u32 app_wrk_index, u32 api_context, session_t *s, u8 is_fail)
static void session_evt_elt_free(session_worker_t *wrk, session_evt_elt_t *elt)
#define VLIB_REGISTER_NODE(x,...)
static u8 svm_fifo_needs_deq_ntf(svm_fifo_t *f, u32 n_last_deq)
Check if fifo needs dequeue notification.
static void session_tx_fifo_chain_tail(vlib_main_t *vm, session_tx_context_t *ctx, vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
static void session_evt_add_to_list(session_worker_t *wrk, session_event_t *evt)
static_always_inline uword vlib_get_thread_index(void)
#define CLIB_PREFETCH(addr, size, type)
static void svm_msg_q_unlock(svm_msg_q_t *mq)
Unlock message queue.
static void session_event_dispatch_io(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, u32 thread_index, int *n_tx_packets)
static_always_inline void vlib_buffer_enqueue_to_next(vlib_main_t *vm, vlib_node_runtime_t *node, u32 *buffers, u16 *nexts, uword count)
#define vec_free(V)
Free vector's memory (no header).
#define SESSION_CONN_HDR_LEN
int app_worker_builtin_rx(app_worker_t *app_wrk, session_t *s)
#define VLIB_MAIN_LOOP_EXIT_FUNCTION(x)
#define clib_warning(format, args...)
static vlib_node_runtime_t * vlib_node_get_runtime(vlib_main_t *vm, u32 node_index)
Get node runtime by node index.
svm_queue_t * q
queue for exchanging messages
struct _transport_connection transport_connection_t
static session_evt_elt_t * session_evt_alloc_new(session_worker_t *wrk)
static_always_inline uword session_queue_pre_input_inline(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
blocking call - best used in combination with condvars, for eventfds we don't yield the cpu ...
u32 * pending_tx_buffers
Vector of buffers to be sent.
static int svm_msg_q_try_lock(svm_msg_q_t *mq)
Try locking message queue.
vlib_main_t vlib_node_runtime_t * node
static u32 session_thread_from_handle(session_handle_t handle)
#define SESSION_Q_PROCESS_STOP
u32 ring_index
ring index, could be u8
#define foreach_session_queue_error
static void session_mq_disconnected_reply_handler(void *data)
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.
static session_t * session_event_get_session(session_event_t *e, u8 thread_index)
static int transport_custom_tx(transport_proto_t tp, void *s, u32 max_burst_size)
void transport_connection_tx_pacer_reset_bucket(transport_connection_t *tc, u32 bucket)
Reset tx pacer bucket.
int ct_session_connect_notify(session_t *ss)
session_tx_context_t ctx
Context for session tx.
void transport_connection_update_tx_bytes(transport_connection_t *tc, u32 bytes)
Update tx bytes for paced transport connection.
int vnet_listen(vnet_listen_args_t *a)
int vnet_application_detach(vnet_app_detach_args_t *a)
Detach application from vpp.
static u32 transport_connection_snd_space(transport_connection_t *tc)
Get maximum tx burst allowed for transport connection.
#define TRANSPORT_PACER_MIN_BURST
enum _transport_proto transport_proto_t
#define clib_llist_elt_is_linked(E, name)
Check if element is linked in a list.
clib_time_type_t last_vlib_time
vlib_time_now last time around the track
struct _vnet_app_detach_args_t vnet_app_detach_args_t
static uword pointer_to_uword(const void *p)
int vnet_connect(vnet_connect_args_t *a)
static u8 * format_session_queue_trace(u8 *s, va_list *args)
static int transport_app_rx_evt(transport_proto_t tp, u32 conn_index, u32 thread_index)
u8 thread_index
Index of the thread that allocated the session.
static void * vlib_add_trace(vlib_main_t *vm, vlib_node_runtime_t *r, vlib_buffer_t *b, u32 n_data_bytes)
static void session_tx_set_dequeue_params(vlib_main_t *vm, session_tx_context_t *ctx, u32 max_segs, u8 peek_data)
u32 app_index
App index in app pool.
struct _vlib_node_registration vlib_node_registration_t
template key/value backing page structure
app_worker_t * app_worker_get(u32 wrk_index)
void svm_msg_q_free_msg(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Free message buffer.
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
void vlib_rpc_call_main_thread(void *callback, u8 *args, u32 arg_size)
static u8 session_tx_not_ready(session_t *s, u8 peek_data)
clib_llist_index_t old_head
Head of list of pending events.
u32 next_buffer
Next buffer for this linked-list of buffers.
volatile u8 session_state
State in session layer state machine.
#define VLIB_BUFFER_TRACE_TRAJECTORY_INIT(b)
VLIB buffer representation.
int vnet_bind_uri(vnet_listen_args_t *a)
static const u32 session_evt_msg_sizes[]
static void session_mq_worker_update_handler(void *data)
static void session_evt_add_old(session_worker_t *wrk, session_evt_elt_t *elt)
int vnet_disconnect_session(vnet_disconnect_args_t *a)
static void session_mq_accepted_reply_handler(void *data)
static void session_evt_add_head_old(session_worker_t *wrk, session_evt_elt_t *elt)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
static session_evt_elt_t * session_evt_alloc_ctrl(session_worker_t *wrk)
u32 elsize
size of an element
struct clib_bihash_value offset
template key/value backing page structure
static void session_mq_disconnected_handler(void *data)
static u8 session_has_transport(session_t *s)
#define clib_llist_prev_index(E, name)
static transport_connection_t * session_tx_get_transport(session_tx_context_t *ctx, u8 peek_data)
u32 app_index
Index of owning app.
transport_connection_t * tc
static void session_flush_pending_tx_buffers(session_worker_t *wrk, vlib_node_runtime_t *node)
int app_worker_own_session(app_worker_t *app_wrk, session_t *s)
void vlib_worker_thread_barrier_release(vlib_main_t *vm)
clib_us_time_t last_vlib_us_time
vlib_time_now rounded to us precision and as u64
static u32 vlib_num_workers()
static u32 svm_msg_q_size(svm_msg_q_t *mq)
Check length of message queue.
u32 app_wrk_index
Index of the app worker that owns the session.
vlib_main_t vlib_node_runtime_t vlib_frame_t * frame
static void session_mq_listen_handler(void *data)
static int session_tx_fifo_read_and_snd_i(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, int *n_tx_packets, u8 peek_data)
#define clib_llist_remove(LP, name, E)
Remove entry from list.
static session_main_t * vnet_get_session_main()
void session_transport_close(session_t *s)
Notify transport the session can be disconnected.
clib_llist_index_t ctrl_head
Head of control events list.
int app_worker_close_notify(app_worker_t *app_wrk, session_t *s)
static void vlib_set_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt, u32 count)
#define CLIB_CACHE_LINE_BYTES
u32 total_length_not_including_first_buffer
Only valid for first buffer in chain.
#define clib_llist_foreach_safe(LP, name, H, E, body)
Walk list starting at head safe.
int vnet_connect_uri(vnet_connect_args_t *a)
int svm_fifo_dequeue_drop(svm_fifo_t *f, u32 len)
Dequeue and drop bytes from fifo.
static u32 vlib_buffer_alloc(vlib_main_t *vm, u32 *buffers, u32 n_buffers)
Allocate buffers into supplied array.
vlib_node_registration_t session_queue_process_node
(constructor) VLIB_REGISTER_NODE (session_queue_process_node)
svm_msg_q_t * event_queue
Application listens for events on this svm queue.
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
#define SESSION_EVT(_evt, _args...)
static void session_mq_reset_reply_handler(void *data)
u32 transport_connection_tx_pacer_burst(transport_connection_t *tc)
Get tx pacer max burst.
static void session_mq_connect_handler(void *data)
void mq_send_unlisten_reply(app_worker_t *app_wrk, session_handle_t sh, u32 context, int rv)
#define app_check_thread_and_barrier(_fn, _arg)
int session_tx_fifo_dequeue_internal(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *e, int *n_tx_packets)
#define clib_llist_entry_index(LP, E)
Get list entry index.
void svm_msg_q_sub_w_lock(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Consumer dequeue one message from queue with mutex held.