u32 n_bytes, uword * si_bits, uword * libc_bits)
{
uword si_bits_set, libc_bits_set;
+ u32 session_index, wrk_index;
vls_handle_t vlsh;
int fd;
vlsh = ldp_fd_to_vlsh (fd);
if (vlsh == VLS_INVALID_HANDLE)
clib_bitmap_set_no_check (*libcb, fd, 1);
- else if (vlsh_to_worker_index (vlsh) != vppcom_worker_index ())
- clib_warning ("migration currently not supported");
else
- *vclb = clib_bitmap_set (*vclb, vlsh_to_session_index (vlsh), 1);
+ {
+ vlsh_to_session_and_worker_index (vlsh, &session_index, &wrk_index);
+ if (wrk_index != vppcom_worker_index ())
+ clib_warning ("migration currently not supported");
+ else
+ *vclb = clib_bitmap_set (*vclb, session_index, 1);
+ }
}
si_bits_set = clib_bitmap_last_set (*vclb) + 1;
clib_bitmap_foreach (si, vclb) {
vlsh = vls_session_index_to_vlsh (si);
- ASSERT (vlsh != VLS_INVALID_HANDLE);
+ if (vlsh == VLS_INVALID_HANDLE)
+ continue;
fd = ldp_vlsh_to_fd (vlsh);
if (PREDICT_FALSE (fd < 0))
{
clib_bitmap_t *listeners; /**< bitmap of wrks actively listening */
} vls_shared_data_t;
+#define foreach_vls_flag _ (APP_CLOSED, "app closed")
+
+enum vls_flags_bits_
+{
+#define _(sym, str) VLS_FLAG_BIT_##sym,
+ foreach_vls_flag
+#undef _
+};
+
+typedef enum vls_flags_
+{
+#define _(sym, str) VLS_F_##sym = 1 << VLS_FLAG_BIT_##sym,
+ foreach_vls_flag
+#undef _
+} vls_flags_t;
+
typedef struct vcl_locked_session_
{
clib_spinlock_t lock; /**< vls lock when in use */
u32 owner_vcl_wrk_index; /**< vcl wrk of the vls wrk at alloc */
uword *vcl_wrk_index_to_session_index; /**< map vcl wrk to session */
int libc_epfd; /**< epoll fd for libc epoll */
+ vls_flags_t flags; /**< vls flags */
} vcl_locked_session_t;
typedef struct vls_worker_
vcl_set_worker_index (vlsl->vls_wrk_index);
/* Only allow new pthread to be cancled in vls_mt_mq_lock */
- int old_state;
if (vlsl->vls_mt_n_threads >= 2)
- pthread_setcancelstate (PTHREAD_CANCEL_DISABLE, &old_state);
+ pthread_setcancelstate (PTHREAD_CANCEL_DISABLE, NULL);
}
static inline void
vls_mt_mq_lock (void)
{
- /* Allow controlled cancelation of thread before grabbing mutex */
- pthread_testcancel ();
pthread_mutex_lock (&vlsl->vls_mt_mq_mlock);
}
+static inline int
+vls_mt_mq_trylock (void)
+{
+ return pthread_mutex_trylock (&vlsl->vls_mt_mq_mlock);
+}
+
static inline void
vls_mt_mq_unlock (void)
{
clib_spinlock_lock (&vls->lock);
}
+static inline int
+vls_trylock (vcl_locked_session_t *vls)
+{
+ if ((vlsl->vls_mt_n_threads > 1) || vls_is_shared (vls))
+ return !clib_spinlock_trylock (&vls->lock);
+ return 0;
+}
+
static inline void
vls_unlock (vcl_locked_session_t * vls)
{
return rv;
}
-vcl_session_handle_t
-vlsh_to_session_index (vls_handle_t vlsh)
-{
- vcl_session_handle_t sh;
- sh = vlsh_to_sh (vlsh);
- return vppcom_session_index (sh);
-}
-
-int
-vlsh_to_worker_index (vls_handle_t vlsh)
+void
+vlsh_to_session_and_worker_index (vls_handle_t vlsh, u32 *session_index,
+ u32 *wrk_index)
{
vcl_locked_session_t *vls;
- u32 wrk_index;
- vls = vls_get_w_dlock (vlsh);
+ vls_mt_pool_rlock ();
+
+ /* Do not lock vls because for mt apps that use select this could
+ * deadlock if multiple threads select on the same vlsh */
+ vls = vls_get (vlsh);
+
if (!vls)
- wrk_index = INVALID_SESSION_ID;
+ {
+ *session_index = INVALID_SESSION_ID;
+ *wrk_index = INVALID_SESSION_ID;
+ }
else
- wrk_index = vls->vcl_wrk_index;
- vls_dunlock (vls);
+ {
+ *session_index = vls->session_index;
+ *wrk_index = vls->vcl_wrk_index;
+ }
- return wrk_index;
+ vls_mt_pool_runlock ();
}
vls_handle_t
vls_share_sessions (vls_parent_wrk, vls_wrk);
}
+static inline u8
+vcl_session_is_write_nonblk (vcl_session_t *s)
+{
+ int rv = vcl_session_write_ready (s);
+
+ if (!s->is_dgram)
+ return rv != 0;
+
+ /* Probably not common, but without knowing the actual write size, this is
+ * the only way we can guarantee the write won't block */
+ return rv < 0 ? 1 : (rv > (64 << 10));
+}
+
static void
vls_mt_acq_locks (vcl_locked_session_t * vls, vls_mt_ops_t op, int *locks_acq)
{
switch (op)
{
case VLS_MT_OP_READ:
- if (!is_nonblk)
- is_nonblk = vcl_session_read_ready (s) != 0;
- if (!is_nonblk)
+ is_nonblk = is_nonblk ?: vcl_session_read_ready (s) != 0;
+ while (!is_nonblk && vls_mt_mq_trylock ())
{
- vls_mt_mq_lock ();
- *locks_acq |= VLS_MT_LOCK_MQ;
+ /* might get data while waiting for lock */
+ is_nonblk = vcl_session_read_ready (s) != 0;
}
+ if (!is_nonblk)
+ *locks_acq |= VLS_MT_LOCK_MQ;
break;
case VLS_MT_OP_WRITE:
ASSERT (s);
- if (!is_nonblk)
- is_nonblk = vcl_session_write_ready (s) != 0;
- if (!is_nonblk)
+ is_nonblk = is_nonblk ?: vcl_session_is_write_nonblk (s);
+ while (!is_nonblk && vls_mt_mq_trylock ())
{
- vls_mt_mq_lock ();
- *locks_acq |= VLS_MT_LOCK_MQ;
+ /* might get space while waiting for lock */
+ is_nonblk = vcl_session_is_write_nonblk (s);
}
+ if (!is_nonblk)
+ *locks_acq |= VLS_MT_LOCK_MQ;
break;
case VLS_MT_OP_XPOLL:
vls_mt_mq_lock ();
int rv;
vls_mt_detect ();
- vls_mt_pool_wlock ();
- vls = vls_get_and_lock (vlsh);
+ /* Notify vcl while holding a reader lock. Allows other threads to
+ * regrab vls and unlock it if needed. */
+ vls_mt_pool_rlock ();
+
+ vls = vls_get (vlsh);
if (!vls)
{
- vls_mt_pool_wunlock ();
+ vls_mt_pool_runlock ();
return VPPCOM_EBADFD;
}
+ /* Notify other threads, if any, that app closed. Do it before
+ * grabbing lock as vls might be already locked */
+ vls->flags |= VLS_F_APP_CLOSED;
+ vls_lock (vls);
+
vls_mt_guard (vls, VLS_MT_OP_SPOOL);
if (vls_is_shared (vls))
if (vls_mt_wrk_supported ())
vls_mt_session_cleanup (vls);
- vls_free (vls);
vls_mt_unguard ();
+ vls_unlock (vls);
+ vls_mt_pool_runlock ();
+
+ /* Drop mt reader lock on pool and acquire writer lock */
+ vls_mt_pool_wlock ();
+
+ vls = vls_get (vlsh);
+
+ /* Other threads might be still using the session */
+ while (vls_trylock (vls))
+ {
+ vls_mt_pool_wunlock ();
+ vls_mt_pool_wlock ();
+ }
+
+ vls_free (vls);
vls_mt_pool_wunlock ();
dst_wrk_index, msg->session_index, msg->origin_vcl_wrk, ret);
}
+static inline void
+vls_mt_mq_wait_lock (vcl_session_handle_t vcl_sh)
+{
+ vcl_locked_session_t *vls;
+ vls_worker_t *wrk;
+ uword *vlshp;
+
+ /* If mt wrk supported or single threaded just return */
+ if (vls_mt_wrk_supported () || (vlsl->vls_mt_n_threads <= 1))
+ return;
+
+ wrk = vls_worker_get_current ();
+ /* Expect current thread to have dropped lock before calling vcl */
+ vls_mt_pool_rlock ();
+
+ vlshp = vls_sh_to_vlsh_table_get (wrk, vcl_sh);
+ if (vlshp)
+ {
+ vls = vls_get (*vlshp);
+ /* Handle case here other threads might've closed the session */
+ if (vls->flags & VLS_F_APP_CLOSED)
+ {
+ vcl_session_t *s;
+ s = vcl_session_get_w_handle (vcl_worker_get_current (), vcl_sh);
+ s->flags |= VCL_SESSION_F_APP_CLOSING;
+ vls_mt_pool_runlock ();
+ return;
+ }
+ }
+
+ vls_mt_mq_unlock ();
+}
+
+static inline void
+vls_mt_mq_wait_unlock (vcl_session_handle_t vcl_sh)
+{
+ if (vls_mt_wrk_supported () || (vlsl->vls_mt_n_threads <= 1))
+ return;
+
+ vls_mt_mq_lock ();
+
+ /* writers can grab lock now */
+ vls_mt_pool_runlock ();
+}
+
int
vls_app_create (char *app_name)
{
clib_rwlock_init (&vlsl->vls_pool_lock);
vls_mt_locks_init ();
vcm->wrk_rpc_fn = vls_rpc_handler;
+ /* For multi threaded apps where sessions are implicitly shared, ask vcl
+ * to use these callbacks prior and after blocking on io operations */
+ vcl_worker_set_wait_mq_fns (vls_mt_mq_wait_lock, vls_mt_mq_wait_unlock);
return VPPCOM_OK;
}
vcl_si_set * except_map, double wait_for_time);
int vls_poll (vcl_poll_t *vp, uint32_t n_sids, double wait_for_time);
vcl_session_handle_t vlsh_to_sh (vls_handle_t vlsh);
-vcl_session_handle_t vlsh_to_session_index (vls_handle_t vlsh);
-int vlsh_to_worker_index (vls_handle_t vlsh);
+void vlsh_to_session_and_worker_index (vls_handle_t vlsh,
+ uint32_t *session_index,
+ uint32_t *wrk_index);
vls_handle_t vls_session_index_to_vlsh (uint32_t session_index);
int vls_app_create (char *app_name);
unsigned char vls_use_eventfd (void);
mqc->mq_fd = mq_fd;
mqc->mq = mq;
+ fcntl (mq_fd, F_SETFL, O_NONBLOCK);
+
e.events = EPOLLIN;
e.data.u32 = mqc_index;
if (epoll_ctl (wrk->mqs_epfd, EPOLL_CTL_ADD, mq_fd, &e) < 0)
hash_free (seg_indices_map);
}
+void
+vcl_worker_set_wait_mq_fns (vcl_worker_wait_mq_fn pre_wait,
+ vcl_worker_wait_mq_fn post_wait)
+{
+ vcl_worker_t *wrk = vcl_worker_get_current ();
+ wrk->pre_wait_fn = pre_wait;
+ wrk->post_wait_fn = post_wait;
+}
+
vcl_worker_t *
vcl_worker_alloc_and_init ()
{
max_deq = svm_fifo_max_dequeue_cons (s->rx_fifo);
if (max_deq <= SESSION_CONN_HDR_LEN)
return 0;
- if (svm_fifo_peek (s->rx_fifo, 0, sizeof (ph), (u8 *) & ph) < 0)
+ if (svm_fifo_peek (s->rx_fifo, 0, sizeof (ph), (u8 *) &ph) < 0)
return 0;
if (ph.data_length + SESSION_CONN_HDR_LEN > max_deq)
return 0;
}
}
+/**
+ * Used as alternative to vcl_session_read_ready to avoid peeking udp sessions.
+ * Multi-threaded applications could select the same session from multiple
+ * threads */
+int
+vcl_session_read_ready2 (vcl_session_t *s)
+{
+ if (vcl_session_is_open (s))
+ {
+ if (vcl_session_is_ct (s))
+ return svm_fifo_max_dequeue_cons (s->ct_rx_fifo);
+
+ if (s->is_dgram)
+ {
+ if (svm_fifo_max_dequeue_cons (s->rx_fifo) <= SESSION_CONN_HDR_LEN)
+ return 0;
+
+ /* Return 1 even if not yet sure if a full datagram was received */
+ return 1;
+ }
+
+ return svm_fifo_max_dequeue_cons (s->rx_fifo);
+ }
+ else if (s->session_state == VCL_STATE_LISTEN)
+ {
+ return clib_fifo_elts (s->accept_evts_fifo);
+ }
+ else
+ {
+ return 1;
+ }
+}
+
int
vcl_session_write_ready (vcl_session_t * s)
{
VCL_SESSION_F_PENDING_DISCONNECT = 1 << 6,
VCL_SESSION_F_PENDING_FREE = 1 << 7,
VCL_SESSION_F_PENDING_LISTEN = 1 << 8,
+ VCL_SESSION_F_APP_CLOSING = 1 << 9,
} __clib_packed vcl_session_flags_t;
+typedef enum vcl_worker_wait_
+{
+ VCL_WRK_WAIT_CTRL,
+ VCL_WRK_WAIT_IO_RX,
+ VCL_WRK_WAIT_IO_TX,
+} vcl_worker_wait_type_t;
+
typedef struct vcl_session_
{
CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
int mq_fd;
} vcl_mq_evt_conn_t;
+typedef void (*vcl_worker_wait_mq_fn) (u32 vcl_sh);
typedef struct vcl_worker_
{
CLIB_CACHE_LINE_ALIGN_MARK (cacheline0);
/** vcl needs next epoll_create to go to libc_epoll */
u8 vcl_needs_real_epoll;
volatile int rpc_done;
+
+ /* functions to be called pre/post wait if vcl managed by vls */
+ vcl_worker_wait_mq_fn pre_wait_fn;
+ vcl_worker_wait_mq_fn post_wait_fn;
} vcl_worker_t;
STATIC_ASSERT (sizeof (session_disconnected_msg_t) <= 16,
void vcl_segment_table_del (u64 segment_handle);
int vcl_session_read_ready (vcl_session_t * session);
+int vcl_session_read_ready2 (vcl_session_t *s);
int vcl_session_write_ready (vcl_session_t * session);
int vcl_session_alloc_ext_cfg (vcl_session_t *s,
transport_endpt_ext_cfg_type_t type, u32 len);
int vcl_session_share_fifos (vcl_session_t *s, svm_fifo_t *rxf,
svm_fifo_t *txf);
void vcl_worker_detach_sessions (vcl_worker_t *wrk);
+void vcl_worker_set_wait_mq_fns (vcl_worker_wait_mq_fn pre_wait,
+ vcl_worker_wait_mq_fn post_wait);
/*
* VCL Binary API
vcl_worker_flush_mq_events (vcl_worker_get_current ());
}
+static inline void
+vcl_worker_wait_mq (vcl_worker_t *wrk, u32 session_handle,
+ vcl_worker_wait_type_t wait)
+{
+ vcl_session_t *s = 0;
+ u32 sleeps = 0;
+
+ if (wrk->pre_wait_fn)
+ wrk->pre_wait_fn (session_handle);
+
+ if (session_handle != VCL_INVALID_SESSION_INDEX)
+ {
+ s = vcl_session_get_w_handle (wrk, session_handle);
+ /* Session might've been closed by another thread if multi-threaded
+ * as opposed to multi-worker app */
+ if (s->flags & VCL_SESSION_F_APP_CLOSING)
+ return;
+ }
+
+ /* Short sleeps waiting on mq notifications. Note that we drop mq lock for
+ * multi-thread apps so we may be getting spurious notifications. Not ideal,
+ * as we'd like to only be woken up on events, but since multiple threads may
+ * be waiting on the same mq, events may be missed. If waiting on session io
+ * events, return if session is ready. Otherwise return if mq has event */
+ while (svm_msg_q_timedwait (wrk->app_event_queue, 1e-3))
+ {
+ if (s)
+ {
+ if ((wait == VCL_WRK_WAIT_IO_RX && vcl_session_read_ready (s)) ||
+ (wait == VCL_WRK_WAIT_IO_TX && vcl_session_write_ready (s)))
+ break;
+ }
+ else
+ {
+ if (!svm_msg_q_is_empty (wrk->app_event_queue))
+ break;
+ }
+ if (++sleeps > 200)
+ break;
+ }
+
+ if (wrk->post_wait_fn)
+ wrk->post_wait_fn (session_handle);
+}
+
static int
vppcom_session_unbind (u32 session_handle)
{
if (svm_msg_q_is_empty (wrk->app_event_queue) && is_nonblocking)
return VPPCOM_EAGAIN;
- svm_msg_q_wait (wrk->app_event_queue, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_wait_mq (wrk, -1, VCL_WRK_WAIT_CTRL);
vcl_worker_flush_mq_events (wrk);
goto again;
}
vcl_session_t *s = 0;
svm_fifo_t *rx_fifo;
session_event_t *e;
- svm_msg_q_t *mq;
u8 is_ct;
if (PREDICT_FALSE (!buf))
is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
is_ct = vcl_session_is_ct (s);
- mq = wrk->app_event_queue;
rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
{
if (vcl_session_is_closing (s))
return vcl_session_closing_error (s);
+ if (s->flags & VCL_SESSION_F_APP_CLOSING)
+ return vcl_session_closed_error (s);
if (is_ct)
svm_fifo_unset_event (s->rx_fifo);
svm_fifo_unset_event (rx_fifo);
- svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_wait_mq (wrk, session_handle, VCL_WRK_WAIT_IO_RX);
vcl_worker_flush_mq_events (wrk);
}
}
int n_read = 0, is_nonblocking;
vcl_session_t *s = 0;
svm_fifo_t *rx_fifo;
- svm_msg_q_t *mq;
u8 is_ct;
s = vcl_session_get_w_handle (wrk, session_handle);
is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
is_ct = vcl_session_is_ct (s);
- mq = wrk->app_event_queue;
rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
{
if (vcl_session_is_closing (s))
return vcl_session_closing_error (s);
+ if (s->flags & VCL_SESSION_F_APP_CLOSING)
+ return vcl_session_closed_error (s);
if (is_ct)
svm_fifo_unset_event (s->rx_fifo);
svm_fifo_unset_event (rx_fifo);
- svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_wait_mq (wrk, session_handle, VCL_WRK_WAIT_IO_RX);
vcl_worker_flush_mq_events (wrk);
}
}
int n_write, is_nonblocking;
session_evt_type_t et;
svm_fifo_t *tx_fifo;
- svm_msg_q_t *mq;
u8 is_ct;
/* Accept zero length writes but just return */
tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
- mq = wrk->app_event_queue;
if (!vcl_fifo_is_writeable (tx_fifo, n, is_dgram))
{
if (is_nonblocking)
svm_fifo_add_want_deq_ntf (tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
if (vcl_session_is_closing (s))
return vcl_session_closing_error (s);
+ if (s->flags & VCL_SESSION_F_APP_CLOSING)
+ return vcl_session_closed_error (s);
- svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_wait_mq (wrk, vcl_session_handle (s), VCL_WRK_WAIT_IO_TX);
vcl_worker_flush_mq_events (wrk);
}
}
int n_write = 0, n_bytes = 0, is_nonblocking;
vcl_session_t *s = 0;
svm_fifo_t *tx_fifo;
- svm_msg_q_t *mq;
u8 is_ct;
u32 i;
is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
is_ct = vcl_session_is_ct (s);
- mq = wrk->app_event_queue;
tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
for (i = 0; i < n_segments; i++)
svm_fifo_add_want_deq_ntf (tx_fifo, SVM_FIFO_WANT_DEQ_NOTIF);
if (vcl_session_is_closing (s))
return vcl_session_closing_error (s);
+ if (s->flags & VCL_SESSION_F_APP_CLOSING)
+ return vcl_session_closed_error (s);
- svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_wait_mq (wrk, session_handle, VCL_WRK_WAIT_IO_TX);
vcl_worker_flush_mq_events (wrk);
}
}
continue;
}
- if (vcl_session_read_ready (s))
+ if (vcl_session_read_ready2 (s))
{
clib_bitmap_set_no_check ((uword *) read_map, sid, 1);
bits_set++;
import signal
import glob
from config import config
-from asfframework import VppAsfTestCase, VppTestRunner, Worker, tag_fixme_ubuntu2404
+from asfframework import VppAsfTestCase, VppTestRunner, Worker
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
iperf3 = "/usr/bin/iperf3"
self.assert_equal(worker_client.result, 0, "Binary test return code")
-@tag_fixme_ubuntu2404
class LDPCutThruTestCase(VCLTestCase):
"""LDP Cut Thru Tests"""
)
-@tag_fixme_ubuntu2404
class LDPThruHostStackIperf(VCLTestCase):
"""LDP Thru Host Stack Iperf"""
)
-@tag_fixme_ubuntu2404
class LDPThruHostStackIperfUdp(VCLTestCase):
"""LDP Thru Host Stack Iperf UDP"""
)
-@tag_fixme_ubuntu2404
class LDPIpv6CutThruTestCase(VCLTestCase):
"""LDP IPv6 Cut Thru Tests"""
Code Review / vpp.git / commitdiff