pool_get_aligned_will_expand (smm->sessions[thread_index], will_expand,
CLIB_CACHE_LINE_BYTES);
/* If we have peekers, let them finish */
- if (PREDICT_FALSE (will_expand))
+ if (PREDICT_FALSE (will_expand && vlib_num_workers ()))
{
- clib_spinlock_lock_if_init (&smm->peekers_write_locks[thread_index]);
+ clib_rwlock_writer_lock (&smm->peekers_rw_locks[thread_index]);
pool_get_aligned (session_manager_main.sessions[thread_index], s,
CLIB_CACHE_LINE_BYTES);
- clib_spinlock_unlock_if_init (&smm->peekers_write_locks[thread_index]);
+ clib_rwlock_writer_unlock (&smm->peekers_rw_locks[thread_index]);
}
else
{
return s;
}
-static void
+void
session_free (stream_session_t * s)
{
pool_put (session_manager_main.sessions[s->thread_index], s);
memset (s, 0xFA, sizeof (*s));
}
-static int
+int
session_alloc_fifos (segment_manager_t * sm, stream_session_t * s)
{
svm_fifo_t *server_rx_fifo = 0, *server_tx_fifo = 0;
return 0;
}
- /* Built-in server? Hand event to the callback... */
- if (app->cb_fns.builtin_server_rx_callback)
- return app->cb_fns.builtin_server_rx_callback (s);
+ /* Built-in app? Hand event to the callback... */
+ if (app->cb_fns.builtin_app_rx_callback)
+ return app->cb_fns.builtin_app_rx_callback (s);
/* If no event, send one */
if (svm_fifo_set_event (s->server_rx_fifo))
int
session_stream_connect_notify (transport_connection_t * tc, u8 is_fail)
{
- application_t *app;
+ u32 opaque = 0, new_ti, new_si;
stream_session_t *new_s = 0;
- u64 handle;
- u32 opaque = 0;
- int error = 0;
segment_manager_t *sm;
+ application_t *app;
u8 alloc_fifos;
+ int error = 0;
+ u64 handle;
/*
* Find connection handle and cleanup half-open table
error = -1;
}
else
- new_s->app_index = app->index;
+ {
+ new_s->app_index = app->index;
+ new_si = new_s->session_index;
+ new_ti = new_s->thread_index;
+ }
}
/*
{
SESSION_DBG ("failed to notify app");
if (!is_fail)
- stream_session_disconnect (new_s);
+ {
+ new_s = session_get (new_si, new_ti);
+ stream_session_disconnect_transport (new_s);
+ }
}
else
{
if (!is_fail)
- new_s->session_state = SESSION_STATE_READY;
+ {
+ new_s = session_get (new_si, new_ti);
+ new_s->session_state = SESSION_STATE_READY;
+ }
}
return error;
return 0;
}
-/**
- * Ask transport to open connection to remote transport endpoint.
- *
- * Stores handle for matching request with reply since the call can be
- * asynchronous. For instance, for TCP the 3-way handshake must complete
- * before reply comes. Session is only created once connection is established.
- *
- * @param app_index Index of the application requesting the connect
- * @param st Session type requested.
- * @param tep Remote transport endpoint
- * @param opaque Opaque data (typically, api_context) the application expects
- * on open completion.
- */
int
-session_open (u32 app_index, session_endpoint_t * rmt, u32 opaque)
+session_open_cl (u32 app_index, session_endpoint_t * rmt, u32 opaque)
{
transport_connection_t *tc;
transport_endpoint_t *tep;
stream_session_t *s;
application_t *app;
int rv;
+
+ tep = session_endpoint_to_transport (rmt);
+ rv = tp_vfts[rmt->transport_proto].open (tep);
+ if (rv < 0)
+ {
+ SESSION_DBG ("Transport failed to open connection.");
+ return VNET_API_ERROR_SESSION_CONNECT;
+ }
+
+ tc = tp_vfts[rmt->transport_proto].get_half_open ((u32) rv);
+
+ /* For dgram type of service, allocate session and fifos now.
+ */
+ app = application_get (app_index);
+ sm = application_get_connect_segment_manager (app);
+
+ if (session_alloc_and_init (sm, tc, 1, &s))
+ return -1;
+ s->app_index = app->index;
+ s->session_state = SESSION_STATE_CONNECTING_READY;
+
+ /* Tell the app about the new event fifo for this session */
+ app->cb_fns.session_connected_callback (app->index, opaque, s, 0);
+
+ return 0;
+}
+
+int
+session_open_vc (u32 app_index, session_endpoint_t * rmt, u32 opaque)
+{
+ transport_connection_t *tc;
+ transport_endpoint_t *tep;
u64 handle;
+ int rv;
+
+ /* TODO until udp is fixed */
+ if (rmt->transport_proto == TRANSPORT_PROTO_UDP)
+ return session_open_cl (app_index, rmt, opaque);
tep = session_endpoint_to_transport (rmt);
rv = tp_vfts[rmt->transport_proto].open (tep);
/* If transport offers a stream service, only allocate session once the
* connection has been established.
+ * Add connection to half-open table and save app and tc index. The
+ * latter is needed to help establish the connection while the former
+ * is needed when the connect notify comes and we have to notify the
+ * external app
*/
- if (transport_is_stream (rmt->transport_proto))
- {
- /* Add connection to half-open table and save app and tc index. The
- * latter is needed to help establish the connection while the former
- * is needed when the connect notify comes and we have to notify the
- * external app
- */
- handle = (((u64) app_index) << 32) | (u64) tc->c_index;
- session_lookup_add_half_open (tc, handle);
-
- /* Store api_context (opaque) for when the reply comes. Not the nicest
- * thing but better than allocating a separate half-open pool.
- */
- tc->s_index = opaque;
- }
- /* For dgram type of service, allocate session and fifos now.
+ handle = (((u64) app_index) << 32) | (u64) tc->c_index;
+ session_lookup_add_half_open (tc, handle);
+
+ /* Store api_context (opaque) for when the reply comes. Not the nicest
+ * thing but better than allocating a separate half-open pool.
*/
- else
- {
- app = application_get (app_index);
- sm = application_get_connect_segment_manager (app);
+ tc->s_index = opaque;
+ return 0;
+}
- if (session_alloc_and_init (sm, tc, 1, &s))
- return -1;
- s->app_index = app->index;
- s->session_state = SESSION_STATE_CONNECTING_READY;
+int
+session_open_app (u32 app_index, session_endpoint_t * rmt, u32 opaque)
+{
+ session_endpoint_extended_t *sep = (session_endpoint_extended_t *) rmt;
+ sep->app_index = app_index;
+ sep->opaque = opaque;
- /* Tell the app about the new event fifo for this session */
- app->cb_fns.session_connected_callback (app->index, opaque, s, 0);
- }
- return 0;
+ return tp_vfts[rmt->transport_proto].open ((transport_endpoint_t *) sep);
+}
+
+typedef int (*session_open_service_fn) (u32, session_endpoint_t *, u32);
+
+/* *INDENT-OFF* */
+static session_open_service_fn session_open_srv_fns[TRANSPORT_N_SERVICES] = {
+ session_open_vc,
+ session_open_cl,
+ session_open_app,
+};
+/* *INDENT-ON* */
+
+/**
+ * Ask transport to open connection to remote transport endpoint.
+ *
+ * Stores handle for matching request with reply since the call can be
+ * asynchronous. For instance, for TCP the 3-way handshake must complete
+ * before reply comes. Session is only created once connection is established.
+ *
+ * @param app_index Index of the application requesting the connect
+ * @param st Session type requested.
+ * @param tep Remote transport endpoint
+ * @param opaque Opaque data (typically, api_context) the application expects
+ * on open completion.
+ */
+int
+session_open (u32 app_index, session_endpoint_t * rmt, u32 opaque)
+{
+ transport_service_type_t tst = tp_vfts[rmt->transport_proto].service_type;
+ return session_open_srv_fns[tst] (app_index, rmt, opaque);
}
/**
* @param tep Local endpoint to be listened on.
*/
int
-stream_session_listen (stream_session_t * s, session_endpoint_t * sep)
+session_listen_vc (stream_session_t * s, session_endpoint_t * sep)
{
transport_connection_t *tc;
u32 tci;
return 0;
}
+int
+session_listen_app (stream_session_t * s, session_endpoint_t * sep)
+{
+ session_endpoint_extended_t esep;
+ clib_memcpy (&esep, sep, sizeof (*sep));
+ esep.app_index = s->app_index;
+
+ return tp_vfts[sep->transport_proto].bind (s->session_index,
+ (transport_endpoint_t *) & esep);
+}
+
+typedef int (*session_listen_service_fn) (stream_session_t *,
+ session_endpoint_t *);
+
+/* *INDENT-OFF* */
+static session_listen_service_fn
+session_listen_srv_fns[TRANSPORT_N_SERVICES] = {
+ session_listen_vc,
+ session_listen_vc,
+ session_listen_app,
+};
+/* *INDENT-ON* */
+
+int
+stream_session_listen (stream_session_t * s, session_endpoint_t * sep)
+{
+ transport_service_type_t tst = tp_vfts[sep->transport_proto].service_type;
+ return session_listen_srv_fns[tst] (s, sep);
+}
+
/**
* Ask transport to stop listening on local transport endpoint.
*
}
/**
- * Disconnect session and propagate to transport. This should eventually
+ * Initialize session disconnect.
+ *
+ * Request is always sent to session node to ensure that all outstanding
+ * requests are served before transport is notified.
+ */
+void
+stream_session_disconnect (stream_session_t * s)
+{
+ if (!s || s->session_state == SESSION_STATE_CLOSED)
+ return;
+ s->session_state = SESSION_STATE_CLOSED;
+ session_send_session_evt_to_thread (session_handle (s),
+ FIFO_EVENT_DISCONNECT, s->thread_index);
+}
+
+/**
+ * Notify transport the session can be disconnected. This should eventually
* result in a delete notification that allows us to cleanup session state.
* Called for both active/passive disconnects.
*
- * Should be called from the session's thread.
+ * Must be called from the session's thread.
*/
void
-stream_session_disconnect (stream_session_t * s)
+stream_session_disconnect_transport (stream_session_t * s)
{
s->session_state = SESSION_STATE_CLOSED;
tp_vfts[session_get_transport_proto (s)].close (s->connection_index,
return 0;
}
+/* *INDENT-OFF* */
+static session_fifo_rx_fn *session_tx_fns[TRANSPORT_TX_N_FNS] = {
+ session_tx_fifo_peek_and_snd,
+ session_tx_fifo_dequeue_and_snd,
+ session_tx_fifo_dequeue_internal
+};
+/* *INDENT-ON* */
+
/**
* Initialize session layer for given transport proto and ip version
*
vec_validate (smm->session_tx_fns, session_type);
/* *INDENT-OFF* */
- foreach_vlib_main (({
- next_index = vlib_node_add_next (this_vlib_main, session_queue_node.index,
- output_node);
- }));
+ if (output_node != ~0)
+ {
+ foreach_vlib_main (({
+ next_index = vlib_node_add_next (this_vlib_main,
+ session_queue_node.index,
+ output_node);
+ }));
+ }
/* *INDENT-ON* */
smm->session_type_to_next[session_type] = next_index;
- session_manager_set_transport_rx_fn (session_type,
- vft->tx_fifo_offset != 0);
+ smm->session_tx_fns[session_type] = session_tx_fns[vft->tx_type];
}
transport_connection_t *
static clib_error_t *
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;
vlib_thread_main_t *vtm = vlib_get_thread_main ();
- u32 num_threads;
- u32 preallocated_sessions_per_worker;
+ u32 num_threads, preallocated_sessions_per_worker;
int i, j;
num_threads = 1 /* main thread */ + vtm->n_threads;
vec_validate (smm->pending_disconnects, num_threads - 1);
vec_validate (smm->free_event_vector, num_threads - 1);
vec_validate (smm->vpp_event_queues, num_threads - 1);
- vec_validate (smm->session_peekers, num_threads - 1);
- vec_validate (smm->peekers_readers_locks, num_threads - 1);
- vec_validate (smm->peekers_write_locks, num_threads - 1);
+ vec_validate (smm->peekers_rw_locks, num_threads - 1);
for (i = 0; i < TRANSPORT_N_PROTO; i++)
for (j = 0; j < num_threads; j++)
vec_validate (smm->pending_disconnects[i], 0);
_vec_len (smm->pending_disconnects[i]) = 0;
if (num_threads > 1)
- {
- clib_spinlock_init (&smm->peekers_readers_locks[i]);
- clib_spinlock_init (&smm->peekers_write_locks[i]);
- }
+ clib_rwlock_init (&smm->peekers_rw_locks[i]);
}
#if SESSION_DBG
session_vpp_event_queues_allocate (smm);
/* Initialize fifo segment main baseva and timeout */
- svm_fifo_segment_init (smm->session_baseva + smm->evt_qs_segment_size,
- smm->segment_timeout);
+ sm_args->baseva = smm->session_baseva + smm->evt_qs_segment_size;
+ sm_args->size = smm->session_va_space_size;
+ segment_manager_main_init (sm_args);
/* Preallocate sessions */
if (smm->preallocated_sessions)
{
session_manager_main_t *smm = &session_manager_main;
smm->session_baseva = 0x200000000ULL;
- smm->segment_timeout = 20;
+ smm->session_va_space_size = (u64) 128 << 30;
smm->evt_qs_segment_size = 64 << 20;
smm->is_enabled = 0;
return 0;
Code Review / vpp.git / blobdiff