path: root/src/stable/myriad.cpp

/*
 * Copyright (C) 2007-2012 Xagasoft, All rights reserved.
 *
 * This file is part of the libbu++ library and is released under the
 * terms of the license contained in the file LICENSE.
 */

#include "bu/myriad.h"
#include "bu/stream.h"
#include "bu/myriadstream.h"
#include "bu/mutexlocker.h"
#include <stdio.h>

#include "bu/sio.h"
using Bu::sio;
using Bu::Fmt;

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

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 &sStore, int iBlockSize, int iPreallocate ) :
	sStore( sStore ),
	iBlockSize( iBlockSize ),
	iBlocks( 0 ),
	iUsed( 0 ),
	bHeaderChanged( false )
{
	try
	{
		initialize();
	}
	catch( Bu::MyriadException &e )
	{
		if( e.getErrorCode() == MyriadException::emptyStream )
		{
			initialize( iBlockSize, iPreallocate );
		}
		else
		{
			throw;
		}
	}
}

Bu::Myriad::~Myriad()
{
	mActiveBlocks.lock();
	if( !hActiveBlocks.isEmpty() )
	{
		sio << "Bu::Myriad::~Myriad(): Error: There are "
			<< hActiveBlocks.getSize() << " unsynced blocks!" << sio.nl;
	}
	mActiveBlocks.unlock();
	sync();

	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		delete *i;
	}
}

void Bu::Myriad::sync()
{
	updateHeader();

	mActiveBlocks.lock();
	for( BlockHash::iterator i = hActiveBlocks.begin(); i; i++ )
	{
		if( (*i)->bChanged )
		{
			syncBlock( *i );
		}
	}
	mActiveBlocks.unlock();
}

void Bu::Myriad::initialize()
{
	MutexLocker mLock( mHeader );
	lFreeBlocks.clear();
	sStore.setPosEnd( 0 );
	Bu::size iSize = sStore.tell();
	sStore.setPos( 0 );

	unsigned char buf[4];
	if( sStore.read( buf, 4 ) < 4 )
		throw MyriadException( MyriadException::emptyStream,
				"Input stream appears to be empty.");
	if( memcmp( buf, Myriad_MAGIC_CODE, 4 ) )
	{
		throw MyriadException( MyriadException::invalidFormat,
			"Stream does not appear to be a valid Myriad format.");
	}
	sStore.read( buf, 2 );
	if( buf[0] != 1 )
		throw MyriadException( MyriadException::badVersion,
			"We can only handle version 1 for now.");
	if( buf[1] != 32 )
		throw MyriadException( MyriadException::invalidWordSize,
			"We can only handle 32-bit words at the moment.");
	sStore.read( &iBlockSize, 4 );
	int iStreams;
	sStore.read( &iStreams, 4 );

	iBlocks = iSize/iBlockSize;
	//sio << "Myriad: iSize=" << iSize << ", iBlockSize=" << iBlockSize
	//	<< ", iBlocks=" << iBlocks << ", iStreams=" << iStreams << sio.nl;

	int iHeaderSize = 14 + 8 + 4;
	int iHeaderBlocks = 0; //blkDiv( iHeaderSize+4, iBlockSize );

	while( iHeaderSize > iHeaderBlocks*iBlockSize )
	{
		iHeaderBlocks = blkDiv( iHeaderSize+4, iBlockSize );
		iHeaderSize = 14 + 8 + 4*iHeaderBlocks;
	}

	//sio << "Myriad: iHeaderSize=" << iHeaderSize
	//	<< ", iHeaderBlocks=" << iHeaderBlocks << sio.nl;

	Stream *pFakeHdr = new Stream;
	pFakeHdr->iId = 0;
	pFakeHdr->iSize = iHeaderSize;
	for( int j = 0; j < iHeaderBlocks; j++ )
	{
		pFakeHdr->aBlocks.append( j );
	}

//	sio << "Blocks: " << iBlocks << " (size = " << iSize << "/" << iBlockSize
//		<< ")" << sio.nl;
	Bu::BitString bsBlockUsed( iBlocks, false );
	bsBlockUsed.clear();

//	bool bCanSkip = false; // Can skip around, post initial header stream i/o
	MyriadStream *pIn = new MyriadStream( *this, pFakeHdr );
	pIn->setPos( sStore.tell() );
	for( int j = 0; j < iStreams; j++ )
	{
		aStreams.append( new Stream() );
		Stream &s = *aStreams[j];
		pIn->read( &s.iId, 4 );
		pIn->read( &s.iSize, 4 );
		int iSBlocks = blkDiv(s.iSize, iBlockSize);
	//	sio << "Myriad: - Stream::iId=" << s.iId
	//		<< ", Stream::iSize=" << s.iSize
	//		<< ", Stream::aBlocks=" << iSBlocks
	//		<< ", pIn->tell()=" << pIn->tell() << sio.nl;
		for( int k = 0; k < iSBlocks; k++ )
		{
			int iBId;
			pIn->read( &iBId, 4 );
	//		sio << "Myriad:   - iBId=" << iBId
	//			<< ", iStartPos=" << iBId*iBlockSize
	//			<< ", pIn->tell()=" << pIn->tell() << sio.nl;
			s.aBlocks.append( iBId );
			bsBlockUsed.setBit( iBId );
			iUsed++;
			if( (j == 0 && k == iHeaderBlocks-1) )
			{
	//			sio << "Myriad:   - End of prepartition, unlocking skipping."
	//				<< sio.nl;
//				bCanSkip = true;
				MyriadStream *pTmp = new MyriadStream( *this, aStreams[0] );
	//			sio << "Myriad    - Position = " << pIn->tell() << sio.nl;
				pTmp->setPos( pIn->tell() );
				delete pIn;
				delete pFakeHdr;
				pIn = pTmp;
			}
		}
	}
	delete pIn;

	for( int j = 0; j < iBlocks; j++ )
	{
		if( bsBlockUsed.getBit( j ) == false )
		{
//			sio << "Preinitialized block " << j << " is free." << sio.nl;
			lFreeBlocks.append( j );
		}
	}
//	sio << "Myriad: Blocks used: " << bsBlockUsed.toString() << sio.nl;
}

