path: root/src/stable/myriad.cpp

#include "bu/myriad.h"
#include "bu/myriadstream.h"

#include "bu/membuf.h"
#include "bu/mutexlocker.h"
#include "bu/util.h"

#include "bu/sio.h"

#define Myriad_MAGIC_CODE       ((unsigned char *)"\x0a\xd3\xfa\x84")

#define MyriadRead( target, size ) if( rBacking.read( target, size ) < size ) \
{ \
    throw Bu::MyriadException( Bu::MyriadException::invalidFormat, \
        "Insufficient data reading myriad data from backing stream."); \
} (void)0

#define ReqRead( stream, target, size ) if( stream.read( target, size ) < size ) \
{ \
    throw Bu::MyriadException( Bu::MyriadException::invalidFormat, \
        "Insufficient data reading from myriad stream."); \
} (void)0

namespace Bu
{
    subExceptionDef( MyriadException )
    template<typename t> t blkDiv( t total, t block ) {
        return (total/block)+((total%block==0)?(0):(1));
    }
}

Bu::Myriad::Myriad( Bu::Stream &rBacking, int iBlockSize,
        int iPreallocateBlocks ) :
    rBacking( rBacking ),
    iBlockSize( iBlockSize ),
    iBlockCount( 0 ),
    bIsNewStream( true ),
    bStructureChanged( false ),
    iLastUsedIndex( -1 )
{
    if( !rBacking.isSeekable() )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidBackingStream,
                "Myriad backing stream must be random access (seekable).");
    }
    if( rBacking.getSize() == 0 )
    {
        createMyriad( iBlockSize, iPreallocateBlocks );
    }
    else
    {
        loadMyriad();
    }
}

Bu::Myriad::~Myriad()
{
    writeHeader();
}

Bu::MyriadStream Bu::Myriad::create( Bu::Myriad::Mode eMode,
        int32_t iPreallocateBytes )
{
    Bu::MutexLocker l( mAccess );

    Stream *pStream = new Stream( *this, ++iLastUsedIndex, 0 );
    int iBlocks = std::max(1, blkDiv( iPreallocateBytes, iBlockSize ));
    for( int j = 0; j < iBlocks; j++ )
    {
        pStream->aBlocks.append( __allocateBlock() );
    }
    mhStream.lock();
    hStream.insert( pStream->iStream, pStream );
    mhStream.unlock();
    bStructureChanged = true;

    return Bu::MyriadStream( *this, pStream, eMode&ReadWrite );
}

Bu::MyriadStream Bu::Myriad::open( Bu::Myriad::StreamId iStream,
        Bu::Myriad::Mode eMode )
{
    Stream *pStream = NULL;
    Bu::MutexLocker l( mhStream );
    if( (eMode&Create) )
    {
        if( hStream.has( iStream ) )
        {
            if( (eMode&Exclusive) )
            {
                throw Bu::MyriadException( MyriadException::noSuchStream,
                    "Stream exists.");
            }
        }
        else
        {
            Bu::MutexLocker l( mAccess );
            if( iStream >= iLastUsedIndex )
            {
                iLastUsedIndex = iStream;
            }
            pStream = new Stream( *this, iStream, 0 );
            pStream->aBlocks.append( __allocateBlock() );
            hStream.insert( pStream->iStream, pStream );
            bStructureChanged = true;
        }
    }
    if( !hStream.has( iStream ) )
    {
        throw Bu::MyriadException( MyriadException::noSuchStream,
            "No such stream.");
    }
    {
        Bu::MutexLocker l2( mBacking );
        if( (eMode&Write) && !rBacking.isWritable() )
        {
            throw Bu::MyriadException( MyriadException::badMode,
                "Backing stream does not support writing.");
        }
    }
    if( pStream == NULL )
    {
        pStream = hStream.get( iStream );
    }
    if( (eMode&Truncate) )
    {
        pStream->setSize( 0 );
    }
    return Bu::MyriadStream( *this, pStream, eMode );
}

