11 files changed, 835 insertions, 580 deletions

diff --git a/src/vnet/session/application.c b/src/vnet/session/application.c
index cf867401ea2..fdd5a0a67cd 100644
--- a/src/vnet/session/application.c
+++ b/src/vnet/session/application.c
@@ -725,6 +725,12 @@ application_get_if_valid (u32 app_index)
   return pool_elt_at_index (app_main.app_pool, app_index);
 }
 
+static int
+_null_app_tx_callback (session_t *s)
+{
+  return 0;
+}
+
 static void
 application_verify_cb_fns (session_cb_vft_t * cb_fns)
 {
@@ -736,6 +742,8 @@ application_verify_cb_fns (session_cb_vft_t * cb_fns)
     clib_warning ("No session disconnect callback function provided");
   if (cb_fns->session_reset_callback == 0)
     clib_warning ("No session reset callback function provided");
+  if (!cb_fns->builtin_app_tx_callback)
+    cb_fns->builtin_app_tx_callback = _null_app_tx_callback;
 }
 
 /**
diff --git a/src/vnet/session/application.h b/src/vnet/session/application.h
index c3d6180022f..5505d91ea09 100644
--- a/src/vnet/session/application.h
+++ b/src/vnet/session/application.h
@@ -77,17 +77,17 @@ typedef struct app_worker_
   /** Pool of half-open session handles. Tracked in case worker detaches */
   session_handle_t *half_open_table;
 
+  /* Per vpp worker fifos of events for app worker */
+  session_event_t **wrk_evts;
+
+  /* Vector of vpp workers mq congestion flags */
+  u8 *wrk_mq_congested;
+
   /** Protects detached seg managers */
   clib_spinlock_t detached_seg_managers_lock;
 
   /** Vector of detached listener segment managers */
   u32 *detached_seg_managers;
-
-  /** Fifo of messages postponed because of mq congestion */
-  app_wrk_postponed_msg_t *postponed_mq_msgs;
-
-  /** Lock to add/sub message from ref @postponed_mq_msgs */
-  clib_spinlock_t postponed_mq_msgs_lock;
 } app_worker_t;
 
 typedef struct app_worker_map_
@@ -317,6 +317,12 @@ void application_enable_rx_mqs_nodes (u8 is_en);
  * App worker
  */
 
+always_inline u8
+app_worker_mq_is_congested (app_worker_t *app_wrk)
+{
+  return app_wrk->mq_congested > 0;
+}
+
 app_worker_t *app_worker_alloc (application_t * app);
 int application_alloc_worker_and_init (application_t * app,
 				       app_worker_t ** wrk);
@@ -331,6 +337,10 @@ session_error_t app_worker_start_listen (app_worker_t *app_wrk,
 					 app_listener_t *lstnr);
 int app_worker_stop_listen (app_worker_t * app_wrk, app_listener_t * al);
 int app_worker_init_accepted (session_t * s);
+int app_worker_listened_notify (app_worker_t *app_wrk, session_handle_t alsh,
+				u32 opaque, int err);
+int app_worker_unlisten_reply (app_worker_t *app_wrk, session_handle_t sh,
+			       u32 opaque, session_error_t err);
 int app_worker_accept_notify (app_worker_t * app_wrk, session_t * s);
 int app_worker_init_connected (app_worker_t * app_wrk, session_t * s);
 int app_worker_connect_notify (app_worker_t * app_wrk, session_t * s,
@@ -343,13 +353,21 @@ int app_worker_transport_closed_notify (app_worker_t * app_wrk,
 int app_worker_reset_notify (app_worker_t * app_wrk, session_t * s);
 int app_worker_cleanup_notify (app_worker_t * app_wrk, session_t * s,
 			       session_cleanup_ntf_t ntf);
+int app_worker_cleanup_notify_custom (app_worker_t *app_wrk, session_t *s,
+				      session_cleanup_ntf_t ntf,
+				      void (*cleanup_cb) (session_t *s));
 int app_worker_migrate_notify (app_worker_t * app_wrk, session_t * s,
 			       session_handle_t new_sh);
-int app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s);
-int app_worker_builtin_tx (app_worker_t * app_wrk, session_t * s);
+int app_worker_rx_notify (app_worker_t *app_wrk, session_t *s);
 int app_worker_session_fifo_tuning (app_worker_t * app_wrk, session_t * s,
 				    svm_fifo_t * f,
 				    session_ft_action_t act, u32 len);
+void app_worker_add_event (app_worker_t *app_wrk, session_t *s,
+			   session_evt_type_t evt_type);
+void app_worker_add_event_custom (app_worker_t *app_wrk, u32 thread_index,
+				  session_event_t *evt);
+int app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_del_all_events (app_worker_t *app_wrk);
 segment_manager_t *app_worker_get_listen_segment_manager (app_worker_t *,
 							  session_t *);
 segment_manager_t *app_worker_get_connect_segment_manager (app_worker_t *);
@@ -364,9 +382,10 @@ void app_wrk_send_ctrl_evt_fd (app_worker_t *app_wrk, u8 evt_type, void *msg,
 			       u32 msg_len, int fd);
 void app_wrk_send_ctrl_evt (app_worker_t *app_wrk, u8 evt_type, void *msg,
 			    u32 msg_len);
-int app_worker_send_event (app_worker_t * app, session_t * s, u8 evt);
-int app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
-				    u8 evt_type);
+u8 app_worker_mq_wrk_is_congested (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_set_mq_wrk_congested (app_worker_t *app_wrk, u32 thread_index);
+void app_worker_unset_wrk_mq_congested (app_worker_t *app_wrk,
+					u32 thread_index);
 session_t *app_worker_proxy_listener (app_worker_t * app, u8 fib_proto,
 				      u8 transport_proto);
 void app_worker_del_detached_sm (app_worker_t * app_wrk, u32 sm_index);
@@ -395,6 +414,12 @@ void sapi_socket_close_w_handle (u32 api_handle);
 crypto_engine_type_t app_crypto_engine_type_add (void);
 u8 app_crypto_engine_n_types (void);
 
+static inline u8
+app_worker_application_is_builtin (app_worker_t *app_wrk)
+{
+  return app_wrk->app_is_builtin;
+}
+
 #endif /* SRC_VNET_SESSION_APPLICATION_H_ */
 
 /*
diff --git a/src/vnet/session/application_local.c b/src/vnet/session/application_local.c
index 6ac4da2c655..192c22b659a 100644
--- a/src/vnet/session/application_local.c
+++ b/src/vnet/session/application_local.c
@@ -1028,6 +1028,17 @@ ct_close_is_reset (ct_connection_t *ct, session_t *s)
 }
 
 static void
+ct_session_cleanup_server_session (session_t *s)
+{
+  ct_connection_t *ct;
+
+  ct = (ct_connection_t *) session_get_transport (s);
+  ct_session_dealloc_fifos (ct, s->rx_fifo, s->tx_fifo);
+  session_free (s);
+  ct_connection_free (ct);
+}
+
+static void
 ct_session_postponed_cleanup (ct_connection_t *ct)
 {
   ct_connection_t *peer_ct;
@@ -1047,33 +1058,38 @@ ct_session_postponed_cleanup (ct_connection_t *ct)
     }
   session_transport_closed_notify (&ct->connection);
 
+  /* It would be cleaner to call session_transport_delete_notify
+   * but then we can't control session cleanup lower */
+  session_set_state (s, SESSION_STATE_TRANSPORT_DELETED);
+  if (app_wrk)
+    app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
+
   if (ct->flags & CT_CONN_F_CLIENT)
     {
-      if (app_wrk)
-	app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
-
       /* Normal free for client session as the fifos are allocated through
        * the connects segment manager in a segment that's not shared with
        * the server */
       ct_session_dealloc_fifos (ct, ct->client_rx_fifo, ct->client_tx_fifo);
-      session_free_w_fifos (s);
+      session_program_cleanup (s);
+      ct_connection_free (ct);
     }
   else
     {
       /* Manual session and fifo segment cleanup to avoid implicit
        * segment manager cleanups and notifications */
-      app_wrk = app_worker_get_if_valid (s->app_wrk_index);
       if (app_wrk)
 	{
-	  app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
-	  app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_SESSION);
+	  /* Remove custom cleanup notify infra when/if switching to normal
+	   * session cleanup. Note that ct is freed in the cb function */
+	  app_worker_cleanup_notify_custom (app_wrk, s,
+					    SESSION_CLEANUP_SESSION,
+					    ct_session_cleanup_server_session);
+	}
+      else
+	{
+	  ct_connection_free (ct);
 	}
-
-      ct_session_dealloc_fifos (ct, s->rx_fifo, s->tx_fifo);
-      session_free (s);
     }
-
-  ct_connection_free (ct);
 }
 
 static void
diff --git a/src/vnet/session/application_worker.c b/src/vnet/session/application_worker.c
index c3941d1fd7b..127963a1eda 100644
--- a/src/vnet/session/application_worker.c
+++ b/src/vnet/session/application_worker.c
@@ -26,6 +26,7 @@ app_worker_t *
 app_worker_alloc (application_t * app)
 {
   app_worker_t *app_wrk;
+
   pool_get (app_workers, app_wrk);
   clib_memset (app_wrk, 0, sizeof (*app_wrk));
   app_wrk->wrk_index = app_wrk - app_workers;
@@ -33,7 +34,8 @@ app_worker_alloc (application_t * app)
   app_wrk->wrk_map_index = ~0;
   app_wrk->connects_seg_manager = APP_INVALID_SEGMENT_MANAGER_INDEX;
   clib_spinlock_init (&app_wrk->detached_seg_managers_lock);
-  clib_spinlock_init (&app_wrk->postponed_mq_msgs_lock);
+  vec_validate (app_wrk->wrk_evts, vlib_num_workers ());
+  vec_validate (app_wrk->wrk_mq_congested, vlib_num_workers ());
   APP_DBG ("New app %v worker %u", app->name, app_wrk->wrk_index);
   return app_wrk;
 }
