session: fix segment alloc/free worker race

Avoid scenarios where a worker allocates a segment but while it drops
the segment manager writer lock and acquires the reader lock another
worker uses the segment and frees it.

Type: fix

Thanks to wanghanlin@corp.netease.com for the report.

Signed-off-by: Florin Coras <fcoras@cisco.com>
Change-Id: I0a88d738c51b33fd07c34916f125c98806861a06

2 files changed, 119 insertions, 70 deletions

diff --git a/src/vnet/session/segment_manager.c b/src/vnet/session/segment_manager.c
index 184d3fd597d..cbf6207060c 100644
--- a/src/vnet/session/segment_manager.c
+++ b/src/vnet/session/segment_manager.c
@@ -89,7 +89,7 @@ segment_manager_segment_index (segment_manager_t * sm, fifo_segment_t * seg)
  */
 static inline int
 segment_manager_add_segment_inline (segment_manager_t *sm, uword segment_size,
-				    u8 notify_app, u8 flags)
+				    u8 notify_app, u8 flags, u8 need_lock)
 {
   segment_manager_main_t *smm = &sm_main;
   segment_manager_props_t *props;
@@ -112,7 +112,7 @@ segment_manager_add_segment_inline (segment_manager_t *sm, uword segment_size,
   /*
    * Allocate fifo segment and grab lock if needed
    */
-  if (vlib_num_workers ())
+  if (need_lock)
     clib_rwlock_writer_lock (&sm->segments_rwlock);
 
   pool_get_zero (sm->segments, fs);
@@ -179,7 +179,7 @@ segment_manager_add_segment_inline (segment_manager_t *sm, uword segment_size,
     }
 done:
 
-  if (vlib_num_workers ())
+  if (need_lock)
     clib_rwlock_writer_unlock (&sm->segments_rwlock);
 
   return fs_index;
@@ -189,14 +189,16 @@ int
 segment_manager_add_segment (segment_manager_t *sm, uword segment_size,
 			     u8 notify_app)
 {
-  return segment_manager_add_segment_inline (sm, segment_size, notify_app, 0);
+  return segment_manager_add_segment_inline (sm, segment_size, notify_app,
+					     0 /* flags */, 0 /* need_lock */);
 }
 
 int
 segment_manager_add_segment2 (segment_manager_t *sm, uword segment_size,
 			      u8 flags)
 {
-  return segment_manager_add_segment_inline (sm, segment_size, 0, flags);
+  return segment_manager_add_segment_inline (sm, segment_size, 0, flags,
+					     vlib_num_workers ());
 }
 
 /**
@@ -333,12 +335,6 @@ segment_manager_segment_reader_unlock (segment_manager_t * sm)
   clib_rwlock_reader_unlock (&sm->segments_rwlock);
 }
 
-void
-segment_manager_segment_writer_unlock (segment_manager_t * sm)
-{
-  clib_rwlock_writer_unlock (&sm->segments_rwlock);
-}
-
 segment_manager_t *
 segment_manager_alloc (void)
 {
@@ -740,92 +736,131 @@ segment_manager_try_alloc_fifos (fifo_segment_t * fifo_segment,
   return 0;
 }
 
+static inline int
+sm_lookup_segment_and_alloc_fifos (segment_manager_t *sm,
+				   segment_manager_props_t *props,
+				   u32 thread_index, svm_fifo_t **rx_fifo,
+				   svm_fifo_t **tx_fifo)
+{
+  uword free_bytes, max_free_bytes;
+  fifo_segment_t *cur, *fs = 0;
+  u32 fs_index;
+  int rv;
+
+  max_free_bytes = props->rx_fifo_size + props->tx_fifo_size - 1;
+
+  pool_foreach (cur, sm->segments)
+    {
+      if (fifo_segment_flags (cur) & FIFO_SEGMENT_F_CUSTOM_USE)
+	continue;
+      free_bytes = fifo_segment_available_bytes (cur);
+      if (free_bytes > max_free_bytes)
+	{
+	  max_free_bytes = free_bytes;
+	  fs = cur;
+	}
+    }
+
+  if (PREDICT_FALSE (!fs))
+    return SESSION_E_SEG_NO_SPACE;
+
+  fs_index = segment_manager_segment_index (sm, fs);
+  rv = segment_manager_try_alloc_fifos (fs, thread_index, props->rx_fifo_size,
+					props->tx_fifo_size, rx_fifo, tx_fifo);
+
+  return rv ? SESSION_E_SEG_NO_SPACE : fs_index;
+}
+
+static int
+sm_lock_and_alloc_segment_and_fifos (segment_manager_t *sm,
+				     segment_manager_props_t *props,
+				     u32 thread_index, svm_fifo_t **rx_fifo,
+				     svm_fifo_t **tx_fifo)
+{
+  int new_fs_index, rv;
+  fifo_segment_t *fs;
+
+  if (!props->add_segment)
+    return SESSION_E_SEG_NO_SPACE;
+
+  clib_rwlock_writer_lock (&sm->segments_rwlock);
+
+  /* Make sure there really is no free space. Another worker might've freed
+   * some fifos or allocated a segment */
+  rv = sm_lookup_segment_and_alloc_fifos (sm, props, thread_index, rx_fifo,
+					  tx_fifo);
+  if (rv > 0)
+    goto done;
+
+  new_fs_index =
+    segment_manager_add_segment (sm, 0 /* segment_size*/, 1 /* notify_app */);
+  if (new_fs_index < 0)
+    {
+      rv = SESSION_E_SEG_CREATE;
+      goto done;
+    }
+  fs = segment_manager_get_segment (sm, new_fs_index);
+  rv = segment_manager_try_alloc_fifos (fs, thread_index, props->rx_fifo_size,
+					props->tx_fifo_size, rx_fifo, tx_fifo);
+  if (rv)
+    {
+      clib_warning ("Added a segment, still can't allocate a fifo");
+      rv = SESSION_E_SEG_NO_SPACE2;
+      goto done;
+    }
+
+  rv = new_fs_index;
+
+done:
+
+  clib_rwlock_writer_unlock (&sm->segments_rwlock);
+
+  return rv;
+}
+
 int
 segment_manager_alloc_session_fifos (segment_manager_t * sm,
 				     u32 thread_index,
 				     svm_fifo_t ** rx_fifo,
 				     svm_fifo_t ** tx_fifo)
 {
-  int alloc_fail = 1, rv = 0, new_fs_index;
-  uword free_bytes, max_free_bytes = 0;
   segment_manager_props_t *props;
-  fifo_segment_t *fs = 0, *cur;
-  u32 sm_index, fs_index;
+  u32 sm_index;
+  int fs_index;
 
   props = segment_manager_properties_get (sm);
+  sm_index = segment_manager_index (sm);
 
   /*
-   * Find the first free segment to allocate the fifos in
+   * Fast path: find the first segment with enough free space and
+   * try to allocate the fifos. Done with reader lock
    */
 
   segment_manager_segment_reader_lock (sm);
 
-  pool_foreach (cur, sm->segments)  {
-      if (fifo_segment_flags (cur) & FIFO_SEGMENT_F_CUSTOM_USE)
-	continue;
-      free_bytes = fifo_segment_available_bytes (cur);
-      if (free_bytes > max_free_bytes)
-	{
-	  max_free_bytes = free_bytes;
-	  fs = cur;
-	}
-  }
-
-  if (fs)
-    {
-      alloc_fail = segment_manager_try_alloc_fifos (fs, thread_index,
-						    props->rx_fifo_size,
-						    props->tx_fifo_size,
-						    rx_fifo, tx_fifo);
-      /* On success, keep lock until fifos are initialized */
-      if (!alloc_fail)
-	goto alloc_success;
-    }
+  fs_index = sm_lookup_segment_and_alloc_fifos (sm, props, thread_index,
+						rx_fifo, tx_fifo);
 
   segment_manager_segment_reader_unlock (sm);
 
   /*
-   * Allocation failed, see if we can add a new segment
+   * Slow path: if no fifo segment or alloc fail grab writer lock and try
+   * to allocate new segment
    */
-  if (props->add_segment)
+  if (PREDICT_FALSE (fs_index < 0))
     {
-      if ((new_fs_index = segment_manager_add_segment (sm, 0, 1)) < 0)
-	{
-	  clib_warning ("Failed to add new segment");
-	  return SESSION_E_SEG_CREATE;
-	}
-      fs = segment_manager_get_segment_w_lock (sm, new_fs_index);
-      alloc_fail = segment_manager_try_alloc_fifos (fs, thread_index,
-						    props->rx_fifo_size,
-						    props->tx_fifo_size,
-						    rx_fifo, tx_fifo);
-      if (alloc_fail)
-	{
-	  clib_warning ("Added a segment, still can't allocate a fifo");
-	  segment_manager_segment_reader_unlock (sm);
-	  return SESSION_E_SEG_NO_SPACE2;
-	}
+      fs_index = sm_lock_and_alloc_segment_and_fifos (sm, props, thread_index,
+						      rx_fifo, tx_fifo);
+      if (fs_index < 0)
+	return fs_index;
     }
-  else
-    {
-      SESSION_DBG ("Can't add new seg and no space to allocate fifos!");
-      return SESSION_E_SEG_NO_SPACE;
-    }
-
-alloc_success:
-  ASSERT (rx_fifo && tx_fifo);
 
-  sm_index = segment_manager_index (sm);
-  fs_index = segment_manager_segment_index (sm, fs);
   (*tx_fifo)->segment_manager = sm_index;
   (*rx_fifo)->segment_manager = sm_index;
   (*tx_fifo)->segment_index = fs_index;
   (*rx_fifo)->segment_index = fs_index;
 
-  /* Drop the lock after app is notified */
-  segment_manager_segment_reader_unlock (sm);
-
-  return rv;
+  return 0;
 }
 
 void
diff --git a/src/vnet/session/segment_manager.h b/src/vnet/session/segment_manager.h
index 5a3d772ff02..8b722a02324 100644
--- a/src/vnet/session/segment_manager.h
+++ b/src/vnet/session/segment_manager.h
@@ -102,8 +102,23 @@ segment_manager_t *segment_manager_get (u32 index);
 segment_manager_t *segment_manager_get_if_valid (u32 index);
 u32 segment_manager_index (segment_manager_t * sm);
 
+/**
+ * Add segment without lock
+ *
+ * @param sm		Segment manager
+ * @param segment_size	Size of segment to be added
+ * @param notify_app	Flag set if app notification requested
+ */
 int segment_manager_add_segment (segment_manager_t *sm, uword segment_size,
 				 u8 notify_app);
+
+/**
+ * Add segment with lock
+ *
+ * @param sm		Segment manager
+ * @param segment_size	Size of segment to be added
+ * @param flags		Flags to be set on segment
+ */
 int segment_manager_add_segment2 (segment_manager_t *sm, uword segment_size,
 				  u8 flags);
 void segment_manager_del_segment (segment_manager_t * sm,
@@ -122,7 +137,6 @@ u64 segment_manager_make_segment_handle (u32 segment_manager_index,
 u64 segment_manager_segment_handle (segment_manager_t * sm,
 				    fifo_segment_t * segment);
 void segment_manager_segment_reader_unlock (segment_manager_t * sm);
-void segment_manager_segment_writer_unlock (segment_manager_t * sm);
 
 int segment_manager_alloc_session_fifos (segment_manager_t * sm,
 					 u32 thread_index,