while (em->state < STATE_DATA_DONE && !em->time_to_stop)
{
- svm_msg_q_lock (mq);
if (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 1))
{
- svm_msg_q_unlock (mq);
continue;
}
- svm_msg_q_sub_w_lock (mq, &msg);
+ svm_msg_q_sub_raw (mq, &msg);
rpc = svm_msg_q_msg_data (mq, &msg);
- svm_msg_q_unlock (mq);
((echo_rpc_t) rpc->fp) (em, &rpc->args);
svm_msg_q_free_msg (mq, &msg);
}
if (em->periodic_stats_delta)
echo_print_periodic_stats (em);
- svm_msg_q_lock (mq);
if (svm_msg_q_is_empty (mq) && svm_msg_q_timedwait (mq, 1))
{
- svm_msg_q_unlock (mq);
continue;
}
for (i = 0; i < svm_msg_q_size (mq); i++)
{
vec_add2 (msg_vec, msg, 1);
- svm_msg_q_sub_w_lock (mq, msg);
+ svm_msg_q_sub_raw (mq, msg);
}
- svm_msg_q_unlock (mq);
for (i = 0; i < vec_len (msg_vec); i++)
{
{
if (!use_eventfd)
{
- svm_msg_q_lock (mq);
- while (svm_msg_q_is_empty (mq))
- svm_msg_q_wait (mq);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
}
else
{
continue;
if (!svm_msg_q_is_empty (mq))
- {
- svm_msg_q_lock (mq);
- break;
- }
+ break;
}
}
}
for (i = 0; i < n_test_msgs; i++)
{
wait_for_event (mq, prod_fd, epfd, use_eventfd);
- svm_msg_q_sub_w_lock (mq, &msg);
+ svm_msg_q_sub_raw (mq, &msg);
svm_msg_q_free_msg (mq, &msg);
svm_msg_q_unlock (mq);
*counter = *counter + 1;
offset = sizeof (*ring) + ring->nitems * ring->elsize;
ring = (void *) ((u8 *) ring + offset);
}
+ clib_spinlock_init (&mq->q.lock);
}
void
svm_msg_q_free (svm_msg_q_t * mq)
{
clib_mem_free (mq->q.shr);
+ clib_spinlock_free (&mq->q.lock);
clib_mem_free (mq);
}
static void
-svm_msg_q_send_signal (svm_msg_q_t *mq)
+svm_msg_q_send_signal (svm_msg_q_t *mq, u8 is_consumer)
{
if (mq->q.evtfd == -1)
{
+ if (is_consumer)
+ {
+ int rv = pthread_mutex_lock (&mq->q.shr->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
+ {
+ rv = pthread_mutex_consistent (&mq->q.shr->mutex);
+ return;
+ }
+ }
+
(void) pthread_cond_broadcast (&mq->q.shr->condvar);
+
+ if (is_consumer)
+ pthread_mutex_unlock (&mq->q.shr->mutex);
}
else
{
svm_msg_q_lock (mq);
while (svm_msg_q_is_full (mq)
|| svm_msg_q_ring_is_full (mq, ring_index))
- svm_msg_q_wait (mq);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_FULL);
*msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
}
return 0;
msg.ring_index = ring - mq->rings;
msg.elt_index = sr->tail;
sr->tail = (sr->tail + 1) % ring->nitems;
- clib_atomic_fetch_add_rel (&sr->cursize, 1);
+ clib_atomic_fetch_add_relax (&sr->cursize, 1);
break;
}
return msg;
{
svm_msg_q_ring_shared_t *sr;
svm_msg_q_ring_t *ring;
- int need_signal;
+ u32 need_signal;
ASSERT (vec_len (mq->rings) > msg->ring_index);
ring = svm_msg_q_ring_inline (mq, msg->ring_index);
}
else
{
- clib_warning ("message out of order");
+ clib_warning ("message out of order: elt %u head %u ring %u",
+ msg->elt_index, sr->head, msg->ring_index);
/* for now, expect messages to be processed in order */
ASSERT (0);
}
- need_signal = sr->cursize == ring->nitems;
- clib_atomic_fetch_sub_rel (&sr->cursize, 1);
+ need_signal = clib_atomic_load_relax_n (&sr->cursize) == ring->nitems;
+ clib_atomic_fetch_sub_relax (&sr->cursize, 1);
if (PREDICT_FALSE (need_signal))
- svm_msg_q_send_signal (mq);
+ svm_msg_q_send_signal (mq, 1 /* is consumer */);
}
static int
sz = clib_atomic_fetch_add_rel (&sq->cursize, 1);
if (!sz)
- svm_msg_q_send_signal (mq);
+ svm_msg_q_send_signal (mq, 0 /* is consumer */);
}
int
if (nowait)
return (-2);
while (svm_msg_q_is_full (mq))
- svm_msg_q_wait (mq);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_FULL);
}
svm_msg_q_add_raw (mq, (u8 *) msg);
svm_msg_q_unlock (mq);
}
-static int
-svm_msg_q_sub_raw (svm_msg_q_t *mq, u8 *elem)
+int
+svm_msg_q_sub_raw (svm_msg_q_t *mq, svm_msg_q_msg_t *elem)
{
svm_msg_q_shared_queue_t *sq = mq->q.shr;
i8 *headp;
sq->head = (sq->head + 1) % sq->maxsize;
- sz = clib_atomic_fetch_sub_rel (&sq->cursize, 1);
+ sz = clib_atomic_fetch_sub_relax (&sq->cursize, 1);
if (PREDICT_FALSE (sz == sq->maxsize))
- svm_msg_q_send_signal (mq);
+ svm_msg_q_send_signal (mq, 1 /* is consumer */);
return 0;
}
int
-svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg,
- svm_q_conditional_wait_t cond, u32 time)
+svm_msg_q_sub_raw_batch (svm_msg_q_t *mq, svm_msg_q_msg_t *msg_buf, u32 n_msgs)
{
- int rc = 0;
+ svm_msg_q_shared_queue_t *sq = mq->q.shr;
+ u32 sz, to_deq;
+ i8 *headp;
- if (cond == SVM_Q_NOWAIT)
+ sz = svm_msg_q_size (mq);
+ ASSERT (sz);
+ to_deq = clib_min (sz, n_msgs);
+
+ headp = (i8 *) (&sq->data[0] + sq->elsize * sq->head);
+
+ if (sq->head + to_deq < sq->maxsize)
{
- /* zero on success */
- if (svm_msg_q_try_lock (mq))
- {
- return (-1);
- }
+ clib_memcpy_fast (msg_buf, headp, sq->elsize * to_deq);
+ sq->head += to_deq;
}
else
- svm_msg_q_lock (mq);
+ {
+ u32 first_batch = sq->maxsize - sq->head;
+ clib_memcpy_fast (msg_buf, headp, sq->elsize * first_batch);
+ clib_memcpy_fast (msg_buf + first_batch, sq->data,
+ sq->elsize * (to_deq - first_batch));
+ sq->head = (sq->head + to_deq) % sq->maxsize;
+ }
+
+ clib_atomic_fetch_sub_relax (&sq->cursize, to_deq);
+ if (PREDICT_FALSE (sz == sq->maxsize))
+ svm_msg_q_send_signal (mq, 1 /* is consumer */);
+
+ return to_deq;
+}
+
+int
+svm_msg_q_sub (svm_msg_q_t *mq, svm_msg_q_msg_t *msg,
+ svm_q_conditional_wait_t cond, u32 time)
+{
+ int rc = 0;
- if (PREDICT_FALSE (svm_msg_q_is_empty (mq)))
+ if (svm_msg_q_is_empty (mq))
{
if (cond == SVM_Q_NOWAIT)
{
- svm_msg_q_unlock (mq);
return (-2);
}
else if (cond == SVM_Q_TIMEDWAIT)
{
- while (svm_msg_q_is_empty (mq) && rc == 0)
- rc = svm_msg_q_timedwait (mq, time);
-
- if (rc == ETIMEDOUT)
- {
- svm_msg_q_unlock (mq);
- return ETIMEDOUT;
- }
+ if ((rc = svm_msg_q_timedwait (mq, time)))
+ return rc;
}
else
{
- while (svm_msg_q_is_empty (mq))
- svm_msg_q_wait (mq);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
}
}
- svm_msg_q_sub_raw (mq, (u8 *) msg);
-
- svm_msg_q_unlock (mq);
+ svm_msg_q_sub_raw (mq, msg);
return 0;
}
-void
-svm_msg_q_sub_w_lock (svm_msg_q_t *mq, svm_msg_q_msg_t *msg)
-{
- svm_msg_q_sub_raw (mq, (u8 *) msg);
-}
-
void
svm_msg_q_set_eventfd (svm_msg_q_t *mq, int fd)
{
return 0;
}
-void
-svm_msg_q_wait (svm_msg_q_t *mq)
+int
+svm_msg_q_wait (svm_msg_q_t *mq, svm_msg_q_wait_type_t type)
{
+ u8 (*fn) (svm_msg_q_t *);
+ int rv;
+
+ fn = (type == SVM_MQ_WAIT_EMPTY) ? svm_msg_q_is_empty : svm_msg_q_is_full;
+
if (mq->q.evtfd == -1)
{
- pthread_cond_wait (&mq->q.shr->condvar, &mq->q.shr->mutex);
+ rv = pthread_mutex_lock (&mq->q.shr->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
+ {
+ rv = pthread_mutex_consistent (&mq->q.shr->mutex);
+ return rv;
+ }
+
+ while (fn (mq))
+ pthread_cond_wait (&mq->q.shr->condvar, &mq->q.shr->mutex);
+
+ pthread_mutex_unlock (&mq->q.shr->mutex);
}
else
{
u64 buf;
- int rv;
- svm_msg_q_unlock (mq);
- while ((rv = read (mq->q.evtfd, &buf, sizeof (buf))) < 0)
+ while (fn (mq))
{
- if (errno != EAGAIN)
+ while ((rv = read (mq->q.evtfd, &buf, sizeof (buf))) < 0)
{
- clib_unix_warning ("read error");
- return;
+ if (errno != EAGAIN)
+ {
+ clib_unix_warning ("read error");
+ return rv;
+ }
}
}
- svm_msg_q_lock (mq);
}
+
+ return 0;
}
int
{
if (mq->q.evtfd == -1)
{
+ svm_msg_q_shared_queue_t *sq = mq->q.shr;
struct timespec ts;
+ u32 sz;
+ int rv;
+
+ rv = pthread_mutex_lock (&sq->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
+ {
+ rv = pthread_mutex_consistent (&sq->mutex);
+ return rv;
+ }
+
+ /* check if we're still in a signalable state after grabbing lock */
+ sz = svm_msg_q_size (mq);
+ if (sz != 0 && sz != sq->maxsize)
+ {
+ pthread_mutex_unlock (&sq->mutex);
+ return 0;
+ }
+
ts.tv_sec = unix_time_now () + (u32) timeout;
ts.tv_nsec = (timeout - (u32) timeout) * 1e9;
- return pthread_cond_timedwait (&mq->q.shr->condvar, &mq->q.shr->mutex,
- &ts);
+ rv = pthread_cond_timedwait (&sq->condvar, &sq->mutex, &ts);
+
+ pthread_mutex_unlock (&sq->mutex);
+ return rv;
}
else
{
setsockopt (mq->q.evtfd, SOL_SOCKET, SO_RCVTIMEO, (const char *) &tv,
sizeof tv);
- svm_msg_q_unlock (mq);
rv = read (mq->q.evtfd, &buf, sizeof (buf));
if (rv < 0)
clib_warning ("read %u", errno);
- svm_msg_q_lock (mq);
return rv < 0 ? errno : 0;
}
#include <vppinfra/clib.h>
#include <vppinfra/error.h>
+#include <vppinfra/lock.h>
#include <svm/queue.h>
typedef struct svm_msg_q_shr_queue_
{
svm_msg_q_shared_queue_t *shr; /**< pointer to shared queue */
int evtfd; /**< producer/consumer eventfd */
+ clib_spinlock_t lock; /**< private lock for multi-producer */
} svm_msg_q_queue_t;
typedef struct svm_msg_q_ring_shared_
} svm_msg_q_msg_t;
#define SVM_MQ_INVALID_MSG { .as_u64 = ~0 }
+
+typedef enum svm_msg_q_wait_type_
+{
+ SVM_MQ_WAIT_EMPTY,
+ SVM_MQ_WAIT_FULL
+} svm_msg_q_wait_type_t;
+
/**
* Allocate message queue
*
* Consumer dequeue one message from queue
*
* This returns the message pointing to the data in the message rings.
+ * Should only be used in single consumer scenarios as no locks are grabbed.
* The consumer is expected to call @ref svm_msg_q_free_msg once it
* finishes processing/copies the message data.
*
svm_q_conditional_wait_t cond, u32 time);
/**
- * Consumer dequeue one message from queue with mutex held
+ * Consumer dequeue one message from queue
*
- * Returns the message pointing to the data in the message rings under the
- * assumption that the message queue lock is already held. The consumer is
- * expected to call @ref svm_msg_q_free_msg once it finishes
+ * Returns the message pointing to the data in the message rings. Should only
+ * be used in single consumer scenarios as no locks are grabbed. The consumer
+ * is expected to call @ref svm_msg_q_free_msg once it finishes
* processing/copies the message data.
*
* @param mq message queue
* @param msg pointer to structure where message is to be received
* @return success status
*/
-void svm_msg_q_sub_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg);
+int svm_msg_q_sub_raw (svm_msg_q_t *mq, svm_msg_q_msg_t *elem);
+
+/**
+ * Consumer dequeue multiple messages from queue
+ *
+ * Returns the message pointing to the data in the message rings. Should only
+ * be used in single consumer scenarios as no locks are grabbed. The consumer
+ * is expected to call @ref svm_msg_q_free_msg once it finishes
+ * processing/copies the message data.
+ *
+ * @param mq message queue
+ * @param msg_buf pointer to array of messages to received
+ * @param n_msgs lengt of msg_buf array
+ * @return number of messages dequeued
+ */
+int svm_msg_q_sub_raw_batch (svm_msg_q_t *mq, svm_msg_q_msg_t *msg_buf,
+ u32 n_msgs);
/**
* Get data for message in queue
static inline int
svm_msg_q_try_lock (svm_msg_q_t * mq)
{
- int rv = pthread_mutex_trylock (&mq->q.shr->mutex);
- if (PREDICT_FALSE (rv == EOWNERDEAD))
- rv = pthread_mutex_consistent (&mq->q.shr->mutex);
- return rv;
+ if (mq->q.evtfd == -1)
+ {
+ int rv = pthread_mutex_trylock (&mq->q.shr->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
+ rv = pthread_mutex_consistent (&mq->q.shr->mutex);
+ return rv;
+ }
+ else
+ {
+ return !clib_spinlock_trylock (&mq->q.lock);
+ }
}
/**
static inline int
svm_msg_q_lock (svm_msg_q_t * mq)
{
- int rv = pthread_mutex_lock (&mq->q.shr->mutex);
- if (PREDICT_FALSE (rv == EOWNERDEAD))
- rv = pthread_mutex_consistent (&mq->q.shr->mutex);
- return rv;
+ if (mq->q.evtfd == -1)
+ {
+ int rv = pthread_mutex_lock (&mq->q.shr->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
+ rv = pthread_mutex_consistent (&mq->q.shr->mutex);
+ return rv;
+ }
+ else
+ {
+ clib_spinlock_lock (&mq->q.lock);
+ return 0;
+ }
}
/**
static inline void
svm_msg_q_unlock (svm_msg_q_t * mq)
{
- pthread_mutex_unlock (&mq->q.shr->mutex);
+ if (mq->q.evtfd == -1)
+ {
+ pthread_mutex_unlock (&mq->q.shr->mutex);
+ }
+ else
+ {
+ clib_spinlock_unlock (&mq->q.lock);
+ }
}
/**
* Must be called with mutex held. The queue only works non-blocking
* with eventfds, so handle blocking calls as an exception here.
*/
-void svm_msg_q_wait (svm_msg_q_t *mq);
+int svm_msg_q_wait (svm_msg_q_t *mq, svm_msg_q_wait_type_t type);
/**
* Timed wait for message queue event
static inline int
vcl_mq_dequeue_batch (vcl_worker_t * wrk, svm_msg_q_t * mq, u32 n_max_msg)
{
- svm_msg_q_msg_t *msg;
- u32 n_msgs;
- int i;
+ u32 n_msgs = 0, sz, len;
- n_msgs = clib_min (svm_msg_q_size (mq), n_max_msg);
- for (i = 0; i < n_msgs; i++)
+ while ((sz = svm_msg_q_size (mq)))
{
- vec_add2 (wrk->mq_msg_vector, msg, 1);
- svm_msg_q_sub_w_lock (mq, msg);
+ len = vec_len (wrk->mq_msg_vector);
+ vec_validate (wrk->mq_msg_vector, len + sz - 1);
+ svm_msg_q_sub_raw_batch (mq, wrk->mq_msg_vector + len, sz);
+ n_msgs += sz;
}
return n_msgs;
}
}
void
-vcl_flush_mq_events (void)
+vcl_worker_flush_mq_events (vcl_worker_t *wrk)
{
- vcl_worker_t *wrk = vcl_worker_get_current ();
svm_msg_q_msg_t *msg;
session_event_t *e;
svm_msg_q_t *mq;
int i;
mq = wrk->app_event_queue;
- svm_msg_q_lock (mq);
vcl_mq_dequeue_batch (wrk, mq, ~0);
- svm_msg_q_unlock (mq);
for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
{
vcl_handle_pending_wrk_updates (wrk);
}
+void
+vcl_flush_mq_events (void)
+{
+ vcl_worker_flush_mq_events (vcl_worker_get_current ());
+}
+
static int
vppcom_session_unbind (u32 session_handle)
{
vcl_session_t *listen_session = 0;
vcl_session_t *client_session = 0;
vcl_session_msg_t *evt;
- svm_msg_q_msg_t msg;
- session_event_t *e;
u8 is_nonblocking;
int rv;
+again:
+
listen_session = vcl_session_get_w_handle (wrk, listen_session_handle);
if (!listen_session)
return VPPCOM_EBADFD;
if (svm_msg_q_is_empty (wrk->app_event_queue) && is_nonblocking)
return VPPCOM_EAGAIN;
- if (svm_msg_q_sub (wrk->app_event_queue, &msg, SVM_Q_WAIT, 0))
- return VPPCOM_EAGAIN;
-
- e = svm_msg_q_msg_data (wrk->app_event_queue, &msg);
- if (e->event_type != SESSION_CTRL_EVT_ACCEPTED)
- {
- vcl_handle_mq_event (wrk, e);
- svm_msg_q_free_msg (wrk->app_event_queue, &msg);
- continue;
- }
- clib_memcpy_fast (&accepted_msg, e->data, sizeof (accepted_msg));
- svm_msg_q_free_msg (wrk->app_event_queue, &msg);
- break;
+ svm_msg_q_wait (wrk->app_event_queue, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_flush_mq_events (wrk);
+ goto again;
}
handle:
return rv;
}
-static u8
-vcl_is_rx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
-{
- return (e->event_type == SESSION_IO_EVT_RX && e->session_index == sid);
-}
-
static inline int
vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
u8 peek)
int rv, n_read = 0, is_nonblocking;
vcl_session_t *s = 0;
svm_fifo_t *rx_fifo;
- svm_msg_q_msg_t msg;
session_event_t *e;
svm_msg_q_t *mq;
u8 is_ct;
if (is_ct)
svm_fifo_unset_event (s->rx_fifo);
svm_fifo_unset_event (rx_fifo);
- svm_msg_q_lock (mq);
- if (svm_msg_q_is_empty (mq))
- svm_msg_q_wait (mq);
-
- svm_msg_q_sub_w_lock (mq, &msg);
- e = svm_msg_q_msg_data (mq, &msg);
- svm_msg_q_unlock (mq);
- if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
- vcl_handle_mq_event (wrk, e);
- svm_msg_q_free_msg (mq, &msg);
+
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ 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_msg_t msg;
- session_event_t *e;
svm_msg_q_t *mq;
u8 is_ct;
if (is_ct)
svm_fifo_unset_event (s->rx_fifo);
svm_fifo_unset_event (rx_fifo);
- svm_msg_q_lock (mq);
- if (svm_msg_q_is_empty (mq))
- svm_msg_q_wait (mq);
-
- svm_msg_q_sub_w_lock (mq, &msg);
- e = svm_msg_q_msg_data (mq, &msg);
- svm_msg_q_unlock (mq);
- if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
- vcl_handle_mq_event (wrk, e);
- svm_msg_q_free_msg (mq, &msg);
+
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_flush_mq_events (wrk);
}
}
s->rx_bytes_pending -= n_bytes;
}
-static u8
-vcl_is_tx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
-{
- return (e->event_type == SESSION_IO_EVT_TX && e->session_index == sid);
-}
-
always_inline u8
vcl_fifo_is_writeable (svm_fifo_t * f, u32 len, u8 is_dgram)
{
{
int n_write, is_nonblocking;
session_evt_type_t et;
- svm_msg_q_msg_t msg;
svm_fifo_t *tx_fifo;
- session_event_t *e;
svm_msg_q_t *mq;
u8 is_ct;
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);
- svm_msg_q_lock (mq);
- if (svm_msg_q_is_empty (mq))
- svm_msg_q_wait (mq);
- svm_msg_q_sub_w_lock (mq, &msg);
- e = svm_msg_q_msg_data (mq, &msg);
- svm_msg_q_unlock (mq);
-
- if (!vcl_is_tx_evt_for_session (e, s->session_index, is_ct))
- vcl_handle_mq_event (wrk, e);
- svm_msg_q_free_msg (mq, &msg);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
+ vcl_worker_flush_mq_events (wrk);
}
}
session_event_t *e;
u32 i;
- svm_msg_q_lock (mq);
if (svm_msg_q_is_empty (mq))
{
if (*bits_set)
- {
- svm_msg_q_unlock (mq);
- return 0;
- }
+ return 0;
if (!time_to_wait)
- {
- svm_msg_q_unlock (mq);
- return 0;
- }
+ return 0;
else if (time_to_wait < 0)
- {
- svm_msg_q_wait (mq);
- }
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
else
{
if (svm_msg_q_timedwait (mq, time_to_wait))
- {
- svm_msg_q_unlock (mq);
- return 0;
- }
+ return 0;
}
}
vcl_mq_dequeue_batch (wrk, mq, ~0);
- svm_msg_q_unlock (mq);
for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
{
if (vec_len (wrk->mq_msg_vector) && svm_msg_q_is_empty (mq))
goto handle_dequeued;
- svm_msg_q_lock (mq);
if (svm_msg_q_is_empty (mq))
{
if (!wait_for_time)
- {
- svm_msg_q_unlock (mq);
- return 0;
- }
+ return 0;
else if (wait_for_time < 0)
- {
- svm_msg_q_wait (mq);
- }
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_EMPTY);
else
{
if (svm_msg_q_timedwait (mq, wait_for_time / 1e3))
- {
- svm_msg_q_unlock (mq);
- return 0;
- }
+ return 0;
}
}
ASSERT (maxevents > *num_ev);
vcl_mq_dequeue_batch (wrk, mq, ~0);
- svm_msg_q_unlock (mq);
handle_dequeued:
for (i = 0; i < vec_len (wrk->mq_msg_vector); i++)
svm_msg_q_lock (mq);
while (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)
|| svm_msg_q_is_full (mq))
- svm_msg_q_wait (mq);
+ svm_msg_q_wait (mq, SVM_MQ_WAIT_FULL);
msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
evt = (session_event_t *) svm_msg_q_msg_data (mq, &msg);
evt->session_index = session_index;
* XXX: we may need priorities here */
mq = wrk->vpp_event_queue;
n_to_dequeue = svm_msg_q_size (mq);
- if (n_to_dequeue && svm_msg_q_try_lock (mq) == 0)
+ if (n_to_dequeue)
{
for (i = 0; i < n_to_dequeue; i++)
{
- svm_msg_q_sub_w_lock (mq, msg);
+ svm_msg_q_sub_raw (mq, msg);
evt = svm_msg_q_msg_data (mq, msg);
session_evt_add_to_list (wrk, evt);
svm_msg_q_free_msg (mq, msg);
}
- svm_msg_q_unlock (mq);
}
SESSION_EVT (SESSION_EVT_DSP_CNTRS, MQ_DEQ, wrk, n_to_dequeue, !i);
Code Review / vpp.git / commitdiff