@@ -65,17 +67,25 @@ app_worker_free (app_worker_t * app_wrk)
   int i;
 
   /*
+   * Cleanup vpp wrk events
+   */
+  app_worker_del_all_events (app_wrk);
+  for (i = 0; i < vec_len (app_wrk->wrk_evts); i++)
+    clib_fifo_free (app_wrk->wrk_evts[i]);
+
+  vec_free (app_wrk->wrk_evts);
+  vec_free (app_wrk->wrk_mq_congested);
+
+  /*
    *  Listener cleanup
    */
 
-  /* *INDENT-OFF* */
   hash_foreach (handle, sm_index, app_wrk->listeners_table, ({
     ls = listen_session_get_from_handle (handle);
     vec_add1 (handles, app_listen_session_handle (ls));
     vec_add1 (sm_indices, sm_index);
     sm = segment_manager_get (sm_index);
   }));
-  /* *INDENT-ON* */
 
   for (i = 0; i < vec_len (handles); i++)
     {
@@ -127,7 +137,6 @@ app_worker_free (app_worker_t * app_wrk)
     }
   vec_free (app_wrk->detached_seg_managers);
   clib_spinlock_free (&app_wrk->detached_seg_managers_lock);
-  clib_spinlock_free (&app_wrk->postponed_mq_msgs_lock);
 
   if (CLIB_DEBUG)
     clib_memset (app_wrk, 0xfe, sizeof (*app_wrk));
@@ -339,7 +348,7 @@ app_worker_init_accepted (session_t * s)
 
   listener = listen_session_get_from_handle (s->listener_handle);
   app_wrk = application_listener_select_worker (listener);
-  if (PREDICT_FALSE (app_wrk->mq_congested))
+  if (PREDICT_FALSE (app_worker_mq_is_congested (app_wrk)))
     return -1;
 
   s->app_wrk_index = app_wrk->wrk_index;
@@ -355,10 +364,39 @@ app_worker_init_accepted (session_t * s)
 }
 
 int
+app_worker_listened_notify (app_worker_t *app_wrk, session_handle_t alsh,
+			    u32 opaque, session_error_t err)
+{
+  session_event_t evt;
+
+  evt.event_type = SESSION_CTRL_EVT_BOUND;
+  evt.session_handle = alsh;
+  evt.as_u64[1] = (u64) opaque << 32 | err;
+
+  app_worker_add_event_custom (app_wrk, 0 /* thread index */, &evt);
+
+  return 0;
+}
+
+int
+app_worker_unlisten_reply (app_worker_t *app_wrk, session_handle_t sh,
+			   u32 opaque, session_error_t err)
+{
+  session_event_t evt = {};
+
+  evt.event_type = SESSION_CTRL_EVT_UNLISTEN_REPLY;
+  evt.session_handle = sh;
+  evt.as_u64[1] = (u64) opaque << 32 | (u32) err;
+
+  app_worker_add_event_custom (app_wrk, 0 /* thread index */, &evt);
+  return 0;
+}
+
+int
 app_worker_accept_notify (app_worker_t * app_wrk, session_t * s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  return app->cb_fns.session_accept_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_CTRL_EVT_ACCEPTED);
+  return 0;
 }
 
 int
@@ -382,9 +420,16 @@ int
 app_worker_connect_notify (app_worker_t * app_wrk, session_t * s,
 			   session_error_t err, u32 opaque)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  return app->cb_fns.session_connected_callback (app_wrk->wrk_index, opaque,
-						 s, err);
+  session_event_t evt = {};
+  u32 thread_index;
+
+  evt.event_type = SESSION_CTRL_EVT_CONNECTED;
+  evt.session_index = s ? s->session_index : ~0;
+  evt.as_u64[1] = (u64) opaque << 32 | (u32) err;
+  thread_index = s ? s->thread_index : vlib_get_thread_index ();
+
+  app_worker_add_event_custom (app_wrk, thread_index, &evt);
+  return 0;
 }
 
 int
@@ -402,36 +447,28 @@ app_worker_add_half_open (app_worker_t *app_wrk, session_handle_t sh)
 int
 app_worker_del_half_open (app_worker_t *app_wrk, session_t *s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  ASSERT (session_vlib_thread_is_cl_thread ());
-  pool_put_index (app_wrk->half_open_table, s->ho_index);
-  if (app->cb_fns.half_open_cleanup_callback)
-    app->cb_fns.half_open_cleanup_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_CTRL_EVT_HALF_CLEANUP);
   return 0;
 }
 
 int
 app_worker_close_notify (app_worker_t * app_wrk, session_t * s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  app->cb_fns.session_disconnect_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_CTRL_EVT_DISCONNECTED);
   return 0;
 }
 
 int
 app_worker_transport_closed_notify (app_worker_t * app_wrk, session_t * s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  if (app->cb_fns.session_transport_closed_callback)
-    app->cb_fns.session_transport_closed_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_CTRL_EVT_TRANSPORT_CLOSED);
   return 0;
 }
 
 int
 app_worker_reset_notify (app_worker_t * app_wrk, session_t * s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  app->cb_fns.session_reset_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_CTRL_EVT_RESET);
   return 0;
 }
 
@@ -439,29 +476,37 @@ int
 app_worker_cleanup_notify (app_worker_t * app_wrk, session_t * s,
 			   session_cleanup_ntf_t ntf)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  if (app->cb_fns.session_cleanup_callback)
-    app->cb_fns.session_cleanup_callback (s, ntf);
+  session_event_t evt;
+
+  evt.event_type = SESSION_CTRL_EVT_CLEANUP;
+  evt.as_u64[0] = (u64) ntf << 32 | s->session_index;
+  evt.as_u64[1] = pointer_to_uword (session_cleanup);
+
+  app_worker_add_event_custom (app_wrk, s->thread_index, &evt);
+
   return 0;
 }
 
 int
-app_worker_builtin_rx (app_worker_t * app_wrk, session_t * s)
+app_worker_cleanup_notify_custom (app_worker_t *app_wrk, session_t *s,
+				  session_cleanup_ntf_t ntf,
+				  void (*cleanup_cb) (session_t *s))
 {
-  application_t *app = application_get (app_wrk->app_index);
-  app->cb_fns.builtin_app_rx_callback (s);
+  session_event_t evt;
+
+  evt.event_type = SESSION_CTRL_EVT_CLEANUP;
+  evt.as_u64[0] = (u64) ntf << 32 | s->session_index;
+  evt.as_u64[1] = pointer_to_uword (cleanup_cb);
+
+  app_worker_add_event_custom (app_wrk, s->thread_index, &evt);
+
   return 0;
 }
 
 int
-app_worker_builtin_tx (app_worker_t * app_wrk, session_t * s)
+app_worker_rx_notify (app_worker_t *app_wrk, session_t *s)
 {
-  application_t *app = application_get (app_wrk->app_index);
-
-  if (!app->cb_fns.builtin_app_tx_callback)
-    return 0;
-
-  app->cb_fns.builtin_app_tx_callback (s);
+  app_worker_add_event (app_wrk, s, SESSION_IO_EVT_RX);
   return 0;
 }
 
@@ -469,8 +514,13 @@ int
 app_worker_migrate_notify (app_worker_t * app_wrk, session_t * s,
 			   session_handle_t new_sh)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  app->cb_fns.session_migrate_callback (s, new_sh);
+  session_event_t evt;
+
+  evt.event_type = SESSION_CTRL_EVT_MIGRATED;
+  evt.session_index = s->session_index;
+  evt.as_u64[1] = new_sh;
+
+  app_worker_add_event_custom (app_wrk, s->thread_index, &evt);
   return 0;
 }
 
