From be64d9868706f22abe82f4558108fde6fa620219 Mon Sep 17 00:00:00 2001
From: George Bosilca <bosilca@icl.utk.edu>
Date: Thu, 19 Jan 2006 06:45:29 +0000
Subject: [PATCH] We're thread safe again. The atomic compare-and-swap was not
 used in the correct way allowing the descriptors to vanish. The PML was
 thinking that they are in the btl_cache when they weren't ... It lead to
 memory consumption on most environments when compiled with thread enabled.
 After modification the latency went down by nearly 0.5 microseconds.

Simple way to trigger the bug: limit the number of maximum items in the free list and run
any communication intensive application (like Netpipe).

This commit was SVN r8741.
---
 ompi/mca/bml/bml.h | 83 ++++++++++++++++++++--------------------------
 1 file changed, 36 insertions(+), 47 deletions(-)

diff --git a/ompi/mca/bml/bml.h b/ompi/mca/bml/bml.h
index 734e832ab6..109730fd87 100644
--- a/ompi/mca/bml/bml.h
+++ b/ompi/mca/bml/bml.h
@@ -232,7 +232,6 @@ static inline void mca_bml_base_free(mca_bml_base_btl_t* bml_btl, mca_btl_base_d
     /* The previous function is supposed to release the des object
      * so we should not touch it anymore.
      */
-    /*des->des_context = NULL;*/
 }
 
 static inline int mca_bml_base_send(mca_bml_base_btl_t* bml_btl, mca_btl_base_descriptor_t* des, mca_btl_base_tag_t tag) { 
@@ -299,65 +298,55 @@ static inline void mca_bml_base_prepare_dst(mca_bml_base_btl_t* bml_btl,
     }
 }
 
+static inline mca_btl_base_descriptor_t*
+mca_bml_base_btl_des_alloc( mca_bml_base_btl_t* bml_btl,
+			    size_t alloc_size,
+			    size_t seg_size )
+{
+    mca_btl_base_descriptor_t* des;
 
-
-
+    if( NULL != (des = bml_btl->btl_cache) ) { 
 #if OMPI_HAVE_THREAD_SUPPORT
-#define MCA_BML_BASE_BTL_DES_ALLOC(bml_btl, des, alloc_size, seg_size)                   \
-do  {                                                                                    \
-     if(NULL != (des = bml_btl->btl_cache)) {                                            \
-        /* atomically acquire the cached descriptor */                                   \
-        if(opal_atomic_cmpset_ptr(&bml_btl->btl_cache, des, NULL) == 0) {                \
-            bml_btl->btl_cache = NULL;                                                   \
-        } else {                                                                         \
-            des = bml_btl->btl_alloc(bml_btl->btl, alloc_size);                          \
-        }                                                                                \
-    } else {                                                                             \
-            des = bml_btl->btl_alloc(bml_btl->btl, alloc_size);                          \
-    }                                                                                    \
-    des->des_src->seg_len = seg_size;                                                    \
-    des->des_context = (void*) bml_btl;                                                  \
-} while(0)                                                                               
+        /* atomically acquire the cached descriptor */
+        if(opal_atomic_cmpset_ptr(&bml_btl->btl_cache, des, NULL) == 0) {
+            des = bml_btl->btl_alloc(bml_btl->btl, alloc_size);
+        }
 #else
-#define MCA_BML_BASE_BTL_DES_ALLOC(bml_btl, descriptor, alloc_size, seg_size)            \
-do {                                                                                     \
-    if(NULL != (descriptor = bml_btl->btl_cache)) {                                      \
-        bml_btl->btl_cache = NULL;                                                       \
-    } else {                                                                             \
-        descriptor = bml_btl->btl_alloc(bml_btl->btl, alloc_size);                       \
-    }                                                                                    \
-    descriptor->des_src->seg_len = seg_size;                                             \
-    descriptor->des_context = (void*) bml_btl;                                           \
-} while(0)
+        bml_btl->btl_cache = NULL;
 #endif
+    } else {
+        des = bml_btl->btl_alloc(bml_btl->btl, alloc_size);
+    }
+    des->des_src->seg_len = seg_size;
+    des->des_context = (void*) bml_btl;
+    return des;
+}
+#define MCA_BML_BASE_BTL_DES_ALLOC(bml_btl, des, alloc_size, seg_size)    \
+    des = mca_bml_base_btl_des_alloc( bml_btl, alloc_size, seg_size )
 
 
 /**
  * Return a descriptor
  */
 
+static inline void mca_bml_base_btl_des_return( mca_bml_base_btl_t* bml_btl,
+						mca_btl_base_descriptor_t* des)
+{
 #if OMPI_HAVE_THREAD_SUPPORT
-#define MCA_BML_BASE_BTL_DES_RETURN( bml_btl, descriptor )                               \
-do {                                                                                     \
-    if(NULL == bml_btl->btl_cache) {                                                     \
-        if(opal_atomic_cmpset_ptr(&bml_btl->btl_cache,NULL,descriptor) == 0) {           \
-             bml_btl->btl_free(bml_btl->btl,descriptor);                                 \
-        }                                                                                \
-    } else {                                                                             \
-        bml_btl->btl_free(bml_btl->btl,descriptor);                                      \
-    }                                                                                    \
-} while(0)
+    if( opal_atomic_cmpset_ptr(&bml_btl->btl_cache,NULL,des) == 0 ) {
 #else
-#define MCA_BML_BASE_BTL_DES_RETURN(bml_btl, descriptor)                                 \
-do {                                                                                     \
-    if(NULL == bml_btl->btl_cache) {                                                     \
-        bml_btl->btl_cache = descriptor;                                                 \
-    } else {                                                                             \
-        bml_btl->btl_free(bml_btl->btl,descriptor);                                      \
-    }                                                                                    \
-} while(0)
+    if( NULL == bml_btl->btl_cache ) {
+        bml_btl->btl_cache = des;
+    } else {
 #endif
-
+        bml_btl->btl_free( bml_btl->btl, des );
+    }
+}
+#define MCA_BML_BASE_BTL_DES_RETURN( bml_btl, descriptor ) \
+do { \
+    mca_bml_base_btl_des_return( bml_btl, descriptor ); \
+    descriptor = NULL; \
+} while(0)
 
 /*
  *  BML component interface functions and datatype.