1 files changed, 456 insertions, 230 deletions

diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 981fa66eae1..0ec158fb429 100644
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -26,12 +26,28 @@
 #include <svm/queue.h>
 #include <sys/timerfd.h>
 
-#define app_check_thread_and_barrier(_fn, _arg)				\
-  if (!vlib_thread_is_main_w_barrier ())				\
-    {									\
-     vlib_rpc_call_main_thread (_fn, (u8 *) _arg, sizeof(*_arg));	\
-      return;								\
-   }
+static inline void
+session_wrk_send_evt_to_main (session_worker_t *wrk, session_evt_elt_t *elt)
+{
+  session_evt_elt_t *he;
+  uword thread_index;
+  u8 is_empty;
+
+  thread_index = wrk->vm->thread_index;
+  he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
+  is_empty = clib_llist_is_empty (wrk->event_elts, evt_list, he);
+  clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
+  if (is_empty)
+    session_send_rpc_evt_to_thread (0, session_wrk_handle_evts_main_rpc,
+				    uword_to_pointer (thread_index, void *));
+}
+
+#define app_check_thread_and_barrier(_wrk, _elt)                              \
+  if (!vlib_thread_is_main_w_barrier ())                                      \
+    {                                                                         \
+      session_wrk_send_evt_to_main (wrk, elt);                                \
+      return;                                                                 \
+    }
 
 static void
 session_wrk_timerfd_update (session_worker_t *wrk, u64 time_ns)
@@ -93,16 +109,17 @@ session_mq_free_ext_config (application_t *app, uword offset)
 }
 
 static void
