Fixed a bug with available space in sender based.

This commit was SVN r15889.

ompi/mca/pml/v/protocol_pessimist/vprotocol_pessimist_sender_based.c

@@ -27,7 +27,7 @@ int vprotocol_pessimist_sender_based_init(const char *mmapfile, size_t size)
     sb.sb_length = size;
     sb.sb_pagesize = getpagesize();
     sb.sb_cursor = sb.sb_addr = (uintptr_t) NULL;
-    sb.sb_vacant = 0;
+    sb.sb_available = 0;
     sb.sb_comm = MPI_COMM_NULL;
     
     sprintf(path, "%s"OPAL_PATH_SEP"%s", orte_process_info.proc_session_dir, 
@@ -81,8 +81,11 @@ void vprotocol_pessimist_sender_based_alloc(size_t len)
     sb.sb_offset -= sb.sb_cursor; 
 
     /* Adjusting sb_length for the largest application message to fit   */
+    len += sb.sb_cursor + sizeof(vprotocol_pessimist_sender_based_header_t);
     if(sb.sb_length < len)
-        sb.sb_length = len + sb.sb_cursor;
+        sb.sb_length = len;
+    /* How much space left for application data */
+    sb.sb_available = sb.sb_length - sb.sb_cursor;
 
     if(-1 == lseek(sb.sb_fd, sb.sb_offset + sb.sb_length, SEEK_SET))
     {
@@ -108,8 +111,6 @@ void vprotocol_pessimist_sender_based_alloc(size_t len)
         close(sb.sb_fd);
         ompi_mpi_abort(MPI_COMM_NULL, MPI_ERR_NO_SPACE, false);
     }
-    sb.sb_vacant = sb.sb_length - sb.sb_cursor - /* make room for the header */
-                        sizeof(vprotocol_pessimist_sender_based_header_t);
     sb.sb_cursor += sb.sb_addr; /* set absolute addr of sender_based buffer */
     V_OUTPUT_VERBOSE(30, "pessimist:\tsb\tgrow\toffset %llu\tlength %llu\tbase %p\tcursor %p", (unsigned long long) sb.sb_offset, (unsigned long long) sb.sb_length, (void *) sb.sb_addr, (void *) sb.sb_cursor);
 }   

ompi/mca/pml/v/protocol_pessimist/vprotocol_pessimist_sender_based.h

@@ -26,7 +26,7 @@ typedef struct vprotocol_pessimist_sender_based_t
     uintptr_t sb_addr;      /* base address of mmaped segment */
     size_t sb_length;       /* length of mmaped segment */
     uintptr_t sb_cursor;    /* current pointer to writeable memory */
-    size_t sb_vacant;       /* available space before end of segment */
+    size_t sb_available;    /* available space before end of segment */
 } vprotocol_pessimist_sender_based_t;
 
 #include "vprotocol_pessimist.h"
@@ -49,7 +49,6 @@ void vprotocol_pessimist_sender_based_finalize(void);
   */
 void vprotocol_pessimist_sender_based_alloc(size_t len);
 
-
 #define __SENDER_BASED_CNVTOR_PACK(req) do {                                  \
     if( 0 != req->req_bytes_packed ) {                                        \
         ompi_convertor_t conv;                                                \
@@ -95,7 +94,7 @@ void vprotocol_pessimist_sender_based_alloc(size_t len);
                                                                               \
     __SENDER_BASED_CNVTOR_PACK(req);                                          \
     mca_vprotocol_pessimist.sender_based.sb_cursor += sbhdr->size;            \
-    mca_vprotocol_pessimist.sender_based.sb_vacant -= (sbhdr->size +          \
+    mca_vprotocol_pessimist.sender_based.sb_available -= (sbhdr->size +       \
             sizeof(vprotocol_pessimist_sender_based_header_t));               \
     V_OUTPUT_VERBOSE(70, "pessimist:\tsb\twrite\t%"PRIpclock"\tsize %lu", VPESSIMIST_REQ(&req->req_base)->reqid, sbhdr->size + sizeof(vprotocol_pessimist_sender_based_header_t); \
 } while(0)
@@ -115,7 +114,7 @@ static inline void __SENDER_BASED_PACK(mca_pml_base_send_request_t *req) {
  
     __SENDER_BASED_CNVTOR_PACK(req); 
     mca_vprotocol_pessimist.sender_based.sb_cursor += sbhdr->size;
-    mca_vprotocol_pessimist.sender_based.sb_vacant -= (sbhdr->size + 
+    mca_vprotocol_pessimist.sender_based.sb_available -= (sbhdr->size + 
             sizeof(vprotocol_pessimist_sender_based_header_t));
     V_OUTPUT_VERBOSE(70, "pessimist:\tsb\twrite\t%"PRIpclock"\tsize %lu", VPESSIMIST_REQ(&req->req_base)->reqid, sbhdr->size + sizeof(vprotocol_pessimist_sender_based_header_t)); 
 }
@@ -126,8 +125,9 @@ static inline void __SENDER_BASED_PACK(mca_pml_base_send_request_t *req) {
   */
 #define VPROTOCOL_PESSIMIST_SENDER_BASED_COPY(REQ) do {                       \
     mca_pml_base_send_request_t *req = (mca_pml_base_send_request_t *) (REQ); \
-    if(req->req_bytes_packed >=                                               \
+    if(req->req_bytes_packed +                                                \
-            mca_vprotocol_pessimist.sender_based.sb_vacant)                   \
+       sizeof(vprotocol_pessimist_sender_based_header_t) >=                   \
+       mca_vprotocol_pessimist.sender_based.sb_available)                     \
     {                                                                         \
         vprotocol_pessimist_sender_based_alloc(req->req_bytes_packed);        \
     }                                                                         \