clog_buffer.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_STRING_H )
#include <string.h>
#endif
#if defined( STDC_HEADERS ) || defined( HAVE_STDLIB_H )
#include <stdlib.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_preamble.h"
#include "clog_buffer.h"
#include "clog_util.h"
#include "clog_record.h"

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

#define CLOG_NULL_FILE       -5

static int clog_buffer_minblocksize;

CLOG_Buffer_t* CLOG_Buffer_create( void )
{
    CLOG_Buffer_t *buffer;

    buffer = (CLOG_Buffer_t *) MALLOC( sizeof(CLOG_Buffer_t) );
    if ( buffer == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_create() - MALLOC() fails.\n" );
        fflush( stderr );
        return NULL;
    }

    buffer->preamble = CLOG_Preamble_create();
    if ( buffer->preamble == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_create() - \n"
                         "\t""CLOG_Preamble_create() returns NULL.\n" );
        fflush( stderr );
        return NULL;
    }

    buffer->head_block       = NULL;
    buffer->curr_block       = NULL;

    buffer->block_size       = 0;
    buffer->num_blocks       = 0;
    buffer->num_used_blocks  = 0;

#if !defined( CLOG_NOMPI )
    buffer->commset          = CLOG_CommSet_create();
    if ( buffer->commset == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_create() - \n"
                         "\t""CLOG_CommSet_create() returns NULL.\n" );
        fflush( stderr );
        return NULL;
    }
#else
    buffer->commset          = NULL;
#endif

    buffer->local_fd         = CLOG_NULL_FILE;
    strcpy( buffer->local_filename, "" );
    buffer->timeshift_fptr   = 0;
    buffer->delete_localfile = CLOG_BOOL_TRUE;
    buffer->log_overhead     = CLOG_BOOL_TRUE;
    buffer->status           = CLOG_UNINIT;

    return buffer;
}

void CLOG_Buffer_free( CLOG_Buffer_t **buffer_handle )
{
    CLOG_Buffer_t *buffer;
    CLOG_Block_t  *block;

    buffer = *buffer_handle;
#if !defined( CLOG_NOMPI )
    CLOG_CommSet_free( &(buffer->commset) );
#endif

    block = buffer->head_block;
    while ( block != NULL ) {
        buffer->head_block = block->next;
        CLOG_Block_free( &block );
        buffer->num_blocks--;
        block = buffer->head_block;
    }
    buffer->block_size        = 0;
    buffer->num_blocks        = 0;
    buffer->num_used_blocks   = 0;
    CLOG_Preamble_free( &(buffer->preamble) );
    if ( buffer != NULL )
        FREE( buffer );
    *buffer_handle = NULL;
}

void CLOG_Buffer_env_init( CLOG_Buffer_t *buffer )
{
    char          *env_delete_localfile;
    char          *env_log_overhead;
#if !defined( CLOG_NOMPI )
    int            ierr;
#endif

    env_delete_localfile = (char *) getenv( "MPE_DELETE_LOCALFILE" );
    if ( env_delete_localfile != NULL ) {
        if (    strcmp( env_delete_localfile, "false" ) == 0
             || strcmp( env_delete_localfile, "FALSE" ) == 0
             || strcmp( env_delete_localfile, "no" ) == 0
             || strcmp( env_delete_localfile, "NO" ) == 0 )
            buffer->delete_localfile = CLOG_BOOL_FALSE;
    }

    env_log_overhead     = (char *) getenv( "MPE_LOG_OVERHEAD" );
    if ( env_log_overhead != NULL ) {
        if (    strcmp( env_log_overhead, "false" ) == 0
             || strcmp( env_log_overhead, "FALSE" ) == 0
             || strcmp( env_log_overhead, "no" ) == 0
             || strcmp( env_log_overhead, "NO" ) == 0 )
            buffer->log_overhead = CLOG_BOOL_FALSE;
    }

#if !defined( CLOG_NOMPI )
    /*  Let everyone in MPI_COMM_WORLD know what root has */
    ierr = PMPI_Bcast( &(buffer->delete_localfile), 1, MPI_INT,
                       0, MPI_COMM_WORLD );
    if ( ierr != MPI_SUCCESS ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_env_init() - \n"
                         "\t""PMPI_Bcast(buffer->delete_localfile) fails!\n" );
        fflush( stderr );
        PMPI_Abort( MPI_COMM_WORLD, 1 );
    }

    ierr = PMPI_Bcast( &(buffer->log_overhead), 1, MPI_INT,
                       0, MPI_COMM_WORLD );
    if ( ierr != MPI_SUCCESS ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_env_init() - \n"
                         "\t""PMPI_Bcast(buffer->log_overhead) fails!\n" );
        fflush( stderr );
        PMPI_Abort( MPI_COMM_WORLD, 1 );
    }
#endif
}

