VMPI_stream.c

Go to the documentation of this file.

/* ############################ MALP License ############################## */
/* # Fri Jan 18 14:00:00 CET 2013                                         # */
/* # Copyright or (C) or Copr. Commissariat a l'Energie Atomique          # */
/* #                                                                      # */
/* # This software is governed by the CeCILL-C license under French law   # */
/* # and abiding by the rules of distribution of free software.  You can  # */
/* # use, modify and/ or redistribute the software under the terms of     # */
/* # the CeCILL-C license as circulated by CEA, CNRS and INRIA at the     # */
/* # following URL http://www.cecill.info.                                # */
/* #                                                                      # */
/* # The fact that you are presently reading this means that you have     # */
/* # had knowledge of the CeCILL-C license and that you accept its        # */
/* # terms.                                                               # */
/* #                                                                      # */
/* # Authors:                                                             # */
/* #   - BESNARD Jean-Baptiste jean-baptiste.besnard@cea.fr               # */
/* #                                                                      # */
/* ######################################################################## */
#include "VMPI_stream.h"

#include <string.h>
#include <stdio.h>
#include <stdlib.h>


int VMPI_Stream_init( VMPI_Stream *st, size_t block_size, VMPI_Stream_load_balance lb)
{
        int i;
        memset( st, 0, sizeof( VMPI_Stream ));
        st->lb = lb;
        st->block_size = block_size;

        st->r_tmp_buffers = NULL;
        st->r_req = NULL;
        st->r_stat = NULL;
        st->r_current_buff = NULL;

        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC ; i++ )
        {
                /* Fill in requests */
                st->w_req[i] = MPI_REQUEST_NULL;


                st->w_tmp_buffers[i] = malloc( block_size );

                if( !st->w_tmp_buffers[i] )
                        return 2;
        }
        
    st->lock = 0;

        return 0;
}




int VMPI_Stream_push_stream( VMPI_Stream *st, int dest, int remote_tag, int local_tag, VMPI_Stream_mode mode )
{
        struct _VMPI_Stream new_stream;
        int i, j;
        
        new_stream.target = dest;
        new_stream.remote_tag = remote_tag;
        new_stream.local_tag = local_tag;
        new_stream.state = 1;
        
        /* Make sure we dont push the same rank twice */
        switch( mode )
        {
                case VMPI_STREAM_R:
                case VMPI_STREAM_RW:
                for( i = 0; i < st->in_count; i++ )
                {
                        /* Check for this stream in incoming */
                        if( st->in[i].target == dest )
                                return 0;

                        if( mode == VMPI_STREAM_RW )
                        {
                                case VMPI_STREAM_W:
                                for( j = 0; j < st->out_count; j++ )
                                {
                                        if( st->out[i].target == dest )
                                                return 0;
                                }
                                
                                if( mode == VMPI_STREAM_W )
                                        break;
                        }
                }
                break;
        }
        
        /* Push a given rank */
        switch( mode )
        {
                case VMPI_STREAM_RW:
                case VMPI_STREAM_R:
                        /* We want to make sure that each stream is constantly
                         * in the read request list to avoid deadlocks
                         * as a consequences readers have to book more memory
                         * 
                         * To do so each stream has VMPI_STREAM_MAX_ASYNC requests
                         * these could have been the same on the writer side but
                         * as the reader relies on these data for computing we
                         * want it to get as much data as possible whereas
                         * the writer wants to get rid of his data as fast
                         * as possible with low memory footprint */

                        st->in_count++;
                        
                        /* Allocate room for streams */
                        st->in = realloc( st->in, st->in_count * sizeof( struct _VMPI_Stream ) );
                        
                        if( !st->in ) 
                                return 2;

                        memcpy(&st->in[ st->in_count - 1 ], &new_stream, sizeof( struct _VMPI_Stream ) );
                        
                        /* Allocate requests */
                        st->r_req = realloc( st->r_req, st->in_count * sizeof( MPI_Request ) * VMPI_STREAM_MAX_ASYNC );

                        if( !st->r_req ) 
                                return 2;

                        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC ; i++ )
                        {
                                st->r_req[ (st->in_count*VMPI_STREAM_MAX_ASYNC) - VMPI_STREAM_MAX_ASYNC + i ] = MPI_REQUEST_NULL;
                        }
                        
                        /* Allocate statuses */
                        st->r_stat = realloc( st->r_stat, st->in_count * sizeof( MPI_Status ) * VMPI_STREAM_MAX_ASYNC );
                        
                        if( !st->r_stat ) 
                                return 2;
                        
                        /* Allocate buffer counter */
                        st->r_current_buff = realloc( st->r_current_buff, st->in_count * sizeof( int ) * VMPI_STREAM_MAX_ASYNC );
                        
                        if( !st->r_current_buff ) 
                                return 2;
                        
                        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC ; i++ )
                        {
                                st->r_current_buff[ (st->in_count*VMPI_STREAM_MAX_ASYNC) - VMPI_STREAM_MAX_ASYNC + i ] = -1;
                        }
                        
                        /* Allocate buffers */
                        st->r_tmp_buffers = realloc( st->r_tmp_buffers, st->in_count * sizeof( char * ) * VMPI_STREAM_MAX_ASYNC );
                        
                        if( !st->r_tmp_buffers ) 
                                return 2;
                        
                        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC ; i++ )
                        {
                                st->r_tmp_buffers[ (st->in_count*VMPI_STREAM_MAX_ASYNC) - VMPI_STREAM_MAX_ASYNC + i ] = malloc( st->block_size );

                                if( !st->r_tmp_buffers[ (st->in_count*VMPI_STREAM_MAX_ASYNC) - VMPI_STREAM_MAX_ASYNC + i ] ) 
                                        return 2;
                        }

                        
                if( mode == VMPI_STREAM_R )
                        break;
                /* otherwise no break */
                case VMPI_STREAM_W:
                        st->out_count++;
                        st->out = realloc( st->out, st->out_count * sizeof( struct _VMPI_Stream ) );
                        
                        if( !st->out ) 
                                return 2;
                        
                        memcpy(&st->out[ st->out_count - 1 ], &new_stream, sizeof( struct _VMPI_Stream ) );
                break;
                default :
                        printf("No such mode \n");
                        abort();
        }

        return 0;
}




