summaryrefslogtreecommitdiffstats
path: root/src/svm/message_queue.c
diff options
context:
space:
mode:
authorFlorin Coras <fcoras@cisco.com>2021-03-25 11:24:33 -0700
committerDave Barach <openvpp@barachs.net>2021-03-25 23:50:14 +0000
commit89c98a4bff101ad4270724bde7ca6f21ba4c8482 (patch)
tree95178e327777d4112e6034ebcf18bab7a0fd1001 /src/svm/message_queue.c
parent22af286a2976339b25e11c23ac273e1781138fb7 (diff)
svm: add producer wait function
Type: refactor Signed-off-by: Florin Coras <fcoras@cisco.com> Change-Id: I9488ad7e045c908b60b5821d9c48583f6d513c2f
Diffstat (limited to 'src/svm/message_queue.c')
-rw-r--r--src/svm/message_queue.c47
1 files changed, 36 insertions, 11 deletions
diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c
index 3163d302967..5c04b19e64c 100644
--- a/src/svm/message_queue.c
+++ b/src/svm/message_queue.c
@@ -247,7 +247,7 @@ svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index,
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_MQ_WAIT_FULL);
+ svm_msg_q_wait_prod (mq);
*msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
}
return 0;
@@ -371,7 +371,7 @@ svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, int nowait)
if (nowait)
return (-2);
while (svm_msg_q_is_full (mq))
- svm_msg_q_wait (mq, SVM_MQ_WAIT_FULL);
+ svm_msg_q_wait_prod (mq);
}
svm_msg_q_add_raw (mq, (u8 *) msg);
@@ -498,21 +498,17 @@ svm_msg_q_wait (svm_msg_q_t *mq, svm_msg_q_wait_type_t type)
if (mq->q.evtfd == -1)
{
- if (type == SVM_MQ_WAIT_EMPTY)
+ rv = pthread_mutex_lock (&mq->q.shr->mutex);
+ if (PREDICT_FALSE (rv == EOWNERDEAD))
{
- rv = pthread_mutex_lock (&mq->q.shr->mutex);
- if (PREDICT_FALSE (rv == EOWNERDEAD))
- {
- rv = pthread_mutex_consistent (&mq->q.shr->mutex);
- return rv;
- }
+ rv = pthread_mutex_consistent (&mq->q.shr->mutex);
+ return rv;
}
while (fn (mq))
pthread_cond_wait (&mq->q.shr->condvar, &mq->q.shr->mutex);
- if (type == SVM_MQ_WAIT_EMPTY)
- pthread_mutex_unlock (&mq->q.shr->mutex);
+ pthread_mutex_unlock (&mq->q.shr->mutex);
}
else
{
@@ -535,6 +531,35 @@ 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)
+{
+ if (mq->q.evtfd == -1)
+ {
+ while (svm_msg_q_is_full (mq))
+ pthread_cond_wait (&mq->q.shr->condvar, &mq->q.shr->mutex);
+ }
+ else
+ {
+ u64 buf;
+ int rv;
+
+ while (svm_msg_q_is_full (mq))
+ {
+ 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)