clog_buffer.c

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

}

off_t CLOG_Buffer_localIO_ftell( CLOG_Buffer_t *buffer )
{
    CLOG_BlockData_t     *blkdata;
    off_t                 fptr;

    /* Determine the current offset within curr_block */
    blkdata  = buffer->curr_block->data;
    fptr     = blkdata->ptr - blkdata->head;

    /* Determine the space taken by the number of filled blocks */
    fptr    += ( buffer->num_used_blocks - 1 ) * buffer->block_size;

    if ( buffer->local_fd != CLOG_NULL_FILE )
        fptr += lseek( buffer->local_fd, 0, SEEK_CUR );
    else
        fptr += CLOG_PREAMBLE_SIZE;
    return fptr;
}

/* Allocate a default CLOG_Rec_Timeshift record in the memory buffer.  */
void CLOG_Buffer_init_timeshift( CLOG_Buffer_t *buffer )
{
    const CLOG_CommIDs_t  *commIDs;
    /*
       The default CLOG_Rec_Timeshift recird needs to be saved before
       CLOG_Buffer_t.timeshift_fptr is updated to account for the possibility
       that CLOG_Buffer_t.curr_block may not have enough room for a
       CLOG_Rec_Timeshift record.  After saving the record, the location
       of the CLOG_Rec_Timeshift is unchangable.
    */
    commIDs  = CLOG_CommSet_get_IDs( buffer->commset, MPI_COMM_WORLD );
    CLOG_Buffer_save_timeshift( buffer, commIDs, 0, 0.0 );
    buffer->timeshift_fptr = CLOG_Buffer_localIO_ftell( buffer )
                           - CLOG_Rec_size( CLOG_REC_TIMESHIFT );
}

void CLOG_Buffer_set_timeshift( CLOG_Buffer_t *buffer,
                                CLOG_Time_t    new_timediff,
                                int            init_next_timeshift )
{
    CLOG_Rec_Header_t    *hdr, hdr_rec;
    CLOG_Rec_Timeshift_t *tshift, tshift_rec;
    CLOG_Block_t         *block;
    CLOG_BlockData_t     *blkdata;
    off_t                 last_timeshift_fptr, curr_fptr, fptr;
    int                   reclen_hdr, reclen_tshift;
    int                   block_ptr, ierr = 0;

    /* Save the CLOG_Buffer_t.timeshift_fptr for later update */
    last_timeshift_fptr    = buffer->timeshift_fptr;

    if ( init_next_timeshift == CLOG_BOOL_TRUE )
        CLOG_Buffer_init_timeshift( buffer );

    /* Save the current file pointer */
    if ( buffer->local_fd != CLOG_NULL_FILE )
        curr_fptr = lseek( buffer->local_fd, 0, SEEK_CUR );
    else
        curr_fptr = CLOG_PREAMBLE_SIZE;
    /* Check if last_timeshift_ptr is in the memory buffer or on the disk */
    if (  last_timeshift_fptr < curr_fptr ) {
        if (buffer->local_fd  == CLOG_NULL_FILE ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                             "\t""buffer->local_fd == NULL_FILE detected.\n" );
            fflush( stderr );
            return;
        }
        /* Locate the last CLOG_Rec_Timeshift_t in the file */
        lseek( buffer->local_fd, last_timeshift_fptr, SEEK_SET );
        reclen_hdr     = CLOG_RECLEN_HEADER;
        reclen_tshift  = CLOG_RECLEN_TIMESHIFT;
        /* Set CLOG_Rec_Header_t & CLOG_Rec_Timeshift_t to local tmp storages */
        hdr            = &hdr_rec;   
        tshift         = &tshift_rec;
        /* Fetch the CLOG_Rec_Header of the CLOG_Rec_Timeshift from the disk */
        ierr = read( buffer->local_fd, hdr, reclen_hdr );
        if ( ierr != reclen_hdr ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                             "\t""read(CLOG_Rec_Header) fails w/ err=%d.\n",
                             ierr );
            fflush( stderr );
            return;
        }
        if ( hdr->rectype != CLOG_REC_TIMESHIFT ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                             "\t""1st disk record is not CLOG_Rec_Timeshift.\n"
                             );
            fflush( stderr );
            return;
        }
        fptr = lseek( buffer->local_fd, 0, SEEK_CUR );
        ierr = read( buffer->local_fd, tshift, reclen_tshift );
        if ( ierr != reclen_tshift ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                             "\t""read(CLOG_Rec_Timeshift) fails w/ err=%d.\n",
                             ierr );
            fflush( stderr );
            return;
        }
        /* Update the CLOG_Rec_Timeshift_t with the new timeshift value */
        tshift->timeshift = new_timediff;
        /* Seek back in file of where CLOG_Rec_Timeshift_t was retrieved */
        lseek( buffer->local_fd, fptr, SEEK_SET );
        ierr = write( buffer->local_fd, tshift, reclen_tshift );
        if ( ierr != reclen_tshift ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                             "\t""write(CLOG_Rec_Timeshift) fails w/ err=%d.\n",
                             ierr );
            fflush( stderr );
            return;
        }
        /* Restore the current file pointer */
        lseek( buffer->local_fd, curr_fptr, SEEK_SET );
    }
    else { /* if (  last_timeshift_fptr >= curr_fptr ) */
        /* Locate the CLOG_Block_t that contains the last CLOG_Rec_Timeshift */
        block       = buffer->head_block;
        block_ptr   = last_timeshift_fptr - curr_fptr;
        while ( block_ptr >= buffer->block_size ) {
            block_ptr -= buffer->block_size;
            if ( block == NULL ) {
                fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() - \n"
                                 "\t""End of memory buffer encountered!\n" );
                fflush( stderr );
                return;
            }
            block      = block->next;
        }

        blkdata     = block->data;
        hdr         = (CLOG_Rec_Header_t *) ( blkdata->head + block_ptr );
        if ( hdr->rectype != CLOG_REC_TIMESHIFT ) {
            fprintf( stderr, __FILE__":CLOG_Buffer_set_timeshift() -\n"
                             "\t""No CLOG_Rec_Timeshift at the expected "
                             "record location %d!\n", block_ptr );
            fflush( stderr );
            return;
        }
        tshift            = (CLOG_Rec_Timeshift_t *) hdr->rest;
        /* Update the CLOG_Rec_Timeshift_t with the new timeshift value */
        tshift->timeshift = new_timediff;
    }
}

