streambtreecontroller.h

b tree code for index in the database design

//////////////////////////////////////////////////////////////////
///
/// (C) 2007: ScalingWeb.com
///
/// Author: Anton Fedoruk <[e-mail]>
///
//////////////////////////////////////////////////////////////////

#ifndef __StreamBTreeController_h__
#define __StreamBTreeController_h__

#include <set>
#include <QSet>
#include <QMap>
#include <map>

#include "BTreeNode.h"
#include "JStream.h"
#include "JBufferedOutputStream.h"
#include "JBufferedInputStream.h"

const int constRemovedAddr = sizeof( int )*2;

///
/// BTreeController with stream as storage implementation
///

template< int NodeSize, typename KeyT, typename DataT >
class StreamBTreeController
{
public:

	typedef BTreeNode< NodeSize, KeyT, DataT > NodeType;

	struct CacheData
	{
		CacheData():
		node( 0 )
		{}

		CacheData( time_t lastAccess ):
		node( 0 )
		{}

		CacheData( NodeType *innode ):
		node( innode )
		{}

		NodeType *node;
	};

	typedef std::set< int > FreeChunksSet;
	typedef QMap< int, CacheData > ChunkCacheMap;

	StreamBTreeController();

	///
	/// Open stream-based BTree storage
	bool open( JStream *stor, guint maxCacheSize, JStream *journal = 0 );

	///
	/// Returns true if tree is opened
	bool isOpen();

	///
	/// Flush changes without cache releasing
	void flushChanges();

	///
	/// Release all cache if size of cache less then max accepted
	void releaseCache();

	///
	/// Clear all cached node, except root node
	void clearCache();

	///
	/// Release only one node
	void releaseNode( int addr );

	///
	/// Return current cache size
	guint cacheSize() const;

	///
	/// Set max accepted cache size
	void setMaxCacheSize( guint maxSize );

	///
	/// Return max accepted cache size
	guint getMaxCacheSize();

	///
	/// Returns true if node is in cache
	bool nodeInCache( int addr );

	///
	/// Returns internal storage pointer
	JStream *storage();

	///
	/// Clear tree fully
	bool clear();

	///
	/// Get root pointer
	typename NodeType* root();

	///
	/// Load node is specified
	bool loadNode( NodeType **node, int addr );

protected:

	// Storage related operations
	NodeType* newNode();
	bool allocFreeSpace();
	bool readAllOffsets( NodeType *node );
	void deleteNode( NodeType *node );
	void saveNode( NodeType *node );
	int rootAddr();
	void rootAddr( int addr );

	bool checkJournal();

	void closeController();

	// Data
	NodeType *root_;
	JStream *storage_;
	JStream *journal_;

	FreeChunksSet freeChunks_;
	ChunkCacheMap cache_;
	guint maxCacheSize_;
	guint rootAddr_;
};

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::StreamBTreeController

template< int NodeSize, typename KeyT, typename DataT >
StreamBTreeController< NodeSize, KeyT, DataT >::StreamBTreeController() : 
root_( 0 ),
storage_( 0 ),
journal_( 0 ),
maxCacheSize_( 0 ),
rootAddr_( 0 )
{}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::open

template< int NodeSize, typename KeyT, typename DataT >
bool StreamBTreeController< NodeSize, KeyT, DataT >::open( JStream *stor, 
								guint maxCacheSize, JStream *journal )
{
	if ( storage_ )
	{
		clearCache();
		delete root_;
		root_ = 0;
		closeController();
	}

	storage_ = stor;
	journal_ = journal;
	maxCacheSize_ = maxCacheSize;

	int sig = 0x77377377;

	if ( !checkJournal() )
	{
		int addr = rootAddr();
		if ( addr )
		{
			if ( !loadNode( &root_, addr ) )
			{
				return false;
			}
		}

		if ( !allocFreeSpace() ) return false;
	}

	if ( 0 == storage_->size() )
	{
		storage_->write( ( char* ) &sig, sizeof( sig ) );
		sig = 0;

		// Root addr
		storage_->write( ( char* ) &sig, sizeof( sig ) );
		// Removed addr
		storage_->write( ( char* ) &sig, sizeof( sig ) );
	}
	else
	{
		storage_->seek( 0 );
		int fileSig = 0;
		storage_->read( ( char* ) &fileSig, sizeof( fileSig ) );

		if ( sig != fileSig )
		{
			storage_->close();
			return false;
		}

		// Load removed nodes
		guint removedListAddr = 0;
		storage_->seek( constRemovedAddr );
		if ( sizeof( removedListAddr ) != storage_->read( 
			( char* ) &removedListAddr, sizeof( removedListAddr ) ) )
		{
			return false;
		}

		if ( removedListAddr )
		{
			if ( removedListAddr > storage_->size() ) return false;
			guint rlSize = storage_->size() - removedListAddr;
			if ( rlSize )
			{
				storage_->seek( removedListAddr );
				QByteArray rblock;
				rblock.resize( rlSize );
				storage_->read( rblock.data(), rlSize );

				guint curAddr = 0;
				for ( guint i = 0; i < rlSize/sizeof( guint ); i++ )
				{
					::memcpy( ( char* ) &curAddr, rblock.data() + sizeof( guint )*i,
						sizeof( curAddr ) );

					freeChunks_.insert( curAddr );
				}

				storage_->resize( removedListAddr );
			}
		}

		int addr = rootAddr();
		if ( addr )
		{
			if ( !loadNode( &root_, addr ) )
			{
				return false;
			}
		}
	}

	return true;
}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::checkJournal

