Session layer refactoring

Major refactoring of the session layer api
- Add attatch api for application binding to the the session layer
- Simplify listen/connect calls
- Update application CLI
- Add transport endpoint to accept callback
- Associate segment manager to application and allow for multiple
  binds/connects per app

Additional:
- svm fifo cleanup
- add fifo free, format fns
- add fifo offset enqueue unit test

Change-Id: Id93a65047de61afc2bf3d58c9b544339c02065af
Signed-off-by: Florin Coras <fcoras@cisco.com>
Signed-off-by: Dave Barach <dave@barachs.net>

7 files changed, 476 insertions, 61 deletions

diff --git a/src/vnet/tcp/builtin_client.c b/src/vnet/tcp/builtin_client.c
index 9e8e1561..f8fbf28c 100644
--- a/src/vnet/tcp/builtin_client.c
+++ b/src/vnet/tcp/builtin_client.c
@@ -237,8 +237,7 @@ tclient_thread_fn (void *arg)
 	  memset (dmp, 0, sizeof (*dmp));
 	  dmp->_vl_msg_id = ntohs (VL_API_DISCONNECT_SESSION);
 	  dmp->client_index = tm->my_client_index;
-	  dmp->session_index = sp->vpp_session_index;
-	  dmp->session_thread_index = sp->vpp_session_thread;
+	  dmp->handle = sp->vpp_session_handle;
 	  vl_msg_api_send_shmem (tm->vl_input_queue, (u8 *) & dmp);
 	  pool_put (tm->sessions, sp);
 	}
@@ -253,9 +252,10 @@ tclient_thread_fn (void *arg)
 static void
 vl_api_memclnt_create_reply_t_handler (vl_api_memclnt_create_reply_t * mp)
 {
+  vlib_main_t *vm = vlib_get_main ();
   tclient_main_t *tm = &tclient_main;
-
   tm->my_client_index = mp->index;
+  vlib_process_signal_event (vm, tm->node_index, 1 /* evt */ , 0 /* data */ );
 }
 
 static void
@@ -264,7 +264,6 @@ vl_api_connect_uri_reply_t_handler (vl_api_connect_uri_reply_t * mp)
   tclient_main_t *tm = &tclient_main;
   session_t *session;
   u32 session_index;
-  u64 key;
   i32 retval = /* clib_net_to_host_u32 ( */ mp->retval /*) */ ;
 
   if (retval < 0)
@@ -291,24 +290,24 @@ vl_api_connect_uri_reply_t_handler (vl_api_connect_uri_reply_t * mp)
   session->server_rx_fifo->client_session_index = session_index;
   session->server_tx_fifo = (svm_fifo_t *) mp->server_tx_fifo;
   session->server_tx_fifo->client_session_index = session_index;
-
-  session->vpp_session_index = mp->session_index;
-  session->vpp_session_thread = mp->session_thread_index;
+  session->vpp_session_handle = mp->handle;
 
   /* Add it to the session lookup table */
-  key = (((u64) mp->session_thread_index) << 32) | (u64) mp->session_index;
-  hash_set (tm->session_index_by_vpp_handles, key, session_index);
+  hash_set (tm->session_index_by_vpp_handles, mp->handle, session_index);
 
   tm->ready_connections++;
 }
 
-static void
+static int
 create_api_loopback (tclient_main_t * tm)
 {
+  vlib_main_t *vm = vlib_get_main ();
   vl_api_memclnt_create_t _m, *mp = &_m;
   extern void vl_api_memclnt_create_t_handler (vl_api_memclnt_create_t *);
   api_main_t *am = &api_main;
   vl_shmem_hdr_t *shmem_hdr;
+  uword *event_data = 0, event_type;
+  int resolved = 0;
 
   /*
    * Create a "loopback" API client connection
@@ -324,6 +323,25 @@ create_api_loopback (tclient_main_t * tm)
   strncpy ((char *) mp->name, "tcp_tester", sizeof (mp->name) - 1);
 
   vl_api_memclnt_create_t_handler (mp);
+
+  /* Wait for reply */
+  tm->node_index = vlib_get_current_process (vm)->node_runtime.node_index;
+  vlib_process_wait_for_event_or_clock (vm, 1.0);
+  event_type = vlib_process_get_events (vm, &event_data);
+  switch (event_type)
+    {
+    case 1:
+      resolved = 1;
+      break;
+    case ~0:
+      /* timed out */
+      break;
+    default:
+      clib_warning ("unknown event_type %d", event_type);
+    }
+  if (!resolved)
+    return -1;
+  return 0;
 }
 
 #define foreach_tclient_static_api_msg       	\