Bu::Myriad::StreamId Bu::Myriad::allocate()
{
    Bu::MutexLocker l( mAccess );

    Stream *pStream = new Stream( *this, ++iLastUsedIndex, 0 );
    mhStream.lock();
    hStream.insert( pStream->iStream, pStream );
    mhStream.unlock();
    bStructureChanged = true;

    return pStream->iStream;
}

void Bu::Myriad::erase( Bu::Myriad::StreamId iStream )
{
    // For now, let's prevent you from erasing a stream if it's open.
    Bu::MutexLocker l( mhStream );
    if( !hStream.has( iStream ) )
    {
        throw Bu::MyriadException( Bu::MyriadException::noSuchStream,
                "No such stream exists.");
    }
    Stream *pStream = hStream.get( iStream );
    Bu::MutexLocker sl( pStream->mAccess );
    if( pStream->iOpenCount > 0 )
    {
        throw Bu::MyriadException( Bu::MyriadException::streamOpen,
                "Cannot currently erase a stream while it is open.");
    }

    for( Bu::Array<int32_t>::iterator i = pStream->aBlocks.begin(); i; i++ )
    {
        releaseBlock( *i, false );
    }
    pStream->aBlocks.clear();
    hStream.erase( iStream );
    delete pStream;
}

void Bu::Myriad::setSize( Bu::Myriad::StreamId iStream,
        int32_t iNewSize )
{
    Stream *pStream;
    {
        Bu::MutexLocker l( mhStream );
        pStream = hStream.get( iStream );
    }
    pStream->setSize( iNewSize );
}

int32_t Bu::Myriad::getSize( StreamId iStream ) const
{
    Bu::MutexLocker l( mhStream );
    return hStream.get( iStream )->getSize();
}

bool Bu::Myriad::exists( StreamId iStream ) const
{
    Bu::MutexLocker l( mhStream );
    return hStream.has( iStream );
}

Bu::String Bu::Myriad::getLocation() const
{
    Bu::MutexLocker l( mAccess );
    Bu::MutexLocker l2( mBacking );
    return Bu::String("myriad(%1,%2):%3")
        .arg( 1 ).arg( iBlockSize ).arg( rBacking.getLocation() );
}

int32_t Bu::Myriad::getBlockSize() const
{
    Bu::MutexLocker l( mAccess );
    return iBlockSize;
}

int32_t Bu::Myriad::getTotalBlocks() const
{
    Bu::MutexLocker l( mAccess );
    return iBlockCount;
}

int32_t Bu::Myriad::getUsedBlocks() const
{
    Bu::MutexLocker l( mAccess );
    return iBlockCount-lFreeBlocks.getSize();
}

int32_t Bu::Myriad::getFreeBlocks() const
{
    Bu::MutexLocker l( mAccess );
    return lFreeBlocks.getSize();
}

int32_t Bu::Myriad::getTotalStreams() const
{
    Bu::MutexLocker l( mhStream );
    return hStream.getSize();
}

int32_t Bu::Myriad::getTotalUsedBytes() const
{
    Bu::MutexLocker l( mhStream );
    int32_t iTotal = 0;
    for( StreamHash::const_iterator i = hStream.begin(); i; i++ )
    {
        iTotal += i.getValue()->getSize();
    }

    return iTotal;
}

int32_t Bu::Myriad::getTotalUnusedBytes(int32_t iAssumeBlockSize ) const
{
    if( iAssumeBlockSize < 0 )
    {
        iAssumeBlockSize = getBlockSize();
    }
    int32_t iTotal = 0;
    {
        Bu::MutexLocker l( mhStream );
        for( StreamHash::const_iterator i = hStream.begin(); i; i++ )
        {
            if( (i.getValue()->getSize()%iAssumeBlockSize) > 0 )
                iTotal += iBlockSize-(i.getValue()->getSize()%iAssumeBlockSize);
        }
    }

    {
        Bu::MutexLocker l( mAccess );
        iTotal += lFreeBlocks.getSize()*iBlockSize;
    }

    return iTotal;
}

Bu::Myriad::StreamIdList Bu::Myriad::getStreamList() const
{
    mhStream.lock();
    StreamIdList lIds = hStream.getKeys();
    mhStream.unlock();
    lIds.sort();
    if( lIds.first() == 0 )
    {
        lIds.eraseFirst();
    }
    return lIds;
}

