vcl: support inter worker rpc
[vpp.git] / src / vcl / vppcom.c
index 579cbc1..41d2f31 100644 (file)
@@ -209,6 +209,8 @@ vcl_send_session_listen (vcl_worker_t * wrk, vcl_session_t * s)
   clib_memcpy_fast (&mp->ip, &s->transport.lcl_ip, sizeof (mp->ip));
   mp->port = s->transport.lcl_port;
   mp->proto = s->session_type;
+  if (s->flags & VCL_SESSION_F_CONNECTED)
+    mp->flags = TRANSPORT_CFG_F_CONNECTED;
   app_send_ctrl_evt_to_vpp (mq, app_evt);
 }
 
@@ -349,6 +351,36 @@ vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
   app_send_ctrl_evt_to_vpp (mq, app_evt);
 }
 
+void
+vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len)
+{
+  app_session_evt_t _app_evt, *app_evt = &_app_evt;
+  session_app_wrk_rpc_msg_t *mp;
+  vcl_worker_t *dst_wrk, *wrk;
+  svm_msg_q_t *mq;
+
+  if (data_len > sizeof (mp->data))
+    goto done;
+
+  clib_spinlock_lock (&vcm->workers_lock);
+
+  dst_wrk = vcl_worker_get_if_valid (dst_wrk_index);
+  if (!dst_wrk)
+    goto done;
+
+  wrk = vcl_worker_get_current ();
+  mq = vcl_worker_ctrl_mq (wrk);
+  app_alloc_ctrl_evt_to_vpp (mq, app_evt, SESSION_CTRL_EVT_APP_WRK_RPC);
+  mp = (session_app_wrk_rpc_msg_t *) app_evt->evt->data;
+  mp->client_index = wrk->my_client_index;
+  mp->wrk_index = dst_wrk->vpp_wrk_index;
+  clib_memcpy (mp->data, data, data_len);
+  app_send_ctrl_evt_to_vpp (mq, app_evt);
+
+done:
+  clib_spinlock_unlock (&vcm->workers_lock);
+}
+
 static u32
 vcl_session_accepted_handler (vcl_worker_t * wrk, session_accepted_msg_t * mp,
                              u32 ls_index)
@@ -550,7 +582,8 @@ vcl_session_reset_handler (vcl_worker_t * wrk,
       return VCL_INVALID_SESSION_INDEX;
     }
 
-  session->session_state = STATE_DISCONNECT;
+  if (session->session_state != STATE_CLOSED)
+    session->session_state = STATE_DISCONNECT;
   VDBG (0, "reset session %u [0x%llx]", sid, reset_msg->handle);
   return sid;
 }
@@ -749,6 +782,17 @@ vcl_session_cleanup_handler (vcl_worker_t * wrk, void *data)
       return;
     }
 
+  if (msg->type == SESSION_CLEANUP_TRANSPORT)
+    {
+      /* Transport was cleaned up before we confirmed close. Probably the
+       * app is still waiting for some data that cannot be delivered.
+       * Confirm close to make sure everything is cleaned up */
+      if (session->session_state == STATE_VPP_CLOSING)
+       vcl_session_cleanup (wrk, session, vcl_session_handle (session),
+                            1 /* do_disconnect */ );
+      return;
+    }
+
   vcl_session_table_del_vpp_handle (wrk, msg->handle);
   /* Should not happen. App did not close the connection so don't free it. */
   if (session->session_state != STATE_CLOSED)
@@ -852,6 +896,15 @@ vcl_session_app_del_segment_handler (vcl_worker_t * wrk, void *data)
   VDBG (1, "Unmapped segment: %d", msg->segment_handle);
 }
 
+static void
+vcl_worker_rpc_handler (vcl_worker_t * wrk, void *data)
+{
+  if (!vcm->wrk_rpc_fn)
+    return;
+
+  (vcm->wrk_rpc_fn) (data);
+}
+
 static int
 vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
 {
@@ -909,6 +962,9 @@ vcl_handle_mq_event (vcl_worker_t * wrk, session_event_t * e)
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       clib_warning ("unhandled %u", e->event_type);
     }
@@ -1148,7 +1204,7 @@ vppcom_app_exit (void)
  * VPPCOM Public API functions
  */
 int
