aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/svm/message_queue.c37
-rw-r--r--src/svm/message_queue.h14
-rw-r--r--src/vnet/session/application_interface.h9
-rw-r--r--src/vnet/session/session.c3
4 files changed, 51 insertions, 12 deletions
diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c
index a6af7962f73..2880645b427 100644
--- a/src/svm/message_queue.c
+++ b/src/svm/message_queue.c
@@ -243,8 +243,7 @@ svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index,
{
if (svm_msg_q_try_lock (mq))
return -1;
- if (PREDICT_FALSE (svm_msg_q_is_full (mq)
- || svm_msg_q_ring_is_full (mq, ring_index)))
+ if (PREDICT_FALSE (svm_msg_q_or_ring_is_full (mq, ring_index)))
{
svm_msg_q_unlock (mq);
return -2;
@@ -254,9 +253,8 @@ svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index,
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_prod (mq);
+ while (svm_msg_q_or_ring_is_full (mq, ring_index))
+ svm_msg_q_or_ring_wait_prod (mq, ring_index);
*msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
}
return 0;
@@ -569,6 +567,35 @@ svm_msg_q_wait_prod (svm_msg_q_t *mq)
}
int
+svm_msg_q_or_ring_wait_prod (svm_msg_q_t *mq, u32 ring_index)
+{
+ if (mq->q.evtfd == -1)
+ {
+ while (svm_msg_q_or_ring_is_full (mq, ring_index))
+ pthread_cond_wait (&mq->q.shr->condvar, &mq->q.shr->mutex);
+ }
+ else
+ {
+ u64 buf;
+ int rv;
+
+ while (svm_msg_q_or_ring_is_full (mq, ring_index))
+ {
+ while ((rv = read (mq->q.evtfd, &buf, sizeof (buf))) < 0)
+ {
+ if (errno != EAGAIN)
+ {
+ clib_unix_warning ("read error");
+ return rv;
+ }
+ }
+ }
+ }
+
+ return 0;
+}
+
+int
svm_msg_q_timedwait (svm_msg_q_t *mq, double timeout)
{
if (mq->q.evtfd == -1)
diff --git a/src/svm/message_queue.h b/src/svm/message_queue.h
index bd76eda5d88..0780cca1c32 100644
--- a/src/svm/message_queue.h
+++ b/src/svm/message_queue.h
@@ -328,6 +328,12 @@ svm_msg_q_ring_is_full (svm_msg_q_t * mq, u32 ring_index)
return (clib_atomic_load_relax_n (&ring->shr->cursize) >= ring->nitems);
}
+static inline u8
+svm_msg_q_or_ring_is_full (svm_msg_q_t *mq, u32 ring_index)
+{
+ return (svm_msg_q_is_full (mq) || svm_msg_q_ring_is_full (mq, ring_index));
+}
+
/**
* Check if message queue is empty
*/
@@ -418,6 +424,14 @@ int svm_msg_q_wait (svm_msg_q_t *mq, svm_msg_q_wait_type_t type);
int svm_msg_q_wait_prod (svm_msg_q_t *mq);
/**
+ * Wait for message queue or ring event as producer
+ *
+ * Similar to @ref svm_msg_q_wait but lock (mutex or spinlock) must
+ * be held. Should only be called by producers.
+ */
+int svm_msg_q_or_ring_wait_prod (svm_msg_q_t *mq, u32 ring_index);
+
+/**
* Timed wait for message queue event
*
* Must be called with mutex held.
diff --git a/src/vnet/session/application_interface.h b/src/vnet/session/application_interface.h
index 733a4627c2e..b1dab322dd7 100644
--- a/src/vnet/session/application_interface.h
+++ b/src/vnet/session/application_interface.h
@@ -613,8 +613,8 @@ app_send_io_evt_to_vpp (svm_msg_q_t * mq, u32 session_index, u8 evt_type,
{
if (svm_msg_q_try_lock (mq))
return -1;
- if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)
- || svm_msg_q_is_full (mq)))
+ if (PREDICT_FALSE (
+ svm_msg_q_or_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
{
svm_msg_q_unlock (mq);
return -2;
@@ -629,9 +629,8 @@ app_send_io_evt_to_vpp (svm_msg_q_t * mq, u32 session_index, u8 evt_type,
else
{
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_prod (mq);
+ while (svm_msg_q_or_ring_is_full (mq, SESSION_MQ_IO_EVT_RING))
+ svm_msg_q_or_ring_wait_prod (mq, SESSION_MQ_IO_EVT_RING);
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;
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 256760079ee..d791c463f90 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -36,8 +36,7 @@ session_send_evt_to_thread (void *data, void *args, u32 thread_index,
mq = wrk->vpp_event_queue;
if (PREDICT_FALSE (svm_msg_q_lock (mq)))
return -1;
- if (PREDICT_FALSE (svm_msg_q_is_full (mq)
- || svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
+ if (PREDICT_FALSE (svm_msg_q_or_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
{
svm_msg_q_unlock (mq);
return -2;