107 u32 index, thread_index;
151 session_event_t *evt;
166 clib_warning (
"could not disconnect session: %llu app: %u",
221 u32 owner_app_wrk_map;
222 session_event_t *evt;
308 s =
format (s,
"SESSION_QUEUE: session index %d, server thread index %d",
313 #define foreach_session_queue_error \ 314 _(TX, "Packets transmitted") \ 315 _(TIMER, "Timer events") \ 316 _(NO_BUFFER, "Out of buffers") 320 #define _(sym,str) SESSION_QUEUE_ERROR_##sym, 327 #define _(sym,string) string, 341 u32 next_index,
u32 * to_next,
u16 n_segs,
348 for (i = 0; i <
clib_min (n_trace, n_segs); i++)
364 u32 chain_bi0, to_deq, left_from_seg;
366 u16 len_to_deq, n_bytes_read;
370 b->
flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
376 to_deq = left_from_seg;
405 ASSERT (n_bytes_read > 0);
418 ASSERT (n_bytes_read == len_to_deq);
424 prev_b->
flags |= VLIB_BUFFER_NEXT_PRESENT;
429 to_deq -= n_bytes_read;
450 b->
flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
460 ASSERT (n_bytes_read > 0);
479 ASSERT (n_bytes_read > 0);
497 ASSERT (n_bytes_read > 0);
556 u32 max_segs,
u8 peek_data)
558 u32 n_bytes_per_buf, n_bytes_per_seg;
625 int *n_tx_packets,
u8 peek_data)
627 u32 next_index, next0, next1, *to_next, n_left_to_next, max_burst;
628 u32 n_trace, n_bufs_needed = 0, n_left, pbi;
631 session_event_t *e = &elt->
evt;
645 next0 = next1 = next_index;
663 *n_tx_packets += n_custom_tx;
667 max_burst -= n_custom_tx;
706 SESSION_QUEUE_ERROR_NO_BUFFER, 1);
754 n_left_to_next, bi0, bi1, next0,
784 n_left_to_next, bi0, next0);
876 switch (e->event_type)
883 clib_warning (
"session %u was freed!", e->session_index);
921 (*fp) (e->rpc_args.arg);
941 clib_warning (
"unhandled event type %d", e->event_type);
960 int i, n_tx_packets = 0;
961 session_event_t *evt;
983 for (i = 0; i < n_to_dequeue; i++)
1008 clib_llist_remove (wrk->event_elts, evt_list, elt);
1009 session_event_dispatch (wrk, node, elt, thread_index, &n_tx_packets);
1021 session_evt_type_t et;
1023 et = elt->evt.event_type;
1024 clib_llist_remove (wrk->event_elts, evt_list, elt);
1027 if (n_tx_packets >= VLIB_FRAME_SIZE
1028 && (et == SESSION_IO_EVT_TX || et == SESSION_IO_EVT_TX_FLUSH))
1030 clib_llist_add (wrk->event_elts, evt_list, elt, new_he);
1059 SESSION_QUEUE_ERROR_TX, n_tx_packets);
1061 SESSION_EVT (SESSION_EVT_DISPATCH_END, wrk, n_tx_packets);
1063 return n_tx_packets;
1070 .name =
"session-queue",
1075 .state = VLIB_NODE_STATE_DISABLED,
1085 session_event_t _e, *e = &_e;
1093 index = mq->
q->head;
1095 for (i = 0; i < mq->
q->cursize; i++)
1101 switch (e->event_type)
1110 fformat (stdout,
"[%04d] disconnect session %d\n", i,
1120 fformat (stdout,
"[%04d] RPC call %llx with %llx\n",
1126 fformat (stdout,
"[%04d] unhandled event type %d\n",
1133 if (index == mq->
q->maxsize)
1142 switch (e->event_type)
1147 if (e->session_index == f->master_session_index)
1173 int i, index, found = 0;
1180 thread_index = f->master_thread_index;
1187 index = mq->
q->head;
1188 for (i = 0; i < mq->
q->cursize; i++)
1196 if (++index == mq->
q->maxsize)
1207 found = session_node_cmp_event (&elt->evt, f);
1210 clib_memcpy_fast (e, &elt->evt, sizeof (*e));
1243 f64 now, timeout = 1.0;
1244 uword *event_data = 0;
1278 .name =
"session-queue-process",
1279 .state = VLIB_NODE_STATE_DISABLED,
1298 .name =
"session-queue-main",
1299 .state = VLIB_NODE_STATE_DISABLED,
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.
vlib_main_t vlib_global_main
void * svm_msg_q_msg_data(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Get data for message in queue.
#define clib_llist_is_empty(LP, name, H)
Check is list is empty.
session_type_t session_type
Type built from transport and network protocol types.
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
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 f64 vlib_time_now(vlib_main_t *vm)
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.
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)
clib_memset(h->entries, 0, sizeof(h->entries[0])*entries)
static uword session_queue_node_fn(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
session_evt_elt_t * event_elts
Pool of session event list elements.
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)
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)
void transport_update_time(f64 time_now, u8 thread_index)
#define vec_reset_length(v)
Reset vector length to zero NULL-pointer tolerant.
#define vlib_worker_thread_barrier_sync(X)
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)
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)
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.
void vl_api_rpc_call_main_thread(void *fp, u8 *data, u32 data_length)
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.
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 * 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 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.
u32 wrk_index
Worker index in global worker pool.
u32 node_index
Node index.
#define vlib_validate_buffer_enqueue_x2(vm, node, next_index, to_next, n_left_to_next, bi0, bi1, next0, next1)
Finish enqueueing two buffers forward in the graph.
session_fifo_rx_fn ** session_tx_fns
Per transport rx function that can either dequeue or peek.
u64 session_segment_handle(session_t *s)
#define vlib_validate_buffer_enqueue_x1(vm, node, next_index, to_next, n_left_to_next, bi0, next0)
Finish enqueueing one buffer forward in the graph.
#define vlib_get_next_frame(vm, node, next_index, vectors, n_vectors_left)
Get pointer to next frame vector data by (vlib_node_runtime_t, next_index).
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 clib_llist_pop_first(LP, name, E, H)
Removes and returns the first element in the list.
The fine-grained event logger allows lightweight, thread-safe event logging at minimum cost...
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_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.
#define SESSION_CONN_HDR_LEN
int app_worker_builtin_rx(app_worker_t *app_wrk, session_t *s)
static void accepted_notify_cb(void *data, u32 data_len)
#define VLIB_MAIN_LOOP_EXIT_FUNCTION(x)
#define clib_warning(format, args...)
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)
void vlib_put_next_frame(vlib_main_t *vm, vlib_node_runtime_t *r, u32 next_index, u32 n_vectors_left)
Release pointer to next frame vector data.
blocking call - best used in combination with condvars, for eventfds we don't yield the cpu ...
static int svm_msg_q_try_lock(svm_msg_q_t *mq)
Try locking message queue.
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)
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.
enum _transport_proto transport_proto_t
#define clib_llist_elt_is_linked(E, name)
Check if element is linked in a list.
u32 transport_connection_snd_space(transport_connection_t *tc, u64 time_now, u16 mss)
Get maximum tx burst allowed for transport connection.
static uword pointer_to_uword(const void *p)
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)
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.
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)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
static void session_event_dispatch(session_worker_t *wrk, vlib_node_runtime_t *node, session_evt_elt_t *elt, u32 thread_index, int *n_tx_packets)
f64 last_vlib_time
vlib_time_now last time around the track
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
int app_worker_own_session(app_worker_t *app_wrk, session_t *s)
void vlib_worker_thread_barrier_release(vlib_main_t *vm)
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.
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)
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 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)
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.