Start the cleanup process. I try to remove all useless operations from the critical path.

Start of the pool of memory that will be used for unexpected messages (just the framework at this moment).

This commit was SVN r3867.

src/mca/ptl/gm/src/ptl_gm.c

@@ -411,7 +411,7 @@ mca_ptl_gm_matched( mca_ptl_base_module_t * ptl,
     
     /* Now update the status of the fragment */
     if( ((mca_ptl_gm_recv_frag_t*)frag)->have_allocated_buffer == true ) {
-        free( ((mca_ptl_gm_recv_frag_t*)frag)->frag_recv.frag_base.frag_addr);
+        gm_release_local_buffer( ((mca_ptl_gm_recv_frag_t*)frag)->frag_recv.frag_base.frag_addr );
         ((mca_ptl_gm_recv_frag_t*)frag)->have_allocated_buffer = false;
     }
     

src/mca/ptl/gm/src/ptl_gm.h

@@ -221,6 +221,13 @@ extern void mca_ptl_gm_matched (struct mca_ptl_base_module_t *ptl,
      */
     extern int  mca_ptl_gm_finalize (struct mca_ptl_base_module_t *ptl);
 
+    /**
+     * Internally allocate memory for the unexpected messages. We will manage a list
+     * of such buffers in order to avoid too many memory allocations.
+     */
+    extern char* gm_get_local_buffer( void );
+    extern void gm_release_local_buffer( char* ptr );
+
 #if defined(c_plusplus) || defined(__cplusplus)
 }
 #endif

src/mca/ptl/gm/src/ptl_gm_component.c

@@ -480,7 +480,6 @@ mca_ptl_gm_component_init (int *num_ptl_modules,
     memcpy (ptls, mca_ptl_gm_component.gm_ptl_modules,
             mca_ptl_gm_component.gm_num_ptl_modules * sizeof(mca_ptl_gm_module_t *));
     *num_ptl_modules = mca_ptl_gm_component.gm_num_ptl_modules;
-
     return ptls;
 }
 
@@ -494,6 +493,15 @@ mca_ptl_gm_component_control (int param, void *value, size_t size)
     return OMPI_SUCCESS;
 }
 
+char* gm_get_local_buffer( void )
+{
+    return malloc( sizeof(char) * GM_BUF_SIZE );
+}
+
+void gm_release_local_buffer( char* ptr )
+{
+    free(ptr);
+}
 
 /*
  *  GM module progress.

src/mca/ptl/gm/src/ptl_gm_priv.c

@@ -105,9 +105,9 @@ int mca_ptl_gm_peer_send_continue( mca_ptl_gm_peer_t *ptl_peer,
     item = (ompi_list_item_t*)fragment->send_buf;
 
     if( (*size) <= mca_ptl_gm_component.gm_max_eager_size ) {  /* small protocol */
-	size_t max_data, remaining_bytes = fragment->send_frag.frag_base.frag_size;
-	int freeAfter;
-	unsigned int in_size;
+	size_t remaining_bytes = fragment->send_frag.frag_base.frag_size;
+	int32_t freeAfter;
+	uint32_t max_data, in_size;
 	struct iovec iov;
 	ompi_convertor_t *convertor = &(fragment->send_frag.frag_base.frag_convertor);
 	
@@ -298,7 +298,6 @@ int mca_ptl_gm_peer_send( mca_ptl_gm_peer_t *ptl_peer,
     return OMPI_SUCCESS;
 }
 