-session_mq_listen_handler (void *data)
+session_mq_listen_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 {
-  session_listen_msg_t *mp = (session_listen_msg_t *) data;
   vnet_listen_args_t _a, *a = &_a;
+  session_listen_msg_t *mp;
   app_worker_t *app_wrk;
   application_t *app;
   int rv;
 
-  app_check_thread_and_barrier (session_mq_listen_handler, mp);
+  app_check_thread_and_barrier (wrk, elt);
 
+  mp = session_evt_ctrl_data (wrk, elt);
   app = application_lookup (mp->client_index);
   if (!app)
     return;
@@ -122,26 +139,31 @@ session_mq_listen_handler (void *data)
     a->sep_ext.ext_cfg = session_mq_get_ext_config (app, mp->ext_config);
 
   if ((rv = vnet_listen (a)))
-    clib_warning ("listen returned: %U", format_session_error, rv);
+    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
-session_mq_listen_uri_handler (void *data)
+session_mq_listen_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 {
-  session_listen_uri_msg_t *mp = (session_listen_uri_msg_t *) data;
   vnet_listen_args_t _a, *a = &_a;
+  session_listen_uri_msg_t *mp;
   app_worker_t *app_wrk;
   application_t *app;
   int rv;
 
-  app_check_thread_and_barrier (session_mq_listen_uri_handler, mp);
+  app_check_thread_and_barrier (wrk, elt);
 
+  mp = session_evt_ctrl_data (wrk, elt);
   app = application_lookup (mp->client_index);
   if (!app)
     return;
@@ -152,7 +174,8 @@ session_mq_listen_uri_handler (void *data)
   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
@@ -160,6 +183,7 @@ session_mq_connect_one (session_connect_msg_t *mp)
 {
   vnet_connect_args_t _a, *a = &_a;
   app_worker_t *app_wrk;
+  session_worker_t *wrk;
   application_t *app;
   int rv;
 
@@ -173,6 +197,7 @@ session_mq_connect_one (session_connect_msg_t *mp)
   a->sep.port = mp->port;
   a->sep.transport_proto = mp->proto;
   a->sep.peer.fib_index = mp->vrf;
+  a->sep.dscp = mp->dscp;
   clib_memcpy_fast (&a->sep.peer.ip, &mp->lcl_ip, sizeof (mp->lcl_ip));
   if (mp->is_ip4)
     {
@@ -192,9 +217,10 @@ session_mq_connect_one (session_connect_msg_t *mp)
 
   if ((rv = vnet_connect (a)))
     {
-      clib_warning ("connect returned: %U", format_session_error, rv);
+      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)
@@ -205,23 +231,20 @@ static void
 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, *wrk;
+  session_worker_t *fwrk;
 
-  ASSERT (vlib_get_thread_index () == 0);
+  ASSERT (session_vlib_thread_is_cl_thread ());
 
   /* Pending connects on linked list pertaining to first worker */
-  fwrk = session_main_get_worker (1);
+  fwrk = session_main_get_worker (transport_cl_thread ());
   if (!fwrk->n_pending_connects)
-    goto update_state;
-
-  vlib_worker_thread_barrier_sync (vm);
+    return;
 
   he = clib_llist_elt (fwrk->event_elts, fwrk->pending_connects);
   elt = clib_llist_next (fwrk->event_elts, evt_list, he);
 
-  /* Avoid holding the barrier for too long */
+  /* Avoid holding the worker for too long */
   while (n_connects < max_connects && elt != he)
     {
       next = clib_llist_next (fwrk->event_elts, evt_list, elt);
@@ -235,45 +258,10 @@ session_mq_handle_connects_rpc (void *arg)
 
   /* Decrement with worker barrier */
   fwrk->n_pending_connects -= n_connects;
-
-  vlib_worker_thread_barrier_release (vm);
-
-update_state:
-
-  /* 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)
+  if (fwrk->n_pending_connects > 0)
     {
-      if (n_connects)
-	{
-	  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);
-	    }
-	}
-      else
-	wrk->no_connect_loops = 0;
-    }
-
-  if (wrk->state == SESSION_WRK_POLLING)
-    {
-      elt = session_evt_alloc_ctrl (wrk);
-      elt->evt.event_type = SESSION_CTRL_EVT_RPC;
-      elt->evt.rpc_args.fp = session_mq_handle_connects_rpc;
+      session_send_rpc_evt_to_thread_force (fwrk->vm->thread_index,
+					    session_mq_handle_connects_rpc, 0);
     }
 }
 
@@ -283,20 +271,28 @@ session_mq_connect_handler (session_worker_t *wrk, session_evt_elt_t *elt)
   u32 thread_index = wrk - session_main.wrk;
   session_evt_elt_t *he;
 
-  /* No workers, so just deal with the connect now */
-  if (PREDICT_FALSE (!thread_index))
+  if (PREDICT_FALSE (thread_index > transport_cl_thread ()))
     {
-      session_mq_connect_one (session_evt_ctrl_data (wrk, elt));
+      clib_warning ("Connect on wrong thread. Dropping");
       return;
     }
 
-  if (PREDICT_FALSE (thread_index != 1))
+  /* If on worker, check if main has any pending messages. Avoids reordering
+   * with other control messages that need to be handled by main
+   */
+  if (thread_index)
     {
-      clib_warning ("Connect on wrong thread. Dropping");
-      return;
+      he = clib_llist_elt (wrk->event_elts, wrk->evts_pending_main);
+
+      /* Events pending on main, postpone to avoid reordering */
+      if (!clib_llist_is_empty (wrk->event_elts, evt_list, he))
+	{
+	  clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
+	  return;
+	}
     }
 
-  /* Add to pending list to be handled by main thread */
+  /* Add to pending list to be handled by first worker */
   he = clib_llist_elt (wrk->event_elts, wrk->pending_connects);
   clib_llist_add_tail (wrk->event_elts, evt_list, elt, he);
 
@@ -304,23 +300,23 @@ session_mq_connect_handler (session_worker_t *wrk, session_evt_elt_t *elt)
   wrk->n_pending_connects += 1;
   if (wrk->n_pending_connects == 1)
     {
-      vlib_node_set_interrupt_pending (vlib_get_main_by_index (0),
-				       session_queue_node.index);
-      session_send_rpc_evt_to_thread (0, session_mq_handle_connects_rpc, 0);
+      session_send_rpc_evt_to_thread_force (thread_index,
+					    session_mq_handle_connects_rpc, 0);
     }
 }
 
 static void
-session_mq_connect_uri_handler (void *data)
+session_mq_connect_uri_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 {
-  session_connect_uri_msg_t *mp = (session_connect_uri_msg_t *) data;
   vnet_connect_args_t _a, *a = &_a;
+  session_connect_uri_msg_t *mp;
   app_worker_t *app_wrk;
   application_t *app;
   int rv;
 
-  app_check_thread_and_barrier (session_mq_connect_uri_handler, mp);
+  app_check_thread_and_barrier (wrk, elt);
 
+  mp = session_evt_ctrl_data (wrk, elt);
   app = application_lookup (mp->client_index);
   if (!app)
     return;
@@ -331,9 +327,9 @@ session_mq_connect_uri_handler (void *data)
   a->app_index = app->app_index;
   if ((rv = vnet_connect_uri (a)))
     {
-      clib_warning ("connect_uri returned: %d", rv);
+      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);
     }
 }
 
@@ -370,14 +366,15 @@ session_mq_disconnect_handler (void *data)
 }
 
 static void
-app_mq_detach_handler (void *data)
+app_mq_detach_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 {
-  session_app_detach_msg_t *mp = (session_app_detach_msg_t *) data;
   vnet_app_detach_args_t _a, *a = &_a;
+  session_app_detach_msg_t *mp;
   application_t *app;
 
-  app_check_thread_and_barrier (app_mq_detach_handler, mp);
+  app_check_thread_and_barrier (wrk, elt);
 
+  mp = session_evt_ctrl_data (wrk, elt);
   app = application_lookup (mp->client_index);
   if (!app)
     return;
@@ -388,18 +385,19 @@ app_mq_detach_handler (void *data)
 }
 
 static void
-session_mq_unlisten_rpc (session_unlisten_msg_t *mp)
+session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
 {
-  vlib_main_t *vm = vlib_get_main ();
   vnet_unlisten_args_t _a, *a = &_a;
+  session_unlisten_msg_t *mp;
   app_worker_t *app_wrk;
   session_handle_t sh;
   application_t *app;
-  u32 context;
   int rv;
 
+  app_check_thread_and_barrier (wrk, elt);
+
+  mp = session_evt_ctrl_data (wrk, elt);
   sh = mp->handle;
-  context = mp->context;
 
   app = application_lookup (mp->client_index);
   if (!app)
@@ -410,65 +408,34 @@ session_mq_unlisten_rpc (session_unlisten_msg_t *mp)
   a->handle = sh;
   a->wrk_map_index = mp->wrk_index;
 
-  vlib_worker_thread_barrier_sync (vm);
-
   if ((rv = vnet_unlisten (a)))
-    clib_warning ("unlisten returned: %d", rv);
-
-  vlib_worker_thread_barrier_release (vm);
+    session_worker_stat_error_inc (wrk, rv, 1);
 
   app_wrk = application_get_worker (app, a->wrk_map_index);
   if (!app_wrk)
     return;
 
-  mq_send_unlisten_reply (app_wrk, sh, context, rv);
-  clib_mem_free (mp);
+  app_worker_unlisten_reply (app_wrk, sh, mp->context, rv);
 }
 
 static void
-session_mq_unlisten_handler (session_worker_t *wrk, session_evt_elt_t *elt)
+session_mq_accepted_reply_handler (session_worker_t *wrk,
+				   session_evt_elt_t *elt)
 {
-  u32 thread_index = wrk - session_main.wrk;
-  session_unlisten_msg_t *mp, *arg;
-
-  mp = session_evt_ctrl_data (wrk, elt);
-  arg = clib_mem_alloc (sizeof (session_unlisten_msg_t));
-  clib_memcpy_fast (arg, mp, sizeof (*arg));
-
-  if (PREDICT_FALSE (!thread_index))
-    {
-      session_mq_unlisten_rpc (arg);
-      return;
-    }
-
-  session_send_rpc_evt_to_thread_force (0, session_mq_unlisten_rpc, arg);
-}
-
-static void
-session_mq_accepted_reply_handler (void *data)
-{
-  session_accepted_reply_msg_t *mp = (session_accepted_reply_msg_t *) data;
   vnet_disconnect_args_t _a = { 0 }, *a = &_a;
+  session_accepted_reply_msg_t *mp;
   session_state_t old_state;
   app_worker_t *app_wrk;
   session_t *s;
 
-  /* Server isn't interested, kill the session */
-  if (mp->retval)
-    {
-      a->app_index = mp->context;
-      a->handle = mp->handle;
-      vnet_disconnect_session (a);
-      return;
-    }
+  mp = session_evt_ctrl_data (wrk, elt);
 
   /* Mail this back from the main thread. We're not polling in main
    * thread so we're using other workers for notifications. */
-  if (vlib_num_workers () && vlib_get_thread_index () != 0
-      && session_thread_from_handle (mp->handle) == 0)
+  if (session_thread_from_handle (mp->handle) == 0 && vlib_num_workers () &&
+      vlib_get_thread_index () != 0)
     {
-      vlib_rpc_call_main_thread (session_mq_accepted_reply_handler,
-				 (u8 *) mp, sizeof (*mp));
+      session_wrk_send_evt_to_main (wrk, elt);
       return;
     }
 
@@ -483,27 +450,36 @@ session_mq_accepted_reply_handler (void *data)
       return;
     }
 
-  if (!session_has_transport (s))
+  /* Server isn't interested, disconnect the session */
+  if (mp->retval)
     {
-      s->session_state = SESSION_STATE_READY;
-      if (ct_session_connect_notify (s, SESSION_E_NONE))
-	return;
+      a->app_index = mp->context;
+      a->handle = mp->handle;
+      vnet_disconnect_session (a);
+      s->app_wrk_index = SESSION_INVALID_INDEX;
+      return;
     }
-  else
+
+  /* Special handling for cut-through sessions */
+  if (!session_has_transport (s))
     {
-      old_state = s->session_state;
-      s->session_state = SESSION_STATE_READY;
+      session_set_state (s, SESSION_STATE_READY);
+      ct_session_connect_notify (s, SESSION_E_NONE);
+      return;
+    }
 
-      if (!svm_fifo_is_empty_prod (s->rx_fifo))
-	app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
+  old_state = s->session_state;
+  session_set_state (s, SESSION_STATE_READY);
 
-      /* Closed while waiting for app to reply. Resend disconnect */
-      if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
-	{
-	  app_worker_close_notify (app_wrk, s);
-	  s->session_state = old_state;
-	  return;
-	}
+  if (!svm_fifo_is_empty_prod (s->rx_fifo))
+    app_worker_rx_notify (app_wrk, s);
+
+  /* Closed while waiting for app to reply. Resend disconnect */
+  if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
+    {
+      app_worker_close_notify (app_wrk, s);
+      session_set_state (s, old_state);
+      return;
     }
 }
 
@@ -515,15 +491,13 @@ session_mq_reset_reply_handler (void *data)
   app_worker_t *app_wrk;
   session_t *s;
   application_t *app;
-  u32 index, thread_index;
 
   mp = (session_reset_reply_msg_t *) data;
   app = application_lookup (mp->context);
   if (!app)
     return;
 
-  session_parse_handle (mp->handle, &index, &thread_index);
-  s = session_get_if_valid (index, thread_index);
+  s = session_get_from_handle_if_valid (mp->handle);
 
   /* No session or not the right session */
   if (!s || s->session_state < SESSION_STATE_TRANSPORT_CLOSING)
@@ -633,6 +607,7 @@ session_mq_worker_update_handler (void *data)
   session_event_t *evt;
   session_t *s;
   application_t *app;
+  int rv;
 
   app = application_lookup (mp->client_index);
   if (!app)
@@ -669,7 +644,9 @@ session_mq_worker_update_handler (void *data)
       return;
     }
 