@@ -514,7 +564,7 @@ int
 app_worker_connect_session (app_worker_t *app_wrk, session_endpoint_cfg_t *sep,
 			    session_handle_t *rsh)
 {
-  if (PREDICT_FALSE (app_wrk->mq_congested))
+  if (PREDICT_FALSE (app_worker_mq_is_congested (app_wrk)))
     return SESSION_E_REFUSED;
 
   sep->app_wrk_index = app_wrk->wrk_index;
@@ -601,24 +651,27 @@ app_worker_proxy_listener (app_worker_t * app_wrk, u8 fib_proto,
 int
 app_worker_add_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 {
-  application_t *app = application_get (app_wrk->app_index);
+  session_event_t evt;
+
+  evt.event_type = SESSION_CTRL_EVT_APP_ADD_SEGMENT;
+  evt.as_u64[1] = segment_handle;
 
-  return app->cb_fns.add_segment_callback (app_wrk->wrk_index,
-					   segment_handle);
+  app_worker_add_event_custom (app_wrk, vlib_get_thread_index (), &evt);
+
+  return 0;
 }
 
 int
 app_worker_del_segment_notify (app_worker_t * app_wrk, u64 segment_handle)
 {
-  application_t *app = application_get (app_wrk->app_index);
-  return app->cb_fns.del_segment_callback (app_wrk->wrk_index,
-					   segment_handle);
-}
+  session_event_t evt;
 
-static inline u8
-app_worker_application_is_builtin (app_worker_t * app_wrk)
-{
-  return app_wrk->app_is_builtin;
+  evt.event_type = SESSION_CTRL_EVT_APP_DEL_SEGMENT;
+  evt.as_u64[1] = segment_handle;
+
+  app_worker_add_event_custom (app_wrk, vlib_get_thread_index (), &evt);
+
+  return 0;
 }
 
 static int
@@ -677,126 +730,38 @@ app_wrk_send_fd (app_worker_t *app_wrk, int fd)
   return 0;
 }
 
-static int
-mq_try_lock_and_alloc_msg (svm_msg_q_t *mq, session_mq_rings_e ring,
-			   svm_msg_q_msg_t *msg)
-{
-  int rv, n_try = 0;
-
-  while (n_try < 75)
-    {
-      rv = svm_msg_q_lock_and_alloc_msg_w_ring (mq, ring, SVM_Q_NOWAIT, msg);
-      if (!rv)
-	return 0;
-      /*
-       * Break the loop if mq is full, usually this is because the
-       * app has crashed or is hanging on somewhere.
-       */
-      if (rv != -1)
-	break;
-      n_try += 1;
-      usleep (1);
-    }
-
-  return -1;
-}
-
-typedef union app_wrk_mq_rpc_args_
-{
-  struct
-  {
-    u32 thread_index;
-    u32 app_wrk_index;
-  };
-  uword as_uword;
-} app_wrk_mq_rpc_ags_t;
-
-static int
-app_wrk_handle_mq_postponed_msgs (void *arg)
+void
+app_worker_add_event (app_worker_t *app_wrk, session_t *s,
+		      session_evt_type_t evt_type)
 {
-  svm_msg_q_msg_t _mq_msg, *mq_msg = &_mq_msg;
-  app_wrk_postponed_msg_t *pm;
-  app_wrk_mq_rpc_ags_t args;
-  u32 max_msg, n_msg = 0;
-  app_worker_t *app_wrk;
   session_event_t *evt;
-  svm_msg_q_t *mq;
-
-  args.as_uword = pointer_to_uword (arg);
-  app_wrk = app_worker_get_if_valid (args.app_wrk_index);
-  if (!app_wrk)
-    return 0;
-
-  mq = app_wrk->event_queue;
-
-  clib_spinlock_lock (&app_wrk->postponed_mq_msgs_lock);
-
-  max_msg = clib_min (32, clib_fifo_elts (app_wrk->postponed_mq_msgs));
-
-  while (n_msg < max_msg)
-    {
-      pm = clib_fifo_head (app_wrk->postponed_mq_msgs);
-      if (mq_try_lock_and_alloc_msg (mq, pm->ring, mq_msg))
-	break;
 
-      evt = svm_msg_q_msg_data (mq, mq_msg);
-      clib_memset (evt, 0, sizeof (*evt));
-      evt->event_type = pm->event_type;
-      clib_memcpy_fast (evt->data, pm->data, pm->len);
-
-      if (pm->fd != -1)
-	app_wrk_send_fd (app_wrk, pm->fd);
-
-      svm_msg_q_add_and_unlock (mq, mq_msg);
-
-      clib_fifo_advance_head (app_wrk->postponed_mq_msgs, 1);
-      n_msg += 1;
-    }
+  ASSERT (s->thread_index == vlib_get_thread_index ());
+  clib_fifo_add2 (app_wrk->wrk_evts[s->thread_index], evt);
+  evt->session_index = s->session_index;
+  evt->event_type = evt_type;
+  evt->postponed = 0;
 
-  if (!clib_fifo_elts (app_wrk->postponed_mq_msgs))
+  /* First event for this app_wrk. Schedule it for handling in session input */
+  if (clib_fifo_elts (app_wrk->wrk_evts[s->thread_index]) == 1)
     {
-      app_wrk->mq_congested = 0;
+      session_worker_t *wrk = session_main_get_worker (s->thread_index);
+      session_wrk_program_app_wrk_evts (wrk, app_wrk->wrk_index);
     }
-  else
-    {
-      session_send_rpc_evt_to_thread_force (
-	args.thread_index, app_wrk_handle_mq_postponed_msgs,
-	uword_to_pointer (args.as_uword, void *));
-    }
-
-  clib_spinlock_unlock (&app_wrk->postponed_mq_msgs_lock);
-
-  return 0;
 }
 
-static void
-app_wrk_add_mq_postponed_msg (app_worker_t *app_wrk, session_mq_rings_e ring,
-			      u8 evt_type, void *msg, u32 msg_len, int fd)
+void
+app_worker_add_event_custom (app_worker_t *app_wrk, u32 thread_index,
+			     session_event_t *evt)
 {
-  app_wrk_postponed_msg_t *pm;
-
-  clib_spinlock_lock (&app_wrk->postponed_mq_msgs_lock);
+  clib_fifo_add1 (app_wrk->wrk_evts[thread_index], *evt);
 
-  app_wrk->mq_congested = 1;
-
-  clib_fifo_add2 (app_wrk->postponed_mq_msgs, pm);
-  clib_memcpy_fast (pm->data, msg, msg_len);
-  pm->event_type = evt_type;
-  pm->ring = ring;
-  pm->len = msg_len;
-  pm->fd = fd;
-
-  if (clib_fifo_elts (app_wrk->postponed_mq_msgs) == 1)
+  /* First event for this app_wrk. Schedule it for handling in session input */
+  if (clib_fifo_elts (app_wrk->wrk_evts[thread_index]) == 1)
     {
-      app_wrk_mq_rpc_ags_t args = { .thread_index = vlib_get_thread_index (),
-				    .app_wrk_index = app_wrk->wrk_index };
-
-      session_send_rpc_evt_to_thread_force (
-	args.thread_index, app_wrk_handle_mq_postponed_msgs,
-	uword_to_pointer (args.as_uword, void *));
+      session_worker_t *wrk = session_main_get_worker (thread_index);
+      session_wrk_program_app_wrk_evts (wrk, app_wrk->wrk_index);
     }
-
-  clib_spinlock_unlock (&app_wrk->postponed_mq_msgs_lock);
 }
 
 always_inline void
@@ -806,14 +771,9 @@ app_wrk_send_ctrl_evt_inline (app_worker_t *app_wrk, u8 evt_type, void *msg,
   svm_msg_q_msg_t _mq_msg, *mq_msg = &_mq_msg;
   svm_msg_q_t *mq = app_wrk->event_queue;
   session_event_t *evt;
-  int rv;
-
-  if (PREDICT_FALSE (app_wrk->mq_congested))
-    goto handle_congestion;
 
-  rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_CTRL_EVT_RING, mq_msg);
-  if (PREDICT_FALSE (rv))
-    goto handle_congestion;
+  ASSERT (!svm_msg_q_or_ring_is_full (mq, SESSION_MQ_CTRL_EVT_RING));
+  *mq_msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_CTRL_EVT_RING);
 
   evt = svm_msg_q_msg_data (mq, mq_msg);
   clib_memset (evt, 0, sizeof (*evt));
@@ -823,14 +783,7 @@ app_wrk_send_ctrl_evt_inline (app_worker_t *app_wrk, u8 evt_type, void *msg,
   if (fd != -1)
     app_wrk_send_fd (app_wrk, fd);
 
-  svm_msg_q_add_and_unlock (mq, mq_msg);
-
-  return;
-
-handle_congestion:
-
-  app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_CTRL_EVT_RING, evt_type,
-				msg, msg_len, fd);
+  svm_msg_q_add_raw (mq, mq_msg);
 }
 
 void
@@ -847,109 +800,26 @@ app_wrk_send_ctrl_evt (app_worker_t *app_wrk, u8 evt_type, void *msg,
   app_wrk_send_ctrl_evt_inline (app_wrk, evt_type, msg, msg_len, -1);
 }
 
-static inline int
-app_send_io_evt_rx (app_worker_t * app_wrk, session_t * s)
+u8
+app_worker_mq_wrk_is_congested (app_worker_t *app_wrk, u32 thread_index)
 {
-  svm_msg_q_msg_t _mq_msg = { 0 }, *mq_msg = &_mq_msg;
-  session_event_t *evt;
-  svm_msg_q_t *mq;
-  u32 app_session;
-  int rv;
-
-  if (app_worker_application_is_builtin (app_wrk))
-    return app_worker_builtin_rx (app_wrk, s);
-
-  if (svm_fifo_has_event (s->rx_fifo))
-    return 0;
-
-  app_session = s->rx_fifo->shr->client_session_index;
-  mq = app_wrk->event_queue;
-
-  if (PREDICT_FALSE (app_wrk->mq_congested))
-    goto handle_congestion;
-
-  rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_IO_EVT_RING, mq_msg);
-
-  if (PREDICT_FALSE (rv))
-    goto handle_congestion;
-
-  evt = svm_msg_q_msg_data (mq, mq_msg);
-  evt->event_type = SESSION_IO_EVT_RX;
-  evt->session_index = app_session;
-
-  (void) svm_fifo_set_event (s->rx_fifo);
-
-  svm_msg_q_add_and_unlock (mq, mq_msg);
-
-  return 0;
-
-handle_congestion:
-
-  app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_IO_EVT_RING,
-				SESSION_IO_EVT_RX, &app_session,
-				sizeof (app_session), -1);
-  return -1;
+  return app_wrk->wrk_mq_congested[thread_index] > 0;
 }
 
-static inline int
-app_send_io_evt_tx (app_worker_t * app_wrk, session_t * s)
+void
+app_worker_set_mq_wrk_congested (app_worker_t *app_wrk, u32 thread_index)
 {
-  svm_msg_q_msg_t _mq_msg = { 0 }, *mq_msg = &_mq_msg;
-  session_event_t *evt;
-  svm_msg_q_t *mq;
-  u32 app_session;
-  int rv;
-
-  if (app_worker_application_is_builtin (app_wrk))
-    return app_worker_builtin_tx (app_wrk, s);
-
-  app_session = s->tx_fifo->shr->client_session_index;
-  mq = app_wrk->event_queue;
-
-  if (PREDICT_FALSE (app_wrk->mq_congested))
-    goto handle_congestion;
-
-  rv = mq_try_lock_and_alloc_msg (mq, SESSION_MQ_IO_EVT_RING, mq_msg);
-
-  if (PREDICT_FALSE (rv))
-    goto handle_congestion;
-
-  evt = svm_msg_q_msg_data (mq, mq_msg);
-  evt->event_type = SESSION_IO_EVT_TX;
-  evt->session_index = app_session;
-
-  svm_msg_q_add_and_unlock (mq, mq_msg);
-
-  return 0;
-
-handle_congestion:
-
-  app_wrk_add_mq_postponed_msg (app_wrk, SESSION_MQ_IO_EVT_RING,
-				SESSION_IO_EVT_TX, &app_session,
-				sizeof (app_session), -1);
-  return -1;
+  clib_atomic_fetch_add_relax (&app_wrk->mq_congested, 1);
+  ASSERT (thread_index == vlib_get_thread_index ());
+  app_wrk->wrk_mq_congested[thread_index] = 1;
 }
 
