vcl/session: tx notifications for cut-thru sessions

[vpp.git] / src / vnet / session / session_node.c

diff --git a/src/vnet/session/session_node.c b/src/vnet/session/session_node.c
index 7cbd0d9..605fd58 100644 (file)
--- a/src/vnet/session/session_node.c
+++ b/src/vnet/session/session_node.c
@@ -82,14 +82,14 @@ session_mq_accepted_reply_handler (void *data)
     {
       old_state = s->session_state;
       s->session_state = SESSION_STATE_READY;
-      if (!svm_fifo_is_empty (s->rx_fifo))
+
+      if (!svm_fifo_is_empty_prod (s->rx_fifo))
        app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
 
       /* Closed while waiting for app to reply. Resend disconnect */
       if (old_state >= SESSION_STATE_TRANSPORT_CLOSING)
        {
-         application_t *app = application_get (app_wrk->app_index);
-         app->cb_fns.session_disconnect_callback (s);
+         app_worker_close_notify (app_wrk, s);
          s->session_state = old_state;
          return;
        }
@@ -289,7 +289,7 @@ session_mq_worker_update_handler (void *data)
     app_worker_lock_and_send_event (app_wrk, s, SESSION_IO_EVT_RX);
 
   if (s->session_state >= SESSION_STATE_TRANSPORT_CLOSING)
-    app->cb_fns.session_disconnect_callback (s);
+    app_worker_close_notify (app_wrk, s);
 }
 
 vlib_node_registration_t session_queue_node;
@@ -558,7 +558,7 @@ session_tx_set_dequeue_params (vlib_main_t * vm, session_tx_context_t * ctx,
                               u32 max_segs, u8 peek_data)
 {
   u32 n_bytes_per_buf, n_bytes_per_seg;
-  ctx->max_dequeue = svm_fifo_max_dequeue (ctx->s->tx_fifo);
+  ctx->max_dequeue = svm_fifo_max_dequeue_cons (ctx->s->tx_fifo);
   if (peek_data)
     {
       /* Offset in rx fifo from where to peek data */
@@ -789,7 +789,7 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
        svm_fifo_overwrite_head (ctx->s->tx_fifo, (u8 *) & ctx->hdr,
                                 sizeof (session_dgram_pre_hdr_t));
       /* More data needs to be read */
-      else if (svm_fifo_max_dequeue (ctx->s->tx_fifo) > 0)
+      else if (svm_fifo_max_dequeue_cons (ctx->s->tx_fifo) > 0)
        if (svm_fifo_set_event (ctx->s->tx_fifo))
          vec_add1 (wrk->pending_event_vector, *e);
     }
@@ -819,13 +819,11 @@ session_tx_fifo_dequeue_internal (vlib_main_t * vm,
                                  session_event_t * e, int *n_tx_pkts)
 {
   session_t *s = wrk->ctx.s;
-  application_t *app;
 
   if (PREDICT_FALSE (s->session_state >= SESSION_STATE_TRANSPORT_CLOSED))
     return 0;
-  app = application_get (s->t_app_index);
   svm_fifo_unset_event (s->tx_fifo);
-  return app->cb_fns.builtin_app_tx_callback (s);
+  return transport_custom_tx (session_get_transport_proto (s), s);
 }
 
 always_inline session_t *
@@ -858,7 +856,6 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   f64 now = vlib_time_now (vm);
   int n_tx_packets = 0, i, rv;
   app_worker_t *app_wrk;
-  application_t *app;
   svm_msg_q_t *mq;
   void (*fp) (void *);
 
@@ -930,6 +927,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
              clib_warning ("session was freed!");
              continue;
            }
+         CLIB_PREFETCH (s->tx_fifo, 2 * CLIB_CACHE_LINE_BYTES, LOAD);
          wrk->ctx.s = s;
          /* Spray packets in per session type frames, since they go to
           * different nodes */
@@ -949,6 +947,13 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
              continue;
            }
          break;
+       case SESSION_IO_EVT_RX:
+         s = session_event_get_session (e, thread_index);
+         if (!s)
+           break;
+         transport_app_rx_evt (session_get_transport_proto (s),
+                               s->connection_index, s->thread_index);
+         break;
        case SESSION_CTRL_EVT_CLOSE:
          s = session_get_from_handle_if_valid (e->session_handle);
          if (PREDICT_FALSE (!s))
@@ -959,7 +964,8 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
           * and the tx queue is still not empty, try to wait for some
           * dispatch cycles */
          if (!e->postponed
-             || (e->postponed < 200 && svm_fifo_max_dequeue (s->tx_fifo)))
+             || (e->postponed < 200
+                 && svm_fifo_max_dequeue_cons (s->tx_fifo)))
            {
              e->postponed += 1;
              vec_add1 (wrk->pending_disconnects, *e);
@@ -974,8 +980,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
            continue;
          svm_fifo_unset_event (s->rx_fifo);
          app_wrk = app_worker_get (s->app_wrk_index);
-         app = application_get (app_wrk->app_index);
-         app->cb_fns.builtin_app_rx_callback (s);
+         app_worker_builtin_rx (app_wrk, s);
          break;
        case SESSION_IO_EVT_BUILTIN_TX:
          s = session_get_from_handle_if_valid (e->session_handle);