iovec datatype representation

opal/datatype/opal_convertor.c

@@ -324,6 +324,195 @@ int32_t opal_convertor_unpack( opal_convertor_t* pConv,
     return pConv->fAdvance( pConv, iov, out_size, max_data );
 }
 
+int32_t
+opal_iovec_set_position( opal_convertor_t *convertor, size_t *position )
+{
+    if( 0 == *position ) {
+        convertor->pStack[0].count = convertor->count;
+        convertor->pStack[0].disp = 0;
+        convertor->pStack[1].index = 0;
+        convertor->pStack[1].disp = 0;
+        convertor->bConverted = 0;
+
+        return 0;
+    }
+
+    const opal_datatype_t *pData = convertor->pDesc;
+    size_t done_count = *position / pData->size;
+
+    struct iovec *iov = pData->iov;
+    uint32_t i = 0;
+
+    convertor->bConverted = *position;
+    convertor->pStack[0].count = convertor->count - done_count;
+
+    size_t track = *position % pData->size;
+    for( i = 0; i < pData->iovcnt; track -= iov[i].iov_len, i++ ){
+        if( track < iov[i].iov_len )
+            break;
+    }
+
+    convertor->pStack[1].disp = track;
+    convertor->pStack[1].index = i;
+    convertor->pStack[0].disp = done_count * (pData->ub - pData->lb);
+
+    return 0;
+}
+
+int32_t
+opal_iovec_pack( opal_convertor_t *convertor,
+        struct iovec *out_iov,
+        uint32_t *out_size,
+        size_t *max_data )
+{
+    const opal_datatype_t *pData = convertor->pDesc;
+    struct iovec *iov = pData->iov;
+    char *dst,
+         *src = convertor->pBaseBuf + convertor->pStack[0].disp;
+    size_t track = *max_data, iov_track;
+    uint32_t i, iov_count = 0;
+
+    dst = out_iov[iov_count].iov_base;
+    iov_track = out_iov[iov_count].iov_len;
+
+    while( convertor->pStack[0].count ){
+        for( i = convertor->pStack[1].index; i < pData->iovcnt; i++ ) {
+restart_pack:
+            if( iov_track < (iov[i].iov_len - convertor->pStack[1].disp) && iov_track < track ){
+                memcpy( dst,
+                        src + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                        iov_track);
+                convertor->pStack[1].disp += iov_track;
+                convertor->pStack[1].index = i;
+                track -= iov_track;
+
+                iov_count++;
+                if( iov_count == *out_size )
+                    goto complete_pack;
+
+                dst = out_iov[iov_count].iov_base;
+                iov_track = out_iov[iov_count].iov_len;
+
+                goto restart_pack;
+            }
+
+            if( track < (iov[i].iov_len - convertor->pStack[1].disp) || track == 0 ){
+                memcpy( dst,
+                        src + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                        track);
+
+                convertor->pStack[1].disp  += track;
+                convertor->pStack[1].index = i;
+                track = 0;
+
+                goto complete_pack;
+            }
+
+            memcpy( dst,
+                    src + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                    iov[i].iov_len - convertor->pStack[1].disp );
+
+            dst += iov[i].iov_len - convertor->pStack[1].disp;
+            iov_track -= iov[i].iov_len - convertor->pStack[1].disp;
+            track -= iov[i].iov_len - convertor->pStack[1].disp;
+            convertor->pStack[1].disp = 0;
+        }
+
+        convertor->pStack[0].disp += pData->ub - pData->lb;
+        convertor->pStack[1].index = 0;
+        convertor->pStack[0].count--;
+
+        src += pData->ub - pData->lb;
+    }
+
+complete_pack:
+    *max_data -= track;
+    convertor->bConverted += *max_data;
+
+    if( convertor->bConverted < convertor->local_size ){
+        return 0;
+    }
+
+    convertor->flags |= CONVERTOR_COMPLETED;
+    return 1;
+}
+
+int32_t
+opal_iovec_unpack( opal_convertor_t *convertor,
+        struct iovec *out_iov,
+        uint32_t *out_size,
+        size_t *max_data )
+{
+    const opal_datatype_t *pData = convertor->pDesc;
+    struct iovec *iov = pData->iov;
+    char *dst = convertor->pBaseBuf + convertor->pStack[0].disp,
+         *src;
+    size_t track = *max_data, iov_track;
+    uint32_t i, iov_count = 0;
+
+    src = out_iov[iov_count].iov_base;
+    iov_track = out_iov[iov_count].iov_len;
+
+    while( convertor->pStack[0].count ) {
+        for( i = convertor->pStack[1].index; i < pData->iovcnt; i++ ) {
+restart_unpack:
+            if( iov_track < (iov[i].iov_len - convertor->pStack[1].disp) && iov_track < track ){
+                memcpy( dst + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                        src,
+                        iov_track);
+                convertor->pStack[1].disp += iov_track;
+                convertor->pStack[1].index = i;
+                track -= iov_track;
+
+                iov_count++;
+                if( iov_count == *out_size )
+                    goto complete_unpack;
+
+                src = out_iov[iov_count].iov_base;
+                iov_track = out_iov[iov_count].iov_len;
+
+                goto restart_unpack;
+            }
+
+            if( track < (iov[i].iov_len - convertor->pStack[1].disp) || track == 0 ) {
+                memcpy( dst + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                        src,
+                        track);
+
+                convertor->pStack[1].disp += track;
+                convertor->pStack[1].index = i;
+                track = 0;
+                goto complete_unpack;
+            }
+            memcpy( dst + (ptrdiff_t)(iov[i].iov_base) + convertor->pStack[1].disp,
+                    src,
+                    iov[i].iov_len - convertor->pStack[1].disp );
+
+            src += iov[i].iov_len - convertor->pStack[1].disp;
+            track -= iov[i].iov_len - convertor->pStack[1].disp;
+            iov_track -= iov[i].iov_len - convertor->pStack[1].disp;
+            convertor->pStack[1].disp = 0;
+        }
+
+        convertor->pStack[0].disp += pData->ub - pData->lb;
+        convertor->pStack[1].index = 0;
+        convertor->pStack[0].count--;
+        dst += pData->ub - pData->lb;
+    }
+
+
+complete_unpack:
+    *max_data -= track;
+    convertor->bConverted += *max_data;
+
+    if( convertor->bConverted < convertor->local_size ){
+        return 0;
+    }
+
+    convertor->flags |= CONVERTOR_COMPLETED;
+    return 1;
+}
+
 static inline int
 opal_convertor_create_stack_with_pos_contig( opal_convertor_t* pConvertor,
                                              size_t starting_point, const size_t* sizes )