-  app_worker_own_session (app_wrk, s);
+  rv = app_worker_own_session (app_wrk, s);
+  if (rv)
+    session_stat_error_inc (rv, 1);
 
   /*
    * Send reply
@@ -696,7 +673,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);
@@ -774,6 +751,67 @@ session_mq_transport_attr_handler (void *data)
   svm_msg_q_add_and_unlock (app_wrk->event_queue, msg);
 }
 
+void
+session_wrk_handle_evts_main_rpc (void *args)
+{
+  vlib_main_t *vm = vlib_get_main ();
+  clib_llist_index_t ei, next_ei;
+  session_evt_elt_t *he, *elt;
+  session_worker_t *fwrk;
+  u32 thread_index;
+
+  vlib_worker_thread_barrier_sync (vm);
+
+  thread_index = pointer_to_uword (args);
+  fwrk = session_main_get_worker (thread_index);
+
+  he = clib_llist_elt (fwrk->event_elts, fwrk->evts_pending_main);
+  ei = clib_llist_next_index (he, evt_list);
+
+  while (ei != fwrk->evts_pending_main)
+    {
+      elt = clib_llist_elt (fwrk->event_elts, ei);
+      next_ei = clib_llist_next_index (elt, evt_list);
+      clib_llist_remove (fwrk->event_elts, evt_list, elt);
+      switch (elt->evt.event_type)
+	{
+	case SESSION_CTRL_EVT_LISTEN:
+	  session_mq_listen_handler (fwrk, elt);
+	  break;
+	case SESSION_CTRL_EVT_UNLISTEN:
+	  session_mq_unlisten_handler (fwrk, elt);
+	  break;
+	case SESSION_CTRL_EVT_APP_DETACH:
+	  app_mq_detach_handler (fwrk, elt);
+	  break;
+	case SESSION_CTRL_EVT_CONNECT_URI:
+	  session_mq_connect_uri_handler (fwrk, elt);
+	  break;
+	case SESSION_CTRL_EVT_ACCEPTED_REPLY:
+	  session_mq_accepted_reply_handler (fwrk, elt);
+	  break;
+	case SESSION_CTRL_EVT_CONNECT:
+	  session_mq_connect_handler (fwrk, elt);
+	  break;
+	default:
+	  clib_warning ("unhandled %u", elt->evt.event_type);
+	  ALWAYS_ASSERT (0);
+	  break;
+	}
+
+      /* Regrab element in case pool moved */
+      elt = clib_llist_elt (fwrk->event_elts, ei);
+      if (!clib_llist_elt_is_linked (elt, evt_list))
+	{
+	  session_evt_ctrl_data_free (fwrk, elt);
+	  clib_llist_put (fwrk->event_elts, elt);
+	}
+      ei = next_ei;
+    }
+
+  vlib_worker_thread_barrier_release (vm);
+}
+
 vlib_node_registration_t session_queue_node;
 
 typedef struct
