clog_merger.c

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

/*
   (C) 2001 by Argonne National Laboratory.
       See COPYRIGHT in top-level directory.
*/
#include "mpe_logging_conf.h"

#if defined( STDC_HEADERS ) || defined( HAVE_STDIO_H )
#include <stdio.h>
#endif
#if defined( STDC_HEADERS ) || defined( HAVE_STDLIB_H )
#include <stdlib.h>
#endif
#if defined( STDC_HEADERS ) || defined( HAVE_STRING_H )
#include <string.h>
#endif
#if defined( HAVE_FCNTL_H )
#include <fcntl.h>
#endif
#if defined( HAVE_UNISTD_H )
#include <unistd.h>
#endif
#ifdef HAVE_SYS_TYPES_H
#include <sys/types.h>
#endif
#ifdef HAVE_IO_H
#include <io.h>
#endif

#include "clog_const.h"
#include "clog_merger.h"
#include "clog_util.h"

#if !defined( CLOG_NOMPI )
#include "mpi.h"
#endif

#define CLOG_RANK_NULL   -1

static int clog_merger_minblocksize;

CLOG_Merger_t* CLOG_Merger_create( unsigned int block_size )
{
    CLOG_Merger_t  *merger;

    merger = (CLOG_Merger_t *) MALLOC( sizeof(CLOG_Merger_t) );
    if ( merger == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Merger_create() - \n"
                         "\t""MALLOC() fails for CLOG_Merger_t!\n" );
        fflush( stderr );
        return NULL;
    }


    merger->left_blk  = (CLOG_BlockData_t *)
                        CLOG_BlockData_create( block_size );
    if ( merger->left_blk == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Merger_create() - \n"
                         "\t""CLOG_BlockData_create(%d) fails for left_blk!",
                         block_size );
        fflush( stderr );
        return NULL;
    }

    merger->right_blk  = (CLOG_BlockData_t *)
                         CLOG_BlockData_create( block_size );
    if ( merger->right_blk == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Merger_create() - \n"
                         "\t""CLOG_BlockData_create(%d) fails for right_blk!",
                         block_size );
        fflush( stderr );
        return NULL;
    }

    merger->sorted_blk  = (CLOG_BlockData_t *)
                          CLOG_BlockData_create( block_size );
    if ( merger->sorted_blk == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Merger_create() - \n"
                         "\t""CLOG_BlockData_create(%d) fails for sorted_blk!",
                         block_size );
        fflush( stderr );
        return NULL;
    }

    merger->block_size        = block_size;
    merger->num_active_blks   = 0;
    merger->world_size        = 1;
    merger->local_world_rank  = 0;
    merger->left_world_rank   = 0;
    merger->right_world_rank  = 0;
    merger->parent_world_rank = 0;

    merger->is_big_endian    = CLOG_BOOL_NULL;
    strncpy( merger->out_filename, "mpe_trace."CLOG_FILE_TYPE,
             CLOG_PATH_STRLEN );
    merger->out_fd           = -1;

    return merger;
}

void CLOG_Merger_free( CLOG_Merger_t **merger_handle )
{
    CLOG_Merger_t  *merger;
    merger = *merger_handle;
    if ( merger != NULL ) {
        CLOG_BlockData_free( &(merger->sorted_blk) );
        CLOG_BlockData_free( &(merger->left_blk) );
        CLOG_BlockData_free( &(merger->right_blk) );
        FREE( merger );
    }
    *merger_handle = NULL;
}

/*@
    CLOG_Merger_set_neighbor_ranks - locally determine parent and children
                                     in a binary tree

Input parameters

+  merger->local_world_rank  - calling process''s id
-  merger->world_size        - total number of processes in tree

Output parameters

+  merger->parent_world_rank - parent of local process
                             (or CLOG_RANK_NULL if root)
.  merger->left_world_rank   - left child of local process in binary tree
                             (or CLOG_RANK_NULL if none)
-  merger->right_world_rank  - right child of local process in binary tree
                             (or CLOG_RANK_NULL if none)

@*/
void CLOG_Merger_set_neighbor_ranks( CLOG_Merger_t *merger )
{
    if ( merger->local_world_rank == 0 )
        merger->parent_world_rank = CLOG_RANK_NULL;
    else
        merger->parent_world_rank = (merger->local_world_rank - 1) / 2;

    merger->left_world_rank = (2 * merger->local_world_rank) + 1;
    if ( merger->left_world_rank > merger->world_size - 1 )
        merger->left_world_rank = CLOG_RANK_NULL;

    merger->right_world_rank = (2 * merger->local_world_rank) + 2;
    if ( merger->right_world_rank > merger->world_size - 1 )
        merger->right_world_rank = CLOG_RANK_NULL;
}

