Bdb packages | Design docs | Source docs | Guidelines | Recent releases
	Search | Site Map .

                 

---

/BdbTransfer/BdbPTChecksum.cc

Go to the documentation of this file.

//------------------------------------------------------------------------------
// File and Version Information:
//      $Id: BdbPTChecksum.cc,v 1.8 1999/07/27 00:03:53 svarovsk Exp $
//
// Description:
//      fast checksumming tool.
//      uses multiple threads and various buffer sizes.
//      optimal number of threads and size of buffer may vary on
//      different systems. run tests to determine them.
//
// Environment:
//      Software developed for the BaBar Detector at the SLAC B-Factory
//
// Author:
//      Gennadi S. Svarovski (svarovsk@slac.stanford.edu)
//          original author
//
// Copyright Information:
//      Copyright (C) 1999      Stanford Linear Accelerator Center
//
//------------------------------------------------------------------------------

#include "BaBar/BaBar.hh"
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>

#include "BdbTransfer/BdbPTosSpecific.h"
#include "BdbTransfer/BdbPThread.h"
#include "BdbTransfer/BdbPTMutex.h"
#include "BdbTransfer/BdbPTChecksumSum.h"
#include "BdbTransfer/BdbPTVerbose.h"

static char cvsid[]="$Id: BdbPTChecksum.cc,v 1.8 1999/07/27 00:03:53 svarovsk Exp $";

// these 2 variables are used by threads and main function
// array to hold checksums of all blocks
static BdbPTChecksumSum *checksums;
// needed for arithmetics
static unsigned extrashift = 0;

// class that does the job
class BdbPTChecksum : public BdbPThread {
    private:
        // arguments
        int num, total, started;
        int fd;
        const char *filename;
        ssize_t bufsize;

        // attribute for threads creation
        BdbPThreadAttr Attr;

    private:
        // mutex to synchronize access to file
        static BdbPTMutex& fileMutex();

    public:
        // default constructor
        BdbPTChecksum( );
        // seeting arguments for call
        int start( int num, int total, int started, int fd, 
                   const char* filename, ssize_t bufsize);
        
    private:
        virtual void* run( void );
};

// mutex to synchronize access to file between threads
BdbPTMutex& BdbPTChecksum::fileMutex() {
    static BdbPTMutex mutex;
    return mutex;
}

BdbPTChecksum::BdbPTChecksum( )
    : num( 0 ), total( 0 ), started( 0 ), fd( 0 ), filename( NULL ), bufsize( 0 )
{
    // make system schedule threads as separate processes
    Attr.setscope( PTHREAD_SCOPE_SYSTEM );
}

int BdbPTChecksum::start( int _num, int _total, int _started, int _fd,
                          const char* _filename, ssize_t _bufsize) {
    num      = _num;
    total    = _total;
    started  = _started;
    fd       = _fd;
    filename = _filename;
    bufsize  = _bufsize;

    return BdbPThread::start( Attr );
}

void* BdbPTChecksum::run( void ) {
    off_t offset;
    ssize_t len;
    unsigned i, blocknum;
    char *buffer;
    
    buffer = new char [bufsize];
    assert( buffer != NULL );

    // main loop across all blocks for this thread
    for( blocknum = num; blocknum < total; blocknum += started ) {
        offset = (off_t)bufsize * blocknum;

        VERBOSEMSG( 3, ("running ( thread=%d, block=%d, file=%s, size=%d )\n", 
                        num, blocknum, filename, bufsize) );

        // sum calculation
        fileMutex().lock();
        lseek( fd, offset, SEEK_SET );
        len = read( fd, buffer, bufsize );
        fileMutex().unlock();

        if( len < 0 ) {
            perror( "pread" );
            ::exit(1);
        }

        checksums[blocknum] = BdbPTChecksumSum::calculate( buffer, len, &extrashift );
        VERBOSEMSG( 2, ( "blockdone ( block=%d, checksum=%08X )\n", 
                         blocknum, checksums[blocknum].value() ) );
    }    

    VERBOSEMSG( 3, ( "finished ( thread=%d )\n", num ) );

    delete[] buffer;

    return NULL;
}    

