slog_irec_write.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

    /*
        Check if all the previuous calls of SLOG_Irec_ReserveSpace() 
        and SLOG_Irec_ToOutputStream() have used up all the space
        on the disk.   In particular, this checks if too many of
        SLOG_Irec_ReserveSpace() calls have been used up or over-reserved
        all the space in the frame buffer. 

        This check can only be done in SLOG_Irec_ToOutputStream() instead
        of SLOG_Irec_ReserveSpace(), because there is no way to reset
        the slog->reserve->Nbytes_tot correctly in SLOG_Irec_ReserveSpace().
    */
    if ( slog->reserve != NULL ) { 
        Nbytes_in_bbufs_patched = SLOG_typesz[ FrameHdr ]
                                + slog->hdr->frame_reserved_size
                                + slog->reserve->Nbytes_tot;

        if ( Nbytes_in_bbufs_patched > fbuf_bufsz( slog->fbuf ) ) {
            fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                              "\t""At frame idx = "fmt_ui32", the total "
                              "allocated size of all _incompleted_ and \n"
                              "\t""_completed_ intervals is too big to fit "
                              "in the frame, and this may \n"
                              "\t""causes problem in the 2nd pass\n",
                              ( slog->frame_dir ).Nframe );
            fprintf( errfile, "\t""Nbytes_in_bbufs_patched ("fmt_ui32") > "
                              "frame_buffer_size("fmt_ui32")\n",
                              Nbytes_in_bbufs_patched,
                              fbuf_bufsz( slog->fbuf ) );
            fflush( errfile );
        }
    }

    /*
        since the new buffer size is NOT too big for the buffer 
        Add/Copy the new interval record to the SLOG Current buffer 
    */
    ierr = SLOG_Bbuf_AddCopiedIntvlRec( slog->cur_bbuf, irec );
    if ( ierr != SLOG_SUCCESS ) {
        fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                          "\t""SLOG_Bbuf_AddCopiedIntvlRec() fails !!\n" );
        fflush( errfile );
        return SLOG_FAIL;
    }

    /*
        Update the record definition table to indicate which definition
        has been used.
    */
#if defined( COMPRESSION )
    recdef = SLOG_RDEF_GetRecDef( slog->rec_defs, irec->intvltype,
                                  irec->bebits[0], irec->bebits[1] );
    if ( recdef == NULL ) {
        fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                          "\t""SLOG_RecDef_SetUsed ("fmt_itype_t","
                          fmt_bebit_t","fmt_bebit_t") fails!\n",
                          irec->intvltype, irec->bebits[0], irec->bebits[1] );
        fflush( errfile );
        return SLOG_FAIL;
    }
    SLOG_RecDef_SetUsed( recdef, SLOG_TRUE );

    intvlinfo = SLOG_PROF_GetIntvlInfo( slog->prof, irec->intvltype,
                                        irec->bebits[0], irec->bebits[1] );
    if ( intvlinfo == NULL ) {
        fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - Warning !!!\n"
                          "\t""SLOG_IntvlInfo_SetUsed ("fmt_itype_t","
                          fmt_bebit_t","fmt_bebit_t") fails!  "
                          "Program Continues!\n",
                          irec->intvltype, irec->bebits[0], irec->bebits[1] );
        fflush( errfile );
        /*  return SLOG_FAIL;  */
    }
    SLOG_IntvlInfo_SetUsed( intvlinfo, SLOG_TRUE );
#endif

    /*
        Check if there is any frame space reservation inconsistency
    */
    if ( slog->reserve != NULL ) {
        slog->reserve->Nrec_eff --;
        if ( slog->reserve->Nrec_eff < 0 ) {
            fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                              "\t""Number of outstanding interval records = "
                              fmt_i32"\n", slog->reserve->Nrec_eff );
            fflush( errfile );
            return SLOG_FAIL;
        }

        ierr = SLOG_RecDef_SizeOfIrecInFile( slog->rec_defs,
                                             irec->rectype, irec->intvltype,
                                             irec->bebits[0], irec->bebits[1] );
        if ( ierr == SLOG_FAIL ) {
            fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                              "\t""SLOG_Irec_SizeOf("fmt_rtype_t","fmt_itype_t
                              ","fmt_bebit_t","fmt_bebit_t") fails!\n",
                              irec->rectype, irec->intvltype,
                              irec->bebits[0], irec->bebits[1] );
            fflush( errfile );
            return SLOG_FAIL;
        }
        irec_bytesize_estimate = ierr;

        if ( irec_bytesize_estimate < irec->bytesize ) {
            fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                              "\t""SLOG_Irec_SizeOf("fmt_rtype_t","fmt_itype_t
                              ","fmt_bebit_t","fmt_bebit_t") returns %d bytes "
                              "< the actual irec->bytesize, "fmt_sz_t" bytes\n",
                              irec->rectype, irec->intvltype,
                              irec->bebits[0], irec->bebits[1],
                              irec_bytesize_estimate, irec->bytesize );
            fprintf( errfile, "\t""Possible space reservation "
                              "inconsistency!\n" ); 
            fflush( errfile );
            return SLOG_FAIL;
        }
        slog->reserve->Nbytes_eff -= irec_bytesize_estimate;

        if ( slog->reserve->Nbytes_eff < 0 ) {
            fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                              "\t""Byte size of all outstanding intervals, "
                              "slog->reserve->Nbytes_eff = "fmt_i32"\n",
                              slog->reserve->Nbytes_eff );
            fflush( errfile );
            return SLOG_FAIL;
        }
    }   /*  Endif ( slog->reserve != NULL )  */

    return SLOG_SUCCESS;
}