@@ -333,17 +351,7 @@ _(CONNECT_URI_REPLY, connect_uri_reply)
 static clib_error_t *
 tclient_api_hookup (vlib_main_t * vm)
 {
-  tclient_main_t *tm = &tclient_main;
   vl_msg_api_msg_config_t _c, *c = &_c;
-  int i;
-
-  /* Init test data */
-  vec_validate (tm->connect_test_data, 64 * 1024 - 1);
-  for (i = 0; i < vec_len (tm->connect_test_data); i++)
-    tm->connect_test_data[i] = i & 0xff;
-
-  tm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
-  vec_validate (tm->rx_buf, vec_len (tm->connect_test_data) - 1);
 
   /* Hook up client-side static APIs to our handlers */
 #define _(N,n) do {                                             \
@@ -365,18 +373,105 @@ tclient_api_hookup (vlib_main_t * vm)
   return 0;
 }
 
-VLIB_API_INIT_FUNCTION (tclient_api_hookup);
+static int
+tcp_test_clients_init (vlib_main_t * vm)
+{
+  tclient_main_t *tm = &tclient_main;
+  int i;
+
+  tclient_api_hookup (vm);
+  if (create_api_loopback (tm))
+    return -1;
+
+  /* Init test data */
+  vec_validate (tm->connect_test_data, 64 * 1024 - 1);
+  for (i = 0; i < vec_len (tm->connect_test_data); i++)
+    tm->connect_test_data[i] = i & 0xff;
+
+  tm->session_index_by_vpp_handles = hash_create (0, sizeof (uword));
+  vec_validate (tm->rx_buf, vec_len (tm->connect_test_data) - 1);
+
+  tm->is_init = 1;
+
+  return 0;
+}
+
+static void
+builtin_session_reset_callback (stream_session_t * s)
+{
+  return;
+}
+
+static int
+builtin_session_connected_callback (u32 app_index, u32 api_context,
+				    stream_session_t * s, u8 code)
+{
+  return 0;
+}
+
+static int
+builtin_session_create_callback (stream_session_t * s)
+{
+  return 0;
+}
+
+static void
+builtin_session_disconnect_callback (stream_session_t * s)
+{
+  return;
+}
+
+static int
+builtin_server_rx_callback (stream_session_t * s)
+{
+  return 0;
+}
+
+/* *INDENT-OFF* */
+static session_cb_vft_t builtin_clients = {
+    .session_reset_callback = builtin_session_reset_callback,
+    .session_connected_callback = builtin_session_connected_callback,
+    .session_accept_callback = builtin_session_create_callback,
+    .session_disconnect_callback = builtin_session_disconnect_callback,
+    .builtin_server_rx_callback = builtin_server_rx_callback
+};
+/* *INDENT-ON* */
+
+static int
+attach_builtin_test_clients ()
+{
+  vnet_app_attach_args_t _a, *a = &_a;
+  u8 segment_name[128];
+  u32 segment_name_length;
+  u64 options[16];
+
+  segment_name_length = ARRAY_LEN (segment_name);
+
+  memset (a, 0, sizeof (*a));
+  memset (options, 0, sizeof (options));
+
+  a->api_client_index = ~0;
+  a->segment_name = segment_name;
+  a->segment_name_length = segment_name_length;
+  a->session_cb_vft = &builtin_clients;
+
+  options[SESSION_OPTIONS_ACCEPT_COOKIE] = 0x12345678;
+  options[SESSION_OPTIONS_SEGMENT_SIZE] = (2 << 30);	/*$$$$ config / arg */
+  a->options = options;
+
+  return vnet_application_attach (a);
+}
 
 static clib_error_t *
 test_tcp_clients_command_fn (vlib_main_t * vm,
 			     unformat_input_t * input,
 			     vlib_cli_command_t * cmd)
 {
+  tclient_main_t *tm = &tclient_main;
   u8 *connect_uri = (u8 *) "tcp://6.0.1.1/1234";
   u8 *uri;
-  tclient_main_t *tm = &tclient_main;
-  int i;
   u32 n_clients = 1;
+  int i;
 
   tm->bytes_to_send = 8192;
   tm->n_iterations = 1;
@@ -397,14 +492,19 @@ test_tcp_clients_command_fn (vlib_main_t * vm,
 				  format_unformat_error, input);
     }
 
