more cleanup

This commit was SVN r3431.

src/mca/ptl/mx/ptl_mx.c

@@ -253,7 +253,6 @@ int mca_ptl_mx_send_continue(
     mca_ptl_base_header_t* hdr;
     ompi_ptr_t match;
     mx_return_t mx_return;
-    uint64_t match_value;
     int rc;
 
     /* allocate fragment */
@@ -445,7 +444,8 @@ void mca_ptl_mx_matched(
             request->req_base.req_addr,     /* users buffer */
             hdr->hdr_frag.hdr_frag_offset,  /* offset in bytes into packed buffer */
             NULL );                         /* not allocating memory */
-                                                                                                             
+        ompi_convertor_get_packed_size(convertor, &request->req_bytes_packed);
+
         iov.iov_base = mx_frag->frag_data;
         iov.iov_len = hdr->hdr_frag.hdr_frag_length;
         ompi_convertor_unpack(convertor, &iov, &iov_count, &bytes_delivered, &free_after );

src/mca/ptl/mx/ptl_mx_module.c

@@ -159,7 +159,8 @@ static void mca_ptl_mx_match(void* context, uint64_t match_value, int size)
     /* first fragment - post a buffer */
     if(match_value == 0) {
 
-        frag->frag_recv.frag_base.frag_size = size - sizeof(mca_ptl_base_header_t);
+        frag->frag_size = size - sizeof(mca_ptl_base_header_t);
+        frag->frag_recv.frag_base.frag_size = frag->frag_size;
         frag->frag_recv.frag_base.frag_addr = frag->frag_data;
         frag->frag_recv.frag_is_buffered = true;
         frag->frag_segment_count = 2;
@@ -175,7 +176,7 @@ static void mca_ptl_mx_match(void* context, uint64_t match_value, int size)
         ompi_convertor_t* convertor = &frag->frag_recv.frag_base.frag_convertor;
 
         frag->frag_size = size;
-        frag->frag_recv.frag_base.frag_size = size - sizeof(mca_ptl_base_header_t);
+        frag->frag_recv.frag_base.frag_size = size;
         frag->frag_recv.frag_base.frag_header.hdr_common.hdr_type =
             MCA_PTL_HDR_TYPE_FRAG;
 
@@ -192,12 +193,29 @@ static void mca_ptl_mx_match(void* context, uint64_t match_value, int size)
                                                                                                           
         /* non-contiguous - allocate buffer for receive */
         if( 1 == ompi_convertor_need_buffers( convertor )  ||
-            request->req_bytes_packed < offset + frag->frag_recv.frag_base.frag_size) {
-            frag->frag_recv.frag_base.frag_addr = malloc(frag->frag_recv.frag_base.frag_size);
+            request->req_bytes_packed < offset + frag->frag_size ) {
+
+            /* TODO - use a fixed fragment size for non-contigous and convert 
+             * this to a free-list of buffers.
+            */
             frag->frag_recv.frag_is_buffered = true;
+            frag->frag_recv.frag_base.frag_addr = malloc(frag->frag_size);
+            if( NULL == frag->frag_recv.frag_base.frag_addr ) {
+                ompi_output(0, "mca_ptl_mx_match: unable to allocate buffer (%d)\n", frag->frag_size);
+                MCA_PTL_MX_RECV_FRAG_RETURN(frag);
+                return;
+            }
+
+            /* check for sending more than receiving */
+            if( offset > request->req_bytes_packed ) {
+                frag->frag_recv.frag_base.frag_size = 0;
+            } else if (offset + frag->frag_size > request->req_bytes_packed ) {
+                frag->frag_recv.frag_base.frag_size = request->req_bytes_packed - offset;
+            }
         /* calculate offset into users buffer */
         } else {
             frag->frag_recv.frag_base.frag_addr = ((unsigned char*)request->req_base.req_addr) + offset;
+            frag->frag_recv.frag_is_buffered = false;
         }
 
         /* dont receive a header */
@@ -205,7 +223,7 @@ static void mca_ptl_mx_match(void* context, uint64_t match_value, int size)
         segments = frag->frag_segments+1;
     }
     frag->frag_segments[1].segment_ptr = frag->frag_recv.frag_base.frag_addr;
-    frag->frag_segments[1].segment_length = frag->frag_recv.frag_base.frag_size;
+    frag->frag_segments[1].segment_length = frag->frag_size;
 
     mx_return = mx_irecv(
         ptl->mx_endpoint,
@@ -217,6 +235,7 @@ static void mca_ptl_mx_match(void* context, uint64_t match_value, int size)
         &frag->frag_request);
     if(mx_return != MX_SUCCESS) {
         ompi_output(0, "mca_ptl_mx_match: mx_irecv() failed with status=%dn", mx_return);
+        MCA_PTL_MX_RECV_FRAG_RETURN(frag);
     }
 }
 

src/mca/ptl/mx/ptl_mx_module.h

@@ -48,7 +48,6 @@ do {
             mca_ptl_mx_recv_frag_t* recvfrag = (mca_ptl_mx_recv_frag_t*)frag;       \
             mca_ptl_base_header_t* hdr =                                            \
                 &recvfrag->frag_recv.frag_base.frag_header;                         \
-            int rc;                                                                 \
             switch(hdr->hdr_common.hdr_type) {                                      \
                 case MCA_PTL_HDR_TYPE_MATCH:                                        \
                 {                                                                   \

src/mca/ptl/mx/ptl_mx_recvfrag.h

@@ -77,6 +77,7 @@ static inline void MCA_PTL_MX_RECV_FRAG_FRAG(mca_ptl_mx_recv_frag_t* frag)
         max_data = iov.iov_len;
         ompi_convertor_unpack( &frag->frag_recv.frag_base.frag_convertor,
                                &iov, &iov_count, &max_data, &free_after );
+        frag->frag_recv.frag_base.frag_size = max_data;
     }
                                                                                                           
     /* progress the request */

src/mca/ptl/mx/ptl_mx_sendfrag.h

@@ -24,7 +24,7 @@ struct mca_ptl_mx_send_frag_t {
    mx_request_t frag_request;
    mx_segment_t frag_segments[2];
    size_t frag_segment_count;
-   uint32_t frag_progress;
+   int32_t frag_progress;
 };
 typedef struct mca_ptl_mx_send_frag_t mca_ptl_mx_send_frag_t;
 
@@ -79,7 +79,7 @@ static inline void MCA_PTL_MX_SEND_FRAG_PROGRESS(mca_ptl_mx_send_frag_t* frag)
 {
    mca_pml_base_send_request_t* request = frag->frag_send.frag_request;
    bool frag_ack;
-   uint32_t frag_progress;
+   int32_t frag_progress;
 
     /* if this is an ack - simply return to pool */
     if(request == NULL) {