int VMPI_Stream_join(VMPI_Stream * dest, VMPI_Stream * src)
{
        int i;
        for( i = 0 ; i < src->in_count ; i++ )
        {
        if (VMPI_Stream_push_stream( dest, src->in[i].target , src->in[i].remote_tag, src->in[i].local_tag, VMPI_STREAM_R) != 0)
            return 2;
        }

        for( i = 0 ; i < src->out_count ; i++ )
        {
        if(VMPI_Stream_push_stream( dest, src->out[i].target , src->out[i].remote_tag, src->out[i].local_tag, VMPI_STREAM_W) != 0)
            return 2;
        }
    return 0;
        /* Todo clean release of src */
}


struct TMP_VMPI_Stream_desc
{
        int rank;               
        int tag;                
        VMPI_Stream_mode mode;  
};


int VMPI_Stream_open( VMPI_Stream *st, int dest, char *smode )
{
        struct TMP_VMPI_Stream_desc tmp_desc;
        struct TMP_VMPI_Stream_desc remote_desc;
        int rank, local_tag;
        VMPI_Stream_mode mode;
        MPI_Status status;
        
        /* Check mode */
        {
                if( !strcmp( smode, "r") )
                {
                        mode = VMPI_STREAM_R;
                }else if( !strcmp( smode, "w") )
                {
                        mode = VMPI_STREAM_W;
                }
                else if( !strcmp( smode, "rw") || !strcmp( smode, "wr") )
                {
                        mode = VMPI_STREAM_RW;
                }
                else
                {
                        printf("No such mode '%s' in %s\n", smode, __FUNCTION__ );
                        abort();
                }
        }

        /* Fill in mode desc */
        
        PMPI_Comm_rank(VMPI_Get_vmpi_comm(), &rank);
        
        /* We allocate a tag localy to
         * identify this stream */
        local_tag = VMPI_get_new_tag();

        tmp_desc.tag = local_tag;
        tmp_desc.rank = rank;
        tmp_desc.mode = mode;
        

        /* Exchange mode descs using the VMPI_NG_ARBITRARY_VALUE + 1 tag*/
        if( PMPI_Sendrecv((void *)&tmp_desc, sizeof(struct TMP_VMPI_Stream_desc), MPI_CHAR, dest, VMPI_NG_ARBITRARY_VALUE + 2,
                                 (void *)&remote_desc, sizeof(struct TMP_VMPI_Stream_desc), MPI_CHAR, dest, VMPI_NG_ARBITRARY_VALUE + 2,
                VMPI_Get_vmpi_comm(), &status) != MPI_SUCCESS )
    {         
        return 1;
    }

    MALP_Spinlock_lock(&st->lock);
    
    /* Check validity of remote desc */

    /* Check rank just in case */
    if( remote_desc.rank != dest )
    {
                printf("Error exchanging MPI_Stream informations\n");
                abort();
        }
    
    /* Check modes */
    if( (mode == VMPI_STREAM_R) && (remote_desc.mode == VMPI_STREAM_R) )
    {
                printf("Error opening a stream with two read endpoints is not possible\n");
                abort();
        }
    
    if( (mode == VMPI_STREAM_W) && (remote_desc.mode == VMPI_STREAM_W) )
    {
                printf("Error opening a stream with two write endpoints is not possible\n");
                abort();
        }

        /* If check passed store the new stream info */
        VMPI_Stream_push_stream( st, dest, remote_desc.tag, local_tag, mode );

    MALP_Spinlock_unlock(&st->lock);

        return 0;
}


