15 #ifndef __included_session_h__ 16 #define __included_session_h__ 25 #define HALF_OPEN_LOOKUP_INVALID_VALUE ((u64)~0) 26 #define INVALID_INDEX ((u32)~0) 27 #define SESSION_PROXY_LISTENER_INDEX ((u32)~0 - 1) 30 #define MAX_HDRS_LEN 100 42 static inline const char *
48 return "FIFO_EVENT_APP_RX";
50 return "FIFO_EVENT_APP_TX";
52 return "FIFO_EVENT_TIMEOUT";
54 return "FIFO_EVENT_DISCONNECT";
56 return "FIFO_EVENT_BUILTIN_RX";
58 return "FIFO_EVENT_RPC";
60 return "UNKNOWN FIFO EVENT";
64 #define foreach_session_input_error \ 65 _(NO_SESSION, "No session drops") \ 66 _(NO_LISTENER, "No listener for dst port drops") \ 67 _(ENQUEUED, "Packets pushed into rx fifo") \ 68 _(NOT_READY, "Session not ready packets") \ 69 _(FIFO_FULL, "Packets dropped for lack of rx fifo space") \ 70 _(EVENT_FIFO_FULL, "Events not sent for lack of event fifo space") \ 71 _(API_QUEUE_FULL, "Sessions not created for lack of API queue space") \ 72 _(NEW_SEG_NO_SPACE, "Created segment, couldn't allocate a fifo pair") \ 73 _(NO_SPACE, "Couldn't allocate a fifo pair") 77 #define _(sym,str) SESSION_ERROR_##sym, 99 }) session_fifo_event_t;
109 u32 thread_index,
int *n_tx_pkts);
116 struct _session_manager_main
122 u32 *session_peekers;
141 session_fifo_event_t **free_event_vector;
144 session_fifo_event_t **pending_event_vector;
147 session_fifo_event_t **pending_disconnects;
153 u32 unique_segment_name_counter;
161 u32 *session_type_to_next;
167 u32 configured_event_queue_length;
174 u32 configured_v4_session_table_buckets;
175 u32 configured_v4_session_table_memory;
176 u32 configured_v4_halfopen_table_buckets;
177 u32 configured_v4_halfopen_table_memory;
178 u32 configured_v6_session_table_buckets;
179 u32 configured_v6_session_table_memory;
180 u32 configured_v6_halfopen_table_buckets;
181 u32 configured_v6_halfopen_table_memory;
184 u32 local_endpoints_table_memory;
185 u32 local_endpoints_table_buckets;
188 u32 preallocated_sessions;
195 f64 *last_event_poll_by_thread;
217 if (s->thread_index != thread_index || s->session_index != si
238 if (thread_index >=
vec_len (session_manager_main.sessions))
251 return ((
u64) s->thread_index << 32) | (
u64) s->session_index;
257 return handle & 0xFFFFFFFF;
285 return (s->session_type >> 1);
291 return (proto << 1 | is_ip4);
311 smm->session_peekers[thread_index] += 1;
312 if (smm->session_peekers[thread_index] == 1)
323 ASSERT (session_manager_main.session_peekers[thread_index] > 0);
325 smm->session_peekers[thread_index] -= 1;
326 if (smm->session_peekers[thread_index] == 0)
364 return s - session_manager_main.listen_sessions[s->session_type];
366 return s - session_manager_main.sessions[s->thread_index];
380 return s->server_rx_fifo->nitems;
386 return (s - session_manager_main.sessions[s->thread_index]);
402 old_s = session_manager_main.sessions[thread_index] + session_index;
405 new_s->thread_index = current_thread_index;
418 u8 queue_event,
u8 is_in_order);
420 u8 proto,
u8 queue_event);
422 u32 offset,
u32 max_bytes);
427 u32 old_thread_index,
430 u32 rx_pointer,
u32 tx_pointer);
444 fifo_event_type_t evt_type,
461 return session_manager_main.vpp_event_queues[thread_index];
470 return ((
u64) s->session_type << 32) | s->session_index;
480 index = handle & 0xFFFFFFFF;
496 memset (s, 0,
sizeof (*s));
498 s->session_type = type;
500 s->session_index = s - session_manager_main.listen_sessions[type];
515 pool_put (session_manager_main.listen_sessions[s->session_type], s);
540 session_manager_main.session_tx_fns[type] = (is_peek) ?
541 session_tx_fifo_peek_and_snd : session_tx_fifo_dequeue_and_snd;
547 return session_manager_main.is_enabled == 1;
550 #define session_cli_return_if_not_enabled() \ 552 if (!session_manager_main.is_enabled) \ 553 return clib_error_return(0, "session layer is not enabled"); \
static stream_session_t * listen_session_get_from_handle(u64 handle)
struct _transport_connection transport_connection_t
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.
uword unformat_transport_connection(unformat_input_t *input, va_list *args)
uword unformat_stream_session(unformat_input_t *input, va_list *args)
static_always_inline void clib_spinlock_unlock_if_init(clib_spinlock_t *p)
struct _transport_proto_vft transport_proto_vft_t
struct _vlib_node_registration vlib_node_registration_t
static stream_session_t * listen_session_new(session_type_t type)
int stream_session_accept(transport_connection_t *tc, u32 listener_index, u8 notify)
Accept a stream session.
int session_stream_connect_notify(transport_connection_t *tc, u8 is_fail)
static u32 svm_fifo_max_enqueue(svm_fifo_t *f)
static void session_pool_remove_peeker(u32 thread_index)
static u8 session_manager_is_enabled()
int listen_session_get_local_session_endpoint(stream_session_t *listener, session_endpoint_t *sep)
static u32 stream_session_max_rx_enqueue(transport_connection_t *tc)
vlib_node_registration_t session_queue_node
(constructor) VLIB_REGISTER_NODE (session_queue_node)
session_fifo_rx_fn session_tx_fifo_peek_and_snd
struct _svm_fifo svm_fifo_t
typedef CLIB_PACKED(struct{union{svm_fifo_t *fifo;u64 session_handle;rpc_args_t rpc_args;};u8 event_type;u8 postponed;}) session_fifo_event_t
int session_open(u32 app_index, session_endpoint_t *tep, u32 opaque)
Ask transport to open connection to remote transport endpoint.
void stream_session_accept_notify(transport_connection_t *tc)
u32 stream_session_dequeue_drop(transport_connection_t *tc, u32 max_bytes)
struct _stream_session_t stream_session_t
u8 session_node_lookup_fifo_event(svm_fifo_t *f, session_fifo_event_t *e)
static transport_proto_t session_get_transport_proto(stream_session_t *s)
static void session_parse_handle(u64 handle, u32 *index, u32 *thread_index)
static void listen_session_del(stream_session_t *s)
static session_manager_main_t * vnet_get_session_manager_main()
#define pool_elt_at_index(p, i)
Returns pointer to element at given index.
static u32 session_thread_from_handle(u64 handle)
struct _session_endpoint session_endpoint_t
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.
static u32 session_get_index(stream_session_t *s)
void stream_session_init_fifos_pointers(transport_connection_t *tc, u32 rx_pointer, u32 tx_pointer)
Init fifo tail and head pointers.
#define pool_put(P, E)
Free an object E in pool P.
static stream_session_t * session_manager_get_listener(u8 session_type, u32 index)
static stream_session_t * session_get_if_valid(u64 si, u32 thread_index)
struct _session_manager_main session_manager_main_t
static stream_session_t * session_get(u32 si, u32 thread_index)
static void session_pool_add_peeker(u32 thread_index)
Acquires a lock that blocks a session pool from expanding.
#define pool_get_aligned(P, E, A)
Allocate an object E from a pool P (general version).
void session_node_enable_disable(u8 is_en)
static unix_shared_memory_queue_t * session_manager_get_vpp_event_queue(u32 thread_index)
int session_manager_flush_enqueue_events(u8 proto, u32 thread_index)
Flushes queue of sessions that are to be notified of new data enqueued events.
session_fifo_rx_fn session_tx_fifo_dequeue_and_snd
static_always_inline uword vlib_get_thread_index(void)
static session_type_t session_type_from_proto_and_ip(transport_proto_t proto, u8 is_ip4)
static u32 stream_session_get_index(stream_session_t *s)
u8 * format_stream_session(u8 *s, va_list *args)
Format stream session as per the following format.
void stream_session_disconnect(stream_session_t *s)
Disconnect session and propagate to transport.
static stream_session_t * session_get_from_handle_safe(u64 handle)
Get session from handle and 'lock' pool resize if not in same thread.
#define clib_memcpy(a, b, c)
int stream_session_stop_listen(stream_session_t *s)
Ask transport to stop listening on local transport endpoint.
#define pool_is_free_index(P, I)
Use free bitmap to query whether given index is free.
session_manager_main_t session_manager_main
static void session_manager_set_transport_rx_fn(session_type_t type, u8 is_peek)
Set peek or dequeue function for given session type.
static u8 stream_session_is_valid(u32 si, u8 thread_index)
clib_error_t * vnet_session_enable_disable(vlib_main_t *vm, u8 is_en)
void stream_session_delete_notify(transport_connection_t *tc)
Notification from transport that connection is being deleted.
void stream_session_cleanup(stream_session_t *s)
Cleanup transport and session state.
void stream_session_reset_notify(transport_connection_t *tc)
Notify application that connection has been reset.
static const char * fifo_event_type_str(fifo_event_type_t et)
void session_send_session_evt_to_thread(u64 session_handle, fifo_event_type_t evt_type, u32 thread_index)
transport_connection_t * session_get_transport(stream_session_t *s)
static stream_session_t * stream_session_listener_get(u8 sst, u64 si)
u32 stream_session_tx_fifo_max_dequeue(transport_connection_t *tc)
int session_enqueue_stream_connection(transport_connection_t *tc, vlib_buffer_t *b, u32 offset, u8 queue_event, u8 is_in_order)
static u32 session_index_from_handle(u64 handle)
static stream_session_t * session_get_from_handle(u64 handle)
static u64 session_handle(stream_session_t *s)
int session_enqueue_dgram_connection(stream_session_t *s, vlib_buffer_t *b, u8 proto, u8 queue_event)
enum _transport_proto transport_proto_t
#define vec_len(v)
Number of elements in vector (rvalue-only, NULL tolerant)
static stream_session_t * listen_session_get(session_type_t type, u32 index)
stream_session_t * session_alloc(u32 thread_index)
static stream_session_t * session_clone_safe(u32 session_index, u32 thread_index)
#define foreach_session_input_error
static u32 stream_session_rx_fifo_size(transport_connection_t *tc)
int stream_session_listen(stream_session_t *s, session_endpoint_t *tep)
Ask transport to listen on local transport endpoint.
int stream_session_peek_bytes(transport_connection_t *tc, u8 *buffer, u32 offset, u32 max_bytes)
int( session_fifo_rx_fn)(vlib_main_t *vm, vlib_node_runtime_t *node, session_manager_main_t *smm, session_fifo_event_t *e0, stream_session_t *s0, u32 thread_index, int *n_tx_pkts)
#define CLIB_CACHE_LINE_BYTES
transport_connection_t * listen_session_get_transport(stream_session_t *s)
static_always_inline void clib_spinlock_lock_if_init(clib_spinlock_t *p)
static u64 listen_session_get_handle(stream_session_t *s)
void stream_session_disconnect_notify(transport_connection_t *tc)
Notification from transport that connection is being closed.
struct _unix_shared_memory_queue unix_shared_memory_queue_t