ssvm_pop_heap (oldheap);
if (f)
{
- fss->num_chunks[fl_index] += 1;
+ clib_atomic_fetch_add_rel (&fss->num_chunks[fl_index], 1);
fsh_free_bytes_sub (fsh, fifo_sz);
goto done;
}
if (c)
{
- fss->num_chunks[fl_index] += 1;
+ clib_atomic_fetch_add_rel (&fss->num_chunks[fl_index], 1);
fsh_free_bytes_sub (fsh, chunk_size + sizeof (*c));
goto done;
}
fsh_slice_collect_chunks (fsh, fss, c);
}
+static inline void
+fss_fifo_add_active_list (fifo_segment_slice_t * fss, svm_fifo_t * f)
+{
+ if (fss->fifos)
+ {
+ fss->fifos->prev = f;
+ f->next = fss->fifos;
+ }
+ fss->fifos = f;
+}
+
+static inline void
+fss_fifo_del_active_list (fifo_segment_slice_t * fss, svm_fifo_t * f)
+{
+ if (f->flags & SVM_FIFO_F_LL_TRACKED)
+ {
+ if (f->prev)
+ f->prev->next = f->next;
+ else
+ fss->fifos = f->next;
+ if (f->next)
+ f->next->prev = f->prev;
+ }
+}
+
/**
* Allocate fifo in fifo segment
*/
* only one. */
if (ftype == FIFO_SEGMENT_RX_FIFO)
{
- if (fss->fifos)
- {
- fss->fifos->prev = f;
- f->next = fss->fifos;
- }
- fss->fifos = f;
+ fss_fifo_add_active_list (fss, f);
f->flags |= SVM_FIFO_F_LL_TRACKED;
svm_fifo_init_ooo_lookup (f, 0 /* ooo enq */ );
/* Remove from active list. Only rx fifos are tracked */
if (f->flags & SVM_FIFO_F_LL_TRACKED)
{
- if (f->prev)
- f->prev->next = f->next;
- else
- fss->fifos = f->next;
- if (f->next)
- f->next->prev = f->prev;
+ fss_fifo_del_active_list (fss, f);
f->flags &= ~SVM_FIFO_F_LL_TRACKED;
}
fsh_active_fifos_update (fsh, -1);
}
+void
+fifo_segment_detach_fifo (fifo_segment_t * fs, svm_fifo_t * f)
+{
+ fifo_segment_slice_t *fss;
+ svm_fifo_chunk_t *c;
+ u32 fl_index;
+
+ ASSERT (f->refcnt == 1);
+
+ fss = fsh_slice_get (fs->h, f->slice_index);
+ fss->virtual_mem -= svm_fifo_size (f);
+ if (f->flags & SVM_FIFO_F_LL_TRACKED)
+ fss_fifo_del_active_list (fss, f);
+
+ c = f->start_chunk;
+ while (c)
+ {
+ fl_index = fs_freelist_for_size (c->length);
+ clib_atomic_fetch_sub_rel (&fss->num_chunks[fl_index], 1);
+ c = c->next;
+ }
+}
+
+void
+fifo_segment_attach_fifo (fifo_segment_t * fs, svm_fifo_t * f,
+ u32 slice_index)
+{
+ fifo_segment_slice_t *fss;
+ svm_fifo_chunk_t *c;
+ u32 fl_index;
+
+ f->slice_index = slice_index;
+ fss = fsh_slice_get (fs->h, f->slice_index);
+ fss->virtual_mem += svm_fifo_size (f);
+ if (f->flags & SVM_FIFO_F_LL_TRACKED)
+ fss_fifo_add_active_list (fss, f);
+
+ c = f->start_chunk;
+ while (c)
+ {
+ fl_index = fs_freelist_for_size (c->length);
+ clib_atomic_fetch_add_rel (&fss->num_chunks[fl_index], 1);
+ c = c->next;
+ }
+}
+
int
fifo_segment_prealloc_fifo_hdrs (fifo_segment_t * fs, u32 slice_index,
u32 batch_size)
*/
void fifo_segment_free_fifo (fifo_segment_t * fs, svm_fifo_t * f);
+void fifo_segment_detach_fifo (fifo_segment_t * fs, svm_fifo_t * f);
+void fifo_segment_attach_fifo (fifo_segment_t * fs, svm_fifo_t * f,
+ u32 slice_index);
+
/**
* Try to preallocate fifo headers
*
segment_manager_segment_reader_unlock (sm);
}
+void
+segment_manager_detach_fifo (segment_manager_t * sm, svm_fifo_t * f)
+{
+ fifo_segment_t *fs;
+
+ fs = segment_manager_get_segment_w_lock (sm, f->segment_index);
+ fifo_segment_detach_fifo (fs, f);
+ segment_manager_segment_reader_unlock (sm);
+}
+
+void
+segment_manager_attach_fifo (segment_manager_t * sm, svm_fifo_t * f,
+ session_t * s)
+{
+ fifo_segment_t *fs;
+
+ fs = segment_manager_get_segment_w_lock (sm, f->segment_index);
+ fifo_segment_attach_fifo (fs, f, s->thread_index);
+ segment_manager_segment_reader_unlock (sm);
+
+ f->master_session_index = s->session_index;
+ f->master_thread_index = s->thread_index;
+}
+
u32
segment_manager_evt_q_expected_size (u32 q_len)
{
svm_fifo_t ** tx_fifo);
void segment_manager_dealloc_fifos (svm_fifo_t * rx_fifo,
svm_fifo_t * tx_fifo);
+void segment_manager_detach_fifo (segment_manager_t * sm, svm_fifo_t * f);
+void segment_manager_attach_fifo (segment_manager_t * sm, svm_fifo_t * f,
+ session_t * s);
void segment_manager_set_watermarks (segment_manager_t * sm,
u8 high_watermark, u8 low_watermark);
return session_stream_connect_notify_inline (tc, err, SESSION_STATE_OPENED);
}
+static void
+session_switch_pool_reply (void *arg)
+{
+ u32 session_index = pointer_to_uword (arg);
+ segment_manager_t *sm;
+ app_worker_t *app_wrk;
+ session_t *s;
+
+ s = session_get_if_valid (session_index, vlib_get_thread_index ());
+ if (!s)
+ return;
+
+ app_wrk = app_worker_get_if_valid (s->app_wrk_index);
+ if (!app_wrk)
+ return;
+
+ /* Attach fifos to the right session and segment slice */
+ sm = app_worker_get_connect_segment_manager (app_wrk);
+ segment_manager_attach_fifo (sm, s->rx_fifo, s);
+ segment_manager_attach_fifo (sm, s->tx_fifo, s);
+
+ /* Notify app that it has data on the new session */
+ session_enqueue_notify (s);
+}
+
typedef struct _session_switch_pool_args
{
u32 session_index;
session_switch_pool (void *cb_args)
{
session_switch_pool_args_t *args = (session_switch_pool_args_t *) cb_args;
+ session_handle_t new_sh;
+ segment_manager_t *sm;
app_worker_t *app_wrk;
session_t *s;
+ void *rargs;
ASSERT (args->thread_index == vlib_get_thread_index ());
s = session_get (args->session_index, args->thread_index);
- s->tx_fifo->master_session_index = args->new_session_index;
- s->tx_fifo->master_thread_index = args->new_thread_index;
+
transport_cleanup (session_get_transport_proto (s), s->connection_index,
s->thread_index);
+ new_sh = session_make_handle (args->new_session_index,
+ args->new_thread_index);
+
app_wrk = app_worker_get_if_valid (s->app_wrk_index);
if (app_wrk)
{
- session_handle_t new_sh;
- new_sh = session_make_handle (args->new_session_index,
- args->new_thread_index);
- app_worker_migrate_notify (app_wrk, s, new_sh);
+ /* Cleanup fifo segment slice state for fifos */
+ sm = app_worker_get_connect_segment_manager (app_wrk);
+ segment_manager_detach_fifo (sm, s->rx_fifo);
+ segment_manager_detach_fifo (sm, s->tx_fifo);
- /* Trigger app read on the new thread */
- session_enqueue_notify_thread (new_sh);
+ /* Notify app, using old session, about the migration event */
+ app_worker_migrate_notify (app_wrk, s, new_sh);
}
+ /* Trigger app read and fifo updates on the new thread */
+ rargs = uword_to_pointer (args->new_session_index, void *);
+ session_send_rpc_evt_to_thread (args->new_thread_index,
+ session_switch_pool_reply, rargs);
+
session_free (s);
clib_mem_free (cb_args);
}
*/
new_s = session_clone_safe (tc->s_index, old_thread_index);
new_s->connection_index = tc->c_index;
- new_s->rx_fifo->master_session_index = new_s->session_index;
- new_s->rx_fifo->master_thread_index = new_s->thread_index;
new_s->session_state = SESSION_STATE_READY;
new_s->flags |= SESSION_F_IS_MIGRATING;
+
session_lookup_add_connection (tc, session_handle (new_s));
/*