int VMPI_Stream_open_map( VMPI_Stream *st, VMPI_Map *map, char *mode )
{
        int i = 0;
        
        for( i = 0 ; i < map->count ; i++ )
        {
                VMPI_Stream_open( st, map->ranks[i], mode );
        }

        return 0;
}


int VMPI_Stream_close( VMPI_Stream * st )
{
        int i, ret;
        int dummy;
        char *buff = malloc( st->block_size );
        VMPI_Partition_desc *desc;
        
        desc = VMPI_Get_desc();
        
        if( !buff )
        {
                return 2;
        }

        /* Ensure everything is written */
        if( st->w_req_count )
                PMPI_Waitall(st->w_req_count, st->w_req, st->w_stat);

        /* Broadcast close to outgoing targets so that read return EOF */
        for( i = 0 ; i < st->out_count ; i++ )
        {
                if( PMPI_Send((void *)&dummy, 0, MPI_INT, st->out[i].target, st->out[i].remote_tag, VMPI_Get_vmpi_comm()) != MPI_SUCCESS )
                        return 1;
        }
        
                
        /* Wait for incoming close data on inbounds streams */
        if( st->in_count )
        {
                do
                {
                        ret = VMPI_Stream_read( st, buff, 1, VMPI_STREAM_BLOCK );
                        
                        if( ret )
                        {
                                printf("Warning : Unprocessed datas in partition %d (%s)\n", desc?desc->id:-1, desc?desc->name:"UNKNOWN");
                                printf("Make sure calls to close are done in both partitions\n");
                                abort();
                        }
                        
                }while( ret );
        }
        
        /* Now that we are done free everything */
        free( st->in );
        st->in = NULL;
        st->in_count = 0;
        free( st->out );
        st->out_count = 0;
        free( st->r_stat );
        st->r_stat = NULL;
        free( st->r_req );
        st->r_req = NULL;
        
        /* Release tmp buffs */
        for( i = 0 ; i < st->in_count * VMPI_STREAM_MAX_ASYNC ; i++ )
        {
                free( st->r_tmp_buffers[i] );
                st->r_tmp_buffers[i] = NULL;
        }

        free( st->r_tmp_buffers );
        st->r_tmp_buffers = NULL;
        
        free( st->r_current_buff );
        st->r_current_buff = NULL;

        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC; i++ )
        {
                free( st->w_tmp_buffers[i] );
                st->w_tmp_buffers[i] = NULL;
        }

        memset( st, 0, sizeof( VMPI_Stream ) );


        return 0;
}

