session: use msg queue for events

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

27 files changed, 773 insertions, 400 deletions

diff --git a/src/svm/message_queue.c b/src/svm/message_queue.c
index 4f3e7642740..89411143c12 100644
--- a/src/svm/message_queue.c
+++ b/src/svm/message_queue.c
@@ -16,6 +16,25 @@
 #include <svm/message_queue.h>
 #include <vppinfra/mem.h>
 
+static inline svm_msg_q_ring_t *
+svm_msg_q_ring_inline (svm_msg_q_t * mq, u32 ring_index)
+{
+  return vec_elt_at_index (mq->rings, ring_index);
+}
+
+svm_msg_q_ring_t *
+svm_msg_q_ring (svm_msg_q_t * mq, u32 ring_index)
+{
+  return svm_msg_q_ring_inline (mq, ring_index);
+}
+
+static inline void *
+svm_msg_q_ring_data (svm_msg_q_ring_t * ring, u32 elt_index)
+{
+  ASSERT (elt_index < ring->nitems);
+  return (ring->data + elt_index * ring->elsize);
+}
+
 svm_msg_q_t *
 svm_msg_q_alloc (svm_msg_q_cfg_t * cfg)
 {
@@ -63,6 +82,53 @@ svm_msg_q_free (svm_msg_q_t * mq)
 }
 
 svm_msg_q_msg_t
+svm_msg_q_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index)
+{
+  svm_msg_q_msg_t msg = {.as_u64 = ~0 };
+  svm_msg_q_ring_t *ring = svm_msg_q_ring_inline (mq, ring_index);
+
+  ASSERT (ring->cursize != ring->nitems);
+  msg.ring_index = ring - mq->rings;
+  msg.elt_index = ring->tail;
+  ring->tail = (ring->tail + 1) % ring->nitems;
+  __sync_fetch_and_add (&ring->cursize, 1);
+  return msg;
+}
+
+int
+svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index,
+				     u8 noblock, svm_msg_q_msg_t * msg)
+{
+  if (noblock)
+    {
+      if (svm_msg_q_try_lock (mq))
+	return -1;
+      if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, ring_index)))
+	{
+	  svm_msg_q_unlock (mq);
+	  return -2;
+	}
+      *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
+      if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (msg)))
+	{
+	  svm_msg_q_unlock (mq);
+	  return -2;
+	}
+    }
+  else
+    {
+      svm_msg_q_lock (mq);
+      *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
+      while (svm_msg_q_msg_is_invalid (msg))
+	{
+	  svm_msg_q_wait (mq);
+	  *msg = svm_msg_q_alloc_msg_w_ring (mq, ring_index);
+	}
+    }
+  return 0;
+}
+
+svm_msg_q_msg_t
 svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes)
 {
   svm_msg_q_msg_t msg = {.as_u64 = ~0 };
@@ -81,23 +147,10 @@ svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes)
   return msg;
 }
 
-static inline svm_msg_q_ring_t *
-svm_msg_q_get_ring (svm_msg_q_t * mq, u32 ring_index)
-{
-  return vec_elt_at_index (mq->rings, ring_index);
-}
-
-static inline void *
-svm_msg_q_ring_data (svm_msg_q_ring_t * ring, u32 elt_index)
-{
-  ASSERT (elt_index < ring->nitems);
-  return (ring->data + elt_index * ring->elsize);
-}
-
 void *
 svm_msg_q_msg_data (svm_msg_q_t * mq, svm_msg_q_msg_t * msg)
 {
-  svm_msg_q_ring_t *ring = svm_msg_q_get_ring (mq, msg->ring_index);
+  svm_msg_q_ring_t *ring = svm_msg_q_ring_inline (mq, msg->ring_index);
   return svm_msg_q_ring_data (ring, msg->elt_index);
 }
 
@@ -131,7 +184,7 @@ svm_msq_q_msg_is_valid (svm_msg_q_t * mq, svm_msg_q_msg_t * msg)
     return 0;
   ring = &mq->rings[msg->ring_index];
 
-  dist1 = ((ring->nitems + msg->ring_index) - ring->head) % ring->nitems;
+  dist1 = ((ring->nitems + msg->elt_index) - ring->head) % ring->nitems;
   if (ring->tail == ring->head)
     dist2 = (ring->cursize == 0) ? 0 : ring->nitems;
   else
@@ -140,10 +193,17 @@ svm_msq_q_msg_is_valid (svm_msg_q_t * mq, svm_msg_q_msg_t * msg)
 }
 
 int
-svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait)
+svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, int nowait)
+{
+  ASSERT (svm_msq_q_msg_is_valid (mq, msg));
+  return svm_queue_add (mq->q, (u8 *) msg, nowait);
+}
+
+void
+svm_msg_q_add_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg)
 {
-  ASSERT (svm_msq_q_msg_is_valid (mq, &msg));
-  return svm_queue_add (mq->q, (u8 *) & msg, nowait);
+  ASSERT (svm_msq_q_msg_is_valid (mq, msg));
+  svm_queue_add_raw (mq->q, (u8 *) msg);
 }
 
 int
@@ -153,6 +213,12 @@ svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg,
   return svm_queue_sub (mq->q, (u8 *) msg, cond, time);
 }
 
+void
+svm_msg_q_sub_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg)
+{
+  svm_queue_sub_raw (mq->q, (u8 *) msg);
+}
+
 /*
  * fd.io coding-style-patch-verification: ON
  *
diff --git a/src/svm/message_queue.h b/src/svm/message_queue.h
index 5ec8547016e..708a03d716e 100644
--- a/src/svm/message_queue.h
+++ b/src/svm/message_queue.h
@@ -21,14 +21,15 @@
 #define SRC_SVM_MESSAGE_QUEUE_H_
 
 #include <vppinfra/clib.h>
+#include <vppinfra/error.h>
 #include <svm/queue.h>
 
 typedef struct svm_msg_q_ring_
 {
   volatile u32 cursize;			/**< current size of the ring */
   u32 nitems;				/**< max size of the ring */
-  u32 head;				/**< current head (for dequeue) */
-  u32 tail;				/**< current tail (for enqueue) */
+  volatile u32 head;			/**< current head (for dequeue) */
+  volatile u32 tail;			/**< current tail (for enqueue) */
   u32 elsize;				/**< size of an element */
   u8 *data;				/**< chunk of memory for msg data */
 } svm_msg_q_ring_t;
@@ -64,6 +65,7 @@ typedef union
   u64 as_u64;
 } svm_msg_q_msg_t;
 
+#define SVM_MQ_INVALID_MSG { .as_u64 = ~0 }
 /**
  * Allocate message queue
  *
@@ -98,6 +100,36 @@ void svm_msg_q_free (svm_msg_q_t * mq);
 svm_msg_q_msg_t svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes);
 
 /**
+ * Allocate message buffer on ring
+ *
+ * Message is allocated, on requested ring. The caller MUST check that
+ * the ring is not full.
+ *
+ * @param mq		message queue
+ * @param ring_index	ring on which the allocation should occur
+ * @return		message structure pointing to the ring and position
+ * 			allocated
+ */
+svm_msg_q_msg_t svm_msg_q_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index);
+
+/**
+ * Lock message queue and allocate message buffer on ring
+ *
+ * This should be used when multiple writers/readers are expected to
+ * compete for the rings/queue. Message should be enqueued by calling
+ * @ref svm_msg_q_add_w_lock and the caller MUST unlock the queue once
+ * the message in enqueued.
+ *
+ * @param mq		message queue
+ * @param ring_index	ring on which the allocation should occur
+ * @param noblock	flag that indicates if request should block
+ * @param msg		pointer to message to be filled in
+ * @return		0 on success, negative number otherwise
+ */
+int svm_msg_q_lock_and_alloc_msg_w_ring (svm_msg_q_t * mq, u32 ring_index,
+					 u8 noblock, svm_msg_q_msg_t * msg);
+
+/**
  * Free message buffer
  *
  * Marks message buffer on ring as free.
@@ -106,6 +138,7 @@ svm_msg_q_msg_t svm_msg_q_alloc_msg (svm_msg_q_t * mq, u32 nbytes);
  * @param msg		message to be freed
  */
 void svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg);
+
 /**
  * Producer enqueue one message to queue
  *
@@ -117,7 +150,20 @@ void svm_msg_q_free_msg (svm_msg_q_t * mq, svm_msg_q_msg_t * msg);
  * @param nowait	flag to indicate if request is blocking or not
  * @return		success status
  */
-int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait);
+int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, int nowait);
+
+/**
+ * Producer enqueue one message to queue with mutex held
+ *
+ * Prior to calling this, the producer should've obtained a message buffer
+ * from one of the rings by calling @ref svm_msg_q_alloc_msg. It assumes
+ * the queue mutex is held.
+ *
+ * @param mq		message queue
+ * @param msg		message (pointer to ring position) to be enqueued
+ * @return		success status
+ */
+void svm_msg_q_add_w_lock (svm_msg_q_t * mq, svm_msg_q_msg_t * msg);
 
 /**
  * Consumer dequeue one message from queue
@@ -129,13 +175,28 @@ int svm_msg_q_add (svm_msg_q_t * mq, svm_msg_q_msg_t msg, int nowait);
  * @param mq		message queue
  * @param msg		pointer to structure where message is to be received
  * @param cond		flag that indicates if request should block or not
+ * @param time		time to wait if condition it SVM_Q_TIMEDWAIT
  * @return		success status
  */
 int svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg,
 		   svm_q_conditional_wait_t cond, u32 time);
 
 /**
- * Get data for message in queu
+ * Consumer dequeue one message from queue with mutex held
+ *
+ * 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
+ * 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);
+
+/**
+ * Get data for message in queue
  *
  * @param mq		message queue
  * @param msg		message for which the data is requested
@@ -143,6 +204,94 @@ int svm_msg_q_sub (svm_msg_q_t * mq, svm_msg_q_msg_t * msg,
  */
 void *svm_msg_q_msg_data (svm_msg_q_t * mq, svm_msg_q_msg_t * msg);
 