+  if (tm->is_init == 0)
+    {
+      if (tcp_test_clients_init (vm))
+	return clib_error_return (0, "failed init");
+    }
+
   tm->ready_connections = 0;
   tm->expected_connections = n_clients;
+
   uri = connect_uri;
   if (tm->connect_uri)
     uri = tm->connect_uri;
 
-  create_api_loopback (tm);
-
 #if TCP_BUILTIN_CLIENT_PTHREAD
   /* Start a transmit thread */
   if (tm->client_thread_handle == 0)
@@ -420,6 +520,7 @@ test_tcp_clients_command_fn (vlib_main_t * vm,
     }
 #endif
   vnet_session_enable_disable (vm, 1 /* turn on TCP, etc. */ );
+  attach_builtin_test_clients ();
 
   /* Fire off connect requests, in something approaching a normal manner */
   for (i = 0; i < n_clients; i++)
@@ -461,6 +562,16 @@ VLIB_CLI_COMMAND (test_clients_command, static) =
 };
 /* *INDENT-ON* */
 
+clib_error_t *
+tcp_test_clients_main_init (vlib_main_t * vm)
+{
+  tclient_main_t *tm = &tclient_main;
+  tm->is_init = 0;
+  return 0;
+}
+
+VLIB_INIT_FUNCTION (tcp_test_clients_main_init);
+
 /*
  * fd.io coding-style-patch-verification: ON
  *
diff --git a/src/vnet/tcp/builtin_client.h b/src/vnet/tcp/builtin_client.h
index 64030302..2bd87c07 100644
--- a/src/vnet/tcp/builtin_client.h
+++ b/src/vnet/tcp/builtin_client.h
@@ -39,8 +39,7 @@ typedef struct
   svm_fifo_t *server_rx_fifo;
   svm_fifo_t *server_tx_fifo;
 
-  u32 vpp_session_index;
-  u32 vpp_session_thread;
+  u64 vpp_session_handle;
 } session_t;
 
 typedef struct
@@ -110,6 +109,10 @@ typedef struct
   u32 client_bytes_received;
   u8 test_return_packets;
 
+  u8 is_init;
+
+  u32 node_index;
+
   /* convenience */
   vlib_main_t *vlib_main;
   vnet_main_t *vnet_main;
diff --git a/src/vnet/tcp/builtin_server.c b/src/vnet/tcp/builtin_server.c
index 917d4bd3..8308e3d9 100644
--- a/src/vnet/tcp/builtin_server.c
+++ b/src/vnet/tcp/builtin_server.c
@@ -18,17 +18,46 @@
 #include <vnet/session/application.h>
 #include <vnet/session/application_interface.h>
 