-/* *INDENT-OFF* */
-typedef int (app_send_evt_handler_fn) (app_worker_t *app,
-				       session_t *s);
-static app_send_evt_handler_fn * const app_send_evt_handler_fns[2] = {
-    app_send_io_evt_rx,
-    app_send_io_evt_tx,
-};
-/* *INDENT-ON* */
-
-/**
- * Send event to application
- *
- * Logic from queue perspective is blocking. However, if queue is full,
- * we return.
- */
-int
-app_worker_lock_and_send_event (app_worker_t * app, session_t * s,
-				u8 evt_type)
+void
+app_worker_unset_wrk_mq_congested (app_worker_t *app_wrk, u32 thread_index)
 {
-  return app_send_evt_handler_fns[evt_type] (app, s);
+  clib_atomic_fetch_sub_relax (&app_wrk->mq_congested, 1);
+  ASSERT (thread_index == vlib_get_thread_index ());
+  app_wrk->wrk_mq_congested[thread_index] = 0;
 }
 
 u8 *
diff --git a/src/vnet/session/segment_manager.c b/src/vnet/session/segment_manager.c
index ad0ba89a1dd..d459b73650a 100644
--- a/src/vnet/session/segment_manager.c
+++ b/src/vnet/session/segment_manager.c
@@ -866,7 +866,7 @@ segment_manager_dealloc_fifos (svm_fifo_t * rx_fifo, svm_fifo_t * tx_fifo)
 
   /* Thread that allocated the fifos must be the one to clean them up */
   ASSERT (rx_fifo->master_thread_index == vlib_get_thread_index () ||
-	  rx_fifo->refcnt > 1);
+	  rx_fifo->refcnt > 1 || vlib_thread_is_main_w_barrier ());
 
   /* It's possible to have no segment manager if the session was removed
    * as result of a detach. */
diff --git a/src/vnet/session/session.c b/src/vnet/session/session.c
index 228234ceefc..b494041f9eb 100644
--- a/src/vnet/session/session.c
+++ b/src/vnet/session/session.c
@@ -240,18 +240,26 @@ session_is_valid (u32 si, u8 thread_index)
       || s->session_state <= SESSION_STATE_LISTENING)
     return 1;
 
-  if (s->session_state == SESSION_STATE_CONNECTING &&
+  if ((s->session_state == SESSION_STATE_CONNECTING ||
+       s->session_state == SESSION_STATE_TRANSPORT_CLOSED) &&
       (s->flags & SESSION_F_HALF_OPEN))
     return 1;
 
   tc = session_get_transport (s);
-  if (s->connection_index != tc->c_index
-      || s->thread_index != tc->thread_index || tc->s_index != si)
+  if (s->connection_index != tc->c_index ||
+      s->thread_index != tc->thread_index || tc->s_index != si)
     return 0;
 
   return 1;
 }
 
+void
+session_cleanup (session_t *s)
+{
+  segment_manager_dealloc_fifos (s->rx_fifo, s->tx_fifo);
+  session_free (s);
+}
+
 static void
 session_cleanup_notify (session_t * s, session_cleanup_ntf_t ntf)
 {
@@ -259,16 +267,21 @@ session_cleanup_notify (session_t * s, session_cleanup_ntf_t ntf)
 
   app_wrk = app_worker_get_if_valid (s->app_wrk_index);
   if (!app_wrk)
-    return;
+    {
+      if (ntf == SESSION_CLEANUP_TRANSPORT)
+	return;
+
+      session_cleanup (s);
+      return;
+    }
   app_worker_cleanup_notify (app_wrk, s, ntf);
 }
 
 void
-session_free_w_fifos (session_t * s)
+session_program_cleanup (session_t *s)
 {
+  ASSERT (s->session_state == SESSION_STATE_TRANSPORT_DELETED);
   session_cleanup_notify (s, SESSION_CLEANUP_SESSION);
-  segment_manager_dealloc_fifos (s->rx_fifo, s->tx_fifo);
-  session_free (s);
 }
 
 /**
@@ -285,7 +298,7 @@ session_delete (session_t * s)
   if ((rv = session_lookup_del_session (s)))
     clib_warning ("session %u hash delete rv %d", s->session_index, rv);
 
-  session_free_w_fifos (s);
+  session_program_cleanup (s);
 }
 
 void
@@ -307,7 +320,7 @@ session_cleanup_half_open (session_handle_t ho_handle)
       transport_cleanup (session_get_transport_proto (ho),
 			 ho->connection_index, ho->app_index /* overloaded */);
     }
-  else
+  else if (ho->session_state != SESSION_STATE_TRANSPORT_DELETED)
     {
       /* Cleanup half-open session lookup table if need be */
       if (ho->session_state != SESSION_STATE_TRANSPORT_CLOSED)
@@ -333,7 +346,8 @@ session_half_open_free (session_t *ho)
   app_wrk = app_worker_get_if_valid (ho->app_wrk_index);
   if (app_wrk)
     app_worker_del_half_open (app_wrk, ho);
-  session_free (ho);
+  else
+    session_free (ho);
 }
 
 static void
@@ -354,6 +368,7 @@ session_half_open_delete_notify (transport_connection_t *tc)
       if (!(tc->flags & TRANSPORT_CONNECTION_F_NO_LOOKUP))
 	session_lookup_del_half_open (tc);
     }
+  session_set_state (ho, SESSION_STATE_TRANSPORT_DELETED);
 
   /* Notification from ctrl thread accepted without rpc */
   if (tc->thread_index == transport_cl_thread ())
@@ -558,10 +573,158 @@ session_fifo_tuning (session_t * s, svm_fifo_t * f,
     }
 }
 
+void
+session_wrk_program_app_wrk_evts (session_worker_t *wrk, u32 app_wrk_index)
+{
+  u8 need_interrupt;
+
+  ASSERT ((wrk - session_main.wrk) == vlib_get_thread_index ());
+  need_interrupt = clib_bitmap_is_zero (wrk->app_wrks_pending_ntf);
+  wrk->app_wrks_pending_ntf =
+    clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk_index, 1);
+
+  if (need_interrupt)
+    vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);
+}
+
+always_inline void
+session_program_io_event (app_worker_t *app_wrk, session_t *s,
+			  session_evt_type_t et, u8 is_cl)
+{
+  if (is_cl)
+    {
+      /* Special events for connectionless sessions */
+      et += SESSION_IO_EVT_BUILTIN_RX - SESSION_IO_EVT_RX;
+
+      ASSERT (s->thread_index == 0);
+      session_event_t evt = {
+	.event_type = et,
+	.session_handle = session_handle (s),
+      };
+
+      app_worker_add_event_custom (app_wrk, vlib_get_thread_index (), &evt);
+    }
+  else
+    {
+      app_worker_add_event (app_wrk, s, et);
+    }
+}
+
+static inline int
+session_notify_subscribers (u32 app_index, session_t *s, svm_fifo_t *f,
+			    session_evt_type_t evt_type)
+{
+  app_worker_t *app_wrk;
+  application_t *app;
+  u8 is_cl;
+  int i;
+
+  app = application_get (app_index);
+  if (!app)
+    return -1;
+
+  is_cl = s->thread_index != vlib_get_thread_index ();
+  for (i = 0; i < f->shr->n_subscribers; i++)
+    {
+      app_wrk = application_get_worker (app, f->shr->subscribers[i]);
+      if (!app_wrk)
+	continue;
+      session_program_io_event (app_wrk, s, evt_type, is_cl ? 1 : 0);
+    }
+
+  return 0;
+}
+
+always_inline int
+session_enqueue_notify_inline (session_t *s, u8 is_cl)
+{
+  app_worker_t *app_wrk;
+
+  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
+  if (PREDICT_FALSE (!app_wrk))
+    return -1;
+
+  session_program_io_event (app_wrk, s, SESSION_IO_EVT_RX, is_cl);
+
+  if (PREDICT_FALSE (svm_fifo_n_subscribers (s->rx_fifo)))
+    return session_notify_subscribers (app_wrk->app_index, s, s->rx_fifo,
+				       SESSION_IO_EVT_RX);
+
+  return 0;
+}
+
+int
+session_enqueue_notify (session_t *s)
+{
+  return session_enqueue_notify_inline (s, 0 /* is_cl */);
+}
+
+int
+session_enqueue_notify_cl (session_t *s)
+{
+  return session_enqueue_notify_inline (s, 1 /* is_cl */);
+}
+
+int
+session_dequeue_notify (session_t *s)
+{
+  app_worker_t *app_wrk;
+  u8 is_cl;
+
+  /* Unset as soon as event is requested */
+  svm_fifo_clear_deq_ntf (s->tx_fifo);
+
+  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
+  if (PREDICT_FALSE (!app_wrk))
+    return -1;
+
+  is_cl = s->session_state == SESSION_STATE_LISTENING ||
+	  s->session_state == SESSION_STATE_OPENED;
+  session_program_io_event (app_wrk, s, SESSION_IO_EVT_TX, is_cl ? 1 : 0);
+
+  if (PREDICT_FALSE (svm_fifo_n_subscribers (s->tx_fifo)))
+    return session_notify_subscribers (app_wrk->app_index, s, s->tx_fifo,
+				       SESSION_IO_EVT_TX);
+
+  return 0;
+}
+
+/**
+ * Flushes queue of sessions that are to be notified of new data
+ * enqueued events.
+ *
+ * @param transport_proto transport protocol for which queue to be flushed
+ * @param thread_index Thread index for which the flush is to be performed.
+ * @return 0 on success or a positive number indicating the number of
+ *         failures due to API queue being full.
+ */
+void
+session_main_flush_enqueue_events (transport_proto_t transport_proto,
+				   u32 thread_index)
+{
+  session_worker_t *wrk = session_main_get_worker (thread_index);
+  session_handle_t *handles;
+  session_t *s;
+  u32 i;
+
+  handles = wrk->session_to_enqueue[transport_proto];
+
+  for (i = 0; i < vec_len (handles); i++)
+    {
+      s = session_get_from_handle (handles[i]);
+      session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED,
+			   0 /* TODO/not needed */);
+      session_enqueue_notify_inline (s,
+				     s->thread_index != thread_index ? 1 : 0);
+    }
+
+  vec_reset_length (handles);
+  wrk->session_to_enqueue[transport_proto] = handles;
+}
+
 /*
- * Enqueue data for delivery to session peer. Does not notify peer of enqueue
- * event but on request can queue notification events for later delivery by
- * calling stream_server_flush_enqueue_events().
+ * Enqueue data for delivery to app. If requested, it queues app notification
+ * event for later delivery.
  *
  * @param tc Transport connection which is to be enqueued data
  * @param b Buffer to be enqueued
@@ -610,15 +773,14 @@ session_enqueue_stream_connection (transport_connection_t * tc,
 
   if (queue_event)
     {
-      /* Queue RX event on this fifo. Eventually these will need to be flushed
-       * by calling stream_server_flush_enqueue_events () */
-      session_worker_t *wrk;
-
-      wrk = session_main_get_worker (s->thread_index);
+      /* Queue RX event on this fifo. Eventually these will need to be
+       * flushed by calling @ref session_main_flush_enqueue_events () */
       if (!(s->flags & SESSION_F_RX_EVT))
 	{
+	  session_worker_t *wrk = session_main_get_worker (s->thread_index);
+	  ASSERT (s->thread_index == vlib_get_thread_index ());
 	  s->flags |= SESSION_F_RX_EVT;
-	  vec_add1 (wrk->session_to_enqueue[tc->proto], s->session_index);
+	  vec_add1 (wrk->session_to_enqueue[tc->proto], session_handle (s));
 	}
 
       session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED, 0);
