mavibot/src/main/java/org/apache/directory/mavibot/btree/AbstractBTree.java - directory-mavibot - Git at Google

 /*
  *  Licensed to the Apache Software Foundation (ASF) under one
  *  or more contributor license agreements.  See the NOTICE file
  *  distributed with this work for additional information
  *  regarding copyright ownership.  The ASF licenses this file
  *  to you under the Apache License, Version 2.0 (the
  *  "License"); you may not use this file except in compliance
  *  with the License.  You may obtain a copy of the License at
  *
  *    http://www.apache.org/licenses/LICENSE-2.0
  *
  *  Unless required by applicable law or agreed to in writing,
  *  software distributed under the License is distributed on an
  *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  *  KIND, either express or implied.  See the License for the
  *  specific language governing permissions and limitations
  *  under the License.
  *
  */
 package org.apache.directory.mavibot.btree;


 import java.io.IOException;
 import java.lang.reflect.Array;
 import java.util.Comparator;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.locks.ReentrantLock;

 import org.apache.directory.mavibot.btree.exception.KeyNotFoundException;
 import org.apache.directory.mavibot.btree.serializer.ElementSerializer;


 /**
  * A BTree abstract class containing the methods shared by the PersistedBTree or the InMemoryBTree
  * implementations.
  *
  * @param <K> The Key type
  * @param <V> The Value type
  *
  * @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
  */
 /* No qualifier*/abstract class AbstractBTree<K, V> implements BTree<K, V>
 {
     /** The read transaction timeout */
     protected long readTimeOut = DEFAULT_READ_TIMEOUT;

     /** The Header for a managed BTree */
     protected BTreeHeader btreeHeader;

     /** The current rootPage */
     protected volatile Page<K, V> rootPage;

     /** The Key serializer used for this tree.*/
     protected ElementSerializer<K> keySerializer;

     /** The Value serializer used for this tree. */
     protected ElementSerializer<V> valueSerializer;

     /** The list of read transactions being executed */
     protected ConcurrentLinkedQueue<ReadTransaction<K, V>> readTransactions;

     /** The size of the buffer used to write data in disk */
     protected int writeBufferSize;

     /** A lock used to protect the write operation against concurrent access */
     protected ReentrantLock writeLock;

     /** Flag to enable duplicate key support */
     private boolean allowDuplicates;

     /** The thread responsible for the cleanup of timed out reads */
     protected Thread readTransactionsThread;

     /** The BTree type : either in-memory, disk backed or persisted */
     private BTreeTypeEnum type;

     /** The current transaction */
     protected WriteTransaction writeTransaction;


     /**
      * Starts a Read Only transaction. If the transaction is not closed, it will be
      * automatically closed after the timeout
      *
      * @return The created transaction
      */
     protected ReadTransaction<K, V> beginReadTransaction()
     {
         ReadTransaction<K, V> readTransaction = new ReadTransaction<K, V>( rootPage, btreeHeader.getRevision() - 1,
             System.currentTimeMillis() );

         readTransactions.add( readTransaction );

         return readTransaction;
     }


     /**
      * {@inheritDoc}
      */
     public TupleCursor<K, V> browse() throws IOException
     {
         ReadTransaction<K, V> transaction = beginReadTransaction();

         // Fetch the root page for this revision
         ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

         TupleCursor<K, V> cursor = rootPage.browse( transaction, stack, 0 );

         // Set the position before the first element
         cursor.beforeFirst();

         return cursor;
     }


     /**
      * {@inheritDoc}
      */
     public TupleCursor<K, V> browse( long revision ) throws IOException, KeyNotFoundException
     {
         ReadTransaction<K, V> transaction = beginReadTransaction();

         // Fetch the root page for this revision
         Page<K, V> revisionRootPage = getRootPage( revision );

         ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

         // And get the cursor
         TupleCursor<K, V> cursor = revisionRootPage.browse( transaction, stack, 0 );

         return cursor;
     }


     /**
      * {@inheritDoc}
      */
     public TupleCursor<K, V> browseFrom( K key ) throws IOException
     {
         ReadTransaction<K, V> transaction = beginReadTransaction();

         // Fetch the root page for this revision
         ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

         TupleCursor<K, V> cursor = rootPage.browse( key, transaction, stack, 0 );

         return cursor;
     }


     /**
      * {@inheritDoc}
      */
     public TupleCursor<K, V> browseFrom( long revision, K key ) throws IOException, KeyNotFoundException
     {
         ReadTransaction<K, V> transaction = beginReadTransaction();

         // Fetch the rootPage for this revision
         Page<K, V> revisionRootPage = getRootPage( revision );

         ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

         // And get the cursor
         TupleCursor<K, V> cursor = revisionRootPage.browse( key, transaction, stack, 0 );

         return cursor;
     }


     /**
      * {@inheritDoc}
      */
     public boolean contains( K key, V value ) throws IOException
     {
         return rootPage.contains( key, value );
     }


     /**
      * {@inheritDoc}
      */
     public boolean contains( long revision, K key, V value ) throws IOException, KeyNotFoundException
     {
         // Fetch the root page for this revision
         Page<K, V> revisionRootPage = getRootPage( revision );

         return revisionRootPage.contains( key, value );
     }


     /**
      * {@inheritDoc}
      */
     public Tuple<K, V> delete( K key ) throws IOException
     {
         if ( key == null )
         {
             throw new IllegalArgumentException( "Key must not be null" );
         }

         long revision = generateRevision();

         Tuple<K, V> deleted = delete( key, revision );

         return deleted;
     }


     /**
      * {@inheritDoc}
      */
     public Tuple<K, V> delete( K key, V value ) throws IOException
     {
         if ( key == null )
         {
             throw new IllegalArgumentException( "Key must not be null" );
         }

         if ( value == null )
         {
             throw new IllegalArgumentException( "Value must not be null" );
         }

         long revision = generateRevision();

         Tuple<K, V> deleted = delete( key, value, revision );

         return deleted;
     }


     /**
      * Delete the entry which key is given as a parameter. If the entry exists, it will
      * be removed from the tree, the old tuple will be returned. Otherwise, null is returned.
      *
      * @param key The key for the entry we try to remove
      * @return A Tuple<K, V> containing the removed entry, or null if it's not found.
      */
     /*no qualifier*/Tuple<K, V> delete( K key, long revision ) throws IOException
     {
         return delete( key, null, revision );
     }


     /*no qualifier*/abstract Tuple<K, V> delete( K key, V value, long revision ) throws IOException;


     /**
      * {@inheritDoc}
      */
     public V insert( K key, V value ) throws IOException
     {
         long revision = generateRevision();

         V existingValue = null;

         try
         {
             if ( writeTransaction == null )
             {
                 writeLock.lock();
             }

             InsertResult<K, V> result = insert( key, value, revision );

             if ( result instanceof ModifyResult )
             {
                 existingValue = ( ( ModifyResult<K, V> ) result ).getModifiedValue();
             }
         }
         finally
         {
             // See above
             if ( writeTransaction == null )
             {
                 writeLock.unlock();
             }
         }

         return existingValue;
     }


     /**
      * {@inheritDoc}
      */
     /* no qualifier */abstract InsertResult<K, V> insert( K key, V value, long revision ) throws IOException;


     /**
      * Flush the latest revision to disk. We will replace the current file by the new one, as
      * we flush in a temporary file.
      */
     public void flush() throws IOException
     {
     }


     /**
      * {@inheritDoc}
      */
     public V get( K key ) throws IOException, KeyNotFoundException
     {
         return rootPage.get( key );
     }


     /**
      * {@inheritDoc}
      */
     public V get( long revision, K key ) throws IOException, KeyNotFoundException
     {
         // Fetch the root page for this revision
         Page<K, V> revisionRootPage = getRootPage( revision );

         return revisionRootPage.get( key );
     }


     /**
      * {@inheritDoc}
      */
     public Page<K, V> getRootPage()
     {
         return rootPage;
     }


     /**
      * {@inheritDoc}
      */
     /* no qualifier */void setRootPage( Page<K, V> root )
     {
         rootPage = root;
     }


     /**
      * {@inheritDoc}
      */
     public ValueCursor<V> getValues( K key ) throws IOException, KeyNotFoundException
     {
         return rootPage.getValues( key );
     }


     /**
      * {@inheritDoc}
      */
     public boolean hasKey( K key ) throws IOException
     {
         if ( key == null )
         {
             return false;
         }

         return rootPage.hasKey( key );
     }


     /**
      * {@inheritDoc}
      */
     public boolean hasKey( long revision, K key ) throws IOException, KeyNotFoundException
     {
         if ( key == null )
         {
             return false;
         }

         // Fetch the root page for this revision
         Page<K, V> revisionRootPage = getRootPage( revision );

         return revisionRootPage.hasKey( key );
     }


     /**
      * {@inheritDoc}
      */
     public ElementSerializer<K> getKeySerializer()
     {
         return keySerializer;
     }


     /**
      * {@inheritDoc}
      */
     public void setKeySerializer( ElementSerializer<K> keySerializer )
     {
         this.keySerializer = keySerializer;
         btreeHeader.setKeySerializerFQCN( keySerializer.getClass().getName() );
     }


     /**
      * {@inheritDoc}
      */
     public String getKeySerializerFQCN()
     {
         return btreeHeader.getKeySerializerFQCN();
     }


     /**
      * {@inheritDoc}
      */
     public ElementSerializer<V> getValueSerializer()
     {
         return valueSerializer;
     }


     /**
      * {@inheritDoc}
      */
     public void setValueSerializer( ElementSerializer<V> valueSerializer )
     {
         this.valueSerializer = valueSerializer;
         btreeHeader.setValueSerializerFQCN( valueSerializer.getClass().getName() );
     }


     /**
      * {@inheritDoc}
      */
     public String getValueSerializerFQCN()
     {
         return btreeHeader.getValueSerializerFQCN();
     }


     /**
      * {@inheritDoc}
      */
     public long getRevision()
     {
         return btreeHeader.getRevision();
     }


     /**
      * {@inheritDoc}
      */
     /* no qualifier */void setRevision( long revision )
     {
         btreeHeader.setRevision( revision );
     }


     /**
      * Generates a new revision number. It's only used by the Page instances.
      *
      * @return a new incremental revision number
      */
     /* no qualifier */long generateRevision()
     {
         return btreeHeader.incrementRevision();
     }


     /**
      * {@inheritDoc}
      */
     public long getReadTimeOut()
     {
         return readTimeOut;
     }


     /**
      * {@inheritDoc}
      */
     public void setReadTimeOut( long readTimeOut )
     {
         this.readTimeOut = readTimeOut;
     }


     /**
      * {@inheritDoc}
      */
     public long getNbElems()
     {
         return btreeHeader.getNbElems();
     }


     /**
      * {@inheritDoc}
      */
     /* no qualifier */void setNbElems( long nbElems )
     {
         btreeHeader.setNbElems( nbElems );
     }


     /**
      * {@inheritDoc}
      */
     public int getPageSize()
     {
         return btreeHeader.getPageSize();
     }


     /**
      * {@inheritDoc}
      */
     public void setPageSize( int pageSize )
     {
         if ( pageSize <= 2 )
         {
             btreeHeader.setPageSize( DEFAULT_PAGE_SIZE );
         }
         else
         {
             btreeHeader.setPageSize( getPowerOf2( pageSize ) );
         }
     }


     /**
      * {@inheritDoc}
      */
     public String getName()
     {
         return btreeHeader.getName();
     }


     /**
      * {@inheritDoc}
      */
     public void setName( String name )
     {
         btreeHeader.setName( name );
     }


     /**
      * {@inheritDoc}
      */
     public Comparator<K> getComparator()
     {
         return keySerializer.getComparator();
     }


     /**
      * {@inheritDoc}
      */
     public int getWriteBufferSize()
     {
         return writeBufferSize;
     }


     /**
      * {@inheritDoc}
      */
     public void setWriteBufferSize( int writeBufferSize )
     {
         this.writeBufferSize = writeBufferSize;
     }


     /**
      * {@inheritDoc}
      */
     public boolean isAllowDuplicates()
     {
         return btreeHeader.isAllowDuplicates();
     }


     /**
      * {@inheritDoc}
      */
     public void setAllowDuplicates( boolean allowDuplicates )
     {
         btreeHeader.setAllowDuplicates( allowDuplicates );
     }


     /**
      * {@inheritDoc}
      */
     public BTreeTypeEnum getType()
     {
         return type;
     }


     /**
      * @param type the type to set
      */
     public void setType( BTreeTypeEnum type )
     {
         this.type = type;
     }


     /**
      * Gets the number which is a power of 2 immediately above the given positive number.
      */
     private int getPowerOf2( int size )
     {
         int newSize = --size;
         newSize |= newSize >> 1;
         newSize |= newSize >> 2;
         newSize |= newSize >> 4;
         newSize |= newSize >> 8;
         newSize |= newSize >> 16;
         newSize++;

         return newSize;
     }


     /**
      * Create a thread that is responsible of cleaning the transactions when
      * they hit the timeout
      */
     /*no qualifier*/void createTransactionManager()
     {
         Runnable readTransactionTask = new Runnable()
         {
             public void run()
             {
                 try
                 {
                     ReadTransaction<K, V> transaction = null;

                     while ( !Thread.currentThread().isInterrupted() )
                     {
                         long timeoutDate = System.currentTimeMillis() - readTimeOut;
                         long t0 = System.currentTimeMillis();
                         int nbTxns = 0;

                         // Loop on all the transactions from the queue
                         while ( ( transaction = readTransactions.peek() ) != null )
                         {
                             nbTxns++;

                             if ( transaction.isClosed() )
                             {
                                 // The transaction is already closed, remove it from the queue
                                 readTransactions.poll();
                                 continue;
                             }

                             // Check if the transaction has timed out
                             if ( transaction.getCreationDate() < timeoutDate )
                             {
                                 transaction.close();
                                 readTransactions.poll();
                                 continue;
                             }

                             // We need to stop now
                             break;
                         }

                         long t1 = System.currentTimeMillis();

                         if ( nbTxns > 0 )
                         {
                             System.out.println( "Processing old txn : " + nbTxns + ", " + ( t1 - t0 ) + "ms" );
                         }

                         // Wait until we reach the timeout
                         Thread.sleep( readTimeOut );
                     }
                 }
                 catch ( InterruptedException ie )
                 {
                     //System.out.println( "Interrupted" );
                 }
                 catch ( Exception e )
                 {
                     throw new RuntimeException( e );
                 }
             }
         };

         readTransactionsThread = new Thread( readTransactionTask );
         readTransactionsThread.setDaemon( true );
         readTransactionsThread.start();
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/
	package org.apache.directory.mavibot.btree;


	import java.io.IOException;
	import java.lang.reflect.Array;
	import java.util.Comparator;
	import java.util.concurrent.ConcurrentLinkedQueue;
	import java.util.concurrent.locks.ReentrantLock;

	import org.apache.directory.mavibot.btree.exception.KeyNotFoundException;
	import org.apache.directory.mavibot.btree.serializer.ElementSerializer;


	/**
	* A BTree abstract class containing the methods shared by the PersistedBTree or the InMemoryBTree
	* implementations.
	*
	* @param <K> The Key type
	* @param <V> The Value type
	*
	* @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
	*/
	/* No qualifier*/abstract class AbstractBTree<K, V> implements BTree<K, V>
	{
	/** The read transaction timeout */
	protected long readTimeOut = DEFAULT_READ_TIMEOUT;

	/** The Header for a managed BTree */
	protected BTreeHeader btreeHeader;

	/** The current rootPage */
	protected volatile Page<K, V> rootPage;

	/** The Key serializer used for this tree.*/
	protected ElementSerializer<K> keySerializer;

	/** The Value serializer used for this tree. */
	protected ElementSerializer<V> valueSerializer;

	/** The list of read transactions being executed */
	protected ConcurrentLinkedQueue<ReadTransaction<K, V>> readTransactions;

	/** The size of the buffer used to write data in disk */
	protected int writeBufferSize;

	/** A lock used to protect the write operation against concurrent access */
	protected ReentrantLock writeLock;

	/** Flag to enable duplicate key support */
	private boolean allowDuplicates;

	/** The thread responsible for the cleanup of timed out reads */
	protected Thread readTransactionsThread;

	/** The BTree type : either in-memory, disk backed or persisted */
	private BTreeTypeEnum type;

	/** The current transaction */
	protected WriteTransaction writeTransaction;


	/**
	* Starts a Read Only transaction. If the transaction is not closed, it will be
	* automatically closed after the timeout
	*
	* @return The created transaction
	*/
	protected ReadTransaction<K, V> beginReadTransaction()
	{
	ReadTransaction<K, V> readTransaction = new ReadTransaction<K, V>( rootPage, btreeHeader.getRevision() - 1,
	System.currentTimeMillis() );

	readTransactions.add( readTransaction );

	return readTransaction;
	}


	/**
	* {@inheritDoc}
	*/
	public TupleCursor<K, V> browse() throws IOException
	{
	ReadTransaction<K, V> transaction = beginReadTransaction();

	// Fetch the root page for this revision
	ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

	TupleCursor<K, V> cursor = rootPage.browse( transaction, stack, 0 );

	// Set the position before the first element
	cursor.beforeFirst();

	return cursor;
	}


	/**
	* {@inheritDoc}
	*/
	public TupleCursor<K, V> browse( long revision ) throws IOException, KeyNotFoundException
	{
	ReadTransaction<K, V> transaction = beginReadTransaction();

	// Fetch the root page for this revision
	Page<K, V> revisionRootPage = getRootPage( revision );

	ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

	// And get the cursor
	TupleCursor<K, V> cursor = revisionRootPage.browse( transaction, stack, 0 );

	return cursor;
	}


	/**
	* {@inheritDoc}
	*/
	public TupleCursor<K, V> browseFrom( K key ) throws IOException
	{
	ReadTransaction<K, V> transaction = beginReadTransaction();

	// Fetch the root page for this revision
	ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

	TupleCursor<K, V> cursor = rootPage.browse( key, transaction, stack, 0 );

	return cursor;
	}


	/**
	* {@inheritDoc}
	*/
	public TupleCursor<K, V> browseFrom( long revision, K key ) throws IOException, KeyNotFoundException
	{
	ReadTransaction<K, V> transaction = beginReadTransaction();

	// Fetch the rootPage for this revision
	Page<K, V> revisionRootPage = getRootPage( revision );

	ParentPos<K, V>[] stack = (ParentPos<K, V>[]) Array.newInstance( ParentPos.class, 32 );

	// And get the cursor
	TupleCursor<K, V> cursor = revisionRootPage.browse( key, transaction, stack, 0 );

	return cursor;
	}


	/**
	* {@inheritDoc}
	*/
	public boolean contains( K key, V value ) throws IOException
	{
	return rootPage.contains( key, value );
	}


	/**
	* {@inheritDoc}
	*/
	public boolean contains( long revision, K key, V value ) throws IOException, KeyNotFoundException
	{
	// Fetch the root page for this revision
	Page<K, V> revisionRootPage = getRootPage( revision );

	return revisionRootPage.contains( key, value );
	}


	/**
	* {@inheritDoc}
	*/
	public Tuple<K, V> delete( K key ) throws IOException
	{
	if ( key == null )
	{
	throw new IllegalArgumentException( "Key must not be null" );
	}

	long revision = generateRevision();

	Tuple<K, V> deleted = delete( key, revision );

	return deleted;
	}


	/**
	* {@inheritDoc}
	*/
	public Tuple<K, V> delete( K key, V value ) throws IOException
	{
	if ( key == null )
	{
	throw new IllegalArgumentException( "Key must not be null" );
	}

	if ( value == null )
	{
	throw new IllegalArgumentException( "Value must not be null" );
	}

	long revision = generateRevision();

	Tuple<K, V> deleted = delete( key, value, revision );

	return deleted;
	}


	/**
	* Delete the entry which key is given as a parameter. If the entry exists, it will
	* be removed from the tree, the old tuple will be returned. Otherwise, null is returned.
	*
	* @param key The key for the entry we try to remove
	* @return A Tuple<K, V> containing the removed entry, or null if it's not found.
	*/
	/no qualifier/Tuple<K, V> delete( K key, long revision ) throws IOException
	{
	return delete( key, null, revision );
	}


	/no qualifier/abstract Tuple<K, V> delete( K key, V value, long revision ) throws IOException;


	/**
	* {@inheritDoc}
	*/
	public V insert( K key, V value ) throws IOException
	{
	long revision = generateRevision();

	V existingValue = null;

	try
	{
	if ( writeTransaction == null )
	{
	writeLock.lock();
	}

	InsertResult<K, V> result = insert( key, value, revision );

	if ( result instanceof ModifyResult )
	{
	existingValue = ( ( ModifyResult<K, V> ) result ).getModifiedValue();
	}
	}
	finally
	{
	// See above
	if ( writeTransaction == null )
	{
	writeLock.unlock();
	}
	}

	return existingValue;
	}


	/**
	* {@inheritDoc}
	*/
	/* no qualifier */abstract InsertResult<K, V> insert( K key, V value, long revision ) throws IOException;


	/**
	* Flush the latest revision to disk. We will replace the current file by the new one, as
	* we flush in a temporary file.
	*/
	public void flush() throws IOException
	{
	}


	/**
	* {@inheritDoc}
	*/
	public V get( K key ) throws IOException, KeyNotFoundException
	{
	return rootPage.get( key );
	}


	/**
	* {@inheritDoc}
	*/
	public V get( long revision, K key ) throws IOException, KeyNotFoundException
	{
	// Fetch the root page for this revision
	Page<K, V> revisionRootPage = getRootPage( revision );

	return revisionRootPage.get( key );
	}


	/**
	* {@inheritDoc}
	*/
	public Page<K, V> getRootPage()
	{
	return rootPage;
	}


	/**
	* {@inheritDoc}
	*/
	/* no qualifier */void setRootPage( Page<K, V> root )
	{
	rootPage = root;
	}


	/**
	* {@inheritDoc}
	*/
	public ValueCursor<V> getValues( K key ) throws IOException, KeyNotFoundException
	{
	return rootPage.getValues( key );
	}


	/**
	* {@inheritDoc}
	*/
	public boolean hasKey( K key ) throws IOException
	{
	if ( key == null )
	{
	return false;
	}

	return rootPage.hasKey( key );
	}


	/**
	* {@inheritDoc}
	*/
	public boolean hasKey( long revision, K key ) throws IOException, KeyNotFoundException
	{
	if ( key == null )
	{
	return false;
	}

	// Fetch the root page for this revision
	Page<K, V> revisionRootPage = getRootPage( revision );

	return revisionRootPage.hasKey( key );
	}


	/**
	* {@inheritDoc}
	*/
	public ElementSerializer<K> getKeySerializer()
	{
	return keySerializer;
	}


	/**
	* {@inheritDoc}
	*/
	public void setKeySerializer( ElementSerializer<K> keySerializer )
	{
	this.keySerializer = keySerializer;
	btreeHeader.setKeySerializerFQCN( keySerializer.getClass().getName() );
	}


	/**
	* {@inheritDoc}
	*/
	public String getKeySerializerFQCN()
	{
	return btreeHeader.getKeySerializerFQCN();
	}


	/**
	* {@inheritDoc}
	*/
	public ElementSerializer<V> getValueSerializer()
	{
	return valueSerializer;
	}


	/**
	* {@inheritDoc}
	*/
	public void setValueSerializer( ElementSerializer<V> valueSerializer )
	{
	this.valueSerializer = valueSerializer;
	btreeHeader.setValueSerializerFQCN( valueSerializer.getClass().getName() );
	}


	/**
	* {@inheritDoc}
	*/
	public String getValueSerializerFQCN()
	{
	return btreeHeader.getValueSerializerFQCN();
	}


	/**
	* {@inheritDoc}
	*/
	public long getRevision()
	{
	return btreeHeader.getRevision();
	}


	/**
	* {@inheritDoc}
	*/
	/* no qualifier */void setRevision( long revision )
	{
	btreeHeader.setRevision( revision );
	}


	/**
	* Generates a new revision number. It's only used by the Page instances.
	*
	* @return a new incremental revision number
	*/
	/* no qualifier */long generateRevision()
	{
	return btreeHeader.incrementRevision();
	}


	/**
	* {@inheritDoc}
	*/
	public long getReadTimeOut()
	{
	return readTimeOut;
	}


	/**
	* {@inheritDoc}
	*/
	public void setReadTimeOut( long readTimeOut )
	{
	this.readTimeOut = readTimeOut;
	}


	/**
	* {@inheritDoc}
	*/
	public long getNbElems()
	{
	return btreeHeader.getNbElems();
	}


	/**
	* {@inheritDoc}
	*/
	/* no qualifier */void setNbElems( long nbElems )
	{
	btreeHeader.setNbElems( nbElems );
	}


	/**
	* {@inheritDoc}
	*/
	public int getPageSize()
	{
	return btreeHeader.getPageSize();
	}


	/**
	* {@inheritDoc}
	*/
	public void setPageSize( int pageSize )
	{
	if ( pageSize <= 2 )
	{
	btreeHeader.setPageSize( DEFAULT_PAGE_SIZE );
	}
	else
	{
	btreeHeader.setPageSize( getPowerOf2( pageSize ) );
	}
	}


	/**
	* {@inheritDoc}
	*/
	public String getName()
	{
	return btreeHeader.getName();
	}


	/**
	* {@inheritDoc}
	*/
	public void setName( String name )
	{
	btreeHeader.setName( name );
	}


	/**
	* {@inheritDoc}
	*/
	public Comparator<K> getComparator()
	{
	return keySerializer.getComparator();
	}


	/**
	* {@inheritDoc}
	*/
	public int getWriteBufferSize()
	{
	return writeBufferSize;
	}


	/**
	* {@inheritDoc}
	*/
	public void setWriteBufferSize( int writeBufferSize )
	{
	this.writeBufferSize = writeBufferSize;
	}


	/**
	* {@inheritDoc}
	*/
	public boolean isAllowDuplicates()
	{
	return btreeHeader.isAllowDuplicates();
	}


	/**
	* {@inheritDoc}
	*/
	public void setAllowDuplicates( boolean allowDuplicates )
	{
	btreeHeader.setAllowDuplicates( allowDuplicates );
	}


	/**
	* {@inheritDoc}
	*/
	public BTreeTypeEnum getType()
	{
	return type;
	}


	/**
	* @param type the type to set
	*/
	public void setType( BTreeTypeEnum type )
	{
	this.type = type;
	}


	/**
	* Gets the number which is a power of 2 immediately above the given positive number.
	*/
	private int getPowerOf2( int size )
	{
	int newSize = --size;
	newSize \|= newSize >> 1;
	newSize \|= newSize >> 2;
	newSize \|= newSize >> 4;
	newSize \|= newSize >> 8;
	newSize \|= newSize >> 16;
	newSize++;

	return newSize;
	}


	/**
	* Create a thread that is responsible of cleaning the transactions when
	* they hit the timeout
	*/
	/no qualifier/void createTransactionManager()
	{
	Runnable readTransactionTask = new Runnable()
	{
	public void run()
	{
	try
	{
	ReadTransaction<K, V> transaction = null;

	while ( !Thread.currentThread().isInterrupted() )
	{
	long timeoutDate = System.currentTimeMillis() - readTimeOut;
	long t0 = System.currentTimeMillis();
	int nbTxns = 0;

	// Loop on all the transactions from the queue
	while ( ( transaction = readTransactions.peek() ) != null )
	{
	nbTxns++;

	if ( transaction.isClosed() )
	{
	// The transaction is already closed, remove it from the queue
	readTransactions.poll();
	continue;
	}

	// Check if the transaction has timed out
	if ( transaction.getCreationDate() < timeoutDate )
	{
	transaction.close();
	readTransactions.poll();
	continue;
	}

	// We need to stop now
	break;
	}

	long t1 = System.currentTimeMillis();

	if ( nbTxns > 0 )
	{
	System.out.println( "Processing old txn : " + nbTxns + ", " + ( t1 - t0 ) + "ms" );
	}

	// Wait until we reach the timeout
	Thread.sleep( readTimeOut );
	}
	}
	catch ( InterruptedException ie )
	{
	//System.out.println( "Interrupted" );
	}
	catch ( Exception e )
	{
	throw new RuntimeException( e );
	}
	}
	};

	readTransactionsThread = new Thread( readTransactionTask );
	readTransactionsThread.setDaemon( true );
	readTransactionsThread.start();
	}
	}