vcl: epoll fix postponed evt handling
[vpp.git] / src / vcl / vppcom.c
index b3efdcd..b74570d 100644 (file)
@@ -471,8 +471,8 @@ vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
                    sizeof (ip46_address_t));
 
   vcl_session_table_add_vpp_handle (wrk, mp->handle, session->session_index);
-  session->transport.lcl_port = listen_session->transport.lcl_port;
-  session->transport.lcl_ip = listen_session->transport.lcl_ip;
+  session->transport.lcl_port = mp->lcl.port;
+  session->transport.lcl_ip = mp->lcl.ip;
   session->session_type = listen_session->session_type;
   session->is_dgram = vcl_proto_is_dgram (session->session_type);
   session->listener_index = listen_session->session_index;
@@ -807,7 +807,7 @@ vcl_session_disconnected_handler (vcl_worker_t * wrk,
 }
 
 int
-vppcom_session_shutdown (uint32_t session_handle)
+vppcom_session_shutdown (uint32_t session_handle, int how)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
   vcl_session_t *session;
@@ -830,13 +830,20 @@ vppcom_session_shutdown (uint32_t session_handle)
       return VPPCOM_EBADFD;
     }
 
+  if (how == SHUT_RD || how == SHUT_RDWR)
+    {
+      session->flags |= VCL_SESSION_F_RD_SHUTDOWN;
+      if (how == SHUT_RD)
+       return VPPCOM_OK;
+    }
+  session->flags |= VCL_SESSION_F_WR_SHUTDOWN;
+
   if (PREDICT_TRUE (state == VCL_STATE_READY))
     {
       VDBG (1, "session %u [0x%llx]: sending shutdown...",
            session->session_index, vpp_handle);
 
       vcl_send_session_shutdown (wrk, session);
-      session->flags |= VCL_SESSION_F_SHUTDOWN;
     }
 
   return VPPCOM_OK;
@@ -1438,6 +1445,18 @@ vppcom_session_create (u8 proto, u8 is_nonblocking)
   return vcl_session_handle (session);
 }
 
+static void
+vcl_epoll_wait_clean_lt (vcl_worker_t *wrk, u32 sid)
+{
+  int i;
+
+  for (i = vec_len (wrk->ep_level_evts) - 1; i >= 0; i--)
+    {
+      if (wrk->ep_level_evts[i] == sid)
+       vec_del1 (wrk->ep_level_evts, i);
+    }
+}
+
 int
 vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * s,
                     vcl_session_handle_t sh, u8 do_disconnect)
@@ -1468,6 +1487,8 @@ vcl_session_cleanup (vcl_worker_t * wrk, vcl_session_t * s,
        VDBG (0, "session %u [0x%llx]: EPOLL_CTL_DEL vep_idx %u "
              "failed! rv %d (%s)", s->session_index, s->vpp_handle,
              s->vep.vep_sh, rv, vppcom_retval_str (rv));
+      if (PREDICT_FALSE (vec_len (wrk->ep_level_evts)))
+       vcl_epoll_wait_clean_lt (wrk, s->session_index);
     }
 
   if (!do_disconnect)
@@ -1948,6 +1969,18 @@ vppcom_session_read_internal (uint32_t session_handle, void *buf, int n,
       return vcl_session_closed_error (s);
     }
 
+  if (PREDICT_FALSE (s->flags & VCL_SESSION_F_RD_SHUTDOWN))
+    {
+      /* Vpp would ack the incoming data and enqueue it for reading.
+       * So even if SHUT_RD is set, we can still read() the data if
+       * the session is ready.
+       */
+      if (!vcl_session_read_ready (s))
+       {
+         return 0;
+       }
+    }
+
   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
   is_ct = vcl_session_is_ct (s);
   mq = wrk->app_event_queue;
@@ -1987,6 +2020,10 @@ read_again:
     rv = app_recv_stream_raw (rx_fifo, buf, n, 0, peek);
 
   ASSERT (rv >= 0);
+
+  if (peek)
+    return rv;
+
   n_read += rv;
 
   if (svm_fifo_is_empty_cons (rx_fifo))
@@ -2148,8 +2185,12 @@ vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
   svm_msg_q_t *mq;
   u8 is_ct;
 
-  if (PREDICT_FALSE (!buf || n == 0))
-    return VPPCOM_EINVAL;
+  /* Accept zero length writes but just return */
+  if (PREDICT_FALSE (!n))
+    return VPPCOM_OK;
+
+  if (PREDICT_FALSE (!buf))
+    return VPPCOM_EFAULT;
 
   if (PREDICT_FALSE (s->flags & VCL_SESSION_F_IS_VEP))
     {
@@ -2158,8 +2199,7 @@ vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
       return VPPCOM_EBADFD;
     }
 