@@ -627,10 +789,11 @@ session_enqueue_stream_connection (transport_connection_t * tc,
   return enqueued;
 }
 
-int
-session_enqueue_dgram_connection (session_t * s,
-				  session_dgram_hdr_t * hdr,
-				  vlib_buffer_t * b, u8 proto, u8 queue_event)
+always_inline int
+session_enqueue_dgram_connection_inline (session_t *s,
+					 session_dgram_hdr_t *hdr,
+					 vlib_buffer_t *b, u8 proto,
+					 u8 queue_event, u32 is_cl)
 {
   int rv;
 
@@ -639,12 +802,10 @@ session_enqueue_dgram_connection (session_t * s,
 
   if (PREDICT_TRUE (!(b->flags & VLIB_BUFFER_NEXT_PRESENT)))
     {
-      /* *INDENT-OFF* */
       svm_fifo_seg_t segs[2] = {
 	  { (u8 *) hdr, sizeof (*hdr) },
 	  { vlib_buffer_get_current (b), b->current_length }
       };
-      /* *INDENT-ON* */
 
       rv = svm_fifo_enqueue_segments (s->rx_fifo, segs, 2,
 				      0 /* allow_partial */ );
@@ -676,15 +837,16 @@ session_enqueue_dgram_connection (session_t * s,
 
   if (queue_event && rv > 0)
     {
-      /* Queue RX event on this fifo. Eventually these will need to be flushed
-       * by calling stream_server_flush_enqueue_events () */
-      session_worker_t *wrk;
-
-      wrk = session_main_get_worker (s->thread_index);
+      /* Queue RX event on this fifo. Eventually these will need to be
+       * flushed by calling @ref session_main_flush_enqueue_events () */
       if (!(s->flags & SESSION_F_RX_EVT))
 	{
+	  u32 thread_index =
+	    is_cl ? vlib_get_thread_index () : s->thread_index;
+	  session_worker_t *wrk = session_main_get_worker (thread_index);
+	  ASSERT (s->thread_index == vlib_get_thread_index () || is_cl);
 	  s->flags |= SESSION_F_RX_EVT;
-	  vec_add1 (wrk->session_to_enqueue[proto], s->session_index);
+	  vec_add1 (wrk->session_to_enqueue[proto], session_handle (s));
 	}
 
       session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED, 0);
@@ -693,6 +855,23 @@ session_enqueue_dgram_connection (session_t * s,
 }
 
 int
+session_enqueue_dgram_connection (session_t *s, session_dgram_hdr_t *hdr,
+				  vlib_buffer_t *b, u8 proto, u8 queue_event)
+{
+  return session_enqueue_dgram_connection_inline (s, hdr, b, proto,
+						  queue_event, 0 /* is_cl */);
+}
+
+int
+session_enqueue_dgram_connection_cl (session_t *s, session_dgram_hdr_t *hdr,
+				     vlib_buffer_t *b, u8 proto,
+				     u8 queue_event)
+{
+  return session_enqueue_dgram_connection_inline (s, hdr, b, proto,
+						  queue_event, 1 /* is_cl */);
+}
+
+int
 session_tx_fifo_peek_bytes (transport_connection_t * tc, u8 * buffer,
 			    u32 offset, u32 max_bytes)
 {
@@ -715,189 +894,6 @@ session_tx_fifo_dequeue_drop (transport_connection_t * tc, u32 max_bytes)
   return rv;
 }
 
-static inline int
-session_notify_subscribers (u32 app_index, session_t * s,
-			    svm_fifo_t * f, session_evt_type_t evt_type)
-{
-  app_worker_t *app_wrk;
-  application_t *app;
-  int i;
-
-  app = application_get (app_index);
-  if (!app)
-    return -1;
-
-  for (i = 0; i < f->shr->n_subscribers; i++)
-    {
-      app_wrk = application_get_worker (app, f->shr->subscribers[i]);
-      if (!app_wrk)
-	continue;
-      if (app_worker_lock_and_send_event (app_wrk, s, evt_type))
-	return -1;
-    }
-
-  return 0;
-}
-
-/**
- * Notify session peer that new data has been enqueued.
- *
- * @param s 	Stream session for which the event is to be generated.
- * @param lock 	Flag to indicate if call should lock message queue.
- *
- * @return 0 on success or negative number if failed to send notification.
- */
-static inline int
-session_enqueue_notify_inline (session_t * s)
-{
-  app_worker_t *app_wrk;
-  u32 session_index;
-  u8 n_subscribers;
-  u32 thread_index;
-
-  session_index = s->session_index;
-  thread_index = s->thread_index;
-  n_subscribers = svm_fifo_n_subscribers (s->rx_fifo);
-
-  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
-  if (PREDICT_FALSE (!app_wrk))
-    {
-      SESSION_DBG ("invalid s->app_index = %d", s->app_wrk_index);
-      return 0;
-    }
-
-  SESSION_EVT (SESSION_EVT_ENQ, s, svm_fifo_max_dequeue_prod (s->rx_fifo));
-
-  s->flags &= ~SESSION_F_RX_EVT;
-
-  /* Application didn't confirm accept yet */
-  if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING))
-    return 0;
-
-  if (PREDICT_FALSE (app_worker_lock_and_send_event (app_wrk, s,
-						     SESSION_IO_EVT_RX)))
-    return -1;
-
-  if (PREDICT_FALSE (n_subscribers))
-    {
-      s = session_get (session_index, thread_index);
-      return session_notify_subscribers (app_wrk->app_index, s,
-					 s->rx_fifo, SESSION_IO_EVT_RX);
-    }
-
-  return 0;
-}
-
-int
-session_enqueue_notify (session_t * s)
-{
-  return session_enqueue_notify_inline (s);
-}
-
-static void
-session_enqueue_notify_rpc (void *arg)
-{
-  u32 session_index = pointer_to_uword (arg);
-  session_t *s;
-
-  s = session_get_if_valid (session_index, vlib_get_thread_index ());
-  if (!s)
-    return;
-
-  session_enqueue_notify (s);
-}
-
-/**
- * Like session_enqueue_notify, but can be called from a thread that does not
- * own the session.
- */
-void
-session_enqueue_notify_thread (session_handle_t sh)
-{
-  u32 thread_index = session_thread_from_handle (sh);
-  u32 session_index = session_index_from_handle (sh);
-
-  /*
-   * Pass session index (u32) as opposed to handle (u64) in case pointers
-   * are not 64-bit.
-   */
-  session_send_rpc_evt_to_thread (thread_index,
-				  session_enqueue_notify_rpc,
-				  uword_to_pointer (session_index, void *));
-}
-
-int
-session_dequeue_notify (session_t * s)
-{
-  app_worker_t *app_wrk;
-
-  svm_fifo_clear_deq_ntf (s->tx_fifo);
-
-  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
-  if (PREDICT_FALSE (!app_wrk))
-    return -1;
-
-  if (PREDICT_FALSE (app_worker_lock_and_send_event (app_wrk, s,
-						     SESSION_IO_EVT_TX)))
-    return -1;
-
-  if (PREDICT_FALSE (s->tx_fifo->shr->n_subscribers))
-    return session_notify_subscribers (app_wrk->app_index, s,
-				       s->tx_fifo, SESSION_IO_EVT_TX);
-
-  return 0;
-}
-
-/**
- * Flushes queue of sessions that are to be notified of new data
- * enqueued events.
- *
- * @param thread_index Thread index for which the flush is to be performed.
- * @return 0 on success or a positive number indicating the number of
- *         failures due to API queue being full.
- */
-int
-session_main_flush_enqueue_events (u8 transport_proto, u32 thread_index)
-{
-  session_worker_t *wrk = session_main_get_worker (thread_index);
-  session_t *s;
-  int i, errors = 0;
-  u32 *indices;
-
-  indices = wrk->session_to_enqueue[transport_proto];
-
-  for (i = 0; i < vec_len (indices); i++)
-    {
-      s = session_get_if_valid (indices[i], thread_index);
-      if (PREDICT_FALSE (!s))
-	{
-	  errors++;
-	  continue;
-	}
-
-      session_fifo_tuning (s, s->rx_fifo, SESSION_FT_ACTION_ENQUEUED,
-			   0 /* TODO/not needed */ );
-
-      if (PREDICT_FALSE (session_enqueue_notify_inline (s)))
-	errors++;
-    }
-
-  vec_reset_length (indices);
-  wrk->session_to_enqueue[transport_proto] = indices;
-
-  return errors;
-}
-
-int
-session_main_flush_all_enqueue_events (u8 transport_proto)
-{
-  vlib_thread_main_t *vtm = vlib_get_thread_main ();
-  int i, errors = 0;
-  for (i = 0; i < 1 + vtm->n_threads; i++)
-    errors += session_main_flush_enqueue_events (transport_proto, i);
-  return errors;
-}
-
 int
 session_stream_connect_notify (transport_connection_t * tc,
 			       session_error_t err)
@@ -951,43 +947,20 @@ session_stream_connect_notify (transport_connection_t * tc,
   return 0;
 }
 
-typedef union session_switch_pool_reply_args_
-{
-  struct
-  {
-    u32 session_index;
-    u16 thread_index;
-    u8 is_closed;
-  };
-  u64 as_u64;
-} session_switch_pool_reply_args_t;
-
-STATIC_ASSERT (sizeof (session_switch_pool_reply_args_t) <= sizeof (uword),
-	       "switch pool reply args size");
-
 static void
-session_switch_pool_reply (void *arg)
+session_switch_pool_closed_rpc (void *arg)
 {
-  session_switch_pool_reply_args_t rargs;
+  session_handle_t sh;
   session_t *s;
 
-  rargs.as_u64 = pointer_to_uword (arg);
-  s = session_get_if_valid (rargs.session_index, rargs.thread_index);
+  sh = pointer_to_uword (arg);
+  s = session_get_from_handle_if_valid (sh);
   if (!s)
     return;
 
-  /* Session closed during migration. Clean everything up */
-  if (rargs.is_closed)
-    {
-      transport_cleanup (session_get_transport_proto (s), s->connection_index,
-			 s->thread_index);
-      segment_manager_dealloc_fifos (s->rx_fifo, s->tx_fifo);
-      session_free (s);
-      return;
-    }
-
-  /* Notify app that it has data on the new session */
-  session_enqueue_notify (s);
+  transport_cleanup (session_get_transport_proto (s), s->connection_index,
+		     s->thread_index);
+  session_cleanup (s);
 }
 
 typedef struct _session_switch_pool_args
@@ -1005,8 +978,7 @@ static void
 session_switch_pool (void *cb_args)
 {
   session_switch_pool_args_t *args = (session_switch_pool_args_t *) cb_args;
-  session_switch_pool_reply_args_t rargs;
-  session_handle_t new_sh;
+  session_handle_t sh, new_sh;
   segment_manager_t *sm;
   app_worker_t *app_wrk;
   session_t *s;
@@ -1014,37 +986,32 @@ session_switch_pool (void *cb_args)
   ASSERT (args->thread_index == vlib_get_thread_index ());
   s = session_get (args->session_index, args->thread_index);
 
-  /* Check if session closed during migration */
-  rargs.is_closed = s->session_state >= SESSION_STATE_TRANSPORT_CLOSING;
+  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
+  if (!app_wrk)
+    goto app_closed;
 
-  transport_cleanup (session_get_transport_proto (s), s->connection_index,
-		     s->thread_index);
+  /* Cleanup fifo segment slice state for fifos */
+  sm = app_worker_get_connect_segment_manager (app_wrk);
+  segment_manager_detach_fifo (sm, &s->rx_fifo);
+  segment_manager_detach_fifo (sm, &s->tx_fifo);
 
-  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
-  if (app_wrk)
-    {
-      /* Cleanup fifo segment slice state for fifos */
-      sm = app_worker_get_connect_segment_manager (app_wrk);
-      segment_manager_detach_fifo (sm, &s->rx_fifo);
-      segment_manager_detach_fifo (sm, &s->tx_fifo);
+  /* Check if session closed during migration */
+  if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
+    goto app_closed;
 
-      /* Notify app, using old session, about the migration event */
-      if (!rargs.is_closed)
-	{
-	  new_sh = session_make_handle (args->new_session_index,
-					args->new_thread_index);
-	  app_worker_migrate_notify (app_wrk, s, new_sh);
-	}
-    }
+  new_sh =
+    session_make_handle (args->new_session_index, args->new_thread_index);
+  app_worker_migrate_notify (app_wrk, s, new_sh);
 
-  /* Trigger app read and fifo updates on the new thread */
-  rargs.session_index = args->new_session_index;
-  rargs.thread_index = args->new_thread_index;
-  session_send_rpc_evt_to_thread (args->new_thread_index,
-				  session_switch_pool_reply,
-				  uword_to_pointer (rargs.as_u64, void *));
+  clib_mem_free (cb_args);
+  return;
 
-  session_free (s);
+app_closed:
+  /* Session closed during migration. Clean everything up */
+  sh = session_handle (s);
+  session_send_rpc_evt_to_thread (args->new_thread_index,
+				  session_switch_pool_closed_rpc,
+				  uword_to_pointer (sh, void *));
   clib_mem_free (cb_args);
 }
 
@@ -1184,6 +1151,7 @@ session_transport_delete_notify (transport_connection_t * tc)
       break;
     case SESSION_STATE_CLOSED:
       session_cleanup_notify (s, SESSION_CLEANUP_TRANSPORT);
+      session_set_state (s, SESSION_STATE_TRANSPORT_DELETED);
       session_delete (s);
       break;
     default:
@@ -1633,7 +1601,7 @@ session_transport_close (session_t * s)
 	session_set_state (s, SESSION_STATE_CLOSED);
       /* If transport is already deleted, just free the session */
       else if (s->session_state >= SESSION_STATE_TRANSPORT_DELETED)
-	session_free_w_fifos (s);
+	session_program_cleanup (s);
       return;
     }
 
@@ -1660,7 +1628,7 @@ session_transport_reset (session_t * s)
       if (s->session_state == SESSION_STATE_TRANSPORT_CLOSED)
 	session_set_state (s, SESSION_STATE_CLOSED);
       else if (s->session_state >= SESSION_STATE_TRANSPORT_DELETED)
-	session_free_w_fifos (s);
+	session_program_cleanup (s);
       return;
     }
 