+/**
+ * Get message queue ring
+ *
+ * @param mq		message queue
+ * @param ring_index	index of ring
+ * @return		pointer to ring
+ */
+svm_msg_q_ring_t *svm_msg_q_ring (svm_msg_q_t * mq, u32 ring_index);
+
+/**
+ * Check if message queue is full
+ */
+static inline u8
+svm_msg_q_is_full (svm_msg_q_t * mq)
+{
+  return (mq->q->cursize == mq->q->maxsize);
+}
+
+static inline u8
+svm_msg_q_ring_is_full (svm_msg_q_t * mq, u32 ring_index)
+{
+  ASSERT (ring_index < vec_len (mq->rings));
+  return (mq->rings[ring_index].cursize == mq->rings[ring_index].nitems);
+}
+
+/**
+ * Check if message queue is empty
+ */
+static inline u8
+svm_msg_q_is_empty (svm_msg_q_t * mq)
+{
+  return (mq->q->cursize == 0);
+}
+
+/**
+ * Check length of message queue
+ */
+static inline u32
+svm_msg_q_size (svm_msg_q_t * mq)
+{
+  return mq->q->cursize;
+}
+
+/**
+ * Check if message is invalid
+ */
+static inline u8
+svm_msg_q_msg_is_invalid (svm_msg_q_msg_t * msg)
+{
+  return (msg->as_u64 == (u64) ~ 0);
+}
+
+/**
+ * Try locking message queue
+ */
+static inline int
+svm_msg_q_try_lock (svm_msg_q_t * mq)
+{
+  return pthread_mutex_trylock (&mq->q->mutex);
+}
+
+/**
+ * Lock, or block trying, the message queue
+ */
+static inline int
+svm_msg_q_lock (svm_msg_q_t * mq)
+{
+  return pthread_mutex_lock (&mq->q->mutex);
+}
+
+static inline void
+svm_msg_q_wait (svm_msg_q_t * mq)
+{
+  pthread_cond_wait (&mq->q->condvar, &mq->q->mutex);
+}
+
+/**
+ * Unlock message queue
+ */
+static inline void
+svm_msg_q_unlock (svm_msg_q_t * mq)
+{
+  /* The other side of the connection is not polling */
+  if (mq->q->cursize < (mq->q->maxsize / 8))
+    (void) pthread_cond_broadcast (&mq->q->condvar);
+  pthread_mutex_unlock (&mq->q->mutex);
+}
+
 #endif /* SRC_SVM_MESSAGE_QUEUE_H_ */
 
 /*
diff --git a/src/svm/queue.c b/src/svm/queue.c
index 96e40fc2aec..8e18f5832e3 100644
--- a/src/svm/queue.c
+++ b/src/svm/queue.c
@@ -154,26 +154,16 @@ svm_queue_add_nolock (svm_queue_t * q, u8 * elem)
   return 0;
 }
 
-int
+void
 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->tail = (q->tail + 1) % q->maxsize;
   q->cursize++;
-
-  if (q->tail == q->maxsize)
-    q->tail = 0;
-  return 0;
 }
 
 
@@ -414,11 +404,9 @@ svm_queue_sub_raw (svm_queue_t * q, u8 * elem)
   headp = (i8 *) (&q->data[0] + q->elsize * q->head);
   clib_memcpy (elem, headp, q->elsize);
 
-  q->head++;
+  q->head = (q->head + 1) % q->maxsize;
   q->cursize--;
 
-  if (q->head == q->maxsize)
-    q->head = 0;
   return 0;
 }
 
diff --git a/src/svm/queue.h b/src/svm/queue.h
index 856c17237d1..68a63d769b6 100644
--- a/src/svm/queue.h
+++ b/src/svm/queue.h
@@ -69,7 +69,13 @@ 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);
+
+/**
+ * Add element to queue with mutex held
+ * @param q		queue
+ * @param elem		pointer element data to add
+ */
+void svm_queue_add_raw (svm_queue_t * q, u8 * elem);
 
 /*
  * DEPRECATED please use svm_queue_t instead
diff --git a/src/svm/svm_fifo.h b/src/svm/svm_fifo.h
index 0d5a08b86ae..40182901db5 100644
--- a/src/svm/svm_fifo.h
+++ b/src/svm/svm_fifo.h
@@ -107,6 +107,12 @@ svm_fifo_max_dequeue (svm_fifo_t * f)
   return f->cursize;
 }
 
+static inline int
+svm_fifo_is_full (svm_fifo_t * f)
+{
+  return (f->cursize == f->nitems);
+}
+
 static inline u32
 svm_fifo_max_enqueue (svm_fifo_t * f)
 {
diff --git a/src/svm/test_svm_message_queue.c b/src/svm/test_svm_message_queue.c
index 69ffd131ac2..758163ffeab 100644
--- a/src/svm/test_svm_message_queue.c
+++ b/src/svm/test_svm_message_queue.c
@@ -88,9 +88,9 @@ test1 (int verbose)
       test1_error ("failed: msg alloc3");
 
   *(u32 *)svm_msg_q_msg_data (mq, &msg2) = 123;
-  svm_msg_q_add (mq, msg2, SVM_Q_NOWAIT);
+  svm_msg_q_add (mq, &msg2, SVM_Q_NOWAIT);
   for (i = 0; i < 12; i++)
-    svm_msg_q_add (mq, msg[i], SVM_Q_NOWAIT);
+    svm_msg_q_add (mq, &msg[i], SVM_Q_NOWAIT);
 
   if (svm_msg_q_sub (mq, &msg2, SVM_Q_NOWAIT, 0))
     test1_error ("failed: dequeue1");
diff --git a/src/tests/vnet/session/tcp_echo.c b/src/tests/vnet/session/tcp_echo.c
index 59314f943f0..f8b75d95ac7 100644
--- a/src/tests/vnet/session/tcp_echo.c
+++ b/src/tests/vnet/session/tcp_echo.c
@@ -48,7 +48,7 @@ typedef struct
   svm_fifo_t *server_rx_fifo;
   svm_fifo_t *server_tx_fifo;
 
-  svm_queue_t *vpp_evt_q;
+  svm_msg_q_t *vpp_evt_q;
 
   u64 vpp_session_handle;
   u64 bytes_sent;
@@ -99,7 +99,7 @@ typedef struct
   int no_return;
 
   /* Our event queue */
-  svm_queue_t *our_event_queue;
+  svm_msg_q_t *our_event_queue;
 
   /* $$$ single thread only for the moment */
   svm_queue_t *vpp_event_queue;
@@ -426,7 +426,7 @@ vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
 
   ASSERT (mp->app_event_queue_address);
   em->our_event_queue = uword_to_pointer (mp->app_event_queue_address,
-					  svm_queue_t *);
+					  svm_msg_q_t *);
   em->state = STATE_ATTACHED;
 }
 
@@ -677,7 +677,6 @@ send_test_chunk (echo_main_t * em, session_t * s)
   svm_fifo_t *tx_fifo = s->server_tx_fifo;
   u8 *test_data = em->connect_test_data;
   u32 enq_space, min_chunk = 16 << 10;
-  session_fifo_event_t evt;
   int written;
 
   test_buf_len = vec_len (test_data);
@@ -698,12 +697,8 @@ send_test_chunk (echo_main_t * em, session_t * s)
       s->bytes_sent += written;
 
       if (svm_fifo_set_event (tx_fifo))
-	{
-	  /* Fabricate TX event, send to vpp */
-	  evt.fifo = tx_fifo;
-	  evt.event_type = FIFO_EVENT_APP_TX;
-	  svm_queue_add (s->vpp_evt_q, (u8 *) & evt, 0 /* wait for mutex */ );
-	}
+	app_send_io_evt_to_vpp (s->vpp_evt_q, tx_fifo, FIFO_EVENT_APP_TX,
+				0 /* do wait for mutex */ );
     }
 }
 
@@ -751,6 +746,7 @@ client_rx_thread_fn (void *arg)
   session_fifo_event_t _e, *e = &_e;
   echo_main_t *em = &echo_main;
   static u8 *rx_buf = 0;
+  svm_msg_q_msg_t msg;
 
   vec_validate (rx_buf, 1 << 20);
 
@@ -759,7 +755,8 @@ client_rx_thread_fn (void *arg)
 
   while (!em->time_to_stop)
     {
-      svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
+      svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
@@ -769,6 +766,7 @@ client_rx_thread_fn (void *arg)
 	  clib_warning ("unknown event type %d", e->event_type);
 	  break;
 	}
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
     }
   pthread_exit (0);
 }
@@ -808,7 +806,7 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
   session->vpp_session_handle = mp->handle;
   session->start = clib_time_now (&em->clib_time);
   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-					 svm_queue_t *);
+					 svm_msg_q_t *);
 
   hash_set (em->session_index_by_vpp_handles, mp->handle, session_index);
 
@@ -869,8 +867,8 @@ client_disconnect (echo_main_t * em, session_t * s)
 static void
 clients_run (echo_main_t * em)
 {
-  session_fifo_event_t _e, *e = &_e;
   f64 start_time, deltat, timeout = 100.0;
+  svm_msg_q_msg_t msg;
   session_t *s;
   int i;
 
@@ -918,9 +916,15 @@ clients_run (echo_main_t * em)
   start_time = clib_time_now (&em->clib_time);
   em->state = STATE_READY;
   while (em->n_active_clients)
-    if (em->our_event_queue->cursize)
-      svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_NOWAIT, 0);
-
+    if (!svm_msg_q_is_empty (em->our_event_queue))
+      {
+	if (svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0))
+	  {
+	    clib_warning ("svm msg q returned");
+	  }
+	else
+	  svm_msg_q_free_msg (em->our_event_queue, &msg);
+      }
 
   for (i = 0; i < em->n_clients; i++)
     {
@@ -1180,10 +1184,12 @@ void
 server_handle_event_queue (echo_main_t * em)
 {
   session_fifo_event_t _e, *e = &_e;
+  svm_msg_q_msg_t msg;
 
   while (1)
     {
-      svm_queue_sub (em->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
+      svm_msg_q_sub (em->our_event_queue, &msg, SVM_Q_WAIT, 0);
+      e = svm_msg_q_msg_data (em->our_event_queue, &msg);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
@@ -1191,6 +1197,7 @@ server_handle_event_queue (echo_main_t * em)
 	  break;
 
 	case FIFO_EVENT_DISCONNECT:
+	  svm_msg_q_free_msg (em->our_event_queue, &msg);
 	  return;
 
 	default:
@@ -1204,6 +1211,7 @@ server_handle_event_queue (echo_main_t * em)
 	  em->time_to_print_stats = 0;
 	  fformat (stdout, "%d connections\n", pool_elts (em->sessions));
 	}
+      svm_msg_q_free_msg (em->our_event_queue, &msg);
     }
 }
 
diff --git a/src/tests/vnet/session/udp_echo.c b/src/tests/vnet/session/udp_echo.c
index 54e00b181bd..d796b6b8de0 100644
--- a/src/tests/vnet/session/udp_echo.c
+++ b/src/tests/vnet/session/udp_echo.c
@@ -87,10 +87,10 @@ typedef struct
   u8 is_connected;
 
   /* Our event queue */
-  svm_queue_t *our_event_queue;
+  svm_msg_q_t *our_event_queue;
 
   /* $$$ single thread only for the moment */
-  svm_queue_t *vpp_event_queue;
+  svm_msg_q_t *vpp_event_queue;
 
   /* $$$$ hack: cut-through session index */
   volatile u32 cut_through_session_index;
@@ -369,7 +369,7 @@ vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
     }
 
   utm->our_event_queue =