@@ -795,21 +833,21 @@ format_session_queue_trace (u8 * s, va_list * args)
   return s;
 }
 
-#define foreach_session_queue_error		\
-_(TX, "Packets transmitted")                  	\
-_(TIMER, "Timer events")			\
-_(NO_BUFFER, "Out of buffers")
+#define foreach_session_queue_error                                           \
+  _ (TX, tx, INFO, "Packets transmitted")                                     \
+  _ (TIMER, timer, INFO, "Timer events")                                      \
+  _ (NO_BUFFER, no_buffer, ERROR, "Out of buffers")
 
 typedef enum
 {
-#define _(sym,str) SESSION_QUEUE_ERROR_##sym,
+#define _(f, n, s, d) SESSION_QUEUE_ERROR_##f,
   foreach_session_queue_error
 #undef _
     SESSION_QUEUE_N_ERROR,
 } session_queue_error_t;
 
-static char *session_queue_error_strings[] = {
-#define _(sym,string) string,
+static vlib_error_desc_t session_error_counters[] = {
+#define _(f, n, s, d) { #n, d, VL_COUNTER_SEVERITY_##s },
   foreach_session_queue_error
 #undef _
 };
@@ -822,36 +860,134 @@ enum
 };
 
 static void
-session_tx_trace_frame (vlib_main_t * vm, vlib_node_runtime_t * node,
-			u32 next_index, u32 * to_next, u16 n_segs,
-			session_t * s, u32 n_trace)
+session_tx_trace_frame (vlib_main_t *vm, vlib_node_runtime_t *node,
+			u32 next_index, vlib_buffer_t **bufs, u16 n_segs,
+			session_t *s, u32 n_trace)
 {
+  vlib_buffer_t **b = bufs;
+
   while (n_trace && n_segs)
     {
-      vlib_buffer_t *b = vlib_get_buffer (vm, to_next[0]);
-      if (PREDICT_TRUE
-	  (vlib_trace_buffer
-	   (vm, node, next_index, b, 1 /* follow_chain */ )))
+      if (PREDICT_TRUE (vlib_trace_buffer (vm, node, next_index, b[0],
+					   1 /* follow_chain */)))
 	{
 	  session_queue_trace_t *t =
-	    vlib_add_trace (vm, node, b, sizeof (*t));
+	    vlib_add_trace (vm, node, b[0], sizeof (*t));
 	  t->session_index = s->session_index;
 	  t->server_thread_index = s->thread_index;
 	  n_trace--;
 	}
-      to_next++;
+      b++;
       n_segs--;
     }
   vlib_set_trace_count (vm, node, n_trace);
 }
 
+always_inline int
+session_tx_fill_dma_transfers (session_worker_t *wrk,
+			       session_tx_context_t *ctx, vlib_buffer_t *b)
+{
+  vlib_main_t *vm = wrk->vm;
+  u32 len_to_deq;
+  u8 *data0 = NULL;
+  int n_bytes_read, len_write;
+  svm_fifo_seg_t data_fs[2];
+
+  u32 n_segs = 2;
+  u16 n_transfers = 0;
+  /*
+   * Start with the first buffer in chain
+   */
+  b->error = 0;
+  b->flags = VNET_BUFFER_F_LOCALLY_ORIGINATED;
+  b->current_data = 0;
+  data0 = vlib_buffer_make_headroom (b, TRANSPORT_MAX_HDRS_LEN);
+  len_to_deq = clib_min (ctx->left_to_snd, ctx->deq_per_first_buf);
+
+  n_bytes_read = svm_fifo_segments (ctx->s->tx_fifo, ctx->sp.tx_offset,
+				    data_fs, &n_segs, len_to_deq);
+
+  len_write = n_bytes_read;
+  ASSERT (n_bytes_read == len_to_deq);
+
+  while (n_bytes_read)
+    {
+      wrk->batch_num++;
+      vlib_dma_batch_add (vm, wrk->batch, data0, data_fs[n_transfers].data,
+			  data_fs[n_transfers].len);
+      data0 += data_fs[n_transfers].len;
+      n_bytes_read -= data_fs[n_transfers].len;
+      n_transfers++;
+    }
+  return len_write;
+}
+
+always_inline int
+session_tx_fill_dma_transfers_tail (session_worker_t *wrk,
+				    session_tx_context_t *ctx,
+				    vlib_buffer_t *b, u32 len_to_deq, u8 *data)
+{
+  vlib_main_t *vm = wrk->vm;
+  int n_bytes_read, len_write;
+  svm_fifo_seg_t data_fs[2];
+  u32 n_segs = 2;
+  u16 n_transfers = 0;
+
+  n_bytes_read = svm_fifo_segments (ctx->s->tx_fifo, ctx->sp.tx_offset,
+				    data_fs, &n_segs, len_to_deq);
+
+  len_write = n_bytes_read;
+
+  ASSERT (n_bytes_read == len_to_deq);
+
+  while (n_bytes_read)
+    {
+      wrk->batch_num++;
+      vlib_dma_batch_add (vm, wrk->batch, data, data_fs[n_transfers].data,
+			  data_fs[n_transfers].len);
+      data += data_fs[n_transfers].len;
+      n_bytes_read -= data_fs[n_transfers].len;
+      n_transfers++;
+    }
+
+  return len_write;
+}
+
+always_inline int
+session_tx_copy_data (session_worker_t *wrk, session_tx_context_t *ctx,
+		      vlib_buffer_t *b, u32 len_to_deq, u8 *data0)
+{
+  int n_bytes_read;
+  if (PREDICT_TRUE (!wrk->dma_enabled))
+    n_bytes_read =
+      svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset, len_to_deq, data0);
+  else
+    n_bytes_read = session_tx_fill_dma_transfers (wrk, ctx, b);
+  return n_bytes_read;
+}
+
+always_inline int
+session_tx_copy_data_tail (session_worker_t *wrk, session_tx_context_t *ctx,
+			   vlib_buffer_t *b, u32 len_to_deq, u8 *data)
+{
+  int n_bytes_read;
+  if (PREDICT_TRUE (!wrk->dma_enabled))
+    n_bytes_read =
+      svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset, len_to_deq, data);
+  else
+    n_bytes_read =
+      session_tx_fill_dma_transfers_tail (wrk, ctx, b, len_to_deq, data);
+  return n_bytes_read;
+}
+
 always_inline void