/*
   Reset CLOG_Buffer_t's localIO to read mode
   If the local_fd has been written, update local buffer from the
   beginning of the file.  If the local_fd has not been used,
   reposition the current block of the buffer to the head block.
*/
void CLOG_Buffer_localIO_reinit4read( CLOG_Buffer_t *buffer )
{
    if ( buffer->local_fd != CLOG_NULL_FILE ) {
        lseek( buffer->local_fd, (off_t) CLOG_PREAMBLE_SIZE, SEEK_SET );
        CLOG_Buffer_localIO_read( buffer );
    }
    else
        buffer->curr_block = buffer->head_block;
}

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

    /* If CLOG_Buffer_t.local_fd isn't initialized, nothing to read from disk */
    if ( buffer->local_fd == CLOG_NULL_FILE )
        return;

    /*
        CLOG_Buffer_t.num_used_blocks is used to indicate if only less than
        CLOG_Buffer_t.num_blocks is stored on the disk
    */
    buffer->num_used_blocks  = 0;
    ierr                     = buffer->block_size;
    for ( block = buffer->head_block; block != NULL; block = block->next ) {
        ierr = read( buffer->local_fd, block->data->head, buffer->block_size );
        if ( ierr > 0 ) {
            if ( ierr != buffer->block_size ) {
                fprintf( stderr, __FILE__":CLOG_Buffer_localIO_read() - \n"
                                 "\t""read(%s) cannot fetch the whole block "
                                 "in one scoop.\n", buffer->local_filename );
                fflush( stderr );
                CLOG_Util_abort( 1 );
            }
        }
        else { /* if ierr <= 0 */ 
            if ( ierr < 0 ) {
                fprintf( stderr, __FILE__":CLOG_Buffer_localIO_read() - \n"
                                 "\t""read(%s) fails with an error=%d.\n",
                                 buffer->local_filename, ierr );
                fflush( stderr );
                CLOG_Util_abort( 1 );
            }
            else /* if ierr == 0 */
                break;
        }
        buffer->num_used_blocks++;
    }

    buffer->curr_block       = buffer->head_block;
}