-    uword_to_pointer (mp->app_event_queue_address, svm_queue_t *);
+    uword_to_pointer (mp->app_event_queue_address, svm_msg_q_t *);
 }
 
 static void
@@ -736,7 +736,7 @@ vl_api_bind_uri_reply_t_handler (vl_api_bind_uri_reply_t * mp)
 	       sizeof (ip46_address_t));
   session->transport.is_ip4 = mp->lcl_is_ip4;
   session->transport.lcl_port = mp->lcl_port;
-  session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_queue_t *);
+  session->vpp_evt_q = uword_to_pointer (mp->vpp_evt_q, svm_msg_q_t *);
 
   utm->state = utm->is_connected ? STATE_BOUND : STATE_READY;
 }
@@ -896,7 +896,7 @@ vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
     start_time = clib_time_now (&utm->clib_time);
 
   utm->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
-					   svm_queue_t *);
+					   svm_msg_q_t *);
   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
   tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
 
@@ -909,7 +909,7 @@ vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
     {
       clib_warning ("cut-through session");
       utm->our_event_queue = uword_to_pointer (mp->server_event_queue_address,
-					       svm_queue_t *);
+					       svm_msg_q_t *);
       rx_fifo->master_session_index = session_index;
       tx_fifo->master_session_index = session_index;
       utm->cut_through_session_index = session_index;
@@ -1012,23 +1012,23 @@ vl_api_connect_session_reply_t_handler (vl_api_connect_session_reply_t * mp)
   session->rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
   session->tx_fifo = uword_to_pointer (mp->server_tx_fifo, svm_fifo_t *);
   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-					 svm_queue_t *);
+					 svm_msg_q_t *);
   /* Cut-through case */
   if (mp->client_event_queue_address)
     {
       clib_warning ("cut-through session");
       utm->cut_through_session_index = session - utm->sessions;
       utm->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
-					       svm_queue_t *);
+					       svm_msg_q_t *);
       utm->our_event_queue = uword_to_pointer (mp->client_event_queue_address,
-					       svm_queue_t *);
+					       svm_msg_q_t *);
       utm->do_echo = 1;
     }
   else
     {
       utm->connected_session = session - utm->sessions;
       utm->vpp_event_queue = uword_to_pointer (mp->vpp_event_queue_address,
-					       svm_queue_t *);
+					       svm_msg_q_t *);
 
       clib_memcpy (&session->transport.lcl_ip, mp->lcl_ip,
 		   sizeof (ip46_address_t));
@@ -1134,14 +1134,20 @@ server_handle_fifo_event_rx (udp_echo_main_t * utm, session_fifo_event_t * e)
 void
 server_handle_event_queue (udp_echo_main_t * utm)
 {
-  session_fifo_event_t _e, *e = &_e;
+  session_fifo_event_t *e;
+  svm_msg_q_msg_t msg;
 
   while (utm->state != STATE_READY)
     sleep (5);
 
   while (1)
     {
-      svm_queue_sub (utm->our_event_queue, (u8 *) e, SVM_Q_WAIT, 0);
+      if (svm_msg_q_sub (utm->our_event_queue, &msg, SVM_Q_WAIT, 0))
+	{
+	  clib_warning ("svm msg q returned");
+	  continue;
+	}
+      e = svm_msg_q_msg_data (utm->our_event_queue, &msg);
       switch (e->event_type)
 	{
 	case FIFO_EVENT_APP_RX:
@@ -1149,12 +1155,14 @@ server_handle_event_queue (udp_echo_main_t * utm)
 	  break;
 
 	case FIFO_EVENT_DISCONNECT:
-	  return;
+	  utm->time_to_stop = 1;
+	  break;
 
 	default:
 	  clib_warning ("unknown event type %d", e->event_type);
 	  break;
 	}
+      svm_msg_q_free_msg (utm->our_event_queue, &msg);
       if (PREDICT_FALSE (utm->time_to_stop == 1))
 	return;
       if (PREDICT_FALSE (utm->time_to_print_stats == 1))
diff --git a/src/vcl/vcl_bapi.c b/src/vcl/vcl_bapi.c
index cc3179d860a..ca65782e9c6 100644
--- a/src/vcl/vcl_bapi.c
+++ b/src/vcl/vcl_bapi.c
@@ -99,7 +99,7 @@ vl_api_application_attach_reply_t_handler (vl_api_application_attach_reply_t *
     }
 
   vcm->app_event_queue =
-    uword_to_pointer (mp->app_event_queue_address, svm_queue_t *);
+    uword_to_pointer (mp->app_event_queue_address, svm_msg_q_t *);
 
   vcm->app_state = STATE_APP_ATTACHED;
 }
@@ -291,7 +291,7 @@ done:
       VCL_IO_SESSIONS_UNLOCK ();
     }
   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-					 svm_queue_t *);
+					 svm_msg_q_t *);
 
   rx_fifo = uword_to_pointer (mp->server_rx_fifo, svm_fifo_t *);
   rx_fifo->client_session_index = session_index;
@@ -431,7 +431,7 @@ vl_api_accept_session_t_handler (vl_api_accept_session_t * mp)
   session->rx_fifo = rx_fifo;
   session->tx_fifo = tx_fifo;
   session->vpp_evt_q = uword_to_pointer (mp->vpp_event_queue_address,
-					 svm_queue_t *);
+					 svm_msg_q_t *);
   session->session_state = STATE_ACCEPT;
   session->transport.rmt_port = mp->port;
   session->transport.is_ip4 = mp->is_ip4;
diff --git a/src/vcl/vcl_private.h b/src/vcl/vcl_private.h
index 4283b6e1167..aba4839f129 100644
--- a/src/vcl/vcl_private.h
+++ b/src/vcl/vcl_private.h
@@ -194,7 +194,7 @@ typedef struct vppcom_main_t_
   clib_bitmap_t *ex_bitmap;
 
   /* Our event queue */
-  svm_queue_t *app_event_queue;
+  svm_msg_q_t *app_event_queue;
 
   /* unique segment name counter */
   u32 unique_segment_index;
diff --git a/src/vcl/vppcom.c b/src/vcl/vppcom.c
index 1fd138e6dd3..0d077caa797 100644
--- a/src/vcl/vppcom.c
+++ b/src/vcl/vppcom.c
@@ -1164,16 +1164,19 @@ vppcom_session_read_ready (vcl_session_t * session, u32 session_index)
     }
   rv = ready;
 
-  if (vcm->app_event_queue->cursize &&
-      !pthread_mutex_trylock (&vcm->app_event_queue->mutex))
+  if (!svm_msg_q_is_empty (vcm->app_event_queue) &&
+      !pthread_mutex_trylock (&vcm->app_event_queue->q->mutex))
     {
-      u32 i, n_to_dequeue = vcm->app_event_queue->cursize;
-      session_fifo_event_t e;
+      u32 i, n_to_dequeue = vcm->app_event_queue->q->cursize;
+      svm_msg_q_msg_t msg;
 
       for (i = 0; i < n_to_dequeue; i++)
-	svm_queue_sub_raw (vcm->app_event_queue, (u8 *) & e);
+	{
+	  svm_queue_sub_raw (vcm->app_event_queue->q, (u8 *) & msg);
+	  svm_msg_q_free_msg (vcm->app_event_queue, &msg);
+	}
 
-      pthread_mutex_unlock (&vcm->app_event_queue->mutex);
+      pthread_mutex_unlock (&vcm->app_event_queue->q->mutex);
     }
 done:
   return rv;
@@ -1184,8 +1187,7 @@ vppcom_session_write (uint32_t session_index, void *buf, size_t n)
 {
   vcl_session_t *session = 0;
   svm_fifo_t *tx_fifo = 0;
-  svm_queue_t *q;
-  session_fifo_event_t evt;
+  svm_msg_q_t *mq;
   session_state_t state;
   int rv, n_write, is_nonblocking;
   u32 poll_et;
@@ -1241,18 +1243,15 @@ vppcom_session_write (uint32_t session_index, void *buf, size_t n)
 
   if ((n_write > 0) && svm_fifo_set_event (tx_fifo))
     {
-      /* Fabricate TX event, send to vpp */
-      evt.fifo = tx_fifo;
-      evt.event_type = FIFO_EVENT_APP_TX;
-
+      /* Send TX event to vpp */
       VCL_SESSION_LOCK_AND_GET (session_index, &session);
-      q = session->vpp_evt_q;
-      ASSERT (q);
-      svm_queue_add (q, (u8 *) & evt, 0 /* do wait for mutex */ );
+      mq = session->vpp_evt_q;
+      ASSERT (mq);
+      app_send_io_evt_to_vpp (mq, tx_fifo, FIFO_EVENT_APP_TX, SVM_Q_WAIT);
       VCL_SESSION_UNLOCK ();
       VDBG (1, "VCL<%d>: vpp handle 0x%llx, sid %u: added FIFO_EVENT_APP_TX "
 	    "to vpp_event_q %p, n_write %d", getpid (),
-	    vpp_handle, session_index, q, n_write);
+	    vpp_handle, session_index, mq, n_write);
     }
 
   if (n_write <= 0)
diff --git a/src/vnet/session-apps/echo_client.c b/src/vnet/session-apps/echo_client.c
index 6ee91f9c54b..3d1af676186 100644
--- a/src/vnet/session-apps/echo_client.c
+++ b/src/vnet/session-apps/echo_client.c
@@ -62,13 +62,8 @@ send_data_chunk (echo_client_main_t * ecm, eclient_session_t * s)
 	  svm_fifo_t *f = s->data.tx_fifo;
 	  rv = clib_min (svm_fifo_max_enqueue (f), bytes_this_chunk);
 	  svm_fifo_enqueue_nocopy (f, rv);