@@ -2157,6 +2125,7 @@ session_node_enable_disable (u8 is_en)
 	  if (!sm->poll_main)
 	    continue;
 	}
+      vlib_node_set_state (vm, session_input_node.index, mstate);
       vlib_node_set_state (vm, session_queue_node.index, state);
     }
 
diff --git a/src/vnet/session/session.h b/src/vnet/session/session.h
index 6b6d1f64499..4de7bb252f0 100644
--- a/src/vnet/session/session.h
+++ b/src/vnet/session/session.h
@@ -100,8 +100,8 @@ typedef struct session_worker_
   /** Convenience pointer to this worker's vlib_main */
   vlib_main_t *vm;
 
-  /** Per-proto vector of sessions to enqueue */
-  u32 **session_to_enqueue;
+  /** Per-proto vector of session handles to enqueue */
+  session_handle_t **session_to_enqueue;
 
   /** Timerfd used to periodically signal wrk session queue node */
   int timerfd;
@@ -148,6 +148,9 @@ typedef struct session_worker_
   /** List head for first worker evts pending handling on main */
   clib_llist_index_t evts_pending_main;
 
+  /** Per-app-worker bitmap of pending notifications */
+  uword *app_wrks_pending_ntf;
+
   int config_index;
   u8 dma_enabled;
   session_dma_transfer *dma_trans;
@@ -275,6 +278,7 @@ typedef struct session_main_
 
 extern session_main_t session_main;
 extern vlib_node_registration_t session_queue_node;
+extern vlib_node_registration_t session_input_node;
 extern vlib_node_registration_t session_queue_process_node;
 extern vlib_node_registration_t session_queue_pre_input_node;
 
@@ -358,7 +362,8 @@ int session_wrk_handle_mq (session_worker_t *wrk, svm_msg_q_t *mq);
 
 session_t *session_alloc (u32 thread_index);
 void session_free (session_t * s);
-void session_free_w_fifos (session_t * s);
+void session_cleanup (session_t *s);
+void session_program_cleanup (session_t *s);
 void session_cleanup_half_open (session_handle_t ho_handle);
 u8 session_is_valid (u32 si, u8 thread_index);
 
@@ -452,8 +457,9 @@ void session_transport_reset (session_t * s);
 void session_transport_cleanup (session_t * s);
 int session_send_io_evt_to_thread (svm_fifo_t * f,
 				   session_evt_type_t evt_type);
-int session_enqueue_notify (session_t * s);
+int session_enqueue_notify (session_t *s);
 int session_dequeue_notify (session_t * s);
