Overall tcp performance improvements (VPP-846)
[vpp.git] / src / vnet / session / node.c
index ce7c386..c0ab1bf 100644 (file)
@@ -47,7 +47,8 @@ vlib_node_registration_t session_queue_node;
 
 #define foreach_session_queue_error            \
 _(TX, "Packets transmitted")                   \
-_(TIMER, "Timer events")
+_(TIMER, "Timer events")                       \
+_(NO_BUFFER, "Out of buffers")
 
 typedef enum
 {
@@ -141,6 +142,7 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
   u8 *data0;
   int i, n_bytes_read;
   u32 n_bytes_per_buf, deq_per_buf;
+  u32 buffers_allocated, buffers_allocated_this_call;
 
   next_index = next0 = session_type_to_next[s0->session_type];
 
@@ -154,7 +156,7 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
   /* Can't make any progress */
   if (snd_space0 == 0 || snd_mss0 == 0)
     {
-      vec_add1 (smm->evts_partially_read[thread_index], *e0);
+      vec_add1 (smm->pending_event_vector[thread_index], *e0);
       return 0;
     }
 
@@ -167,12 +169,12 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
   /* Check how much we can pull. If buffering, subtract the offset */
   max_dequeue0 = svm_fifo_max_dequeue (s0->server_tx_fifo) - rx_offset;
 
-  /* Allow enqueuing of a new event */
-  svm_fifo_unset_event (s0->server_tx_fifo);
-
   /* Nothing to read return */
   if (max_dequeue0 == 0)
-    return 0;
+    {
+      svm_fifo_unset_event (s0->server_tx_fifo);
+      return 0;
+    }
 
   /* Ensure we're not writing more than transport window allows */
   if (max_dequeue0 < snd_space0)
@@ -187,8 +189,8 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
       max_len_to_snd0 = snd_space0;
     }
 
-  n_bytes_per_buf = vlib_buffer_free_list_buffer_size (vm,
-                                                      VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
+  n_bytes_per_buf = vlib_buffer_free_list_buffer_size
+    (vm, VLIB_BUFFER_DEFAULT_FREE_LIST_INDEX);
   n_bytes_per_seg = MAX_HDRS_LEN + snd_mss0;
   n_bufs_per_seg = ceil ((double) n_bytes_per_seg / n_bytes_per_buf);
   n_bufs_per_evt = (ceil ((double) max_len_to_snd0 / n_bytes_per_seg))
@@ -205,24 +207,33 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
       if (PREDICT_FALSE (n_bufs < VLIB_FRAME_SIZE))
        {
          vec_validate (smm->tx_buffers[thread_index],
-                       n_bufs + VLIB_FRAME_SIZE - 1);
-         n_bufs += vlib_buffer_alloc (vm,
-                                      &smm->tx_buffers[thread_index][n_bufs],
-                                      VLIB_FRAME_SIZE);
-
-         /* buffer shortage
-          * XXX 0.9 because when debugging we might not get a full frame */
-         if (PREDICT_FALSE (n_bufs < 0.9 * VLIB_FRAME_SIZE))
+                       n_bufs + 2 * VLIB_FRAME_SIZE - 1);
+
+         buffers_allocated = 0;
+         do
            {
-             if (svm_fifo_set_event (s0->server_tx_fifo))
-               {
-                 vec_add1 (smm->evts_partially_read[thread_index], *e0);
-               }
-             return -1;
+             buffers_allocated_this_call =
+               vlib_buffer_alloc
+               (vm,
+                &smm->tx_buffers[thread_index][n_bufs + buffers_allocated],
+                2 * VLIB_FRAME_SIZE - buffers_allocated);
+             buffers_allocated += buffers_allocated_this_call;
            }
+         while (buffers_allocated_this_call > 0
+                && ((buffers_allocated + n_bufs < VLIB_FRAME_SIZE)));
+
+         n_bufs += buffers_allocated;
 
          _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
+
+         if (PREDICT_FALSE (n_bufs < VLIB_FRAME_SIZE))
+           {
+             vec_add1 (smm->pending_event_vector[thread_index], *e0);
+             return -1;
+           }
        }