-  if (PREDICT_FALSE (!vcl_session_is_open (s) ||
-                    s->flags & VCL_SESSION_F_SHUTDOWN))
+  if (PREDICT_FALSE (!vcl_session_is_open (s)))
     {
       VDBG (1, "session %u [0x%llx]: is not open! state 0x%x (%s)",
            s->session_index, s->vpp_handle, s->session_state,
@@ -2167,6 +2207,14 @@ vppcom_session_write_inline (vcl_worker_t * wrk, vcl_session_t * s, void *buf,
       return vcl_session_closed_error (s);;
     }
 
+  if (PREDICT_FALSE (s->flags & VCL_SESSION_F_WR_SHUTDOWN))
+    {
+      VDBG (1, "session %u [0x%llx]: is shutdown! state 0x%x (%s)",
+           s->session_index, s->vpp_handle, s->session_state,
+           vppcom_session_state_str (s->session_state));
+      return VPPCOM_EPIPE;
+    }
+
   is_ct = vcl_session_is_ct (s);
   tx_fifo = is_ct ? s->ct_tx_fifo : s->tx_fifo;
   is_nonblocking = vcl_session_has_attr (s, VCL_SESS_ATTR_NONBLOCK);
@@ -2667,8 +2715,8 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
                  struct epoll_event *event)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
+  int rv = VPPCOM_OK, add_evt = 0;
   vcl_session_t *vep_session;
-  int rv = VPPCOM_OK;
   vcl_session_t *s;
   svm_fifo_t *txf;
 
@@ -2713,6 +2761,12 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          VDBG (0, "EPOLL_CTL_ADD: NULL pointer to epoll_event structure!");
          return VPPCOM_EINVAL;
        }
+      if (s->flags & VCL_SESSION_F_IS_VEP_SESSION)
+       {
+         VDBG (0, "EPOLL_CTL_ADD: %u already epolled!", s->session_index);
+         rv = VPPCOM_EEXIST;
+         goto done;
+       }
       if (vep_session->vep.next_sh != ~0)
        {
          vcl_session_t *next_session;
@@ -2747,6 +2801,7 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          e.event_type = SESSION_IO_EVT_TX;
          e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
+         add_evt = 1;
        }
       /* Generate EPOLLIN if rx fifo has data */
       if ((event->events & EPOLLIN) && (vcl_session_read_ready (s) > 0))
@@ -2755,6 +2810,19 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          e.event_type = SESSION_IO_EVT_RX;
          e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
+         s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
+         add_evt = 1;
+       }
+      if (!add_evt && vcl_session_is_closing (s))
+       {
+         session_event_t e = { 0 };
+         if (s->session_state == VCL_STATE_VPP_CLOSING)
+           e.event_type = SESSION_CTRL_EVT_DISCONNECTED;
+         else
+           e.event_type = SESSION_CTRL_EVT_RESET;
+         e.session_index = s->session_index;
+         e.postponed = 1;
+         vec_add1 (wrk->unhandled_evts_vector, e);
        }
       VDBG (1, "EPOLL_CTL_ADD: vep_sh %u, sh %u, events 0x%x, data 0x%llx!",
            vep_handle, session_handle, event->events, event->data.u64);
@@ -2771,7 +2839,7 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
       else if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP_SESSION)))
        {
          VDBG (0, "sh %u EPOLL_CTL_MOD: not a vep session!", session_handle);
-         rv = VPPCOM_EINVAL;
+         rv = VPPCOM_ENOENT;
          goto done;
        }
       else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
@@ -2782,15 +2850,25 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
          goto done;
        }
 
-      /* Generate EPOLLOUT when tx_fifo/ct_tx_fifo not full */
-      if ((event->events & EPOLLOUT) &&
-         !(s->vep.ev.events & EPOLLOUT) && (vcl_session_write_ready (s) > 0))
+      /* Generate EPOLLOUT if session write ready nd event was not on */
+      if ((event->events & EPOLLOUT) && !(s->vep.ev.events & EPOLLOUT) &&
+         (vcl_session_write_ready (s) > 0))
        {
          session_event_t e = { 0 };
          e.event_type = SESSION_IO_EVT_TX;
          e.session_index = s->session_index;
          vec_add1 (wrk->unhandled_evts_vector, e);
        }