Bu::BitString Bu::Myriad::buildBlockUseMap() const
{
    Bu::MutexLocker l( mAccess );
    Bu::BitString bsMap( iBlockCount );
    bsMap.fill();
    for( IndexList::const_iterator i = lFreeBlocks.begin(); i; i++ )
    {
        bsMap.setBit( *i, false );
    }
    return bsMap;
}

Bu::Myriad::StreamIdArray Bu::Myriad::buildBlockMap() const
{
    Bu::MutexLocker l( mAccess );
    StreamIdArray bm( iBlockCount );
    for( int j = 0; j < iBlockCount; j++ )
    {
        bm.append( -1 );
    }
    Bu::MutexLocker l2( mhStream );
    for( StreamHash::const_iterator iStream = hStream.begin();
         iStream; iStream++ )
    {
        int32_t iId = iStream.getKey();
        Stream *pStream = iStream.getValue();
        for( Bu::Array<int32_t>::const_iterator iBlock =
             pStream->aBlocks.begin(); iBlock; iBlock++ )
        {
            bm[*iBlock] = iId;
        }
    }
    return bm;
}

void Bu::Myriad::sync()
{
    writeHeader();
}

bool Bu::Myriad::loadMyriad()
{
    //Bu::println("Load myriad!");
    char sMagicCode[4];
    rBacking.setPos( 0 );
    MyriadRead( sMagicCode, 4 );
    if( memcmp( sMagicCode, Myriad_MAGIC_CODE, 4 ) )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "Backing stream does not seem to be a Myriad structure.");
    }
    uint8_t uVer;
    uint8_t uBitsPerInt;
    MyriadRead( &uVer, 1 );
    if( uVer != 1 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "Only version 1 myriad structures are supported.");
    }
    MyriadRead( &uBitsPerInt, 1 );
    if( uBitsPerInt != 32 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "Only 32 bits per int are supported at this time.");
    }
    MyriadRead( &iBlockSize, 4 );

    iBlockCount = rBacking.getSize()/iBlockSize;
    if( (rBacking.getSize()%iBlockSize) != 0 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "Backing stream is not cleanly divisibly by the block size.");
    }

    Bu::Hash<int32_t,bool> hUnusedBlocks;
    for( int32_t j = 0; j < iBlockCount; j++ )
    {
        hUnusedBlocks.insert( j, true );
    }

    int iStreamCount;
    MyriadRead( &iStreamCount, 4 );

    //
    // Read stream data -- Bootstrap the zero stream
    //
    StreamId iStream;
    MyriadRead( &iStream, 4 );
    if( iStream != 0 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "The first stream defined must be the header/zero stream.");
    }
    iLastUsedIndex = iStream;
    int32_t iHeaderStreamBytes;
    MyriadRead( &iHeaderStreamBytes, 4 );

    Stream *pHeaderStream = new Stream( *this, iStream, iHeaderStreamBytes );
    hStream.insert( iStream, pHeaderStream );
    int iHeaderStreamBlocks = blkDiv(iHeaderStreamBytes, iBlockSize );
    MyriadStream sHeader( *this, pHeaderStream, Read );

    // We need to read enough so that we can gurantee that we're within a block
    // that we have read the index to, plus one index.
    for( int32_t j = 0; j < iHeaderStreamBlocks; j++ )
    {
        int32_t iBlockIndex;
        MyriadRead( &iBlockIndex, 4 );
        hUnusedBlocks.erase( iBlockIndex );
        pHeaderStream->aBlocks.append( iBlockIndex );
        if( rBacking.tell()+4 <= (j+1)*iBlockSize )
            break;
    }

    // Bootstrap now using the header stream to read the rest of the data.
    sHeader.setPos( rBacking.tell() );
    while( pHeaderStream->aBlocks.getSize() < iHeaderStreamBlocks )
    {
        int32_t iBlockIndex;
        ReqRead( sHeader, &iBlockIndex, 4 );
        hUnusedBlocks.erase( iBlockIndex );
        pHeaderStream->aBlocks.append( iBlockIndex );
    }

    // Ok, now we can read the rest of the header in.
    for( int j = 1; j < iStreamCount; j++ )
    {
        int32_t iStreamBytes;
        ReqRead( sHeader, &iStream, 4 );
        ReqRead( sHeader, &iStreamBytes, 4 );
        Stream *pStream = new Stream( *this, iStream, iStreamBytes );
        int32_t iBlocks = blkDiv(iStreamBytes, iBlockSize );
        for( int k = 0; k < iBlocks; k++ )
        {
            int32_t iBlockIndex;
            ReqRead( sHeader, &iBlockIndex, 4 );
            hUnusedBlocks.erase( iBlockIndex );
            pStream->aBlocks.append( iBlockIndex );
        }
        hStream.insert( iStream, pStream );
        if( iLastUsedIndex < iStream )
            iLastUsedIndex = iStream;
    }

    lFreeBlocks = hUnusedBlocks.getKeys();
    //Bu::println("Free blocks: %1").arg( lFreeBlocks.getSize() );
    
    bIsNewStream = false;

    return true;
}