int VMPI_Stream_write( VMPI_Stream * st, char *buff, unsigned int block_count )
{
        size_t len;
        char *dat_ptr;
        char *tmp_buff;
        MPI_Request *req;
        int index, a_value;
        MPI_Status status;
        int written_size;
        struct _VMPI_Stream *dest;
        
        len = block_count * st->block_size;
        
        
        if( len == 0 )
                return 0;

        written_size = 0;
        dat_ptr = buff;

    MALP_Spinlock_lock(&st->lock);

        while( 1 )
        {
                /* Did we write everything ? */
                if( len == 0 )
                {
                        break;
                }

                /* Find a free request */
                if( st->w_req_count < VMPI_STREAM_MAX_ASYNC )
                {
                        /* We can allocate a new request */
                        req = &st->w_req[st->w_req_count];
                        tmp_buff = st->w_tmp_buffers[st->w_req_count];
                        st->w_req_count ++;
                }
                else
                {
                        /* We can use an old one */
                        PMPI_Waitany(st->w_req_count, st->w_req, &index, &status);

                        if( index != MPI_UNDEFINED )
                        {
                                req = &st->w_req[index];
                                tmp_buff = st->w_tmp_buffers[index];
                        }
                        else
                        {
                                continue;
                        }
                }

                /* We now have a free request */
                
                /* Pick a dest according to policy */
                switch( st->lb )
                {
                        case VMPI_STREAM_BALANCE_NONE :
                                dest = &st->out[0];
                                break;
                        case VMPI_STREAM_BALANCE_RANDOM :
                                a_value = rand()%st->out_count;
                                dest = &st->out[a_value];
                                break;
                        case VMPI_STREAM_BALANCE_ROUND_ROBIN :
                                st->current_out = (st->current_out + 1)%st->out_count;
                                dest = &st->out[st->current_out];
                                break;
                }
                

                /* Store in tmp Buff to relax user buffer */
                memcpy( tmp_buff, dat_ptr, st->block_size );
                

                /* Now lets send datas */               
                /* Do we have enough datas to fill a block_size ? */

                //printf("WRITE send to <%d;%d>\n", dest->target, dest->remote_tag );
                if( PMPI_Isend((void *)tmp_buff, st->block_size, MPI_CHAR, dest->target, dest->remote_tag, VMPI_Get_vmpi_comm(), req) != MPI_SUCCESS )
        {
                MALP_Spinlock_unlock(&st->lock);
                                return -1;
        }

                dat_ptr += st->block_size;
                len -= st->block_size;
                written_size += st->block_size;
        }

    MALP_Spinlock_unlock(&st->lock);

        return written_size;
}


