api: refactor vlibmemory

- separate client/server code for both memory and socket apis
- separate memory api code from generic vlib api code
- move unix_shared_memory_fifo to svm and rename to svm_fifo_t
- overall declutter

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

2 files changed, 478 insertions, 0 deletions

diff --git a/src/svm/queue.c b/src/svm/queue.c
new file mode 100644
index 00000000000..aef409277db
--- /dev/null
+++ b/src/svm/queue.c
@@ -0,0 +1,410 @@
+/*
+ *------------------------------------------------------------------
+ * svm_queue.c - unidirectional shared-memory queues
+ *
+ * Copyright (c) 2009 Cisco and/or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *------------------------------------------------------------------
+ */
+
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include <vppinfra/mem.h>
+#include <vppinfra/format.h>
+#include <vppinfra/cache.h>
+#include <svm/queue.h>
+#include <signal.h>
+
+/*
+ * svm_queue_init
+ *
+ * nels = number of elements on the queue
+ * elsize = element size, presumably 4 and cacheline-size will
+ *          be popular choices.
+ * pid   = consumer pid
+ *
+ * The idea is to call this function in the queue consumer,
+ * and e-mail the queue pointer to the producer(s).
+ *
+ * The vpp process / main thread allocates one of these
+ * at startup; its main input queue. The vpp main input queue
+ * has a pointer to it in the shared memory segment header.
+ *
+ * You probably want to be on an svm data heap before calling this
+ * function.
+ */
+svm_queue_t *
+svm_queue_init (int nels,
+		int elsize, int consumer_pid, int signal_when_queue_non_empty)
+{
+  svm_queue_t *q;
+  pthread_mutexattr_t attr;
+  pthread_condattr_t cattr;
+
+  q = clib_mem_alloc_aligned (sizeof (svm_queue_t)
+			      + nels * elsize, CLIB_CACHE_LINE_BYTES);
+  memset (q, 0, sizeof (*q));
+
+  q->elsize = elsize;
+  q->maxsize = nels;
+  q->consumer_pid = consumer_pid;
+  q->signal_when_queue_non_empty = signal_when_queue_non_empty;
+
+  memset (&attr, 0, sizeof (attr));
+  memset (&cattr, 0, sizeof (cattr));
+
+  if (pthread_mutexattr_init (&attr))
+    clib_unix_warning ("mutexattr_init");
+  if (pthread_mutexattr_setpshared (&attr, PTHREAD_PROCESS_SHARED))
+    clib_unix_warning ("pthread_mutexattr_setpshared");
+  if (pthread_mutex_init (&q->mutex, &attr))
+    clib_unix_warning ("mutex_init");
+  if (pthread_mutexattr_destroy (&attr))
+    clib_unix_warning ("mutexattr_destroy");
+  if (pthread_condattr_init (&cattr))
+    clib_unix_warning ("condattr_init");
+  /* prints funny-looking messages in the Linux target */
+  if (pthread_condattr_setpshared (&cattr, PTHREAD_PROCESS_SHARED))
+    clib_unix_warning ("condattr_setpshared");
+  if (pthread_cond_init (&q->condvar, &cattr))
+    clib_unix_warning ("cond_init1");
+  if (pthread_condattr_destroy (&cattr))
+    clib_unix_warning ("cond_init2");
+
+  return (q);
+}
+
+/*
+ * svm_queue_free
+ */
+void
+svm_queue_free (svm_queue_t * q)
+{
+  (void) pthread_mutex_destroy (&q->mutex);
+  (void) pthread_cond_destroy (&q->condvar);
+  clib_mem_free (q);
+}
+
+void
+svm_queue_lock (svm_queue_t * q)
+{
+  pthread_mutex_lock (&q->mutex);
+}
+
+void
+svm_queue_unlock (svm_queue_t * q)
+{
+  pthread_mutex_unlock (&q->mutex);
+}
+
+int
+svm_queue_is_full (svm_queue_t * q)
+{
+  return q->cursize == q->maxsize;
+}
+
+/*
+ * svm_queue_add_nolock
+ */
+int
+svm_queue_add_nolock (svm_queue_t * q, u8 * elem)
+{
+  i8 *tailp;
+  int need_broadcast = 0;
+
+  if (PREDICT_FALSE (q->cursize == q->maxsize))
+    {
+      while (q->cursize == q->maxsize)
+	{
+	  (void) pthread_cond_wait (&q->condvar, &q->mutex);
+	}
+    }
+
+  tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
+  clib_memcpy (tailp, elem, q->elsize);
+
+  q->tail++;
+  q->cursize++;
+
+  need_broadcast = (q->cursize == 1);
+
+  if (q->tail == q->maxsize)
+    q->tail = 0;
+
+  if (need_broadcast)
+    {
+      (void) pthread_cond_broadcast (&q->condvar);
+      if (q->signal_when_queue_non_empty)
+	kill (q->consumer_pid, q->signal_when_queue_non_empty);
+    }
+  return 0;
+}
+
+int
+svm_queue_add_raw (svm_queue_t * q, u8 * elem)
+{
+  i8 *tailp;
+
+  if (PREDICT_FALSE (q->cursize == q->maxsize))
+    {
+      while (q->cursize == q->maxsize)
+	;
+    }
+
+  tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
+  clib_memcpy (tailp, elem, q->elsize);
+
+  q->tail++;
+  q->cursize++;
+
+  if (q->tail == q->maxsize)
+    q->tail = 0;
+  return 0;
+}
+
+
+/*
+ * svm_queue_add
+ */
+int
+svm_queue_add (svm_queue_t * q, u8 * elem, int nowait)
+{
+  i8 *tailp;
+  int need_broadcast = 0;
+
+  if (nowait)
+    {
+      /* zero on success */
+      if (pthread_mutex_trylock (&q->mutex))
+	{
+	  return (-1);
+	}
+    }
+  else
+    pthread_mutex_lock (&q->mutex);
+
+  if (PREDICT_FALSE (q->cursize == q->maxsize))
+    {
+      if (nowait)
+	{
+	  pthread_mutex_unlock (&q->mutex);
+	  return (-2);
+	}
+      while (q->cursize == q->maxsize)
+	{
+	  (void) pthread_cond_wait (&q->condvar, &q->mutex);
+	}
+    }
+
+  tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
+  clib_memcpy (tailp, elem, q->elsize);
+
+  q->tail++;
+  q->cursize++;
+
+  need_broadcast = (q->cursize == 1);
+
+  if (q->tail == q->maxsize)
+    q->tail = 0;
+
+  if (need_broadcast)
+    {
+      (void) pthread_cond_broadcast (&q->condvar);
+      if (q->signal_when_queue_non_empty)
+	kill (q->consumer_pid, q->signal_when_queue_non_empty);
+    }
+  pthread_mutex_unlock (&q->mutex);
+
+  return 0;
+}
+
+/*
+ * svm_queue_add2
+ */
+int
+svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait)
+{
+  i8 *tailp;
+  int need_broadcast = 0;
+
+  if (nowait)
+    {
+      /* zero on success */
+      if (pthread_mutex_trylock (&q->mutex))
+	{
+	  return (-1);
+	}
+    }
+  else
+    pthread_mutex_lock (&q->mutex);
+
+  if (PREDICT_FALSE (q->cursize + 1 == q->maxsize))
+    {
+      if (nowait)
+	{
+	  pthread_mutex_unlock (&q->mutex);
+	  return (-2);
+	}
+      while (q->cursize + 1 == q->maxsize)
+	{
+	  (void) pthread_cond_wait (&q->condvar, &q->mutex);
+	}
+    }
+
+  tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
+  clib_memcpy (tailp, elem, q->elsize);
+
+  q->tail++;
+  q->cursize++;
+
+  if (q->tail == q->maxsize)
+    q->tail = 0;
+
+  need_broadcast = (q->cursize == 1);
+
+  tailp = (i8 *) (&q->data[0] + q->elsize * q->tail);
+  clib_memcpy (tailp, elem2, q->elsize);
+
+  q->tail++;
+  q->cursize++;
+
+  if (q->tail == q->maxsize)
+    q->tail = 0;
+
+  if (need_broadcast)
+    {
+      (void) pthread_cond_broadcast (&q->condvar);
+      if (q->signal_when_queue_non_empty)
+	kill (q->consumer_pid, q->signal_when_queue_non_empty);
+    }
+  pthread_mutex_unlock (&q->mutex);
+
+  return 0;
+}
+
+/*
+ * svm_queue_sub
+ */
+int
+svm_queue_sub (svm_queue_t * q, u8 * elem, int nowait)
+{
+  i8 *headp;
+  int need_broadcast = 0;
+
+  if (nowait)
+    {
+      /* zero on success */
+      if (pthread_mutex_trylock (&q->mutex))
+	{
+	  return (-1);
+	}
+    }
+  else
+    pthread_mutex_lock (&q->mutex);
+
+  if (PREDICT_FALSE (q->cursize == 0))
+    {
+      if (nowait)
+	{
+	  pthread_mutex_unlock (&q->mutex);
+	  return (-2);
+	}
+      while (q->cursize == 0)
+	{
+	  (void) pthread_cond_wait (&q->condvar, &q->mutex);
+	}
+    }
+
+  headp = (i8 *) (&q->data[0] + q->elsize * q->head);
+  clib_memcpy (elem, headp, q->elsize);
+
+  q->head++;
+  /* $$$$ JFC shouldn't this be == 0? */
+  if (q->cursize == q->maxsize)
+    need_broadcast = 1;
+
+  q->cursize--;
+
+  if (q->head == q->maxsize)
+    q->head = 0;
+
+  if (need_broadcast)
+    (void) pthread_cond_broadcast (&q->condvar);
+
+  pthread_mutex_unlock (&q->mutex);
+
+  return 0;
+}
+
+int
+svm_queue_sub2 (svm_queue_t * q, u8 * elem)
+{
+  int need_broadcast;
+  i8 *headp;
+
+  pthread_mutex_lock (&q->mutex);
+  if (q->cursize == 0)
+    {
+      pthread_mutex_unlock (&q->mutex);
+      return -1;
+    }
+
+  headp = (i8 *) (&q->data[0] + q->elsize * q->head);
+  clib_memcpy (elem, headp, q->elsize);
+
+  q->head++;
+  need_broadcast = (q->cursize == q->maxsize / 2);
+  q->cursize--;
+
+  if (PREDICT_FALSE (q->head == q->maxsize))
+    q->head = 0;
+  pthread_mutex_unlock (&q->mutex);
+
+  if (need_broadcast)
+    (void) pthread_cond_broadcast (&q->condvar);
+
+  return 0;
+}
+
+int
+svm_queue_sub_raw (svm_queue_t * q, u8 * elem)
+{
+  i8 *headp;
+
+  if (PREDICT_FALSE (q->cursize == 0))
+    {
+      while (q->cursize == 0)
+	;
+    }
+
+  headp = (i8 *) (&q->data[0] + q->elsize * q->head);
+  clib_memcpy (elem, headp, q->elsize);
+
+  q->head++;
+  q->cursize--;
+
+  if (q->head == q->maxsize)
+    q->head = 0;
+  return 0;
+}
+
+/*
+ * fd.io coding-style-patch-verification: ON
+ *
+ * Local Variables:
+ * eval: (c-set-style "gnu")
+ * End:
+ */
diff --git a/src/svm/queue.h b/src/svm/queue.h
new file mode 100644
index 00000000000..dc1fc36e002
--- /dev/null
+++ b/src/svm/queue.h
@@ -0,0 +1,68 @@
+/*
+ *------------------------------------------------------------------
+ * svm_queue.h - shared-memory queues
+ *
+ * Copyright (c) 2009 Cisco and/or its affiliates.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *------------------------------------------------------------------
+ */
+
+#ifndef included_svm_queue_h
+#define included_svm_queue_h
+
+#include <pthread.h>
+
+typedef struct _svm_queue
+{
+  pthread_mutex_t mutex;	/* 8 bytes */
+  pthread_cond_t condvar;	/* 8 bytes */
+  int head;
+  int tail;
+  volatile int cursize;
+  int maxsize;
+  int elsize;
+  int consumer_pid;
+  int signal_when_queue_non_empty;
+  char data[0];
+} svm_queue_t;
+
+svm_queue_t *svm_queue_init (int nels,
+			     int elsize,
+			     int consumer_pid,
+			     int signal_when_queue_non_empty);
+void svm_queue_free (svm_queue_t * q);
+int svm_queue_add (svm_queue_t * q, u8 * elem, int nowait);
+int svm_queue_add2 (svm_queue_t * q, u8 * elem, u8 * elem2, int nowait);
+int svm_queue_sub (svm_queue_t * q, u8 * elem, int nowait);
+int svm_queue_sub2 (svm_queue_t * q, u8 * elem);
+void svm_queue_lock (svm_queue_t * q);
+void svm_queue_unlock (svm_queue_t * q);
+int svm_queue_is_full (svm_queue_t * q);
+int svm_queue_add_nolock (svm_queue_t * q, u8 * elem);
+int svm_queue_sub_raw (svm_queue_t * q, u8 * elem);
+int svm_queue_add_raw (svm_queue_t * q, u8 * elem);
+
+/*
+ * DEPRECATED please use svm_queue_t instead
+ */
+typedef svm_queue_t unix_shared_memory_queue_t;
+
+#endif /* included_svm_queue_h */
+
+/*
+ * fd.io coding-style-patch-verification: ON
+ *
+ * Local Variables:
+ * eval: (c-set-style "gnu")
+ * End:
+ */