1 files changed, 43 insertions, 38 deletions

diff --git a/src/vcl/vppcom.c b/src/vcl/vppcom.c
index 3abde98288a..fa37a1da68f 100644
--- a/src/vcl/vppcom.c
+++ b/src/vcl/vppcom.c
@@ -649,7 +649,7 @@ vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
     {
     case SESSION_IO_EVT_RX:
     case SESSION_IO_EVT_TX:
-      session = vcl_session_get (wrk, e->fifo->client_session_index);
+      session = vcl_session_get (wrk, e->session_index);
       if (!session || !(session->session_state & STATE_OPEN))
 	break;
       vec_add1 (wrk->unhandled_evts_vector, *e);
@@ -1491,8 +1491,7 @@ vppcom_session_connect (uint32_t session_handle, vppcom_endpt_t * server_ep)
 static u8
 vcl_is_rx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
 {
-  return (e->event_type == SESSION_IO_EVT_RX
-	  && e->fifo->client_session_index == sid);
+  return (e->event_type == SESSION_IO_EVT_RX && e->session_index == sid);
 }
 
 static inline int
@@ -1529,20 +1528,19 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
   rx_fifo = is_ct ? s->ct_rx_fifo : s->rx_fifo;
   s->has_rx_evt = 0;
 
-  if (is_ct)
-    svm_fifo_unset_event (s->rx_fifo);
-
   if (svm_fifo_is_empty (rx_fifo))
     {
-      svm_fifo_unset_event (rx_fifo);
       if (is_nonblocking)
-	return VPPCOM_EWOULDBLOCK;
+	{
+	  svm_fifo_unset_event (s->rx_fifo);
+	  return VPPCOM_EWOULDBLOCK;
+	}
       while (svm_fifo_is_empty (rx_fifo))
 	{
 	  if (vcl_session_is_closing (s))
 	    return vcl_session_closing_error (s);
 
-	  svm_fifo_unset_event (rx_fifo);
+	  svm_fifo_unset_event (s->rx_fifo);
 	  svm_msg_q_lock (mq);
 	  if (svm_msg_q_is_empty (mq))
 	    svm_msg_q_wait (mq);
@@ -1551,10 +1549,7 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
 	  e = svm_msg_q_msg_data (mq, &msg);
 	  svm_msg_q_unlock (mq);
 	  if (!vcl_is_rx_evt_for_session (e, s->session_index, is_ct))
-	    {
-	      clib_warning ("THIS ONE type %u", e->event_type);
-	      vcl_handle_mq_event (wrk, e);
-	    }
+	    vcl_handle_mq_event (wrk, e);
 	  svm_msg_q_free_msg (mq, &msg);
 	}
     }
@@ -1565,7 +1560,7 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
     n_read = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
 
   if (svm_fifo_is_empty (rx_fifo))
-    svm_fifo_unset_event (rx_fifo);
+    svm_fifo_unset_event (s->rx_fifo);
 
   VDBG (2, "vpp handle 0x%llx, sid %u: read %d bytes from (%p)",
 	s->vpp_handle, session_handle, n_read, rx_fifo);
@@ -1676,8 +1671,7 @@ vppcom_data_segment_copy (void *buf, vppcom_data_segments_t ds, u32 max_bytes)
 static u8
 vcl_is_tx_evt_for_session (session_event_t * e, u32 sid, u8 is_ct)
 {
-  return (e->event_type == SESSION_IO_EVT_TX
-	  && e->fifo->client_session_index == sid);
+  return (e->event_type == SESSION_IO_EVT_TX && e->session_index == sid);
 }
 
 static inline int
@@ -1758,7 +1752,8 @@ vppcom_session_write_inline (uint32_t session_handle, void *buf, size_t n,
 				   !is_ct /* do_evt */ , SVM_Q_WAIT);
 
   if (is_ct && svm_fifo_set_event (s->tx_fifo))
-    app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo, et, SVM_Q_WAIT);
+    app_send_io_evt_to_vpp (s->vpp_evt_q, s->tx_fifo->master_session_index,
+			    et, SVM_Q_WAIT);
 
   ASSERT (n_write > 0);
 
@@ -1782,13 +1777,22 @@ vppcom_session_write_msg (uint32_t session_handle, void *buf, size_t n)
 				      1 /* is_flush */ );
 }
 