void Bu::Myriad::createMyriad( int32_t iBlockSize, int32_t iPreallocateBlocks )
{
    if( iBlockSize < 8 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidParameter,
                "iBlockSize cannot be below 8");
    }
    if( rBacking.getSize() )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "Backing stream contains data, but not a myriad structure.");
    }

    // Start with the bytes for the file header and initial stream header
    int iHeaderStreamBytes
        = 14 // Base header
        + 8; // Stream header
             
    // Pick the block count that matches our current estimate for the header
    // plus one block index.
    int iHeaderStreamBlocks = blkDiv(iHeaderStreamBytes+4, iBlockSize );

    //Bu::println("Initial estimate: %1 bytes / %2 cur blocks, %3 computed blocks (%4 target bytes).").arg( iHeaderStreamBytes+(iHeaderStreamBlocks*4) ).arg( iHeaderStreamBlocks ).arg( blkDiv((iHeaderStreamBytes+(iHeaderStreamBlocks*4)), iBlockSize) ).arg( iHeaderStreamBlocks*iBlockSize );
    while( iHeaderStreamBytes+(iHeaderStreamBlocks*4)
           > iHeaderStreamBlocks*iBlockSize )
    {
        iHeaderStreamBlocks = blkDiv((iHeaderStreamBytes+((iHeaderStreamBlocks+1)*4)), iBlockSize);
        if( iHeaderStreamBlocks > 100 )
            break;
        //Bu::println("      Adjustment: %1 bytes / %2 cur blocks, %3 computed blocks (%4 target bytes).").arg( iHeaderStreamBytes+(iHeaderStreamBlocks*4) ).arg( iHeaderStreamBlocks ).arg( blkDiv((iHeaderStreamBytes+(iHeaderStreamBlocks*4)), iBlockSize) ).arg( iHeaderStreamBlocks*iBlockSize );
    }

    if( iPreallocateBlocks < iHeaderStreamBlocks )
    {
        iPreallocateBlocks = iHeaderStreamBlocks;
    }
    rBacking.setSize( iBlockSize*iPreallocateBlocks );

    //
    // Write Myriad header
    //
    uint8_t uVer = 1;
    uint8_t uBpi = 32;
    int32_t iStreamCount = 1;
    rBacking.setPos( 0 );
    rBacking.write( Myriad_MAGIC_CODE, 4 );
    rBacking.write( &uVer, 1 );
    rBacking.write( &uBpi, 1 );
    rBacking.write( &iBlockSize, 4 );
    rBacking.write( &iStreamCount, 4 );

    Stream *pHeadStream = new Stream( *this, 0, Bu::Myriad::ReadWrite );
    //
    // Write stream header
    //
    uint32_t uStreamId = 0;
    uint32_t uStreamSize = iHeaderStreamBytes+iHeaderStreamBlocks*4;
    rBacking.write( &uStreamId, 4 );
    rBacking.write( &uStreamSize, 4 );
    for( int iBlockIndex = 0; iBlockIndex < iHeaderStreamBlocks; iBlockIndex++ )
    {
        rBacking.write( &iBlockIndex, 4 );
        pHeadStream->aBlocks.append( iBlockIndex );
    }
    rBacking.flush();

    hStream.insert( pHeadStream->iStream, pHeadStream );

    for( int32_t j = iHeaderStreamBlocks; j < iPreallocateBlocks; j++ )
    {
        lFreeBlocks.append( j );
    }
    iLastUsedIndex = 0;
    iBlockCount = iPreallocateBlocks;
}