-	  if (svm_fifo_set_event (f))
-	    {
-	      session_fifo_event_t evt;
-	      evt.fifo = f;
-	      evt.event_type = FIFO_EVENT_APP_TX;
-	      svm_queue_add (s->data.vpp_evt_q, (u8 *) & evt, 0);
-	    }
+	  session_send_io_evt_to_thread_custom (f, s->thread_index,
+						FIFO_EVENT_APP_TX);
 	}
       else
 	rv = app_send_stream (&s->data, test_data + test_buf_offset,
@@ -98,13 +93,8 @@ send_data_chunk (echo_client_main_t * ecm, eclient_session_t * s)
 	  hdr.lcl_port = at->lcl_port;
 	  svm_fifo_enqueue_nowait (f, sizeof (hdr), (u8 *) & hdr);
 	  svm_fifo_enqueue_nocopy (f, rv);
-	  if (svm_fifo_set_event (f))
-	    {
-	      session_fifo_event_t evt;
-	      evt.fifo = f;
-	      evt.event_type = FIFO_EVENT_APP_TX;
-	      svm_queue_add (s->data.vpp_evt_q, (u8 *) & evt, 0);
-	    }
+	  session_send_io_evt_to_thread_custom (f, s->thread_index,
+						FIFO_EVENT_APP_TX);
 	}
       else
 	rv = app_send_dgram (&s->data, test_data + test_buf_offset,
@@ -462,18 +452,9 @@ echo_clients_rx_callback (stream_session_t * s)
 
   if (svm_fifo_max_dequeue (s->server_rx_fifo))
     {
-      session_fifo_event_t evt;
-      svm_queue_t *q;
       if (svm_fifo_set_event (s->server_rx_fifo))
-	{
-	  evt.fifo = s->server_rx_fifo;
-	  evt.event_type = FIFO_EVENT_BUILTIN_RX;
-	  q = session_manager_get_vpp_event_queue (s->thread_index);
-	  if (PREDICT_FALSE (q->cursize == q->maxsize))
-	    clib_warning ("out of event queue space");
-	  else if (svm_queue_add (q, (u8 *) & evt, 0))
-	    clib_warning ("failed to enqueue self-tap");
-	}
+	session_send_io_evt_to_thread (s->server_rx_fifo,
+				       FIFO_EVENT_BUILTIN_RX);
     }
   return 0;
 }
diff --git a/src/vnet/session-apps/echo_client.h b/src/vnet/session-apps/echo_client.h
index f3fc8d2b59e..db5ba163628 100644
--- a/src/vnet/session-apps/echo_client.h
+++ b/src/vnet/session-apps/echo_client.h
@@ -46,7 +46,7 @@ typedef struct
    * Application setup parameters
    */
   svm_queue_t *vl_input_queue;		/**< vpe input queue */
-  svm_queue_t **vpp_event_queue;
+  svm_msg_q_t **vpp_event_queue;
 
   u32 cli_node_index;			/**< cli process node index */
   u32 my_client_index;			/**< loopback API client handle */
diff --git a/src/vnet/session-apps/echo_server.c b/src/vnet/session-apps/echo_server.c
index 7d1ae5a4d49..770f4ba7337 100644
--- a/src/vnet/session-apps/echo_server.c
+++ b/src/vnet/session-apps/echo_server.c
@@ -23,7 +23,7 @@ typedef struct
   /*
    * Server app parameters
    */
-  svm_queue_t **vpp_queue;
+  svm_msg_q_t **vpp_queue;
   svm_queue_t *vl_input_queue;	/**< Sever's event queue */
 
   u32 app_index;		/**< Server app index */
@@ -145,10 +145,8 @@ echo_server_rx_callback (stream_session_t * s)
   int actual_transfer;
   svm_fifo_t *tx_fifo, *rx_fifo;
   echo_server_main_t *esm = &echo_server_main;
-  session_fifo_event_t evt;
   u32 thread_index = vlib_get_thread_index ();
   app_session_transport_t at;
-  svm_queue_t *q;
 
   ASSERT (s->thread_index == thread_index);
 
@@ -170,8 +168,9 @@ echo_server_rx_callback (stream_session_t * s)
       max_dequeue = ph.data_length - ph.data_offset;
       if (!esm->vpp_queue[s->thread_index])
 	{
-	  q = session_manager_get_vpp_event_queue (s->thread_index);
-	  esm->vpp_queue[s->thread_index] = q;
+	  svm_msg_q_t *mq;
+	  mq = session_manager_get_vpp_event_queue (s->thread_index);
+	  esm->vpp_queue[s->thread_index] = mq;
 	}
       max_enqueue -= sizeof (session_dgram_hdr_t);
     }
@@ -191,13 +190,7 @@ echo_server_rx_callback (stream_session_t * s)
       /* Program self-tap to retry */
       if (svm_fifo_set_event (rx_fifo))
 	{
-	  evt.fifo = rx_fifo;
-	  evt.event_type = FIFO_EVENT_BUILTIN_RX;
-
-	  q = esm->vpp_queue[s->thread_index];
-	  if (PREDICT_FALSE (q->cursize == q->maxsize))
-	    clib_warning ("out of event queue space");
-	  else if (svm_queue_add (q, (u8 *) & evt, 0))
+	  if (session_send_io_evt_to_thread (rx_fifo, FIFO_EVENT_BUILTIN_RX))
 	    clib_warning ("failed to enqueue self-tap");
 
 	  vec_validate (esm->rx_retries[s->thread_index], s->session_index);
diff --git a/src/vnet/session-apps/http_server.c b/src/vnet/session-apps/http_server.c
index 9ad1297b901..ef9f760c992 100644
--- a/src/vnet/session-apps/http_server.c
+++ b/src/vnet/session-apps/http_server.c
@@ -32,7 +32,7 @@ typedef struct
 typedef struct
 {
   u8 **rx_buf;
-  svm_queue_t **vpp_queue;
+  svm_msg_q_t **vpp_queue;
   u64 byte_index;
 
   uword *handler_by_get_request;
@@ -140,7 +140,6 @@ http_cli_output (uword arg, u8 * buffer, uword buffer_bytes)
 void
 send_data (stream_session_t * s, u8 * data)
 {
-  session_fifo_event_t evt;
   u32 offset, bytes_to_send;
   f64 delay = 10e-3;
   http_server_main_t *hsm = &http_server_main;
@@ -178,14 +177,8 @@ send_data (stream_session_t * s, u8 * data)
 	  bytes_to_send -= actual_transfer;
 
 	  if (svm_fifo_set_event (s->server_tx_fifo))
-	    {
-	      /* Fabricate TX event, send to vpp */
-	      evt.fifo = s->server_tx_fifo;
-	      evt.event_type = FIFO_EVENT_APP_TX;
-
-	      svm_queue_add (hsm->vpp_queue[s->thread_index],
-			     (u8 *) & evt, 0 /* do wait for mutex */ );
-	    }
+	    session_send_io_evt_to_thread (s->server_tx_fifo,
+					   FIFO_EVENT_APP_TX);
 	  delay = 10e-3;
 	}
     }
@@ -379,15 +372,7 @@ http_server_rx_callback (stream_session_t * s)
 
   /* Send an RPC request via the thread-0 input node */
   if (vlib_get_thread_index () != 0)
-    {
-      session_fifo_event_t evt;
-      evt.rpc_args.fp = alloc_http_process_callback;
-      evt.rpc_args.arg = args;
-      evt.event_type = FIFO_EVENT_RPC;
-      svm_queue_add
-	(session_manager_get_vpp_event_queue (0 /* main thread */ ),
-	 (u8 *) & evt, 0 /* do wait for mutex */ );
-    }
+    session_send_rpc_evt_to_thread (0, alloc_http_process_callback, args);
   else
     alloc_http_process (args);
   return 0;
diff --git a/src/vnet/session-apps/proxy.c b/src/vnet/session-apps/proxy.c
index 78aa0de2840..6260ad350f0 100644
--- a/src/vnet/session-apps/proxy.c
+++ b/src/vnet/session-apps/proxy.c
@@ -194,7 +194,6 @@ proxy_rx_callback (stream_session_t * s)
   int proxy_index;
   uword *p;
   svm_fifo_t *active_open_tx_fifo;
-  session_fifo_event_t evt;
 
   ASSERT (s->thread_index == thread_index);
 
@@ -212,10 +211,9 @@ proxy_rx_callback (stream_session_t * s)
       if (svm_fifo_set_event (active_open_tx_fifo))
 	{
 	  u32 ao_thread_index = active_open_tx_fifo->master_thread_index;
-	  evt.fifo = active_open_tx_fifo;
-	  evt.event_type = FIFO_EVENT_APP_TX;
-	  if (svm_queue_add (pm->active_open_event_queue[ao_thread_index],
-			     (u8 *) & evt, 0 /* do wait for mutex */ ))
+	  if (session_send_io_evt_to_thread_custom (active_open_tx_fifo,
+						    ao_thread_index,
+						    FIFO_EVENT_APP_TX))
 	    clib_warning ("failed to enqueue tx evt");
 	}
     }