/*
    CLOG_Buffer_localIO_finalize -
    Close CLOG_Buffer_t.local_fd (assuming the local_fd is > 0 ).
    Also, delete the CLOG_Buffer_t.local_filename (assuming file already opened)
*/
void CLOG_Buffer_localIO_finalize( CLOG_Buffer_t *buffer )
{
    if ( buffer->local_fd != CLOG_NULL_FILE ) {
        close( buffer->local_fd );
        buffer->local_fd = CLOG_NULL_FILE;
        if (    buffer->delete_localfile == CLOG_BOOL_TRUE
             && strlen( buffer->local_filename ) != 0 )
            unlink( buffer->local_filename );
    }
}

/*
    clog_buffer_minblocksize is defined in CLOG_Buffer_init()
*/
int CLOG_Buffer_reserved_block_size( unsigned int rectype )
{
    if ( rectype < CLOG_REC_NUM )
        return CLOG_Rec_size( rectype ) + clog_buffer_minblocksize;
    else {
        fprintf( stderr, __FILE__":CLOG_Buffer_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_buffer_minblocksize;
    }
}




void CLOG_Buffer_save_endlog( CLOG_Buffer_t *buffer )
{
    const CLOG_CommIDs_t  *commIDs;
    if ( buffer->status == CLOG_INIT_AND_ON ) {
        commIDs  = CLOG_CommSet_get_IDs( buffer->commset, MPI_COMM_WORLD );
        CLOG_Buffer_save_header_0chk( buffer, commIDs, 0, CLOG_REC_ENDLOG );
    }
    else if ( buffer->status == CLOG_UNINIT ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_save_endlog() - \n"
                         "\t""CLOG is used before being initialized.\n" );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
}

void CLOG_Buffer_save_endblock( CLOG_Buffer_t *buffer )
{
    const CLOG_CommIDs_t  *commIDs;
    if ( buffer->status == CLOG_INIT_AND_ON ) {
        commIDs  = CLOG_CommSet_get_IDs( buffer->commset, MPI_COMM_WORLD );
        CLOG_Buffer_save_header_0chk( buffer, commIDs, 0, CLOG_REC_ENDBLOCK );
    }
    else if ( buffer->status == CLOG_UNINIT ) {
        fprintf( stderr, __FILE__":CLOG_Buffer_save_endblock() - \n"
                         "\t""CLOG is used before being initialized.\n" );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
}

void CLOG_Buffer_save_header_0chk( CLOG_Buffer_t *buffer,
                                   const CLOG_CommIDs_t *commIDs,
                                   CLOG_ThreadLID_t thd,
                                   int rectype )
{
    CLOG_BlockData_t   *blkdata;
    CLOG_Rec_Header_t  *hdr;

    blkdata          = buffer->curr_block->data;
    hdr              = (CLOG_Rec_Header_t *) blkdata->ptr;
    hdr->time        = CLOG_Timer_get();
#if !defined( CLOG_NOMPI )
    hdr->icomm       = commIDs->local_ID;
    hdr->rank        = commIDs->comm_rank;
#else
    hdr->icomm       = 0;
    hdr->rank        = 0;
#endif
    hdr->thread      = thd;
    hdr->rectype     = rectype;
    blkdata->ptr     = hdr->rest;  /* advance to next available space */
}

void CLOG_Buffer_save_header( CLOG_Buffer_t *buffer,
                              const CLOG_CommIDs_t *commIDs,
                              CLOG_ThreadLID_t thd,
                              int rectype )
{
    CLOG_BlockData_t   *blkdata;
    CLOG_Rec_Header_t  *hdr;

    blkdata          = buffer->curr_block->data;
    if (    blkdata->ptr + CLOG_Buffer_reserved_block_size( rectype )
         >= blkdata->tail ) {
        CLOG_Buffer_advance_block( buffer );
        blkdata      = buffer->curr_block->data;
    }
    hdr              = (CLOG_Rec_Header_t *) blkdata->ptr;
    hdr->time        = CLOG_Timer_get();
#if !defined( CLOG_NOMPI )
    hdr->icomm       = commIDs->local_ID;
    hdr->rank        = commIDs->comm_rank;
#else
    hdr->icomm       = 0;
    hdr->rank        = 0;
#endif
    hdr->thread      = thd;
    hdr->rectype     = rectype;
    blkdata->ptr     = hdr->rest;  /* advance to next available space */
}

void CLOG_Buffer_save_statedef( CLOG_Buffer_t *buffer,
                                const CLOG_CommIDs_t *commIDs,
                                CLOG_ThreadLID_t thd,
                                int stateID, int startetype, int finaletype,