}
}
+static inline u64
+ct_client_seg_handle (u64 server_sh, u32 client_wrk_index)
+{
+ return (((u64) client_wrk_index << 56) | server_sh);
+}
+
static void
ct_session_dealloc_fifos (ct_connection_t *ct, svm_fifo_t *rx_fifo,
svm_fifo_t *tx_fifo)
segment_manager_t *csm;
csm = app_worker_get_connect_segment_manager (app_wrk);
if (!segment_manager_app_detached (csm))
- app_worker_del_segment_notify (app_wrk, ct->segment_handle);
+ app_worker_del_segment_notify (
+ app_wrk, ct_client_seg_handle (ct->segment_handle, ct->client_wrk));
}
/* Notify server app and free segment */
session_set_state (cs, SESSION_STATE_CONNECTING);
cs->app_wrk_index = client_wrk->wrk_index;
cs->connection_index = cct->c_c_index;
+ cs->opaque = opaque;
cct->c_s_index = cs->session_index;
/* This will allocate fifos for the session. They won't be used for
segment_manager_t *sm, u32 client_wrk_index)
{
u32 seg_ctx_index = ~0, sm_index, pair_bytes;
+ u64 seg_size, seg_handle, client_seg_handle;
segment_manager_props_t *props;
const u32 margin = 16 << 10;
ct_segments_ctx_t *seg_ctx;
app_worker_t *client_wrk;
- u64 seg_size, seg_handle;
application_t *server;
ct_segment_t *ct_seg;
uword *spp;
goto error;
client_wrk = app_worker_get (client_wrk_index);
- if (app_worker_add_segment_notify (client_wrk, seg_handle))
+ /* Make sure client workers do not have overlapping segment handles.
+ * Ideally, we should attach fs to client worker segment manager and
+ * create a new handle but that's not currently possible. */
+ client_seg_handle = ct_client_seg_handle (seg_handle, client_wrk_index);
+ if (app_worker_add_segment_notify (client_wrk, client_seg_handle))
{
app_worker_del_segment_notify (server_wrk, seg_handle);
goto error;
cct->client_tx_fifo = ss->rx_fifo;
cct->client_rx_fifo->refcnt++;
cct->client_tx_fifo->refcnt++;
- cct->segment_handle = sct->segment_handle;
+ cct->segment_handle =
+ ct_client_seg_handle (sct->segment_handle, cct->client_wrk);
session_set_state (ss, SESSION_STATE_ACCEPTING);
if (app_worker_accept_notify (server_wrk, ss))
ct_worker_t *wrk;
u8 need_rpc;
- fwrk_index = pointer_to_uword (rpc_args);
- ASSERT (fwrk_index == cm->fwrk_thread);
+ fwrk_index = cm->fwrk_thread;
n_workers = vec_len (cm->fwrk_pending_connects);
for (thread_index = fwrk_index; thread_index < n_workers; thread_index++)
{
+ if (!vec_len (cm->fwrk_pending_connects[thread_index]))
+ continue;
+
wrk = ct_worker_get (thread_index);
/* Connects can be done without worker barrier, grab dst worker lock */
goto global_scope;
ll = listen_session_get_from_handle (lh);
- al = app_listener_get_w_session (ll);
+ al = app_listener_get (ll->al_index);
/*
* Break loop if rule in local table points to connecting app. This
ll = session_lookup_listener_wildcard (table_index, sep);
/* Avoid connecting app to own listener */
- if (ll && ll->app_index != app->app_index)
- return ct_connect (app_wrk, ll, sep_ext);
+ if (ll)
+ {
+ al = app_listener_get (ll->al_index);
+ if (al->app_index != app->app_index)
+ return ct_connect (app_wrk, ll, sep_ext);
+ }
/* Failed to connect but no error */
return SESSION_E_LOCAL_CONNECT;
static inline int
ct_close_is_reset (ct_connection_t *ct, session_t *s)
{
+ if (ct->flags & CT_CONN_F_RESET)
+ return 1;
if (ct->flags & CT_CONN_F_CLIENT)
return (svm_fifo_max_dequeue (ct->client_rx_fifo) > 0);
else
return (svm_fifo_max_dequeue (s->rx_fifo) > 0);
}
+static void
+ct_session_cleanup_server_session (session_t *s)
+{
+ ct_connection_t *ct;
+
+ ct = (ct_connection_t *) session_get_transport (s);
+ ct_session_dealloc_fifos (ct, s->rx_fifo, s->tx_fifo);
+ session_free (s);
+ ct_connection_free (ct);
+}
+
static void
ct_session_postponed_cleanup (ct_connection_t *ct)
{
}
session_transport_closed_notify (&ct->connection);
+ /* It would be cleaner to call session_transport_delete_notify
+ * but then we can't control session cleanup lower */
+ session_set_state (s, SESSION_STATE_TRANSPORT_DELETED);
+ if (app_wrk)
+ app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
+
if (ct->flags & CT_CONN_F_CLIENT)
{
- if (app_wrk)
- app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
-
/* Normal free for client session as the fifos are allocated through
* the connects segment manager in a segment that's not shared with
* the server */
ct_session_dealloc_fifos (ct, ct->client_rx_fifo, ct->client_tx_fifo);
- session_free_w_fifos (s);
+ session_program_cleanup (s);
+ ct_connection_free (ct);
}
else
{
/* Manual session and fifo segment cleanup to avoid implicit
* segment manager cleanups and notifications */
- app_wrk = app_worker_get_if_valid (s->app_wrk_index);
if (app_wrk)
{
- app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
- app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_SESSION);
+ /* Remove custom cleanup notify infra when/if switching to normal
+ * session cleanup. Note that ct is freed in the cb function */
+ app_worker_cleanup_notify_custom (app_wrk, s,
+ SESSION_CLEANUP_SESSION,
+ ct_session_cleanup_server_session);
+ }
+ else
+ {
+ ct_connection_free (ct);
}
-
- ct_session_dealloc_fifos (ct, s->rx_fifo, s->tx_fifo);
- session_free (s);
}
-
- ct_connection_free (ct);
}
static void
clib_fifo_sub2 (wrk->pending_cleanups, req);
ct = ct_connection_get (req->ct_index, thread_index);
s = session_get (ct->c_s_index, ct->c_thread_index);
- if (!svm_fifo_has_event (s->tx_fifo))
- ct_session_postponed_cleanup (ct);
- else
+ if (svm_fifo_has_event (s->tx_fifo) || (s->flags & SESSION_F_RX_EVT))
clib_fifo_add1 (wrk->pending_cleanups, *req);
+ else
+ ct_session_postponed_cleanup (ct);
n_to_handle -= 1;
}
ct_program_cleanup (ct);
}
+static void
+ct_session_reset (u32 ct_index, u32 thread_index)
+{
+ ct_connection_t *ct;
+ ct = ct_connection_get (ct_index, thread_index);
+ ct->flags |= CT_CONN_F_RESET;
+ ct_session_close (ct_index, thread_index);
+}
+
static transport_connection_t *
ct_session_get (u32 ct_index, u32 thread_index)
{
return 0;
}
-/* *INDENT-OFF* */
static const transport_proto_vft_t cut_thru_proto = {
.enable = ct_enable_disable,
.start_listen = ct_start_listen,
.cleanup_ho = ct_cleanup_ho,
.connect = ct_session_connect,
.close = ct_session_close,
+ .reset = ct_session_reset,
.custom_tx = ct_custom_tx,
.app_rx_evt = ct_app_rx_evt,
.format_listener = format_ct_listener,
.service_type = TRANSPORT_SERVICE_VC,
},
};
-/* *INDENT-ON* */
static inline int
ct_session_can_tx (session_t *s)
peer_s = session_get (peer_ct->c_s_index, peer_ct->c_thread_index);
if (peer_s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
return 0;
+ peer_s->flags |= SESSION_F_RX_EVT;
return session_enqueue_notify (peer_s);
}