typedef struct vls_worker_
{
+ clib_rwlock_t sh_to_vlsh_table_lock; /** valid for multithread workers */
vcl_locked_session_t *vls_pool;
uword *session_handle_to_vlsh_table;
u32 wrk_index;
pthread_mutex_t vls_mt_mq_mlock;
pthread_mutex_t vls_mt_spool_mlock;
volatile u8 select_mp_check;
- volatile u8 epoll_mp_check;
} vls_process_local_t;
static vls_process_local_t vls_local;
/** Pool of data shared by sessions owned by different workers */
vls_shared_data_t *shared_data_pool;
clib_rwlock_t shared_data_lock;
+ /** Lock to protect rpc among workers */
+ clib_spinlock_t worker_rpc_lock;
} vls_main_t;
vls_main_t *vlsm;
+typedef enum
+{
+ VLS_RPC_STATE_INIT,
+ VLS_RPC_STATE_SUCCESS,
+ VLS_RPC_STATE_SESSION_NOT_EXIST,
+} vls_rpc_state_e;
+
typedef enum vls_rpc_msg_type_
{
VLS_RPC_CLONE_AND_SHARE,
void vls_send_session_cleanup_rpc (vcl_worker_t * wrk,
u32 dst_wrk_index, u32 dst_session_index);
-void vls_send_clone_and_share_rpc (vcl_worker_t * wrk,
- vcl_locked_session_t * vls,
+void vls_send_clone_and_share_rpc (vcl_worker_t *wrk, u32 origin_vls_index,
u32 session_index, u32 vls_wrk_index,
u32 dst_wrk_index, u32 dst_vls_index,
u32 dst_session_index);
-
static inline u32
vls_get_worker_index (void)
{
* vcl worker with vpp. Otherwise, all threads use the same vcl worker, so
* update the vcl worker's thread local worker index variable */
if (vls_mt_wrk_supported ())
- vls_register_vcl_worker ();
+ {
+ if (vppcom_worker_register () != VPPCOM_OK)
+ VERR ("failed to register worker");
+ }
else
vcl_set_worker_index (vlsl->vls_wrk_index);
}
vls_worker_t *wrk;
pool_get_zero (vlsm->workers, wrk);
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_init (&wrk->sh_to_vlsh_table_lock);
wrk->wrk_index = vcl_get_worker_index ();
}
vls_worker_free (vls_worker_t * wrk)
{
hash_free (wrk->session_handle_to_vlsh_table);
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_free (&wrk->sh_to_vlsh_table_lock);
pool_free (wrk->vls_pool);
pool_put (vlsm->workers, wrk);
}
return pool_elt_at_index (vlsm->workers, wrk_index);
}
+static void
+vls_sh_to_vlsh_table_add (vls_worker_t *wrk, vcl_session_handle_t sh, u32 vlsh)
+{
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_writer_lock (&wrk->sh_to_vlsh_table_lock);
+ hash_set (wrk->session_handle_to_vlsh_table, sh, vlsh);
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_writer_unlock (&wrk->sh_to_vlsh_table_lock);
+}
+
+static void
+vls_sh_to_vlsh_table_del (vls_worker_t *wrk, vcl_session_handle_t sh)
+{
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_writer_lock (&wrk->sh_to_vlsh_table_lock);
+ hash_unset (wrk->session_handle_to_vlsh_table, sh);
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_writer_unlock (&wrk->sh_to_vlsh_table_lock);
+}
+
+static uword *
+vls_sh_to_vlsh_table_get (vls_worker_t *wrk, vcl_session_handle_t sh)
+{
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_reader_lock (&wrk->sh_to_vlsh_table_lock);
+ uword *vlshp = hash_get (wrk->session_handle_to_vlsh_table, sh);
+ if (vls_mt_wrk_supported ())
+ clib_rwlock_reader_unlock (&wrk->sh_to_vlsh_table_lock);
+ return vlshp;
+}
+
static vls_handle_t
vls_alloc (vcl_session_handle_t sh)
{
vls->worker_index = vppcom_session_worker (sh);
vls->vls_index = vls - wrk->vls_pool;
vls->shared_data_index = ~0;
- hash_set (wrk->session_handle_to_vlsh_table, sh, vls->vls_index);
+ vls_sh_to_vlsh_table_add (wrk, sh, vls->vls_index);
if (vls_mt_wrk_supported ())
{
hash_set (vls->vcl_wrk_index_to_session_index, vls->worker_index,
vls_worker_t *wrk = vls_worker_get_current ();
ASSERT (vls != 0);
- hash_unset (wrk->session_handle_to_vlsh_table,
- vcl_session_handle_from_index (vls->session_index));
+ vls_sh_to_vlsh_table_del (
+ wrk, vcl_session_handle_from_index (vls->session_index));
clib_spinlock_free (&vls->lock);
pool_put (wrk->vls_pool, vls);
}
vls_si_wi_to_vlsh (u32 session_index, u32 vcl_wrk_index)
{
vls_worker_t *wrk = vls_worker_get_current ();
- uword *vlshp;
- vlshp = hash_get (wrk->session_handle_to_vlsh_table,
- vcl_session_handle_from_wrk_session_index (session_index,
- vcl_wrk_index));
+ uword *vlshp = vls_sh_to_vlsh_table_get (
+ wrk,
+ vcl_session_handle_from_wrk_session_index (session_index, vcl_wrk_index));
+
return vlshp ? *vlshp : VLS_INVALID_HANDLE;
}
vcl_locked_session_t *vls, *parent_vls;
/* *INDENT-OFF* */
- pool_foreach (vls, vls_wrk->vls_pool, ({
+ pool_foreach (vls, vls_wrk->vls_pool) {
/* Initialize sharing on parent session */
if (vls->shared_data_index == ~0)
{
vls->shared_data_index = parent_vls->shared_data_index;
}
vls_share_session (vls_wrk, vls);
- }));
+ }
/* *INDENT-ON* */
}
&& vls->worker_index != vcl_get_worker_index ());
}
-static void
-vls_mt_session_migrate (vcl_locked_session_t * vls)
+static vcl_locked_session_t *
+vls_mt_session_migrate (vcl_locked_session_t *vls)
{
u32 wrk_index = vcl_get_worker_index ();
vcl_worker_t *wrk;
- u32 src_sid, sid;
+ vls_worker_t *vls_wrk = vls_worker_get_current ();
+ u32 src_sid, sid, vls_index, own_vcl_wrk_index;
vcl_session_t *session;
uword *p;
{
vls->worker_index = wrk_index;
vls->session_index = (u32) p[0];
- return;
+ return vls;
}
/*
{
VERR ("session in owner worker(%u) is free", vls->owner_vcl_wrk_index);
ASSERT (0);
- return;
+ vls_unlock (vls);
+ vls_mt_table_runlock ();
+ return 0;
}
src_sid = (u32) p[0];
wrk = vcl_worker_get_current ();
session = vcl_session_alloc (wrk);
sid = session->session_index;
- vls_send_clone_and_share_rpc (wrk, vls, sid, vls_get_worker_index (),
- vls->owner_vcl_wrk_index, vls->vls_index,
- src_sid);
- session->session_index = sid;
- vls->worker_index = wrk_index;
- vls->session_index = sid;
- hash_set (vls->vcl_wrk_index_to_session_index, wrk_index, sid);
VDBG (1, "migrate session of worker (session): %u (%u) -> %u (%u)",
vls->owner_vcl_wrk_index, src_sid, wrk_index, sid);
- if (PREDICT_FALSE ((session->flags & VCL_SESSION_F_IS_VEP)
- && session->vep.next_sh != ~0))
+ /* Drop lock to prevent dead lock when dst wrk trying to get lock. */
+ vls_index = vls->vls_index;
+ own_vcl_wrk_index = vls->owner_vcl_wrk_index;
+ vls_unlock (vls);
+ vls_mt_table_runlock ();
+ vls_send_clone_and_share_rpc (wrk, vls_index, sid, vls_get_worker_index (),
+ own_vcl_wrk_index, vls_index, src_sid);
+
+ if (PREDICT_FALSE (wrk->rpc_done == VLS_RPC_STATE_SESSION_NOT_EXIST))
+ {
+ VWRN ("session %u not exist", src_sid);
+ goto err;
+ }
+ else if (PREDICT_FALSE (wrk->rpc_done == VLS_RPC_STATE_INIT))
+ {
+ VWRN ("failed to wait rpc response");
+ goto err;
+ }
+ else if (PREDICT_FALSE ((session->flags & VCL_SESSION_F_IS_VEP) &&
+ session->vep.next_sh != ~0))
{
- /* TODO: rollback? */
VERR ("can't migrate nonempty epoll session");
ASSERT (0);
- return;
+ goto err;
}
else if (PREDICT_FALSE (!(session->flags & VCL_SESSION_F_IS_VEP) &&
session->session_state != VCL_STATE_CLOSED))
{
- /* TODO: rollback? */
VERR ("migrate NOT supported, session_status (%u)",
session->session_state);
ASSERT (0);
- return;
+ goto err;
}
+
+ vls = vls_get_w_dlock (vls_index);
+ if (PREDICT_FALSE (!vls))
+ {
+ VWRN ("failed to get vls %u", vls_index);
+ goto err;
+ }
+
+ session->session_index = sid;
+ vls->worker_index = wrk_index;
+ vls->session_index = sid;
+ hash_set (vls->vcl_wrk_index_to_session_index, wrk_index, sid);
+ vls_sh_to_vlsh_table_add (vls_wrk, vcl_session_handle (session),
+ vls->vls_index);
+ return vls;
+
+err:
+ vcl_session_free (wrk, session);
+ return 0;
}
static inline void
vls_mt_add ();
}
-#define vls_mt_guard(_vls, _op) \
- int _locks_acq = 0; \
- if (vls_mt_wrk_supported ()) \
- { \
- if (PREDICT_FALSE (_vls \
- && ((vcl_locked_session_t *)_vls)->worker_index != \
- vcl_get_worker_index ())) \
- vls_mt_session_migrate (_vls); \
- } \
- else \
- { \
- if (PREDICT_FALSE (vlsl->vls_mt_n_threads > 1)) \
- vls_mt_acq_locks (_vls, _op, &_locks_acq); \
- } \
+#define vls_mt_guard(_vls, _op) \
+ int _locks_acq = 0; \
+ if (vls_mt_wrk_supported ()) \
+ { \
+ if (PREDICT_FALSE (_vls && \
+ ((vcl_locked_session_t *) _vls)->worker_index != \
+ vcl_get_worker_index ())) \
+ { \
+ _vls = vls_mt_session_migrate (_vls); \
+ if (PREDICT_FALSE (!_vls)) \
+ return VPPCOM_EBADFD; \
+ } \
+ } \
+ else \
+ { \
+ if (PREDICT_FALSE (vlsl->vls_mt_n_threads > 1)) \
+ vls_mt_acq_locks (_vls, _op, &_locks_acq); \
+ }
#define vls_mt_unguard() \
if (PREDICT_FALSE (_locks_acq)) \
if (!(vls = vls_get_w_dlock (vlsh)))
return VPPCOM_EBADFD;
if (vls_mt_session_should_migrate (vls))
- vls_mt_session_migrate (vls);
+ {
+ vls = vls_mt_session_migrate (vls);
+ if (PREDICT_FALSE (!vls))
+ return VPPCOM_EBADFD;
+ }
rv = vppcom_session_attr (vls_to_sh_tu (vls), op, buffer, buflen);
vls_get_and_unlock (vlsh);
return rv;
{
vcl_session_handle_t sh;
vls_handle_t vlsh;
+ vcl_locked_session_t *vls = NULL;
vls_mt_detect ();
- vls_mt_guard (0, VLS_MT_OP_SPOOL);
+ vls_mt_guard (vls, VLS_MT_OP_SPOOL);
sh = vppcom_session_create (proto, is_nonblocking);
vls_mt_unguard ();
if (sh == INVALID_SESSION_ID)
return rv;
}
+int
+vls_shutdown (vls_handle_t vlsh, int how)
+{
+ vcl_locked_session_t *vls;
+ int rv;
+
+ vls_mt_detect ();
+ if (!(vls = vls_get_w_dlock (vlsh)))
+ return VPPCOM_EBADFD;
+
+ vls_mt_guard (vls, VLS_MT_OP_SPOOL);
+ rv = vppcom_session_shutdown (vls_to_sh (vls), how);
+ vls_mt_unguard ();
+ vls_get_and_unlock (vlsh);
+
+ return rv;
+}
+
vls_handle_t
vls_epoll_create (void)
{
static void
vls_epoll_ctl_mp_checks (vcl_locked_session_t * vls, int op)
{
- if (vcl_n_workers () <= 1)
- {
- vlsl->epoll_mp_check = 1;
- return;
- }
-
- if (op == EPOLL_CTL_MOD)
+ if (vcl_n_workers () <= 1 || op == EPOLL_CTL_MOD)
return;
- vlsl->epoll_mp_check = 1;
vls_mp_checks (vls, op == EPOLL_CTL_ADD);
}
vls_mt_detect ();
vls_mt_table_rlock ();
ep_vls = vls_get_and_lock (ep_vlsh);
- vls = vls_get_and_lock (vlsh);
if (vls_mt_session_should_migrate (ep_vls))
- vls_mt_session_migrate (ep_vls);
+ {
+ ep_vls = vls_mt_session_migrate (ep_vls);
+ if (PREDICT_FALSE (!ep_vls))
+ return VPPCOM_EBADFD;
+ }
ep_sh = vls_to_sh (ep_vls);
+ vls = vls_get_and_lock (vlsh);
sh = vls_to_sh (vls);
- if (PREDICT_FALSE (!vlsl->epoll_mp_check))
- vls_epoll_ctl_mp_checks (vls, op);
-
+ vls_epoll_ctl_mp_checks (vls, op);
vls_mt_table_runlock ();
-
rv = vppcom_epoll_ctl (ep_sh, op, sh, event);
vls_mt_table_rlock ();
vls_epoll_wait (vls_handle_t ep_vlsh, struct epoll_event *events,
int maxevents, double wait_for_time)
{
- vcl_locked_session_t *vls;
+ vcl_locked_session_t *vls, *vls_tmp = NULL;
int rv;
vls_mt_detect ();
if (!(vls = vls_get_w_dlock (ep_vlsh)))
return VPPCOM_EBADFD;
- vls_mt_guard (0, VLS_MT_OP_XPOLL);
+ vls_mt_guard (vls_tmp, VLS_MT_OP_XPOLL);
rv = vppcom_epoll_wait (vls_to_sh_tu (vls), events, maxevents,
wait_for_time);
vls_mt_unguard ();
wrk = vcl_worker_get_current ();
/* *INDENT-OFF* */
- clib_bitmap_foreach (si, read_map, ({
+ clib_bitmap_foreach (si, read_map) {
s = vcl_session_get (wrk, si);
if (s->session_state == VCL_STATE_LISTEN)
{
vls = vls_get (vls_session_index_to_vlsh (si));
vls_mp_checks (vls, 1 /* is_add */);
}
- }));
+ }
/* *INDENT-ON* */
}
vcl_si_set * except_map, double wait_for_time)
{
int rv;
+ vcl_locked_session_t *vls = NULL;
vls_mt_detect ();
- vls_mt_guard (0, VLS_MT_OP_XPOLL);
+ vls_mt_guard (vls, VLS_MT_OP_XPOLL);
if (PREDICT_FALSE (!vlsl->select_mp_check))
vls_select_mp_checks (read_map);
rv = vppcom_select (n_bits, read_map, write_map, except_map, wait_for_time);
is_current = current_wrk == wrk->wrk_index;
/* *INDENT-OFF* */
- pool_foreach (s, wrk->sessions, ({
+ pool_foreach (s, wrk->sessions) {
vls = vls_get (vls_si_wi_to_vlsh (s->session_index, wrk->wrk_index));
if (vls && (is_current || vls_is_shared_by_wrk (vls, current_wrk)))
vls_unshare_session (vls, wrk);
- }));
+ }
/* *INDENT-ON* */
}
vls_incercept_sigchld ()
{
struct sigaction sa;
+ if (old_sa.sa_sigaction)
+ {
+ VDBG (0, "have intercepted sigchld");
+ return;
+ }
clib_memset (&sa, 0, sizeof (sa));
sa.sa_sigaction = vls_intercept_sigchld_handler;
sa.sa_flags = SA_SIGINFO;
vlsl->vls_mt_n_threads = 0;
vlsl->vls_wrk_index = vcl_get_worker_index ();
vlsl->select_mp_check = 0;
- vlsl->epoll_mp_check = 0;
vls_mt_locks_init ();
VDBG (0, "forked child main worker initialized");
vcl_worker_t *vcl_wrk = vcl_worker_get_current (), *dst_vcl_wrk;
vcl_session_t *s, *dst_s;
- vls = vls_session_get (wrk, msg->vls_index);
-
- if (!vls_mt_wrk_supported ())
- vls_init_share_session (wrk, vls);
+ VDBG (1, "process session clone of worker (session): %u (%u) -> %u (%u)",
+ vcl_wrk->wrk_index, msg->session_index, msg->origin_vcl_wrk,
+ msg->origin_session_index);
- s = vcl_session_get (vcl_wrk, msg->session_index);
- dst_wrk = vls_worker_get (msg->origin_vls_wrk);
+ /* VCL locked session can't been protected, so DONT touch it.
+ * VCL session may been free, check it.
+ */
dst_vcl_wrk = vcl_worker_get (msg->origin_vcl_wrk);
- dst_vls = vls_session_get (dst_wrk, msg->origin_vls_index);
- dst_vls->shared_data_index = vls->shared_data_index;
+ s = vcl_session_get (vcl_wrk, msg->session_index);
+ if (PREDICT_FALSE (!s))
+ {
+ dst_vcl_wrk->rpc_done = VLS_RPC_STATE_SESSION_NOT_EXIST;
+ return;
+ }
+
+ if (!vls_mt_wrk_supported ())
+ {
+ vls = vls_session_get (wrk, msg->vls_index);
+ vls_init_share_session (wrk, vls);
+ dst_wrk = vls_worker_get (msg->origin_vls_wrk);
+ dst_vls = vls_session_get (dst_wrk, msg->origin_vls_index);
+ dst_vls->shared_data_index = vls->shared_data_index;
+ }
dst_s = vcl_session_get (dst_vcl_wrk, msg->origin_session_index);
clib_memcpy (dst_s, s, sizeof (*s));
- dst_vcl_wrk->rpc_done = 1;
-
- VDBG (1, "proces session clone of worker (session): %u (%u) -> %u (%u)",
- vcl_wrk->wrk_index, msg->session_index, dst_vcl_wrk->wrk_index,
- msg->origin_session_index);
+ dst_vcl_wrk->rpc_done = VLS_RPC_STATE_SUCCESS;
}
static void
{
vls_sess_cleanup_msg_t *msg = (vls_sess_cleanup_msg_t *) args;
vcl_worker_t *wrk = vcl_worker_get_current ();
- vcl_worker_t *dst_wrk = vcl_worker_get (msg->origin_vcl_wrk);
-
- vppcom_session_close (vcl_session_handle_from_index (msg->session_index));
+ vls_worker_t *vls_wrk = vls_worker_get_current ();
+ vcl_session_handle_t sh = vcl_session_handle_from_index (msg->session_index);
- dst_wrk->rpc_done = 1;
+ VDBG (1, "process session cleanup of worker (session): %u (%u) from %u ()",
+ wrk->wrk_index, msg->session_index, msg->origin_vcl_wrk);
- VDBG (1, "proces session cleanup of worker (session): %u (%u) from %u ()",
- wrk->wrk_index, msg->session_index, dst_wrk->wrk_index);
+ vppcom_session_close (sh);
+ vls_sh_to_vlsh_table_del (vls_wrk, sh);
}
static void
}
void
-vls_send_clone_and_share_rpc (vcl_worker_t * wrk,
- vcl_locked_session_t * vls, u32 session_index,
- u32 vls_wrk_index, u32 dst_wrk_index,
- u32 dst_vls_index, u32 dst_session_index)
+vls_send_clone_and_share_rpc (vcl_worker_t *wrk, u32 origin_vls_index,
+ u32 session_index, u32 vls_wrk_index,
+ u32 dst_wrk_index, u32 dst_vls_index,
+ u32 dst_session_index)
{
u8 data[sizeof (u8) + sizeof (vls_clone_and_share_msg_t)];
vls_clone_and_share_msg_t *msg;
vls_rpc_msg_t *rpc;
int ret;
+ f64 timeout = clib_time_now (&wrk->clib_time) + VLS_WORKER_RPC_TIMEOUT;
rpc = (vls_rpc_msg_t *) & data;
rpc->type = VLS_RPC_CLONE_AND_SHARE;
msg = (vls_clone_and_share_msg_t *) & rpc->data;
msg->origin_vls_wrk = vls_wrk_index;
- msg->origin_vls_index = vls->vls_index;
+ msg->origin_vls_index = origin_vls_index;
msg->origin_vcl_wrk = wrk->wrk_index;
msg->origin_session_index = session_index;
msg->vls_index = dst_vls_index;
msg->session_index = dst_session_index;
- wrk->rpc_done = 0;
+ /* Try lock and handle rpcs if two threads send each other
+ * clone requests at the same time.
+ */
+ wrk->rpc_done = VLS_RPC_STATE_INIT;
+ while (!clib_spinlock_trylock (&vlsm->worker_rpc_lock))
+ vcl_flush_mq_events ();
ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
VDBG (1, "send session clone to wrk (session): %u (%u) -> %u (%u), ret=%d",
dst_wrk_index, msg->session_index, msg->origin_vcl_wrk,
msg->origin_session_index, ret);
- while (!ret && !wrk->rpc_done)
+ while (!ret && wrk->rpc_done == VLS_RPC_STATE_INIT &&
+ clib_time_now (&wrk->clib_time) < timeout)
;
+ clib_spinlock_unlock (&vlsm->worker_rpc_lock);
}
void
msg->origin_vcl_wrk = wrk->wrk_index;
msg->session_index = dst_session_index;
- wrk->rpc_done = 0;
ret = vcl_send_worker_rpc (dst_wrk_index, rpc, sizeof (data));
VDBG (1, "send session cleanup to wrk (session): %u (%u) from %u, ret=%d",
dst_wrk_index, msg->session_index, msg->origin_vcl_wrk, ret);
- while (!ret && !wrk->rpc_done)
- ;
}
int
clib_memset (vlsm, 0, sizeof (*vlsm));
clib_rwlock_init (&vlsm->vls_table_lock);
clib_rwlock_init (&vlsm->shared_data_lock);
+ clib_spinlock_init (&vlsm->worker_rpc_lock);
pool_alloc (vlsm->workers, vcm->cfg.max_workers);
pthread_atfork (vls_app_pre_fork, vls_app_fork_parent_handler,
void
vls_register_vcl_worker (void)
{
- if (vppcom_worker_register () != VPPCOM_OK)
- {
- VERR ("failed to register worker");
- return;
- }
+ vls_mt_add ();
}
/*
Code Review / vpp.git / blobdiff