+      /* Generate EPOLLIN if session read ready and event was not on */
+      if ((event->events & EPOLLIN) && !(s->vep.ev.events & EPOLLIN) &&
+         (vcl_session_read_ready (s) > 0))
+       {
+         session_event_t e = { 0 };
+         e.event_type = SESSION_IO_EVT_RX;
+         e.session_index = s->session_index;
+         vec_add1 (wrk->unhandled_evts_vector, e);
+         s->flags &= ~VCL_SESSION_F_HAS_RX_EVT;
+       }
       s->vep.et_mask = VEP_DEFAULT_ET_MASK;
       s->vep.ev = *event;
       txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
@@ -2809,7 +2887,7 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
       if (PREDICT_FALSE (!(s->flags & VCL_SESSION_F_IS_VEP_SESSION)))
        {
          VDBG (0, "EPOLL_CTL_DEL: %u not a vep session!", session_handle);
-         rv = VPPCOM_EINVAL;
+         rv = VPPCOM_ENOENT;
          goto done;
        }
       else if (PREDICT_FALSE (s->vep.vep_sh != vep_handle))
@@ -2855,9 +2933,12 @@ vppcom_epoll_ctl (uint32_t vep_handle, int op, uint32_t session_handle,
       s->vep.vep_sh = ~0;
       s->flags &= ~VCL_SESSION_F_IS_VEP_SESSION;
 
-      txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
-      if (txf)
-       svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
+      if (vcl_session_is_open (s))
+       {
+         txf = vcl_session_is_ct (s) ? s->ct_tx_fifo : s->tx_fifo;
+         if (txf)
+           svm_fifo_del_want_deq_ntf (txf, SVM_FIFO_WANT_DEQ_NOTIF_IF_FULL);
+       }
 
       VDBG (1, "EPOLL_CTL_DEL: vep_idx %u, sh %u!", vep_handle,
            session_handle);
@@ -2954,9 +3035,17 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
        events[*num_ev].events |= EPOLLHUP;
       break;
     case SESSION_CTRL_EVT_DISCONNECTED:
-      disconnected_msg = (session_disconnected_msg_t *) e->data;
-      s = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (vcl_session_is_closed (s))
+      if (!e->postponed)
+       {
+         disconnected_msg = (session_disconnected_msg_t *) e->data;
+         s = vcl_session_disconnected_handler (wrk, disconnected_msg);
+       }
+      else
+       {
+         s = vcl_session_get (wrk, e->session_index);
+       }
+      if (vcl_session_is_closed (s) ||
+         !(s->flags & VCL_SESSION_F_IS_VEP_SESSION))
        break;
       sid = s->session_index;
       session_events = s->vep.ev.events;
@@ -2965,9 +3054,13 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       session_evt_data = s->vep.ev.data.u64;
       break;
     case SESSION_CTRL_EVT_RESET:
-      sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
+      if (!e->postponed)
+       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
+      else
+       sid = e->session_index;
       s = vcl_session_get (wrk, sid);
-      if (vcl_session_is_closed (s))
+      if (vcl_session_is_closed (s) ||
+         !(s->flags & VCL_SESSION_F_IS_VEP_SESSION))
        break;
       session_events = s->vep.ev.events;
       add_event = 1;
@@ -3011,6 +3104,10 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
          s = vcl_session_get (wrk, sid);
          s->vep.ev.events = 0;
        }
+      if (!(EPOLLET & session_events))
+       {
+         vec_add1 (wrk->ep_level_evts, sid);
+       }
       *num_ev += 1;
     }
 }
@@ -3125,13 +3222,73 @@ vppcom_epoll_wait_eventfd (vcl_worker_t *wrk, struct epoll_event *events,
   return 0;
 }
 