+      /* Allow enqueuing of a new event */
+      svm_fifo_unset_event (s0->server_tx_fifo);
 
       vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
       while (left_to_snd0 && n_left_to_next >= n_bufs_per_seg)
@@ -232,7 +243,9 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
           */
 
          /* Get free buffer */
+         ASSERT (n_bufs >= 1);
          bi0 = smm->tx_buffers[thread_index][--n_bufs];
+         ASSERT (bi0);
          _vec_len (smm->tx_buffers[thread_index]) = n_bufs;
 
          b0 = vlib_get_buffer (vm, bi0);
@@ -324,7 +337,7 @@ session_tx_fifo_read_and_snd_i (vlib_main_t * vm, vlib_node_runtime_t * node,
       /* If we don't already have new event */
       if (svm_fifo_set_event (s0->server_tx_fifo))
        {
-         vec_add1 (smm->evts_partially_read[thread_index], *e0);
+         vec_add1 (smm->pending_event_vector[thread_index], *e0);
        }
     }
   return 0;
@@ -338,7 +351,7 @@ dequeue_fail:
 
   if (svm_fifo_set_event (s0->server_tx_fifo))
     {
-      vec_add1 (smm->evts_partially_read[thread_index], *e0);
+      vec_add1 (smm->pending_event_vector[thread_index], *e0);
     }
   vlib_put_next_frame (vm, node, next_index, n_left_to_next + 1);
   _vec_len (smm->tx_buffers[thread_index]) += 1;
@@ -383,17 +396,75 @@ session_event_get_session (session_fifo_event_t * e0, u8 thread_index)
 
   s0 = stream_session_get_if_valid (session_index0, thread_index);
 
-  ASSERT (s0->thread_index == thread_index);
+  ASSERT (s0 == 0 || s0->thread_index == thread_index);
 
   return s0;
 }
 
