clog_cache.c

fortran并行计算包

/*
   (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
#ifdef HAVE_SYS_TYPES_H
#include <sys/types.h>
#endif
#if defined( HAVE_UNISTD_H )
#include <unistd.h>
#endif

#include "clog_mem.h"
#include "clog_cache.h"
#include "clog_preamble.h"
#include "clog_commset.h"
#include "clog_util.h"

static int clog_cache_minblocksize; /* constant needed for writing to cache */

CLOG_Cache_t* CLOG_Cache_create( void )
{
    CLOG_Cache_t *cache;

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

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

    cache->commset     = CLOG_CommSet_create();
    if ( cache->commset == NULL ) {
        fprintf( stderr, __FILE__":CLOG_Cache_create() - \n"
                         "\t""CLOG_CommSet_create() returns NULL.\n" );
        fflush( stderr );
        return NULL;
    }

    cache->blockdata             = NULL;
    cache->block_size            = -9999;

    cache->is_runtime_bigendian  = CLOG_Util_is_runtime_bigendian();

    cache->local_fd              = CLOG_NULL_FILE;
    strcpy( cache->local_filename, "" );

    cache->local_timediff        = CLOG_MAXTIME;    /* Set an invalid time */

    return cache;
}

void CLOG_Cache_free( CLOG_Cache_t **cache_handle )
{
    CLOG_Cache_t  *cache;

    cache = *cache_handle;

    if ( cache != NULL ) {
        CLOG_CommSet_free( &(cache->commset) );
        CLOG_BlockData_free( &(cache->blockdata) );
        cache->block_size = 0;
        CLOG_Preamble_free( &(cache->preamble) );
        FREE( cache );
    }
    *cache_handle = NULL;
}

/* fill the block with data from the disk */
static void CLOG_Cache_fillblock( CLOG_Cache_t *cache )
{
    int ierr;

    ierr = read( cache->local_fd, cache->blockdata->head, cache->block_size );
    if ( ierr != cache->block_size ) {
        fprintf( stderr, __FILE__":CLOG_Cache_fillblock() - \n"
                         "\t""read(cache->blockdata) fails w/ err=%d.\n",
                         ierr );
        fflush( stderr );
        exit( 1 );
    }
    /* Do byteswapping if necessary then initialize the CLOG_BlockData_t */
    if ( cache->preamble->is_big_endian != cache->is_runtime_bigendian )
        CLOG_BlockData_swap_bytes_first( cache->blockdata );

    /* fixup the local communicator ID in each record */
    if ( cache->preamble->is_finalized != CLOG_BOOL_TRUE )
        CLOG_BlockData_patch_all( cache->blockdata, &(cache->local_timediff),
                                  cache->commset->table );
    else
        CLOG_BlockData_patch_comm( cache->blockdata, cache->commset->table );

    CLOG_BlockData_reset( cache->blockdata );
}

/* flush the block's data to the disk */
static void CLOG_Cache_flushblock( CLOG_Cache_t *cache )
{
    int ierr;

    /* CLOG_Cache_t only writes big-endian file */
    /* if ( cache->is_runtime_bigendian != CLOG_BOOL_TRUE ) */
    if ( cache->preamble->is_big_endian != CLOG_BOOL_TRUE )
        CLOG_BlockData_swap_bytes_last( cache->blockdata );

    ierr = write( cache->local_fd, cache->blockdata->head, cache->block_size );
    if ( ierr != cache->block_size ) {
        fprintf( stderr, __FILE__":CLOG_Cache_flushblock() - \n"
                         "\t""Fail to write to the logfile %s.\n"
                         "\t""Check if the filesystem where the logfile "
                         "resides is full.\n", cache->local_filename );
        fflush( stderr );
        exit( 1 );
    }
    CLOG_BlockData_reset( cache->blockdata );
}

void CLOG_Cache_flushlastblock( CLOG_Cache_t *cache )
{
    CLOG_BlockData_t   *blkdata;
    CLOG_Rec_Header_t  *last_hdr;

    blkdata = cache->blockdata;

    /* Write a CLOG_REC_ENDLOG record */
    last_hdr = (CLOG_Rec_Header_t *) blkdata->ptr;
    last_hdr->time    = CLOG_MAXTIME;
    last_hdr->icomm   = 0;                  /* arbitary */
    last_hdr->rank    = 0;                  /* arbitary */
    last_hdr->thread  = 0;                  /* arbitary */
    last_hdr->rectype = CLOG_REC_ENDLOG;

    CLOG_Cache_flushblock( cache );
}