-vppcom_app_create (char *app_name)
+vppcom_app_create (const char *app_name)
 {
   vppcom_cfg_t *vcl_cfg = &vcm->cfg;
   int rv;
@@ -1211,8 +1267,8 @@ vppcom_app_create (char *app_name)
 void
 vppcom_app_destroy (void)
 {
+  vcl_worker_t *wrk, *current_wrk;
   struct dlmallinfo mi;
-  vcl_worker_t *wrk;
   mspace heap;
 
   if (!pool_elts (vcm->workers))
@@ -1220,16 +1276,19 @@ vppcom_app_destroy (void)
 
   vcl_evt (VCL_EVT_DETACH, vcm);
 
-  vcl_send_app_detach (vcl_worker_get_current ());
+  current_wrk = vcl_worker_get_current ();
 
   /* *INDENT-OFF* */
   pool_foreach (wrk, vcm->workers, ({
-    if (pool_elts (vcm->workers) == 1)
-      vl_client_disconnect_from_vlib ();
-    vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
+    if (current_wrk != wrk)
+      vcl_worker_cleanup (wrk, 0 /* notify vpp */ );
   }));
   /* *INDENT-ON* */
 
+  vcl_send_app_detach (current_wrk);
+  vppcom_disconnect_from_vpp ();
+  vcl_worker_cleanup (current_wrk, 0 /* notify vpp */ );
+
   vcl_elog_stop (vcm);
 
   /*
@@ -1240,6 +1299,7 @@ vppcom_app_destroy (void)
   munmap (mspace_least_addr (heap), mi.arena);
 
   vcm = &_vppcom_main;
+  vcm->is_init = 0;
 }
 
 int
@@ -1541,10 +1601,6 @@ vppcom_unformat_proto (uint8_t * proto, char *proto_str)
     *proto = VPPCOM_PROTO_UDP;
   else if (!strcmp (proto_str, "udp"))
     *proto = VPPCOM_PROTO_UDP;
-  else if (!strcmp (proto_str, "UDPC"))
-    *proto = VPPCOM_PROTO_UDPC;
-  else if (!strcmp (proto_str, "udpc"))
-    *proto = VPPCOM_PROTO_UDPC;
   else if (!strcmp (proto_str, "TLS"))
     *proto = VPPCOM_PROTO_TLS;
   else if (!strcmp (proto_str, "tls"))
@@ -2245,6 +2301,9 @@ vcl_select_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       clib_warning ("unhandled: %u", e->event_type);
       break;
@@ -2400,12 +2459,12 @@ vppcom_select (int n_bits, vcl_si_set * read_map, vcl_si_set * write_map,
   clib_bitmap_foreach (sid, wrk->wr_bitmap, ({
     if (!(session = vcl_session_get (wrk, sid)))
       {
-        if (except_map && sid < minbits)
-          clib_bitmap_set_no_check (except_map, sid, 1);
-        continue;
+       clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
+       bits_set++;
+       continue;
       }
 
-    if (vcl_session_write_ready (session) > 0)
+    if (vcl_session_write_ready (session))
       {
         clib_bitmap_set_no_check ((uword*)write_map, sid, 1);
         bits_set++;
@@ -2421,12 +2480,12 @@ check_rd:
   clib_bitmap_foreach (sid, wrk->rd_bitmap, ({
     if (!(session = vcl_session_get (wrk, sid)))
       {
-        if (except_map && sid < minbits)
-          clib_bitmap_set_no_check (except_map, sid, 1);
-        continue;
+       clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
+       bits_set++;
+       continue;
       }
 
-    if (vcl_session_read_ready (session) > 0)
+    if (vcl_session_read_ready (session))
       {
         clib_bitmap_set_no_check ((uword*)read_map, sid, 1);
         bits_set++;
@@ -2765,7 +2824,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     {
     case SESSION_IO_EVT_RX:
       sid = e->session_index;
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       vcl_fifo_rx_evt_valid_or_break (session);
       session_events = session->vep.ev.events;
@@ -2778,7 +2838,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_IO_EVT_TX:
       sid = e->session_index;
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       if (!(EPOLLOUT & session_events))
@@ -2806,7 +2867,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       connected_msg = (session_connected_msg_t *) e->data;
       sid = vcl_session_connected_handler (wrk, connected_msg);
       /* Generate EPOLLOUT because there's no connected event */
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       if (!(EPOLLOUT & session_events))
@@ -2820,7 +2882,7 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_DISCONNECTED:
       disconnected_msg = (session_disconnected_msg_t *) e->data;
       session = vcl_session_disconnected_handler (wrk, disconnected_msg);
-      if (!session)
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       add_event = 1;
@@ -2829,7 +2891,8 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
       break;
     case SESSION_CTRL_EVT_RESET:
       sid = vcl_session_reset_handler (wrk, (session_reset_msg_t *) e->data);
-      if (!(session = vcl_session_get (wrk, sid)))
+      session = vcl_session_get (wrk, sid);
+      if (vcl_session_is_closed (session))
        break;
       session_events = session->vep.ev.events;
       add_event = 1;
@@ -2857,6 +2920,9 @@ vcl_epoll_wait_handle_mq_event (vcl_worker_t * wrk, session_event_t * e,
     case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
       vcl_session_app_del_segment_handler (wrk, e->data);
       break;
+    case SESSION_CTRL_EVT_RPC:
+      vcl_worker_rpc_handler (wrk, e->data);
+      break;
     default:
       VDBG (0, "unhandled: %u", e->event_type);
       break;
@@ -2908,7 +2974,7 @@ vcl_epoll_wait_handle_mq (vcl_worker_t * wrk, svm_msg_q_t * mq,
        }
     }
   ASSERT (maxevents > *num_ev);
-  vcl_mq_dequeue_batch (wrk, mq, maxevents - *num_ev);
+  vcl_mq_dequeue_batch (wrk, mq, ~0);
   svm_msg_q_unlock (mq);
 
 handle_dequeued:
@@ -2916,7 +2982,10 @@ handle_dequeued:
     {
       msg = vec_elt_at_index (wrk->mq_msg_vector, i);
       e = svm_msg_q_msg_data (mq, msg);
-      vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
+      if (*num_ev < maxevents)
+       vcl_epoll_wait_handle_mq_event (wrk, e, events, num_ev);
+      else
+       vcl_handle_mq_event (wrk, e);
       svm_msg_q_free_msg (mq, msg);
     }
   vec_reset_length (wrk->mq_msg_vector);
@@ -2946,7 +3015,7 @@ vppcom_epoll_wait_condvar (vcl_worker_t * wrk, struct epoll_event *events,
        continue;
 
       now = clib_time_now (&wrk->clib_time);
-      wait -= now - start;
+      wait -= (now - start) * 1e3;
       start = now;
     }
   while (wait > 0);
@@ -3060,8 +3129,10 @@ vppcom_session_attr (uint32_t session_handle, uint32_t op,
     case VPPCOM_ATTR_GET_FLAGS:
       if (PREDICT_TRUE (buffer && buflen && (*buflen >= sizeof (*flags))))
        {
-         *flags = O_RDWR | (VCL_SESS_ATTR_TEST (session->attr,
-                                                VCL_SESS_ATTR_NONBLOCK));
+         *flags =
+           O_RDWR |
+           (VCL_SESS_ATTR_TEST (session->attr, VCL_SESS_ATTR_NONBLOCK) ?
+            O_NONBLOCK : 0);
          *buflen = sizeof (*flags);
          VDBG (2, "VPPCOM_ATTR_GET_FLAGS: sh %u, flags = 0x%08x, "
                "is_nonblocking = %u", session_handle, *flags,
@@ -3578,6 +3649,11 @@ vppcom_session_attr (uint32_t session_handle, uint32_t op,
        *(int *) buffer = SHUT_RDWR;
       *buflen = sizeof (int);
       break;
+
+    case VPPCOM_ATTR_SET_CONNECTED:
+      session->flags |= VCL_SESSION_F_CONNECTED;
+      break;
+
     default:
       rv = VPPCOM_EINVAL;
       break;
@@ -3816,6 +3892,12 @@ vppcom_worker_unregister (void)
   vcl_set_worker_index (~0);
 }
 
+void
+vppcom_worker_index_set (int index)
+{
+  vcl_set_worker_index (index);
+}
+
 int
 vppcom_worker_index (void)
 {
@@ -3868,6 +3950,95 @@ vppcom_session_n_accepted (uint32_t session_handle)
   return session->n_accepted_sessions;
 }
 
+const char *
+vppcom_proto_str (vppcom_proto_t proto)
+{
+  char const *proto_str;
+
+  switch (proto)
+    {
+    case VPPCOM_PROTO_TCP:
+      proto_str = "TCP";
+      break;
+    case VPPCOM_PROTO_UDP:
+      proto_str = "UDP";
+      break;
+    case VPPCOM_PROTO_TLS:
+      proto_str = "TLS";
+      break;
+    case VPPCOM_PROTO_QUIC:
+      proto_str = "QUIC";
+      break;
+    default:
+      proto_str = "UNKNOWN";
+      break;
+    }
+  return proto_str;
+}
+
+const char *
+vppcom_retval_str (int retval)
+{
+  char const *st;
+
+  switch (retval)
+    {
+    case VPPCOM_OK:
+      st = "VPPCOM_OK";
+      break;
+
+    case VPPCOM_EAGAIN:
+      st = "VPPCOM_EAGAIN";
+      break;
+
+    case VPPCOM_EFAULT:
+      st = "VPPCOM_EFAULT";
+      break;
+
+    case VPPCOM_ENOMEM:
+      st = "VPPCOM_ENOMEM";
+      break;
+
+    case VPPCOM_EINVAL:
+      st = "VPPCOM_EINVAL";
+      break;
+
+    case VPPCOM_EBADFD:
+      st = "VPPCOM_EBADFD";
+      break;
+
+    case VPPCOM_EAFNOSUPPORT:
+      st = "VPPCOM_EAFNOSUPPORT";
+      break;
+
+    case VPPCOM_ECONNABORTED:
+      st = "VPPCOM_ECONNABORTED";
+      break;
+
+    case VPPCOM_ECONNRESET:
+      st = "VPPCOM_ECONNRESET";
+      break;
+
+    case VPPCOM_ENOTCONN:
+      st = "VPPCOM_ENOTCONN";
+      break;
+
+    case VPPCOM_ECONNREFUSED:
+      st = "VPPCOM_ECONNREFUSED";
+      break;
+
+    case VPPCOM_ETIMEDOUT:
+      st = "VPPCOM_ETIMEDOUT";
+      break;
+
+    default:
+      st = "UNKNOWN_STATE";
+      break;
+    }
+
+  return st;
+}
+
 /*
  * fd.io coding-style-patch-verification: ON
  *