1 files changed, 105 insertions, 18 deletions

diff --git a/src/vnet/session/application_local.c b/src/vnet/session/application_local.c
index 0abf03d0c15..88536dc27c6 100644
--- a/src/vnet/session/application_local.c
+++ b/src/vnet/session/application_local.c
@@ -41,9 +41,16 @@ typedef struct ct_segments_
   ct_segment_t *segments;
 } ct_segments_ctx_t;
 
+typedef struct ct_cleanup_req_
+{
+  u32 ct_index;
+} ct_cleanup_req_t;
+
 typedef struct ct_main_
 {
   ct_connection_t **connections;	/**< Per-worker connection pools */
+  ct_cleanup_req_t **pending_cleanups;
+  u8 *heave_cleanups;
   u32 n_workers;			/**< Number of vpp workers */
   u32 n_sessions;			/**< Cumulative sessions counter */
   u32 *ho_reusable;			/**< Vector of reusable ho indices */
@@ -864,31 +871,19 @@ global_scope:
 }
 
 static void
-ct_session_close (u32 ct_index, u32 thread_index)
+ct_session_postponed_cleanup (ct_connection_t *ct)
 {
-  ct_connection_t *ct, *peer_ct;
   app_worker_t *app_wrk;
   session_t *s;
 
-  ct = ct_connection_get (ct_index, thread_index);
   s = session_get (ct->c_s_index, ct->c_thread_index);
-  peer_ct = ct_connection_get (ct->peer_index, thread_index);
-  if (peer_ct)
-    {
-      peer_ct->peer_index = ~0;
-      /* Make sure session was allocated */
-      if (peer_ct->flags & CT_CONN_F_HALF_OPEN)
-	{
-	  ct_session_connect_notify (s, SESSION_E_REFUSED);
-	}
-      else if (peer_ct->c_s_index != ~0)
-	session_transport_closing_notify (&peer_ct->connection);
-      else
-	ct_connection_free (peer_ct);
-    }
+  app_wrk = app_worker_get_if_valid (s->app_wrk_index);
 
   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 */
@@ -901,7 +896,10 @@ ct_session_close (u32 ct_index, u32 thread_index)
        * 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_SESSION);
+	{
+	  app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_TRANSPORT);
+	  app_worker_cleanup_notify (app_wrk, s, SESSION_CLEANUP_SESSION);
+	}
 
       ct_session_dealloc_fifos (ct, s->rx_fifo, s->tx_fifo);
       session_free (s);
@@ -910,6 +908,93 @@ ct_session_close (u32 ct_index, u32 thread_index)
   ct_connection_free (ct);
 }
 
+static void
+ct_handle_cleanups (void *args)
+{
+  uword thread_index = pointer_to_uword (args);
+  const u32 max_cleanups = 100;
+  ct_main_t *cm = &ct_main;
+  ct_cleanup_req_t *req;
+  ct_connection_t *ct;
+  u32 n_to_handle = 0;
+  session_t *s;
+
+  cm->heave_cleanups[thread_index] = 0;
+  n_to_handle = clib_fifo_elts (cm->pending_cleanups[thread_index]);
+  n_to_handle = clib_min (n_to_handle, max_cleanups);
+
+  while (n_to_handle)
+    {
+      clib_fifo_sub2 (cm->pending_cleanups[thread_index], req);
+      ct = ct_connection_get (req->ct_index, thread_index);
+      s = session_get (ct->c_s_index, ct->c_thread_index);
+      if (!svm_fifo_has_event (s->tx_fifo))
+	ct_session_postponed_cleanup (ct);
+      else
+	clib_fifo_add1 (cm->pending_cleanups[thread_index], *req);
+      n_to_handle -= 1;
+    }
+
+  if (clib_fifo_elts (cm->pending_cleanups[thread_index]))
+    {
+      cm->heave_cleanups[thread_index] = 1;
+      session_send_rpc_evt_to_thread_force (
+	thread_index, ct_handle_cleanups,
+	uword_to_pointer (thread_index, void *));
+    }
+}
+
+static void
+ct_program_cleanup (ct_connection_t *ct)
+{
+  ct_main_t *cm = &ct_main;
+  ct_cleanup_req_t *req;
+  uword thread_index;
+
+  thread_index = ct->c_thread_index;
+  clib_fifo_add2 (cm->pending_cleanups[thread_index], req);
+  req->ct_index = ct->c_c_index;
+
+  if (cm->heave_cleanups[thread_index])
+    return;
+
+  cm->heave_cleanups[thread_index] = 1;
+  session_send_rpc_evt_to_thread_force (
+    thread_index, ct_handle_cleanups, uword_to_pointer (thread_index, void *));
+}
+
+static void
+ct_session_close (u32 ct_index, u32 thread_index)
+{
+  ct_connection_t *ct, *peer_ct;
+  session_t *s;
+
+  ct = ct_connection_get (ct_index, thread_index);
+  s = session_get (ct->c_s_index, ct->c_thread_index);
+  peer_ct = ct_connection_get (ct->peer_index, thread_index);
+  if (peer_ct)
+    {
+      peer_ct->peer_index = ~0;
+      /* Make sure session was allocated */
+      if (peer_ct->flags & CT_CONN_F_HALF_OPEN)
+	{
+	  ct_session_connect_notify (s, SESSION_E_REFUSED);
+	}
+      else if (peer_ct->c_s_index != ~0)
+	session_transport_closing_notify (&peer_ct->connection);
+      else
+	{
+	  /* should not happen */
+	  clib_warning ("ct peer without session");
+	  ct_connection_free (peer_ct);
+	}
+    }
+
+  /* Do not send closed notify to make sure pending tx events are
+   * still delivered and program cleanup */
+  ct_program_cleanup (ct);
+}
+
 static transport_connection_t *
 ct_session_get (u32 ct_index, u32 thread_index)
 {
@@ -1046,6 +1131,8 @@ ct_enable_disable (vlib_main_t * vm, u8 is_en)
 
   cm->n_workers = vlib_num_workers ();
   vec_validate (cm->connections, cm->n_workers);
+  vec_validate (cm->pending_cleanups, cm->n_workers);
+  vec_validate (cm->heave_cleanups, cm->n_workers);
   clib_spinlock_init (&cm->ho_reuseable_lock);
   clib_rwlock_init (&cm->app_segs_lock);
   return 0;