vcl: support multi-threads with session migration

Currently, mutlti-threads only support one dispatch thread and multiple
worker threads, eventually only dispatch thread is a vcl worker and can
interact with epoll.
This patch will register all threads as vcl worker, and then each
thread can interact with epoll now. Moreover, session migration also
supported, such as socket created in thread A and used (bind, connect
and etc.) in thread B.

Type: feature

Signed-off-by: hanlin <hanlin_wang@163.com>
Change-Id: Iab0b43a33466968c1423d7d20faf1460c8589d91

src/vcl/ldp.c

@@ -2223,7 +2223,7 @@ epoll_create1 (int flags)
   if ((errno = -ldp_init ()))
     return -1;
 
-  if (ldp->vcl_needs_real_epoll)
+  if (ldp->vcl_needs_real_epoll || vls_use_real_epoll ())
     {
       /* Make sure workers have been allocated */
       if (!ldp->workers)
@@ -2425,7 +2425,7 @@ static inline int
 ldp_epoll_pwait_eventfd (int epfd, struct epoll_event *events,
 			 int maxevents, int timeout, const sigset_t * sigmask)
 {
-  ldp_worker_ctx_t *ldpw = ldp_worker_get_current ();
+  ldp_worker_ctx_t *ldpw;
   int libc_epfd, rv = 0, num_ev;
   vls_handle_t ep_vlsh;
 
@@ -2438,6 +2438,11 @@ ldp_epoll_pwait_eventfd (int epfd, struct epoll_event *events,
       return -1;
     }
 
+  if (vls_mt_wrk_supported ())
+    if (PREDICT_FALSE (vppcom_worker_index () == ~0))
+      vls_register_vcl_worker ();
+
+  ldpw = ldp_worker_get_current ();
   if (epfd == ldpw->vcl_mq_epfd)
     return libc_epoll_pwait (epfd, events, maxevents, timeout, sigmask);
 
@@ -2497,7 +2502,7 @@ ldp_epoll_pwait_eventfd (int epfd, struct epoll_event *events,
   rv = vls_epoll_wait (ep_vlsh, events, maxevents, 0);
   if (rv > 0)
     goto done;
-  else if (rv < 0)
+  else if (PREDICT_FALSE (rv < 0))
     {
       errno = -rv;
       rv = -1;

src/vcl/vcl_cfg.c

@@ -498,6 +498,11 @@ vppcom_cfg_read_file (char *conf_fname)
 	      VCFG_DBG (0, "VCL<%d>: configured tls-engine %u (0x%x)",
 			getpid (), vcl_cfg->tls_engine, vcl_cfg->tls_engine);
 	    }
+	  else if (unformat (line_input, "multi-thread"))
+	    {
+	      vcl_cfg->mt_supported = 1;
+	      VCFG_DBG (0, "VCL<%d>: configured with multithread", getpid ());
+	    }
 	  else if (unformat (line_input, "}"))
 	    {
 	      vc_cfg_input = 0;

src/vcl/vcl_locked.h

@@ -51,6 +51,9 @@ vcl_session_handle_t vlsh_to_session_index (vls_handle_t vlsh);
 vls_handle_t vls_session_index_to_vlsh (uint32_t session_index);
 int vls_app_create (char *app_name);
 unsigned char vls_use_eventfd (void);
+unsigned char vls_mt_wrk_supported (void);
+int vls_use_real_epoll (void);
+void vls_register_vcl_worker (void);
 
 #endif /* SRC_VCL_VCL_LOCKED_H_ */
 

src/vcl/vcl_private.c

@@ -217,7 +217,9 @@ vcl_worker_alloc_and_init ()
   wrk->mqs_epfd = -1;
   if (vcm->cfg.use_mq_eventfd)
     {
+      wrk->vcl_needs_real_epoll = 1;
       wrk->mqs_epfd = epoll_create (1);
+      wrk->vcl_needs_real_epoll = 0;
       if (wrk->mqs_epfd < 0)
 	{
 	  clib_unix_warning ("epoll_create() returned");

src/vcl/vcl_private.h

@@ -218,6 +218,7 @@ typedef struct vppcom_cfg_t_
   u8 *vpp_api_socket_name;
   u8 *vpp_api_chroot;
   u32 tls_engine;
+  u8 mt_supported;
 } vppcom_cfg_t;
 
 void vppcom_cfg (vppcom_cfg_t * vcl_cfg);
@@ -307,6 +308,10 @@ typedef struct vcl_worker_
   socket_client_main_t bapi_sock_ctx;
   memory_client_main_t bapi_shm_ctx;
   api_main_t bapi_api_ctx;
+
+  /** vcl needs next epoll_create to go to libc_epoll */
+  u8 vcl_needs_real_epoll;
+  volatile int rpc_done;
 } vcl_worker_t;
 
 typedef void (vcl_rpc_fn_t) (void *args);
@@ -659,7 +664,7 @@ vcl_session_vpp_evt_q (vcl_worker_t * wrk, vcl_session_t * s)
 
 void vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
 				     u32 wrk_index);
-void vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len);
+int vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len);
 
 /*
  * VCL Binary API

src/vcl/vppcom.c

@@ -351,13 +351,14 @@ vcl_send_session_worker_update (vcl_worker_t * wrk, vcl_session_t * s,
   app_send_ctrl_evt_to_vpp (mq, app_evt);
 }
 
-void
+int
 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;
+  int ret = -1;
 
   if (data_len > sizeof (mp->data))
     goto done;
@@ -376,9 +377,11 @@ vcl_send_worker_rpc (u32 dst_wrk_index, void *data, u32 data_len)
   mp->wrk_index = dst_wrk->vpp_wrk_index;
   clib_memcpy (mp->data, data, data_len);
   app_send_ctrl_evt_to_vpp (mq, app_evt);
+  ret = 0;
 
 done:
   clib_spinlock_unlock (&vcm->workers_lock);
+  return ret;
 }
 
 static u32
@@ -902,7 +905,7 @@ vcl_worker_rpc_handler (vcl_worker_t * wrk, void *data)
   if (!vcm->wrk_rpc_fn)
     return;
 
-  (vcm->wrk_rpc_fn) (data);
+  (vcm->wrk_rpc_fn) (((session_app_wrk_rpc_msg_t *) data)->data);
 }
 
 static int
@@ -962,7 +965,7 @@ 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:
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
       vcl_worker_rpc_handler (wrk, e->data);
       break;
     default:
@@ -2301,7 +2304,7 @@ 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:
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
       vcl_worker_rpc_handler (wrk, e->data);
       break;
     default:
@@ -2920,7 +2923,7 @@ 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:
+    case SESSION_CTRL_EVT_APP_WRK_RPC:
       vcl_worker_rpc_handler (wrk, e->data);
       break;
     default:

src/vnet/session/application_interface.h

@@ -531,7 +531,7 @@ typedef struct session_app_wrk_rpc_msg_
 {
   u32 client_index;	/**< app client index */
   u32 wrk_index;	/**< dst worker index */
-  u8 data[252];		/**< rpc data */
+  u8 data[64];		/**< rpc data */
 } __clib_packed session_app_wrk_rpc_msg_t;
 
 typedef struct app_session_event_

src/vnet/session/session_types.h

@@ -384,6 +384,7 @@ typedef enum
   _(APP_DEL_SEGMENT, app_del_segment)				\
   _(MIGRATED, migrated)						\
   _(CLEANUP, cleanup)						\
+  _(APP_WRK_RPC, app_wrk_rpc)					\
 
 /* Deprecated and will be removed. Use types above */
 #define FIFO_EVENT_APP_RX SESSION_IO_EVT_RX