+/* define message IDs */
+#include <vpp/api/vpe_msg_enum.h>
+
+/* define message structures */
+#define vl_typedefs
+#include <vpp/api/vpe_all_api_h.h>
+#undef vl_typedefs
+
+/* define generated endian-swappers */
+#define vl_endianfun
+#include <vpp/api/vpe_all_api_h.h>
+#undef vl_endianfun
+
+/* instantiate all the print functions we know about */
+#define vl_print(handle, ...) vlib_cli_output (handle, __VA_ARGS__)
+#define vl_printfun
+#include <vpp/api/vpe_all_api_h.h>
+#undef vl_printfun
+
 typedef struct
 {
   u8 *rx_buf;
   unix_shared_memory_queue_t **vpp_queue;
-  u32 byte_index;
+  u64 byte_index;
+
+  /* Sever's event queue */
+  unix_shared_memory_queue_t *vl_input_queue;
+
+  /* API client handle */
+  u32 my_client_index;
+
+  u32 app_index;
+
+  /* process node index for evnt scheduling */
+  u32 node_index;
   vlib_main_t *vlib_main;
 } builtin_server_main_t;
 
 builtin_server_main_t builtin_server_main;
 
-
 int
 builtin_session_accept_callback (stream_session_t * s)
 {
@@ -45,9 +74,13 @@ builtin_session_accept_callback (stream_session_t * s)
 void
 builtin_session_disconnect_callback (stream_session_t * s)
 {
+  builtin_server_main_t *bsm = &builtin_server_main;
+  vnet_disconnect_args_t _a, *a = &_a;
   clib_warning ("called...");
 
-  vnet_disconnect_session (s->session_index, s->thread_index);
+  a->handle = stream_session_handle (s);
+  a->app_index = bsm->app_index;
+  vnet_disconnect_session (a);
 }
 
 void
@@ -60,7 +93,7 @@ builtin_session_reset_callback (stream_session_t * s)
 
 
 int
-builtin_session_connected_callback (u32 client_index,
+builtin_session_connected_callback (u32 app_index, u32 api_context,
 				    stream_session_t * s, u8 is_fail)
 {
   clib_warning ("called...");
@@ -91,7 +124,7 @@ test_bytes (builtin_server_main_t * bsm, int actual_transfer)
     {
       if (bsm->rx_buf[i] != ((bsm->byte_index + i) & 0xff))
 	{
-	  clib_warning ("at %d expected %d got %d", bsm->byte_index + i,
+	  clib_warning ("at %lld expected %d got %d", bsm->byte_index + i,
 			(bsm->byte_index + i) & 0xff, bsm->rx_buf[i]);
 	}
     }
@@ -190,23 +223,66 @@ static session_cb_vft_t builtin_session_cb_vft = {
   .session_reset_callback = builtin_session_reset_callback
 };
 
+/* Abuse VPP's input queue */
 static int
-server_create (vlib_main_t * vm)
+create_api_loopback (vlib_main_t * vm)
 {
-  vnet_bind_args_t _a, *a = &_a;
-  u64 options[SESSION_OPTIONS_N_OPTIONS];
-  char segment_name[128];
-  u32 num_threads;
-  vlib_thread_main_t *vtm = vlib_get_thread_main ();
+  builtin_server_main_t *bsm = &builtin_server_main;
+  vl_api_memclnt_create_t _m, *mp = &_m;
+  extern void vl_api_memclnt_create_t_handler (vl_api_memclnt_create_t *);
+  api_main_t *am = &api_main;
+  vl_shmem_hdr_t *shmem_hdr;
+  uword *event_data = 0, event_type;
+  int resolved = 0;
 
-  num_threads = 1 /* main thread */  + vtm->n_threads;
-  vec_validate (builtin_server_main.vpp_queue, num_threads - 1);
+  /*
+   * Create a "loopback" API client connection
+   * Don't do things like this unless you know what you're doing...
+   */
+
+  shmem_hdr = am->shmem_hdr;
+  bsm->vl_input_queue = shmem_hdr->vl_input_queue;
+  memset (mp, 0, sizeof (*mp));
+  mp->_vl_msg_id = VL_API_MEMCLNT_CREATE;
+  mp->context = 0xFEEDFACE;
+  mp->input_queue = (u64) bsm->vl_input_queue;
+  strncpy ((char *) mp->name, "tcp_test_server", sizeof (mp->name) - 1);
+
+  vl_api_memclnt_create_t_handler (mp);
+
+  /* Wait for reply */
+  bsm->node_index = vlib_get_current_process (vm)->node_runtime.node_index;
+  vlib_process_wait_for_event_or_clock (vm, 1.0);
+  event_type = vlib_process_get_events (vm, &event_data);
+  switch (event_type)
+    {
+    case 1:
+      resolved = 1;
+      break;
+    case ~0:
+      /* timed out */
+      break;
+    default:
+      clib_warning ("unknown event_type %d", event_type);
+    }
+  if (!resolved)
+    return -1;
+
+  return 0;
+}
+
+static int
+server_attach ()
+{
+  builtin_server_main_t *bsm = &builtin_server_main;
+  u8 segment_name[128];
+  u64 options[SESSION_OPTIONS_N_OPTIONS];
+  vnet_app_attach_args_t _a, *a = &_a;
 
   memset (a, 0, sizeof (*a));
   memset (options, 0, sizeof (options));
 
-  a->uri = "tcp://0.0.0.0/1234";
-  a->api_client_index = ~0;
+  a->api_client_index = bsm->my_client_index;
   a->session_cb_vft = &builtin_session_cb_vft;
   a->options = options;
   a->options[SESSION_OPTIONS_SEGMENT_SIZE] = 128 << 20;
@@ -215,9 +291,94 @@ server_create (vlib_main_t * vm)
   a->segment_name = segment_name;
   a->segment_name_length = ARRAY_LEN (segment_name);
 
+  if (vnet_application_attach (a))
+    {
+      clib_warning ("failed to attach server");
+      return -1;
+    }
+  bsm->app_index = a->app_index;
+  return 0;
+}
+
+static int
+server_listen ()
+{
+  builtin_server_main_t *bsm = &builtin_server_main;
+  vnet_bind_args_t _a, *a = &_a;
+  memset (a, 0, sizeof (*a));
+  a->app_index = bsm->app_index;
+  a->uri = "tcp://0.0.0.0/1234";
   return vnet_bind_uri (a);
 }
 
+static int
+server_create (vlib_main_t * vm)
+{
+  builtin_server_main_t *bsm = &builtin_server_main;
+  u32 num_threads;
+  vlib_thread_main_t *vtm = vlib_get_thread_main ();
+
+  if (bsm->my_client_index == (u32) ~ 0)
+    {
+      if (create_api_loopback (vm))
+	return -1;
+    }
+
+  num_threads = 1 /* main thread */  + vtm->n_threads;
+  vec_validate (builtin_server_main.vpp_queue, num_threads - 1);
+
+  if (server_attach ())
+    {
+      clib_warning ("failed to attach server");
+      return -1;
+    }
+  if (server_listen ())
+    {
+      clib_warning ("failed to start listening");
+      return -1;
+    }
+  return 0;
+}
+
+/* Get our api client index */
+static void
+vl_api_memclnt_create_reply_t_handler (vl_api_memclnt_create_reply_t * mp)
+{
+  vlib_main_t *vm = vlib_get_main ();
+  builtin_server_main_t *bsm = &builtin_server_main;
+  bsm->my_client_index = mp->index;
+  vlib_process_signal_event (vm, bsm->node_index, 1 /* evt */ ,
+			     0 /* data */ );
+}
+
+#define foreach_tcp_builtin_server_api_msg      		\
+_(MEMCLNT_CREATE_REPLY, memclnt_create_reply)   		\
+
+static clib_error_t *
+tcp_builtin_server_api_hookup (vlib_main_t * vm)
+{
+  vl_msg_api_msg_config_t _c, *c = &_c;
+
+  /* Hook up client-side static APIs to our handlers */
+#define _(N,n) do {                                             \
+    c->id = VL_API_##N;                                         \
+    c->name = #n;                                               \
+    c->handler = vl_api_##n##_t_handler;                        \
+    c->cleanup = vl_noop_handler;                               \
+    c->endian = vl_api_##n##_t_endian;                          \
+    c->print = vl_api_##n##_t_print;                            \
+    c->size = sizeof(vl_api_##n##_t);                           \
+    c->traced = 1; /* trace, so these msgs print */             \
+    c->replay = 0; /* don't replay client create/delete msgs */ \
+    c->message_bounce = 0; /* don't bounce this message */	\
+    vl_msg_api_config(c);} while (0);
+
+  foreach_tcp_builtin_server_api_msg;
+#undef _
+
+  return 0;
+}
+
 static clib_error_t *
 server_create_command_fn (vlib_main_t * vm,
 			  unformat_input_t * input, vlib_cli_command_t * cmd)
@@ -234,6 +395,7 @@ server_create_command_fn (vlib_main_t * vm,
     }
 #endif
 
+  tcp_builtin_server_api_hookup (vm);
   vnet_session_enable_disable (vm, 1 /* turn on TCP, etc. */ );
   rv = server_create (vm);
   switch (rv)
@@ -249,12 +411,22 @@ server_create_command_fn (vlib_main_t * vm,
 /* *INDENT-OFF* */
 VLIB_CLI_COMMAND (server_create_command, static) =
 {
-  .path = "test server",
-  .short_help = "test server",
+  .path = "test tcp server",
+  .short_help = "test tcp server",
   .function = server_create_command_fn,
 };
 /* *INDENT-ON* */
 
+clib_error_t *
+builtin_tcp_server_main_init (vlib_main_t * vm)
+{
+  builtin_server_main_t *bsm = &builtin_server_main;
+  bsm->my_client_index = ~0;
+  return 0;
+}
+
+VLIB_INIT_FUNCTION (builtin_tcp_server_main_init);
+
 /*
 * fd.io coding-style-patch-verification: ON
 *
diff --git a/src/vnet/tcp/tcp.c b/src/vnet/tcp/tcp.c
index b6c34828..a0c66b9f 100644
--- a/src/vnet/tcp/tcp.c
+++ b/src/vnet/tcp/tcp.c
@@ -34,14 +34,19 @@ tcp_connection_bind (u32 session_index, ip46_address_t * ip,
   listener->c_lcl_port = clib_host_to_net_u16 (port_host_byte_order);
 
   if (is_ip4)
-    listener->c_lcl_ip4.as_u32 = ip->ip4.as_u32;
+    {
+      listener->c_lcl_ip4.as_u32 = ip->ip4.as_u32;
+      listener->c_is_ip4 = 1;
+      listener->c_proto = SESSION_TYPE_IP4_TCP;
+    }
   else
-    clib_memcpy (&listener->c_lcl_ip6, &ip->ip6, sizeof (ip6_address_t));
+    {
+      clib_memcpy (&listener->c_lcl_ip6, &ip->ip6, sizeof (ip6_address_t));
+      listener->c_proto = SESSION_TYPE_IP6_TCP;
+    }
 
   listener->c_s_index = session_index;
-  listener->c_proto = SESSION_TYPE_IP4_TCP;
   listener->state = TCP_STATE_LISTEN;
-  listener->c_is_ip4 = 1;
 
   tcp_connection_timers_init (listener);
 
@@ -62,7 +67,6 @@ tcp_session_bind_ip6 (u32 session_index, ip46_address_t * ip,
 		      u16 port_host_byte_order)
 {
   return tcp_connection_bind (session_index, ip, port_host_byte_order, 0);
-
 }
 
 static void
@@ -397,6 +401,7 @@ tcp_connection_open (ip46_address_t * rmt_addr, u16 rmt_port, u8 is_ip4)
   tc->c_lcl_port = clib_host_to_net_u16 (lcl_port);
   tc->c_c_index = tc - tm->half_open_connections;
   tc->c_is_ip4 = is_ip4;
+  tc->c_proto = is_ip4 ? SESSION_TYPE_IP4_TCP : SESSION_TYPE_IP6_TCP;
 
   /* The other connection vars will be initialized after SYN ACK */
   tcp_connection_timers_init (tc);
@@ -518,7 +523,10 @@ format_tcp_session (u8 * s, va_list * args)
   tcp_connection_t *tc;
 
   tc = tcp_connection_get (tci, thread_index);
-  return format (s, "%U", format_tcp_connection, tc);
+  if (tc)
+    return format (s, "%U", format_tcp_connection, tc);
+  else
+    return format (s, "empty");
 }
 
 u8 *
diff --git a/src/vnet/tcp/tcp.h b/src/vnet/tcp/tcp.h
index 2f5da108..93f3245d 100644
--- a/src/vnet/tcp/tcp.h
+++ b/src/vnet/tcp/tcp.h
@@ -100,8 +100,6 @@ extern timer_expiration_handler tcp_timer_retransmit_syn_handler;
 #define TCP_RTO_SYN_RETRIES 3	/* SYN retries without doubling RTO */
 #define TCP_RTO_INIT 1 * THZ	/* Initial retransmit timer */
 
-void tcp_update_time (f64 now, u32 thread_index);
-
 /** TCP connection flags */
 #define foreach_tcp_connection_flag             \
   _(SNDACK, "Send ACK")                         \
@@ -481,6 +479,13 @@ tcp_time_now (void)
   return clib_cpu_time_now () * tcp_main.tstamp_ticks_per_clock;
 }
 
+always_inline void
+tcp_update_time (f64 now, u32 thread_index)
+{
+  tw_timer_expire_timers_16t_2w_512sl (&tcp_main.timer_wheels[thread_index],
+				       now);
+}
+
 u32 tcp_push_header (transport_connection_t * tconn, vlib_buffer_t * b);
 
 u32
diff --git a/src/vnet/tcp/tcp_input.c b/src/vnet/tcp/tcp_input.c
index 7e9fa47b..ae1f92d5 100644
--- a/src/vnet/tcp/tcp_input.c
+++ b/src/vnet/tcp/tcp_input.c
@@ -1841,6 +1841,7 @@ tcp46_rcv_process_inline (vlib_main_t * vm, vlib_node_runtime_t * node,
 	    case TCP_STATE_ESTABLISHED:
 	    case TCP_STATE_FIN_WAIT_1:
 	    case TCP_STATE_FIN_WAIT_2:
+	      vlib_buffer_advance (b0, n_advance_bytes0);
 	      error0 = tcp_segment_rcv (tm, tc0, b0, n_data_bytes0, &next0);
 	      break;
 	    case TCP_STATE_CLOSE_WAIT:
@@ -2410,12 +2411,6 @@ VLIB_REGISTER_NODE (tcp6_input_node) =
 /* *INDENT-ON* */
 
 VLIB_NODE_FUNCTION_MULTIARCH (tcp6_input_node, tcp6_input);
-void
-tcp_update_time (f64 now, u32 thread_index)
-{
-  tcp_main_t *tm = vnet_get_tcp_main ();
-  tw_timer_expire_timers_16t_2w_512sl (&tm->timer_wheels[thread_index], now);
-}
 
 static void
 tcp_dispatch_table_init (tcp_main_t * tm)
diff --git a/src/vnet/tcp/tcp_test.c b/src/vnet/tcp/tcp_test.c
index 0725bb04..3dbbdf6f 100644
--- a/src/vnet/tcp/tcp_test.c
+++ b/src/vnet/tcp/tcp_test.c
@@ -12,7 +12,6 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 #include <vnet/tcp/tcp.h>
 
 #define TCP_TEST_I(_cond, _comment, _args...)			\
@@ -174,6 +173,118 @@ tcp_test_sack ()
   return 0;
 }
 
+static int
+tcp_test_fifo (vlib_main_t * vm, unformat_input_t * input)
+{
+  svm_fifo_t *f;
+  u32 fifo_size = 1 << 20;
+  u32 *test_data = 0;
+  u32 offset;
+  int i, rv;
+  u32 data_word, test_data_len;
+
+  /* $$$ parse args */
+  test_data_len = fifo_size / sizeof (u32);
+  vec_validate (test_data, test_data_len - 1);
+
+  for (i = 0; i < vec_len (test_data); i++)
+    test_data[i] = i;
+
+  f = svm_fifo_create (fifo_size);
+
+  /* Paint fifo data vector with -1's */
+  memset (f->data, 0xFF, test_data_len);
+
+  /* Enqueue an initial (un-dequeued) chunk */
+  rv = svm_fifo_enqueue_nowait (f, 0 /* pid */ ,
+				sizeof (u32), (u8 *) test_data);
+
+  if (rv != sizeof (u32))
+    {
+      clib_warning ("enqueue returned %d", rv);
+      goto out;
+    }
+
+  /*
+   * Create 3 chunks in the future. The offsets are relative
+   * to the current fifo tail
+   */
+  for (i = 0; i < 3; i++)
+    {
+      offset = (2 * i + 1) * sizeof (u32);
+      vlib_cli_output (vm, "add offset %d", offset);
+
+      rv = svm_fifo_enqueue_with_offset
+	(f, 0 /* pid */ , offset, sizeof (u32),
+	 (u8 *) (test_data + ((offset + sizeof (u32)) / sizeof (u32))));
+
+      if (rv)
+	{
+	  clib_warning ("enqueue returned %d", rv);
+	  goto out;
+	}
+    }
+
+  /* Paint missing data backwards */
+  for (i = 3; i > 0; i--)
+    {
+      offset = (2 * i + 0) * sizeof (u32);
+
+      vlib_cli_output (vm, "add offset %d", offset);
+
+      rv = svm_fifo_enqueue_with_offset
+	(f, 0 /* pid */ , offset, sizeof (u32),
+	 (u8 *) (test_data + ((offset + sizeof (u32)) / sizeof (u32))));
+
+      if (rv)
+	{
+	  clib_warning ("enqueue returned %d", rv);
+	  goto out;
+	}
+    }
+
+  vlib_cli_output (vm, "fifo before missing link: %U",
+		   format_svm_fifo, f, 1 /* verbose */ );
+
+  /* Enqueue the missing u32 */
+  rv = svm_fifo_enqueue_nowait (f, 0 /* pid */ ,
+				sizeof (u32), (u8 *) (test_data + 1));
+  if (rv != 7 * sizeof (u32))
+    {
+      clib_warning ("enqueue returned %d", rv);
+      goto out;
+    }
+
+  vlib_cli_output (vm, "fifo after missing link: %U",
+		   format_svm_fifo, f, 1 /* verbose */ );
+
+  /* Collect results */
+  for (i = 0; i < 7; i++)
+    {
+      rv = svm_fifo_dequeue_nowait (f, 0 /* pid */ , sizeof (u32),
+				    (u8 *) & data_word);
+      if (rv != sizeof (u32))
+	{
+	  clib_warning ("dequeue returned %d", rv);
+	  goto out;
+	}
+      if (data_word != test_data[i])
+	{
+	  clib_warning ("recovered data %d not %d", data_word, test_data[i]);
+	  goto out;
+	}
+    }
+
+  clib_warning ("test complete...");
+
+out:
+  svm_fifo_free (f);
+  vec_free (test_data);
+  return 0;
+}
+
+
+
 static clib_error_t *
 tcp_test (vlib_main_t * vm,
 	  unformat_input_t * input, vlib_cli_command_t * cmd_arg)
@@ -186,6 +297,10 @@ tcp_test (vlib_main_t * vm,
 	{
 	  res = tcp_test_sack ();
 	}
+      else if (unformat (input, "fifo"))
+	{
+	  res = tcp_test_fifo (vm, input);
+	}
       else
 	{
 	  return clib_error_return (0, "unknown input `%U'",
@@ -203,10 +318,16 @@ tcp_test (vlib_main_t * vm,
     }
 }
 
+/* *INDENT-OFF* */
 VLIB_CLI_COMMAND (tcp_test_command, static) =
 {
-.path = "test tcp",.short_help = "internal tcp unit tests",.function =
-    tcp_test,};
+  .path = "test tcp",
+  .short_help = "internal tcp unit tests",
+  .function = tcp_test,
+};
+/* *INDENT-ON* */
+
+
 /*
  * fd.io coding-style-patch-verification: ON
  *