@@ -278,7 +276,6 @@ active_open_connected_callback (u32 app_index, u32 opaque,
   proxy_main_t *pm = &proxy_main;
   proxy_session_t *ps;
   u8 thread_index = vlib_get_thread_index ();
-  session_fifo_event_t evt;
 
   if (is_fail)
     {
@@ -320,15 +317,9 @@ active_open_connected_callback (u32 app_index, u32 opaque,
   /*
    * Send event for active open tx fifo
    */
+  ASSERT (s->thread_index == thread_index);
   if (svm_fifo_set_event (s->server_tx_fifo))
-    {
-      evt.fifo = s->server_tx_fifo;
-      evt.event_type = FIFO_EVENT_APP_TX;
-      if (svm_queue_add
-	  (pm->active_open_event_queue[thread_index], (u8 *) & evt,
-	   0 /* do wait for mutex */ ))
-	clib_warning ("failed to enqueue tx evt");
-    }
+    session_send_io_evt_to_thread (s->server_tx_fifo, FIFO_EVENT_APP_TX);
 
   return 0;
 }
@@ -354,8 +345,6 @@ active_open_disconnect_callback (stream_session_t * s)
 static int
 active_open_rx_callback (stream_session_t * s)
 {
-  proxy_main_t *pm = &proxy_main;
-  session_fifo_event_t evt;
   svm_fifo_t *proxy_tx_fifo;
 
   proxy_tx_fifo = s->server_rx_fifo;
@@ -365,12 +354,10 @@ active_open_rx_callback (stream_session_t * s)
    */
   if (svm_fifo_set_event (proxy_tx_fifo))
     {
-      u32 p_thread_index = proxy_tx_fifo->master_thread_index;
-      evt.fifo = proxy_tx_fifo;
-      evt.event_type = FIFO_EVENT_APP_TX;
-      if (svm_queue_add (pm->server_event_queue[p_thread_index], (u8 *) & evt,
-			 0 /* do wait for mutex */ ))
-	clib_warning ("failed to enqueue server rx evt");
+      u8 thread_index = proxy_tx_fifo->master_thread_index;
+      return session_send_io_evt_to_thread_custom (proxy_tx_fifo,
+						   thread_index,
+						   FIFO_EVENT_APP_TX);
     }
 
   return 0;
diff --git a/src/vnet/session-apps/proxy.h b/src/vnet/session-apps/proxy.h
index 4bca0a02cd8..c221a5e75f2 100644
--- a/src/vnet/session-apps/proxy.h
+++ b/src/vnet/session-apps/proxy.h
@@ -40,8 +40,8 @@ typedef struct
 {
   svm_queue_t *vl_input_queue;	/**< vpe input queue */
   /** per-thread vectors */
-  svm_queue_t **server_event_queue;
-  svm_queue_t **active_open_event_queue;
+  svm_msg_q_t **server_event_queue;
+  svm_msg_q_t **active_open_event_queue;
   u8 **rx_buf;				/**< intermediate rx buffers */
 
   u32 cli_node_index;			/**< cli process node index */
diff --git a/src/vnet/session/application.c b/src/vnet/session/application.c
index 60019dd7961..6041b49712d 100644
--- a/src/vnet/session/application.c
+++ b/src/vnet/session/application.c
@@ -807,6 +807,143 @@ application_get_segment_manager_properties (u32 app_index)
   return &app->sm_properties;
 }
 
+static inline int
+app_enqueue_evt (svm_msg_q_t * mq, svm_msg_q_msg_t * msg, u8 lock)
+{
+  if (PREDICT_TRUE (!svm_msg_q_is_full (mq)))
+    {
+      if (lock)
+	{
+	  svm_msg_q_add_w_lock (mq, msg);
+	  svm_msg_q_unlock (mq);
+	}
+      else if (svm_msg_q_add (mq, msg, SVM_Q_WAIT))
+	{
+	  clib_warning ("msg q add returned");
+	  if (lock)
+	    svm_msg_q_unlock (mq);
+	  return -1;
+	}
+    }
+  else
+    {
+      clib_warning ("evt q full");
+      svm_msg_q_free_msg (mq, msg);
+      if (lock)
+	svm_msg_q_unlock (mq);
+      return -1;
+    }
+  return 0;
+}
+
+static inline int
+app_send_io_evt_rx (application_t * app, stream_session_t * s, u8 lock)
+{
+  session_fifo_event_t *evt;
+  svm_msg_q_msg_t msg;
+  svm_msg_q_t *mq;
+
+  if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED))
+    {
+      /* Session is closed so app will never clean up. Flush rx fifo */
+      svm_fifo_dequeue_drop_all (s->server_rx_fifo);
+      return 0;
+    }
+
+  /* Built-in app? Hand event to the callback... */
+  if (app->cb_fns.builtin_app_rx_callback)
+    return app->cb_fns.builtin_app_rx_callback (s);
+
+  /* If no need for event, return */
+  if (!svm_fifo_set_event (s->server_rx_fifo))
+    return 0;
+
+  mq = app->event_queue;
+  if (lock)
+    svm_msg_q_lock (mq);
+
+  if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
+    {
+      clib_warning ("evt q rings full");
+      if (lock)
+	svm_msg_q_unlock (mq);
+      return -1;
+    }
+
+  msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+  ASSERT (!svm_msg_q_msg_is_invalid (&msg));
+
+  evt = (session_fifo_event_t *) svm_msg_q_msg_data (mq, &msg);
+  evt->fifo = s->server_rx_fifo;
+  evt->event_type = FIFO_EVENT_APP_RX;
+
+  return app_enqueue_evt (mq, &msg, lock);
+}
+
+static inline int
+app_send_io_evt_tx (application_t * app, stream_session_t * s, u8 lock)
+{
+  svm_msg_q_t *mq;
+  session_fifo_event_t *evt;
+  svm_msg_q_msg_t msg;
+
+  if (application_is_builtin (app))
+    return 0;
+
+  mq = app->event_queue;
+  if (lock)
+    svm_msg_q_lock (mq);
+
+  if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
+    {
+      clib_warning ("evt q rings full");
+      if (lock)
+	svm_msg_q_unlock (mq);
+      return -1;
+    }
+
+  msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+  ASSERT (!svm_msg_q_msg_is_invalid (&msg));
+
+  evt = (session_fifo_event_t *) svm_msg_q_msg_data (mq, &msg);
+  evt->event_type = FIFO_EVENT_APP_TX;
+  evt->fifo = s->server_tx_fifo;
+
+  return app_enqueue_evt (mq, &msg, lock);
+}
+
+/* *INDENT-OFF* */
+typedef int (app_send_evt_handler_fn) (application_t *app,
+				       stream_session_t *s,
+				       u8 lock);
+static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
+    app_send_io_evt_rx,
+    app_send_io_evt_tx,
+};
+/* *INDENT-ON* */
+
+/**
+ * Send event to application
+ *
+ * Logic from queue perspective is non-blocking. That is, if there's
+ * not enough space to enqueue a message, we return. However, if the lock
+ * flag is set, we do wait for queue mutex.
+ */
+int
+application_send_event (application_t * app, stream_session_t * s,
+			u8 evt_type)
+{
+  ASSERT (app && evt_type <= FIFO_EVENT_APP_TX);
+  return app_send_evt_handler_fns[evt_type] (app, s, 0 /* lock */ );
+}
+
+int
+application_lock_and_send_event (application_t * app, stream_session_t * s,
+				 u8 evt_type)
+{
+  return app_send_evt_handler_fns[evt_type] (app, s, 1 /* lock */ );
+}
+
 local_session_t *
 application_alloc_local_session (application_t * app)
 {
@@ -949,14 +1086,14 @@ application_local_session_connect (u32 table_index, application_t * client,
   svm_fifo_segment_private_t *seg;
   segment_manager_t *sm;
   local_session_t *ls;
-  svm_queue_t *sq, *cq;
+  svm_msg_q_t *sq, *cq;
 
   ls = application_alloc_local_session (server);
 
   props = application_segment_manager_properties (server);
   cprops = application_segment_manager_properties (client);
   evt_q_elts = props->evt_q_size + cprops->evt_q_size;
-  evt_q_sz = evt_q_elts * sizeof (session_fifo_event_t);
+  evt_q_sz = segment_manager_evt_q_expected_size (evt_q_elts);
   seg_size = props->rx_fifo_size + props->tx_fifo_size + evt_q_sz + margin;
 
   has_transport = session_has_transport ((stream_session_t *) ll);
diff --git a/src/vnet/session/application.h b/src/vnet/session/application.h
index aad7089ba6f..f6c81275826 100644
--- a/src/vnet/session/application.h
+++ b/src/vnet/session/application.h
@@ -72,7 +72,7 @@ typedef struct _application
   u32 ns_index;
 
   /** Application listens for events on this svm queue */
-  svm_queue_t *event_queue;
+  svm_msg_q_t *event_queue;
 
   /*
    * Callbacks: shoulder-taps for the server/client
@@ -207,6 +207,11 @@ int application_local_session_disconnect_w_index (u32 app_index,
 						  u32 ls_index);
 void application_local_sessions_del (application_t * app);
 
+int application_send_event (application_t * app, stream_session_t * s,
+			    u8 evt);
+int application_lock_and_send_event (application_t * app,
+				     stream_session_t * s, u8 evt_type);
+
 always_inline u32
 local_session_id (local_session_t * ll)
 {
diff --git a/src/vnet/session/application_interface.h b/src/vnet/session/application_interface.h
index 2c171487e70..50c043493f2 100644
--- a/src/vnet/session/application_interface.h
+++ b/src/vnet/session/application_interface.h
@@ -136,7 +136,7 @@ typedef enum
   _(IS_BUILTIN, "Application is builtin")			\
   _(IS_PROXY, "Application is proxying")			\
   _(USE_GLOBAL_SCOPE, "App can use global session scope")	\
-  _(USE_LOCAL_SCOPE, "App can use local session scope")
+  _(USE_LOCAL_SCOPE, "App can use local session scope")		\
 
 typedef enum _app_options
 {
@@ -187,7 +187,7 @@ typedef struct app_session_transport_
   _(volatile u8, session_state)		/**< session state */		\
   _(u32, session_index)			/**< index in owning pool */	\
   _(app_session_transport_t, transport)	/**< transport info */		\
-  _(svm_queue_t, *vpp_evt_q)		/**< vpp event queue  */	\
+  _(svm_msg_q_t, *vpp_evt_q)		/**< vpp event queue  */	\
   _(u8, is_dgram)			/**< flag for dgram mode */	\
 
 typedef struct
@@ -197,13 +197,73 @@ typedef struct
 #undef _
 } app_session_t;
 
+/**
+ * Send fifo io event to vpp worker thread
+ *
+ * Because there may be multiple writers to one of vpp's queues, this
+ * protects message allocation and enqueueing.
+ *
+ * @param mq		vpp message queue
+ * @param f		fifo for which the event is sent
+ * @param evt_type	type of event
+ * @param noblock	flag to indicate is request is blocking or not
+ * @return		0 if success, negative integer otherwise
+ */
+static inline int
+app_send_io_evt_to_vpp (svm_msg_q_t * mq, svm_fifo_t * f, u8 evt_type,
+			u8 noblock)
+{
+  session_fifo_event_t *evt;
+  svm_msg_q_msg_t msg;
+
+  if (noblock)
+    {
+      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_unlock (mq);
+	  return -2;
+	}
+      msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+      if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (&msg)))
+	{
+	  svm_msg_q_unlock (mq);
+	  return -2;
+	}
+      evt = (session_fifo_event_t *) svm_msg_q_msg_data (mq, &msg);
+      evt->fifo = f;
+      evt->event_type = evt_type;
+      svm_msg_q_add_w_lock (mq, &msg);
+      svm_msg_q_unlock (mq);
+      return 0;
+    }
+  else
+    {
+      svm_msg_q_lock (mq);
+      msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+      while (svm_msg_q_msg_is_invalid (&msg))
+	{
+	  svm_msg_q_wait (mq);
+	  msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+	}
+      evt = (session_fifo_event_t *) svm_msg_q_msg_data (mq, &msg);
+      evt->fifo = f;
+      evt->event_type = evt_type;
+      if (svm_msg_q_is_full (mq))
+	svm_msg_q_wait (mq);
+      svm_msg_q_add_w_lock (mq, &msg);
+      svm_msg_q_unlock (mq);
+      return 0;
+    }
+}
+
 always_inline int
 app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at,
