static int
tcp_test_lookup (vlib_main_t * vm, unformat_input_t * input)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
tcp_main_t *tm = &tcp_main;
transport_connection_t _tc1, *tc1 = &_tc1, _tc2, *tc2 = &_tc2, *tconn;
tcp_connection_t *tc;
vlib_put_next_frame (vm, node, next_index, n_left_to_next);
}
- errors = session_manager_flush_enqueue_events (TRANSPORT_PROTO_SCTP,
- my_thread_index);
+ errors = session_main_flush_enqueue_events (TRANSPORT_PROTO_SCTP,
+ my_thread_index);
sctp_node_inc_counter (vm, is_ip4, sctp4_established_phase_node.index,
sctp6_established_phase_node.index,
* Make sure we can retransmit something
*/
available_bytes =
- session_tx_fifo_max_dequeue (&sctp_conn->sub_conn[idx].connection);
+ transport_max_tx_dequeue (&sctp_conn->sub_conn[idx].connection);
ASSERT (available_bytes >= offset);
available_bytes -= offset;
if (!available_bytes)
if (PREDICT_TRUE (seg_size <= tm->bytes_per_buffer))
{
n_bytes =
- stream_session_peek_bytes (&sctp_conn->sub_conn[idx].connection, data,
- offset, max_deq_bytes);
+ session_tx_fifo_peek_bytes (&sctp_conn->sub_conn[idx].connection,
+ data, offset, max_deq_bytes);
ASSERT (n_bytes == max_deq_bytes);
b[0]->current_length = n_bytes;
sctp_push_hdr_i (sctp_conn, *b, sctp_conn->state);
if (!ecm->vpp_event_queue[thread_index])
ecm->vpp_event_queue[thread_index] =
- session_manager_get_vpp_event_queue (thread_index);
+ session_main_get_vpp_event_queue (thread_index);
/*
* Setup session
vnet_disconnect_args_t _a = { 0 }, *a = &_a;
if (s->session_state == SESSION_STATE_READY)
- clib_warning ("Reset active connection %U", format_stream_session, s, 2);
+ clib_warning ("Reset active connection %U", format_session, s, 2);
a->handle = session_handle (s);
a->app_index = ecm->app_index;
echo_server_main_t *esm = &echo_server_main;
esm->vpp_queue[s->thread_index] =
- session_manager_get_vpp_event_queue (s->thread_index);
+ session_main_get_vpp_event_queue (s->thread_index);
s->session_state = SESSION_STATE_READY;
esm->byte_index = 0;
ASSERT (vec_len (esm->rx_retries) > s->thread_index);
{
echo_server_main_t *esm = &echo_server_main;
vnet_disconnect_args_t _a = { 0 }, *a = &_a;
- clib_warning ("Reset session %U", format_stream_session, s, 2);
+ clib_warning ("Reset session %U", format_session, s, 2);
a->handle = session_handle (s);
a->app_index = esm->app_index;
vnet_disconnect_session (a);
if (!esm->vpp_queue[s->thread_index])
{
svm_msg_q_t *mq;
- mq = session_manager_get_vpp_event_queue (s->thread_index);
+ mq = session_main_get_vpp_event_queue (s->thread_index);
esm->vpp_queue[s->thread_index] = mq;
}
max_enqueue -= sizeof (session_dgram_hdr_t);
vec_validate (esm->rx_retries[s->thread_index], s->session_index);
if (esm->rx_retries[thread_index][s->session_index] == 500000)
{
- clib_warning ("session stuck: %U", format_stream_session, s, 2);
+ clib_warning ("session stuck: %U", format_session, s, 2);
}
if (esm->rx_retries[thread_index][s->session_index] < 500001)
esm->rx_retries[thread_index][s->session_index]++;
vec_validate (esm->rx_retries, num_threads - 1);
for (i = 0; i < vec_len (esm->rx_retries); i++)
vec_validate (esm->rx_retries[i],
- pool_elts (session_manager_main.wrk[i].sessions));
+ pool_elts (session_main.wrk[i].sessions));
esm->rcv_buffer_size = clib_max (esm->rcv_buffer_size, esm->fifo_size);
for (i = 0; i < num_threads; i++)
vec_validate (esm->rx_buf[i], esm->rcv_buffer_size);
http_session_t *hs;
hsm->vpp_queue[s->thread_index] =
- session_manager_get_vpp_event_queue (s->thread_index);
+ session_main_get_vpp_event_queue (s->thread_index);
if (!hsm->is_static)
http_server_sessions_writer_lock ();
static void
proxy_reset_callback (session_t * s)
{
- clib_warning ("Reset session %U", format_stream_session, s, 2);
+ clib_warning ("Reset session %U", format_session, s, 2);
delete_proxy_session (s, 0 /* is_active_open */ );
}
for (i = 0; i < num_threads; i++)
{
- pm->active_open_event_queue[i] =
- session_manager_get_vpp_event_queue (i);
+ pm->active_open_event_queue[i] = session_main_get_vpp_event_queue (i);
ASSERT (pm->active_open_event_queue[i]);
- pm->server_event_queue[i] = session_manager_get_vpp_event_queue (i);
+ pm->server_event_queue[i] = session_main_get_vpp_event_queue (i);
}
return 0;
u8 is_valid;
if (st == SSVM_SEGMENT_MEMFD)
{
- is_valid = (session_manager_get_evt_q_segment () != 0);
+ is_valid = (session_main_get_evt_q_segment () != 0);
if (!is_valid)
clib_warning ("memfd seg: vpp's event qs IN binary api svm region");
return is_valid;
}
else if (st == SSVM_SEGMENT_SHM)
{
- is_valid = (session_manager_get_evt_q_segment () == 0);
+ is_valid = (session_main_get_evt_q_segment () == 0);
if (!is_valid)
clib_warning ("shm seg: vpp's event qs NOT IN binary api svm region");
return is_valid;
u32 app_worker_n_listeners (app_worker_t * app);
session_t *app_worker_first_listener (app_worker_t * app,
u8 fib_proto, u8 transport_proto);
-u8 app_worker_application_is_builtin (app_worker_t * app_wrk);
int app_worker_send_event (app_worker_t * app, session_t * s, u8 evt);
int app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
u8 evt_type);
segment_handle);
}
-u8
+static inline u8
app_worker_application_is_builtin (app_worker_t * app_wrk)
{
return app_wrk->app_is_builtin;
app_name = application_name_from_index (app_wrk->app_index);
listener = listen_session_get_from_handle (handle);
- str = format (0, "%U", format_stream_session, listener, verbose);
+ str = format (0, "%U", format_session, listener, verbose);
if (verbose)
{
thread_index = fifo->master_thread_index;
session = session_get (session_index, thread_index);
- str = format (0, "%U", format_stream_session, session, verbose);
+ str = format (0, "%U", format_session, session, verbose);
if (verbose)
s = format (s, "%-40s%-20s%-15u%-10u", str, app_name,
#include <vnet/dpo/load_balance.h>
#include <vnet/fib/ip4_fib.h>
-session_manager_main_t session_manager_main;
+session_main_t session_main;
static inline int
session_send_evt_to_thread (void *data, void *args, u32 thread_index,
svm_msg_q_t *mq;
u32 tries = 0, max_tries;
- mq = session_manager_get_vpp_event_queue (thread_index);
+ mq = session_main_get_vpp_event_queue (thread_index);
while (svm_msg_q_try_lock (mq))
{
max_tries = vlib_get_current_process (vlib_get_main ())? 1e6 : 3;
session_program_transport_close (session_t * s)
{
u32 thread_index = vlib_get_thread_index ();
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
session_event_t *evt;
if (!session_has_transport (s))
* held, just append a new event to pending disconnects vector. */
if (vlib_thread_is_main_w_barrier () || thread_index == s->thread_index)
{
- wrk = session_manager_get_worker (s->thread_index);
+ wrk = session_main_get_worker (s->thread_index);
vec_add2 (wrk->pending_disconnects, evt, 1);
clib_memset (evt, 0, sizeof (*evt));
evt->session_handle = session_handle (s);
session_t *
session_alloc (u32 thread_index)
{
- session_manager_worker_t *wrk = &session_manager_main.wrk[thread_index];
+ session_worker_t *wrk = &session_main.wrk[thread_index];
session_t *s;
u8 will_expand = 0;
pool_get_aligned_will_expand (wrk->sessions, will_expand,
void
session_free (session_t * s)
{
- pool_put (session_manager_main.wrk[s->thread_index].sessions, s);
+ pool_put (session_main.wrk[s->thread_index].sessions, s);
if (CLIB_DEBUG)
clib_memset (s, 0xFA, sizeof (*s));
}
{
/* Queue RX event on this fifo. Eventually these will need to be flushed
* by calling stream_server_flush_enqueue_events () */
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
- wrk = session_manager_get_worker (s->thread_index);
+ wrk = session_main_get_worker (s->thread_index);
if (s->enqueue_epoch != wrk->current_enqueue_epoch[tc->proto])
{
s->enqueue_epoch = wrk->current_enqueue_epoch[tc->proto];
{
/* Queue RX event on this fifo. Eventually these will need to be flushed
* by calling stream_server_flush_enqueue_events () */
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
- wrk = session_manager_get_worker (s->thread_index);
+ wrk = session_main_get_worker (s->thread_index);
if (s->enqueue_epoch != wrk->current_enqueue_epoch[proto])
{
s->enqueue_epoch = wrk->current_enqueue_epoch[proto];
return enqueued;
}
-/** Check if we have space in rx fifo to push more bytes */
-u8
-stream_session_no_space (transport_connection_t * tc, u32 thread_index,
- u16 data_len)
-{
- session_t *s = session_get (tc->s_index, thread_index);
-
- if (PREDICT_FALSE (s->session_state != SESSION_STATE_READY))
- return 1;
-
- if (data_len > svm_fifo_max_enqueue (s->rx_fifo))
- return 1;
-
- return 0;
-}
-
-u32
-session_tx_fifo_max_dequeue (transport_connection_t * tc)
-{
- session_t *s = session_get (tc->s_index, tc->thread_index);
- if (!s->tx_fifo)
- return 0;
- return svm_fifo_max_dequeue (s->tx_fifo);
-}
-
int
-stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
- u32 offset, u32 max_bytes)
+session_tx_fifo_peek_bytes (transport_connection_t * tc, u8 * buffer,
+ u32 offset, u32 max_bytes)
{
session_t *s = session_get (tc->s_index, tc->thread_index);
return svm_fifo_peek (s->tx_fifo, offset, max_bytes, buffer);
}
u32
-session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
+session_tx_fifo_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
{
session_t *s = session_get (tc->s_index, tc->thread_index);
return svm_fifo_dequeue_drop (s->tx_fifo, max_bytes);
* failures due to API queue being full.
*/
int
-session_manager_flush_enqueue_events (u8 transport_proto, u32 thread_index)
+session_main_flush_enqueue_events (u8 transport_proto, u32 thread_index)
{
- session_manager_worker_t *wrk = session_manager_get_worker (thread_index);
+ session_worker_t *wrk = session_main_get_worker (thread_index);
session_t *s;
int i, errors = 0;
u32 *indices;
}
int
-session_manager_flush_all_enqueue_events (u8 transport_proto)
+session_main_flush_all_enqueue_events (u8 transport_proto)
{
vlib_thread_main_t *vtm = vlib_get_thread_main ();
int i, errors = 0;
for (i = 0; i < 1 + vtm->n_threads; i++)
- errors += session_manager_flush_enqueue_events (transport_proto, i);
+ errors += session_main_flush_enqueue_events (transport_proto, i);
return errors;
}
-/**
- * Init fifo tail and head pointers
- *
- * Useful if transport uses absolute offsets for tracking ooo segments.
- */
-void
-stream_session_init_fifos_pointers (transport_connection_t * tc,
- u32 rx_pointer, u32 tx_pointer)
-{
- session_t *s;
- s = session_get (tc->s_index, tc->thread_index);
- svm_fifo_init_pointers (s->rx_fifo, rx_pointer);
- svm_fifo_init_pointers (s->tx_fifo, tx_pointer);
-}
-
int
session_stream_connect_notify (transport_connection_t * tc, u8 is_fail)
{
* vpp uses api svm region for event queues.
*/
void
-session_vpp_event_queues_allocate (session_manager_main_t * smm)
+session_vpp_event_queues_allocate (session_main_t * smm)
{
u32 evt_q_length = 2048, evt_size = sizeof (session_event_t);
ssvm_private_t *eqs = &smm->evt_qs_segment;
}
ssvm_private_t *
-session_manager_get_evt_q_segment (void)
+session_main_get_evt_q_segment (void)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
if (smm->evt_qs_use_memfd_seg)
return &smm->evt_qs_segment;
return 0;
}
+u64
+session_segment_handle (session_t * s)
+{
+ svm_fifo_t *f;
+
+ if (s->session_state == SESSION_STATE_LISTENING)
+ return SESSION_INVALID_HANDLE;
+
+ f = s->rx_fifo;
+ return segment_manager_make_segment_handle (f->segment_manager,
+ f->segment_index);
+}
+
/* *INDENT-OFF* */
static session_fifo_rx_fn *session_tx_fns[TRANSPORT_TX_N_FNS] = {
session_tx_fifo_peek_and_snd,
const transport_proto_vft_t * vft, u8 is_ip4,
u32 output_node)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
session_type_t session_type;
u32 next_index = ~0;
s->connection_index);
}
-int
-listen_session_get_local_session_endpoint (session_t * listener,
- session_endpoint_t * sep)
-{
- transport_connection_t *tc;
- tc = listen_session_get_transport (listener);
- if (!tc)
- {
- clib_warning ("no transport");
- return -1;
- }
-
- /* N.B. The ip should not be copied because this is the local endpoint */
- sep->port = tc->lcl_port;
- sep->transport_proto = tc->proto;
- sep->is_ip4 = tc->is_ip4;
- return 0;
-}
-
void
session_flush_frames_main_thread (vlib_main_t * vm)
{
session_manager_main_enable (vlib_main_t * vm)
{
segment_manager_main_init_args_t _sm_args = { 0 }, *sm_args = &_sm_args;
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
vlib_thread_main_t *vtm = vlib_get_thread_main ();
u32 num_threads, preallocated_sessions_per_worker;
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
int i, j;
num_threads = 1 /* main thread */ + vtm->n_threads;
clib_error_t *error = 0;
if (is_en)
{
- if (session_manager_main.is_enabled)
+ if (session_main.is_enabled)
return 0;
session_node_enable_disable (is_en);
}
else
{
- session_manager_main.is_enabled = 0;
+ session_main.is_enabled = 0;
session_node_enable_disable (is_en);
}
clib_error_t *
session_manager_main_init (vlib_main_t * vm)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
smm->session_baseva = HIGH_SEGMENT_BASEVA;
#if (HIGH_SEGMENT_BASEVA > (4ULL << 30))
smm->session_va_space_size = 128ULL << 30;
static clib_error_t *
session_config_fn (vlib_main_t * vm, unformat_input_t * input)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
u32 nitems;
uword tmp;
#include <vnet/session/session_types.h>
#include <vnet/session/session_lookup.h>
#include <vnet/session/session_debug.h>
-#include <vnet/session/segment_manager.h>
#include <svm/message_queue.h>
+#include <svm/ssvm.h>
#define SESSION_PROXY_LISTENER_INDEX ((u8)~0 - 1)
session_dgram_hdr_t hdr;
} session_tx_context_t;
-typedef struct session_manager_worker_
+typedef struct session_worker_
{
CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
u32 last_tx_packets;
-} session_manager_worker_t;
+} session_worker_t;
typedef int (session_fifo_rx_fn) (vlib_main_t * vm,
vlib_node_runtime_t * node,
- session_manager_worker_t * wrk,
+ session_worker_t * wrk,
session_event_t * e, int *n_tx_pkts);
extern session_fifo_rx_fn session_tx_fifo_peek_and_snd;
u8 session_node_lookup_fifo_event (svm_fifo_t * f, session_event_t * e);
-typedef struct session_manager_main_
+typedef struct session_main_
{
/** Worker contexts */
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
/** Event queues memfd segment initialized only if so configured */
ssvm_private_t evt_qs_segment;
f64 *last_event_poll_by_thread;
#endif
-} session_manager_main_t;
+} session_main_t;
-extern session_manager_main_t session_manager_main;
+extern session_main_t session_main;
extern vlib_node_registration_t session_queue_node;
extern vlib_node_registration_t session_queue_process_node;
#define SESSION_Q_PROCESS_STOP 2
always_inline u8
-stream_session_is_valid (u32 si, u8 thread_index)
+session_is_valid (u32 si, u8 thread_index)
{
session_t *s;
- s = pool_elt_at_index (session_manager_main.wrk[thread_index].sessions, si);
- if (s->thread_index != thread_index || s->session_index != si
- /* || s->server_rx_fifo->master_session_index != si
- || s->server_tx_fifo->master_session_index != si
- || s->server_rx_fifo->master_thread_index != thread_index
- || s->server_tx_fifo->master_thread_index != thread_index */ )
+ s = pool_elt_at_index (session_main.wrk[thread_index].sessions, si);
+ if (s->thread_index != thread_index || s->session_index != si)
return 0;
return 1;
}
always_inline session_t *
session_get (u32 si, u32 thread_index)
{
- ASSERT (stream_session_is_valid (si, thread_index));
- return pool_elt_at_index (session_manager_main.wrk[thread_index].sessions,
- si);
+ ASSERT (session_is_valid (si, thread_index));
+ return pool_elt_at_index (session_main.wrk[thread_index].sessions, si);
}
always_inline session_t *
session_get_if_valid (u64 si, u32 thread_index)
{
- if (thread_index >= vec_len (session_manager_main.wrk))
+ if (thread_index >= vec_len (session_main.wrk))
return 0;
- if (pool_is_free_index (session_manager_main.wrk[thread_index].sessions,
- si))
+ if (pool_is_free_index (session_main.wrk[thread_index].sessions, si))
return 0;
- ASSERT (stream_session_is_valid (si, thread_index));
- return pool_elt_at_index (session_manager_main.wrk[thread_index].sessions,
- si);
+ ASSERT (session_is_valid (si, thread_index));
+ return pool_elt_at_index (session_main.wrk[thread_index].sessions, si);
}
always_inline session_t *
session_get_from_handle (session_handle_t handle)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
u32 session_index, thread_index;
session_parse_handle (handle, &session_index, &thread_index);
return pool_elt_at_index (smm->wrk[thread_index].sessions, session_index);
return session_get_if_valid (session_index, thread_index);
}
-always_inline u64
-session_segment_handle (session_t * s)
-{
- svm_fifo_t *f;
-
- if (s->session_state == SESSION_STATE_LISTENING)
- return SESSION_INVALID_HANDLE;
-
- f = s->rx_fifo;
- return segment_manager_make_segment_handle (f->segment_manager,
- f->segment_index);
-}
+u64 session_segment_handle (session_t * s);
/**
* Acquires a lock that blocks a session pool from expanding.
always_inline void
session_pool_add_peeker (u32 thread_index)
{
- session_manager_worker_t *wrk = &session_manager_main.wrk[thread_index];
+ session_worker_t *wrk = &session_main.wrk[thread_index];
if (thread_index == vlib_get_thread_index ())
return;
clib_rwlock_reader_lock (&wrk->peekers_rw_locks);
always_inline void
session_pool_remove_peeker (u32 thread_index)
{
- session_manager_worker_t *wrk = &session_manager_main.wrk[thread_index];
+ session_worker_t *wrk = &session_main.wrk[thread_index];
if (thread_index == vlib_get_thread_index ())
return;
clib_rwlock_reader_unlock (&wrk->peekers_rw_locks);
session_get_from_handle_safe (u64 handle)
{
u32 thread_index = session_thread_from_handle (handle);
- session_manager_worker_t *wrk = &session_manager_main.wrk[thread_index];
+ session_worker_t *wrk = &session_main.wrk[thread_index];
if (thread_index == vlib_get_thread_index ())
{
always_inline u32
session_get_index (session_t * s)
{
- return (s - session_manager_main.wrk[s->thread_index].sessions);
+ return (s - session_main.wrk[s->thread_index].sessions);
}
always_inline session_t *
*/
session_pool_add_peeker (thread_index);
new_s = session_alloc (current_thread_index);
- old_s = session_manager_main.wrk[thread_index].sessions + session_index;
+ old_s = session_main.wrk[thread_index].sessions + session_index;
clib_memcpy_fast (new_s, old_s, sizeof (*new_s));
session_pool_remove_peeker (thread_index);
new_s->thread_index = current_thread_index;
transport_connection_t *session_get_transport (session_t * s);
-u8 *format_stream_session (u8 * s, va_list * args);
-uword unformat_stream_session (unformat_input_t * input, va_list * args);
+u8 *format_session (u8 * s, va_list * args);
+uword unformat_session (unformat_input_t * input, va_list * args);
uword unformat_transport_connection (unformat_input_t * input,
va_list * args);
session_dgram_hdr_t * hdr,
vlib_buffer_t * b, u8 proto,
u8 queue_event);
-int stream_session_peek_bytes (transport_connection_t * tc, u8 * buffer,
- u32 offset, u32 max_bytes);
-u32 session_dequeue_drop (transport_connection_t * tc, u32 max_bytes);
-
int session_stream_connect_notify (transport_connection_t * tc, u8 is_fail);
int session_dgram_connect_notify (transport_connection_t * tc,
u32 old_thread_index,
session_t ** new_session);
-void stream_session_init_fifos_pointers (transport_connection_t * tc,
- u32 rx_pointer, u32 tx_pointer);
-
int session_stream_accept_notify (transport_connection_t * tc);
void session_transport_closing_notify (transport_connection_t * tc);
void session_transport_delete_notify (transport_connection_t * tc);
void session_transport_reset_notify (transport_connection_t * tc);
int session_stream_accept (transport_connection_t * tc, u32 listener_index,
u8 notify);
-u32 session_tx_fifo_max_dequeue (transport_connection_t * tc);
void session_register_transport (transport_proto_t transport_proto,
const transport_proto_vft_t * vft, u8 is_ip4,
u32 output_node);
+int session_tx_fifo_peek_bytes (transport_connection_t * tc, u8 * buffer,
+ u32 offset, u32 max_bytes);
+u32 session_tx_fifo_dequeue_drop (transport_connection_t * tc, u32 max_bytes);
always_inline u32
transport_max_rx_enqueue (transport_connection_t * tc)
always_inline f64
transport_dispatch_period (u32 thread_index)
{
- return session_manager_main.wrk[thread_index].dispatch_period;
+ return session_main.wrk[thread_index].dispatch_period;
}
always_inline f64
transport_time_now (u32 thread_index)
{
- return session_manager_main.wrk[thread_index].last_vlib_time;
+ return session_main.wrk[thread_index].last_vlib_time;
}
always_inline void
transport_connection_t *listen_session_get_transport (session_t * s);
-int
-listen_session_get_local_session_endpoint (session_t * listener,
- session_endpoint_t * sep);
-
/*
- * Session manager functions
+ * Session layer functions
*/
-always_inline session_manager_main_t *
-vnet_get_session_manager_main ()
+always_inline session_main_t *
+vnet_get_session_main ()
{
- return &session_manager_main;
+ return &session_main;
}
-always_inline session_manager_worker_t *
-session_manager_get_worker (u32 thread_index)
+always_inline session_worker_t *
+session_main_get_worker (u32 thread_index)
{
- return &session_manager_main.wrk[thread_index];
+ return &session_main.wrk[thread_index];
}
always_inline svm_msg_q_t *
-session_manager_get_vpp_event_queue (u32 thread_index)
+session_main_get_vpp_event_queue (u32 thread_index)
{
- return session_manager_main.wrk[thread_index].vpp_event_queue;
+ return session_main.wrk[thread_index].vpp_event_queue;
}
always_inline u8
-session_manager_is_enabled ()
+session_main_is_enabled ()
{
- return session_manager_main.is_enabled == 1;
+ return session_main.is_enabled == 1;
}
#define session_cli_return_if_not_enabled() \
do { \
- if (!session_manager_main.is_enabled) \
+ if (!session_main.is_enabled) \
return clib_error_return(0, "session layer is not enabled"); \
} while (0)
-int session_manager_flush_enqueue_events (u8 proto, u32 thread_index);
-int session_manager_flush_all_enqueue_events (u8 transport_proto);
+int session_main_flush_enqueue_events (u8 proto, u32 thread_index);
+int session_main_flush_all_enqueue_events (u8 transport_proto);
void session_flush_frames_main_thread (vlib_main_t * vm);
-ssvm_private_t *session_manager_get_evt_q_segment (void);
+ssvm_private_t *session_main_get_evt_q_segment (void);
void session_node_enable_disable (u8 is_en);
clib_error_t *vnet_session_enable_disable (vlib_main_t * vm, u8 is_en);
if (listener)
mp->listener_handle = listen_session_get_handle (listener);
}
- vpp_queue = session_manager_get_vpp_event_queue (s->thread_index);
+ vpp_queue = session_main_get_vpp_event_queue (s->thread_index);
mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
mp->handle = session_handle (s);
tc = transport_get_connection (session_get_transport_proto (s),
goto done;
}
- vpp_queue = session_manager_get_vpp_event_queue (s->thread_index);
+ vpp_queue = session_main_get_vpp_event_queue (s->thread_index);
mp->handle = session_handle (s);
mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
clib_memcpy_fast (mp->lcl_ip, &tc->lcl_ip, sizeof (tc->lcl_ip));
if (listener)
mp->listener_handle = listen_session_get_handle (listener);
}
- vpp_queue = session_manager_get_vpp_event_queue (s->thread_index);
+ vpp_queue = session_main_get_vpp_event_queue (s->thread_index);
mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
mp->handle = session_handle (s);
tc = transport_get_connection (session_get_transport_proto (s),
mp->is_ip4 = session_type_is_ip4 (listener->session_type);
mp->handle = session_handle (s);
mp->port = ct->c_rmt_port;
- vpp_queue = session_manager_get_vpp_event_queue (main_thread);
+ vpp_queue = session_main_get_vpp_event_queue (main_thread);
mp->vpp_event_queue_address = pointer_to_uword (vpp_queue);
mp->client_event_queue_address = ct->client_evt_q;
mp->server_event_queue_address = ct->server_evt_q;
goto done;
}
- vpp_mq = session_manager_get_vpp_event_queue (s->thread_index);
+ vpp_mq = session_main_get_vpp_event_queue (s->thread_index);
mp->handle = session_handle (s);
mp->vpp_event_queue_address = pointer_to_uword (vpp_mq);
clib_memcpy_fast (mp->lcl_ip, &tc->lcl_ip, sizeof (tc->lcl_ip));
mp->handle = session_handle (s);
mp->lcl_port = cct->c_lcl_port;
- vpp_mq = session_manager_get_vpp_event_queue (main_thread);
+ vpp_mq = session_main_get_vpp_event_queue (main_thread);
mp->vpp_event_queue_address = pointer_to_uword (vpp_mq);
mp->client_event_queue_address = cct->client_evt_q;
mp->server_event_queue_address = cct->server_evt_q;
mp->lcl_is_ip4 = tc->is_ip4;
clib_memcpy_fast (mp->lcl_ip, &tc->lcl_ip, sizeof (tc->lcl_ip));
- vpp_evt_q = session_manager_get_vpp_event_queue (0);
+ vpp_evt_q = session_main_get_vpp_event_queue (0);
mp->vpp_evt_q = pointer_to_uword (vpp_evt_q);
if (session_transport_service_type (ls) == TRANSPORT_SERVICE_CL)
if (!reg)
return;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
vec_free (a->namespace_id);
/* Send event queues segment */
- if ((evt_q_segment = session_manager_get_evt_q_segment ()))
+ if ((evt_q_segment = session_main_get_evt_q_segment ()))
{
fd_flags |= SESSION_FD_F_VPP_MQ_SEGMENT;
fds[n_fds] = evt_q_segment->fd;
vnet_app_detach_args_t _a, *a = &_a;
application_t *app;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
app_worker_t *app_wrk;
int rv;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
{
rmp->rx_fifo = pointer_to_uword (s->rx_fifo);
rmp->tx_fifo = pointer_to_uword (s->tx_fifo);
- vpp_evt_q = session_manager_get_vpp_event_queue (0);
+ vpp_evt_q = session_main_get_vpp_event_queue (0);
rmp->vpp_evt_q = pointer_to_uword (vpp_evt_q);
}
}
vnet_unlisten_args_t _a, *a = &_a;
int rv;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app;
int rv = 0;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app;
int rv = 0;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
session_t *s;
int rv = 0;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
{
rmp->rx_fifo = pointer_to_uword (s->rx_fifo);
rmp->tx_fifo = pointer_to_uword (s->tx_fifo);
- vpp_evt_q = session_manager_get_vpp_event_queue (0);
+ vpp_evt_q = session_main_get_vpp_event_queue (0);
rmp->vpp_evt_q = pointer_to_uword (vpp_evt_q);
}
}
application_t *app = 0;
int rv = 0;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app = 0;
int rv = 0;
- if (session_manager_is_enabled () == 0)
+ if (session_main_is_enabled () == 0)
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app;
u8 fd_flags = 0;
- if (!session_manager_is_enabled ())
+ if (!session_main_is_enabled ())
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
u32 appns_index = 0;
u8 *ns_id = 0;
int rv = 0;
- if (!session_manager_is_enabled ())
+ if (!session_main_is_enabled ())
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app;
u32 cert_len;
int rv = 0;
- if (!session_manager_is_enabled ())
+ if (!session_main_is_enabled ())
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
application_t *app;
u32 key_len;
int rv = 0;
- if (!session_manager_is_enabled ())
+ if (!session_main_is_enabled ())
{
rv = VNET_API_ERROR_FEATURE_DISABLED;
goto done;
* "Connection"
*/
u8 *
-format_stream_session (u8 * s, va_list * args)
+format_session (u8 * s, va_list * args)
{
session_t *ss = va_arg (*args, session_t *);
int verbose = va_arg (*args, int);
}
uword
-unformat_stream_session (unformat_input_t * input, va_list * args)
+unformat_session (unformat_input_t * input, va_list * args)
{
session_t **result = va_arg (*args, session_t **);
u32 lcl_port = 0, rmt_port = 0, fib_index = 0;
vlib_cli_command_t * cmd)
{
u8 one_session = 0, do_listeners = 0, sst, do_elog = 0;
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
u32 transport_proto = ~0, track_index;
session_t *pool, *s;
transport_connection_t *tc;
else if (unformat (input, "listeners %U", unformat_transport_proto,
&transport_proto))
do_listeners = 1;
- else if (unformat (input, "%U", unformat_stream_session, &s))
+ else if (unformat (input, "%U", unformat_session, &s))
{
one_session = 1;
}
if (one_session)
{
- u8 *str = format (0, "%U", format_stream_session, s, 3);
+ u8 *str = format (0, "%U", format_session, s, 3);
if (do_elog && s->session_state != SESSION_STATE_LISTENING)
{
elog_main_t *em = &vm->elog_main;
continue;
app_wrk = app_worker_get (s->app_wrk_index);
app_name = application_name_from_index (app_wrk->app_index);
- vlib_cli_output (vm, "%U%-25v%", format_stream_session, s, 0,
+ vlib_cli_output (vm, "%U%-25v%", format_session, s, 0,
app_name);
}));
/* *INDENT-ON* */
n_closed += 1;
continue;
}
- vlib_cli_output (vm, "%U", format_stream_session, s, verbose);
+ vlib_cli_output (vm, "%U", format_session, s, verbose);
}));
/* *INDENT-ON* */
clear_session_command_fn (vlib_main_t * vm, unformat_input_t * input,
vlib_cli_command_t * cmd)
{
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
u32 thread_index = 0, clear_all = 0;
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
u32 session_index = ~0;
session_t *session;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (input, "%U", unformat_stream_session, &s))
+ if (unformat (input, "%U", unformat_session, &s))
;
else if (unformat (input, "rx"))
is_rx = 1;
while (unformat_check_input (input) != UNFORMAT_END_OF_INPUT)
{
- if (unformat (input, "%U", unformat_stream_session, &s))
+ if (unformat (input, "%U", unformat_session, &s))
;
else if (unformat (input, "rx"))
is_rx = 1;
{
vlib_buffer_t *chain_b, *prev_b;
u32 chain_bi0, to_deq, left_from_seg;
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
u16 len_to_deq, n_bytes_read;
u8 *data, j;
- wrk = session_manager_get_worker (ctx->s->thread_index);
+ wrk = session_main_get_worker (ctx->s->thread_index);
b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
b->total_length_not_including_first_buffer = 0;
always_inline int
session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_worker_t * wrk,
+ session_worker_t * wrk,
session_event_t * e, int *n_tx_packets,
u8 peek_data)
{
u32 next_index, next0, next1, *to_next, n_left_to_next, n_left, pbi;
u32 n_trace = vlib_get_trace_count (vm, node), n_bufs_needed = 0;
- session_manager_main_t *smm = &session_manager_main;
+ session_main_t *smm = &session_main;
session_tx_context_t *ctx = &wrk->ctx;
transport_proto_t tp;
vlib_buffer_t *pb;
int
session_tx_fifo_peek_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_worker_t * wrk,
+ session_worker_t * wrk,
session_event_t * e, int *n_tx_pkts)
{
return session_tx_fifo_read_and_snd_i (vm, node, wrk, e, n_tx_pkts, 1);
int
session_tx_fifo_dequeue_and_snd (vlib_main_t * vm, vlib_node_runtime_t * node,
- session_manager_worker_t * wrk,
+ session_worker_t * wrk,
session_event_t * e, int *n_tx_pkts)
{
return session_tx_fifo_read_and_snd_i (vm, node, wrk, e, n_tx_pkts, 0);
int
session_tx_fifo_dequeue_internal (vlib_main_t * vm,
vlib_node_runtime_t * node,
- session_manager_worker_t * wrk,
+ session_worker_t * wrk,
session_event_t * e, int *n_tx_pkts)
{
session_t *s = wrk->ctx.s;
}
static void
-session_update_dispatch_period (session_manager_worker_t * wrk, f64 now,
+session_update_dispatch_period (session_worker_t * wrk, f64 now,
u32 thread_index)
{
if (wrk->last_tx_packets)
session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
vlib_frame_t * frame)
{
- session_manager_main_t *smm = vnet_get_session_manager_main ();
+ session_main_t *smm = vnet_get_session_main ();
u32 thread_index = vm->thread_index, n_to_dequeue, n_events;
- session_manager_worker_t *wrk = &smm->wrk[thread_index];
+ session_worker_t *wrk = &smm->wrk[thread_index];
session_event_t *e, *fifo_events;
svm_msg_q_msg_t _msg, *msg = &_msg;
f64 now = vlib_time_now (vm);
void
dump_thread_0_event_queue (void)
{
- session_manager_main_t *smm = vnet_get_session_manager_main ();
+ session_main_t *smm = vnet_get_session_main ();
vlib_main_t *vm = &vlib_global_main;
u32 my_thread_index = vm->thread_index;
session_event_t _e, *e = &_e;
session_node_lookup_fifo_event (svm_fifo_t * f, session_event_t * e)
{
session_event_t *pending_event_vector, *evt;
- session_manager_worker_t *wrk;
+ session_worker_t *wrk;
int i, index, found = 0;
svm_msg_q_msg_t *msg;
svm_msg_q_ring_t *ring;
ASSERT (e);
thread_index = f->master_thread_index;
- wrk = session_manager_get_worker (thread_index);
+ wrk = session_main_get_worker (thread_index);
/*
* Search evt queue
#undef _
#define _(af,table,parm,value) \
- if (session_manager_main.configured_##af##_##table##_table_##parm) \
+ if (session_main.configured_##af##_##table##_table_##parm) \
configured_##af##_##table##_table_##parm = \
- session_manager_main.configured_##af##_##table##_table_##parm;
+ session_main.configured_##af##_##table##_table_##parm;
foreach_hash_table_parameter;
#undef _
transport_init (void)
{
vlib_thread_main_t *vtm = vlib_get_thread_main ();
- session_manager_main_t *smm = vnet_get_session_manager_main ();
+ session_main_t *smm = vnet_get_session_main ();
u32 num_threads;
if (smm->local_endpoints_table_buckets == 0)
tcp_timer_update (tc, TCP_TIMER_WAITCLOSE, TCP_FINWAIT1_TIME);
break;
case TCP_STATE_ESTABLISHED:
- if (!session_tx_fifo_max_dequeue (&tc->connection))
+ if (!transport_max_tx_dequeue (&tc->connection))
tcp_send_fin (tc);
else
tc->flags |= TCP_CONN_FINPNDG;
tcp_timer_set (tc, TCP_TIMER_WAITCLOSE, TCP_FINWAIT1_TIME);
break;
case TCP_STATE_CLOSE_WAIT:
- if (!session_tx_fifo_max_dequeue (&tc->connection))
+ if (!transport_max_tx_dequeue (&tc->connection))
{
tcp_send_fin (tc);
tcp_connection_timers_reset (tc);
#define tcp_validate_txf_size(_tc, _a) \
ASSERT(_tc->state != TCP_STATE_ESTABLISHED \
- || session_tx_fifo_max_dequeue (&_tc->connection) >= _a)
+ || transport_max_tx_dequeue (&_tc->connection) >= _a)
void tcp_rcv_sacks (tcp_connection_t * tc, u32 ack);
u8 *tcp_scoreboard_replay (u8 * s, tcp_connection_t * tc, u8 verbose);
continue;
/* Dequeue the newly ACKed bytes */
- session_dequeue_drop (&tc->connection, tc->burst_acked);
+ session_tx_fifo_dequeue_drop (&tc->connection, tc->burst_acked);
tc->burst_acked = 0;
tcp_validate_txf_size (tc, tc->snd_una_max - tc->snd_una);
tcp_inc_err_counter (err_counters, error0, 1);
}
- errors = session_manager_flush_enqueue_events (TRANSPORT_PROTO_TCP,
- thread_index);
+ errors = session_main_flush_enqueue_events (TRANSPORT_PROTO_TCP,
+ thread_index);
err_counters[TCP_ERROR_MSG_QUEUE_FULL] = errors;
tcp_store_err_counters (established, err_counters);
tcp_handle_postponed_dequeues (wrk);
}
}
- errors = session_manager_flush_enqueue_events (TRANSPORT_PROTO_TCP,
- my_thread_index);
+ errors = session_main_flush_enqueue_events (TRANSPORT_PROTO_TCP,
+ my_thread_index);
tcp_inc_counter (syn_sent, TCP_ERROR_MSG_QUEUE_FULL, errors);
vlib_buffer_free (vm, first_buffer, from_frame->n_vectors);
if (tc0->flags & TCP_CONN_FINPNDG)
{
/* TX fifo finally drained */
- max_dequeue = session_tx_fifo_max_dequeue (&tc0->connection);
+ max_dequeue = transport_max_tx_dequeue (&tc0->connection);
if (max_dequeue <= tc0->burst_acked)
tcp_send_fin (tc0);
}
/* Don't try to deq the FIN acked */
if (tc0->burst_acked > 1)
- session_dequeue_drop (&tc0->connection, tc0->burst_acked - 1);
+ session_tx_fifo_dequeue_drop (&tc0->connection,
+ tc0->burst_acked - 1);
tc0->burst_acked = 0;
}
break;
if (tc0->flags & TCP_CONN_FINPNDG)
{
/* TX fifo finally drained */
- if (!session_tx_fifo_max_dequeue (&tc0->connection))
+ if (!transport_max_tx_dequeue (&tc0->connection))
{
tcp_send_fin (tc0);
tcp_connection_timers_reset (tc0);
if (tc0->flags & TCP_CONN_FINPNDG)
{
/* Drop all outstanding tx data. */
- session_dequeue_drop (&tc0->connection,
- transport_max_tx_dequeue
- (&tc0->connection));
+ session_tx_fifo_dequeue_drop (&tc0->connection,
+ transport_max_tx_dequeue
+ (&tc0->connection));
tcp_send_fin (tc0);
}
else
}
}
- errors = session_manager_flush_enqueue_events (TRANSPORT_PROTO_TCP,
- thread_index);
+ errors = session_main_flush_enqueue_events (TRANSPORT_PROTO_TCP,
+ thread_index);
tcp_inc_counter (rcv_process, TCP_ERROR_MSG_QUEUE_FULL, errors);
tcp_handle_postponed_dequeues (wrk);
vlib_buffer_free (vm, first_buffer, from_frame->n_vectors);
return 0;
*b = vlib_get_buffer (vm, bi);
data = tcp_init_buffer (vm, *b);
- n_bytes = stream_session_peek_bytes (&tc->connection, data, offset,
- max_deq_bytes);
+ n_bytes = session_tx_fifo_peek_bytes (&tc->connection, data, offset,
+ max_deq_bytes);
ASSERT (n_bytes == max_deq_bytes);
b[0]->current_length = n_bytes;
tcp_push_hdr_i (tc, *b, tc->state, /* compute opts */ 0, /* burst */ 0);
*b = vlib_get_buffer (vm, wrk->tx_buffers[--n_bufs]);
data = tcp_init_buffer (vm, *b);
- n_bytes = stream_session_peek_bytes (&tc->connection, data, offset,
- bytes_per_buffer -
- TRANSPORT_MAX_HDRS_LEN);
+ n_bytes = session_tx_fifo_peek_bytes (&tc->connection, data, offset,
+ bytes_per_buffer -
+ TRANSPORT_MAX_HDRS_LEN);
b[0]->current_length = n_bytes;
b[0]->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
b[0]->total_length_not_including_first_buffer = 0;
chain_b = vlib_get_buffer (vm, chain_bi);
chain_b->current_data = 0;
data = vlib_buffer_get_current (chain_b);
- n_peeked = stream_session_peek_bytes (&tc->connection, data,
- offset + n_bytes, len_to_deq);
+ n_peeked = session_tx_fifo_peek_bytes (&tc->connection, data,
+ offset + n_bytes,
+ len_to_deq);
ASSERT (n_peeked == len_to_deq);
n_bytes += n_peeked;
chain_b->current_length = n_peeked;
/*
* Make sure we can retransmit something
*/
- available_bytes = session_tx_fifo_max_dequeue (&tc->connection);
+ available_bytes = transport_max_tx_dequeue (&tc->connection);
ASSERT (available_bytes >= offset);
available_bytes -= offset;
if (!available_bytes)
|| tc->snd_wnd > tc->snd_mss)
return;
- available_bytes = session_tx_fifo_max_dequeue (&tc->connection);
+ available_bytes = transport_max_tx_dequeue (&tc->connection);
offset = tc->snd_una_max - tc->snd_una;
/* Reprogram persist if no new bytes available to send. We may have data
max_snd_bytes =
clib_min (tc->snd_mss, tm->bytes_per_buffer - TRANSPORT_MAX_HDRS_LEN);
n_bytes =
- stream_session_peek_bytes (&tc->connection, data, offset, max_snd_bytes);
+ session_tx_fifo_peek_bytes (&tc->connection, data, offset, max_snd_bytes);
b->current_length = n_bytes;
ASSERT (n_bytes != 0 && (tcp_timer_is_active (tc, TCP_TIMER_RETRANSMIT)
|| tc->snd_nxt == tc->snd_una_max
sb = &tc->sack_sb;
hole = scoreboard_get_hole (sb, sb->cur_rxt_hole);
- max_deq = session_tx_fifo_max_dequeue (&tc->connection);
+ max_deq = transport_max_tx_dequeue (&tc->connection);
max_deq -= tc->snd_una_max - tc->snd_una;
while (snd_space > 0 && n_segs < burst_size)
if (snd_space < tc->snd_mss || tc->snd_mss == 0)
goto done;
- max_deq = session_tx_fifo_max_dequeue (&tc->connection);
+ max_deq = transport_max_tx_dequeue (&tc->connection);
max_deq -= tc->snd_una_max - tc->snd_una;
if (max_deq)
{
vlib_put_next_frame (vm, node, next_index, n_left_to_next);
}
- errors = session_manager_flush_all_enqueue_events (TRANSPORT_PROTO_UDP);
+ errors = session_main_flush_all_enqueue_events (TRANSPORT_PROTO_UDP);
udp_input_inc_counter (vm, is_ip4, UDP_ERROR_EVENT_FIFO_FULL, errors);
return frame->n_vectors;
}
Code Review / vpp.git / commitdiff