void Bu::Myriad::writeHeader()
{
    Bu::MutexLocker l( mAccess );
    if( !rBacking.isWritable() )
        return;
    //Bu::println("Writing stream breakdown:");
    Bu::MemBuf mbHeader;
    {
        Bu::MutexLocker l2( mhStream );

        int32_t iHdrStreamSize = __calcHeaderSize();
        // Maybe just do stream surgery here.
        {
            Stream *psHeader = hStream.get( 0 );
            Bu::MutexLocker l2( psHeader->mAccess );
            int iNewBlocks = Bu::blkDiv( iHdrStreamSize, iBlockSize );
            if( iHdrStreamSize < psHeader->iSize )
            {
                while( psHeader->aBlocks.getSize() > iNewBlocks )
                {
                    __releaseBlock( psHeader->aBlocks.last(), false );
                    psHeader->aBlocks.eraseLast();
                }
            }
            else if( iHdrStreamSize > psHeader->iSize )
            {
                while( psHeader->aBlocks.getSize() < iNewBlocks )
                {
                    psHeader->aBlocks.append( __allocateBlock() );
                }
            }
            psHeader->iSize = iHdrStreamSize;
        }

        //Bu::println("Computed header size: %1 bytes. Ver=%2, Bpi=%3, BlockSize=%4").arg( iHdrStreamSize ).arg( 1 ).arg( 32 ).arg( iBlockSize );
        
        uint8_t uVer = 1;
        uint8_t uBpi = 32;
        int32_t iStreamCount = hStream.getSize();

        mbHeader.write( Myriad_MAGIC_CODE, 4 );
        mbHeader.write( &uVer, 1 );
        mbHeader.write( &uBpi, 1 );
        mbHeader.write( &iBlockSize, 4 );
        mbHeader.write( &iStreamCount, 4 );
        StreamHash::KeyList lStreamId = hStream.getKeys();
        lStreamId.sort();
        if( lStreamId.first() != 0 )
        {
            throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                    "There doesn't appear to be a zero (header) stream.");
        }
        for( StreamHash::KeyList::iterator i = lStreamId.begin(); i; i++ )
        {
            uint32_t uStreamId = *i;
            Stream *pStream = hStream.get( uStreamId );
            uint32_t uStreamSize = pStream->getSize();
            mbHeader.write( &uStreamId, 4 );
            mbHeader.write( &uStreamSize, 4 );
            int32_t iBlocks = Bu::blkDiv( uStreamSize, (uint32_t)iBlockSize );
            Bu::Array<int32_t> aBlocks = pStream->getBlockList();
            
            //Bu::println("  Stream %1 is %2 bytes %3 blocks (%4 blocks computed)").arg( *i ).arg( uStreamSize ).arg( aBlocks.getSize() ).arg( Bu::blkDiv( (int)uStreamSize, (int)iBlockSize ) );

//            for( Bu::Array<int32_t>::iterator i = aBlocks.begin(); i; i++ )
            for( int j = 0; j < iBlocks; j++ )
            {
                mbHeader.write( &aBlocks[j], 4 );
            }
        }


    }

    Bu::MyriadStream sHeader( *this, hStream.get( 0 ), Bu::Myriad::Write );
    sHeader.write( mbHeader.getString() );
    bStructureChanged = false;
}