-		    svm_queue_t * vpp_evt_q, u8 * data, u32 len, u8 noblock)
+		    svm_msg_q_t * vpp_evt_q, u8 * data, u32 len, u8 noblock)
 {
   u32 max_enqueue, actual_write;
   session_dgram_hdr_t hdr;
-  session_fifo_event_t evt;
   int rv;
 
   max_enqueue = svm_fifo_max_enqueue (f);
@@ -225,11 +285,7 @@ app_send_dgram_raw (svm_fifo_t * f, app_session_transport_t * at,
   if ((rv = svm_fifo_enqueue_nowait (f, actual_write, data)) > 0)
     {
       if (svm_fifo_set_event (f))
-	{
-	  evt.fifo = f;
-	  evt.event_type = FIFO_EVENT_APP_TX;
-	  svm_queue_add (vpp_evt_q, (u8 *) & evt, noblock);
-	}
+	app_send_io_evt_to_vpp (vpp_evt_q, f, FIFO_EVENT_APP_TX, noblock);
     }
   ASSERT (rv);
   return rv;
@@ -243,20 +299,15 @@ app_send_dgram (app_session_t * s, u8 * data, u32 len, u8 noblock)
 }
 
 always_inline int
-app_send_stream_raw (svm_fifo_t * f, svm_queue_t * vpp_evt_q, u8 * data,
+app_send_stream_raw (svm_fifo_t * f, svm_msg_q_t * vpp_evt_q, u8 * data,
 		     u32 len, u8 noblock)
 {
-  session_fifo_event_t evt;
   int rv;
 
   if ((rv = svm_fifo_enqueue_nowait (f, len, data)) > 0)
     {
       if (svm_fifo_set_event (f))
-	{
-	  evt.fifo = f;
-	  evt.event_type = FIFO_EVENT_APP_TX;
-	  svm_queue_add (vpp_evt_q, (u8 *) & evt, noblock);
-	}
+	app_send_io_evt_to_vpp (vpp_evt_q, f, FIFO_EVENT_APP_TX, noblock);
     }
   return rv;
 }
diff --git a/src/vnet/session/segment_manager.c b/src/vnet/session/segment_manager.c
index 31bb0c3aab5..b00bcd5bbc8 100644
--- a/src/vnet/session/segment_manager.c
+++ b/src/vnet/session/segment_manager.c
@@ -599,25 +599,52 @@ segment_manager_dealloc_fifos (u32 segment_index, svm_fifo_t * rx_fifo,
     segment_manager_segment_reader_unlock (sm);
 }
 
+u32
+segment_manager_evt_q_expected_size (u32 q_len)
+{
+  u32 fifo_evt_size, notif_q_size, q_hdrs;
+  u32 msg_q_sz, fifo_evt_ring_sz, session_ntf_ring_sz;
+
+  fifo_evt_size = 1 << max_log2 (sizeof (session_fifo_event_t));
+  notif_q_size = clib_max (16, q_len >> 4);
+
+  msg_q_sz = q_len * sizeof (svm_msg_q_msg_t);
+  fifo_evt_ring_sz = q_len * fifo_evt_size;
+  session_ntf_ring_sz = notif_q_size * 256;
+  q_hdrs = sizeof (svm_queue_t) + sizeof (svm_msg_q_t);
+
+  return (msg_q_sz + fifo_evt_ring_sz + session_ntf_ring_sz + q_hdrs);
+}
+
 /**
  * Allocates shm queue in the first segment
  *
  * Must be called with lock held
  */
-svm_queue_t *
+svm_msg_q_t *
 segment_manager_alloc_queue (svm_fifo_segment_private_t * segment,
 			     u32 queue_size)
 {
-  ssvm_shared_header_t *sh;
-  svm_queue_t *q;
+  u32 fifo_evt_size, session_evt_size = 256, notif_q_size;
+  svm_msg_q_cfg_t _cfg, *cfg = &_cfg;
+  svm_msg_q_t *q;
   void *oldheap;
 
-  sh = segment->ssvm.sh;
+  fifo_evt_size = sizeof (session_fifo_event_t);
+  notif_q_size = clib_max (16, queue_size >> 4);
+  /* *INDENT-OFF* */
+  svm_msg_q_ring_cfg_t rc[SESSION_MQ_N_RINGS] = {
+    {queue_size, fifo_evt_size, 0},
+    {notif_q_size, session_evt_size, 0}
+  };
+  /* *INDENT-ON* */
+  cfg->consumer_pid = 0;
+  cfg->n_rings = 2;
+  cfg->q_nitems = queue_size;
+  cfg->ring_cfgs = rc;
 
-  oldheap = ssvm_push_heap (sh);
-  q = svm_queue_init (queue_size, sizeof (session_fifo_event_t),
-		      0 /* consumer pid */ ,
-		      0 /* signal when queue non-empty */ );
+  oldheap = ssvm_push_heap (segment->ssvm.sh);
+  q = svm_msg_q_alloc (cfg);
   ssvm_pop_heap (oldheap);
   return q;
 }
diff --git a/src/vnet/session/segment_manager.h b/src/vnet/session/segment_manager.h
index 62e5e97e703..73cb4827a8b 100644
--- a/src/vnet/session/segment_manager.h
+++ b/src/vnet/session/segment_manager.h
@@ -17,7 +17,7 @@
 
 #include <vnet/vnet.h>
 #include <svm/svm_fifo_segment.h>
-#include <svm/queue.h>
+#include <svm/message_queue.h>
 #include <vlibmemory/api.h>
 #include <vppinfra/lock.h>
 #include <vppinfra/valloc.h>
@@ -61,7 +61,7 @@ typedef struct _segment_manager
   /**
    * App event queue allocated in first segment
    */
-  svm_queue_t *event_queue;
+  svm_msg_q_t *event_queue;
 } segment_manager_t;
 
 #define segment_manager_foreach_segment_w_lock(VAR, SM, BODY)		\
@@ -114,7 +114,7 @@ segment_manager_index (segment_manager_t * sm)
   return sm - segment_manager_main.segment_managers;
 }
 