@@ -427,9 +616,12 @@ int32_t opal_convertor_set_position_nocheck( opal_convertor_t* convertor,
     } else {
         if( (0 == (*position)) || ((*position) < convertor->bConverted) ) {
             rc = opal_convertor_create_stack_at_begining( convertor, opal_datatype_local_sizes );
-            if( 0 == (*position) ) return rc;
+            if( 0 == (*position) ){
+                opal_iovec_set_position( convertor, position );
+                return rc;
+            }
         }
-        rc = opal_convertor_generic_simple_position( convertor, position );
+        rc = opal_iovec_set_position( convertor, position );
         /**
          * If we have a non-contigous send convertor don't allow it move in the middle
          * of a predefined datatype, it won't be able to copy out the left-overs
@@ -438,7 +630,7 @@ int32_t opal_convertor_set_position_nocheck( opal_convertor_t* convertor,
          * case, we should be accepted by any receiver convertor.
          */
         if( CONVERTOR_SEND & convertor->flags ) {
-            convertor->bConverted -= convertor->partial_length;
+            convertor->bConverted = *position;
             convertor->partial_length = 0;
         }
     }
@@ -598,7 +790,7 @@ int32_t opal_convertor_prepare_for_recv( opal_convertor_t* convertor,
             if( convertor->pDesc->flags & OPAL_DATATYPE_FLAG_CONTIGUOUS ) {
                 convertor->fAdvance = opal_unpack_homogeneous_contig;
             } else {
-                convertor->fAdvance = opal_generic_simple_unpack;
+                convertor->fAdvance = opal_iovec_unpack;
             }
         }
     return OPAL_SUCCESS;
