57 clib_warning (
"server %u doesn't own local handle 0x%lx",
78 old_state = s->session_state;
87 app->
cb_fns.session_disconnect_callback (s);
88 s->session_state = old_state;
102 u32 index, thread_index;
146 session_event_t *evt;
161 clib_warning (
"could not disconnect session: %llu app: %u",
217 u32 owner_app_wrk_map;
218 session_event_t *evt;
287 app->
cb_fns.session_disconnect_callback (s);
306 s =
format (s,
"SESSION_QUEUE: session index %d, server thread index %d",
311 #define foreach_session_queue_error \ 312 _(TX, "Packets transmitted") \ 313 _(TIMER, "Timer events") \ 314 _(NO_BUFFER, "Out of buffers") 318 #define _(sym,str) SESSION_QUEUE_ERROR_##sym, 325 #define _(sym,string) string, 339 u32 next_index,
u32 * to_next,
u16 n_segs,
346 for (i = 0; i <
clib_min (n_trace, n_segs); i++)
362 u32 chain_bi0, to_deq, left_from_seg;
364 u16 len_to_deq, n_bytes_read;
368 b->
flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
374 to_deq = left_from_seg;
398 session_dgram_hdr_t *hdr = &ctx->
hdr;
400 deq_now =
clib_min (hdr->data_length - hdr->data_offset,
404 ASSERT (n_bytes_read > 0);
406 hdr->data_offset += n_bytes_read;
407 if (hdr->data_offset == hdr->data_length)
417 ASSERT (n_bytes_read == len_to_deq);
423 prev_b->
flags |= VLIB_BUFFER_NEXT_PRESENT;
428 to_deq -= n_bytes_read;
440 u32 thread_index,
u16 * n_bufs,
u32 wanted)
463 b->
flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
473 ASSERT (n_bytes_read > 0);
482 session_dgram_hdr_t *hdr = &ctx->
hdr;
487 ASSERT (hdr->data_length > hdr->data_offset);
488 deq_now =
clib_min (hdr->data_length - hdr->data_offset,
492 ASSERT (n_bytes_read > 0);
496 ip_copy (&ctx->
tc->rmt_ip, &hdr->rmt_ip, ctx->
tc->is_ip4);
497 ctx->
tc->rmt_port = hdr->rmt_port;
499 hdr->data_offset += n_bytes_read;
500 if (hdr->data_offset == hdr->data_length)
510 ASSERT (n_bytes_read > 0);
524 ed->data[0] = FIFO_EVENT_APP_TX;
525 ed->data[1] = ctx->max_dequeue;
526 ed->data[2] = len_to_deq;
527 ed->data[3] = ctx->left_to_snd;
552 return ctx->
transport_vft->get_connection (ctx->
s->connection_index,
553 ctx->
s->thread_index);
558 return ctx->
transport_vft->get_listener (ctx->
s->connection_index);
561 return ctx->
transport_vft->get_connection (ctx->
s->connection_index,
562 ctx->
s->thread_index);
569 u32 max_segs,
u8 peek_data)
571 u32 n_bytes_per_buf, n_bytes_per_seg;
636 session_event_t * e,
int *n_tx_packets,
639 u32 next_index, next0, next1, *to_next, n_left_to_next;
656 next0 = next1 = next_index;
695 if (n_bufs < n_bufs_needed)
750 n_left_to_next, bi0, bi1, next0,
780 n_left_to_next, bi0, next0);
815 session_event_t * e,
int *n_tx_pkts)
823 session_event_t * e,
int *n_tx_pkts)
832 session_event_t * e,
int *n_tx_pkts)
841 return app->
cb_fns.builtin_app_tx_callback (s);
869 session_event_t *e, *fifo_events;
872 int n_tx_packets = 0,
i, rv;
899 for (
i = 0;
i < n_to_dequeue;
i++)
918 n_events =
vec_len (fifo_events);
922 for (
i = 0;
i < n_events;
i++)
929 switch (e->event_type)
961 SESSION_QUEUE_ERROR_NO_BUFFER, 1);
975 || (e->postponed < 200
992 app->
cb_fns.builtin_app_rx_callback (s);
1002 fp = e->rpc_args.fp;
1003 (*fp) (e->rpc_args.arg);
1023 clib_warning (
"unhandled event type %d", e->event_type);
1027 _vec_len (fifo_events) = 0;
1032 SESSION_QUEUE_ERROR_TX, n_tx_packets);
1036 return n_tx_packets;
1043 .name =
"session-queue",
1048 .state = VLIB_NODE_STATE_DISABLED,
1058 session_event_t _e, *e = &_e;
1066 index = mq->
q->head;
1068 for (i = 0; i < mq->
q->cursize; i++)
1074 switch (e->event_type)
1078 fformat (stdout,
"[%04d] TX session %d\n", i, s0->session_index);
1083 fformat (stdout,
"[%04d] disconnect session %d\n", i,
1089 fformat (stdout,
"[%04d] builtin_rx %d\n", i, s0->session_index);
1093 fformat (stdout,
"[%04d] RPC call %llx with %llx\n",
1099 fformat (stdout,
"[%04d] unhandled event type %d\n",
1106 if (index == mq->
q->maxsize)
1115 switch (e->event_type)
1132 if (s->server_rx_fifo == f || s->server_tx_fifo == f)
1144 session_event_t *pending_event_vector, *evt;
1146 int i, index, found = 0;
1153 thread_index = f->master_thread_index;
1160 index = mq->
q->head;
1161 for (i = 0; i < mq->
q->cursize; i++)
1169 if (++index == mq->
q->maxsize)
1211 f64 now, timeout = 1.0;
1212 uword *event_data = 0;
1246 .name =
"session-queue-process",
1247 .state = VLIB_NODE_STATE_DISABLED,
static int session_output_try_get_buffers(vlib_main_t *vm, session_manager_worker_t *wrk, u32 thread_index, u16 *n_bufs, u32 wanted)
f64 dispatch_period
Our approximation of a "complete" dispatch loop period.
app_worker_t * app_worker_get(u32 wrk_index)
static u64 session_segment_handle(stream_session_t *s)
vlib_main_t vlib_global_main
static void session_tx_trace_frame(vlib_main_t *vm, vlib_node_runtime_t *node, u32 next_index, u32 *to_next, u16 n_segs, stream_session_t *s, u32 n_trace)
void * svm_msg_q_msg_data(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Get data for message in queue.
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)
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...
#define SESSION_CONN_HDR_LEN
struct _transport_connection transport_connection_t
#define clib_memcpy_fast(a, b, c)
session_manager_main_t session_manager_main
clib_memset(h->entries, 0, sizeof(h->entries[0]) *entries)
static f64 vlib_time_now(vlib_main_t *vm)
int app_worker_own_session(app_worker_t *app_wrk, stream_session_t *s)
local_session_t * application_get_local_session_from_handle(session_handle_t handle)
#define vec_add1(V, E)
Add 1 element to end of vector (unspecified alignment).
application_t * application_lookup(u32 api_client_index)
session_event_t * pending_event_vector
Vector of active event vectors.
void svm_fifo_overwrite_head(svm_fifo_t *f, u8 *data, u32 len)
u32 wrk_map_index
Worker index in app's map pool.
#define vec_add2(V, P, N)
Add N elements to end of vector V, return pointer to new elements in P.
session_event_t * postponed_event_vector
Vector of postponed events.
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)
static void session_parse_handle(session_handle_t handle, u32 *index, u32 *thread_index)
#define vec_validate_aligned(V, I, A)
Make sure vector is long enough for given index (no header, specified alignment)
static stream_session_t * session_get_from_handle(session_handle_t handle)
void session_node_enable_disable(u8 is_en)
vlib_main_t ** vlib_mains
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.
static stream_session_t * session_event_get_session(session_event_t *e, u8 thread_index)
session_event_t * pending_disconnects
Vector of postponed disconnects.
#define vlib_worker_thread_barrier_sync(X)
struct _svm_fifo svm_fifo_t
int session_tx_fifo_peek_and_snd(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_worker_t *wrk, session_event_t *e, int *n_tx_pkts)
static void vlib_trace_buffer(vlib_main_t *vm, vlib_node_runtime_t *r, u32 next_index, vlib_buffer_t *b, int follow_chain)
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)
void dump_thread_0_event_queue(void)
i16 current_data
signed offset in data[], pre_data[] that we are currently processing.
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)
int application_local_session_connect_notify(local_session_t *ls)
static u32 svm_fifo_max_dequeue(svm_fifo_t *f)
int session_dequeue_notify(stream_session_t *s)
#define vlib_prefetch_buffer_header(b, type)
Prefetch buffer metadata.
struct _stream_session_t stream_session_t
int session_send_io_evt_to_thread(svm_fifo_t *f, session_evt_type_t evt_type)
int session_tx_fifo_dequeue_and_snd(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_worker_t *wrk, session_event_t *e, int *n_tx_pkts)
static char * session_queue_error_strings[]
transport_proto_vft_t * transport_protocol_get_vft(transport_proto_t transport_proto)
Get transport virtual function table.
session_event_t * free_event_vector
Vector of partially read events.
static u8 svm_fifo_needs_tx_ntf(svm_fifo_t *f, u32 n_last_deq)
static transport_proto_t session_get_transport_proto(stream_session_t *s)
f64 last_vlib_time
vlib_time_now last time around the track
static session_manager_main_t * vnet_get_session_manager_main()
u16 current_length
Nbytes between current data and the end of this buffer.
static void * vlib_buffer_make_headroom(vlib_buffer_t *b, u8 size)
Make head room, typically for packet headers.
transport_proto_vft_t * transport_vft
static u8 session_handle_is_local(session_handle_t handle)
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)
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)
Unsets fifo event flag.
u32 wrk_index
Worker index in global worker pool.
static stream_session_t * session_get_if_valid(u64 si, u32 thread_index)
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.
#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).
volatile u8 session_state
State.
vlib_error_t error
Error code for buffers to be enqueued to error handler.
static void vlib_node_increment_counter(vlib_main_t *vm, u32 node_index, u32 counter_index, u64 increment)
static u8 svm_fifo_set_event(svm_fifo_t *f)
Sets fifo event flag.
The fine-grained event logger allows lightweight, thread-safe event logging at minimum cost...
#define SESSION_EVT_DBG(_evt, _args...)
session_tx_context_t ctx
Context for session tx.
#define VLIB_REGISTER_NODE(x,...)
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 session_manager_worker_t * session_manager_get_worker(u32 thread_index)
static void svm_msg_q_unlock(svm_msg_q_t *mq)
Unlock message queue.
#define VLIB_MAIN_LOOP_EXIT_FUNCTION(x)
#define clib_warning(format, args...)
svm_queue_t * q
queue for exchanging messages
int svm_msg_q_add(svm_msg_q_t *mq, svm_msg_q_msg_t *msg, int nowait)
Producer enqueue one message to queue.
int vnet_disconnect_session(vnet_disconnect_args_t *a)
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 ...
u32 app_wrk_index
Server index.
void transport_connection_update_tx_stats(transport_connection_t *tc, u32 bytes)
Update tx byte stats for transport connection.
void session_transport_close(stream_session_t *s)
Notify transport the session can be disconnected.
static int svm_msg_q_try_lock(svm_msg_q_t *mq)
Try locking message queue.
#define VLIB_BUFFER_DATA_SIZE
application_t * application_get(u32 app_index)
int app_worker_lock_and_send_event(app_worker_t *app, stream_session_t *s, u8 evt_type)
Send event to application.
static stream_session_t * session_get_from_handle_if_valid(session_handle_t handle)
#define SESSION_Q_PROCESS_STOP
u32 ring_index
ring index, could be u8
session_fifo_rx_fn ** session_tx_fns
Per transport rx function that can either dequeue or peek.
#define foreach_session_queue_error
static void session_mq_disconnected_reply_handler(void *data)
u32 * tx_buffers
Vector of tx buffer free lists.
u32 next_buffer
Next buffer for this linked-list of buffers.
int session_tx_fifo_dequeue_internal(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_worker_t *wrk, session_event_t *e, int *n_tx_pkts)
session_cb_vft_t cb_fns
Callbacks: shoulder-taps for the server/client.
u32 * session_type_to_next
Per session type output nodes.
#define vec_append(v1, v2)
Append v2 after v1.
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 void * vlib_add_trace(vlib_main_t *vm, vlib_node_runtime_t *r, vlib_buffer_t *b, u32 n_data_bytes)
static int session_tx_fifo_read_and_snd_i(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_worker_t *wrk, session_event_t *e, int *n_tx_packets, u8 peek_data)
u32 total_length_not_including_first_buffer
Only valid for first buffer in chain.
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
int svm_fifo_dequeue_drop(svm_fifo_t *f, u32 max_bytes)
static u8 session_tx_not_ready(stream_session_t *s, u8 peek_data)
session_manager_worker_t * wrk
Worker contexts.
void svm_msg_q_free_msg(svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
Free message buffer.
enum _transport_proto transport_proto_t
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
svm_msg_q_t * vpp_event_queue
vpp event message queue for worker
#define VLIB_BUFFER_TRACE_TRAJECTORY_INIT(b)
static void session_mq_worker_update_handler(void *data)
static void session_mq_accepted_reply_handler(void *data)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
u32 elsize
size of an element
static void session_update_dispatch_period(session_manager_worker_t *wrk, f64 now, u32 thread_index)
struct clib_bihash_value offset
template key/value backing page structure
static void session_mq_disconnected_handler(void *data)
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
void vlib_worker_thread_barrier_release(vlib_main_t *vm)
static u32 svm_msg_q_size(svm_msg_q_t *mq)
Check length of message queue.
#define vec_foreach(var, vec)
Vector iterator.
static void vlib_set_trace_count(vlib_main_t *vm, vlib_node_runtime_t *rt, u32 count)
#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 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.
int svm_fifo_peek(svm_fifo_t *f, u32 relative_offset, u32 max_bytes, u8 *copy_here)
int svm_fifo_dequeue_nowait(svm_fifo_t *f, u32 max_bytes, u8 *copy_here)
static vlib_buffer_t * vlib_get_buffer(vlib_main_t *vm, u32 buffer_index)
Translate buffer index into buffer pointer.
static void session_mq_reset_reply_handler(void *data)
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.