commit 89c98a4bff101ad4270724bde7ca6f21ba4c8482
author Florin Coras <fcoras@cisco.com> Thu Mar 25 11:24:33 2021 -0700
committer Dave Barach <openvpp@barachs.net> Thu Mar 25 23:50:14 2021 +0000
tree 95178e327777d4112e6034ebcf18bab7a0fd1001
parent 22af286a2976339b25e11c23ac273e1781138fb7

svm: add producer wait function

Type: refactor

Signed-off-by: Florin Coras <fcoras@cisco.com>
Change-Id: I9488ad7e045c908b60b5821d9c48583f6d513c2f

src/svm/message_queue.c

diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c
index 3163d30..5c04b19 100644
--- a/src/svm/message_queue.c
+++ b/src/svm/message_queue.c
@@ -247,7 +247,7 @@
       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 @@
       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 @@
 
   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 @@
 }
 
 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)