@@ -646,7 +838,7 @@ int32_t opal_convertor_prepare_for_send( opal_convertor_t* convertor,
                 else
                     convertor->fAdvance = opal_pack_homogeneous_contig_with_gaps;
             } else {
-                convertor->fAdvance = opal_generic_simple_pack;
+                convertor->fAdvance = opal_iovec_pack;
             }
         }
     return OPAL_SUCCESS;

opal/datatype/opal_convertor.h

@@ -133,6 +133,12 @@ static inline uint32_t opal_convertor_get_checksum( opal_convertor_t* convertor
     return convertor->checksum;
 }
 
+OPAL_DECLSPEC int32_t opal_generate_iovec( opal_datatype_t *pData );
+OPAL_DECLSPEC int32_t opal_iovec_set_position( opal_convertor_t *pConv, size_t *position );
+OPAL_DECLSPEC int32_t opal_iovec_pack( opal_convertor_t *outside_convertor, struct iovec *out_iov,
+                uint32_t *out_size, size_t *max_data );
+OPAL_DECLSPEC int32_t opal_iovec_unpack( opal_convertor_t *outside_convertor, struct iovec *out_iov,
+                uint32_t *out_size, size_t *max_data );
 
 /*
  *

opal/datatype/opal_datatype.h

@@ -126,6 +126,9 @@ struct opal_datatype_t {
     dt_type_desc_t     opt_desc; /**< short description of the data used when conversion is useless
                                       or in the send case (without conversion) */
 
+    struct iovec       *iov;     /**< iovec description */
+    uint32_t           iovcnt;   /**< number of iovec */
+
     size_t             *ptypes;  /**< array of basic predefined types that facilitate the computing
                                       of the remote size in heterogeneous environments. The length of the
                                       array is dependent on the maximum number of predefined datatypes of

opal/datatype/opal_datatype_create.c

@@ -55,6 +55,8 @@ static void opal_datatype_construct( opal_datatype_t* pData )
 
     pData->ptypes             = NULL;
     pData->loops              = 0;
+
+    pData->iov                = NULL;
 }
 
 static void opal_datatype_destruct( opal_datatype_t* datatype )

opal/datatype/opal_datatype_destroy.c

@@ -27,6 +27,8 @@ int32_t opal_datatype_destroy( opal_datatype_t** dt )
 {
     opal_datatype_t* pData = *dt;
 
+    free( pData->iov );
+
     if( (pData->flags & OPAL_DATATYPE_FLAG_PREDEFINED) &&
         (pData->super.obj_reference_count <= 1) )
         return OPAL_ERROR;

opal/datatype/opal_datatype_optimize.c

@@ -310,5 +310,39 @@ int32_t opal_datatype_commit( opal_datatype_t * pData )
         pLast->first_elem_disp = first_elem_disp;
         pLast->size            = pData->size;
     }
+
+    /* generate iovec */
+    if( pData->iov == NULL )
+        opal_generate_iovec( pData );
+
     return OPAL_SUCCESS;
 }
+
+int32_t
+opal_generate_iovec( opal_datatype_t *pData )
+{
+    opal_convertor_t *local_convertor = opal_convertor_create( opal_local_arch, 0 );
+
+    size_t max = SIZE_MAX;
+    int rc;
+
+    opal_convertor_prepare_for_send( local_convertor, pData, 1, (void*)0 );
+
+    pData->iovcnt = 512;
+    uint32_t save_iov = 0;
+    uint32_t leftover_iovec = 0;
+    do {
+        pData->iovcnt *= 2;
+        pData->iov = realloc( pData->iov, sizeof(struct iovec) * pData->iovcnt );
+        leftover_iovec = pData->iovcnt - save_iov;
+        rc = opal_convertor_raw( local_convertor, pData->iov + save_iov, &leftover_iovec, &max );
+        leftover_iovec += save_iov;  /* for the case we leave the loop */
+        save_iov = pData->iovcnt;
+
+    } while (0 == rc);
+
+    pData->iov = realloc( pData->iov, sizeof(struct iovec) * leftover_iovec );
+    pData->iovcnt = leftover_iovec;
+
+    return 1;
+}