path: root/src/stable/myriad.cpp

#include "bu/myriad.h"
#include "bu/myriadstream.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, \
        "Insufficent data reading myriad data from backing 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 )
{
    if( !rBacking.isSeekable() )
    {
        throw Bu::MyriadException( Bu::MyriadException::invalidBackingStream,
                "Myriad backing stream must be random access (seekable).");
    }
    if( !loadMyriad() )
    {
        createMyriad( iBlockSize, iPreallocateBlocks );
    }
}

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

Bu::MyriadStream Bu::Myriad::create( Bu::Myriad::Mode /*eMode*/,
        int32_t /*iPreallocateBytes*/ )
{
    return Bu::MyriadStream( *this, NULL, (Mode)0 );
}

Bu::MyriadStream Bu::Myriad::open( Bu::Myriad::StreamId iStream,
        Bu::Myriad::Mode eMode )
{
    Bu::MutexLocker l( mAccess );
    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.");
        }
    }
    return Bu::MyriadStream( *this, hStream.get( iStream ), eMode );
}

bool Bu::Myriad::erase( Bu::Myriad::StreamId /*iStream*/ )
{
    return false;
}

bool Bu::Myriad::setSize( Bu::Myriad::StreamId /*iStream*/,
        int32_t /*iNewSize*/ )
{
    return false;
}

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() );
}

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 );
    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.");
    }
    int32_t iHeaderStreamBytes;
    MyriadRead( &iHeaderStreamBytes, 4 );

    Stream *pHeaderStream = new Stream( *this, iStream, iHeaderStreamBytes );
    int iHeaderStreamBlocks = blkDiv(iHeaderStreamBytes+4, iBlockSize );

    while( iHeaderStreamBytes+(iHeaderStreamBlocks*4)
           > iHeaderStreamBlocks*iBlockSize )
    {
        iHeaderStreamBlocks = blkDiv(
            (iHeaderStreamBytes+((iHeaderStreamBlocks+1)*4)), iBlockSize
            );
    }

    for( int32_t j = 0; j < iHeaderStreamBlocks; j++ )
    {
        int32_t iBlockIndex;
        MyriadRead( &iBlockIndex, 4 );
        pHeaderStream->aBlocks.append( iBlockIndex );
    }

    // Bootstrap now using the header stream to read the rest of the data.

    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.");
}
/*
    struct {
        char sMagicCode[4];
        uint8_t uVer;
        uint8_t uBitsPerInt;
        uint32_t uBlockSize;
        uint32_t uStreamCount;
    } sHeader;

    struct {
        uint32_t uStreamId;
        uint32_t uStreamSize;
    } sStreamHeader;
    
    Bu::println("sHeader = %1, sStreamHeader = %2").arg( sizeof(sHeader) ).arg( sizeof(sStreamHeader) );
*/

    // 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 )
    {
        rBacking.setSize( iBlockSize*iPreallocateBlocks );
    }
    else
    {
        rBacking.setSize( iBlockSize*iHeaderStreamBlocks );
    }

    //
    // 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 );
    }
}

int32_t Bu::Myriad::allocateBlock()
{
    Bu::MutexLocker l( mAccess );
    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 );
    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( mAccess );
    hStream.get( id )->open();
}

void Bu::Myriad::closeStream( StreamId id )
{
    Bu::MutexLocker l( mAccess );
    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 ),
    bStructureChanged( false )
{
}

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 );
    if( iNewSize < iSize )
    {
        // Shrink it
        int iNewBlocks = Bu::blkDiv( iNewSize, rParent.iBlockSize );
        while( aBlocks.getSize() > iNewBlocks )
        {
            rParent.releaseBlock( aBlocks.last(), false );
            aBlocks.eraseLast();
        }
        iSize = iNewSize;
    }
    else if( iNewSize > iSize )
    {
        // Grow it
        int iNewBlocks = Bu::blkDiv( iNewSize, rParent.iBlockSize );
        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 );
    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;
        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;
    }

    return iWrite;
}

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

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

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