-session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
-			    vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
+session_tx_fifo_chain_tail (session_worker_t *wrk, session_tx_context_t *ctx,
+			    vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
 {
+  vlib_main_t *vm = wrk->vm;
   vlib_buffer_t *chain_b, *prev_b;
   u32 chain_bi0, to_deq, left_from_seg;
-  u16 len_to_deq, n_bytes_read;
+  int len_to_deq, n_bytes_read;
   u8 *data, j;
 
   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
@@ -873,8 +1009,8 @@ session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
       data = vlib_buffer_get_current (chain_b);
       if (peek_data)
 	{
-	  n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo,
-					ctx->sp.tx_offset, len_to_deq, data);
+	  n_bytes_read =
+	    session_tx_copy_data_tail (wrk, ctx, b, len_to_deq, data);
 	  ctx->sp.tx_offset += n_bytes_read;
 	}
       else
@@ -931,13 +1067,12 @@ session_tx_fifo_chain_tail (vlib_main_t * vm, session_tx_context_t * ctx,
 }
 
 always_inline void
-session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
-			vlib_buffer_t * b, u16 * n_bufs, u8 peek_data)
+session_tx_fill_buffer (session_worker_t *wrk, session_tx_context_t *ctx,
+			vlib_buffer_t *b, u16 *n_bufs, u8 peek_data)
 {
   u32 len_to_deq;
   u8 *data0;
   int n_bytes_read;
-
   /*
    * Start with the first buffer in chain
    */
@@ -950,8 +1085,7 @@ session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
 
   if (peek_data)
     {
-      n_bytes_read = svm_fifo_peek (ctx->s->tx_fifo, ctx->sp.tx_offset,
-				    len_to_deq, data0);
+      n_bytes_read = session_tx_copy_data (wrk, ctx, b, len_to_deq, data0);
       ASSERT (n_bytes_read > 0);
       /* Keep track of progress locally, transport is also supposed to
        * increment it independently when pushing the header */
@@ -975,8 +1109,8 @@ session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
 
 	  if (transport_connection_is_cless (ctx->tc))
 	    {
-	      ip_copy (&ctx->tc->rmt_ip, &hdr->rmt_ip, ctx->tc->is_ip4);
-	      ctx->tc->rmt_port = hdr->rmt_port;
+	      clib_memcpy_fast (data0 - sizeof (session_dgram_hdr_t), hdr,
+				sizeof (*hdr));
 	    }
 	  hdr->data_offset += n_bytes_read;
 	  if (hdr->data_offset == hdr->data_length)
@@ -998,6 +1132,7 @@ session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
 	  ASSERT (n_bytes_read > 0);
 	}
     }
+
   b->current_length = n_bytes_read;
   ctx->left_to_snd -= n_bytes_read;
 
@@ -1005,7 +1140,7 @@ session_tx_fill_buffer (vlib_main_t * vm, session_tx_context_t * ctx,
    * Fill in the remaining buffers in the chain, if any
    */
   if (PREDICT_FALSE (ctx->n_bufs_per_seg > 1 && ctx->left_to_snd))
-    session_tx_fifo_chain_tail (vm, ctx, b, n_bufs, peek_data);
+    session_tx_fifo_chain_tail (wrk, ctx, b, n_bufs, peek_data);
 }
 
 always_inline u8
@@ -1037,6 +1172,11 @@ session_tx_not_ready (session_t * s, u8 peek_data)
 	  return 2;
 	}
     }
+  else
+    {
+      if (s->session_state == SESSION_STATE_TRANSPORT_DELETED)
+	return 2;
+    }
   return 0;
 }
 
@@ -1093,9 +1233,28 @@ session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
 
 	  svm_fifo_peek (ctx->s->tx_fifo, 0, sizeof (ctx->hdr),
 			 (u8 *) & ctx->hdr);
+	  /* Zero length dgrams not supported */
+	  if (PREDICT_FALSE (ctx->hdr.data_length == 0))
+	    {
+	      svm_fifo_dequeue_drop (ctx->s->tx_fifo, sizeof (ctx->hdr));
+	      ctx->max_len_to_snd = 0;
+	      return;
+	    }
+	  /* We cannot be sure apps have not enqueued incomplete dgrams */
+	  if (PREDICT_FALSE (ctx->max_dequeue <
+			     ctx->hdr.data_length + sizeof (ctx->hdr)))
+	    {
+	      ctx->max_len_to_snd = 0;
+	      return;
+	    }
 	  ASSERT (ctx->hdr.data_length > ctx->hdr.data_offset);
 	  len = ctx->hdr.data_length - ctx->hdr.data_offset;
 
+	  if (ctx->hdr.gso_size)
+	    {
+	      ctx->sp.snd_mss = clib_min (ctx->sp.snd_mss, ctx->hdr.gso_size);
+	    }
+
 	  /* Process multiple dgrams if smaller than min (buf_space, mss).
 	   * This avoids handling multiple dgrams if they require buffer
 	   * chains */