-always_inline svm_queue_t *
+always_inline svm_msg_q_t *
 segment_manager_event_queue (segment_manager_t * sm)
 {
   return sm->event_queue;
@@ -152,7 +152,8 @@ int segment_manager_try_alloc_fifos (svm_fifo_segment_private_t * fs,
 				     svm_fifo_t ** tx_fifo);
 void segment_manager_dealloc_fifos (u32 segment_index, svm_fifo_t * rx_fifo,
 				    svm_fifo_t * tx_fifo);
-svm_queue_t *segment_manager_alloc_queue (svm_fifo_segment_private_t * fs,
+u32 segment_manager_evt_q_expected_size (u32 q_size);
+svm_msg_q_t *segment_manager_alloc_queue (svm_fifo_segment_private_t * fs,
 					  u32 queue_size);
 void segment_manager_dealloc_queue (segment_manager_t * sm, svm_queue_t * q);
 void segment_manager_app_detach (segment_manager_t * sm);
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 26bc70e6fd7..38a0521af94 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -27,49 +27,90 @@
 session_manager_main_t session_manager_main;
 extern transport_proto_vft_t *tp_vfts;
 
-static void
-session_send_evt_to_thread (u64 session_handle, fifo_event_type_t evt_type,
-			    u32 thread_index, void *fp, void *rpc_args)
+static inline int
+session_send_evt_to_thread (void *data, void *args, u32 thread_index,
+			    session_evt_type_t evt_type)
 {
-  session_fifo_event_t evt = { {0}, };
-  svm_queue_t *q;
+  session_fifo_event_t *evt;
+  svm_msg_q_msg_t msg;
+  svm_msg_q_t *mq;
   u32 tries = 0, max_tries;
 
-  evt.event_type = evt_type;
-  if (evt_type == FIFO_EVENT_RPC)
-    {
-      evt.rpc_args.fp = fp;
-      evt.rpc_args.arg = rpc_args;
-    }
-  else
-    evt.session_handle = session_handle;
-
-  q = session_manager_get_vpp_event_queue (thread_index);
-  while (svm_queue_add (q, (u8 *) & evt, 1))
+  mq = session_manager_get_vpp_event_queue (thread_index);
+  while (svm_msg_q_try_lock (mq))
     {
       max_tries = vlib_get_current_process (vlib_get_main ())? 1e6 : 3;
       if (tries++ == max_tries)
 	{
 	  SESSION_DBG ("failed to enqueue evt");
-	  break;
+	  return -1;
 	}
     }
+  if (PREDICT_FALSE (svm_msg_q_ring_is_full (mq, SESSION_MQ_IO_EVT_RING)))
+    {
+      svm_msg_q_unlock (mq);
+      return -2;
+    }
+  msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+  if (PREDICT_FALSE (svm_msg_q_msg_is_invalid (&msg)))
+    {
+      svm_msg_q_unlock (mq);
+      return -2;
+    }
+  evt = (session_fifo_event_t *) svm_msg_q_msg_data (mq, &msg);
+  evt->event_type = evt_type;
+  switch (evt_type)
+    {
+    case FIFO_EVENT_RPC:
+      evt->rpc_args.fp = data;
+      evt->rpc_args.arg = args;
+      break;
+    case FIFO_EVENT_APP_TX:
+    case FIFO_EVENT_BUILTIN_RX:
+      evt->fifo = data;
+      break;
+    case FIFO_EVENT_DISCONNECT:
+      evt->session_handle = session_handle ((stream_session_t *) data);
+      break;
+    default:
+      clib_warning ("evt unhandled!");
+      svm_msg_q_unlock (mq);
+      return -1;
+    }
+
+  svm_msg_q_add_w_lock (mq, &msg);
+  svm_msg_q_unlock (mq);
+  return 0;
 }
 
-void
-session_send_session_evt_to_thread (u64 session_handle,
-				    fifo_event_type_t evt_type,
-				    u32 thread_index)
+int
+session_send_io_evt_to_thread (svm_fifo_t * f, session_evt_type_t evt_type)
+{
+  return session_send_evt_to_thread (f, 0, f->master_thread_index, evt_type);
+}
+
+int
+session_send_io_evt_to_thread_custom (svm_fifo_t * f, u32 thread_index,
+				      session_evt_type_t evt_type)
 {
-  session_send_evt_to_thread (session_handle, evt_type, thread_index, 0, 0);
+  return session_send_evt_to_thread (f, 0, thread_index, evt_type);
+}
+
+int
+session_send_ctrl_evt_to_thread (stream_session_t * s,
+				 session_evt_type_t evt_type)
+{
+  /* only event supported for now is disconnect */
+  ASSERT (evt_type == FIFO_EVENT_DISCONNECT);
+  return session_send_evt_to_thread (s, 0, s->thread_index,
+				     FIFO_EVENT_DISCONNECT);
 }
 
 void
 session_send_rpc_evt_to_thread (u32 thread_index, void *fp, void *rpc_args)
 {
   if (thread_index != vlib_get_thread_index ())
-    session_send_evt_to_thread (0, FIFO_EVENT_RPC, thread_index, fp,
-				rpc_args);
+    session_send_evt_to_thread (fp, rpc_args, thread_index, FIFO_EVENT_RPC);
   else
     {
       void (*fnp) (void *) = fp;
@@ -440,24 +481,15 @@ stream_session_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
 /**
  * Notify session peer that new data has been enqueued.
  *
- * @param s Stream session for which the event is to be generated.
- * @param block Flag to indicate if call should block if event queue is full.
+ * @param s 	Stream session for which the event is to be generated.
+ * @param lock 	Flag to indicate if call should lock message queue.
  *
- * @return 0 on succes or negative number if failed to send notification.
+ * @return 0 on success or negative number if failed to send notification.
  */
-static int
-session_enqueue_notify (stream_session_t * s, u8 block)
+static inline int
+session_enqueue_notify (stream_session_t * s, u8 lock)
 {
   application_t *app;
-  session_fifo_event_t evt;
-  svm_queue_t *q;
-
-  if (PREDICT_FALSE (s->session_state == SESSION_STATE_CLOSED))
-    {
-      /* Session is closed so app will never clean up. Flush rx fifo */
-      svm_fifo_dequeue_drop_all (s->server_rx_fifo);
-      return 0;
-    }
 
   app = application_get_if_valid (s->app_index);
   if (PREDICT_FALSE (app == 0))
@@ -466,68 +498,32 @@ session_enqueue_notify (stream_session_t * s, u8 block)
       return 0;
     }
 
-  /* Built-in app? Hand event to the callback... */
-  if (app->cb_fns.builtin_app_rx_callback)
-    return app->cb_fns.builtin_app_rx_callback (s);
-
-  /* If no event, send one */
-  if (svm_fifo_set_event (s->server_rx_fifo))
-    {
-      /* Fabricate event */
-      evt.fifo = s->server_rx_fifo;
-      evt.event_type = FIFO_EVENT_APP_RX;
-
-      /* Add event to server's event queue */
-      q = app->event_queue;
-
-      /* Based on request block (or not) for lack of space */
-      if (block || PREDICT_TRUE (q->cursize < q->maxsize))
-	svm_queue_add (app->event_queue, (u8 *) & evt,
-		       0 /* do wait for mutex */ );
-      else
-	{
-	  clib_warning ("fifo full");
-	  return -1;
-	}
-    }
-
   /* *INDENT-OFF* */
   SESSION_EVT_DBG(SESSION_EVT_ENQ, s, ({
-      ed->data[0] = evt.event_type;
+      ed->data[0] = FIFO_EVENT_APP_RX;
       ed->data[1] = svm_fifo_max_dequeue (s->server_rx_fifo);
   }));
   /* *INDENT-ON* */
 
-  return 0;
+  if (lock)
+    return application_lock_and_send_event (app, s, FIFO_EVENT_APP_RX);
+
+  return application_send_event (app, s, FIFO_EVENT_APP_RX);
 }
 
 int
 session_dequeue_notify (stream_session_t * s)
 {
   application_t *app;
-  svm_queue_t *q;
 
   app = application_get_if_valid (s->app_index);
   if (PREDICT_FALSE (!app))
     return -1;
 
-  if (application_is_builtin (app))
-    return 0;
+  if (session_transport_service_type (s) == TRANSPORT_SERVICE_CL)
+    return application_lock_and_send_event (app, s, FIFO_EVENT_APP_RX);
 
-  q = app->event_queue;
-  if (PREDICT_TRUE (q->cursize < q->maxsize))
-    {
-      session_fifo_event_t evt = {
-	.event_type = FIFO_EVENT_APP_TX,
-	.fifo = s->server_tx_fifo
-      };
-      svm_queue_add (app->event_queue, (u8 *) & evt, SVM_Q_WAIT);
-    }
-  else
-    {
-      return -1;
-    }
-  return 0;
+  return application_send_event (app, s, FIFO_EVENT_APP_TX);
 }
 
 /**
@@ -542,16 +538,24 @@ int
 session_manager_flush_enqueue_events (u8 transport_proto, u32 thread_index)
 {
   session_manager_main_t *smm = &session_manager_main;
-  u32 *indices;
+  transport_service_type_t tp_service;
+  int i, errors = 0, lock;
   stream_session_t *s;
-  int i, errors = 0;
+  u32 *indices;
 
   indices = smm->session_to_enqueue[transport_proto][thread_index];
+  tp_service = transport_protocol_service_type (transport_proto);
+  lock = tp_service == TRANSPORT_SERVICE_CL;
 
   for (i = 0; i < vec_len (indices); i++)
     {
       s = session_get_if_valid (indices[i], thread_index);
-      if (s == 0 || session_enqueue_notify (s, 0 /* don't block */ ))
+      if (PREDICT_FALSE (!s))
+	{
+	  errors++;
+	  continue;
+	}
+      if (PREDICT_FALSE (session_enqueue_notify (s, lock)))
 	errors++;
     }
 
@@ -1118,9 +1122,7 @@ stream_session_disconnect (stream_session_t * s)
       evt->event_type = FIFO_EVENT_DISCONNECT;
     }
   else
-    session_send_session_evt_to_thread (session_handle (s),
-					FIFO_EVENT_DISCONNECT,
-					s->thread_index);
+    session_send_ctrl_evt_to_thread (s, FIFO_EVENT_DISCONNECT);
 }
 
 /**
@@ -1231,8 +1233,18 @@ session_vpp_event_queues_allocate (session_manager_main_t * smm)
 
   for (i = 0; i < vec_len (smm->vpp_event_queues); i++)
     {
-      smm->vpp_event_queues[i] = svm_queue_init (evt_q_length, evt_size,
-						 vpp_pid, 0);
+      svm_msg_q_cfg_t _cfg, *cfg = &_cfg;
+      u32 notif_q_size = clib_max (16, evt_q_length >> 4);
+      svm_msg_q_ring_cfg_t rc[SESSION_MQ_N_RINGS] = {
+	{evt_q_length, evt_size, 0}
+	,
+	{notif_q_size, 256, 0}
+      };
+      cfg->consumer_pid = 0;
+      cfg->n_rings = 2;
+      cfg->q_nitems = evt_q_length;
+      cfg->ring_cfgs = rc;
+      smm->vpp_event_queues[i] = svm_msg_q_alloc (cfg);
     }
 
   if (smm->evt_qs_use_memfd_seg)
diff --git a/src/vnet/session/session.h b/src/vnet/session/session.h
index fe3477b63e3..879b3823e5d 100644
--- a/src/vnet/session/session.h
+++ b/src/vnet/session/session.h
@@ -38,10 +38,10 @@ typedef enum
   FIFO_EVENT_DISCONNECT,
   FIFO_EVENT_BUILTIN_RX,
   FIFO_EVENT_RPC,
-} fifo_event_type_t;
+} session_evt_type_t;
 
 static inline const char *
-fifo_event_type_str (fifo_event_type_t et)
+fifo_event_type_str (session_evt_type_t et)
 {
   switch (et)
     {
@@ -62,6 +62,13 @@ fifo_event_type_str (fifo_event_type_t et)
     }
 }
 
+typedef enum
+{
+  SESSION_MQ_IO_EVT_RING,
+  SESSION_MQ_CTRL_EVT_RING,
+  SESSION_MQ_N_RINGS
+} session_mq_rings_e;
+
 #define foreach_session_input_error                                    	\
 _(NO_SESSION, "No session drops")                                       \
 _(NO_LISTENER, "No listener for dst port drops")                        \
@@ -86,23 +93,30 @@ typedef struct
 {
   void *fp;
   void *arg;
-} rpc_args_t;
+} session_rpc_args_t;
 
 typedef u64 session_handle_t;
 
 /* *INDENT-OFF* */
-typedef CLIB_PACKED (struct {
+typedef struct
+{
+  u8 event_type;
+  u8 postponed;
   union
+  {
+    svm_fifo_t *fifo;
+    session_handle_t session_handle;
+    session_rpc_args_t rpc_args;
+    struct
     {
-      svm_fifo_t * fifo;
-      session_handle_t session_handle;
-      rpc_args_t rpc_args;
+      u8 data[0];
     };
-  u8 event_type;
-  u8 postponed;
-}) session_fifo_event_t;
+  };
+} __clib_packed session_fifo_event_t;
 /* *INDENT-ON* */
 