int32_t Bu::Myriad::__calcHeaderSize()
{
    int32_t iHdrSize = 4+1+1+4+4;

    StreamHash::KeyList lStreamId = hStream.getKeys();
    lStreamId.sort();
    if( lStreamId.first() != 0 )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidFormat,
                "There doesn't appear to be a zero (header) stream.");
    }
    for( StreamHash::KeyList::iterator i = lStreamId.begin(); i; i++ )
    {
        iHdrSize += 4+4;
        int32_t iStreamSize = hStream.get( *i )->getSize();
        if( (*i) != 0 )
        {
            iHdrSize += Bu::blkDiv( iStreamSize, iBlockSize )*4;
        }
    }

    //Bu::println("HeaderCalc:");
    //Bu::println("  Base (no header stream): %1").arg( iHdrSize );
    int32_t iNewSize = iHdrSize;
    int32_t iOldSize;

    do {
        iOldSize = iNewSize;
        iNewSize = iHdrSize + Bu::blkDiv(iNewSize, iBlockSize)*4;
        //Bu::println("  Recomp: %1").arg( iNewSize );
    } while( iOldSize != iNewSize );

    return iNewSize;
}

int32_t Bu::Myriad::allocateBlock()
{
    Bu::MutexLocker l( mAccess );
    return __allocateBlock();
}

int32_t Bu::Myriad::__allocateBlock()
{
    bStructureChanged = true;
    if( lFreeBlocks.isEmpty() )
    {
        // Increase the size of the backing stream
        int32_t iIndex = iBlockCount++;
        rBacking.setSize( iBlockCount*iBlockSize );
        return iIndex;
    }
    else
    {
        // Provide an existing free block.
        return lFreeBlocks.peekPop();
    }
}

void Bu::Myriad::releaseBlock( int32_t iBlockId, bool bBlank )
{
    Bu::MutexLocker l( mAccess );
    __releaseBlock( iBlockId, bBlank );
}

void Bu::Myriad::__releaseBlock( int32_t iBlockId, bool bBlank )
{
    bStructureChanged = true;
    lFreeBlocks.append( iBlockId );
    if( bBlank )
    {
        blankBlock( iBlockId );
    }
}

void Bu::Myriad::blankBlock( int32_t iBlockId )
{
    Bu::MutexLocker l( mBacking );
    rBacking.setPos( iBlockId*iBlockSize );
    int32_t iChunk = std::min( iBlockSize, 4096 );
    uint8_t *pChunk = new uint8_t[iChunk];
    memset( pChunk, 0, iChunk );
    int iLeft = iBlockSize;
    while( iLeft > 0 )
    {
        int32_t iWrite = rBacking.write( pChunk, std::min( iChunk, iLeft ) );
        iLeft -= iWrite;
    }
    delete[] pChunk;
}

void Bu::Myriad::openStream( StreamId id )
{
    Bu::MutexLocker l( mhStream );
    hStream.get( id )->open();
}

void Bu::Myriad::closeStream( StreamId id )
{
    Bu::MutexLocker l( mhStream );
    hStream.get( id )->close();
}

int32_t Bu::Myriad::blockRead( int32_t iBlock, int32_t iStart,
        void *pTarget, int32_t iSize )
{
    int32_t iUpperSize = iBlockSize - (iStart%iBlockSize);
/*    Bu::println("Max read within block: %1 vs %2  (start=%3, blocksize=%4)")
        .arg( iUpperSize ).arg( iSize )
        .arg( iStart ).arg( iBlockSize );
*/
    int32_t iAmnt = std::min( iSize, iUpperSize );
    Bu::MutexLocker l( mBacking );
    rBacking.setPos( iBlockSize*iBlock + iStart );

    return rBacking.read( pTarget, iAmnt );
}

int32_t Bu::Myriad::blockWrite( int32_t iBlock, int32_t iStart,
        const void *pTarget, int32_t iSize )
{
    int32_t iUpperSize = iBlockSize - (iStart%iBlockSize);
/*    Bu::println("Max write within block: %1 vs %2  (start=%3, blocksize=%4)")
        .arg( iUpperSize ).arg( iSize )
        .arg( iStart ).arg( iBlockSize );
*/
    int32_t iAmnt = std::min( iSize, iUpperSize );
    Bu::MutexLocker l( mBacking );
    rBacking.setPos( iBlock*iBlockSize + iStart );

    return rBacking.write( pTarget, iAmnt );
}

/////////
// Bu::Myriad::Stream
//