void Bu::Myriad::initialize( int iBlockSize, int iPreAllocate )
{
	MutexLocker mLock( mHeader );
	lFreeBlocks.clear();

	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		delete *i;
	}
	aStreams.clear();
	iUsed = 0;

	int iHeaderSize = 14 + 8 + 4;
	int iHeaderBlocks = 0; //blkDiv( iHeaderSize+4, iBlockSize );
	char cBuf = 1;
	int iBuf = 0;
	
	Stream *pStr = new Stream;
	pStr->iId = 0;

	while( iHeaderSize > iHeaderBlocks*iBlockSize )
	{
		iHeaderBlocks = blkDiv( iHeaderSize+4, iBlockSize );
		iHeaderSize = 14 + 8 + 4*iHeaderBlocks;
	}

	iPreAllocate += iHeaderBlocks;

	//sio << "Myriad: iHeaderSize=" << iHeaderSize << ", iBlockSize="
	//	<< iBlockSize << ", iHeaderBlocks=" << iHeaderBlocks << sio.nl;
	
//	bsBlockUsed.setSize( iPreAllocate, true );
	iUsed++;

	char *pBlock = new char[iBlockSize];
	memset( pBlock, 0, iBlockSize );
	for( int j = 0; j < iPreAllocate; j++ )
	{
		sStore.write( pBlock, iBlockSize );
	}
	delete[] (char *)pBlock;

	sStore.setPos( 0 );

	// Magic number
	sStore.write( Myriad_MAGIC_CODE, 4 );

	// Version (0)
	sStore.write( &cBuf, 1 );

	// Bits per int
	cBuf = 32;
	sStore.write( &cBuf, 1 );

	// The size of each block
	sStore.write( &iBlockSize, 4 );

	iBuf = 1;
	// The number of streams
	sStore.write( &iBuf, 4 );
	
	// Stream header
	iBuf = 0;
	sStore.write( &iBuf, 4 );
	sStore.write( &iHeaderSize, 4 );
	for( iBuf = 0; iBuf < iHeaderBlocks; iBuf++ )
	{
		sStore.write( &iBuf, 4 );
	}

	this->iBlockSize = iBlockSize;
	this->iBlocks = iPreAllocate;
	
	pStr->iSize = sStore.tell();
//	sio << "Myriad: Actual end of header stream = " << pStr->iSize << sio.nl;

	pStr->iSize = iHeaderSize;
	for( int j = 0; j < iHeaderBlocks; j++ )
	{
//		sio << "Started block " << j << " is header." << sio.nl;
		pStr->aBlocks.append( j );
//		bsBlockUsed.setBit( j );
		iUsed++;
	}
	for( int j = iHeaderBlocks; j < this->iBlocks; j++ )
	{
//		sio << "Started block " << j << " is free." << sio.nl;
		lFreeBlocks.append( j );
	}

	aStreams.append( pStr );

	//sio << bsBlockUsed.toString() << " - " << pStr->aBlocks << sio.nl;

	bHeaderChanged = true;
	//hStreams.insert( 0, BlockArray( 0 ) );
}