+void
+dump_thread_0_event_queue (void)
+{
+  session_manager_main_t *smm = vnet_get_session_manager_main ();
+  vlib_main_t *vm = &vlib_global_main;
+  u32 my_thread_index = vm->thread_index;
+  session_fifo_event_t _e, *e = &_e;
+  stream_session_t *s0;
+  int i, index;
+  i8 *headp;
+
+  unix_shared_memory_queue_t *q;
+  q = smm->vpp_event_queues[my_thread_index];
+
+  index = q->head;
+
+  for (i = 0; i < q->cursize; i++)
+    {
+      headp = (i8 *) (&q->data[0] + q->elsize * index);
+      clib_memcpy (e, headp, q->elsize);
+
+      switch (e->event_type)
+       {
+       case FIFO_EVENT_APP_TX:
+         s0 = session_event_get_session (e, my_thread_index);
+         fformat (stdout, "[%04d] TX session %d\n", i, s0->session_index);
+         break;
+
+       case FIFO_EVENT_DISCONNECT:
+         s0 = stream_session_get_from_handle (e->session_handle);
+         fformat (stdout, "[%04d] disconnect session %d\n", i,
+                  s0->session_index);
+         break;
+
+       case FIFO_EVENT_BUILTIN_RX:
+         s0 = session_event_get_session (e, my_thread_index);
+         fformat (stdout, "[%04d] builtin_rx %d\n", i, s0->session_index);
+         break;
+
+       case FIFO_EVENT_RPC:
+         fformat (stdout, "[%04d] RPC call %llx with %llx\n",
+                  i, (u64) (e->rpc_args.fp), (u64) (e->rpc_args.arg));
+         break;
+
+       default:
+         fformat (stdout, "[%04d] unhandled event type %d\n",
+                  i, e->event_type);
+         break;
+       }
+
+      index++;
+
+      if (index == q->maxsize)
+       index = 0;
+    }
+}
+
 static uword
 session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
                       vlib_frame_t * frame)
 {
   session_manager_main_t *smm = vnet_get_session_manager_main ();
-  session_fifo_event_t *my_fifo_events, *e;
+  session_fifo_event_t *my_pending_event_vector, *e;
+  session_fifo_event_t *my_fifo_events;
   u32 n_to_dequeue, n_events;
   unix_shared_memory_queue_t *q;
   application_t *app;
@@ -401,6 +472,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   u32 my_thread_index = vm->thread_index;
   int i, rv;
   f64 now = vlib_time_now (vm);
+  void (*fp) (void *);
 
   SESSION_EVT_DBG (SESSION_EVT_POLL_GAP_TRACK, smm, my_thread_index);
 
@@ -416,11 +488,13 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
   if (PREDICT_FALSE (q == 0))
     return 0;
 
+  my_fifo_events = smm->free_event_vector[my_thread_index];
+
   /* min number of events we can dequeue without blocking */
   n_to_dequeue = q->cursize;
-  my_fifo_events = smm->fifo_events[my_thread_index];
+  my_pending_event_vector = smm->pending_event_vector[my_thread_index];
 
-  if (n_to_dequeue == 0 && vec_len (my_fifo_events) == 0)
+  if (n_to_dequeue == 0 && vec_len (my_pending_event_vector) == 0)
     return 0;
 
   SESSION_EVT_DBG (SESSION_EVT_DEQ_NODE, 0);
@@ -430,7 +504,7 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
    * over them again without dequeuing new ones.
    */
   /* XXX: Block senders to sessions that can't keep up */
-  if (vec_len (my_fifo_events) >= 100)
+  if (0 && vec_len (my_pending_event_vector) >= 100)
     {
       clib_warning ("too many fifo events unsolved");
       goto skip_dequeue;
@@ -451,7 +525,10 @@ session_queue_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
     (void) pthread_cond_broadcast (&q->condvar);
   pthread_mutex_unlock (&q->mutex);
 
-  smm->fifo_events[my_thread_index] = my_fifo_events;
+  vec_append (my_fifo_events, my_pending_event_vector);
+
+  _vec_len (my_pending_event_vector) = 0;
+  smm->pending_event_vector[my_thread_index] = my_pending_event_vector;
 
 skip_dequeue:
   n_events = vec_len (my_fifo_events);
@@ -481,9 +558,12 @@ skip_dequeue:
                                                        my_thread_index,
                                                        &n_tx_packets);
          /* Out of buffers */
-         if (rv < 0)
-           goto done;
-
+         if (PREDICT_FALSE (rv < 0))
+           {
+             vlib_node_increment_counter (vm, node->node_index,
+                                          SESSION_QUEUE_ERROR_NO_BUFFER, 1);
+             continue;
+           }
          break;
        case FIFO_EVENT_DISCONNECT:
          s0 = stream_session_get_from_handle (e0->session_handle);
@@ -496,30 +576,18 @@ skip_dequeue:
          app = application_get (s0->app_index);
          app->cb_fns.builtin_server_rx_callback (s0);
          break;
+       case FIFO_EVENT_RPC:
+         fp = e0->rpc_args.fp;
+         (*fp) (e0->rpc_args.arg);
+         break;
+
        default:
          clib_warning ("unhandled event type %d", e0->event_type);
        }
     }
 
-done:
-
-  /* Couldn't process all events. Probably out of buffers */
-  if (PREDICT_FALSE (i < n_events))
-    {
-      session_fifo_event_t *partially_read =
-       smm->evts_partially_read[my_thread_index];
-      vec_add (partially_read, &my_fifo_events[i], n_events - i);
-      vec_free (my_fifo_events);
-      smm->fifo_events[my_thread_index] = partially_read;
-      smm->evts_partially_read[my_thread_index] = 0;
-    }
-  else
-    {
-      vec_free (smm->fifo_events[my_thread_index]);
-      smm->fifo_events[my_thread_index] =
-       smm->evts_partially_read[my_thread_index];
-      smm->evts_partially_read[my_thread_index] = 0;
-    }
+  _vec_len (my_fifo_events) = 0;
+  smm->free_event_vector[my_thread_index] = my_fifo_events;
 
   vlib_node_increment_counter (vm, session_queue_node.index,
                               SESSION_QUEUE_ERROR_TX, n_tx_packets);