datatype_unpack.c

MPI stands for the Message Passing Interface. Written by the MPI Forum (a large committee comprising

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*
 * Copyright (c) 2004-2006 The Trustees of Indiana University and Indiana
 *                         University Research and Technology
 *                         Corporation.  All rights reserved.
 * Copyright (c) 2004-2006 The University of Tennessee and The University
 *                         of Tennessee Research Foundation.  All rights
 *                         reserved.
 * Copyright (c) 2004-2006 High Performance Computing Center Stuttgart,
 *                         University of Stuttgart.  All rights reserved.
 * Copyright (c) 2004-2006 The Regents of the University of California.
 *                         All rights reserved.
 * $COPYRIGHT$
 *
 * Additional copyrights may follow
 *
 * $HEADER$
 */

#include "ompi_config.h"
#include "ompi/datatype/convertor_internal.h"
#include "ompi/datatype/datatype_internal.h"

#if OMPI_ENABLE_DEBUG
extern int ompi_unpack_debug;
#define DO_DEBUG(INST)  if( ompi_unpack_debug ) { INST }
#else
#define DO_DEBUG(INST)
#endif  /* OMPI_ENABLE_DEBUG */

#include "ompi/datatype/datatype_checksum.h"
#include "ompi/datatype/datatype_unpack.h"
#include "ompi/datatype/datatype_prototypes.h"

#if defined(CHECKSUM)
#define ompi_unpack_general_function            ompi_unpack_general_checksum
#define ompi_unpack_homogeneous_contig_function ompi_unpack_homogeneous_contig_checksum
#define ompi_generic_simple_unpack_function     ompi_generic_simple_unpack_checksum
#else
#define ompi_unpack_general_function            ompi_unpack_general
#define ompi_unpack_homogeneous_contig_function ompi_unpack_homogeneous_contig
#define ompi_generic_simple_unpack_function     ompi_generic_simple_unpack
#endif  /* defined(CHECKSUM) */

/*
 *  Remember that the first item in the stack (ie. position 0) is the number
 * of times the datatype is involved in the operation (ie. the count argument
 * in the MPI_ call).
 */
/* Convert data from multiple input buffers (as received from the network layer)
 * to a contiguous output buffer with a predefined size.
 * return OMPI_SUCCESS if everything went OK and if there is still room before the complete
 *          conversion of the data (need additional call with others input buffers )
 *        1 if everything went fine and the data was completly converted
 *       -1 something wrong occurs.
 */
int32_t
ompi_unpack_general_function( ompi_convertor_t* pConvertor,
                              struct iovec* iov,
                              uint32_t* out_size,
                              size_t* max_data )
{
    dt_stack_t* pStack;      /* pointer to the position on the stack */
    uint32_t pos_desc;       /* actual position in the description of the derived datatype */
    int32_t count_desc;      /* the number of items already done in the actual pos_desc */
    int type = DT_CHAR;      /* type at current position */
    ptrdiff_t advance;       /* number of bytes that we should advance the buffer */
    ptrdiff_t disp_desc = 0; /* compute displacement for truncated data */
    size_t bConverted = 0;   /* number of bytes converted this time */
    const ompi_convertor_master_t* master = pConvertor->master;
    dt_elem_desc_t* description;
    ptrdiff_t extent = pConvertor->pDesc->ub - pConvertor->pDesc->lb;
    size_t oCount = extent * pConvertor->count;
    size_t iCount, total_bytes_converted = 0;
    char* pInput;
    int32_t rc;
    uint32_t iov_count;

    /* For the general case always use the user data description */
    description = pConvertor->use_desc->desc;

    pStack = pConvertor->pStack + pConvertor->stack_pos;
    pos_desc   = pStack->index;
    count_desc = (int32_t)pStack->count;
    disp_desc  = pStack->disp;
    pStack--;
    pConvertor->stack_pos--;

    DDT_DUMP_STACK( pConvertor->pStack, pConvertor->stack_pos, description, "starting" );

    for( iov_count = 0; iov_count < (*out_size); iov_count++ ) {
        bConverted = 0;
        pInput = iov[iov_count].iov_base;
        iCount = iov[iov_count].iov_len;
        while( 1 ) {
            if( DT_END_LOOP == description[pos_desc].elem.common.type ) { /* end of the current loop */
                if( --(pStack->count) == 0 ) { /* end of loop */
                    if( pConvertor->stack_pos == 0 ) {
                        goto save_and_return;  /* completed */
                    }
                    pConvertor->stack_pos--;
                    pStack--;
                    pos_desc++;
                } else {
                    pos_desc = pStack->index + 1;
                    if( pStack->index == -1 ) {
                        pStack->disp += extent;
                    } else {
                        assert( DT_LOOP == description[pStack->index].elem.common.type );
                        pStack->disp += description[pStack->index].loop.extent;
                    }
                }
                count_desc = description[pos_desc].elem.count;
                disp_desc = description[pos_desc].elem.disp;
            }
            if( DT_LOOP == description[pos_desc].elem.common.type ) {
                do {
                    PUSH_STACK( pStack, pConvertor->stack_pos,
                                pos_desc, DT_LOOP, description[pos_desc].loop.loops, pStack->disp );
                    pos_desc++;
                } while( DT_LOOP == description[pos_desc].loop.common.type ); /* let's start another loop */
                DDT_DUMP_STACK( pConvertor->pStack, pConvertor->stack_pos, description, "advance loops" );
                /* update the current state */
                count_desc = description[pos_desc].elem.count;
                disp_desc = description[pos_desc].elem.disp;
            }
            while( description[pos_desc].elem.common.flags & DT_FLAG_DATA ) {
                /* now here we have a basic datatype */
                type = description[pos_desc].elem.common.type;
                rc = master->pFunctions[type]( pConvertor, count_desc,
                                               pInput, iCount, ompi_ddt_basicDatatypes[type]->size,
                                               pConvertor->pBaseBuf + pStack->disp + disp_desc,
                                               oCount, description[pos_desc].elem.extent, &advance );
                iCount -= advance;      /* decrease the available space in the buffer */
                pInput += advance;      /* increase the pointer to the buffer */
                bConverted += advance;
                if( rc != count_desc ) {
                    /* not all data has been converted. Keep the state */
                    count_desc -= rc;
                    disp_desc += rc * description[pos_desc].elem.extent;
                    if( iCount != 0 )
                        printf( "unpack there is still room in the input buffer %ld bytes\n", (long)iCount );
                    goto save_and_return;
                }
                pos_desc++;  /* advance to the next data */
                count_desc = description[pos_desc].elem.count;
                disp_desc = description[pos_desc].elem.disp;
                if( iCount == 0 )
                    goto save_and_return;  /* break if there is no more data in the buffer */
            }
        }
    save_and_return:
        pConvertor->bConverted += bConverted;  /* update the # of bytes already converted */
        iov[iov_count].iov_len = bConverted;           /* update the iovec length */
        total_bytes_converted += bConverted;
    }
    *max_data = total_bytes_converted;
    /* out of the loop: we have complete the data conversion or no more space
     * in the buffer.
     */
    if( pConvertor->remote_size == pConvertor->bConverted ) {
        pConvertor->flags |= CONVERTOR_COMPLETED;
        return 1;  /* I'm done */
    }

    /* I complete an element, next step I should go to the next one */
    PUSH_STACK( pStack, pConvertor->stack_pos, pos_desc, type,
                count_desc, disp_desc );

    return 0;
}

/**
 * This function will be used to unpack all datatypes that have the contiguous flag set.
 * Several types of datatypes match this criterion, not only the contiguous one, but
 * the ones that have gaps in the beginning and/or at the end but where the data to
 * be unpacked is contiguous. However, this function only work for homogeneous cases
 * and the datatype that are contiguous and where the extent is equal to the size are
 * taken in account directly in the ompi_convertor_unpack function (in convertor.c) for
 * the homogeneous case.
 */
int32_t
ompi_unpack_homogeneous_contig_function( ompi_convertor_t* pConv,
                                         struct iovec* iov,
                                         uint32_t* out_size,
                                         size_t* max_data )
{
    const ompi_datatype_t *pData = pConv->pDesc;
    char *user_memory, *packed_buffer;
    uint32_t iov_count, i;
    size_t bConverted, remaining, length, initial_bytes_converted = pConv->bConverted;
    dt_stack_t* stack = pConv->pStack;
    ptrdiff_t extent = pData->ub - pData->lb;
    ptrdiff_t initial_displ = pConv->use_desc->desc[pConv->use_desc->used].end_loop.first_elem_disp;

    DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( pBaseBuf %p, iov_count %d )\n",
                           pConv->pBaseBuf, *out_size ); );
    for( iov_count = 0; iov_count < (*out_size); iov_count++ ) {
        packed_buffer = (char*)iov[iov_count].iov_base;
        remaining = pConv->local_size - pConv->bConverted;
        if( remaining > (uint32_t)iov[iov_count].iov_len )
            remaining = iov[iov_count].iov_len;
        bConverted = remaining; /* how much will get unpacked this time */
        user_memory = pConv->pBaseBuf + initial_displ;

        if( (ptrdiff_t)pData->size == extent ) {
            user_memory += pConv->bConverted;
            DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( user_memory %p, packed_buffer %p length %lu\n",
                                   user_memory, packed_buffer, (unsigned long)remaining ); );

            /* contiguous data or basic datatype with count */
            OMPI_DDT_SAFEGUARD_POINTER( user_memory, remaining,
                                        pConv->pBaseBuf, pData, pConv->count );
            DO_DEBUG( opal_output( 0, "1. unpack contig dest %p src %p length %lu\n",
                                   user_memory, packed_buffer, (unsigned long)remaining ); );
            MEMCPY_CSUM( user_memory, packed_buffer, remaining, pConv );
        } else {
            user_memory += stack[0].disp + stack[1].disp;

            DO_DEBUG( opal_output( 0, "unpack_homogeneous_contig( user_memory %p, packed_buffer %p length %lu\n",
                                   user_memory, packed_buffer, (unsigned long)remaining ); );

            length = pConv->bConverted / pData->size;  /* already done */
            length = pConv->bConverted - length * pData->size;  /* how much of the last data we convert */

            /* complete the last copy */
            if( length != 0 ) {
                length = pData->size - length;
                if( length <= remaining ) {
                    OMPI_DDT_SAFEGUARD_POINTER( user_memory, length, pConv->pBaseBuf,
                                                pData, pConv->count );
                    DO_DEBUG( opal_output( 0, "2. unpack dest %p src %p length %lu\n",
                                           user_memory, packed_buffer, (unsigned long)length ); );
                    MEMCPY_CSUM( user_memory, packed_buffer, length, pConv );
                    packed_buffer += length;
                    user_memory   += (extent - (pData->size - length));
                    remaining     -= length;
                }
            }
            for( i = 0; pData->size <= remaining; i++ ) {
                OMPI_DDT_SAFEGUARD_POINTER( user_memory, pData->size, pConv->pBaseBuf,
                                            pData, pConv->count );
                DO_DEBUG( opal_output( 0, "3. unpack dest %p src %p length %lu\n",
                                       user_memory, packed_buffer, (unsigned long)pData->size ); );
                MEMCPY_CSUM( user_memory, packed_buffer, pData->size, pConv );
                packed_buffer += pData->size;
                user_memory   += extent;
                remaining     -= pData->size;
            }
            stack[0].disp = (intptr_t)user_memory - (intptr_t)pConv->pBaseBuf - initial_displ;
            stack[1].disp = remaining;
            /* copy the last bits */
            if( remaining != 0 ) {
                OMPI_DDT_SAFEGUARD_POINTER( user_memory, remaining, pConv->pBaseBuf,