void Bu::Myriad::updateHeader()
{
	MutexLocker mLock( mHeader );

	if( bHeaderChanged == false )
		return;
	if( !sStore.canWrite() )
		return;

	char cBuf;
	int iBuf;

	//for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	//{
	//	sio << "Myriad: Stream " << Fmt(4) << (*i)->iId << ": " << (*i)->aBlocks << sio.nl;
	//}

	// Compute the new size of the header.
	int iHeaderSize = 14 + 8*aStreams.getSize();
//	sio << "Myriad: updateHeader: aStreams.getSize() = " << aStreams.getSize()
//		<< sio.nl;
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		iHeaderSize += 4*(*i)->aBlocks.getSize();
//		sio << "Myriad: updateHeader: (*i)->aBlocks.getSize() = "
//			<< (*i)->aBlocks.getSize() << sio.nl;
	}
	int iNewBlocks = blkDiv( iHeaderSize, iBlockSize );
	while( iNewBlocks > aStreams[0]->aBlocks.getSize() )
	{
		int iBlock = findEmptyBlock();
//		sio << "Myriad: updateHeader: Appending block " << iBlock
//			<< " to header." << sio.nl;
		aStreams[0]->aBlocks.append( iBlock );
//		bsBlockUsed.setBit( iBlock );
		iUsed++;
		iHeaderSize += 4;
		iNewBlocks = blkDiv( iHeaderSize, iBlockSize );
	}
	aStreams[0]->iSize = iHeaderSize;
//	sio << "Myriad: updateHeader: iHeaderSize=" << iHeaderSize
//		<< ", iNewBlocks=" << iNewBlocks << ", curBlocks="
//		<< aStreams[0]->aBlocks.getSize() << sio.nl;

	MyriadStream sHdr( *this, aStreams[0] );
	sHdr.write( Myriad_MAGIC_CODE, 4 );

	// Version (1)
	cBuf = 1;
	sHdr.write( &cBuf, 1 );

	// Bits per int
	cBuf = 32;
	sHdr.write( &cBuf, 1 );

	// The size of each block
	sHdr.write( &iBlockSize, 4 );

	iBuf = aStreams.getSize();
	// The number of streams
	sHdr.write( &iBuf, 4 );
	
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		sHdr.write( &(*i)->iId, 4 );
		sHdr.write( &(*i)->iSize, 4 );
		int iUsedBlocks = blkDiv( (*i)->iSize, iBlockSize );
//		for( BlockArray::iterator j = (*i)->aBlocks.begin(); j; j++ )
		for( int j = 0; j < iUsedBlocks; j++ )
		{
			sHdr.write( &(*i)->aBlocks[j], 4 );
		}
	}
	
	bHeaderChanged = false;
}

int Bu::Myriad::createStream( int iPreAllocate )
{
	MutexLocker mLock( mHeader );

	Stream *pStr = new Stream();
	pStr->iId = aStreams.last()->iId+1;
	//sio << "Myriad: New stream id=" << pStr->iId << ", iPreAllocate="
	//	<< iPreAllocate << sio.nl;
	pStr->iSize = 0;
	aStreams.append( pStr );

	for( int j = 0; j < iPreAllocate; j++ )
	{
		int iFreeBlock = findEmptyBlock();
//		sio << "Myriad: Adding block " << iFreeBlock << sio.nl;
		pStr->aBlocks.append( iFreeBlock );
//		bsBlockUsed.setBit( iFreeBlock );
		iUsed++;
	}

	bHeaderChanged = true;

	return pStr->iId;
}