-#define vcl_fifo_rx_evt_valid_or_break(_fifo)			\
-if (PREDICT_FALSE (svm_fifo_is_empty (_fifo)))			\
-  {								\
-    svm_fifo_unset_event (_fifo);				\
-    if (svm_fifo_is_empty (_fifo))				\
-      break;							\
-  }								\
+#define vcl_fifo_rx_evt_valid_or_break(_s)				\
+if (PREDICT_FALSE (svm_fifo_is_empty (_s->rx_fifo)))			\
+  {									\
+    if (!vcl_session_is_ct (_s))					\
+      {									\
+	svm_fifo_unset_event (_s->rx_fifo);				\
+	if (svm_fifo_is_empty (_s->rx_fifo))				\
+	  break;							\
+      }									\
+    else if (svm_fifo_is_empty (_s->ct_rx_fifo))			\
+      {									\
+	svm_fifo_unset_event (_s->ct_rx_fifo);				\
+	if (svm_fifo_is_empty (_s->ct_rx_fifo))				\
+	  break;							\
+      }									\
+  }									\
 
 static void
 vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
@@ -1803,10 +1807,10 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
 
   switch (e->event_type)
     {
-    case FIFO_EVENT_APP_RX:
-      vcl_fifo_rx_evt_valid_or_break (e->fifo);
-      sid = e->fifo->client_session_index;
+    case SESSION_IO_EVT_RX:
+      sid = e->session_index;
       session = vcl_session_get (wrk, sid);
+      vcl_fifo_rx_evt_valid_or_break (session);
       if (!session)
 	break;
       if (sid < n_bits && read_map)
@@ -1816,7 +1820,7 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
 	}
       break;
     case FIFO_EVENT_APP_TX:
-      sid = e->fifo->client_session_index;
+      sid = e->session_index;
       session = vcl_session_get (wrk, sid);
       if (!session)
 	break;
@@ -1937,8 +1941,10 @@ vppcom_select_condvar (vcl_worker_t * wrk, int n_bits,
 		       u32 * bits_set)
 {
   time_to_wait = (time_to_wait == -1) ? 1e6 : time_to_wait;
-  return vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
-			       write_map, except_map, time_to_wait, bits_set);
+  vcl_select_handle_mq (wrk, wrk->app_event_queue, n_bits, read_map,
+			write_map, except_map, (bits_set ? 0 : time_to_wait),
+			bits_set);
+  return *bits_set;
 }
 
 static int
@@ -2069,7 +2075,7 @@ vep_verify_epoll_chain (vcl_worker_t * wrk, u32 vep_idx)
   vppcom_epoll_t *vep;
   u32 sid = vep_idx;
 
-  if (VPPCOM_DEBUG <= 1)
+  if (VPPCOM_DEBUG <= 2)
     return;
 
   /* Assumes caller has acquired spinlock: vcm->sessions_lockp */
@@ -2354,11 +2360,10 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
   switch (e->event_type)
     {
     case SESSION_IO_EVT_RX:
-      ASSERT (e->fifo->client_thread_index == vcl_get_worker_index ());
-      vcl_fifo_rx_evt_valid_or_break (e->fifo);
-      sid = e->fifo->client_session_index;
+      sid = e->session_index;
       if (!(session = vcl_session_get (wrk, sid)))
 	break;
+      vcl_fifo_rx_evt_valid_or_break (session);
       session_events = session->vep.ev.events;
       if (!(EPOLLIN & session->vep.ev.events) || session->has_rx_evt)
 	break;
@@ -2368,7 +2373,7 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       session->has_rx_evt = 1;
       break;
     case SESSION_IO_EVT_TX:
-      sid = e->fifo->client_session_index;
+      sid = e->session_index;
       if (!(session = vcl_session_get (wrk, sid)))
 	break;
       session_events = session->vep.ev.events;
@@ -2513,8 +2518,9 @@ vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
 			   int maxevents, u32 n_evts, double wait_for_time)
 {
   wait_for_time = (wait_for_time == -1) ? (double) 1e6 : wait_for_time;
-  return vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events,
-				   maxevents, wait_for_time, &n_evts);
+  vcl_epoll_wait_handle_mq (wrk, wrk->app_event_queue, events, maxevents,
+			    (n_evts ? 0 : wait_for_time), &n_evts);
+  return n_evts;
 }
 
 static int
@@ -2583,7 +2589,6 @@ vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
 	      break;
 	    }
 	}
-
       vec_delete (wrk->unhandled_evts_vector, i, 0);
     }