/* CLOG_Cache_open4read() has to be called before CLOG_Cache_init4read() */
void CLOG_Cache_open4read( CLOG_Cache_t *cache, const char *filename )
{
    CLOG_BOOL_T  do_byte_swap;
    int          ierr;

    if ( filename != NULL && strlen(filename) != 0 )
        strcpy( cache->local_filename, filename );
    else {
        fprintf( stderr, __FILE__":CLOG_Cache_open4read() - \n"
                         "\t""input file name is empty.\n" );
        fflush( stderr );
        exit( 1 );
    }

    cache->local_fd = OPEN( cache->local_filename, O_RDONLY, 0 );
    if ( cache->local_fd == -1 ) {
        fprintf( stderr, __FILE__":CLOG_Cache_open4read() - \n"
                         "\t""Fail to open the file %s for reading.\n",
                         cache->local_filename );
        fflush( stderr );
        exit( 1 );
    }

    CLOG_Preamble_read( cache->preamble, cache->local_fd );
    if ( cache->preamble->commtable_fptr < CLOG_PREAMBLE_SIZE ) {
        fprintf( stderr, __FILE__":CLOG_Cache_open4read() - Warning!\n"
                         "\t""Invalid commtable_fptr, "i64fmt", < "
                         "CLOG_PREAMBLE_SIZE, %d.\n"
                         "\t""This logfile could be incomplete. "
                         "clog2_repair may be able to fix it.\n",
                         cache->preamble->commtable_fptr, CLOG_PREAMBLE_SIZE );
        fflush( stderr );
        return;
    }

    lseek( cache->local_fd, cache->preamble->commtable_fptr, SEEK_SET );
    do_byte_swap = (    cache->preamble->is_big_endian
                     != cache->is_runtime_bigendian );
    ierr = CLOG_CommSet_read( cache->commset, cache->local_fd, do_byte_swap );
    if ( ierr <= 0 ) {
        fprintf( stderr, __FILE__":CLOG_Cache_open4read() - Warning!\n"
                         "\t""CLOG_CommSet_read() return an error code, %d.\n"
                         "\t""This logfile could be incomplete. "
                         "clog2_repair may be able to fix it.\n",
                         ierr );
        fflush( stderr );
    }
}

/*
    CLOG_Cache_t's input interface is separated into 2 functions:
    CLOG_Cache_open4read() and CLOG_Cache_init4read() because the
    synchronization of local commIDs, CLOG_Joiner_sync_commIDs(),
    needs to be inserted between the 2 functions.
*/
/* CLOG_Cache_init4read() has to be called after CLOG_Cache_open4read() */
void CLOG_Cache_init4read( CLOG_Cache_t *cache )
{
    /* Create CLOG_BlockData_t based on CLOG_Preamble_t */
    cache->block_size  = cache->preamble->block_size;
    cache->blockdata   = CLOG_BlockData_create( cache->block_size );

    /* Initialize timediff if CLOG_Cache_t is for a local file */
    if ( cache->preamble->is_finalized != CLOG_BOOL_TRUE )
        cache->local_timediff = 0.0;

    /* Initial update of CLOG_BlockData_t with data from the disk */
    lseek( cache->local_fd, (off_t) CLOG_PREAMBLE_SIZE, SEEK_SET );
    CLOG_Cache_fillblock( cache );
}

/* CLOG_Cache_open4write() has to be called before CLOG_Cache_init4write() */
void CLOG_Cache_open4write( CLOG_Cache_t *cache, const char *filename )
{
    if ( filename != NULL && strlen(filename) != 0 )
        strcpy( cache->local_filename, filename );
    else {
        fprintf( stderr, __FILE__":CLOG_Cache_open4write() - \n"
                         "\t""input file name is empty.\n" );
        fflush( stderr );
        exit( 1 );
    }

    cache->local_fd = OPEN( cache->local_filename, O_CREAT|O_WRONLY|O_TRUNC,
                            0664 );
    if ( cache->local_fd == -1 ) {
        fprintf( stderr, __FILE__":CLOG_Cache_open4write() - \n"
                         "\t""Fail to open the file %s for writing.\n",
                         cache->local_filename );
        fflush( stderr );
        exit( 1 );
    }

    /*
       Set the OS's native byte ordering to the CLOG_Preamble_t,
       so this is consistent with CLOG_Merger_t.
    */
    cache->preamble->is_big_endian = cache->is_runtime_bigendian;
    CLOG_Preamble_write( cache->preamble, CLOG_BOOL_TRUE, CLOG_BOOL_TRUE,
                         cache->local_fd );
}

/* CLOG_Cache_init4write() has to be called after CLOG_Cache_open4write() */
void CLOG_Cache_init4write( CLOG_Cache_t *cache )
{
    /* Create CLOG_BlockData_t based on CLOG_Preamble_t */
    cache->block_size  = cache->preamble->block_size;
    cache->blockdata   = CLOG_BlockData_create( cache->block_size );

    /*
       Initialize the CLOG_Cache_t's minimum block size: CLOG_REC_ENDBLOCK
    */
    clog_cache_minblocksize = CLOG_Rec_size( CLOG_REC_ENDBLOCK );
}

/* CLOG_Cache_open4readwrite() has to be called before CLOG_Cache_init4read() */
void CLOG_Cache_open4readwrite( CLOG_Cache_t *cache, const char *filename )
{
    CLOG_BOOL_T  do_byte_swap;
    int          ierr;

    if ( filename != NULL && strlen(filename) != 0 )
        strcpy( cache->local_filename, filename );
    else {
        fprintf( stderr, __FILE__":CLOG_Cache_open4readwrite() - \n"
                         "\t""input file name is empty.\n" );
        fflush( stderr );
        exit( 1 );
    }

    cache->local_fd = OPEN( cache->local_filename, O_RDWR, 0664 );
    if ( cache->local_fd == -1 ) {
        fprintf( stderr, __FILE__":CLOG_Cache_open4readwrite() - \n"
                         "\t""Fail to open the file %s for reading.\n",
                         cache->local_filename );
        fflush( stderr );