int Bu::Myriad::createStreamWithId( int iId, int iPreAllocate )
{
	MutexLocker mLock( mHeader );

	try
	{
		findStream( iId );
		throw MyriadException( MyriadException::streamExists,
			"There is already a stream with the given id.");
	}
	catch( MyriadException &e )
	{
		Stream *pStr = new Stream();
		pStr->iId = iId;
		//sio << "Myriad: New stream id=" << pStr->iId << ", iPreAllocate="
		//	<< iPreAllocate << sio.nl;
		pStr->iSize = 0;
		if( aStreams.last()->iId < iId )
		{
			aStreams.append( pStr );
		}
		else
		{
			for( StreamArray::iterator i = aStreams.begin(); i; i++ )
			{
				if( (*i)->iId > iId )
				{
					aStreams.insert( i, pStr );
					break;
				}
			}
		}

		for( int j = 0; j < iPreAllocate; j++ )
		{
			int iFreeBlock = findEmptyBlock();
	//		sio << "Myriad: Adding block " << iFreeBlock << sio.nl;
			pStr->aBlocks.append( iFreeBlock );
//			bsBlockUsed.setBit( iFreeBlock );
			iUsed++;
		}

		bHeaderChanged = true;

		return pStr->iId;
	}
}

int Bu::Myriad::findEmptyBlock()
{
	bHeaderChanged = true;

	if( lFreeBlocks.isEmpty() )
	{
		sStore.setSize( (iBlocks+1)*(Bu::size)iBlockSize );
		return iBlocks++;
	}
	else
	{
		return lFreeBlocks.dequeue();
	}
}

void Bu::Myriad::deleteStream( int iId )
{
	MutexLocker mLock( mHeader );

	if( iId < 0 )
		throw MyriadException( MyriadException::invalidStreamId,
				"Invalid stream id.");
	if( iId == 0 )
		throw MyriadException( MyriadException::protectedStream,
				"You cannot delete stream zero, it is protected.");
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		if( (*i)->iId == iId )
		{
			Stream *pStream = *i;
			for( BlockArray::iterator j = pStream->aBlocks.begin(); j; j++ )
			{
				lFreeBlocks.append( *j );
//				bsBlockUsed.setBit( *j, false );
				iUsed--;
			}
			aStreams.erase( i );
			bHeaderChanged = true;
			delete pStream;
			return;
		}
	}
}

Bu::Array<int> Bu::Myriad::getStreamIds()
{
	MutexLocker mLock( mHeader );

	Bu::Array<int> aRet( aStreams.getSize() );
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		aRet.append( (*i)->iId );
	}

	return aRet;
}

int Bu::Myriad::getStreamSize( int iId )
{
	MutexLocker mLock( mHeader );

	return findStream( iId )->iSize;
}

bool Bu::Myriad::hasStream( int iId )
{
	MutexLocker mLock( mHeader );

	try
	{
		findStream( iId );
		return true;
	}catch(...)
	{
		return false;
	}
}

Bu::MyriadStream Bu::Myriad::openStream( int iId )
{
	MutexLocker mLock( mHeader );

	//sio << "Myriad:  Request to open stream: " << iId << sio.nl;
	return MyriadStream( *this, findStream( iId ) );
}

int Bu::Myriad::getNumStreams()
{
	MutexLocker mLock( mHeader );

	return aStreams.getSize();
}

int Bu::Myriad::getBlockSize()
{
	return iBlockSize;
}

int Bu::Myriad::getNumBlocks()
{
	return iBlocks;
}

int Bu::Myriad::getNumUsedBlocks()
{
	return iUsed;
}

Bu::size Bu::Myriad::getTotalUsedBytes()
{
	MutexLocker mLock( mHeader );

	Bu::size iTotalSize = 0;
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		iTotalSize += (*i)->iSize;
	}
	return iTotalSize;
}

Bu::size Bu::Myriad::getTotalUnusedBytes()
{
	MutexLocker mLock( mHeader );

	Bu::size iTotalSize = (iBlocks-iUsed)*iBlockSize;
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		iTotalSize += iBlockSize - ((Bu::size)(*i)->iSize%iBlockSize);
	}
	return iTotalSize;
}

Bu::size Bu::Myriad::getTotalUnusedBytes( int iFakeBlockSize )
{
	MutexLocker mLock( mHeader );

	Bu::size iTotalSize = (iBlocks-iUsed)*iFakeBlockSize;
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		iTotalSize += iFakeBlockSize - ((*i)->iSize%iFakeBlockSize);
	}
	return iTotalSize;
}

Bu::Myriad::Stream *Bu::Myriad::findStream( int iId )
{
	for( StreamArray::iterator i = aStreams.begin(); i; i++ )
	{
		if( (*i)->iId == iId )
			return *i;
	}

	throw MyriadException( MyriadException::noSuchStream,
		"The requested stream doesn't exist and cannot be opened." );

	return NULL;
}