+int session_enqueue_notify_cl (session_t *s);
 int session_send_io_evt_to_thread_custom (void *data, u32 thread_index,
 					  session_evt_type_t evt_type);
 void session_send_rpc_evt_to_thread (u32 thread_index, void *fp,
@@ -485,6 +491,10 @@ int session_enqueue_dgram_connection (session_t * s,
 				      session_dgram_hdr_t * hdr,
 				      vlib_buffer_t * b, u8 proto,
 				      u8 queue_event);
+int session_enqueue_dgram_connection_cl (session_t *s,
+					 session_dgram_hdr_t *hdr,
+					 vlib_buffer_t *b, u8 proto,
+					 u8 queue_event);
 int session_stream_connect_notify (transport_connection_t * tc,
 				   session_error_t err);
 int session_dgram_connect_notify (transport_connection_t * tc,
@@ -502,6 +512,7 @@ int session_stream_accept (transport_connection_t * tc, u32 listener_index,
 			   u32 thread_index, u8 notify);
 int session_dgram_accept (transport_connection_t * tc, u32 listener_index,
 			  u32 thread_index);
+
 /**
  * Initialize session layer for given transport proto and ip version
  *
@@ -765,8 +776,8 @@ do {									\
       return clib_error_return (0, "session layer is not enabled");	\
 } while (0)
 
-int session_main_flush_enqueue_events (u8 proto, u32 thread_index);
-int session_main_flush_all_enqueue_events (u8 transport_proto);
+void session_main_flush_enqueue_events (transport_proto_t transport_proto,
+					u32 thread_index);
 void session_queue_run_on_main_thread (vlib_main_t * vm);
 
 /**
@@ -799,6 +810,8 @@ fifo_segment_t *session_main_get_wrk_mqs_segment (void);
 void session_node_enable_disable (u8 is_en);
 clib_error_t *vnet_session_enable_disable (vlib_main_t * vm, u8 is_en);
 void session_wrk_handle_evts_main_rpc (void *);
+void session_wrk_program_app_wrk_evts (session_worker_t *wrk,
+				       u32 app_wrk_index);
 
 session_t *session_alloc_for_connection (transport_connection_t * tc);
 session_t *session_alloc_for_half_open (transport_connection_t *tc);
diff --git a/src/vnet/session/session_api.c b/src/vnet/session/session_api.c
index ff11bcb690a..55fc72ee4c2 100644
--- a/src/vnet/session/session_api.c
+++ b/src/vnet/session/session_api.c
@@ -460,6 +460,52 @@ mq_send_session_cleanup_cb (session_t * s, session_cleanup_ntf_t ntf)
   app_wrk_send_ctrl_evt (app_wrk, SESSION_CTRL_EVT_CLEANUP, &m, sizeof (m));
 }
 
+static int
+mq_send_io_rx_event (session_t *s)
+{
+  session_event_t *mq_evt;
+  svm_msg_q_msg_t mq_msg;
+  app_worker_t *app_wrk;
+  svm_msg_q_t *mq;
+
+  if (svm_fifo_has_event (s->rx_fifo))
+    return 0;
+
+  app_wrk = app_worker_get (s->app_wrk_index);
+  mq = app_wrk->event_queue;
+
+  mq_msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+  mq_evt = svm_msg_q_msg_data (mq, &mq_msg);
+
+  mq_evt->event_type = SESSION_IO_EVT_RX;
+  mq_evt->session_index = s->rx_fifo->shr->client_session_index;
+
+  (void) svm_fifo_set_event (s->rx_fifo);
+
+  svm_msg_q_add_raw (mq, &mq_msg);
+
+  return 0;
+}
+
+static int
+mq_send_io_tx_event (session_t *s)
+{
+  app_worker_t *app_wrk = app_worker_get (s->app_wrk_index);
+  svm_msg_q_t *mq = app_wrk->event_queue;
+  session_event_t *mq_evt;
+  svm_msg_q_msg_t mq_msg;
+
+  mq_msg = svm_msg_q_alloc_msg_w_ring (mq, SESSION_MQ_IO_EVT_RING);
+  mq_evt = svm_msg_q_msg_data (mq, &mq_msg);
+
+  mq_evt->event_type = SESSION_IO_EVT_TX;
+  mq_evt->session_index = s->tx_fifo->shr->client_session_index;
+
+  svm_msg_q_add_raw (mq, &mq_msg);
+
+  return 0;
+}
+
 static session_cb_vft_t session_mq_cb_vft = {
   .session_accept_callback = mq_send_session_accepted_cb,
   .session_disconnect_callback = mq_send_session_disconnected_cb,
@@ -469,6 +515,8 @@ static session_cb_vft_t session_mq_cb_vft = {
   .session_cleanup_callback = mq_send_session_cleanup_cb,
   .add_segment_callback = mq_send_add_segment_cb,
   .del_segment_callback = mq_send_del_segment_cb,
+  .builtin_app_rx_callback = mq_send_io_rx_event,
+  .builtin_app_tx_callback = mq_send_io_tx_event,
 };
 
 static void
@@ -1246,6 +1294,8 @@ static session_cb_vft_t session_mq_sapi_cb_vft = {
   .session_cleanup_callback = mq_send_session_cleanup_cb,
   .add_segment_callback = mq_send_add_segment_sapi_cb,
   .del_segment_callback = mq_send_del_segment_sapi_cb,
+  .builtin_app_rx_callback = mq_send_io_rx_event,
+  .builtin_app_tx_callback = mq_send_io_tx_event,
 };
 
 static void
diff --git a/src/vnet/session/session_input.c b/src/vnet/session/session_input.c
new file mode 100644
index 00000000000..8c1f11ca96e
--- /dev/null
+++ b/src/vnet/session/session_input.c
@@ -0,0 +1,296 @@
+/* SPDX-License-Identifier: Apache-2.0
+ * Copyright(c) 2023 Cisco Systems, Inc.
+ */
+
+#include <vnet/session/session.h>
+#include <vnet/session/application.h>
+
+static inline int
+mq_try_lock (svm_msg_q_t *mq)
+{
+  int rv, n_try = 0;
+
+  while (n_try < 100)
+    {
+      rv = svm_msg_q_try_lock (mq);
+      if (!rv)
+	return 0;
+      n_try += 1;
+      usleep (1);
+    }
+
+  return -1;
+}
+
+always_inline u8
+mq_event_ring_index (session_evt_type_t et)
+{
+  return (et >= SESSION_CTRL_EVT_RPC ? SESSION_MQ_CTRL_EVT_RING :
+					     SESSION_MQ_IO_EVT_RING);
+}
+
+void
+app_worker_del_all_events (app_worker_t *app_wrk)
+{
+  session_worker_t *wrk;
+  session_event_t *evt;
+  u32 thread_index;
+  session_t *s;
+
+  for (thread_index = 0; thread_index < vec_len (app_wrk->wrk_evts);
+       thread_index++)
+    {
+      while (clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
+	{
+	  clib_fifo_sub2 (app_wrk->wrk_evts[thread_index], evt);
+	  switch (evt->event_type)
+	    {
+	    case SESSION_CTRL_EVT_MIGRATED:
+	      s = session_get (evt->session_index, thread_index);
+	      transport_cleanup (session_get_transport_proto (s),
+				 s->connection_index, s->thread_index);
+	      session_free (s);
+	      break;
+	    case SESSION_CTRL_EVT_CLEANUP:
+	      s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
+	      if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
+		break;
+	      uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
+	      break;
+	    case SESSION_CTRL_EVT_HALF_CLEANUP:
+	      s = ho_session_get (evt->session_index);
+	      pool_put_index (app_wrk->half_open_table, s->ho_index);
+	      session_free (s);
+	      break;
+	    default:
+	      break;
+	    }
+	}
+      wrk = session_main_get_worker (thread_index);
+      clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
+    }
+}
+
+always_inline int
+app_worker_flush_events_inline (app_worker_t *app_wrk, u32 thread_index,
+				u8 is_builtin)
+{
+  application_t *app = application_get (app_wrk->app_index);
+  svm_msg_q_t *mq = app_wrk->event_queue;
+  session_event_t *evt;
+  u32 n_evts = 128, i;
+  u8 ring_index, mq_is_cong;
+  session_t *s;
+
+  n_evts = clib_min (n_evts, clib_fifo_elts (app_wrk->wrk_evts[thread_index]));
+
+  if (!is_builtin)
+    {
+      mq_is_cong = app_worker_mq_is_congested (app_wrk);
+      if (mq_try_lock (mq))
+	{
+	  app_worker_set_mq_wrk_congested (app_wrk, thread_index);
+	  return 0;
+	}
+    }
+
+  for (i = 0; i < n_evts; i++)
+    {
+      evt = clib_fifo_head (app_wrk->wrk_evts[thread_index]);
+      if (!is_builtin)
+	{
+	  ring_index = mq_event_ring_index (evt->event_type);
+	  if (svm_msg_q_or_ring_is_full (mq, ring_index))
+	    {
+	      app_worker_set_mq_wrk_congested (app_wrk, thread_index);
+	      break;
+	    }
+	}
+
+      switch (evt->event_type)
+	{
+	case SESSION_IO_EVT_RX:
+	  s = session_get (evt->session_index, thread_index);
+	  s->flags &= ~SESSION_F_RX_EVT;
+	  /* Application didn't confirm accept yet */
+	  if (PREDICT_FALSE (s->session_state == SESSION_STATE_ACCEPTING))
+	    break;
+	  app->cb_fns.builtin_app_rx_callback (s);
+	  break;
+	/* Handle sessions that might not be on current thread */
+	case SESSION_IO_EVT_BUILTIN_RX:
+	  s = session_get_from_handle_if_valid (evt->session_handle);
+	  if (!s || s->session_state == SESSION_STATE_ACCEPTING)
+	    break;
+	  app->cb_fns.builtin_app_rx_callback (s);
+	  break;
+	case SESSION_IO_EVT_TX:
+	  s = session_get (evt->session_index, thread_index);
+	  app->cb_fns.builtin_app_tx_callback (s);
+	  break;
+	case SESSION_IO_EVT_TX_MAIN:
+	  s = session_get_from_handle_if_valid (evt->session_handle);
+	  if (!s)
+	    break;
+	  app->cb_fns.builtin_app_tx_callback (s);
+	  break;
+	case SESSION_CTRL_EVT_BOUND:
+	  /* No app cb function currently */
+	  if (is_builtin)
+	    break;
+	  mq_send_session_bound_cb (app_wrk->wrk_index, evt->as_u64[1] >> 32,
+				    evt->session_handle,
+				    evt->as_u64[1] & 0xffffffff);
+	  break;
+	case SESSION_CTRL_EVT_ACCEPTED:
+	  s = session_get (evt->session_index, thread_index);
+	  app->cb_fns.session_accept_callback (s);
+	  break;
+	case SESSION_CTRL_EVT_CONNECTED:
+	  if (!(evt->as_u64[1] & 0xffffffff))
+	    s = session_get (evt->session_index, thread_index);
+	  else
+	    s = 0;
+	  app->cb_fns.session_connected_callback (app_wrk->wrk_index,
+						  evt->as_u64[1] >> 32, s,
+						  evt->as_u64[1] & 0xffffffff);
+	  break;
+	case SESSION_CTRL_EVT_DISCONNECTED:
+	  s = session_get (evt->session_index, thread_index);
+	  app->cb_fns.session_disconnect_callback (s);
+	  break;
+	case SESSION_CTRL_EVT_RESET:
+	  s = session_get (evt->session_index, thread_index);
+	  app->cb_fns.session_reset_callback (s);
+	  break;
+	case SESSION_CTRL_EVT_UNLISTEN_REPLY:
+	  if (is_builtin)
+	    break;
+	  mq_send_unlisten_reply (app_wrk, evt->session_handle,
+				  evt->as_u64[1] >> 32,
+				  evt->as_u64[1] & 0xffffffff);
+	  break;
+	case SESSION_CTRL_EVT_MIGRATED:
+	  s = session_get (evt->session_index, thread_index);
+	  app->cb_fns.session_migrate_callback (s, evt->as_u64[1]);
+	  transport_cleanup (session_get_transport_proto (s),
+			     s->connection_index, s->thread_index);
+	  session_free (s);
+	  /* Notify app that it has data on the new session */
+	  s = session_get_from_handle (evt->as_u64[1]);
+	  session_send_io_evt_to_thread (s->rx_fifo,
+					 SESSION_IO_EVT_BUILTIN_RX);
+	  break;
+	case SESSION_CTRL_EVT_TRANSPORT_CLOSED:
+	  s = session_get (evt->session_index, thread_index);
+	  if (app->cb_fns.session_transport_closed_callback)
+	    app->cb_fns.session_transport_closed_callback (s);
+	  break;
+	case SESSION_CTRL_EVT_CLEANUP:
+	  s = session_get (evt->as_u64[0] & 0xffffffff, thread_index);
+	  if (app->cb_fns.session_cleanup_callback)
+	    app->cb_fns.session_cleanup_callback (s, evt->as_u64[0] >> 32);
+	  if (evt->as_u64[0] >> 32 != SESSION_CLEANUP_SESSION)
+	    break;
+	  uword_to_pointer (evt->as_u64[1], void (*) (session_t * s)) (s);
+	  break;
+	case SESSION_CTRL_EVT_HALF_CLEANUP:
+	  s = ho_session_get (evt->session_index);
+	  ASSERT (session_vlib_thread_is_cl_thread ());
+	  if (app->cb_fns.half_open_cleanup_callback)
+	    app->cb_fns.half_open_cleanup_callback (s);
+	  pool_put_index (app_wrk->half_open_table, s->ho_index);
+	  session_free (s);
+	  break;
+	case SESSION_CTRL_EVT_APP_ADD_SEGMENT:
+	  app->cb_fns.add_segment_callback (app_wrk->wrk_index,
+					    evt->as_u64[1]);
+	  break;
+	case SESSION_CTRL_EVT_APP_DEL_SEGMENT:
+	  app->cb_fns.del_segment_callback (app_wrk->wrk_index,
+					    evt->as_u64[1]);
+	  break;
+	default:
+	  clib_warning ("unexpected event: %u", evt->event_type);
+	  ASSERT (0);
+	  break;
+	}
+      clib_fifo_advance_head (app_wrk->wrk_evts[thread_index], 1);
+    }
+
+  if (!is_builtin)
+    {
+      svm_msg_q_unlock (mq);
+      if (mq_is_cong && i == n_evts)
+	app_worker_unset_wrk_mq_congested (app_wrk, thread_index);
+    }
+
+  return 0;
+}
+
+int
+app_wrk_flush_wrk_events (app_worker_t *app_wrk, u32 thread_index)
+{
+  if (app_worker_application_is_builtin (app_wrk))
+    return app_worker_flush_events_inline (app_wrk, thread_index,
+					   1 /* is_builtin */);
+  else
+    return app_worker_flush_events_inline (app_wrk, thread_index,
+					   0 /* is_builtin */);
+}
+
+static inline int
+session_wrk_flush_events (session_worker_t *wrk)
+{
+  app_worker_t *app_wrk;
+  uword app_wrk_index;
+  u32 thread_index;
+
+  thread_index = wrk->vm->thread_index;
+  app_wrk_index = clib_bitmap_first_set (wrk->app_wrks_pending_ntf);
+
+  while (app_wrk_index != ~0)
+    {
+      app_wrk = app_worker_get_if_valid (app_wrk_index);
+      /* app_wrk events are flushed on free, so should be valid here */
+      ASSERT (app_wrk != 0);
+      app_wrk_flush_wrk_events (app_wrk, thread_index);
+
+      if (!clib_fifo_elts (app_wrk->wrk_evts[thread_index]))
+	clib_bitmap_set (wrk->app_wrks_pending_ntf, app_wrk->wrk_index, 0);
+
+      app_wrk_index =
+	clib_bitmap_next_set (wrk->app_wrks_pending_ntf, app_wrk_index + 1);
+    }
+
+  if (!clib_bitmap_is_zero (wrk->app_wrks_pending_ntf))
+    vlib_node_set_interrupt_pending (wrk->vm, session_input_node.index);
+
+  return 0;
+}
+
+VLIB_NODE_FN (session_input_node)
+(vlib_main_t *vm, vlib_node_runtime_t *node, vlib_frame_t *frame)
+{
+  u32 thread_index = vm->thread_index;
+  session_worker_t *wrk;
+
+  wrk = session_main_get_worker (thread_index);
+  session_wrk_flush_events (wrk);
+
+  return 0;
+}
+
+VLIB_REGISTER_NODE (session_input_node) = {
+  .name = "session-input",
+  .type = VLIB_NODE_TYPE_INPUT,
+  .state = VLIB_NODE_STATE_DISABLED,
+};
+
+/*
+ * fd.io coding-style-patch-verification: ON
+ *
+ * Local Variables:
+ * eval: (c-set-style "gnu")
+ * End:
+ */
\ No newline at end of file
diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index e15625e37ca..4f2cae4d196 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -142,10 +142,14 @@ session_mq_listen_handler (session_worker_t *wrk, session_evt_elt_t *elt)
     session_worker_stat_error_inc (wrk, rv, 1);
 
   app_wrk = application_get_worker (app, mp->wrk_index);
-  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+  app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
 
   if (mp->ext_config)
     session_mq_free_ext_config (app, mp->ext_config);
+
+  /* Make sure events are flushed before releasing barrier, to avoid
+   * potential race with accept. */
+  app_wrk_flush_wrk_events (app_wrk, 0);
 }
 
 static void
@@ -170,7 +174,8 @@ session_mq_listen_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
   rv = vnet_bind_uri (a);
 
   app_wrk = application_get_worker (app, 0);
-  mq_send_session_bound_cb (app_wrk->wrk_index, mp->context, a->handle, rv);
+  app_worker_listened_notify (app_wrk, a->handle, mp->context, rv);
+  app_wrk_flush_wrk_events (app_wrk, 0);
 }
 
 static void
@@ -215,7 +220,7 @@ session_mq_connect_one (session_connect_msg_t *mp)
       wrk = session_main_get_worker (vlib_get_thread_index ());
       session_worker_stat_error_inc (wrk, rv, 1);
       app_wrk = application_get_worker (app, mp->wrk_index);
-      mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
+      app_worker_connect_notify (app_wrk, 0, rv, mp->context);
     }
 
   if (mp->ext_config)