template< int NodeSize, typename KeyT, typename DataT >
bool StreamBTreeController< NodeSize, KeyT, DataT >::checkJournal()
{
	if ( !journal_ || 0 == journal_->size() ) return true;

	JBufferedInputStream< JStream > journal( journal_ );
	journal.seek( 0 );

	guint label = 0;
	if ( sizeof( label ) != 
		journal.read( ( char* ) &label, sizeof( label ) ) )
	{
		journal.setSource( 0 );
		journal_->resize( 0 );
		return false;
	}

	if ( ~0 != label )
	{
		journal.setSource( 0 );
		journal_->resize( 0 );
		return false;
	}

	// Root
	label = 0;
	if ( sizeof( label ) != 
		journal.read( ( char* ) &label, sizeof( label ) ) )
	{
		journal.setSource( 0 );
		journal_->resize( 0 );
		return false;
	}

	if ( label )
	{
		storage_->seek( sizeof( int ) );
		storage_->write( ( char* ) &label, sizeof( label ) );
	}

	// Nodes
	NodeType node;
	while ( !journal.atEnd() )
	{
		journal.read( ( char* ) &node, sizeof( node ) );

		// Move out the node
		storage_->seek( node.addr_ );
		storage_->write( ( char* ) &node, sizeof( node ) );
	}

	journal.setSource( 0 );
	journal_->resize( 0 );
	return false;
}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::isOpen

template< int NodeSize, typename KeyT, typename DataT >
bool StreamBTreeController< NodeSize, KeyT, DataT >::isOpen()
{
	return ( storage_ != 0 );
}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::newNode

template< int NodeSize, typename KeyT, typename DataT >
typename StreamBTreeController< NodeSize, KeyT, DataT >::NodeType* 
				StreamBTreeController< NodeSize, KeyT, DataT >::newNode()
{
	int addr = 0;

	if ( !freeChunks_.empty() )
	{
		freeChunks_.erase( freeChunks_.begin() );
	}
	else
	{
		addr = storage_->size();
	}

	NodeType *newnode = new NodeType();
	newnode->addr_ = addr;
	newnode->flags_ = NodeType::NodeChanged;

	CacheData cacheData( newnode );
	saveNode( newnode );
	cache_[ addr ] = cacheData;

	return newnode;
}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::allocFreeSpace

template< int NodeSize, typename KeyT, typename DataT >
bool StreamBTreeController< NodeSize, KeyT, DataT >::allocFreeSpace()
{
	gint64 fullSize = storage_->size() - sizeof( int )*3;
	if ( !fullSize ) return true;

	qint32 nodeCount = fullSize / sizeof( NodeType );
	gint64 offset = sizeof( int )*3;

	for ( int i = 0; i < nodeCount; i++ )
	{
		freeChunks_.insert( offset );
		offset += sizeof( NodeType );
	}

	if ( !root_ ) return false;

	if ( !readAllOffsets( root_ ) ) return false;
	freeChunks_.erase( root_->addr_ );
	releaseCache();
	return true;
}

//	===================================================================
//	StreamBTreeController< NodeSize, KeyT, DataT >::readAllOffsets

template< int NodeSize, typename KeyT, typename DataT >
bool StreamBTreeController< NodeSize, KeyT, DataT >::readAllOffsets( 
				NodeType *node )
{
	if ( !node ) return true;