Bu::Myriad::Block *Bu::Myriad::getBlock( int iBlock )
{
//	sio << "Myriad:  Reading block " << iBlock << ", bytes "
//		<< iBlockSize*iBlock << "-" << iBlockSize*(iBlock+1) << sio.nl;
	Block *pBlock = new Block;
	pBlock->pData = new char[iBlockSize];
	sStore.setPos( iBlockSize * (Bu::size)iBlock );
	sStore.read( pBlock->pData, iBlockSize );
	pBlock->bChanged = false;
	pBlock->iBlockIndex = iBlock;

	mActiveBlocks.lock();
	hActiveBlocks.insert( iBlock, pBlock );
	mActiveBlocks.unlock();

	return pBlock;
}

void Bu::Myriad::releaseBlock( Bu::Myriad::Block *pBlock )
{
	if( pBlock == NULL )
		return;
//	sio << "Myriad:  Releasing block " << pBlock->iBlockIndex << sio.nl;
	syncBlock( pBlock );
	mActiveBlocks.lock();
	hActiveBlocks.erase( pBlock->iBlockIndex );
	mActiveBlocks.unlock();

	delete[] pBlock->pData;
	delete pBlock;
}

void Bu::Myriad::syncBlock( Block *pBlock )
{
	if( pBlock->bChanged )
	{
//		sio << "Myriad:    - Block changed, writing back to stream." << sio.nl;
		sStore.setPos( iBlockSize * (Bu::size)pBlock->iBlockIndex );
		sStore.write( pBlock->pData, iBlockSize );
		pBlock->bChanged = false;
	}
}

int Bu::Myriad::streamAddBlock( Stream *pStream )
{
	MutexLocker mLock( mHeader );

	int iBlock = findEmptyBlock();
	pStream->aBlocks.append( iBlock );
//	bsBlockUsed.setBit( iBlock );
//	bHeaderChanged = true;
	iUsed++;
	return iBlock;
}

void Bu::Myriad::setStreamSize( Stream *pStream, long iSize )
{
	MutexLocker mLock( mHeader );

	if( pStream->iSize == iSize )
	{
		return;
	}
	else if( pStream->iSize > iSize )
	{
		// Shrink
		for( int iNewSize = pStream->aBlocks.getSize()*iBlockSize;
			 iNewSize-iBlockSize > iSize; iNewSize -= iBlockSize )
		{
//			if( bsBlockUsed.getBit( pStream->aBlocks.last() ) )
				iUsed--;
//			else
//				sio << "Unused block used in stream? " << pStream->aBlocks.last() << sio.nl;
			lFreeBlocks.enqueue( pStream->aBlocks.last() );
//			bsBlockUsed.setBit( pStream->aBlocks.last(), false );
			pStream->aBlocks.eraseLast();
		}
		pStream->iSize = iSize;
		bHeaderChanged = true;
	}
	else
	{
		// Grow
		for( int iNewSize = pStream->aBlocks.getSize()*iBlockSize;
			 iNewSize < iSize; iNewSize += iBlockSize )
		{
			//streamAddBlock( pStream );
			int iBlock = findEmptyBlock();
			pStream->aBlocks.append( iBlock );
//			bsBlockUsed.setBit( iBlock );
//			bHeaderChanged = true;
			iUsed++;
		}
		pStream->iSize = iSize;
		bHeaderChanged = true;
	}
}

void Bu::Myriad::headerChanged()
{
	bHeaderChanged = true;
}

bool Bu::Myriad::isMyriad( Bu::Stream &sStore )
{
	uint8_t uTmp;

	return isMyriad( sStore, uTmp );
}

bool Bu::Myriad::isMyriad( Bu::Stream &sStore, uint8_t &uTmp )
{
	sStore.setPos( 0 );

	unsigned char buf[4];
	if( sStore.read( buf, 4 ) < 4 )
		throw MyriadException( MyriadException::emptyStream,
				"Input stream appears to be empty.");
	sStore.read( &uTmp, 1 );
	sStore.setPos( 0 );
	if( memcmp( buf, Myriad_MAGIC_CODE, 4 ) )
	{
		return false;
	}
	return true;
}

const Bu::BitString Bu::Myriad::getBlocksUsed() const
{
	Bu::BitString bs( iBlocks, false );
	for( int j = 0; j < iBlocks; j++ )
		bs.setBit( j );
	for( IndexList::const_iterator i = lFreeBlocks.begin(); i; i++ )
		bs.setBit( *i, false );
	return bs;
}