@@ -324,7 +329,7 @@ session_mq_connect_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
     {
       session_worker_stat_error_inc (wrk, rv, 1);
       app_wrk = application_get_worker (app, 0 /* default wrk only */ );
-      mq_send_session_connected_cb (app_wrk->wrk_index, mp->context, 0, rv);
+      app_worker_connect_notify (app_wrk, 0, rv, mp->context);
     }
 }
 
@@ -410,7 +415,7 @@ session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
   if (!app_wrk)
     return;
 
-  mq_send_unlisten_reply (app_wrk, sh, mp->context, rv);
+  app_worker_unlisten_reply (app_wrk, sh, mp->context, rv);
 }
 
 static void
@@ -466,7 +471,7 @@ session_mq_accepted_reply_handler (session_worker_t *wrk,
   session_set_state (s, SESSION_STATE_READY);
 
   if (!svm_fifo_is_empty_prod (s->rx_fifo))
-    app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
+    app_worker_rx_notify (app_wrk, s);
 
   /* Closed while waiting for app to reply. Resend disconnect */
   if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
@@ -669,7 +674,7 @@ session_mq_worker_update_handler (void *data)
     session_send_io_evt_to_thread (s->tx_fifo, SESSION_IO_EVT_TX);
 
   if (s->rx_fifo && !svm_fifo_is_empty (s->rx_fifo))
-    app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
+    app_worker_rx_notify (app_wrk, s);
 
   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
     app_worker_close_notify (app_wrk, s);
@@ -1790,7 +1795,7 @@ session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
 	break;
       svm_fifo_unset_event (s->rx_fifo);
       app_wrk = app_worker_get (s->app_wrk_index);
-      app_worker_builtin_rx (app_wrk, s);
+      app_worker_rx_notify (app_wrk, s);
       break;
     case SESSION_IO_EVT_TX_MAIN:
       s = session_get_if_valid (e->session_index, 0 /* main thread */);
diff --git a/src/vnet/session/session_types.h b/src/vnet/session/session_types.h
index 8755a146bce..4fe0c7cf09a 100644
--- a/src/vnet/session/session_types.h
+++ b/src/vnet/session/session_types.h
@@ -379,6 +379,8 @@ typedef enum
   SESSION_CTRL_EVT_APP_WRK_RPC,
   SESSION_CTRL_EVT_TRANSPORT_ATTR,
   SESSION_CTRL_EVT_TRANSPORT_ATTR_REPLY,
+  SESSION_CTRL_EVT_TRANSPORT_CLOSED,
+  SESSION_CTRL_EVT_HALF_CLEANUP,
 } session_evt_type_t;
 
 #define foreach_session_ctrl_evt                                              \
@@ -437,6 +439,7 @@ typedef struct
     session_handle_t session_handle;
     session_rpc_args_t rpc_args;
     u32 ctrl_data_index;
+    u64 as_u64[2];
     struct
     {
       u8 data[0];