int VMPI_Stream_read( VMPI_Stream * st, char *buff, unsigned int block_count, VMPI_Stream_blocking block )
{
        size_t len;
        int read_size, i, dat_size, has_data;
        struct _VMPI_Stream *src;
        char *dat_ptr;
        int src_id, target_index, flag;

        len = block_count * st->block_size;
        dat_ptr = buff;
        read_size = 0;

        if( len == 0 )
                return VMPI_SUCCESS;

    MALP_Spinlock_lock(&st->lock);
        
        /* Enter Read LOOP */
        read_size = 0;
        while( 1 )
        {
                if( len == 0 )
        {
            MALP_Spinlock_unlock(&st->lock);
                        return read_size;
        }

                /* Check if all streams are closed */
                has_data = 0;
                for( i = 0 ; i < st->in_count; i++ )
                {
                        has_data |= st->in[i].state;
                }

                /* All streams are closed */
                if( !has_data )
                {
            MALP_Spinlock_unlock(&st->lock);
                        return 0;
                }

                src_id = 0;
                has_data = 0;

                /* Pick an open stream */
                while( !has_data )
                {
                        /* Pick a source to process according to policy */
                        switch( st->lb )
                        {
                                case VMPI_STREAM_BALANCE_NONE :
                                        src_id = 0;

                                        /* Is the only stream from which we can read closed */
                                        if( st->in[src_id].state == 0 )
                                        {
                        MALP_Spinlock_unlock(&st->lock);
                                                return 0;
                                        }
                                        
                                        break;
                                case VMPI_STREAM_BALANCE_RANDOM :
                                        src_id = rand()%st->in_count;
                                        break;
                                case VMPI_STREAM_BALANCE_ROUND_ROBIN :
                                        st->current_in = (st->current_in + 1)%st->in_count;
                                        src_id = st->current_in;
                                        break;
                        }
                        
                        has_data = st->in[src_id].state;
                }

                src = &st->in[ src_id ];
                
                if( st->r_current_buff[ src_id ] == -1 )
                {
                        /* There is no receive in this buffer pool */
                        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC; i++ )
                        {
                                //printf("READ INIT Receive from <%d;%d>\n", src->target,  src->local_tag );
                                if( PMPI_Irecv((void *)st->r_tmp_buffers[ src_id * VMPI_STREAM_MAX_ASYNC + i ] , 
                                                                st->block_size, MPI_CHAR, src->target, src->local_tag, VMPI_Get_vmpi_comm(), &st->r_req[ src_id * VMPI_STREAM_MAX_ASYNC + i ]) != MPI_SUCCESS )
                {
                    MALP_Spinlock_unlock(&st->lock);
                                        return VMPI_ERROR;
                }
                        }
                        
                        /* Current buff is now 0 */
                        st->r_current_buff[src_id] = 0;
                }
                
                /* Read */
                
                target_index = src_id * VMPI_STREAM_MAX_ASYNC + st->r_current_buff[ src_id ];

                if( block == VMPI_STREAM_NONBLOCK )
                {
                        flag = 0;
                        PMPI_Test(  &st->r_req[ target_index ], &flag, &st->r_stat[ target_index ]);
                        
                        if( !flag )
                        {
                                /* Operation not completed */
                MALP_Spinlock_unlock(&st->lock);
                                return VMPI_EAGAIN;
                        }
                }
                else
                {
                        PMPI_Wait( &st->r_req[ target_index ], &st->r_stat[ target_index ]);
                }
                
                /* Get incoming size */
                PMPI_Get_count( &st->r_stat[ target_index ], MPI_CHAR, &dat_size );
                
                /* We received EOF */
                if( dat_size == 0 )
                {
                        /* Flag stream closed */
                        st->in[src_id].state = 0;

                        /* Cancel all other receives */
                        for( i = 0 ; i < VMPI_STREAM_MAX_ASYNC ; i++ )
                        {
                                 if( st->r_current_buff[ src_id ] != i )
                                 {
                                        target_index = src_id * VMPI_STREAM_MAX_ASYNC + i;
                                        PMPI_Cancel( &st->r_req[ target_index ] );
                                 }
                                
                        }
                        
                        continue;
                }
                
                /* We have data */
                memcpy( dat_ptr, (void *)st->r_tmp_buffers[ target_index ], st->block_size );
        
                /* Repost receive */
                //printf("READ Receive from <%d;%d>\n", src->target,  src->local_tag );
                if( PMPI_Irecv((void *)st->r_tmp_buffers[target_index] , 
                                                                st->block_size, MPI_CHAR, src->target, src->local_tag, VMPI_Get_vmpi_comm(), &st->r_req[ target_index ]) != MPI_SUCCESS )
        {
                MALP_Spinlock_unlock(&st->lock);
                                return VMPI_ERROR;
        }

                /* Switch to next buff */
                st->r_current_buff[ src_id ] = (st->r_current_buff[ src_id ] + 1)%VMPI_STREAM_MAX_ASYNC;
                
                len -= st->block_size;
                read_size += st->block_size;
                dat_ptr += st->block_size;
        }

    MALP_Spinlock_unlock(&st->lock);

        return 0;
}

int VMPI_Stream_test_read( VMPI_Stream * st )
{
        int src;
        int ret = 0;
        int target_index;

        switch( st->lb )
        {
                case VMPI_STREAM_BALANCE_NONE :
                        src = 0;
                        break;
                case VMPI_STREAM_BALANCE_RANDOM :
                        /* As we cannot know the target stream we return ready for read */
                        return 1;
                case VMPI_STREAM_BALANCE_ROUND_ROBIN :
                        src = (st->current_in + 1)%st->in_count;
                        break;
        }
        
        target_index = src * VMPI_STREAM_MAX_ASYNC + st->r_current_buff[ src ];
        
        if( st->r_req )
                if( st->r_req[target_index] != MPI_REQUEST_NULL )
                        PMPI_Test(&st->r_req[target_index], &ret, MPI_STATUS_IGNORE);

        return ret;
}

int VMPI_Stream_test_write( VMPI_Stream * st )
{
        int dest;

        int ret = 0;
        switch( st->lb )
        {
                case VMPI_STREAM_BALANCE_NONE :
                        dest = 0;
                        break;
                case VMPI_STREAM_BALANCE_RANDOM :
                        /* As we cannot know the target stream we return ready for write */
                        return 1;
                case VMPI_STREAM_BALANCE_ROUND_ROBIN :
                        dest = (st->current_out + 1)%st->out_count;
                        break;
        }
        
        if( st->w_req )
                if( st->w_req[dest] != MPI_REQUEST_NULL )
                        PMPI_Test(&st->w_req[dest], &ret, MPI_STATUS_IGNORE);
        
        return ret;
}