/*@C
    SLOG_CloseOutputStream - Close the specified SLOG_STREAM. It frees 
                             all the internally allocated memory used 
                             in SLOG_STREAM

  Modified Input Variables :
. slog - pointer to the SLOG_STREAM to be closed.

  Modifies Output Variables :
. returned value - integer status code

  Include File Needed :
  slog.h

.N SLOG_RETURN_STATUS
@*/
int SLOG_CloseOutputStream( SLOG_STREAM *slog )
{
    /*  SLOG_uint32 Nbytes_written;  */
    SLOG_uint32 count_irec_in_bbufs;
    int         IsLastFrame = SLOG_TRUE;
    int         ierr;

#if defined( DEBUG )
    SLOG_uint32     old_buf_Nbytes_in_file;
    SLOG_uint32     Nbytes_in_bbufs_patched;
#endif

    if ( slog->HasIrec2IOStreamBeenUsed == SLOG_TRUE ) {

#if defined( DEBUG )
        old_buf_Nbytes_in_file = SLOG_typesz[ FrameHdr ]
                               + slog->hdr->frame_reserved_size
                               + ( slog->cur_bbuf )->Nbytes_in_file
                               + ( slog->tmp_bbuf )->Nbytes_in_file;
        if ( slog->reserve != NULL ) 
            old_buf_Nbytes_in_file += slog->reserve->Nbytes_eff;
        fprintf( stdout, __FILE__":SLOG_CloseOutputStream() - "
                 "Starting SLOG_STM_UpdateFrameDirEntry_Forward()\n" );
        fprintf( stdout, "\t""FrameHdr + sum_reserved_space "
                         "+ _cur_ + _tmp_ = "fmt_ui32" bytes\n",
                         old_buf_Nbytes_in_file );
        fflush( stdout );
#endif

#if defined( DEBUG )
        if ( slog->reserve != NULL ) {
            slog->reserve->Nbytes_tot -= slog->cur_bbuf->Nbytes_in_file;
            slog->reserve->Nrec_tot   -= slog->cur_bbuf->count_irec;
        }
        if ( slog->reserve != NULL ) {
            Nbytes_in_bbufs_patched = SLOG_typesz[ FrameHdr ]
                                    + slog->hdr->frame_reserved_size
                                    + slog->reserve->Nbytes_tot;
    
            if ( Nbytes_in_bbufs_patched > fbuf_bufsz( slog->fbuf ) ) {
                fprintf( errfile, __FILE__":SLOG_Irec_ToOutputStream() - \n"
                                  "\t""At frame idx = "fmt_ui32", the total "
                                  "allocated size of all _incompleted_ and \n"
                                  "\t""_completed_ intervals is too big to fit "
                                  "in the frame, and this may \n"
                                  "\t""causes problem in the 2nd pass\n",
                                  ( slog->frame_dir ).Nframe );
                fprintf( errfile, "\t""Nbytes_in_bbufs_patched ("fmt_ui32") > "
                                  "frame buffer size("fmt_ui32")\n",
                                  Nbytes_in_bbufs_patched,
                                  fbuf_bufsz( slog->fbuf ) );
                fflush( errfile );
            }
        }
        if ( slog->reserve != NULL ) {
            fprintf( stdout, "\t""reserve->Nbytes_eff = "fmt_i32", "
                             "reserve->Nbytes_tot = "fmt_i32"\n",
                              slog->reserve->Nbytes_eff,
                              slog->reserve->Nbytes_tot );
        }
#endif

        count_irec_in_bbufs = ( slog->cur_bbuf )->count_irec
                            + ( slog->tmp_bbuf )->count_irec;

        if ( count_irec_in_bbufs > 0 ) { 
            ierr = SLOG_STM_UpdateFrameDirEntry_Forward( slog, IsLastFrame );
            if ( ierr != SLOG_SUCCESS ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                                  "SLOG_STM_UpdateFrameDirEntry_Forward() "
                                  "fails!!\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }

            ierr = SLOG_STM_UpdateFRAME_Forward( slog );
            if ( ierr != SLOG_SUCCESS ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                                  "SLOG_STM_UpdateFRAME_Forward( slog ) "
                                  "fails!!\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }

            ierr = SLOG_STM_WriteFRAME( slog );
            if ( ierr == SLOG_FAIL ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                                  "SLOG_STM_WriteFRAME( slog ) fails!!\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }
            /*  Nbytes_written = ierr;  */
        
            ierr = SLOG_STM_UpdateWriteFRAMEDIR( slog );
            if ( ierr != SLOG_SUCCESS ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                         "SLOG_STM_UpdateWriteFRAMEDIR( slog ) fails!!\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }

            /*  SLOG_STM_InitAgainBeforeAddRec( slog );  */
        }   /*  Endof    if ( count_irec_in_bbufs > 0 )  */

#if defined( CHECKRECDEFS )
        ierr = SLOG_RDEF_IsBebitIntvlsConsistent( slog->rec_defs );
        if ( ierr != SLOG_TRUE ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - Warning!\n"
                              "\t""SLOG_RDEF_IsBebitIntvlsConsistent() "
                              "fails, program continues.\n" );
            fflush( errfile );
            /* return SLOG_FAIL; */
        }
#endif

#if defined( CHECKPROFILE )
        ierr = SLOG_PROF_IsBebitIntvlsConsistent( slog->prof );
        if ( ierr != SLOG_TRUE ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - Warning!\n"
                              "\t""SLOG_PROF_IsBebitIntvlsConsistent() "
                              "fails, program contines.\n" );
            fflush( errfile );
            /* return SLOG_FAIL; */
        }
#endif

        /*
            Compress the various file headers
        */
#if defined( COMPRESSION )
        ierr = SLOG_RDEF_CompressTable( slog->rec_defs );
        if ( ierr != SLOG_SUCCESS ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                              "SLOG_RDEF_CompressTable() fails\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }
        ierr = SLOG_RDEF_CompressedTableToFile( slog );
        if ( ierr != SLOG_SUCCESS ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                              "SLOG_RDEF_CompressdTableToFile() fails\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }

        ierr = SLOG_PROF_CompressTable( slog->prof );
        if ( ierr != SLOG_SUCCESS ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                              "SLOG_PROF_CompressTable() fails\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }
        ierr = SLOG_PROF_CompressedTableToFile( slog );
        if ( ierr != SLOG_SUCCESS ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                              "SLOG_PROF_CompressdTableToFile() fails\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }
#endif

#if defined( CHECKPROFILE ) && defined( CHECKRECDEFS )
        ierr = SLOG_STM_IsPROFConsistentWithRDEF( slog->prof, slog->rec_defs );
        if ( ierr != SLOG_TRUE ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - \n"
                              "\t""The Display Profile is Inconsistent with "
                              "the Record Definition Table\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }
#endif

        /*  Initialize SLOG's Preview Statistics Array Table  */
        ierr = SLOG_PSTAT_Init( slog );
        if ( ierr != SLOG_SUCCESS ) {
            fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                              "SLOG_PSTAT_Init() fails\n" );
            fflush( errfile );
            return SLOG_FAIL;
        }

        /*
            Process the logfile backward in time to patch the pseudo record
            holes created during the 1st forward pass of the logfile 
        */
        if ( slog->hdr->IsIncreasingEndtime == SLOG_TRUE ) {
            ierr = SLOG_PSTAT_SetBinWidthInTime( slog );
            if ( ierr != SLOG_SUCCESS ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                                  "SLOG_PSTAT_SetBinWidthInTime() fails\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }

            ierr = SLOG_STM_2ndPass( slog );
            if ( ierr != SLOG_SUCCESS ) {
                fprintf( errfile, __FILE__":SLOG_CloseOutputStream() - "
                                  "SLOG_STM_2ndPass() fails\n" );
                fflush( errfile );
                return SLOG_FAIL;
            }
        }   /*  Endof  if ( slog->hdr->IsIncreasingEndtime == SLOG_TRUE )  */

        /*
            Check if all the pseudo record holes has been patched.
            i.e. check if the SLOG_Irec_ReserveSpace() and