+static void
+vcl_epoll_swap_lt_lists (vcl_worker_t *wrk)
+{
+  u32 *le;
+
+  le = wrk->ep_level_evts;
+  wrk->ep_level_evts = wrk->ep_level_evts_fl;
+  wrk->ep_level_evts_fl = le;
+}
+
+static void
+vcl_epoll_wait_handle_lt (vcl_worker_t *wrk, struct epoll_event *events,
+                         int maxevents, u32 *n_evts)
+{
+  u32 *sid, add_event = 0, *le = wrk->ep_level_evts_fl;
+  vcl_session_t *s;
+  u64 evt_data;
+
+  if (*n_evts >= maxevents)
+    {
+      vec_add (wrk->ep_level_evts, le, vec_len (le));
+      vec_reset_length (wrk->ep_level_evts_fl);
+      return;
+    }
+
+  vec_foreach (sid, le)
+    {
+      s = vcl_session_get (wrk, sid[0]);
+      if (!s)
+       continue;
+      if ((s->vep.ev.events & EPOLLIN) && vcl_session_read_ready (s))
+       {
+         add_event = 1;
+         events[*n_evts].events |= EPOLLIN;
+         evt_data = s->vep.ev.data.u64;
+       }
+      if ((s->vep.ev.events & EPOLLOUT) && vcl_session_write_ready (s))
+       {
+         add_event = 1;
+         events[*n_evts].events |= EPOLLOUT;
+         evt_data = s->vep.ev.data.u64;
+       }
+      if (add_event)
+       {
+         events[*n_evts].data.u64 = evt_data;
+         *n_evts += 1;
+         add_event = 0;
+         vec_add1 (wrk->ep_level_evts, sid[0]);
+         if (*n_evts == maxevents)
+           {
+             u32 pos = (sid - le) + 1;
+             vec_add (wrk->ep_level_evts, &le[pos], vec_len (le) - pos);
+             break;
+           }
+       }
+    }
+
+  vec_reset_length (wrk->ep_level_evts_fl);
+}
+
 int
 vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
                   int maxevents, double wait_for_time)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
   vcl_session_t *vep_session;
-  u32 n_evts = 0;
+  u32 n_evts = 0, do_lt = 0;
   int i;
 
   if (PREDICT_FALSE (maxevents <= 0))
@@ -3170,12 +3327,23 @@ vppcom_epoll_wait (uint32_t vep_handle, struct epoll_event *events,
   if ((int) wait_for_time == -2)
     return n_evts;
 
+  if (PREDICT_FALSE (vec_len (wrk->ep_level_evts)))
+    {
+      vcl_epoll_swap_lt_lists (wrk);
+      do_lt = 1;
+    }
+
   if (vcm->cfg.use_mq_eventfd)
-    return vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
-                                     wait_for_time);
+    n_evts = vppcom_epoll_wait_eventfd (wrk, events, maxevents, n_evts,
+                                       wait_for_time);
+  else
+    n_evts = vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
+                                       wait_for_time);
 
-  return vppcom_epoll_wait_condvar (wrk, events, maxevents, n_evts,
-                                   wait_for_time);
+  if (PREDICT_FALSE (do_lt))
+    vcl_epoll_wait_handle_lt (wrk, events, maxevents, &n_evts);
+
+  return n_evts;
 }
 
 int
@@ -3183,7 +3351,7 @@ vppcom_session_attr (uint32_t session_handle, uint32_t op,
                     void *buffer, uint32_t * buflen)
 {
   vcl_worker_t *wrk = vcl_worker_get_current ();
-  u32 *flags = buffer, tmp_flags = 0;
+  u32 *flags = buffer;
   vppcom_endpt_t *ep = buffer;
   transport_endpt_attr_t tea;
   vcl_session_t *session;
@@ -3721,27 +3889,6 @@ vppcom_session_attr (uint32_t session_handle, uint32_t op,
            *buflen);
       break;
 
-    case VPPCOM_ATTR_SET_SHUT:
-      if (*flags == SHUT_RD || *flags == SHUT_RDWR)
-       vcl_session_set_attr (session, VCL_SESS_ATTR_SHUT_RD);
-      if (*flags == SHUT_WR || *flags == SHUT_RDWR)
-       vcl_session_set_attr (session, VCL_SESS_ATTR_SHUT_WR);
-      break;
-
-    case VPPCOM_ATTR_GET_SHUT:
-      if (vcl_session_has_attr (session, VCL_SESS_ATTR_SHUT_RD))
-       tmp_flags = 1;
-      if (vcl_session_has_attr (session, VCL_SESS_ATTR_SHUT_WR))
-       tmp_flags |= 2;
-      if (tmp_flags == 1)
-       *(int *) buffer = SHUT_RD;
-      else if (tmp_flags == 2)
-       *(int *) buffer = SHUT_WR;
-      else if (tmp_flags == 3)
-       *(int *) buffer = SHUT_RDWR;
-      *buflen = sizeof (int);
-      break;
-
     case VPPCOM_ATTR_SET_CONNECTED:
       session->flags |= VCL_SESSION_F_CONNECTED;
       break;
@@ -3870,12 +4017,9 @@ vppcom_session_sendto (uint32_t session_handle, void *buffer,
   vcl_session_t *s;
 
   s = vcl_session_get_w_handle (wrk, session_handle);
-  if (!s)
+  if (PREDICT_FALSE (!s))
     return VPPCOM_EBADFD;
 
-  if (!buffer)
-    return VPPCOM_EINVAL;
-
   if (ep)
     {
       if (!vcl_session_is_cl (s))