session: poll main thread if pending connects

Type: improvement

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

src/vnet/session/session.c

@@ -1822,6 +1822,7 @@ session_manager_main_enable (vlib_main_t * vm)
       wrk->vm = vlib_get_main_by_index (i);
       wrk->last_vlib_time = vlib_time_now (vm);
       wrk->last_vlib_us_time = wrk->last_vlib_time * CLIB_US_TIME_FREQ;
+      wrk->timerfd = -1;
       vec_validate (wrk->session_to_enqueue, smm->last_transport_proto_type);
 
       if (num_threads > 1)
@@ -1899,6 +1900,7 @@ session_node_enable_disable (u8 is_en)
       if (i == 0 && n_vlibs > 1)
 	{
 	  vlib_node_set_state (vm, session_queue_node.index, mstate);
+	  session_main_get_worker (0)->state = SESSION_WRK_INTERRUPT;
 	  if (is_en)
 	    {
 	      vlib_node_set_state (vm, session_queue_process_node.index,

src/vnet/session/session.h

@@ -151,6 +151,9 @@ typedef struct session_worker_
   /** Flag that is set if main thread signaled to handle connects */
   u32 pending_connects_ntf;
 
+  /** Main thread loops in poll mode without a connect */
+  u32 no_connect_loops;
+
 #if SESSION_DEBUG
   /** last event poll time by thread */
   clib_time_type_t last_event_poll;

src/vnet/session/session_node.c

@@ -33,6 +33,43 @@
       return;								\
    }
 
+static void
+session_wrk_timerfd_update (session_worker_t *wrk, u64 time_ns)
+{
+  struct itimerspec its;
+
+  its.it_value.tv_sec = 0;
+  its.it_value.tv_nsec = time_ns;
+  its.it_interval.tv_sec = 0;
+  its.it_interval.tv_nsec = its.it_value.tv_nsec;
+
+  if (timerfd_settime (wrk->timerfd, 0, &its, NULL) == -1)
+    clib_warning ("timerfd_settime");
+}
+
+always_inline u64
+session_wrk_tfd_timeout (session_wrk_state_t state, u32 thread_index)
+{
+  if (state == SESSION_WRK_INTERRUPT)
+    return thread_index ? 1e6 : vlib_num_workers () ? 5e8 : 1e6;
+  else if (state == SESSION_WRK_IDLE)
+    return thread_index ? 1e8 : vlib_num_workers () ? 5e8 : 1e8;
+  else
+    return 0;
+}
+
+static inline void
+session_wrk_set_state (session_worker_t *wrk, session_wrk_state_t state)
+{
+  u64 time_ns;
+
+  wrk->state = state;
+  if (wrk->timerfd == -1)
+    return;
+  time_ns = session_wrk_tfd_timeout (state, wrk->vm->thread_index);
+  session_wrk_timerfd_update (wrk, time_ns);
+}
+
 static transport_endpt_ext_cfg_t *
 session_mq_get_ext_config (application_t *app, uword offset)
 {
@@ -170,8 +207,7 @@ session_mq_handle_connects_rpc (void *arg)
   u32 max_connects = 32, n_connects = 0;
   vlib_main_t *vm = vlib_get_main ();
   session_evt_elt_t *he, *elt, *next;
-  session_worker_t *fwrk;
-  u8 need_reschedule = 1;
+  session_worker_t *fwrk, *wrk;
 
   ASSERT (vlib_get_thread_index () == 0);
 
@@ -194,21 +230,43 @@ session_mq_handle_connects_rpc (void *arg)
       n_connects += 1;
     }
 
-  if (clib_llist_is_empty (fwrk->event_elts, evt_list, he))
+  /* Switch worker to poll mode if it was in interrupt mode and had work or
+   * back to interrupt if threshold of loops without a connect is passed.
+   * While in poll mode, reprogram connects rpc */
+  wrk = session_main_get_worker (0);
+  if (wrk->state != SESSION_WRK_POLLING)
     {
-      fwrk->pending_connects_ntf = 0;
-      need_reschedule = 0;
+      if (!n_connects)
+	goto done;
+
+      session_wrk_set_state (wrk, SESSION_WRK_POLLING);
+      vlib_node_set_state (vm, session_queue_node.index,
+			   VLIB_NODE_STATE_POLLING);
+      wrk->no_connect_loops = 0;
+    }
+  else
+    {
+      if (!n_connects)
+	{
+	  if (++wrk->no_connect_loops > 1e5)
+	    {
+	      session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
+	      vlib_node_set_state (vm, session_queue_node.index,
+				   VLIB_NODE_STATE_INTERRUPT);
+	      fwrk->pending_connects_ntf = 0;
+	      goto done;
+	    }
+	}
+      else
+	wrk->no_connect_loops = 0;
     }
 
+  elt = session_evt_alloc_ctrl (wrk);
+  elt->evt.event_type = SESSION_CTRL_EVT_RPC;
+  elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
+
+done:
   vlib_worker_thread_barrier_release (vm);
-
-  if (need_reschedule)
-    {
-      vlib_node_set_interrupt_pending (vm, session_queue_node.index);
-      elt = session_evt_alloc_ctrl (session_main_get_worker (0));
-      elt->evt.event_type = SESSION_CTRL_EVT_RPC;
-      elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
-    }
 }
 
 static void
@@ -1583,41 +1641,6 @@ session_wrk_handle_mq (session_worker_t *wrk, svm_msg_q_t *mq)
   return n_to_dequeue;
 }
 
-static void
-session_wrk_timerfd_update (session_worker_t *wrk, u64 time_ns)
-{
-  struct itimerspec its;
-
-  its.it_value.tv_sec = 0;
-  its.it_value.tv_nsec = time_ns;
-  its.it_interval.tv_sec = 0;
-  its.it_interval.tv_nsec = its.it_value.tv_nsec;
-
-  if (timerfd_settime (wrk->timerfd, 0, &its, NULL) == -1)
-    clib_warning ("timerfd_settime");
-}
-
-always_inline u64
-session_wrk_tfd_timeout (session_wrk_state_t state, u32 thread_index)
-{
-  if (state == SESSION_WRK_INTERRUPT)
-    return thread_index ? 1e6 : vlib_num_workers () ? 5e8 : 1e6;
-  else if (state == SESSION_WRK_IDLE)
-    return thread_index ? 1e8 : vlib_num_workers () ? 5e8 : 1e8;
-  else
-    return 0;
-}
-
-static inline void
-session_wrk_set_state (session_worker_t *wrk, session_wrk_state_t state)
-{
-  u64 time_ns;
-
-  wrk->state = state;
-  time_ns = session_wrk_tfd_timeout (state, wrk->vm->thread_index);
-  session_wrk_timerfd_update (wrk, time_ns);
-}
-
 static void
 session_wrk_update_state (session_worker_t *wrk)
 {