/* Initialize the CLOG_Merger_t */
void CLOG_Merger_init(       CLOG_Merger_t    *merger,
                       const CLOG_Preamble_t  *preamble,
                       const char             *merged_file_prefix )
{
    /* Set up the binary tree MPI ranks ID locally for merging */
#if !defined( CLOG_NOMPI )
    PMPI_Comm_size( MPI_COMM_WORLD, &(merger->world_size) );
    PMPI_Comm_rank( MPI_COMM_WORLD, &(merger->local_world_rank) );
#else
    merger->world_size       = 1;
    merger->local_world_rank = 0;
#endif
    CLOG_Merger_set_neighbor_ranks( merger );

    merger->is_big_endian  = preamble->is_big_endian;
    /* Open the merged output file at root process */
    if ( merger->parent_world_rank == CLOG_RANK_NULL ) {
        strncpy( merger->out_filename, merged_file_prefix, CLOG_PATH_STRLEN );
        strcat( merger->out_filename, "."CLOG_FILE_TYPE );

        merger->out_fd = OPEN( merger->out_filename, O_CREAT|O_WRONLY|O_TRUNC,
                               0664 );
        if ( merger->out_fd == -1 ) {
            fprintf( stderr, __FILE__":CLOG_Merger_init() - \n"
                             "\t""Could not open file %s for merging!\n",
                             merger->out_filename );
            fflush( stderr );
            CLOG_Util_abort( 1 );
        }
        CLOG_Preamble_write( preamble, CLOG_BOOL_TRUE, merger->out_fd );
    }

    /*
       Initialize the CLOG_Merger's minimum block size: CLOG_REC_ENDBLOCK
    */
    clog_merger_minblocksize = CLOG_Rec_size( CLOG_REC_ENDBLOCK );
}
                                                                                
/* Finalize the CLOG_Merger_t */
void CLOG_Merger_finalize( CLOG_Merger_t *merger )
{
    if ( merger->out_fd != -1 )
        close( merger->out_fd );
}

/*
    Flush the sorted buffer to the output logfile.
*/
void CLOG_Merger_flush( CLOG_Merger_t *merger )
{
    int  ierr;

    if ( merger->is_big_endian != CLOG_BOOL_TRUE )
        CLOG_BlockData_swap_bytes_last( merger->sorted_blk );

    ierr = write( merger->out_fd, merger->sorted_blk->head,
                  merger->block_size );
    if ( ierr != merger->block_size ) {
        fprintf( stderr, __FILE__":CLOG_Merger_flush() - \n"
                         "\t""Fail to write to the logfile %s, ierr = %d.\n",
                         merger->out_filename, ierr );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
}

/*
    clog_merger_minblocksize is defined in CLOG_Merger_init()
*/
int CLOG_Merger_reserved_block_size( unsigned int rectype )
{
    if ( rectype < CLOG_REC_NUM )
        return CLOG_Rec_size( rectype ) + clog_merger_minblocksize;
    else {
        fprintf( stderr, __FILE__":CLOG_Merger_reserved_block_size() - Warning!"
                         "\t""Unknown record type %d\n", rectype );
        fflush( stderr );
        /*
           size to guarantee enough room in each block for the longest record
           + trailer(endblock rec) + internal record(CLOG_Buffer_write2disk)
        */
        return CLOG_Rec_size_max() +  clog_merger_minblocksize;
    }
}

/*
   Save the CLOG record marked by CLOG_Rec_Header_t to the output file.
*/
void CLOG_Merger_save_rec( CLOG_Merger_t *merger, CLOG_Rec_Header_t *hdr )
{
    CLOG_BlockData_t   *sorted_blk;
    CLOG_Rec_Header_t  *sorted_hdr;
    int                 reclen;

    sorted_blk = merger->sorted_blk;


    /*
        If the sorted_blk is full, send sorted_blk to the process's parent
        if the parent exists.  If the parent process does not exist, it must
        be the root.  Then write the sorted_blk to the merged logfile,