Bu::Myriad::Stream::Stream( Bu::Myriad &rParent, Bu::Myriad::StreamId iStream,
        int32_t iSize ) :
    rParent( rParent ),
    iStream( iStream ),
    iSize( iSize ),
    iOpenCount( 0 )
{
}

Bu::Myriad::Stream::~Stream()
{
}

int32_t Bu::Myriad::Stream::getSize() const
{
    Bu::MutexLocker l( mAccess );
    return iSize;
}

int32_t Bu::Myriad::Stream::getBlockSize() const
{
    Bu::MutexLocker l( mAccess );
    return rParent.iBlockSize;
}

Bu::Myriad::StreamId Bu::Myriad::Stream::getStreamId() const
{
    return iStream;
}

int32_t Bu::Myriad::Stream::getOpenCount() const
{
    Bu::MutexLocker l( mAccess );
    return iOpenCount;
}

void Bu::Myriad::Stream::setSize( int32_t iNewSize )
{
    // Two possible modes, shrink or grow.
    Bu::MutexLocker l( mAccess );
    int iNewBlocks = Bu::blkDiv( iNewSize, rParent.iBlockSize );
    if( iNewSize < iSize )
    {
        // Shrink it
        while( aBlocks.getSize() > iNewBlocks )
        {
            rParent.releaseBlock( aBlocks.last(), false );
            aBlocks.eraseLast();
        }
        iSize = iNewSize;
    }
    else if( iNewSize > iSize )
    {
        // Grow it
        while( aBlocks.getSize() < iNewBlocks )
        {
            aBlocks.append( rParent.allocateBlock() );
        }
        iSize = iNewSize;
    }
}

int32_t Bu::Myriad::Stream::read( int32_t iStart, void *pTarget,
        int32_t iSize )
{
    int32_t iRead = 0;
    Bu::MutexLocker l( mAccess );

    if( iStart >= this->iSize )
        return 0;

    if( iStart+iSize >= this->iSize )
    {
        iSize = this->iSize-iStart;
    }

    while( iSize > 0 )
    {
        int32_t iBlock = aBlocks[iStart/rParent.iBlockSize];
        int32_t iChunkRead = rParent.blockRead(
                iBlock, iStart%rParent.iBlockSize, pTarget, iSize
                );
        if( iChunkRead == 0 )
            break;
        iRead += iChunkRead;
        iStart += iChunkRead;
        reinterpret_cast<ptrdiff_t &>(pTarget) += iChunkRead;
        iSize -= iChunkRead;
    }

    return iRead;
}

int32_t Bu::Myriad::Stream::write( int32_t iStart, const void *pTarget,
        int32_t iSize )
{
    int32_t iWrite = 0;
    Bu::MutexLocker l( mAccess );
    while( iSize > 0 )
    {
        int32_t iBlockIdx = iStart/rParent.iBlockSize;
        while( iBlockIdx >= aBlocks.getSize() )
        {
            aBlocks.append( rParent.allocateBlock() );
        }
        int32_t iBlock = aBlocks[iBlockIdx];
        int32_t iChunkWrite = rParent.blockWrite(
                iBlock, iStart%rParent.iBlockSize, pTarget, iSize
                );
        if( iChunkWrite == 0 )
            break;
        iWrite += iChunkWrite;
        iStart += iChunkWrite;
        reinterpret_cast<ptrdiff_t &>(pTarget) += iChunkWrite;
        iSize -= iChunkWrite;
    }
    if( this->iSize < iStart )
        this->iSize = iStart;

    return iWrite;
}

Bu::String Bu::Myriad::Stream::getLocation() const
{
    Bu::MutexLocker l( mAccess );
    return Bu::String("%1:stream %2")\
        .arg( rParent.getLocation() ).arg( iStream );
}

Bu::Array<int32_t> Bu::Myriad::Stream::getBlockList() const
{
    Bu::MutexLocker l( mAccess );
    return aBlocks.clone();
}

void Bu::Myriad::Stream::open()
{
    Bu::MutexLocker l( mAccess );
    iOpenCount++;
}

bool Bu::Myriad::Stream::close()
{
    Bu::MutexLocker l( mAccess );
    return (bool)(--iOpenCount);
}