void usage( void ) {
    printf("Usage: BdbPTChecksum [-v] [-t <threadnum>] [-b <buffersize>] <filename>\n");
    exit(1);
}

int main( int argc, char *argv[] ) {
    char options[] = "vt:b:h?";
    int o;
    char *filename;
    unsigned bufsize = 1024*1024, threadnum = 2, threadstart;
    unsigned extrablock;
    int fd;
    unsigned long count, i, blocks, extrablocksize;
    struct stat stat;
    
    if( argc < 2 ) usage();

    // command-line options processing
    while( ( o = getopt( argc, argv, options ) ) != EOF ) {
        switch ( o ) {
            case 'v':
                VERBOSEINCREASE();
                break;
            case 't':
                threadnum = atoi( optarg );
                break;
            case 'b':
                bufsize   = atoi( optarg );
                break;
            default:
                usage();
        }
    }

    // must have at least one argument - filename
    if( optind >= argc ) usage();
    
    filename = argv[optind];

    if( threadnum < 1 ) {
        printf( "illegal thread number %d. must be at least 1\n", threadnum );
        exit(1);
    }

    // requirement for checksumming algorithm
    if( bufsize % 128 != 0 || bufsize < 1 ) {
        printf("buffersize must be more than 1 and multiply of 128\n");
        exit(1);
    }

    VERBOSEMSG( 2, ( "Called with ( threadnum=%d, file=%s, size=%d )\n", 
                     threadnum, filename, bufsize ) );

    if( ( fd = open( filename, O_RDONLY ) ) == -1 ) {
        perror( filename );
        exit(1);
    }

    if( fstat( fd, &stat ) == -1 ) {
        perror( "fstat" );
        exit(1);
    }

    // need to know how many full blocks in file 
    // and if there is a non-full block at the end
    extrablock = 0;
    blocks = stat.st_size / bufsize;
    extrablocksize = stat.st_size % bufsize;
    if( extrablocksize != 0 )
        extrablock = 1;

    checksums = new BdbPTChecksumSum [ blocks+extrablock ];
    assert( checksums != NULL );

    // start minimum of requested thread number and number of blocks threads
    threadstart = ( threadnum < blocks+extrablock ) ? threadnum : blocks+extrablock;
    VERBOSEMSG( 2, ( "need to run %ld blocks, %d threads. will run %d at a time\n", 
                     blocks + extrablock, threadnum, threadstart ) );

    BdbPTChecksum* Threads = new BdbPTChecksum [threadstart];

    // loop to start all threads
    for( i = 0; i < threadstart; i++ )
        Threads[i].start( i, blocks+extrablock, threadstart, fd, filename, bufsize);

    // loop to wait all them finish
    for( i = 0; i < threadstart; i++ )
        Threads[i].join(NULL);

    delete[] Threads;

    // all done with threads. now need to join all subsums together

    // adding full block sums together
    // no shift needed since blocksize is multiple of 128
    BdbPTChecksumSum sum;
    for( i = 0; i < blocks; i++ ) {
        sum += checksums[i];
        VERBOSEMSG( 3, ( "adding block %d checksum %08X (result: %08X)\n", 
                         i, checksums[i].value(), sum.value() ) );
    }
    
    // adding checksum for last (not full) block - shift required
    // thread that processed last block 
    // was supposed to initialize 'extrashift' variable
    if( extrablock ) {
        // non-trivial assumption on how many shifts we need :)
        int shiftnumber = (extrablocksize/4+extrashift) % 32;
        VERBOSEMSG( 3, ( "shifting %08X %d times", sum.value(), shiftnumber ) );

        sum.rotate( shiftnumber );
        VERBOSEMSG( 3, ( " (result %08X)\n", sum.value() ) );

        // adding last checksum at last
        sum += checksums[blocks];
        VERBOSEMSG( 3, ( "adding block %d checksum %08X (result: %08X)\n",
                    blocks, checksums[blocks].value(), sum.value() ) );
    }

    delete[] checksums;

    printf( "%ld+%d blocks of file %s read\n", 
            blocks, extrablock, filename );

    printf( "%08X\n", sum.value() );

    return 0;
}

 

---

BaBar Public Site | SLAC | News | Links | Who's Who | Contact Us

Page Owner: Jacek Becla
Last Update: October 04, 2002