@@ -1115,11 +1274,13 @@ session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
 		{
 		  svm_fifo_peek (ctx->s->tx_fifo, offset, sizeof (ctx->hdr),
 				 (u8 *) & hdr);
-		  ASSERT (hdr.data_length > hdr.data_offset);
 		  dgram_len = hdr.data_length - hdr.data_offset;
-		  if (len + dgram_len > ctx->max_dequeue
-		      || first_dgram_len != dgram_len)
+		  if (offset + sizeof (hdr) + hdr.data_length >
+			ctx->max_dequeue ||
+		      first_dgram_len != dgram_len)
 		    break;
+		  /* Assert here to allow test above with zero length dgrams */
+		  ASSERT (hdr.data_length > hdr.data_offset);
 		  len += dgram_len;
 		  offset += sizeof (hdr) + hdr.data_length;
 		}
@@ -1188,8 +1349,30 @@ session_tx_maybe_reschedule (session_worker_t * wrk,
 
   svm_fifo_unset_event (s->tx_fifo);
   if (svm_fifo_max_dequeue_cons (s->tx_fifo) > ctx->sp.tx_offset)
-    if (svm_fifo_set_event (s->tx_fifo))
-      session_evt_add_head_old (wrk, elt);
+    {
+      if (svm_fifo_set_event (s->tx_fifo))
+	session_evt_add_head_old (wrk, elt);
+    }
+  else
+    {
+      transport_connection_deschedule (ctx->tc);
+    }
+}
+
+always_inline void
+session_tx_add_pending_buffer (session_worker_t *wrk, u32 bi, u32 next_index)
+{
+  if (PREDICT_TRUE (!wrk->dma_enabled))
+    {
+      vec_add1 (wrk->pending_tx_buffers, bi);
+      vec_add1 (wrk->pending_tx_nexts, next_index);
+    }
+  else
+    {
+      session_dma_transfer *dma_transfer = &wrk->dma_trans[wrk->trans_tail];
+      vec_add1 (dma_transfer->pending_tx_buffers, bi);
+      vec_add1 (dma_transfer->pending_tx_nexts, next_index);
+    }
 }
 
 always_inline int
@@ -1235,9 +1418,12 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
       ctx->sp.max_burst_size = max_burst;
       n_custom_tx = ctx->transport_vft->custom_tx (ctx->tc, &ctx->sp);
       *n_tx_packets += n_custom_tx;
-      if (PREDICT_FALSE
-	  (ctx->s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
-	return SESSION_TX_OK;
+      if (PREDICT_FALSE (ctx->s->session_state >=
+			 SESSION_STATE_TRANSPORT_CLOSED))
+	{
+	  svm_fifo_unset_event (ctx->s->tx_fifo);
+	  return SESSION_TX_OK;
+	}
       max_burst -= n_custom_tx;
       if (!max_burst || (ctx->s->flags & SESSION_F_CUSTOM_TX))
 	{
@@ -1246,6 +1432,11 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
 	}
     }
 
+  /* Connection previously descheduled because it had no data to send.
+   * Clear descheduled flag and reset pacer if in use */
+  if (transport_connection_is_descheduled (ctx->tc))
+    transport_connection_clear_descheduled (ctx->tc);
+
   transport_connection_snd_params (ctx->tc, &ctx->sp);
 
   if (!ctx->sp.snd_space)
@@ -1308,6 +1499,8 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
   ctx->left_to_snd = ctx->max_len_to_snd;
   n_left = ctx->n_segs_per_evt;
 
+  vec_validate (ctx->transport_pending_bufs, n_left);
+
   while (n_left >= 4)
     {
       vlib_buffer_t *b0, *b1;
@@ -1326,18 +1519,15 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
       b0 = vlib_get_buffer (vm, bi0);
       b1 = vlib_get_buffer (vm, bi1);
 
-      session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
-      session_tx_fill_buffer (vm, ctx, b1, &n_bufs, peek_data);
-
-      ctx->transport_vft->push_header (ctx->tc, b0);
-      ctx->transport_vft->push_header (ctx->tc, b1);
+      session_tx_fill_buffer (wrk, ctx, b0, &n_bufs, peek_data);
+      session_tx_fill_buffer (wrk, ctx, b1, &n_bufs, peek_data);
 
+      ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left] = b0;
+      ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left + 1] = b1;
       n_left -= 2;
 
-      vec_add1 (wrk->pending_tx_buffers, bi0);
-      vec_add1 (wrk->pending_tx_buffers, bi1);
-      vec_add1 (wrk->pending_tx_nexts, next_index);
-      vec_add1 (wrk->pending_tx_nexts, next_index);
+      session_tx_add_pending_buffer (wrk, bi0, next_index);
+      session_tx_add_pending_buffer (wrk, bi1, next_index);
     }
   while (n_left)
     {
@@ -1353,20 +1543,20 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
 
       bi0 = ctx->tx_buffers[--n_bufs];
       b0 = vlib_get_buffer (vm, bi0);
-      session_tx_fill_buffer (vm, ctx, b0, &n_bufs, peek_data);
-
-      /* Ask transport to push header after current_length and
-       * total_length_not_including_first_buffer are updated */
-      ctx->transport_vft->push_header (ctx->tc, b0);
+      session_tx_fill_buffer (wrk, ctx, b0, &n_bufs, peek_data);
 
+      ctx->transport_pending_bufs[ctx->n_segs_per_evt - n_left] = b0;
       n_left -= 1;
 
-      vec_add1 (wrk->pending_tx_buffers, bi0);
-      vec_add1 (wrk->pending_tx_nexts, next_index);
+      session_tx_add_pending_buffer (wrk, bi0, next_index);
     }
 
+  /* Ask transport to push headers */
+  ctx->transport_vft->push_header (ctx->tc, ctx->transport_pending_bufs,
+				   ctx->n_segs_per_evt);
+
   if (PREDICT_FALSE ((n_trace = vlib_get_trace_count (vm, node)) > 0))