/*
    If local_tmpfile_name is NULL, the local temporary filename will be chosen
    randomly
*/
void CLOG_Buffer_init( CLOG_Buffer_t *buffer, const char *local_tmpfile_name )
{
    CLOG_Block_t  *block;
    int            num_buffered_blocks;

    /* Initialize CLOG_Preamble_t from the environmental variables */
    CLOG_Preamble_env_init( buffer->preamble );

    /* Initialize core CLOG_Buffer_t's from the environmental variables */
    CLOG_Buffer_env_init( buffer );

    /* Initialize CLOG_Buffer_t's blocks[] based on CLOG_Preamble_t */
    buffer->block_size   = buffer->preamble->block_size;
    num_buffered_blocks  = buffer->preamble->num_buffered_blocks;

    block                = CLOG_Block_create( buffer->block_size );
    buffer->head_block   = block;
    for ( buffer->num_blocks = 1;
          buffer->num_blocks < num_buffered_blocks;
          buffer->num_blocks++ ) {
        block->next = CLOG_Block_create( buffer->block_size );
        block       = block->next;
    }
    buffer->curr_block       = buffer->head_block;
    buffer->num_used_blocks  = 1;

#if !defined( CLOG_NOMPI )
    PMPI_Comm_size( MPI_COMM_WORLD, &(buffer->world_size) );
    PMPI_Comm_rank( MPI_COMM_WORLD, &(buffer->world_rank) );
    CLOG_CommSet_init( buffer->commset );
#else
    buffer->world_size  = 1;
    buffer->world_rank  = 0;
#endif

    if ( local_tmpfile_name != NULL )
        strcpy( buffer->local_filename, local_tmpfile_name );
    if ( strlen( buffer->local_filename ) == 0 ) {
        CLOG_Util_set_tmpfilename( buffer->local_filename );
        if ( strlen( buffer->local_filename ) == 0 ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_init() - \n"
                             "\t""CLOG_Util_set_tmpfilename() fails.\n" );
            fflush( stderr );
            CLOG_Util_abort( 1 );
        }
    }
    buffer->status          = CLOG_INIT_AND_ON;

    /*
       Initialize the CLOG_Buffer's minimum block size:
       CLOG_REC_ENDBLOCK + CLOG_REC_BAREEVT for CLOG_Buffer_write2disk
    */
    clog_buffer_minblocksize = CLOG_Rec_size( CLOG_REC_ENDBLOCK );

    if ( buffer->log_overhead == CLOG_BOOL_TRUE )
        clog_buffer_minblocksize += CLOG_Rec_size( CLOG_REC_BAREEVT );
}

/*
    CLOG_Buffer_localIO_init4write - Open CLOG_Buffer_t.local_fd
    and initialize it with the CLOG_Preamble_t.
    (assuming the local_filename is known ).
*/
void CLOG_Buffer_localIO_init4write( CLOG_Buffer_t *buffer )
{
    buffer->local_fd = OPEN( buffer->local_filename,
                             O_RDWR|O_CREAT|O_TRUNC, 0600 );
    if ( buffer->local_fd == -1 ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_localIO_init4write() - \n"
                         "\t""Fail to open the temporary logfile %s.\n"
                         "\t""Check if the directory where the logfile "
                         "resides exists\n"
                         "\t""and the corresponding file system is "
                         "NOT full.\n"
                         "If not so, set environmental variable TMPDIR to "
                         "a bigger filesystem.\n",
                         buffer->local_filename );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
    /*
       Save the CLOG_Preamble_t at the beginning of the file, where
       CLOG's user_stateID_count & user_solo_eventID_count are set with
       some typical values and they are set in CLOG_Preamble_env_init().
    */
    CLOG_Preamble_write( buffer->preamble, CLOG_BOOL_FALSE, buffer->local_fd );
}

void CLOG_Buffer_localIO_write( CLOG_Buffer_t *buffer )
{
    CLOG_Block_t  *block;
    int            ierr;

    /*  Write the filled block in the buffer to the disk  */
    for ( block = buffer->head_block;
          block != NULL && buffer->num_used_blocks > 0;
          block = block->next ) {
        ierr = write( buffer->local_fd, block->data->head, buffer->block_size );
        if ( ierr != buffer->block_size ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_localIO_write() - \n"
                             "\t""Fail to write to the temporary logfile %s.\n"
                             "\t""Check if the filesystem where the logfile "
                             "resides is full.\n"
                             "If so, set environmental variable TMPDIR to "
                             "a bigger filesystem.\n",
                             buffer->local_filename );
            fflush( stderr );
            CLOG_Util_abort( 1 );
        }
        buffer->num_used_blocks--;
    }

    /*  Reinit the CLOG_Buffer_t for future processing  */
    buffer->curr_block       = buffer->head_block;
    buffer->num_used_blocks  = 1;
}

void CLOG_Buffer_advance_block( CLOG_Buffer_t *buffer )
{
    const CLOG_CommIDs_t  *commIDs;

    if ( buffer->curr_block->next != NULL ) {
        CLOG_Buffer_save_endblock( buffer );

        buffer->curr_block = buffer->curr_block->next;
        buffer->num_used_blocks++;
        CLOG_Block_reset( buffer->curr_block );
    }
    else {
        /*
           Assume space has already been reserved for CLOG_Buffer_write2disk,
           So save the state CLOG_Buffer_write2disk without checking.
           Can't check anyway, circular logic!
        */
        commIDs  = NULL;
        if ( buffer->log_overhead == CLOG_BOOL_TRUE ) {
            commIDs  = CLOG_CommSet_get_IDs( buffer->commset, MPI_COMM_WORLD );
            CLOG_Buffer_save_bareevt_0chk( buffer, commIDs, 0,
                                           CLOG_EVT_BUFFERWRITE_START );
        }
        CLOG_Buffer_save_endblock( buffer );

        if ( buffer->local_fd == CLOG_NULL_FILE )
            CLOG_Buffer_localIO_init4write( buffer );
        CLOG_Buffer_localIO_write( buffer );
        CLOG_Block_reset( buffer->curr_block );

        /*  Assume commIDs is non-NULL and has been updated previously */
        if ( buffer->log_overhead == CLOG_BOOL_TRUE ) {
            CLOG_Buffer_save_bareevt( buffer, commIDs, 0,
                                      CLOG_EVT_BUFFERWRITE_FINAL );
        }
    }
}

void CLOG_Buffer_localIO_flush( CLOG_Buffer_t *buffer )
{
    if ( buffer->local_fd != CLOG_NULL_FILE ) {
        CLOG_Buffer_localIO_write( buffer );
    }