+#define SESSION_MSG_NULL { }
+
 typedef struct session_dgram_pre_hdr_
 {
   u32 data_length;
@@ -193,7 +207,7 @@ struct _session_manager_main
   session_tx_context_t *ctx;
 
   /** vpp fifo event queue */
-  svm_queue_t **vpp_event_queues;
+  svm_msg_q_t **vpp_event_queues;
 
   /** Event queues memfd segment initialized only if so configured */
   ssvm_private_t evt_qs_segment;
@@ -533,9 +547,12 @@ int stream_session_stop_listen (stream_session_t * s);
 void stream_session_disconnect (stream_session_t * s);
 void stream_session_disconnect_transport (stream_session_t * s);
 void stream_session_cleanup (stream_session_t * s);
-void session_send_session_evt_to_thread (u64 session_handle,
-					 fifo_event_type_t evt_type,
-					 u32 thread_index);
+int session_send_io_evt_to_thread (svm_fifo_t * f,
+				   session_evt_type_t evt_type);
+int session_send_io_evt_to_thread_custom (svm_fifo_t * f, u32 thread_index,
+					  session_evt_type_t evt_type);
+void session_send_rpc_evt_to_thread (u32 thread_index, void *fp,
+				     void *rpc_args);
 ssvm_private_t *session_manager_get_evt_q_segment (void);
 
 u8 *format_stream_session (u8 * s, va_list * args);
@@ -549,7 +566,7 @@ void session_register_transport (transport_proto_t transport_proto,
 
 clib_error_t *vnet_session_enable_disable (vlib_main_t * vm, u8 is_en);
 
-always_inline svm_queue_t *
+always_inline svm_msg_q_t *
 session_manager_get_vpp_event_queue (u32 thread_index)
 {
   return session_manager_main.vpp_event_queues[thread_index];
diff --git a/src/vnet/session/session_api.c b/src/vnet/session/session_api.c
index 1a41dbd9e08..f9fddea6148 100755
--- a/src/vnet/session/session_api.c
+++ b/src/vnet/session/session_api.c
@@ -155,7 +155,7 @@ send_session_accept_callback (stream_session_t * s)
   vl_api_registration_t *reg;
   transport_connection_t *tc;
   stream_session_t *listener;
-  svm_queue_t *vpp_queue;
+  svm_msg_q_t *vpp_queue;
 
   reg = vl_mem_api_client_index_to_registration (server->api_client_index);
   if (!reg)
@@ -300,7 +300,7 @@ send_session_connected_callback (u32 app_index, u32 api_context,
   vl_api_connect_session_reply_t *mp;
   transport_connection_t *tc;
   vl_api_registration_t *reg;
-  svm_queue_t *vpp_queue;
+  svm_msg_q_t *vpp_queue;
   application_t *app;
 
   app = application_get (app_index);
@@ -485,7 +485,7 @@ vl_api_bind_uri_t_handler (vl_api_bind_uri_t * mp)
   vl_api_bind_uri_reply_t *rmp;
   stream_session_t *s;
   application_t *app = 0;
-  svm_queue_t *vpp_evt_q;
+  svm_msg_q_t *vpp_evt_q;
   int rv;
 
   if (session_manager_is_enabled () == 0)
@@ -759,7 +759,7 @@ vl_api_bind_sock_t_handler (vl_api_bind_sock_t * mp)
   stream_session_t *s;
   transport_connection_t *tc = 0;
   ip46_address_t *ip46;
-  svm_queue_t *vpp_evt_q;
+  svm_msg_q_t *vpp_evt_q;
 
   if (session_manager_is_enabled () == 0)
     {
diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 85fd28db7a9..350282bd902 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -575,15 +575,14 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 		       vlib_frame_t * frame)
 {
   session_manager_main_t *smm = vnet_get_session_manager_main ();
+  u32 thread_index = vm->thread_index, n_to_dequeue, n_events;
   session_fifo_event_t *pending_events, *e;
   session_fifo_event_t *fifo_events;
-  u32 n_to_dequeue, n_events;
-  svm_queue_t *q;
-  application_t *app;
-  int n_tx_packets = 0;
-  u32 thread_index = vm->thread_index;
-  int i, rv;
+  svm_msg_q_msg_t _msg, *msg = &_msg;
   f64 now = vlib_time_now (vm);
+  int n_tx_packets = 0, i, rv;
+  application_t *app;
+  svm_msg_q_t *mq;
   void (*fp) (void *);
 
   SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, thread_index);
@@ -594,16 +593,11 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   transport_update_time (now, thread_index);
 
   /*
-   * Get vpp queue events
+   * Get vpp queue events that we can dequeue without blocking
    */
-  q = smm->vpp_event_queues[thread_index];
-  if (PREDICT_FALSE (q == 0))
-    return 0;
-
+  mq = smm->vpp_event_queues[thread_index];
   fifo_events = smm->free_event_vector[thread_index];
-
-  /* min number of events we can dequeue without blocking */
-  n_to_dequeue = q->cursize;
+  n_to_dequeue = svm_msg_q_size (mq);
   pending_events = smm->pending_event_vector[thread_index];
 
   if (!n_to_dequeue && !vec_len (pending_events)
@@ -624,21 +618,19 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
     }
 
   /* See you in the next life, don't be late
-   * XXX: we may need priorities here
-   */
-  if (pthread_mutex_trylock (&q->mutex))
+   * XXX: we may need priorities here */
+  if (svm_msg_q_try_lock (mq))
     return 0;
 
   for (i = 0; i < n_to_dequeue; i++)
     {
       vec_add2 (fifo_events, e, 1);
-      svm_queue_sub_raw (q, (u8 *) e);
+      svm_msg_q_sub_w_lock (mq, msg);
+      clib_memcpy (e, svm_msg_q_msg_data (mq, msg), sizeof (*e));
+      svm_msg_q_free_msg (mq, msg);
     }
 
-  /* The other side of the connection is not polling */
-  if (q->cursize < (q->maxsize / 8))
-    (void) pthread_cond_broadcast (&q->condvar);
-  pthread_mutex_unlock (&q->mutex);
+  svm_msg_q_unlock (mq);
 
   vec_append (fifo_events, pending_events);
   vec_append (fifo_events, smm->pending_disconnects[thread_index]);
@@ -760,19 +752,20 @@ dump_thread_0_event_queue (void)
   vlib_main_t *vm = &vlib_global_main;
   u32 my_thread_index = vm->thread_index;
   session_fifo_event_t _e, *e = &_e;
+  svm_msg_q_ring_t *ring;
   stream_session_t *s0;
+  svm_msg_q_msg_t *msg;
+  svm_msg_q_t *mq;
   int i, index;
-  i8 *headp;
-
-  svm_queue_t *q;
-  q = smm->vpp_event_queues[my_thread_index];
 
-  index = q->head;
+  mq = smm->vpp_event_queues[my_thread_index];
+  index = mq->q->head;
 
-  for (i = 0; i < q->cursize; i++)
+  for (i = 0; i < mq->q->cursize; i++)
     {
-      headp = (i8 *) (&q->data[0] + q->elsize * index);
-      clib_memcpy (e, headp, q->elsize);
+      msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
+      ring = svm_msg_q_ring (mq, msg->ring_index);
+      clib_memcpy (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
 
       switch (e->event_type)
 	{
@@ -805,7 +798,7 @@ dump_thread_0_event_queue (void)
 
       index++;
 
-      if (index == q->maxsize)
+      if (index == mq->q->maxsize)
 	index = 0;
     }
 }
@@ -844,10 +837,11 @@ u8
 session_node_lookup_fifo_event (svm_fifo_t * f, session_fifo_event_t * e)
 {
   session_manager_main_t *smm = vnet_get_session_manager_main ();
-  svm_queue_t *q;
+  svm_msg_q_t *mq;
   session_fifo_event_t *pending_event_vector, *evt;
   int i, index, found = 0;
-  i8 *headp;
+  svm_msg_q_msg_t *msg;
+  svm_msg_q_ring_t *ring;
   u8 thread_index;
 
   ASSERT (e);
@@ -855,16 +849,17 @@ session_node_lookup_fifo_event (svm_fifo_t * f, session_fifo_event_t * e)
   /*
    * Search evt queue
    */
-  q = smm->vpp_event_queues[thread_index];
-  index = q->head;
-  for (i = 0; i < q->cursize; i++)
+  mq = smm->vpp_event_queues[thread_index];
+  index = mq->q->head;
+  for (i = 0; i < mq->q->cursize; i++)
     {
-      headp = (i8 *) (&q->data[0] + q->elsize * index);
-      clib_memcpy (e, headp, q->elsize);
+      msg = (svm_msg_q_msg_t *) (&mq->q->data[0] + mq->q->elsize * index);
+      ring = svm_msg_q_ring (mq, msg->ring_index);
+      clib_memcpy (e, svm_msg_q_msg_data (mq, msg), ring->elsize);
       found = session_node_cmp_event (e, f);
       if (found)
 	return 1;
-      if (++index == q->maxsize)
+      if (++index == mq->q->maxsize)
 	index = 0;
     }
   /*
diff --git a/src/vnet/tls/tls.c b/src/vnet/tls/tls.c
index 492b3cc7a2a..cea449163f8 100644
--- a/src/vnet/tls/tls.c
+++ b/src/vnet/tls/tls.c
@@ -41,63 +41,15 @@ tls_get_available_engine (void)
 int
 tls_add_vpp_q_evt (svm_fifo_t * f, u8 evt_type)
 {
-  session_fifo_event_t evt;
-  svm_queue_t *q;
-
   if (svm_fifo_set_event (f))
-    {
-      evt.fifo = f;
-      evt.event_type = evt_type;
-
-      q = session_manager_get_vpp_event_queue (f->master_thread_index);
-      if (PREDICT_TRUE (q->cursize < q->maxsize))
-	{
-	  svm_queue_add (q, (u8 *) & evt, 0 /* do wait for mutex */ );
-	}
-      else
-	{
-	  clib_warning ("vpp's evt q full");
-	  return -1;
-	}
-    }
+    session_send_io_evt_to_thread (f, evt_type);
   return 0;
 }
 
 static inline int
 tls_add_app_q_evt (application_t * app, stream_session_t * app_session)
 {
-  session_fifo_event_t evt;
-  svm_queue_t *q;
-
-  if (PREDICT_FALSE (app_session->session_state == SESSION_STATE_CLOSED))
-    {
-      /* Session is closed so app will never clean up. Flush rx fifo */
-      u32 to_dequeue = svm_fifo_max_dequeue (app_session->server_rx_fifo);
-      if (to_dequeue)
-	svm_fifo_dequeue_drop (app_session->server_rx_fifo, to_dequeue);
-      return 0;
-    }
-
-  if (app->cb_fns.builtin_app_rx_callback)
-    return app->cb_fns.builtin_app_rx_callback (app_session);
-
-  if (svm_fifo_set_event (app_session->server_rx_fifo))
-    {
-      evt.fifo = app_session->server_rx_fifo;
-      evt.event_type = FIFO_EVENT_APP_RX;
-      q = app->event_queue;
-
-      if (PREDICT_TRUE (q->cursize < q->maxsize))
-	{
-	  svm_queue_add (q, (u8 *) & evt, 0 /* do wait for mutex */ );
-	}
-      else
-	{
-	  clib_warning ("app evt q full");
-	  return -1;
-	}
-    }
-  return 0;
+  return application_send_event (app, app_session, FIFO_EVENT_APP_RX);
 }
 
 u32