root/opal/datatype/opal_datatype_unpack.c

DEFINITIONS

1. opal_unpack_homogeneous_contig_function
2. opal_unpack_partial_datatype
3. opal_generic_simple_unpack_function
4. opal_unpack_general_function

   1 /* -*- Mode: C; c-basic-offset:4 ; -*- */
   2 /*
   3  * Copyright (c) 2004-2006 The Trustees of Indiana University and Indiana
   4  *                         University Research and Technology
   5  *                         Corporation.  All rights reserved.
   6  * Copyright (c) 2004-2017 The University of Tennessee and The University
   7  *                         of Tennessee Research Foundation.  All rights
   8  *                         reserved.
   9  * Copyright (c) 2004-2006 High Performance Computing Center Stuttgart,
  10  *                         University of Stuttgart.  All rights reserved.
  11  * Copyright (c) 2004-2006 The Regents of the University of California.
  12  *                         All rights reserved.
  13  * Copyright (c) 2008-2009 Oak Ridge National Labs.  All rights reserved.
  14  * Copyright (c) 2011      NVIDIA Corporation.  All rights reserved.
  15  * Copyright (c) 2013      Cisco Systems, Inc.  All rights reserved.
  16  * Copyright (c) 2017-2018 Research Organization for Information Science
  17  *                         and Technology (RIST).  All rights reserved.
  18  * $COPYRIGHT$
  19  *
  20  * Additional copyrights may follow
  21  *
  22  * $HEADER$
  23  */
  24 
  25 #include "opal_config.h"
  26 
  27 #include <stddef.h>
  28 #include <stdio.h>
  29 
  30 #include "opal/datatype/opal_convertor_internal.h"
  31 #include "opal/datatype/opal_datatype_internal.h"
  32 
  33 #if OPAL_ENABLE_DEBUG
  34 #include "opal/util/output.h"
  35 
  36 #define DO_DEBUG(INST)  if( opal_unpack_debug ) { INST }
  37 #else
  38 #define DO_DEBUG(INST)
  39 #endif  /* OPAL_ENABLE_DEBUG */
  40 
  41 #include "opal/datatype/opal_datatype_checksum.h"
  42 #include "opal/datatype/opal_datatype_unpack.h"
  43 #include "opal/datatype/opal_datatype_prototypes.h"
  44 
  45 #if defined(CHECKSUM)
  46 #define opal_unpack_general_function            opal_unpack_general_checksum
  47 #define opal_unpack_homogeneous_contig_function opal_unpack_homogeneous_contig_checksum
  48 #define opal_generic_simple_unpack_function     opal_generic_simple_unpack_checksum
  49 #else
  50 #define opal_unpack_general_function            opal_unpack_general
  51 #define opal_unpack_homogeneous_contig_function opal_unpack_homogeneous_contig
  52 #define opal_generic_simple_unpack_function     opal_generic_simple_unpack
  53 #endif  /* defined(CHECKSUM) */
  54 
  55 
  56 /**
  57  * This function will be used to unpack all datatypes that have the contiguous flag set.
  58  * Several types of datatypes match this criterion, not only the contiguous one, but
  59  * the ones that have gaps in the beginning and/or at the end but where the data to
  60  * be unpacked is contiguous. However, this function only work for homogeneous cases
  61  * and the datatype that are contiguous and where the extent is equal to the size are
  62  * taken in account directly in the opal_convertor_unpack function (in convertor.c) for
  63  * the homogeneous case.
  64  */
  65 int32_t
  66 opal_unpack_homogeneous_contig_function( opal_convertor_t* pConv,
  67                                          struct iovec* iov,
  68                                          uint32_t* out_size,
  69                                          size_t* max_data )
  70 {
  71     const opal_datatype_t *pData = pConv->pDesc;
  72     unsigned char *user_memory, *packed_buffer;
  73     uint32_t iov_count, i;
  74     size_t bConverted, remaining, length, initial_bytes_converted = pConv->bConverted;
  75     dt_stack_t* stack = pConv->pStack;
  76     ptrdiff_t extent = pData->ub - pData->lb;
  77     ptrdiff_t initial_displ = pConv->use_desc->desc[pConv->use_desc->used].end_loop.first_elem_disp;
  78 
  79     DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( pBaseBuf %p, iov_count %d )\n",
  80                            (void*)pConv->pBaseBuf, *out_size ); );
  81     if( stack[1].type != opal_datatype_uint1.id ) {
  82         stack[1].count *= opal_datatype_basicDatatypes[stack[1].type]->size;
  83         stack[1].type = opal_datatype_uint1.id;
  84     }
  85     for( iov_count = 0; iov_count < (*out_size); iov_count++ ) {
  86         remaining = pConv->local_size - pConv->bConverted;
  87         if( 0 == remaining ) break;  /* we're done this time */
  88         if( remaining > iov[iov_count].iov_len )
  89             remaining = iov[iov_count].iov_len;
  90         packed_buffer = (unsigned char*)iov[iov_count].iov_base;
  91         bConverted = remaining; /* how much will get unpacked this time */
  92         user_memory = pConv->pBaseBuf + initial_displ;
  93 
  94         if( (ptrdiff_t)pData->size == extent ) {
  95             user_memory += pConv->bConverted;
  96             DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( user_memory %p, packed_buffer %p length %lu\n",
  97                                    (void*)user_memory, (void*)packed_buffer, (unsigned long)remaining ); );
  98 
  99             /* contiguous data or basic datatype with count */
 100             OPAL_DATATYPE_SAFEGUARD_POINTER( user_memory, remaining,
 101                                              pConv->pBaseBuf, pData, pConv->count );
 102             DO_DEBUG( opal_output( 0, "1. unpack contig dest %p src %p length %lu\n",
 103                                    (void*)user_memory, (void*)packed_buffer, (unsigned long)remaining ); );
 104             MEMCPY_CSUM( user_memory, packed_buffer, remaining, pConv );
 105         } else {
 106             user_memory += stack[0].disp + stack[1].disp;
 107 
 108             DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( user_memory %p, packed_buffer %p length %lu\n",
 109                                    (void*)user_memory, (void*)packed_buffer, (unsigned long)remaining ); );
 110 
 111             length = (0 == pConv->stack_pos ? 0 : stack[1].count);  /* left over from the last unpack */
 112             /* complete the last copy */
 113             if( (0 != length) && (length <= remaining) ) {
 114                 OPAL_DATATYPE_SAFEGUARD_POINTER( user_memory, length, pConv->pBaseBuf,
 115                                                  pData, pConv->count );
 116                 DO_DEBUG( opal_output( 0, "2. unpack dest %p src %p length %lu\n",
 117                                        (void*)user_memory, (void*)packed_buffer, (unsigned long)length ); );
 118                 MEMCPY_CSUM( user_memory, packed_buffer, length, pConv );
 119                 packed_buffer  += length;
 120                 user_memory    += (extent - (pData->size - length));
 121                 remaining      -= length;
 122                 stack[1].count -= length;
 123                 if( 0 == stack[1].count) { /* one completed element */
 124                     stack[0].count--;
 125                     stack[0].disp += extent;
 126                     if( 0 != stack[0].count ) {  /* not yet done */
 127                         stack[1].count = pData->size;
 128                         stack[1].disp = 0;
 129                     }
 130                 }
 131             }
 132             for( i = 0; pData->size <= remaining; i++ ) {
 133                 OPAL_DATATYPE_SAFEGUARD_POINTER( user_memory, pData->size, pConv->pBaseBuf,
 134                                                  pData, pConv->count );
 135                 DO_DEBUG( opal_output( 0, "3. unpack dest %p src %p length %lu\n",
 136                                        (void*)user_memory, (void*)packed_buffer, (unsigned long)pData->size ); );
 137                 MEMCPY_CSUM( user_memory, packed_buffer, pData->size, pConv );
 138                 packed_buffer += pData->size;
 139                 user_memory   += extent;
 140                 remaining     -= pData->size;
 141             }
 142             stack[0].count -= i;
 143             stack[0].disp  += (i * extent);
 144             stack[1].disp  += remaining;
 145             /* copy the last bits */
 146             if( 0 != remaining ) {
 147                 OPAL_DATATYPE_SAFEGUARD_POINTER( user_memory, remaining, pConv->pBaseBuf,
 148                                                  pData, pConv->count );
 149                 DO_DEBUG( opal_output( 0, "4. unpack dest %p src %p length %lu\n",
 150                                        (void*)user_memory, (void*)packed_buffer, (unsigned long)remaining ); );
 151                 MEMCPY_CSUM( user_memory, packed_buffer, remaining, pConv );
 152                 user_memory += remaining;
 153                 stack[1].count -= remaining;
 154             }
 155         }
 156         pConv->bConverted += bConverted;
 157     }
 158     *out_size = iov_count; /* we only reach this line after the for loop succesfully complete */
 159     *max_data = (pConv->bConverted - initial_bytes_converted);
 160     if( pConv->bConverted == pConv->local_size ) {
 161         pConv->flags |= CONVERTOR_COMPLETED;
 162         return 1;
 163     }
 164     return 0;
 165 }
 166 
 167 /**
 168  * This function handle partial types. Depending on the send operation it might happens
 169  * that we receive only a partial type (always predefined type). In fact the outcome is
 170  * that the unpack has to be done in 2 steps. As there is no way to know if the other
 171  * part of the datatype is already received, we need to use a trick to handle this special
 172  * case. The trick is to fill the missing part with some well known value, unpack the data
 173  * as if it was completely received, and then move into the user memory only the bytes
 174  * that don't match the well known value. This approach work as long as there is no need
 175  * for more than structural changes. They will not work for cases where we will have to
 176  * change the content of the data (as in all conversions that require changing the size
 177  * of the exponent or mantissa).
 178  */
 179 static inline void
 180 opal_unpack_partial_datatype( opal_convertor_t* pConvertor, dt_elem_desc_t* pElem,
 181                               unsigned char* partial_data,
 182                               ptrdiff_t start_position, ptrdiff_t length,
 183                               unsigned char** user_buffer )
 184 {
 185     char unused_byte = 0x7F, saved_data[16];
 186     unsigned char temporary[16], *temporary_buffer = temporary;
 187     unsigned char* user_data = *user_buffer + pElem->elem.disp;
 188     size_t count_desc = 1;
 189     size_t data_length = opal_datatype_basicDatatypes[pElem->elem.common.type]->size;
 190 
 191     DO_DEBUG( opal_output( 0, "unpack partial data start %lu end %lu data_length %lu user %p\n"
 192                            "\tbConverted %lu total_length %lu count %ld\n",
 193                            (unsigned long)start_position, (unsigned long)start_position + length, (unsigned long)data_length, (void*)*user_buffer,
 194                            (unsigned long)pConvertor->bConverted, (unsigned long)pConvertor->local_size, pConvertor->count ); );
 195 
 196     /* Find a byte that is not used in the partial buffer */
 197  find_unused_byte:
 198     for(ptrdiff_t i = 0; i < length; i++ ) {
 199         if( unused_byte == partial_data[i] ) {
 200             unused_byte--;
 201             goto find_unused_byte;
 202         }
 203     }
 204 
 205     /* Copy and fill the rest of the buffer with the unused byte */
 206     memset( temporary, unused_byte, data_length );
 207     MEMCPY( temporary + start_position, partial_data, length );
 208 
 209 #if OPAL_CUDA_SUPPORT
 210     /* In the case where the data is being unpacked from device memory, need to
 211      * use the special host to device memory copy.  Note this code path was only
 212      * seen on large receives of noncontiguous data via buffered sends. */
 213     pConvertor->cbmemcpy(saved_data, user_data, data_length, pConvertor );
 214 #else
 215     /* Save the content of the user memory */
 216     MEMCPY( saved_data, user_data, data_length );
 217 #endif
 218 
 219     /* Then unpack the data into the user memory */
 220     UNPACK_PREDEFINED_DATATYPE( pConvertor, pElem, count_desc,
 221                                 temporary_buffer, *user_buffer, data_length );
 222 
 223     /* reload the length as it is reset by the macro */
 224     data_length = opal_datatype_basicDatatypes[pElem->elem.common.type]->size;
 225 
 226     /* For every occurence of the unused byte move data from the saved
 227      * buffer back into the user memory.
 228      */
 229 #if OPAL_CUDA_SUPPORT
 230     /* Need to copy the modified user_data again so we can see which
 231      * bytes need to be converted back to their original values.  Note
 232      * this code path was only seen on large receives of noncontiguous
 233      * data via buffered sends. */
 234     {
 235         char resaved_data[16];
 236         pConvertor->cbmemcpy(resaved_data, user_data, data_length, pConvertor );
 237         for(size_t i = 0; i < data_length; i++ ) {
 238             if( unused_byte == resaved_data[i] )
 239                 pConvertor->cbmemcpy(&user_data[i], &saved_data[i], 1, pConvertor);
 240         }
 241     }
 242 #else
 243     for(size_t i = 0; i < data_length; i++ ) {
 244         if( unused_byte == user_data[i] )
 245             user_data[i] = saved_data[i];
 246     }
 247 #endif
 248 }
 249 
 250 /* The pack/unpack functions need a cleanup. I have to create a proper interface to access
 251  * all basic functionalities, hence using them as basic blocks for all conversion functions.
 252  *
 253  * But first let's make some global assumptions:
 254  * - a datatype (with the flag DT_DATA set) will have the contiguous flags set if and only if
 255  *   the data is really contiguous (extent equal with size)
 256  * - for the OPAL_DATATYPE_LOOP type the DT_CONTIGUOUS flag set means that the content of the loop is
 257  *   contiguous but with a gap in the begining or at the end.
 258  * - the DT_CONTIGUOUS flag for the type OPAL_DATATYPE_END_LOOP is meaningless.
 259  */
 260 int32_t
 261 opal_generic_simple_unpack_function( opal_convertor_t* pConvertor,
 262                                      struct iovec* iov, uint32_t* out_size,
 263                                      size_t* max_data )
 264 {
 265     dt_stack_t* pStack;                /* pointer to the position on the stack */
 266     uint32_t pos_desc;                 /* actual position in the description of the derived datatype */
 267     size_t count_desc;                 /* the number of items already done in the actual pos_desc */
 268     size_t total_unpacked = 0;         /* total size unpacked this time */
 269     dt_elem_desc_t* description;
 270     dt_elem_desc_t* pElem;
 271     const opal_datatype_t *pData = pConvertor->pDesc;
 272     unsigned char *conv_ptr, *iov_ptr;
 273     size_t iov_len_local;
 274     uint32_t iov_count;
 275 
 276     DO_DEBUG( opal_output( 0, "opal_convertor_generic_simple_unpack( %p, {%p, %lu}, %u )\n",
 277                            (void*)pConvertor, (void*)iov[0].iov_base, (unsigned long)iov[0].iov_len, *out_size ); );
 278 
 279     description = pConvertor->use_desc->desc;
 280 
 281     /* For the first step we have to add both displacement to the source. After in the
 282      * main while loop we will set back the source_base to the correct value. This is
 283      * due to the fact that the convertor can stop in the middle of a data with a count
 284      */
 285     pStack     = pConvertor->pStack + pConvertor->stack_pos;
 286     pos_desc   = pStack->index;
 287     conv_ptr   = pConvertor->pBaseBuf + pStack->disp;
 288     count_desc = pStack->count;
 289     pStack--;
 290     pConvertor->stack_pos--;
 291     pElem = &(description[pos_desc]);
 292 
 293     DO_DEBUG( opal_output( 0, "unpack start pos_desc %d count_desc %" PRIsize_t " disp %ld\n"
 294                            "stack_pos %d pos_desc %d count_desc %" PRIsize_t " disp %ld\n",
 295                            pos_desc, count_desc, (long)(conv_ptr - pConvertor->pBaseBuf),
 296                            pConvertor->stack_pos, pStack->index, pStack->count, (long)(pStack->disp) ); );
 297 
 298     for( iov_count = 0; iov_count < (*out_size); iov_count++ ) {
 299         iov_ptr = (unsigned char *) iov[iov_count].iov_base;
 300         iov_len_local = iov[iov_count].iov_len;
 301         if( 0 != pConvertor->partial_length ) {
 302             size_t element_length = opal_datatype_basicDatatypes[pElem->elem.common.type]->size;
 303             size_t missing_length = element_length - pConvertor->partial_length;
 304 
 305             assert( pElem->elem.common.flags & OPAL_DATATYPE_FLAG_DATA );
 306             COMPUTE_CSUM( iov_ptr, missing_length, pConvertor );
 307             opal_unpack_partial_datatype( pConvertor, pElem,
 308                                           iov_ptr,
 309                                           pConvertor->partial_length, element_length - pConvertor->partial_length,
 310                                           &conv_ptr );
 311             --count_desc;
 312             if( 0 == count_desc ) {
 313                 conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 314                 pos_desc++;  /* advance to the next data */
 315                 UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 316             }
 317             iov_ptr       += missing_length;
 318             iov_len_local -= missing_length;
 319             pConvertor->partial_length = 0;  /* nothing more inside */
 320         }
 321         while( 1 ) {
 322             while( pElem->elem.common.flags & OPAL_DATATYPE_FLAG_DATA ) {
 323                 /* now here we have a basic datatype */
 324                 UNPACK_PREDEFINED_DATATYPE( pConvertor, pElem, count_desc,
 325                                             iov_ptr, conv_ptr, iov_len_local );
 326                 if( 0 == count_desc ) {  /* completed */
 327                     conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 328                     pos_desc++;  /* advance to the next data */
 329                     UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 330                     continue;
 331                 }
 332                 assert( pElem->elem.common.type < OPAL_DATATYPE_MAX_PREDEFINED );
 333                 if( 0 != iov_len_local ) {
 334                     unsigned char* temp = conv_ptr;
 335                     /* We have some partial data here. Let's copy it into the convertor
 336                      * and keep it hot until the next round.
 337                      */
 338                     assert( iov_len_local < opal_datatype_basicDatatypes[pElem->elem.common.type]->size );
 339                     COMPUTE_CSUM( iov_ptr, iov_len_local, pConvertor );
 340 
 341                     opal_unpack_partial_datatype( pConvertor, pElem,
 342                                                   iov_ptr, 0, iov_len_local,
 343                                                   &temp );
 344 
 345                     pConvertor->partial_length = iov_len_local;
 346                     iov_len_local = 0;
 347                 }
 348                 goto complete_loop;
 349             }
 350             if( OPAL_DATATYPE_END_LOOP == pElem->elem.common.type ) { /* end of the current loop */
 351                 DO_DEBUG( opal_output( 0, "unpack end_loop count %" PRIsize_t " stack_pos %d pos_desc %d disp %ld space %lu\n",
 352                                        pStack->count, pConvertor->stack_pos, pos_desc,
 353                                        pStack->disp, (unsigned long)iov_len_local ); );
 354                 if( --(pStack->count) == 0 ) { /* end of loop */
 355                     if( 0 == pConvertor->stack_pos ) {
 356                         /* Do the same thing as when the loop is completed */
 357                         iov[iov_count].iov_len -= iov_len_local;  /* update the amount of valid data */
 358                         total_unpacked += iov[iov_count].iov_len;
 359                         iov_count++;  /* go to the next */
 360                         goto complete_conversion;
 361                     }
 362                     pConvertor->stack_pos--;
 363                     pStack--;
 364                     pos_desc++;
 365                 } else {
 366                     pos_desc = pStack->index + 1;
 367                     if( pStack->index == -1 ) {
 368                         pStack->disp += (pData->ub - pData->lb);
 369                     } else {
 370                         assert( OPAL_DATATYPE_LOOP == description[pStack->index].loop.common.type );
 371                         pStack->disp += description[pStack->index].loop.extent;
 372                     }
 373                 }
 374                 conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 375                 UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 376                 DO_DEBUG( opal_output( 0, "unpack new_loop count %" PRIsize_t " stack_pos %d pos_desc %d disp %ld space %lu\n",
 377                                        pStack->count, pConvertor->stack_pos, pos_desc,
 378                                        pStack->disp, (unsigned long)iov_len_local ); );
 379             }
 380             if( OPAL_DATATYPE_LOOP == pElem->elem.common.type ) {
 381                 ptrdiff_t local_disp = (ptrdiff_t)conv_ptr;
 382                 if( pElem->loop.common.flags & OPAL_DATATYPE_FLAG_CONTIGUOUS ) {
 383                     UNPACK_CONTIGUOUS_LOOP( pConvertor, pElem, count_desc,
 384                                             iov_ptr, conv_ptr, iov_len_local );
 385                     if( 0 == count_desc ) {  /* completed */
 386                         pos_desc += pElem->loop.items + 1;
 387                         goto update_loop_description;
 388                     }
 389                     /* Save the stack with the correct last_count value. */
 390                 }
 391                 local_disp = (ptrdiff_t)conv_ptr - local_disp;
 392                 PUSH_STACK( pStack, pConvertor->stack_pos, pos_desc, OPAL_DATATYPE_LOOP, count_desc,
 393                             pStack->disp + local_disp);
 394                 pos_desc++;
 395             update_loop_description:  /* update the current state */
 396                 conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 397                 UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 398                 DDT_DUMP_STACK( pConvertor->pStack, pConvertor->stack_pos, pElem, "advance loop" );
 399                 continue;
 400             }
 401         }
 402     complete_loop:
 403         iov[iov_count].iov_len -= iov_len_local;  /* update the amount of valid data */
 404         total_unpacked += iov[iov_count].iov_len;
 405     }
 406  complete_conversion:
 407     *max_data = total_unpacked;
 408     pConvertor->bConverted += total_unpacked;  /* update the already converted bytes */
 409     *out_size = iov_count;
 410     if( pConvertor->bConverted == pConvertor->remote_size ) {
 411         pConvertor->flags |= CONVERTOR_COMPLETED;
 412         return 1;
 413     }
 414     /* Save the global position for the next round */
 415     PUSH_STACK( pStack, pConvertor->stack_pos, pos_desc, pElem->elem.common.type, count_desc,
 416                 conv_ptr - pConvertor->pBaseBuf );
 417     DO_DEBUG( opal_output( 0, "unpack save stack stack_pos %d pos_desc %d count_desc %" PRIsize_t " disp %ld\n",
 418                            pConvertor->stack_pos, pStack->index, pStack->count, (long)pStack->disp ); );
 419     return 0;
 420 }
 421 
 422 /*
 423  *  Remember that the first item in the stack (ie. position 0) is the number
 424  * of times the datatype is involved in the operation (ie. the count argument
 425  * in the MPI_ call).
 426  */
 427 /* Convert data from multiple input buffers (as received from the network layer)
 428  * to a contiguous output buffer with a predefined size.
 429  * return OPAL_SUCCESS if everything went OK and if there is still room before the complete
 430  *          conversion of the data (need additional call with others input buffers )
 431  *        1 if everything went fine and the data was completly converted
 432  *       -1 something wrong occurs.
 433  */
 434 int32_t
 435 opal_unpack_general_function( opal_convertor_t* pConvertor,
 436                               struct iovec* iov, uint32_t* out_size,
 437                               size_t* max_data )
 438 {
 439     dt_stack_t* pStack;                /* pointer to the position on the stack */
 440     uint32_t pos_desc;                 /* actual position in the description of the derived datatype */
 441     size_t count_desc;                 /* the number of items already done in the actual pos_desc */
 442     uint16_t type = OPAL_DATATYPE_MAX_PREDEFINED; /* type at current position */
 443     size_t total_unpacked = 0;         /* total size unpacked this time */
 444     dt_elem_desc_t* description;
 445     dt_elem_desc_t* pElem;
 446     const opal_datatype_t *pData = pConvertor->pDesc;
 447     unsigned char *conv_ptr, *iov_ptr;
 448     uint32_t iov_count;
 449     size_t iov_len_local;
 450 
 451     const opal_convertor_master_t* master = pConvertor->master;
 452     ptrdiff_t advance;       /* number of bytes that we should advance the buffer */
 453     size_t rc;
 454 
 455     DO_DEBUG( opal_output( 0, "opal_convertor_general_unpack( %p, {%p, %lu}, %d )\n",
 456                            (void*)pConvertor, (void*)iov[0].iov_base, (unsigned long)iov[0].iov_len, *out_size ); );
 457 
 458     description = pConvertor->use_desc->desc;
 459 
 460     /* For the first step we have to add both displacement to the source. After in the
 461      * main while loop we will set back the source_base to the correct value. This is
 462      * due to the fact that the convertor can stop in the middle of a data with a count
 463      */
 464     pStack     = pConvertor->pStack + pConvertor->stack_pos;
 465     pos_desc   = pStack->index;
 466     conv_ptr   = pConvertor->pBaseBuf + pStack->disp;
 467     count_desc = pStack->count;
 468     pStack--;
 469     pConvertor->stack_pos--;
 470     pElem = &(description[pos_desc]);
 471 
 472     DO_DEBUG( opal_output( 0, "unpack start pos_desc %d count_desc %" PRIsize_t " disp %ld\n"
 473                            "stack_pos %d pos_desc %d count_desc %" PRIsize_t " disp %ld\n",
 474                            pos_desc, count_desc, (long)(conv_ptr - pConvertor->pBaseBuf),
 475                            pConvertor->stack_pos, pStack->index, pStack->count, (long)(pStack->disp) ); );
 476 
 477     for( iov_count = 0; iov_count < (*out_size); iov_count++ ) {
 478         iov_ptr = (unsigned char *) iov[iov_count].iov_base;
 479         iov_len_local = iov[iov_count].iov_len;
 480         assert( 0 == pConvertor->partial_length );
 481         while( 1 ) {
 482             while( pElem->elem.common.flags & OPAL_DATATYPE_FLAG_DATA ) {
 483                 /* now here we have a basic datatype */
 484                 type = description[pos_desc].elem.common.type;
 485                 OPAL_DATATYPE_SAFEGUARD_POINTER( conv_ptr + pElem->elem.disp, pData->size, pConvertor->pBaseBuf,
 486                                                  pData, pConvertor->count );
 487                 DO_DEBUG( opal_output( 0, "unpack (%p, %ld) -> (%p:%ld, %" PRIsize_t ", %ld) type %s\n",
 488                                        (void*)iov_ptr, iov_len_local,
 489                                        (void*)pConvertor->pBaseBuf, conv_ptr + pElem->elem.disp - pConvertor->pBaseBuf,
 490                                        count_desc, description[pos_desc].elem.extent,
 491                                        opal_datatype_basicDatatypes[type]->name ); );
 492                 rc = master->pFunctions[type]( pConvertor, count_desc,
 493                                                iov_ptr, iov_len_local, opal_datatype_basicDatatypes[type]->size,
 494                                                conv_ptr + pElem->elem.disp,
 495                                                (pConvertor->pDesc->ub - pConvertor->pDesc->lb) * pConvertor->count,
 496                                                description[pos_desc].elem.extent, &advance );
 497                 iov_len_local -= advance;  /* decrease the available space in the buffer */
 498                 iov_ptr += advance;        /* increase the pointer to the buffer */
 499                 count_desc -= rc;          /* compute leftovers */
 500                 if( 0 == count_desc ) {  /* completed */
 501                     conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 502                     pos_desc++;  /* advance to the next data */
 503                     UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 504                     if( 0 == iov_len_local ) goto complete_loop;  /* escape if we're done */
 505                     continue;
 506                 }
 507                 conv_ptr += rc * description[pos_desc].elem.extent;
 508                 assert( pElem->elem.common.type < OPAL_DATATYPE_MAX_PREDEFINED );
 509                 assert( 0 == iov_len_local );
 510                 if( 0 != iov_len_local ) {
 511                     unsigned char* temp = conv_ptr;
 512                     /* We have some partial data here. Let's copy it into the convertor
 513                      * and keep it hot until the next round.
 514                      */
 515                     assert( iov_len_local < opal_datatype_basicDatatypes[pElem->elem.common.type]->size );
 516                     COMPUTE_CSUM( iov_ptr, iov_len_local, pConvertor );
 517 
 518                     opal_unpack_partial_datatype( pConvertor, pElem,
 519                                                   iov_ptr, 0, iov_len_local,
 520                                                   &temp );
 521 
 522                     pConvertor->partial_length = iov_len_local;
 523                     iov_len_local = 0;
 524                 }
 525                 goto complete_loop;
 526             }
 527             if( OPAL_DATATYPE_END_LOOP == pElem->elem.common.type ) { /* end of the current loop */
 528                 DO_DEBUG( opal_output( 0, "unpack end_loop count %" PRIsize_t " stack_pos %d pos_desc %d disp %ld space %lu\n",
 529                                        pStack->count, pConvertor->stack_pos, pos_desc,
 530                                        pStack->disp, (unsigned long)iov_len_local ); );
 531                 if( --(pStack->count) == 0 ) { /* end of loop */
 532                     if( 0 == pConvertor->stack_pos ) {
 533                         /* Do the same thing as when the loop is completed */
 534                         iov[iov_count].iov_len -= iov_len_local;  /* update the amount of valid data */
 535                         total_unpacked += iov[iov_count].iov_len;
 536                         iov_count++;  /* go to the next */
 537                         goto complete_conversion;
 538                     }
 539                     pConvertor->stack_pos--;
 540                     pStack--;
 541                     pos_desc++;
 542                 } else {
 543                     pos_desc = pStack->index + 1;
 544                     if( pStack->index == -1 ) {
 545                         pStack->disp += (pData->ub - pData->lb);
 546                     } else {
 547                         assert( OPAL_DATATYPE_LOOP == description[pStack->index].loop.common.type );
 548                         pStack->disp += description[pStack->index].loop.extent;
 549                     }
 550                 }
 551                 conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 552                 UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 553                 DO_DEBUG( opal_output( 0, "unpack new_loop count %" PRIsize_t " stack_pos %d pos_desc %d disp %ld space %lu\n",
 554                                        pStack->count, pConvertor->stack_pos, pos_desc,
 555                                        pStack->disp, (unsigned long)iov_len_local ); );
 556             }
 557             if( OPAL_DATATYPE_LOOP == pElem->elem.common.type ) {
 558                 PUSH_STACK( pStack, pConvertor->stack_pos, pos_desc, OPAL_DATATYPE_LOOP, count_desc,
 559                             pStack->disp );
 560                 pos_desc++;
 561                 conv_ptr = pConvertor->pBaseBuf + pStack->disp;
 562                 UPDATE_INTERNAL_COUNTERS( description, pos_desc, pElem, count_desc );
 563                 DDT_DUMP_STACK( pConvertor->pStack, pConvertor->stack_pos, pElem, "advance loop" );
 564                 continue;
 565             }
 566         }
 567     complete_loop:
 568         iov[iov_count].iov_len -= iov_len_local;  /* update the amount of valid data */
 569         total_unpacked += iov[iov_count].iov_len;
 570     }
 571  complete_conversion:
 572     *max_data = total_unpacked;
 573     pConvertor->bConverted += total_unpacked;  /* update the already converted bytes */
 574     *out_size = iov_count;
 575     if( pConvertor->bConverted == pConvertor->remote_size ) {
 576         pConvertor->flags |= CONVERTOR_COMPLETED;
 577         return 1;
 578     }
 579     /* Save the global position for the next round */
 580     PUSH_STACK( pStack, pConvertor->stack_pos, pos_desc, pElem->elem.common.type, count_desc,
 581                 conv_ptr - pConvertor->pBaseBuf );
 582     DO_DEBUG( opal_output( 0, "unpack save stack stack_pos %d pos_desc %d count_desc %" PRIsize_t" disp %ld\n",
 583                            pConvertor->stack_pos, pStack->index, pStack->count, (long)pStack->disp ); );
 584     return 0;
 585 }