-
 void put_callback(struct gm_port *port,void * context, gm_status_t status)
 {
     mca_ptl_gm_module_t *ptl;
@@ -454,13 +453,9 @@ mca_ptl_gm_recv_frag_match( struct mca_ptl_gm_module_t *ptl,
     
     /* allocate a receive fragment */
     recv_frag = mca_ptl_gm_alloc_recv_frag( (struct mca_ptl_base_module_t*)ptl );
-    
-    recv_frag->frag_recv.frag_base.frag_owner = (struct mca_ptl_base_module_t*)ptl;
-    recv_frag->frag_recv.frag_base.frag_peer = NULL;
+    /*recv_frag->frag_recv.frag_base.frag_peer = NULL;
     recv_frag->frag_recv.frag_request = NULL;
-    recv_frag->frag_recv.frag_is_buffered = false;
-    
-    recv_frag->frag_recv.frag_base.frag_header.hdr_rndv = hdr->hdr_rndv;
+    */
     if( MCA_PTL_HDR_TYPE_MATCH == hdr->hdr_rndv.hdr_match.hdr_common.hdr_type ) {
         recv_frag->frag_recv.frag_base.frag_addr =
             (char *) hdr + sizeof(mca_ptl_base_match_header_t);
@@ -472,27 +467,34 @@ mca_ptl_gm_recv_frag_match( struct mca_ptl_gm_module_t *ptl,
 	recv_frag->frag_recv.frag_base.frag_size = hdr->hdr_rndv.hdr_frag_length;
     }
     
-    recv_frag->matched = false;
+    recv_frag->frag_recv.frag_is_buffered = false;
     recv_frag->have_allocated_buffer = false;
-    recv_frag->frag_ack_pending = false;
-    recv_frag->frag_progressed = 0;
-    recv_frag->frag_offset = 0;  /* initial frgment */
-    recv_frag->frag_bytes_processed = 0;
+    recv_frag->frag_offset = 0;  /* initial fragment */
 
+    recv_frag->frag_recv.frag_base.frag_header.hdr_rndv = hdr->hdr_rndv;
     matched = ptl->super.ptl_match( &(ptl->super),
                                     &(recv_frag->frag_recv),
                                     &(recv_frag->frag_recv.frag_base.frag_header.hdr_match) );
     if( !matched ) {
         size_t length = recv_frag->frag_recv.frag_base.frag_size;
+        
 	/* get some memory and copy the data inside. We can then release the receive buffer */
         if( 0 != length ) {
-            char* ptr = (char*)malloc( sizeof(char) * length );
+            char* ptr = (char*)gm_get_local_buffer();
+	    if (NULL == ptr) {
+		ompi_output(0, "[%s:%d] error in allocating memory \n", __FILE__, __LINE__);
+	    }
             recv_frag->have_allocated_buffer = true;
             memcpy( ptr, recv_frag->frag_recv.frag_base.frag_addr, length );
             recv_frag->frag_recv.frag_base.frag_addr = ptr;
         } else {
             recv_frag->frag_recv.frag_base.frag_addr = NULL;
         }
+        recv_frag->matched = false;
+        recv_frag->frag_ack_pending = false;
+        recv_frag->frag_progressed = 0;
+        recv_frag->frag_bytes_processed = 0;
+
         return recv_frag;
     }
     return NULL;
@@ -592,7 +594,6 @@ mca_ptl_gm_recv_frag_frag( struct mca_ptl_gm_module_t *ptl,
 	    recv_frag = NULL;
 	} else {  /* large message => we have to create a receive fragment */
 	    recv_frag = mca_ptl_gm_alloc_recv_frag( (struct mca_ptl_base_module_t*)ptl );
-	    recv_frag->frag_recv.frag_base.frag_owner = (struct mca_ptl_base_module_t*)ptl;
 	    recv_frag->frag_recv.frag_request = request;
 	    recv_frag->frag_recv.frag_base.frag_header.hdr_frag = hdr->hdr_frag;
 	    recv_frag->frag_recv.frag_base.frag_peer =
@@ -745,19 +746,6 @@ int mca_ptl_gm_analyze_recv_event( struct mca_ptl_gm_module_t* ptl, gm_recv_even
     case GM_HIGH_PEER_RECV_EVENT:
 	mesg = gm_ntohp(event->recv.buffer);
 	frag = ptl_gm_handle_recv( ptl, event );
-	if( (frag != NULL) && !(frag->matched) ) {
-	    /* allocate temporary buffer: temporary until the fragment will be finally matched */
-	    char* buffer = malloc( GM_BUF_SIZE );
-	    if (NULL == buffer) {
-		ompi_output(0, "[%s:%d] error in allocating memory \n", __FILE__, __LINE__);
-	    }
-	    /* copy the data from the registered buffer to the newly allocated one */
-	    memcpy( buffer, mesg, gm_ntohl(event->recv.length) );
-	    /* associate the buffer with the unexpected fragment */
-	    frag->frag_recv.frag_base.frag_addr = (void *)buffer;
-	    /* mark the fragment as having pending buffers */
-	    frag->have_allocated_buffer = true;
-	}
 	gm_provide_receive_buffer( ptl->gm_port, mesg, GM_SIZE, GM_LOW_PRIORITY );
 	break;
     case GM_NO_RECV_EVENT:

src/mca/ptl/gm/src/ptl_gm_sendfrag.c

@@ -90,7 +90,6 @@ mca_ptl_gm_alloc_send_frag( struct mca_ptl_gm_module_t *ptl,
     return sendfrag;
 }
 
-
 int mca_ptl_gm_send_frag_done( mca_ptl_gm_send_frag_t * frag,
                                mca_pml_base_send_request_t * req )
 {
@@ -141,7 +140,6 @@ int mca_ptl_gm_send_ack_init( struct mca_ptl_gm_send_frag_t* ack,
     return OMPI_SUCCESS;
 }
 
-
 int mca_ptl_gm_put_frag_init( struct mca_ptl_gm_send_frag_t* putfrag,
 			      struct mca_ptl_gm_peer_t* ptl_peer,
 			      struct mca_ptl_gm_module_t* gm_ptl,
@@ -203,10 +201,13 @@ mca_ptl_gm_alloc_recv_frag( struct mca_ptl_base_module_t *ptl )
 {
     int rc;
     ompi_list_item_t* item;
+    mca_ptl_gm_recv_frag_t* frag;
 
     OMPI_FREE_LIST_GET( &(((mca_ptl_gm_module_t *)ptl)->gm_recv_frags_free), item, rc );
 
-    return (mca_ptl_gm_recv_frag_t *)item;
+    frag = (mca_ptl_gm_recv_frag_t*)item;
+    frag->frag_recv.frag_base.frag_owner = (struct mca_ptl_base_module_t*)ptl;
+    return frag;
 
 }