-    session_tx_trace_frame (vm, node, next_index, wrk->pending_tx_buffers,
+    session_tx_trace_frame (vm, node, next_index, ctx->transport_pending_bufs,
 			    ctx->n_segs_per_evt, ctx->s, n_trace);
 
   if (PREDICT_FALSE (n_bufs))
@@ -1375,7 +1565,7 @@ session_tx_fifo_read_and_snd_i (session_worker_t * wrk,
   *n_tx_packets += ctx->n_segs_per_evt;
 
   SESSION_EVT (SESSION_EVT_DEQ, ctx->s, ctx->max_len_to_snd, ctx->max_dequeue,
-	       ctx->s->tx_fifo->has_event, wrk->last_vlib_time);
+	       ctx->s->tx_fifo->shr->has_event, wrk->last_vlib_time);
 
   ASSERT (ctx->left_to_snd == 0);
 
@@ -1420,20 +1610,30 @@ session_tx_fifo_dequeue_internal (session_worker_t * wrk,
 {
   transport_send_params_t *sp = &wrk->ctx.sp;
   session_t *s = wrk->ctx.s;
+  clib_llist_index_t ei;
   u32 n_packets;
 
-  if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
+  if (PREDICT_FALSE ((s->session_state >= SESSION_STATE_TRANSPORT_CLOSED) ||
+		     (s->session_state == SESSION_STATE_CONNECTING &&
+		      (s->flags & SESSION_F_HALF_OPEN))))
     return 0;
 
   /* Clear custom-tx flag used to request reschedule for tx */
   s->flags &= ~SESSION_F_CUSTOM_TX;
 
+  sp->flags = 0;
+  sp->bytes_dequeued = 0;
   sp->max_burst_size = clib_min (SESSION_NODE_FRAME_SIZE - *n_tx_packets,
 				 TRANSPORT_PACER_MAX_BURST_PKTS);
 
+  /* Grab elt index since app transports can enqueue events on tx */
+  ei = clib_llist_entry_index (wrk->event_elts, elt);
+
   n_packets = transport_custom_tx (session_get_transport_proto (s), s, sp);
   *n_tx_packets += n_packets;
 
+  elt = clib_llist_elt (wrk->event_elts, ei);
+
   if (s->flags & SESSION_F_CUSTOM_TX)
     {
       session_evt_add_old (wrk, elt);
@@ -1446,8 +1646,8 @@ session_tx_fifo_dequeue_internal (session_worker_t * wrk,
 	  session_evt_add_head_old (wrk, elt);
     }
 
-  if (sp->max_burst_size &&
-      svm_fifo_needs_deq_ntf (s->tx_fifo, sp->max_burst_size))
+  if (sp->bytes_dequeued &&
+      svm_fifo_needs_deq_ntf (s->tx_fifo, sp->bytes_dequeued))
     session_dequeue_notify (s);
 
   return n_packets;
@@ -1499,10 +1699,10 @@ session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
       session_transport_reset (s);
       break;
     case SESSION_CTRL_EVT_LISTEN:
-      session_mq_listen_handler (session_evt_ctrl_data (wrk, elt));
+      session_mq_listen_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_LISTEN_URI:
-      session_mq_listen_uri_handler (session_evt_ctrl_data (wrk, elt));
+      session_mq_listen_uri_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_UNLISTEN:
       session_mq_unlisten_handler (wrk, elt);
@@ -1511,7 +1711,7 @@ session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
       session_mq_connect_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_CONNECT_URI:
-      session_mq_connect_uri_handler (session_evt_ctrl_data (wrk, elt));
+      session_mq_connect_uri_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_SHUTDOWN:
       session_mq_shutdown_handler (session_evt_ctrl_data (wrk, elt));
@@ -1523,7 +1723,7 @@ session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
       session_mq_disconnected_handler (session_evt_ctrl_data (wrk, elt));
       break;
     case SESSION_CTRL_EVT_ACCEPTED_REPLY:
-      session_mq_accepted_reply_handler (session_evt_ctrl_data (wrk, elt));
+      session_mq_accepted_reply_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_DISCONNECTED_REPLY:
       session_mq_disconnected_reply_handler (session_evt_ctrl_data (wrk,
@@ -1536,7 +1736,7 @@ session_event_dispatch_ctrl (session_worker_t * wrk, session_evt_elt_t * elt)
       session_mq_worker_update_handler (session_evt_ctrl_data (wrk, elt));
       break;
     case SESSION_CTRL_EVT_APP_DETACH:
-      app_mq_detach_handler (session_evt_ctrl_data (wrk, elt));
+      app_mq_detach_handler (wrk, elt);
       break;
     case SESSION_CTRL_EVT_APP_WRK_RPC:
       session_mq_app_wrk_rpc_handler (session_evt_ctrl_data (wrk, elt));
@@ -1588,7 +1788,7 @@ session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
       break;
     case SESSION_IO_EVT_RX:
       s = session_event_get_session (wrk, e);
-      if (!s)
+      if (!s || s->session_state >= SESSION_STATE_TRANSPORT_CLOSED)
 	break;
       transport_app_rx_evt (session_get_transport_proto (s),
 			    s->connection_index, s->thread_index);
@@ -1599,19 +1799,21 @@ 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_BUILTIN_TX:
-      s = session_get_from_handle_if_valid (e->session_handle);
+    case SESSION_IO_EVT_TX_MAIN:
+      s = session_get_if_valid (e->session_index, 0 /* main thread */);
+      if (PREDICT_FALSE (!s))
+	break;
       wrk->ctx.s = s;
       if (PREDICT_TRUE (s != 0))
-	session_tx_fifo_dequeue_internal (wrk, node, elt, n_tx_packets);
+	(smm->session_tx_fns[s->session_type]) (wrk, node, elt, n_tx_packets);
       break;
     default:
       clib_warning ("unhandled event type %d", e->event_type);
     }
 
-  SESSION_EVT (SESSION_IO_EVT_COUNTS, e->event_type, 1, wrk);
+  SESSION_EVT (SESSION_EVT_IO_EVT_COUNTS, e->event_type, 1, wrk);
 
   /* Regrab elements in case pool moved */
   elt = clib_llist_elt (wrk->event_elts, ei);
@@ -1619,14 +1821,22 @@ session_event_dispatch_io (session_worker_t * wrk, vlib_node_runtime_t * node,
     clib_llist_put (wrk->event_elts, elt);
 }
 
-/* *INDENT-OFF* */
 static const u32 session_evt_msg_sizes[] = {
 #define _(symc, sym) 							\
   [SESSION_CTRL_EVT_ ## symc] = sizeof (session_ ## sym ##_msg_t),
   foreach_session_ctrl_evt
 #undef _
 };
-/* *INDENT-ON* */
+
+always_inline void
+session_update_time_subscribers (session_main_t *smm, clib_time_type_t now,
+				 u32 thread_index)
+{
+  session_update_time_fn *fn;
+
+  vec_foreach (fn, smm->update_time_fns)
+    (*fn) (now, thread_index);
+}
 
 always_inline void
 session_evt_add_to_list (session_worker_t * wrk, session_event_t * evt)
@@ -1693,7 +1903,7 @@ session_wrk_update_state (session_worker_t *wrk)
 
   if (wrk->state == SESSION_WRK_POLLING)
     {
-      if (clib_llist_elts (wrk->event_elts) == 4 &&
+      if (clib_llist_elts (wrk->event_elts) == 5 &&
 	  vlib_last_vectors_per_main_loop (vm) < 1)
 	{
 	  session_wrk_set_state (wrk, SESSION_WRK_INTERRUPT);
@@ -1703,7 +1913,7 @@ session_wrk_update_state (session_worker_t *wrk)
     }
   else if (wrk->state == SESSION_WRK_INTERRUPT)
     {
-      if (clib_llist_elts (wrk->event_elts) > 4 ||
+      if (clib_llist_elts (wrk->event_elts) > 5 ||
 	  vlib_last_vectors_per_main_loop (vm) > 1)
 	{
 	  session_wrk_set_state (wrk, SESSION_WRK_POLLING);
@@ -1742,10 +1952,19 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   /*
    *  Update transport time
    */
-  transport_update_time (wrk->last_vlib_time, thread_index);
+  session_update_time_subscribers (smm, wrk->last_vlib_time, thread_index);
   n_tx_packets = vec_len (wrk->pending_tx_buffers);
   SESSION_EVT (SESSION_EVT_DSP_CNTRS, UPDATE_TIME, wrk);
 
+  if (PREDICT_FALSE (wrk->dma_enabled))
+    {
+      if (wrk->trans_head == ((wrk->trans_tail + 1) & (wrk->trans_size - 1)))
+	return 0;
+      wrk->batch = vlib_dma_batch_new (vm, wrk->config_index);
+      if (!wrk->batch)
+	return 0;
+    }
+
   /*
    *  Dequeue new internal mq events
    */
@@ -1815,6 +2034,20 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 	};
     }
 
+  if (PREDICT_FALSE (wrk->dma_enabled))
+    {
+      if (wrk->batch_num)
+	{
+	  vlib_dma_batch_set_cookie (vm, wrk->batch, wrk->trans_tail);
+	  wrk->batch_num = 0;
+	  wrk->trans_tail++;
+	  if (wrk->trans_tail == wrk->trans_size)
+	    wrk->trans_tail = 0;
+	}
+
+      vlib_dma_batch_submit (vm, wrk->batch);
+    }
+
   SESSION_EVT (SESSION_EVT_DSP_CNTRS, OLD_IO_EVTS, wrk);
 
   if (vec_len (wrk->pending_tx_buffers))
@@ -1831,19 +2064,16 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   return n_tx_packets;
 }
 
-/* *INDENT-OFF* */
-VLIB_REGISTER_NODE (session_queue_node) =
-{
+VLIB_REGISTER_NODE (session_queue_node) = {
   .function = session_queue_node_fn,
   .flags = VLIB_NODE_FLAG_TRACE_SUPPORTED,
   .name = "session-queue",
   .format_trace = format_session_queue_trace,
   .type = VLIB_NODE_TYPE_INPUT,
-  .n_errors = ARRAY_LEN (session_queue_error_strings),
-  .error_strings = session_queue_error_strings,
+  .n_errors = SESSION_QUEUE_N_ERROR,
+  .error_counters = session_error_counters,
   .state = VLIB_NODE_STATE_DISABLED,
 };
-/* *INDENT-ON* */
 
 static clib_error_t *
 session_wrk_tfd_read_ready (clib_file_t *cf)
@@ -1947,7 +2177,6 @@ session_queue_process (vlib_main_t * vm, vlib_node_runtime_t * rt,
   return 0;
 }
 
-/* *INDENT-OFF* */
 VLIB_REGISTER_NODE (session_queue_process_node) =
 {
   .function = session_queue_process,
@@ -1955,7 +2184,6 @@ VLIB_REGISTER_NODE (session_queue_process_node) =
   .name = "session-queue-process",
   .state = VLIB_NODE_STATE_DISABLED,
 };
-/* *INDENT-ON* */
 
 static_always_inline uword
 session_queue_pre_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
@@ -1968,7 +2196,6 @@ session_queue_pre_input_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
   return session_queue_node_fn (vm, node, frame);
 }
 
-/* *INDENT-OFF* */
 VLIB_REGISTER_NODE (session_queue_pre_input_node) =
 {
   .function = session_queue_pre_input_inline,
@@ -1976,7 +2203,6 @@ VLIB_REGISTER_NODE (session_queue_pre_input_node) =
   .name = "session-queue-main",
   .state = VLIB_NODE_STATE_DISABLED,
 };
-/* *INDENT-ON* */
 
 /*
  * fd.io coding-style-patch-verification: ON