Google Git
Sign in
apache / directory-mavibot / 23a47f40fb2f0ac18c43c2cb169bd6becfd70464 / . / mavibot / src / main / java / org / apache / directory / mavibot / btree / RecordManager.java
blob: 4137bab5cf0ac9743d3eea15d7d910ea09f82f4f [file] [log] [blame]
/*
* 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.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.directory.mavibot.btree.exception.BTreeAlreadyManagedException;
import org.apache.directory.mavibot.btree.exception.BTreeCreationException;
import org.apache.directory.mavibot.btree.exception.EndOfFileExceededException;
import org.apache.directory.mavibot.btree.exception.FileException;
import org.apache.directory.mavibot.btree.exception.InvalidOffsetException;
import org.apache.directory.mavibot.btree.exception.KeyNotFoundException;
import org.apache.directory.mavibot.btree.exception.RecordManagerException;
import org.apache.directory.mavibot.btree.serializer.ElementSerializer;
import org.apache.directory.mavibot.btree.serializer.IntSerializer;
import org.apache.directory.mavibot.btree.serializer.LongArraySerializer;
import org.apache.directory.mavibot.btree.serializer.LongSerializer;
import org.apache.directory.mavibot.btree.util.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The RecordManager is used to manage the file in which we will store the B-trees.
* A RecordManager will manage more than one B-tree.<br/>
*
* It stores data in fixed size pages (default size is 512 bytes), which may be linked one to
* the other if the data we want to store is too big for a page.
*
* @author <a href="mailto:dev@directory.apache.org">Apache Directory Project</a>
*/
public class RecordManager extends AbstractTransactionManager
{
/** The LoggerFactory used by this class */
protected static final Logger LOG = LoggerFactory.getLogger( RecordManager.class );
/** The LoggerFactory used to trace TXN operations */
protected static final Logger TXN_LOG = LoggerFactory.getLogger( "TXN_LOG" );
/** The LoggerFactory used by this class */
protected static final Logger LOG_PAGES = LoggerFactory.getLogger( "org.apache.directory.mavibot.LOG_PAGES" );
/** A dedicated logger for the check */
protected static final Logger LOG_CHECK = LoggerFactory.getLogger( "org.apache.directory.mavibot.LOG_CHECK" );
/** The associated file */
private File file;
/** The channel used to read and write data */
/* no qualifier */FileChannel fileChannel;
/** The number of managed B-trees */
/* no qualifier */int nbBtree;
/** The first and last free page */
/* no qualifier */long firstFreePage;
/** Some counters to track the number of free pages */
public AtomicLong nbFreedPages = new AtomicLong( 0 );
public AtomicLong nbCreatedPages = new AtomicLong( 0 );
public AtomicLong nbReusedPages = new AtomicLong( 0 );
public AtomicLong nbUpdateRMHeader = new AtomicLong( 0 );
public AtomicLong nbUpdateBtreeHeader = new AtomicLong( 0 );
public AtomicLong nbUpdatePageIOs = new AtomicLong( 0 );
/** The offset of the end of the file */
private long endOfFileOffset;
/**
* A B-tree used to manage the page that has been copied in a new version.
* Those pages can be reclaimed when the associated version is dead.
**/
/* no qualifier */ BTree<RevisionName, long[]> copiedPageBtree;
/** A constant for an offset on a non existing page */
public static final long NO_PAGE = -1L;
/** The number of bytes used to store the size of a page */
private static final int PAGE_SIZE = 4;
/** The size of the link to next page */
private static final int LINK_SIZE = 8;
/** Some constants */
private static final int BYTE_SIZE = 1;
/* no qualifier */static final int INT_SIZE = 4;
/* no qualifier */static final int LONG_SIZE = 8;
/** The default page size */
public static final int DEFAULT_PAGE_SIZE = 512;
/** The minimal page size. Can't be below 64, as we have to store many thing sin the RMHeader */
private static final int MIN_PAGE_SIZE = 64;
/** The RecordManager header size */
/* no qualifier */static int RECORD_MANAGER_HEADER_SIZE = DEFAULT_PAGE_SIZE;
/** A global buffer used to store the RecordManager header */
private ByteBuffer RECORD_MANAGER_HEADER_BUFFER;
/** A static buffer used to store the RecordManager header */
private byte[] RECORD_MANAGER_HEADER_BYTES;
/** The length of an Offset, as a negative value */
private byte[] LONG_LENGTH = new byte[]
{ ( byte ) 0xFF, ( byte ) 0xFF, ( byte ) 0xFF, ( byte ) 0xF8 };
/** The RecordManager underlying page size. */
/* no qualifier */int pageSize = DEFAULT_PAGE_SIZE;
/** The set of managed B-trees */
private Map<String, BTree<Object, Object>> managedBtrees;
/** The queue of recently closed transactions */
private Queue<RevisionName> closedTransactionsQueue = new LinkedBlockingQueue<RevisionName>();
/** The default file name */
private static final String DEFAULT_FILE_NAME = "mavibot.db";
/** A flag set to true if we want to keep old revisions */
private boolean keepRevisions;
/** A flag used by internal btrees */
public static final boolean INTERNAL_BTREE = true;
/** A flag used by internal btrees */
public static final boolean NORMAL_BTREE = false;
/** The B-tree of B-trees */
/* no qualifier */ BTree<NameRevision, Long> btreeOfBtrees;
/** The B-tree of B-trees management btree name */
/* no qualifier */static final String BTREE_OF_BTREES_NAME = "_btree_of_btrees_";
/** The CopiedPages management btree name */
/* no qualifier */static final String COPIED_PAGE_BTREE_NAME = "_copiedPageBtree_";
/** The current B-tree of B-trees header offset */
/* no qualifier */long currentBtreeOfBtreesOffset;
/** The previous B-tree of B-trees header offset */
private long previousBtreeOfBtreesOffset = NO_PAGE;
/** The offset on the current copied pages B-tree */
/* no qualifier */ long currentCopiedPagesBtreeOffset = NO_PAGE;
/** The offset on the previous copied pages B-tree */
private long previousCopiedPagesBtreeOffset = NO_PAGE;
/** A lock to protect the transaction handling */
private ReentrantLock transactionLock = new ReentrantLock();
/** A ThreadLocalStorage used to store the current transaction */
private static final ThreadLocal<Integer> context = new ThreadLocal<Integer>();
/** The list of PageIO that can be freed after a commit */
List<PageIO> freedPages = new ArrayList<PageIO>();
/** The list of PageIO that can be freed after a roolback */
private List<PageIO> allocatedPages = new ArrayList<PageIO>();
/** A Map keeping the latest revisions for each managed BTree */
private Map<String, BTreeHeader<?, ?>> currentBTreeHeaders = new HashMap<String, BTreeHeader<?, ?>>();
/** A Map storing the new revisions when some change have been made in some BTrees */
private Map<String, BTreeHeader<?, ?>> newBTreeHeaders = new HashMap<String, BTreeHeader<?, ?>>();
/** A lock to protect the BtreeHeader maps */
private ReadWriteLock btreeHeadersLock = new ReentrantReadWriteLock();
/** A value stored into the transaction context for rollbacked transactions */
private static final int ROLLBACKED_TXN = 0;
/** A lock to protect the freepage pointers */
private ReentrantLock freePageLock = new ReentrantLock();
/** the space reclaimer */
private PageReclaimer reclaimer;
/** variable to keep track of the write commit count */
private int commitCount = 0;
/** the threshold at which the PageReclaimer will be run to free the copied pages */
// FIXME the below value is derived after seeing that anything higher than that
// is resulting in a "This thread does not hold the transactionLock" error
private int pageReclaimerThreshold = 70;
/* a flag used to disable the free page reclaimer (used for internal testing only) */
private boolean disableReclaimer = false;
public Map<Long, Integer> writeCounter = new HashMap<Long, Integer>();
/**
* Create a Record manager which will either create the underlying file
* or load an existing one. If a folder is provided, then we will create
* a file with a default name : mavibot.db
*
* @param name The file name, or a folder name
*/
public RecordManager( String fileName )
{
this( fileName, DEFAULT_PAGE_SIZE );
}
/**
* Create a Record manager which will either create the underlying file
* or load an existing one. If a folder is provider, then we will create
* a file with a default name : mavibot.db
*
* @param name The file name, or a folder name
* @param pageSize the size of a page on disk, in bytes
*/
public RecordManager( String fileName, int pageSize )
{
managedBtrees = new LinkedHashMap<String, BTree<Object, Object>>();
if ( pageSize < MIN_PAGE_SIZE )
{
this.pageSize = MIN_PAGE_SIZE;
}
else
{
this.pageSize = pageSize;
}
RECORD_MANAGER_HEADER_BUFFER = ByteBuffer.allocate( this.pageSize );
RECORD_MANAGER_HEADER_BYTES = new byte[this.pageSize];
RECORD_MANAGER_HEADER_SIZE = this.pageSize;
// Open the file or create it
File tmpFile = new File( fileName );
if ( tmpFile.isDirectory() )
{
// It's a directory. Check that we don't have an existing mavibot file
tmpFile = new File( tmpFile, DEFAULT_FILE_NAME );
}
// We have to create a new file, if it does not already exist
boolean isNewFile = createFile( tmpFile );
try
{
RandomAccessFile randomFile = new RandomAccessFile( file, "rw" );
fileChannel = randomFile.getChannel();
// get the current end of file offset
endOfFileOffset = fileChannel.size();
if ( isNewFile )
{
initRecordManager();
}
else
{
loadRecordManager();
}
reclaimer = new PageReclaimer( this );
runReclaimer();
}
catch ( Exception e )
{
LOG.error( "Error while initializing the RecordManager : {}", e.getMessage() );
LOG.error( "", e );
throw new RecordManagerException( e );
}
}
/**
* runs the PageReclaimer to free the copied pages
*/
private void runReclaimer()
{
if ( disableReclaimer )
{
LOG.warn( "Free page reclaimer is disabled, this should not be disabled on production systems." );
return;
}
try
{
commitCount = 0;
reclaimer.reclaim();
// must update the headers after reclaim operation
updateRecordManagerHeader();
}
catch ( Exception e )
{
LOG.warn( "PageReclaimer failed to free the pages", e );
}
}
/**
* Create the mavibot file if it does not exist
*/
private boolean createFile( File mavibotFile )
{
try
{
boolean creation = mavibotFile.createNewFile();
file = mavibotFile;
if ( mavibotFile.length() == 0 )
{
return true;
}
else
{
return creation;
}
}
catch ( IOException ioe )
{
LOG.error( "Cannot create the file {}", mavibotFile.getName() );
return false;
}
}
/**
* We will create a brand new RecordManager file, containing nothing, but the RecordManager header,
* a B-tree to manage the old revisions we want to keep and
* a B-tree used to manage pages associated with old versions.
* <br/>
* The RecordManager header contains the following details :
* <pre>
* +--------------------------+
* | PageSize | 4 bytes : The size of a physical page (default to 512)
* +--------------------------+
* | NbTree | 4 bytes : The number of managed B-trees (zero or more)
* +--------------------------+
* | FirstFree | 8 bytes : The offset of the first free page
* +--------------------------+
* | current BoB offset | 8 bytes : The offset of the current BoB
* +--------------------------+
* | previous BoB offset | 8 bytes : The offset of the previous BoB
* +--------------------------+
* | current CP btree offset | 8 bytes : The offset of the current BoB
* +--------------------------+
* | previous CP btree offset | 8 bytes : The offset of the previous BoB
* +--------------------------+
* </pre>
*
* We then store the B-tree managing the pages that have been copied when we have added
* or deleted an element in the B-tree. They are associated with a version.
*
* Last, we add the bTree that keep a track on each revision we can have access to.
*/
private void initRecordManager() throws IOException
{
// Create a new Header
nbBtree = 0;
firstFreePage = NO_PAGE;
currentBtreeOfBtreesOffset = NO_PAGE;
updateRecordManagerHeader();
// Set the offset of the end of the file
endOfFileOffset = fileChannel.size();
// First, create the btree of btrees <NameRevision, Long>
createBtreeOfBtrees();
// Now, initialize the Copied Page B-tree
createCopiedPagesBtree();
// Inject these B-trees into the RecordManager. They are internal B-trees.
try
{
manageSubBtree( btreeOfBtrees );
currentBtreeOfBtreesOffset = ( ( PersistedBTree<NameRevision, Long> ) btreeOfBtrees ).getBtreeHeader()
.getBTreeHeaderOffset();
updateRecordManagerHeader();
// Inject the BtreeOfBtrees into the currentBtreeHeaders map
currentBTreeHeaders.put( BTREE_OF_BTREES_NAME,
( ( PersistedBTree<NameRevision, Long> ) btreeOfBtrees ).getBtreeHeader() );
newBTreeHeaders.put( BTREE_OF_BTREES_NAME,
( ( PersistedBTree<NameRevision, Long> ) btreeOfBtrees ).getBtreeHeader() );
// The FreePage B-tree
manageSubBtree( copiedPageBtree );
currentCopiedPagesBtreeOffset = ( ( PersistedBTree<RevisionName, long[]> ) copiedPageBtree ).getBtreeHeader().getBTreeHeaderOffset();
updateRecordManagerHeader();
// Inject the CopiedPagesBTree into the currentBtreeHeaders map
currentBTreeHeaders.put( COPIED_PAGE_BTREE_NAME, ( ( PersistedBTree<RevisionName, long[]> ) copiedPageBtree ).getBtreeHeader() );
newBTreeHeaders.put( COPIED_PAGE_BTREE_NAME, ( ( PersistedBTree<RevisionName, long[]> ) copiedPageBtree ).getBtreeHeader() );
}
catch ( BTreeAlreadyManagedException btame )
{
// Can't happen here.
}
// We are all set ! Verify the file
if ( LOG_CHECK.isDebugEnabled() )
{
MavibotInspector.check( this );
}
}
/**
* Create the B-treeOfBtrees
*/
private void createBtreeOfBtrees()
{
PersistedBTreeConfiguration<NameRevision, Long> configuration = new PersistedBTreeConfiguration<NameRevision, Long>();
configuration.setKeySerializer( NameRevisionSerializer.INSTANCE );
configuration.setName( BTREE_OF_BTREES_NAME );
configuration.setValueSerializer( LongSerializer.INSTANCE );
configuration.setBtreeType( BTreeTypeEnum.BTREE_OF_BTREES );
configuration.setCacheSize( PersistedBTree.DEFAULT_CACHE_SIZE );
btreeOfBtrees = BTreeFactory.createPersistedBTree( configuration );
}
/**
* Create the CopiedPagesBtree
*/
private void createCopiedPagesBtree()
{
PersistedBTreeConfiguration<RevisionName, long[]> configuration = new PersistedBTreeConfiguration<RevisionName, long[]>();
configuration.setKeySerializer( RevisionNameSerializer.INSTANCE );
configuration.setName( COPIED_PAGE_BTREE_NAME );
configuration.setValueSerializer( LongArraySerializer.INSTANCE );
configuration.setBtreeType( BTreeTypeEnum.COPIED_PAGES_BTREE );
configuration.setCacheSize( PersistedBTree.DEFAULT_CACHE_SIZE );
copiedPageBtree = BTreeFactory.createPersistedBTree( configuration );
}
/**
* Load the BTrees from the disk.
*
* @throws InstantiationException
* @throws IllegalAccessException
* @throws ClassNotFoundException
* @throws NoSuchFieldException
* @throws SecurityException
* @throws IllegalArgumentException
*/
private void loadRecordManager() throws IOException, ClassNotFoundException, IllegalAccessException,
InstantiationException, IllegalArgumentException, SecurityException, NoSuchFieldException, KeyNotFoundException
{
if ( fileChannel.size() != 0 )
{
ByteBuffer recordManagerHeader = ByteBuffer.allocate( RECORD_MANAGER_HEADER_SIZE );
// The file exists, we have to load the data now
fileChannel.read( recordManagerHeader );
recordManagerHeader.rewind();
// read the RecordManager Header :
// +---------------------+
// | PageSize | 4 bytes : The size of a physical page (default to 4096)
// +---------------------+
// | NbTree | 4 bytes : The number of managed B-trees (at least 1)
// +---------------------+
// | FirstFree | 8 bytes : The offset of the first free page
// +---------------------+
// | current BoB offset | 8 bytes : The offset of the current B-tree of B-trees
// +---------------------+
// | previous BoB offset | 8 bytes : The offset of the previous B-tree of B-trees
// +---------------------+
// | current CP offset | 8 bytes : The offset of the current Copied Pages B-tree
// +---------------------+
// | previous CP offset | 8 bytes : The offset of the previous Copied Pages B-tree
// +---------------------+
// The page size
pageSize = recordManagerHeader.getInt();
// The number of managed B-trees
nbBtree = recordManagerHeader.getInt();
// The first and last free page
firstFreePage = recordManagerHeader.getLong();
// Read all the free pages
checkFreePages();
// The current BOB offset
currentBtreeOfBtreesOffset = recordManagerHeader.getLong();
// The previous BOB offset
previousBtreeOfBtreesOffset = recordManagerHeader.getLong();
// The current Copied Pages B-tree offset
currentCopiedPagesBtreeOffset = recordManagerHeader.getLong();
// The previous Copied Pages B-tree offset
previousCopiedPagesBtreeOffset = recordManagerHeader.getLong();
// read the B-tree of B-trees
PageIO[] bobHeaderPageIos = readPageIOs( currentBtreeOfBtreesOffset, Long.MAX_VALUE );
btreeOfBtrees = BTreeFactory.<NameRevision, Long> createPersistedBTree( BTreeTypeEnum.BTREE_OF_BTREES );
//BTreeFactory.<NameRevision, Long> setBtreeHeaderOffset( ( PersistedBTree<NameRevision, Long> )btreeOfBtrees, currentBtreeOfBtreesOffset );
loadBtree( bobHeaderPageIos, btreeOfBtrees );
// read the copied page B-tree
PageIO[] copiedPagesPageIos = readPageIOs( currentCopiedPagesBtreeOffset, Long.MAX_VALUE );
copiedPageBtree = BTreeFactory.<RevisionName, long[]> createPersistedBTree( BTreeTypeEnum.COPIED_PAGES_BTREE );
//( ( PersistedBTree<RevisionName, long[]> ) copiedPageBtree ).setBtreeHeaderOffset( currentCopiedPagesBtreeOffset );
loadBtree( copiedPagesPageIos, copiedPageBtree );
// Now, read all the B-trees from the btree of btrees
TupleCursor<NameRevision, Long> btreeCursor = btreeOfBtrees.browse();
Map<String, Long> loadedBtrees = new HashMap<String, Long>();
// loop on all the btrees we have, and keep only the latest revision
long currentRevision = -1L;
while ( btreeCursor.hasNext() )
{
Tuple<NameRevision, Long> btreeTuple = btreeCursor.next();
NameRevision nameRevision = btreeTuple.getKey();
long btreeOffset = btreeTuple.getValue();
long revision = nameRevision.getValue();
// Check if we already have processed this B-tree
Long loadedBtreeRevision = loadedBtrees.get( nameRevision.getName() );
if ( loadedBtreeRevision != null )
{
// The btree has already been loaded. The revision is necessarily higher
if ( revision > currentRevision )
{
// We have a newer revision : switch to the new revision (we keep the offset atm)
loadedBtrees.put( nameRevision.getName(), btreeOffset );
currentRevision = revision;
}
}
else
{
// This is a new B-tree
loadedBtrees.put( nameRevision.getName(), btreeOffset );
currentRevision = nameRevision.getRevision();
}
}
// TODO : clean up the old revisions...
// Now, we can load the real btrees using the offsets
for ( String btreeName : loadedBtrees.keySet() )
{
long btreeOffset = loadedBtrees.get( btreeName );
PageIO[] btreePageIos = readPageIOs( btreeOffset, Long.MAX_VALUE );
BTree<?, ?> btree = BTreeFactory.<NameRevision, Long> createPersistedBTree();
//( ( PersistedBTree<NameRevision, Long> ) btree ).setBtreeHeaderOffset( btreeOffset );
loadBtree( btreePageIos, btree );
// Add the btree into the map of managed B-trees
managedBtrees.put( btreeName, ( BTree<Object, Object> ) btree );
}
// We are done ! Let's finish with the last initialization parts
endOfFileOffset = fileChannel.size();
}
}
/**
* Starts a transaction
*/
public void beginTransaction()
{
if ( TXN_LOG.isDebugEnabled() )
{
TXN_LOG.debug( "Begining a new transaction on thread {}, TxnLevel {}",
Thread.currentThread().getName(), getTxnLevel() );
}
// First, take the lock if it's not already taken
if ( !( ( ReentrantLock ) transactionLock ).isHeldByCurrentThread() )
{
TXN_LOG.debug( "--> Lock taken" );
transactionLock.lock();
}
else
{
TXN_LOG.debug( "..o The current thread already holds the lock" );
}
// Now, check the TLS state
incrementTxnLevel();
}
/**
* Commits a transaction
*/
public void commit()
{
// We *must* own the transactionLock
if ( !transactionLock.isHeldByCurrentThread() )
{
String name = Thread.currentThread().getName();
String err = "This thread, '" + name + "' does not hold the transactionLock ";
TXN_LOG.error( err );
throw new RecordManagerException( err );
}
if ( TXN_LOG.isDebugEnabled() )
{
TXN_LOG.debug( "Committing a transaction on thread {}, TxnLevel {}",
Thread.currentThread().getName(), getTxnLevel() );
}
if ( !fileChannel.isOpen() )
{
// Still we have to decrement the TransactionLevel
int txnLevel = decrementTxnLevel();
if ( txnLevel == 0 )
{
// We can safely release the lock
// The file has been closed, nothing remains to commit, let's get out
transactionLock.unlock();
}
return;
}
int nbTxnStarted = context.get();
switch ( nbTxnStarted )
{
case ROLLBACKED_TXN:
// The transaction was rollbacked, quit immediatelly
transactionLock.unlock();
return;
case 1:
// We are done with the transaction, we can update the RMHeader and swap the BTreeHeaders
// First update the RMHeader to be sure that we have a way to restore from a crash
updateRecordManagerHeader();
// Swap the BtreeHeaders maps
swapCurrentBtreeHeaders();
// We can now free pages
for ( PageIO pageIo : freedPages )
{
try
{
free( pageIo );
}
catch ( IOException ioe )
{
throw new RecordManagerException( ioe.getMessage() );
}
}
// Release the allocated and freed pages list
freedPages.clear();
allocatedPages.clear();
// And update the RMHeader again, removing the old references to BOB and CPB b-tree headers
// here, we have to erase the old references to keep only the new ones.
updateRecordManagerHeader();
commitCount++;
if ( commitCount >= pageReclaimerThreshold )
{
runReclaimer();
}
// Finally, decrement the number of started transactions
// and release the global lock if possible
int txnLevel = decrementTxnLevel();
if ( txnLevel == 0 )
{
transactionLock.unlock();
}
return;
default:
// We are inner an existing transaction. Just update the necessary elements
// Update the RMHeader to be sure that we have a way to restore from a crash
updateRecordManagerHeader();
// Swap the BtreeHeaders maps
//swapCurrentBtreeHeaders();
// We can now free pages
for ( PageIO pageIo : freedPages )
{
try
{
free( pageIo );
}
catch ( IOException ioe )
{
throw new RecordManagerException( ioe.getMessage() );
}
}
// Release the allocated and freed pages list
freedPages.clear();
allocatedPages.clear();
// And update the RMHeader again, removing the old references to BOB and CPB b-tree headers
// here, we have to erase the old references to keep only the new ones.
updateRecordManagerHeader();
commitCount++;
if ( commitCount >= pageReclaimerThreshold )
{
runReclaimer();
}
// Finally, decrement the number of started transactions
// and release the global lock
txnLevel = decrementTxnLevel();
if ( txnLevel == 0 )
{
transactionLock.unlock();
}
return;
}
}
public boolean isContextOk()
{
return ( context == null ? true : ( context.get() == 0 ) );
}
/**
* Get the transactionLevel, ie the number of encapsulated update ops
*/
private int getTxnLevel()
{
Integer nbTxnLevel = context.get();
if ( nbTxnLevel == null )
{
return -1;
}
return nbTxnLevel;
}
/**
* Increment the transactionLevel
*/
private void incrementTxnLevel()
{
Integer nbTxnLevel = context.get();
if ( nbTxnLevel == null )
{
context.set( 1 );
}
else
{
// And increment the counter of inner txn.
context.set( nbTxnLevel + 1 );
}
if ( TXN_LOG.isDebugEnabled() )
{
TXN_LOG.debug( "Incrementing the TxnLevel : {}", context.get() );
}
}
/**
* Decrement the transactionLevel
*/
private int decrementTxnLevel()
{
int nbTxnStarted = context.get() - 1;
context.set( nbTxnStarted );
if ( TXN_LOG.isDebugEnabled() )
{
TXN_LOG.debug( "Decrementing the TxnLevel : {}", context.get() );
}
return nbTxnStarted;
}
/**
* Rollback a transaction
*/
public void rollback()
{
// We *must* own the transactionLock
if ( !transactionLock.isHeldByCurrentThread() )
{
TXN_LOG.error( "This thread does not hold the transactionLock" );
throw new RecordManagerException( "This thread does not hold the transactionLock" );
}
if ( TXN_LOG.isDebugEnabled() )
{
TXN_LOG.debug( "Rollbacking a new transaction on thread {}, TxnLevel {}",
Thread.currentThread().getName(), getTxnLevel() );
}
// Reset the counter
context.set( ROLLBACKED_TXN );
// We can now free allocated pages, this is the end of the transaction
for ( PageIO pageIo : allocatedPages )
{
try
{
free( pageIo );
}
catch ( IOException ioe )
{
throw new RecordManagerException( ioe.getMessage() );
}
}
// Release the allocated and freed pages list
freedPages.clear();
allocatedPages.clear();
// And update the RMHeader
updateRecordManagerHeader();
// And restore the BTreeHeaders new Map to the current state
revertBtreeHeaders();
// This is an all-of-nothing operation : we can't have a transaction within
// a transaction that would survive an inner transaction rollback.
transactionLock.unlock();
}
/**
* Reads all the PageIOs that are linked to the page at the given position, including
* the first page.
*
* @param position The position of the first page
* @param limit The maximum bytes to read. Set this value to -1 when the size is unknown.
* @return An array of pages
*/
/*no qualifier*/PageIO[] readPageIOs( long position, long limit ) throws IOException, EndOfFileExceededException
{
LOG.debug( "Read PageIOs at position {}", position );
if ( limit <= 0 )
{
limit = Long.MAX_VALUE;
}
PageIO firstPage = fetchPage( position );
firstPage.setSize();
List<PageIO> listPages = new ArrayList<PageIO>();
listPages.add( firstPage );
long dataRead = pageSize - LONG_SIZE - INT_SIZE;
// Iterate on the pages, if needed
long nextPage = firstPage.getNextPage();
if ( ( dataRead < limit ) && ( nextPage != NO_PAGE ) )
{
while ( dataRead < limit )
{
PageIO page = fetchPage( nextPage );
listPages.add( page );
nextPage = page.getNextPage();
dataRead += pageSize - LONG_SIZE;
if ( nextPage == NO_PAGE )
{
page.setNextPage( NO_PAGE );
break;
}
}
}
LOG.debug( "Nb of PageIOs read : {}", listPages.size() );
// Return
return listPages.toArray( new PageIO[]
{} );
}
/**
* Check the offset to be sure it's a valid one :
* <ul>
* <li>It's >= 0</li>
* <li>It's below the end of the file</li>
* <li>It's a multipl of the pageSize
* </ul>
* @param offset The offset to check
* @throws InvalidOffsetException If the offset is not valid
*/
/* no qualifier */void checkOffset( long offset )
{
if ( ( offset < 0 ) || ( offset > endOfFileOffset ) || ( ( offset % pageSize ) != 0 ) )
{
throw new InvalidOffsetException( "Bad Offset : " + offset );
}
}
/**
* Read a B-tree from the disk. The meta-data are at the given position in the list of pages.
* We load a B-tree in two steps : first, we load the B-tree header, then the common informations
*
* @param pageIos The list of pages containing the meta-data
* @param btree The B-tree we have to initialize
* @throws InstantiationException
* @throws IllegalAccessException
* @throws ClassNotFoundException
* @throws NoSuchFieldException
* @throws SecurityException
* @throws IllegalArgumentException
*/
private <K, V> void loadBtree( PageIO[] pageIos, BTree<K, V> btree ) throws EndOfFileExceededException,
IOException, ClassNotFoundException, IllegalAccessException, InstantiationException, IllegalArgumentException,
SecurityException, NoSuchFieldException
{
loadBtree( pageIos, btree, null );
}
/**
* Read a B-tree from the disk. The meta-data are at the given position in the list of pages.
* We load a B-tree in two steps : first, we load the B-tree header, then the common informations
*
* @param pageIos The list of pages containing the meta-data
* @param btree The B-tree we have to initialize
* @throws InstantiationException
* @throws IllegalAccessException
* @throws ClassNotFoundException
* @throws NoSuchFieldException
* @throws SecurityException
* @throws IllegalArgumentException
*/
/* no qualifier */<K, V> void loadBtree( PageIO[] pageIos, BTree btree, BTree<K, V> parentBTree )
throws EndOfFileExceededException,
IOException, ClassNotFoundException, IllegalAccessException, InstantiationException, IllegalArgumentException,
SecurityException, NoSuchFieldException
{
long dataPos = 0L;
// Process the B-tree header
BTreeHeader<K, V> btreeHeader = new BTreeHeader<K, V>();
btreeHeader.setBtree( btree );
// The BtreeHeader offset
btreeHeader.setBTreeHeaderOffset( pageIos[0].getOffset() );
// The B-tree current revision
long revision = readLong( pageIos, dataPos );
btreeHeader.setRevision( revision );
dataPos += LONG_SIZE;
// The nb elems in the tree
long nbElems = readLong( pageIos, dataPos );
btreeHeader.setNbElems( nbElems );
dataPos += LONG_SIZE;
// The B-tree rootPage offset
long rootPageOffset = readLong( pageIos, dataPos );
btreeHeader.setRootPageOffset( rootPageOffset );
dataPos += LONG_SIZE;
// The B-tree information offset
long btreeInfoOffset = readLong( pageIos, dataPos );
// Now, process the common informations
PageIO[] infoPageIos = readPageIOs( btreeInfoOffset, Long.MAX_VALUE );
( ( PersistedBTree<K, V> ) btree ).setBtreeInfoOffset( infoPageIos[0].getOffset() );
dataPos = 0L;
// The B-tree page size
int btreePageSize = readInt( infoPageIos, dataPos );
BTreeFactory.setPageSize( btree, btreePageSize );
dataPos += INT_SIZE;
// The tree name
ByteBuffer btreeNameBytes = readBytes( infoPageIos, dataPos );
dataPos += INT_SIZE + btreeNameBytes.limit();
String btreeName = Strings.utf8ToString( btreeNameBytes );
BTreeFactory.setName( btree, btreeName );
// The keySerializer FQCN
ByteBuffer keySerializerBytes = readBytes( infoPageIos, dataPos );
dataPos += INT_SIZE + keySerializerBytes.limit();
String keySerializerFqcn = "";
if ( keySerializerBytes != null )
{
keySerializerFqcn = Strings.utf8ToString( keySerializerBytes );
}
BTreeFactory.setKeySerializer( btree, keySerializerFqcn );
// The valueSerialier FQCN
ByteBuffer valueSerializerBytes = readBytes( infoPageIos, dataPos );
String valueSerializerFqcn = "";
dataPos += INT_SIZE + valueSerializerBytes.limit();
if ( valueSerializerBytes != null )
{
valueSerializerFqcn = Strings.utf8ToString( valueSerializerBytes );
}
BTreeFactory.setValueSerializer( btree, valueSerializerFqcn );
// The B-tree allowDuplicates flag
int allowDuplicates = readInt( infoPageIos, dataPos );
( ( PersistedBTree<K, V> ) btree ).setAllowDuplicates( allowDuplicates != 0 );
dataPos += INT_SIZE;
// Set the recordManager in the btree
( ( PersistedBTree<K, V> ) btree ).setRecordManager( this );
// Set the current revision to the one stored in the B-tree header
// Here, we have to tell the BTree to keep this revision in the
// btreeRevisions Map, thus the 'true' parameter at the end.
( ( PersistedBTree<K, V> ) btree ).storeRevision( btreeHeader, true );
// Now, init the B-tree
( ( PersistedBTree<K, V> ) btree ).init( parentBTree );
// Update the BtreeHeaders Maps
currentBTreeHeaders.put( btree.getName(), ( ( PersistedBTree<K, V> ) btree ).getBtreeHeader() );
newBTreeHeaders.put( btree.getName(), ( ( PersistedBTree<K, V> ) btree ).getBtreeHeader() );
// Read the rootPage pages on disk
PageIO[] rootPageIos = readPageIOs( rootPageOffset, Long.MAX_VALUE );
Page<K, V> btreeRoot = readPage( btree, rootPageIos );
BTreeFactory.setRecordManager( btree, this );
BTreeFactory.setRootPage( btree, btreeRoot );
}
/**
* Deserialize a Page from a B-tree at a give position
*
* @param btree The B-tree we want to read a Page from
* @param offset The position in the file for this page
* @return The read page
* @throws EndOfFileExceededException If we have reached the end of the file while reading the page
*/
public <K, V> Page<K, V> deserialize( BTree<K, V> btree, long offset ) throws EndOfFileExceededException,
IOException
{
checkOffset( offset );
PageIO[] rootPageIos = readPageIOs( offset, Long.MAX_VALUE );
Page<K, V> page = readPage( btree, rootPageIos );
return page;
}
/**
* Read a page from some PageIO for a given B-tree
* @param btree The B-tree we want to read a page for
* @param pageIos The PageIO containing the raw data
* @return The read Page if successful
* @throws IOException If the deserialization failed
*/
private <K, V> Page<K, V> readPage( BTree<K, V> btree, PageIO[] pageIos ) throws IOException
{
// Deserialize the rootPage now
long position = 0L;
// The revision
long revision = readLong( pageIos, position );
position += LONG_SIZE;
// The number of elements in the page
int nbElems = readInt( pageIos, position );
position += INT_SIZE;
// The size of the data containing the keys and values
Page<K, V> page = null;
// Reads the bytes containing all the keys and values, if we have some
// We read big blog of data into ByteBuffer, then we will process
// this ByteBuffer
ByteBuffer byteBuffer = readBytes( pageIos, position );
// Now, deserialize the data block. If the number of elements
// is positive, it's a Leaf, otherwise it's a Node
// Note that only a leaf can have 0 elements, and it's the root page then.
if ( nbElems >= 0 )
{
// It's a leaf
page = readLeafKeysAndValues( btree, nbElems, revision, byteBuffer, pageIos );
}
else
{
// It's a node
page = readNodeKeysAndValues( btree, -nbElems, revision, byteBuffer, pageIos );
}
( ( AbstractPage<K, V> ) page ).setOffset( pageIos[0].getOffset() );
if ( pageIos.length > 1 )
{
( ( AbstractPage<K, V> ) page ).setLastOffset( pageIos[pageIos.length - 1].getOffset() );
}
return page;
}
/**
* Deserialize a Leaf from some PageIOs
*/
private <K, V> PersistedLeaf<K, V> readLeafKeysAndValues( BTree<K, V> btree, int nbElems, long revision,
ByteBuffer byteBuffer, PageIO[] pageIos )
{
// Its a leaf, create it
PersistedLeaf<K, V> leaf = ( PersistedLeaf<K, V> ) BTreeFactory.createLeaf( btree, revision, nbElems );
// Store the page offset on disk
leaf.setOffset( pageIos[0].getOffset() );
leaf.setLastOffset( pageIos[pageIos.length - 1].getOffset() );
int[] keyLengths = new int[nbElems];
int[] valueLengths = new int[nbElems];
boolean isNotSubTree = ( btree.getType() != BTreeTypeEnum.PERSISTED_SUB );
// Read each key and value
for ( int i = 0; i < nbElems; i++ )
{
if ( isNotSubTree )
{
// Read the number of values
int nbValues = byteBuffer.getInt();
PersistedValueHolder<V> valueHolder = null;
if ( nbValues < 0 )
{
// This is a sub-btree
byte[] btreeOffsetBytes = new byte[LONG_SIZE];
byteBuffer.get( btreeOffsetBytes );
// Create the valueHolder. As the number of values is negative, we have to switch
// to a positive value but as we start at -1 for 0 value, add 1.
valueHolder = new PersistedValueHolder<V>( btree, 1 - nbValues, btreeOffsetBytes );
}
else
{
// This is an array
// Read the value's array length
valueLengths[i] = byteBuffer.getInt();
// This is an Array of values, read the byte[] associated with it
byte[] arrayBytes = new byte[valueLengths[i]];
byteBuffer.get( arrayBytes );
valueHolder = new PersistedValueHolder<V>( btree, nbValues, arrayBytes );
}
BTreeFactory.setValue( btree, leaf, i, valueHolder );
}
keyLengths[i] = byteBuffer.getInt();
byte[] data = new byte[keyLengths[i]];
byteBuffer.get( data );
BTreeFactory.setKey( btree, leaf, i, data );
}
return leaf;
}
/**
* Deserialize a Node from some PageIos
*/
private <K, V> PersistedNode<K, V> readNodeKeysAndValues( BTree<K, V> btree, int nbElems, long revision,
ByteBuffer byteBuffer, PageIO[] pageIos ) throws IOException
{
PersistedNode<K, V> node = ( PersistedNode<K, V> ) BTreeFactory.createNode( btree, revision, nbElems );
// Read each value and key
for ( int i = 0; i < nbElems; i++ )
{
// This is an Offset
long offset = LongSerializer.INSTANCE.deserialize( byteBuffer );
long lastOffset = LongSerializer.INSTANCE.deserialize( byteBuffer );
PersistedPageHolder<K, V> valueHolder = new PersistedPageHolder<K, V>( btree, null, offset, lastOffset );
node.setValue( i, valueHolder );
// Read the key length
int keyLength = byteBuffer.getInt();
int currentPosition = byteBuffer.position();
// and the key value
K key = btree.getKeySerializer().deserialize( byteBuffer );
// Set the new position now
byteBuffer.position( currentPosition + keyLength );
BTreeFactory.setKey( btree, node, i, key );
}
// and read the last value, as it's a node
long offset = LongSerializer.INSTANCE.deserialize( byteBuffer );
long lastOffset = LongSerializer.INSTANCE.deserialize( byteBuffer );
PersistedPageHolder<K, V> valueHolder = new PersistedPageHolder<K, V>( btree, null, offset, lastOffset );
node.setValue( nbElems, valueHolder );
return node;
}
/**
* Read a byte[] from pages.
*
* @param pageIos The pages we want to read the byte[] from
* @param position The position in the data stored in those pages
* @return The byte[] we have read
*/
/* no qualifier */ByteBuffer readBytes( PageIO[] pageIos, long position )
{
// Read the byte[] length first
int length = readInt( pageIos, position );
position += INT_SIZE;
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
// Check that the length is correct : it should fit in the provided pageIos
int pageEnd = computePageNb( position + length );
if ( pageEnd > pageIos.length )
{
// This is wrong...
LOG.error( "Wrong size : {}, it's larger than the number of provided pages {}", length, pageIos.length );
throw new ArrayIndexOutOfBoundsException();
}
ByteBuffer pageData = pageIos[pageNb].getData();
int remaining = pageData.capacity() - pagePos;
if ( length == 0 )
{
// No bytes to read : return null;
return null;
}
else
{
ByteBuffer bytes = ByteBuffer.allocate( length );
while ( length > 0 )
{
if ( length <= remaining )
{
pageData.mark();
pageData.position( pagePos );
int oldLimit = pageData.limit();
pageData.limit( pagePos + length );
bytes.put( pageData );
pageData.limit( oldLimit );
pageData.reset();
bytes.rewind();
return bytes;
}
pageData.mark();
pageData.position( pagePos );
int oldLimit = pageData.limit();
pageData.limit( pagePos + remaining );
bytes.put( pageData );
pageData.limit( oldLimit );
pageData.reset();
pageNb++;
pagePos = LINK_SIZE;
pageData = pageIos[pageNb].getData();
length -= remaining;
remaining = pageData.capacity() - pagePos;
}
bytes.rewind();
return bytes;
}
}
/**
* Read an int from pages
* @param pageIos The pages we want to read the int from
* @param position The position in the data stored in those pages
* @return The int we have read
*/
/* no qualifier */int readInt( PageIO[] pageIos, long position )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
ByteBuffer pageData = pageIos[pageNb].getData();
int remaining = pageData.capacity() - pagePos;
int value = 0;
if ( remaining >= INT_SIZE )
{
value = pageData.getInt( pagePos );
}
else
{
value = 0;
switch ( remaining )
{
case 3:
value += ( ( pageData.get( pagePos + 2 ) & 0x00FF ) << 8 );
// Fallthrough !!!
case 2:
value += ( ( pageData.get( pagePos + 1 ) & 0x00FF ) << 16 );
// Fallthrough !!!
case 1:
value += ( pageData.get( pagePos ) << 24 );
break;
}
// Now deal with the next page
pageData = pageIos[pageNb + 1].getData();
pagePos = LINK_SIZE;
switch ( remaining )
{
case 1:
value += ( pageData.get( pagePos ) & 0x00FF ) << 16;
// fallthrough !!!
case 2:
value += ( pageData.get( pagePos + 2 - remaining ) & 0x00FF ) << 8;
// fallthrough !!!
case 3:
value += ( pageData.get( pagePos + 3 - remaining ) & 0x00FF );
break;
}
}
return value;
}
/**
* Read a byte from pages
* @param pageIos The pages we want to read the byte from
* @param position The position in the data stored in those pages
* @return The byte we have read
*/
private byte readByte( PageIO[] pageIos, long position )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
ByteBuffer pageData = pageIos[pageNb].getData();
byte value = 0;
value = pageData.get( pagePos );
return value;
}
/**
* Read a long from pages
* @param pageIos The pages we want to read the long from
* @param position The position in the data stored in those pages
* @return The long we have read
*/
/* no qualifier */long readLong( PageIO[] pageIos, long position )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
ByteBuffer pageData = pageIos[pageNb].getData();
int remaining = pageData.capacity() - pagePos;
long value = 0L;
if ( remaining >= LONG_SIZE )
{
value = pageData.getLong( pagePos );
}
else
{
switch ( remaining )
{
case 7:
value += ( ( ( long ) pageData.get( pagePos + 6 ) & 0x00FF ) << 8 );
// Fallthrough !!!
case 6:
value += ( ( ( long ) pageData.get( pagePos + 5 ) & 0x00FF ) << 16 );
// Fallthrough !!!
case 5:
value += ( ( ( long ) pageData.get( pagePos + 4 ) & 0x00FF ) << 24 );
// Fallthrough !!!
case 4:
value += ( ( ( long ) pageData.get( pagePos + 3 ) & 0x00FF ) << 32 );
// Fallthrough !!!
case 3:
value += ( ( ( long ) pageData.get( pagePos + 2 ) & 0x00FF ) << 40 );
// Fallthrough !!!
case 2:
value += ( ( ( long ) pageData.get( pagePos + 1 ) & 0x00FF ) << 48 );
// Fallthrough !!!
case 1:
value += ( ( long ) pageData.get( pagePos ) << 56 );
break;
}
// Now deal with the next page
pageData = pageIos[pageNb + 1].getData();
pagePos = LINK_SIZE;
switch ( remaining )
{
case 1:
value += ( ( long ) pageData.get( pagePos ) & 0x00FF ) << 48;
// fallthrough !!!
case 2:
value += ( ( long ) pageData.get( pagePos + 2 - remaining ) & 0x00FF ) << 40;
// fallthrough !!!
case 3:
value += ( ( long ) pageData.get( pagePos + 3 - remaining ) & 0x00FF ) << 32;
// fallthrough !!!
case 4:
value += ( ( long ) pageData.get( pagePos + 4 - remaining ) & 0x00FF ) << 24;
// fallthrough !!!
case 5:
value += ( ( long ) pageData.get( pagePos + 5 - remaining ) & 0x00FF ) << 16;
// fallthrough !!!
case 6:
value += ( ( long ) pageData.get( pagePos + 6 - remaining ) & 0x00FF ) << 8;
// fallthrough !!!
case 7:
value += ( ( long ) pageData.get( pagePos + 7 - remaining ) & 0x00FF );
break;
}
}
return value;
}
/**
* Manage a B-tree. The btree will be added and managed by this RecordManager. We will create a
* new RootPage for this added B-tree, which will contain no data.<br/>
* This method is threadsafe.
* Managing a btree is a matter of storing an reference to the managed B-tree in the B-tree Of B-trees.
* We store a tuple of NameRevision (where revision is 0L) and a offset to the B-tree header.
* At the same time, we keep a track of the managed B-trees in a Map.
*
* @param btree The new B-tree to manage.
* @param treeType flag indicating if this is an internal tree
*
* @throws BTreeAlreadyManagedException If the B-tree is already managed
* @throws IOException if there was a problem while accessing the file
*/
public synchronized <K, V> void manage( BTree<K, V> btree ) throws BTreeAlreadyManagedException, IOException
{
beginTransaction();
try
{
LOG.debug( "Managing the btree {}", btree.getName() );
BTreeFactory.setRecordManager( btree, this );
String name = btree.getName();
if ( managedBtrees.containsKey( name ) )
{
// There is already a B-tree with this name in the recordManager...
LOG.error( "There is already a B-tree named '{}' managed by this recordManager", name );
rollback();
throw new BTreeAlreadyManagedException( name );
}
// Now, write the B-tree informations
long btreeInfoOffset = writeBtreeInfo( btree );
BTreeHeader<K, V> btreeHeader = ( ( AbstractBTree<K, V> ) btree ).getBtreeHeader();
( ( PersistedBTree<K, V> ) btree ).setBtreeInfoOffset( btreeInfoOffset );
// Serialize the B-tree root page
Page<K, V> rootPage = btreeHeader.getRootPage();
PageIO[] rootPageIos = serializePage( btree, btreeHeader.getRevision(), rootPage );
// Get the reference on the first page
long rootPageOffset = rootPageIos[0].getOffset();
// Store the rootPageOffset into the Btree header and into the rootPage
btreeHeader.setRootPageOffset( rootPageOffset );
( ( PersistedLeaf<K, V> ) rootPage ).setOffset( rootPageOffset );
LOG.debug( "Flushing the newly managed '{}' btree rootpage", btree.getName() );
flushPages( rootPageIos );
// And the B-tree header
long btreeHeaderOffset = writeBtreeHeader( btree, btreeHeader );
// Now, if this is a new B-tree, add it to the B-tree of B-trees
// Add the btree into the map of managed B-trees
managedBtrees.put( name, ( BTree<Object, Object> ) btree );
// And in the Map of currentBtreeHeaders and newBtreeHeaders
currentBTreeHeaders.put( name, btreeHeader );
newBTreeHeaders.put( name, btreeHeader );
// We can safely increment the number of managed B-trees
nbBtree++;
// Create the new NameRevision
NameRevision nameRevision = new NameRevision( name, 0L );
// Inject it into the B-tree of B-tree
btreeOfBtrees.insert( nameRevision, btreeHeaderOffset );
commit();
}
catch ( IOException ioe )
{
rollback();
throw ioe;
}
}
/**
* Managing a btree is a matter of storing an reference to the managed B-tree in the B-tree Of B-trees.
* We store a tuple of NameRevision (where revision is 0L) and a offset to the B-tree header.
* At the same time, we keep a track of the managed B-trees in a Map.
*
* @param btree The new B-tree to manage.
* @param treeType flag indicating if this is an internal tree
*
* @throws BTreeAlreadyManagedException If the B-tree is already managed
* @throws IOException
*/
public synchronized <K, V> void manageSubBtree( BTree<K, V> btree )
throws BTreeAlreadyManagedException, IOException
{
LOG.debug( "Managing the sub-btree {}", btree.getName() );
BTreeFactory.setRecordManager( btree, this );
String name = btree.getName();
if ( managedBtrees.containsKey( name ) )
{
// There is already a subB-tree with this name in the recordManager...
LOG.error( "There is already a sub-B-tree named '{}' managed by this recordManager", name );
throw new BTreeAlreadyManagedException( name );
}
// Now, write the subB-tree informations
long btreeInfoOffset = writeBtreeInfo( btree );
BTreeHeader<K, V> btreeHeader = ( ( AbstractBTree<K, V> ) btree ).getBtreeHeader();
( ( PersistedBTree<K, V> ) btree ).setBtreeInfoOffset( btreeInfoOffset );
// Serialize the B-tree root page
Page<K, V> rootPage = btreeHeader.getRootPage();
PageIO[] rootPageIos = serializePage( btree, btreeHeader.getRevision(), rootPage );
// Get the reference on the first page
long rootPageOffset = rootPageIos[0].getOffset();
// Store the rootPageOffset into the Btree header and into the rootPage
btreeHeader.setRootPageOffset( rootPageOffset );
( ( AbstractPage<K, V> ) rootPage ).setOffset( rootPageOffset );
LOG.debug( "Flushing the newly managed '{}' btree rootpage", btree.getName() );
flushPages( rootPageIos );
// And the B-tree header
long btreeHeaderOffset = writeBtreeHeader( btree, btreeHeader );
// Now, if this is a new B-tree, add it to the B-tree of B-trees
// Add the btree into the map of managed B-trees
if ( ( btree.getType() != BTreeTypeEnum.BTREE_OF_BTREES ) &&
( btree.getType() != BTreeTypeEnum.COPIED_PAGES_BTREE ) &&
( btree.getType() != BTreeTypeEnum.PERSISTED_SUB ) )
{
managedBtrees.put( name, ( BTree<Object, Object> ) btree );
}
// And in the Map of currentBtreeHeaders and newBtreeHeaders
currentBTreeHeaders.put( name, btreeHeader );
newBTreeHeaders.put( name, btreeHeader );
// Create the new NameRevision
NameRevision nameRevision = new NameRevision( name, 0L );
// Inject it into the B-tree of B-tree
if ( ( btree.getType() != BTreeTypeEnum.BTREE_OF_BTREES ) &&
( btree.getType() != BTreeTypeEnum.COPIED_PAGES_BTREE ) &&
( btree.getType() != BTreeTypeEnum.PERSISTED_SUB ) )
{
// We can safely increment the number of managed B-trees
nbBtree++;
btreeOfBtrees.insert( nameRevision, btreeHeaderOffset );
}
updateRecordManagerHeader();
}
/**
* Serialize a new Page. It will contain the following data :<br/>
* <ul>
* <li>the revision : a long</li>
* <li>the number of elements : an int (if <= 0, it's a Node, otherwise it's a Leaf)</li>
* <li>the size of the values/keys when serialized
* <li>the keys : an array of serialized keys</li>
* <li>the values : an array of references to the children pageIO offset (stored as long)
* if it's a Node, or a list of values if it's a Leaf</li>
* <li></li>
* </ul>
*
* @param revision The node revision
* @param keys The keys to serialize
* @param children The references to the children
* @return An array of pages containing the serialized node
* @throws IOException
*/
private <K, V> PageIO[] serializePage( BTree<K, V> btree, long revision, Page<K, V> page ) throws IOException
{
int nbElems = page.getNbElems();
boolean isNotSubTree = ( btree.getType() != BTreeTypeEnum.PERSISTED_SUB );
if ( nbElems == 0 )
{
return serializeRootPage( revision );
}
else
{
// Prepare a list of byte[] that will contain the serialized page
int nbBuffers = 1 + 1 + 1 + nbElems * 3;
int dataSize = 0;
int serializedSize = 0;
if ( page.isNode() )
{
// A Node has one more value to store
nbBuffers++;
}
// Now, we can create the list with the right size
List<byte[]> serializedData = new ArrayList<byte[]>( nbBuffers );
// The revision
byte[] buffer = LongSerializer.serialize( revision );
serializedData.add( buffer );
serializedSize += buffer.length;
// The number of elements
// Make it a negative value if it's a Node
int pageNbElems = nbElems;
if ( page.isNode() )
{
pageNbElems = -nbElems;
}
buffer = IntSerializer.serialize( pageNbElems );
serializedData.add( buffer );
serializedSize += buffer.length;
// Iterate on the keys and values. We first serialize the value, then the key
// until we are done with all of them. If we are serializing a page, we have
// to serialize one more value
for ( int pos = 0; pos < nbElems; pos++ )
{
// Start with the value
if ( page.isNode() )
{
dataSize += serializeNodeValue( ( PersistedNode<K, V> ) page, pos, serializedData );
dataSize += serializeNodeKey( ( PersistedNode<K, V> ) page, pos, serializedData );
}
else
{
if ( isNotSubTree )
{
dataSize += serializeLeafValue( ( PersistedLeaf<K, V> ) page, pos, serializedData );
}
dataSize += serializeLeafKey( ( PersistedLeaf<K, V> ) page, pos, serializedData );
}
}
// Nodes have one more value to serialize
if ( page.isNode() )
{
dataSize += serializeNodeValue( ( PersistedNode<K, V> ) page, nbElems, serializedData );
}
// Store the data size
buffer = IntSerializer.serialize( dataSize );
serializedData.add( 2, buffer );
serializedSize += buffer.length;
serializedSize += dataSize;
// We are done. Allocate the pages we need to store the data
PageIO[] pageIos = getFreePageIOs( serializedSize );
// And store the data into those pages
long position = 0L;
for ( byte[] bytes : serializedData )
{
position = storeRaw( position, bytes, pageIos );
}
return pageIos;
}
}
/**
* Serialize a Node's key
*/
private <K, V> int serializeNodeKey( PersistedNode<K, V> node, int pos, List<byte[]> serializedData )
{
KeyHolder<K> holder = node.getKeyHolder( pos );
byte[] buffer = ( ( PersistedKeyHolder<K> ) holder ).getRaw();
// We have to store the serialized key length
byte[] length = IntSerializer.serialize( buffer.length );
serializedData.add( length );
// And store the serialized key now if not null
if ( buffer.length != 0 )
{
serializedData.add( buffer );
}
return buffer.length + INT_SIZE;
}
/**
* Serialize a Node's Value. We store the two offsets of the child page.
*/
private <K, V> int serializeNodeValue( PersistedNode<K, V> node, int pos, List<byte[]> serializedData )
throws IOException
{
// For a node, we just store the children's offsets
Page<K, V> child = node.getReference( pos );
// The first offset
byte[] buffer = LongSerializer.serialize( ( ( AbstractPage<K, V> ) child ).getOffset() );
serializedData.add( buffer );
int dataSize = buffer.length;
// The last offset
buffer = LongSerializer.serialize( ( ( AbstractPage<K, V> ) child ).getLastOffset() );
serializedData.add( buffer );
dataSize += buffer.length;
return dataSize;
}
/**
* Serialize a Leaf's key
*/
private <K, V> int serializeLeafKey( PersistedLeaf<K, V> leaf, int pos, List<byte[]> serializedData )
{
int dataSize = 0;
KeyHolder<K> keyHolder = leaf.getKeyHolder( pos );
byte[] keyData = ( ( PersistedKeyHolder<K> ) keyHolder ).getRaw();
if ( keyData != null )
{
// We have to store the serialized key length
byte[] length = IntSerializer.serialize( keyData.length );
serializedData.add( length );
// And the key data
serializedData.add( keyData );
dataSize += keyData.length + INT_SIZE;
}
else
{
serializedData.add( IntSerializer.serialize( 0 ) );
dataSize += INT_SIZE;
}
return dataSize;
}
/**
* Serialize a Leaf's Value.
*/
private <K, V> int serializeLeafValue( PersistedLeaf<K, V> leaf, int pos, List<byte[]> serializedData )
throws IOException
{
// The value can be an Array or a sub-btree, but we don't care
// we just iterate on all the values
ValueHolder<V> valueHolder = leaf.getValue( pos );
int dataSize = 0;
int nbValues = valueHolder.size();
if ( nbValues == 0 )
{
// No value.
byte[] buffer = IntSerializer.serialize( nbValues );
serializedData.add( buffer );
return buffer.length;
}
if ( !valueHolder.isSubBtree() )
{
// Write the nb elements first
byte[] buffer = IntSerializer.serialize( nbValues );
serializedData.add( buffer );
dataSize = INT_SIZE;
// We have a serialized value. Just flush it
byte[] data = ( ( PersistedValueHolder<V> ) valueHolder ).getRaw();
dataSize += data.length;
// Store the data size
buffer = IntSerializer.serialize( data.length );
serializedData.add( buffer );
dataSize += INT_SIZE;
// and add the data if it's not 0
if ( data.length > 0 )
{
serializedData.add( data );
}
}
else
{
// Store the nbVlues as a negative number. We add 1 so that 0 is not confused with an Array value
byte[] buffer = IntSerializer.serialize( -( nbValues + 1 ) );
serializedData.add( buffer );
dataSize += buffer.length;
// the B-tree offset
buffer = LongSerializer.serialize( ( ( PersistedValueHolder<V> ) valueHolder ).getOffset() );
serializedData.add( buffer );
dataSize += buffer.length;
}
return dataSize;
}
/**
* Write a root page with no elements in it
*/
private PageIO[] serializeRootPage( long revision ) throws IOException
{
// We will have 1 single page if we have no elements
PageIO[] pageIos = new PageIO[1];
// This is either a new root page or a new page that will be filled later
PageIO newPage = fetchNewPage();
// We need first to create a byte[] that will contain all the data
// For the root page, this is easy, as we only have to store the revision,
// and the number of elements, which is 0.
long position = 0L;
position = store( position, revision, newPage );
position = store( position, 0, newPage );
// Update the page size now
newPage.setSize( ( int ) position );
// Insert the result into the array of PageIO
pageIos[0] = newPage;
return pageIos;
}
/**
* Update the RecordManager header, injecting the following data :
*
* <pre>
* +---------------------+
* | PageSize | 4 bytes : The size of a physical page (default to 4096)
* +---------------------+
* | NbTree | 4 bytes : The number of managed B-trees (at least 1)
* +---------------------+
* | FirstFree | 8 bytes : The offset of the first free page
* +---------------------+
* | current BoB offset | 8 bytes : The offset of the current B-tree of B-trees
* +---------------------+
* | previous BoB offset | 8 bytes : The offset of the previous B-tree of B-trees
* +---------------------+
* | current CP offset | 8 bytes : The offset of the current CopiedPages B-tree
* +---------------------+
* | previous CP offset | 8 bytes : The offset of the previous CopiedPages B-tree
* +---------------------+
* </pre>
*/
public void updateRecordManagerHeader()
{
// The page size
int position = writeData( RECORD_MANAGER_HEADER_BYTES, 0, pageSize );
// The number of managed B-tree
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, nbBtree );
// The first free page
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, firstFreePage );
// The offset of the current B-tree of B-trees
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, currentBtreeOfBtreesOffset );
// The offset of the copied pages B-tree
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, previousBtreeOfBtreesOffset );
// The offset of the current B-tree of B-trees
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, currentCopiedPagesBtreeOffset );
// The offset of the copied pages B-tree
position = writeData( RECORD_MANAGER_HEADER_BYTES, position, previousCopiedPagesBtreeOffset );
// Write the RecordManager header on disk
RECORD_MANAGER_HEADER_BUFFER.put( RECORD_MANAGER_HEADER_BYTES );
RECORD_MANAGER_HEADER_BUFFER.flip();
LOG.debug( "Update RM header" );
if ( LOG_PAGES.isDebugEnabled() )
{
StringBuilder sb = new StringBuilder();
sb.append( "First free page : 0x" ).append( Long.toHexString( firstFreePage ) ).append( "\n" );
sb.append( "Current BOB header : 0x" ).append( Long.toHexString( currentBtreeOfBtreesOffset ) ).append( "\n" );
sb.append( "Previous BOB header : 0x" ).append( Long.toHexString( previousBtreeOfBtreesOffset ) ).append( "\n" );
sb.append( "Current CPB header : 0x" ).append( Long.toHexString( currentCopiedPagesBtreeOffset ) ).append( "\n" );
sb.append( "Previous CPB header : 0x" ).append( Long.toHexString( previousCopiedPagesBtreeOffset ) ).append( "\n" );
if ( firstFreePage != NO_PAGE )
{
long freePage = firstFreePage;
sb.append( "free pages list : " );
boolean isFirst = true;
while ( freePage != NO_PAGE )
{
if ( isFirst )
{
isFirst = false;
}
else
{
sb.append( " -> " );
}
sb.append( "0x" ).append( Long.toHexString( freePage ) );
try
{
PageIO[] freePageIO = readPageIOs( freePage, 8 );
freePage = freePageIO[0].getNextPage();
}
catch ( EndOfFileExceededException e )
{
// TODO Auto-generated catch block
e.printStackTrace();
}
catch ( IOException e )
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
LOG_PAGES.debug( "Update RM Header : \n{}", sb.toString() );
}
try
{
Integer nbTxnStarted = context.get();
if ( ( nbTxnStarted == null ) || ( nbTxnStarted <= 1 ) )
{
//System.out.println( "Writing page at 0000" );
writeCounter.put( 0L, writeCounter.containsKey( 0L ) ? writeCounter.get( 0L ) + 1 : 1 );
fileChannel.write( RECORD_MANAGER_HEADER_BUFFER, 0 );
}
}
catch ( IOException ioe )
{
throw new FileException( ioe.getMessage() );
}
RECORD_MANAGER_HEADER_BUFFER.clear();
// Reset the old versions
previousBtreeOfBtreesOffset = -1L;
previousCopiedPagesBtreeOffset = -1L;
nbUpdateRMHeader.incrementAndGet();
}
/**
* Update the RecordManager header, injecting the following data :
*
* <pre>
* +---------------------+
* | PageSize | 4 bytes : The size of a physical page (default to 4096)
* +---------------------+
* | NbTree | 4 bytes : The number of managed B-trees (at least 1)
* +---------------------+
* | FirstFree | 8 bytes : The offset of the first free page
* +---------------------+
* | current BoB offset | 8 bytes : The offset of the current B-tree of B-trees
* +---------------------+
* | previous BoB offset | 8 bytes : The offset of the previous B-tree of B-trees
* +---------------------+
* | current CP offset | 8 bytes : The offset of the current CopiedPages B-tree
* +---------------------+
* | previous CP offset | 8 bytes : The offset of the previous CopiedPages B-tree
* +---------------------+
* </pre>
*/
public void updateRecordManagerHeader( long newBtreeOfBtreesOffset, long newCopiedPageBtreeOffset )
{
if ( newBtreeOfBtreesOffset != -1L )
{
previousBtreeOfBtreesOffset = currentBtreeOfBtreesOffset;
currentBtreeOfBtreesOffset = newBtreeOfBtreesOffset;
}
if ( newCopiedPageBtreeOffset != -1L )
{
previousCopiedPagesBtreeOffset = currentCopiedPagesBtreeOffset;
currentCopiedPagesBtreeOffset = newCopiedPageBtreeOffset;
}
}
/**
* Inject an int into a byte[] at a given position.
*/
private int writeData( byte[] buffer, int position, int value )
{
RECORD_MANAGER_HEADER_BYTES[position] = ( byte ) ( value >>> 24 );
RECORD_MANAGER_HEADER_BYTES[position + 1] = ( byte ) ( value >>> 16 );
RECORD_MANAGER_HEADER_BYTES[position + 2] = ( byte ) ( value >>> 8 );
RECORD_MANAGER_HEADER_BYTES[position + 3] = ( byte ) ( value );
return position + 4;
}
/**
* Inject a long into a byte[] at a given position.
*/
private int writeData( byte[] buffer, int position, long value )
{
RECORD_MANAGER_HEADER_BYTES[position] = ( byte ) ( value >>> 56 );
RECORD_MANAGER_HEADER_BYTES[position + 1] = ( byte ) ( value >>> 48 );
RECORD_MANAGER_HEADER_BYTES[position + 2] = ( byte ) ( value >>> 40 );
RECORD_MANAGER_HEADER_BYTES[position + 3] = ( byte ) ( value >>> 32 );
RECORD_MANAGER_HEADER_BYTES[position + 4] = ( byte ) ( value >>> 24 );
RECORD_MANAGER_HEADER_BYTES[position + 5] = ( byte ) ( value >>> 16 );
RECORD_MANAGER_HEADER_BYTES[position + 6] = ( byte ) ( value >>> 8 );
RECORD_MANAGER_HEADER_BYTES[position + 7] = ( byte ) ( value );
return position + 8;
}
/**
* Add a new <btree, revision> tuple into the B-tree of B-trees.
*
* @param name The B-tree name
* @param revision The B-tree revision
* @param btreeHeaderOffset The B-tree offset
* @throws IOException If the update failed
*/
/* no qualifier */<K, V> void addInBtreeOfBtrees( String name, long revision, long btreeHeaderOffset )
throws IOException
{
checkOffset( btreeHeaderOffset );
NameRevision nameRevision = new NameRevision( name, revision );
btreeOfBtrees.insert( nameRevision, btreeHeaderOffset );
// Update the B-tree of B-trees offset
currentBtreeOfBtreesOffset = getNewBTreeHeader( BTREE_OF_BTREES_NAME ).getBTreeHeaderOffset();
}
/**
* Add a new <btree, revision> tuple into the CopiedPages B-tree.
*
* @param name The B-tree name
* @param revision The B-tree revision
* @param btreeHeaderOffset The B-tree offset
* @throws IOException If the update failed
*/
/* no qualifier */<K, V> void addInCopiedPagesBtree( String name, long revision, List<Page<K, V>> pages )
throws IOException
{
RevisionName revisionName = new RevisionName( revision, name );
long[] pageOffsets = new long[pages.size()];
int pos = 0;
for ( Page<K, V> page : pages )
{
pageOffsets[pos++] = ( ( AbstractPage<K, V> ) page ).getOffset();
}
copiedPageBtree.insert( revisionName, pageOffsets );
// Update the CopiedPageBtree offset
currentCopiedPagesBtreeOffset = ( ( AbstractBTree<RevisionName, long[]> ) copiedPageBtree ).getBtreeHeader().getBTreeHeaderOffset();
}
/**
* Internal method used to update the B-tree of B-trees offset
* @param btreeOfBtreesOffset The new offset
*/
/* no qualifier */void setBtreeOfBtreesOffset( long btreeOfBtreesOffset )
{
checkOffset( btreeOfBtreesOffset );
this.currentBtreeOfBtreesOffset = btreeOfBtreesOffset;
}
/**
* Write the B-tree header on disk. We will write the following informations :
* <pre>
* +------------+
* | revision | The B-tree revision
* +------------+
* | nbElems | The B-tree number of elements
* +------------+
* | rootPage | The root page offset
* +------------+
* | BtreeInfo | The B-tree info offset
* +------------+
* </pre>
* @param btree The B-tree which header has to be written
* @param btreeInfoOffset The offset of the B-tree informations
* @return The B-tree header offset
* @throws IOException If we weren't able to write the B-tree header
*/
/* no qualifier */<K, V> long writeBtreeHeader( BTree<K, V> btree, BTreeHeader<K, V> btreeHeader )
throws IOException
{
int bufferSize =
LONG_SIZE + // The revision
LONG_SIZE + // the number of element
LONG_SIZE + // The root page offset
LONG_SIZE; // The B-tree info page offset
// Get the pageIOs we need to store the data. We may need more than one.
PageIO[] btreeHeaderPageIos = getFreePageIOs( bufferSize );
// Store the B-tree header Offset into the B-tree
long btreeHeaderOffset = btreeHeaderPageIos[0].getOffset();
// Now store the B-tree data in the pages :
// - the B-tree revision
// - the B-tree number of elements
// - the B-tree root page offset
// - the B-tree info page offset
// Starts at 0
long position = 0L;
// The B-tree current revision
position = store( position, btreeHeader.getRevision(), btreeHeaderPageIos );
// The nb elems in the tree
position = store( position, btreeHeader.getNbElems(), btreeHeaderPageIos );
// Now, we can inject the B-tree rootPage offset into the B-tree header
position = store( position, btreeHeader.getRootPageOffset(), btreeHeaderPageIos );
// The B-tree info page offset
position = store( position, ( ( PersistedBTree<K, V> ) btree ).getBtreeInfoOffset(), btreeHeaderPageIos );
// And flush the pages to disk now
LOG.debug( "Flushing the newly managed '{}' btree header", btree.getName() );
if ( LOG_PAGES.isDebugEnabled() )
{
LOG_PAGES.debug( "Writing BTreeHeader revision {} for {}", btreeHeader.getRevision(), btree.getName() );
StringBuilder sb = new StringBuilder();
sb.append( "Offset : " ).append( Long.toHexString( btreeHeaderOffset ) ).append( "\n" );
sb.append( " Revision : " ).append( btreeHeader.getRevision() ).append( "\n" );
sb.append( " NbElems : " ).append( btreeHeader.getNbElems() ).append( "\n" );
sb.append( " RootPage : 0x" ).append( Long.toHexString( btreeHeader.getRootPageOffset() ) )
.append( "\n" );
sb.append( " Info : 0x" )
.append( Long.toHexString( ( ( PersistedBTree<K, V> ) btree ).getBtreeInfoOffset() ) ).append( "\n" );
LOG_PAGES.debug( "Btree Header[{}]\n{}", btreeHeader.getRevision(), sb.toString() );
}
flushPages( btreeHeaderPageIos );
btreeHeader.setBTreeHeaderOffset( btreeHeaderOffset );
return btreeHeaderOffset;
}
/**
* Write the B-tree informations on disk. We will write the following informations :
* <pre>
* +------------+
* | pageSize | The B-tree page size (ie, the number of elements per page max)
* +------------+
* | nameSize | The B-tree name size
* +------------+
* | name | The B-tree name
* +------------+
* | keySerSize | The keySerializer FQCN size
* +------------+
* | keySerFQCN | The keySerializer FQCN
* +------------+
* | valSerSize | The Value serializer FQCN size
* +------------+
* | valSerKQCN | The valueSerializer FQCN
* +------------+
* | dups | The flags that tell if the dups are allowed
* +------------+
* </pre>
* @param btree The B-tree which header has to be written
* @return The B-tree header offset
* @throws IOException If we weren't able to write the B-tree header
*/
private <K, V> long writeBtreeInfo( BTree<K, V> btree ) throws IOException
{
// We will add the newly managed B-tree at the end of the header.
byte[] btreeNameBytes = Strings.getBytesUtf8( btree.getName() );
byte[] keySerializerBytes = Strings.getBytesUtf8( btree.getKeySerializerFQCN() );
byte[] valueSerializerBytes = Strings.getBytesUtf8( btree.getValueSerializerFQCN() );
int bufferSize =
INT_SIZE + // The page size
INT_SIZE + // The name size
btreeNameBytes.length + // The name
INT_SIZE + // The keySerializerBytes size
keySerializerBytes.length + // The keySerializerBytes
INT_SIZE + // The valueSerializerBytes size
valueSerializerBytes.length + // The valueSerializerBytes
INT_SIZE; // The allowDuplicates flag
// Get the pageIOs we need to store the data. We may need more than one.
PageIO[] btreeHeaderPageIos = getFreePageIOs( bufferSize );
// Keep the B-tree header Offset into the B-tree
long btreeInfoOffset = btreeHeaderPageIos[0].getOffset();
// Now store the B-tree information data in the pages :
// - the B-tree page size
// - the B-tree name
// - the keySerializer FQCN
// - the valueSerializer FQCN
// - the flags that tell if the dups are allowed
// Starts at 0
long position = 0L;
// The B-tree page size
position = store( position, btree.getPageSize(), btreeHeaderPageIos );
// The tree name
position = store( position, btreeNameBytes, btreeHeaderPageIos );
// The keySerializer FQCN
position = store( position, keySerializerBytes, btreeHeaderPageIos );
// The valueSerialier FQCN
position = store( position, valueSerializerBytes, btreeHeaderPageIos );
// The allowDuplicates flag
position = store( position, ( btree.isAllowDuplicates() ? 1 : 0 ), btreeHeaderPageIos );
// And flush the pages to disk now
LOG.debug( "Flushing the newly managed '{}' btree header", btree.getName() );
flushPages( btreeHeaderPageIos );
return btreeInfoOffset;
}
/**
* Update the B-tree header after a B-tree modification. This will make the latest modification
* visible.<br/>
* We update the following fields :
* <ul>
* <li>the revision</li>
* <li>the number of elements</li>
* <li>the B-tree root page offset</li>
* </ul>
* <br/>
* As a result, a new version of the BtreHeader will be created, which will replace the previous
* B-tree header
* @param btree TheB-tree to update
* @param btreeHeaderOffset The offset of the modified btree header
* @return The offset of the new B-tree Header
* @throws IOException If we weren't able to write the file on disk
* @throws EndOfFileExceededException If we tried to write after the end of the file
*/
/* no qualifier */<K, V> long updateBtreeHeader( BTree<K, V> btree, long btreeHeaderOffset )
throws EndOfFileExceededException, IOException
{
return updateBtreeHeader( btree, btreeHeaderOffset, false );
}
/**
* Update the B-tree header after a B-tree modification. This will make the latest modification
* visible.<br/>
* We update the following fields :
* <ul>
* <li>the revision</li>
* <li>the number of elements</li>
* <li>the reference to the current B-tree revisions</li>
* <li>the reference to the old B-tree revisions</li>
* </ul>
* <br/>
* As a result, we new version of the BtreHeader will be created
* @param btree The B-tree to update
* @param btreeHeaderOffset The offset of the modified btree header
* @return The offset of the new B-tree Header if it has changed (ie, when the onPlace flag is set to true)
* @throws IOException
* @throws EndOfFileExceededException
*/
/* no qualifier */<K, V> void updateBtreeHeaderOnPlace( BTree<K, V> btree, long btreeHeaderOffset )
throws EndOfFileExceededException,
IOException
{
updateBtreeHeader( btree, btreeHeaderOffset, true );
}
/**
* Update the B-tree header after a B-tree modification. This will make the latest modification
* visible.<br/>
* We update the following fields :
* <ul>
* <li>the revision</li>
* <li>the number of elements</li>
* <li>the reference to the current B-tree revisions</li>
* <li>the reference to the old B-tree revisions</li>
* </ul>
* <br/>
* As a result, a new version of the BtreHeader will be created, which may replace the previous
* B-tree header (if the onPlace flag is set to true) or a new set of pageIos will contain the new
* version.
*
* @param btree The B-tree to update
* @param rootPageOffset The offset of the modified rootPage
* @param onPlace Tells if we modify the B-tree on place, or if we create a copy
* @return The offset of the new B-tree Header if it has changed (ie, when the onPlace flag is set to true)
* @throws EndOfFileExceededException If we tried to write after the end of the file
* @throws IOException If tehre were some error while writing the data on disk
*/
private <K, V> long updateBtreeHeader( BTree<K, V> btree, long btreeHeaderOffset, boolean onPlace )
throws EndOfFileExceededException, IOException
{
// Read the pageIOs associated with this B-tree
PageIO[] pageIos;
long newBtreeHeaderOffset = NO_PAGE;
long offset = ( ( PersistedBTree<K, V> ) btree ).getBtreeOffset();
if ( onPlace )
{
// We just have to update the existing BTreeHeader
long headerSize = LONG_SIZE + LONG_SIZE + LONG_SIZE;
pageIos = readPageIOs( offset, headerSize );
// Now, update the revision
long position = 0;
position = store( position, btree.getRevision(), pageIos );
position = store( position, btree.getNbElems(), pageIos );
position = store( position, btreeHeaderOffset, pageIos );
// Write the pages on disk
if ( LOG.isDebugEnabled() )
{
LOG.debug( "-----> Flushing the '{}' B-treeHeader", btree.getName() );
LOG.debug( " revision : " + btree.getRevision() + ", NbElems : " + btree.getNbElems()
+ ", btreeHeader offset : 0x"
+ Long.toHexString( btreeHeaderOffset ) );
}
// Get new place on disk to store the modified BTreeHeader if it's not onPlace
// Rewrite the pages at the same place
LOG.debug( "Rewriting the B-treeHeader on place for B-tree " + btree.getName() );
flushPages( pageIos );
}
else
{
// We have to read and copy the existing BTreeHeader and to create a new one
pageIos = readPageIOs( offset, Long.MAX_VALUE );
// Now, copy every read page
PageIO[] newPageIOs = new PageIO[pageIos.length];
int pos = 0;
for ( PageIO pageIo : pageIos )
{
// Fetch a free page
newPageIOs[pos] = fetchNewPage();
// keep a track of the allocated and copied pages so that we can
// free them when we do a commit or rollback, if the btree is an management one
if ( ( btree.getType() == BTreeTypeEnum.BTREE_OF_BTREES )
|| ( btree.getType() == BTreeTypeEnum.COPIED_PAGES_BTREE ) )
{
freedPages.add( pageIo );
allocatedPages.add( newPageIOs[pos] );
}
pageIo.copy( newPageIOs[pos] );
if ( pos > 0 )
{
newPageIOs[pos - 1].setNextPage( newPageIOs[pos].getOffset() );
}
pos++;
}
// store the new btree header offset
// and update the revision
long position = 0;
position = store( position, btree.getRevision(), newPageIOs );
position = store( position, btree.getNbElems(), newPageIOs );
position = store( position, btreeHeaderOffset, newPageIOs );
// Get new place on disk to store the modified BTreeHeader if it's not onPlace
// Flush the new B-treeHeader on disk
LOG.debug( "Rewriting the B-treeHeader on place for B-tree " + btree.getName() );
flushPages( newPageIOs );
newBtreeHeaderOffset = newPageIOs[0].getOffset();
}
nbUpdateBtreeHeader.incrementAndGet();
if ( LOG_CHECK.isDebugEnabled() )
{
MavibotInspector.check( this );
}
return newBtreeHeaderOffset;
}
/**
* Write the pages on disk, either at the end of the file, or at
* the position they were taken from.
*
* @param pageIos The list of pages to write
* @throws IOException If the write failed
*/
private void flushPages( PageIO... pageIos ) throws IOException
{
if ( LOG.isDebugEnabled() )
{
for ( PageIO pageIo : pageIos )
{
dump( pageIo );
}
}
for ( PageIO pageIo : pageIos )
{
pageIo.getData().rewind();
long pos = pageIo.getOffset();
if ( fileChannel.size() < ( pageIo.getOffset() + pageSize ) )
{
LOG.debug( "Adding a page at the end of the file" );
// This is a page we have to add to the file
pos = fileChannel.size();
fileChannel.write( pageIo.getData(), pos );
//fileChannel.force( false );
}
else
{
LOG.debug( "Writing a page at position {}", pageIo.getOffset() );
fileChannel.write( pageIo.getData(), pageIo.getOffset() );
//fileChannel.force( false );
}
//System.out.println( "Writing page at " + Long.toHexString( pos ) );
writeCounter.put( pos, writeCounter.containsKey( pos ) ? writeCounter.get( pos ) + 1 : 1 );
nbUpdatePageIOs.incrementAndGet();
pageIo.getData().rewind();
}
}
/**
* Compute the page in which we will store data given an offset, when
* we have a list of pages.
*
* @param offset The position in the data
* @return The page number in which the offset will start
*/
private int computePageNb( long offset )
{
long pageNb = 0;
offset -= pageSize - LINK_SIZE - PAGE_SIZE;
if ( offset < 0 )
{
return ( int ) pageNb;
}
pageNb = 1 + offset / ( pageSize - LINK_SIZE );
return ( int ) pageNb;
}
/**
* Stores a byte[] into one ore more pageIO (depending if the long is stored
* across a boundary or not)
*
* @param position The position in a virtual byte[] if all the pages were contiguous
* @param bytes The byte[] to serialize
* @param pageIos The pageIOs we have to store the data in
* @return The new offset
*/
private long store( long position, byte[] bytes, PageIO... pageIos )
{
if ( bytes != null )
{
// Write the bytes length
position = store( position, bytes.length, pageIos );
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Get back the buffer in this page
ByteBuffer pageData = pageIos[pageNb].getData();
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
// Compute the remaining size in the page
int remaining = pageData.capacity() - pagePos;
int nbStored = bytes.length;
// And now, write the bytes until we have none
while ( nbStored > 0 )
{
if ( remaining > nbStored )
{
pageData.mark();
pageData.position( pagePos );
pageData.put( bytes, bytes.length - nbStored, nbStored );
pageData.reset();
nbStored = 0;
}
else
{
pageData.mark();
pageData.position( pagePos );
pageData.put( bytes, bytes.length - nbStored, remaining );
pageData.reset();
pageNb++;
pageData = pageIos[pageNb].getData();
pagePos = LINK_SIZE;
nbStored -= remaining;
remaining = pageData.capacity() - pagePos;
}
}
// We are done
position += bytes.length;
}
else
{
// No bytes : write 0 and return
position = store( position, 0, pageIos );
}
return position;
}
/**
* Stores a byte[] into one ore more pageIO (depending if the long is stored
* across a boundary or not). We don't add the byte[] size, it's already present
* in the received byte[].
*
* @param position The position in a virtual byte[] if all the pages were contiguous
* @param bytes The byte[] to serialize
* @param pageIos The pageIOs we have to store the data in
* @return The new offset
*/
private long storeRaw( long position, byte[] bytes, PageIO... pageIos )
{
if ( bytes != null )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Get back the buffer in this page
ByteBuffer pageData = pageIos[pageNb].getData();
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
// Compute the remaining size in the page
int remaining = pageData.capacity() - pagePos;
int nbStored = bytes.length;
// And now, write the bytes until we have none
while ( nbStored > 0 )
{
if ( remaining > nbStored )
{
pageData.mark();
pageData.position( pagePos );
pageData.put( bytes, bytes.length - nbStored, nbStored );
pageData.reset();
nbStored = 0;
}
else
{
pageData.mark();
pageData.position( pagePos );
pageData.put( bytes, bytes.length - nbStored, remaining );
pageData.reset();
pageNb++;
if ( pageNb == pageIos.length )
{
// We can stop here : we have reach the end of the page
break;
}
pageData = pageIos[pageNb].getData();
pagePos = LINK_SIZE;
nbStored -= remaining;
remaining = pageData.capacity() - pagePos;
}
}
// We are done
position += bytes.length;
}
else
{
// No bytes : write 0 and return
position = store( position, 0, pageIos );
}
return position;
}
/**
* Stores an Integer into one ore more pageIO (depending if the int is stored
* across a boundary or not)
*
* @param position The position in a virtual byte[] if all the pages were contiguous
* @param value The int to serialize
* @param pageIos The pageIOs we have to store the data in
* @return The new offset
*/
private long store( long position, int value, PageIO... pageIos )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
// Get back the buffer in this page
ByteBuffer pageData = pageIos[pageNb].getData();
// Compute the remaining size in the page
int remaining = pageData.capacity() - pagePos;
if ( remaining < INT_SIZE )
{
// We have to copy the serialized length on two pages
switch ( remaining )
{
case 3:
pageData.put( pagePos + 2, ( byte ) ( value >>> 8 ) );
// Fallthrough !!!
case 2:
pageData.put( pagePos + 1, ( byte ) ( value >>> 16 ) );
// Fallthrough !!!
case 1:
pageData.put( pagePos, ( byte ) ( value >>> 24 ) );
break;
}
// Now deal with the next page
pageData = pageIos[pageNb + 1].getData();
pagePos = LINK_SIZE;
switch ( remaining )
{
case 1:
pageData.put( pagePos, ( byte ) ( value >>> 16 ) );
// fallthrough !!!
case 2:
pageData.put( pagePos + 2 - remaining, ( byte ) ( value >>> 8 ) );
// fallthrough !!!
case 3:
pageData.put( pagePos + 3 - remaining, ( byte ) ( value ) );
break;
}
}
else
{
// Store the value in the page at the selected position
pageData.putInt( pagePos, value );
}
// Increment the position to reflect the addition of an Int (4 bytes)
position += INT_SIZE;
return position;
}
/**
* Stores a Long into one ore more pageIO (depending if the long is stored
* across a boundary or not)
*
* @param position The position in a virtual byte[] if all the pages were contiguous
* @param value The long to serialize
* @param pageIos The pageIOs we have to store the data in
* @return The new offset
*/
private long store( long position, long value, PageIO... pageIos )
{
// Compute the page in which we will store the data given the
// current position
int pageNb = computePageNb( position );
// Compute the position in the current page
int pagePos = ( int ) ( position + ( pageNb + 1 ) * LONG_SIZE + INT_SIZE ) - pageNb * pageSize;
// Get back the buffer in this page
ByteBuffer pageData = pageIos[pageNb].getData();
// Compute the remaining size in the page
int remaining = pageData.capacity() - pagePos;
if ( remaining < LONG_SIZE )
{
// We have to copy the serialized length on two pages
switch ( remaining )
{
case 7:
pageData.put( pagePos + 6, ( byte ) ( value >>> 8 ) );
// Fallthrough !!!
case 6:
pageData.put( pagePos + 5, ( byte ) ( value >>> 16 ) );
// Fallthrough !!!
case 5:
pageData.put( pagePos + 4, ( byte ) ( value >>> 24 ) );
// Fallthrough !!!
case 4:
pageData.put( pagePos + 3, ( byte ) ( value >>> 32 ) );
// Fallthrough !!!
case 3:
pageData.put( pagePos + 2, ( byte ) ( value >>> 40 ) );
// Fallthrough !!!
case 2:
pageData.put( pagePos + 1, ( byte ) ( value >>> 48 ) );
// Fallthrough !!!
case 1:
pageData.put( pagePos, ( byte ) ( value >>> 56 ) );
break;
}
// Now deal with the next page
pageData = pageIos[pageNb + 1].getData();
pagePos = LINK_SIZE;
switch ( remaining )
{
case 1:
pageData.put( pagePos, ( byte ) ( value >>> 48 ) );
// fallthrough !!!
case 2:
pageData.put( pagePos + 2 - remaining, ( byte ) ( value >>> 40 ) );
// fallthrough !!!
case 3:
pageData.put( pagePos + 3 - remaining, ( byte ) ( value >>> 32 ) );
// fallthrough !!!
case 4:
pageData.put( pagePos + 4 - remaining, ( byte ) ( value >>> 24 ) );
// fallthrough !!!
case 5:
pageData.put( pagePos + 5 - remaining, ( byte ) ( value >>> 16 ) );
// fallthrough !!!
case 6:
pageData.put( pagePos + 6 - remaining, ( byte ) ( value >>> 8 ) );
// fallthrough !!!
case 7:
pageData.put( pagePos + 7 - remaining, ( byte ) ( value ) );
break;
}
}
else
{
// Store the value in the page at the selected position
pageData.putLong( pagePos, value );
}
// Increment the position to reflect the addition of an Long (8 bytes)
position += LONG_SIZE;
return position;
}
/**
* Write the page in a serialized form.
*
* @param btree The persistedBtree we will create a new PageHolder for
* @param newPage The page to write on disk
* @param newRevision The page's revision
* @return A PageHolder containing the copied page
* @throws IOException If the page can't be written on disk
*/
/* No qualifier*/<K, V> PageHolder<K, V> writePage( BTree<K, V> btree, Page<K, V> newPage,
long newRevision ) throws IOException
{
// We first need to save the new page on disk
PageIO[] pageIos = serializePage( btree, newRevision, newPage );
if ( LOG_PAGES.isDebugEnabled() )
{
LOG_PAGES.debug( "Write data for '{}' btree", btree.getName() );
logPageIos( pageIos );
}
// Write the page on disk
flushPages( pageIos );
// Build the resulting reference
long offset = pageIos[0].getOffset();
long lastOffset = pageIos[pageIos.length - 1].getOffset();
PersistedPageHolder<K, V> pageHolder = new PersistedPageHolder<K, V>( btree, newPage, offset,
lastOffset );
return pageHolder;
}
/* No qualifier */static void logPageIos( PageIO[] pageIos )
{
int pageNb = 0;
for ( PageIO pageIo : pageIos )
{
StringBuilder sb = new StringBuilder();
sb.append( "PageIO[" ).append( pageNb ).append( "]:0x" );
sb.append( Long.toHexString( pageIo.getOffset() ) ).append( "/" );
sb.append( pageIo.getSize() );
pageNb++;
ByteBuffer data = pageIo.getData();
int position = data.position();
int dataLength = ( int ) pageIo.getSize() + 12;
if ( dataLength > data.limit() )
{
dataLength = data.limit();
}
byte[] bytes = new byte[dataLength];
data.get( bytes );
data.position( position );
int pos = 0;
for ( byte b : bytes )
{
int mod = pos % 16;
switch ( mod )
{
case 0:
sb.append( "\n " );
// No break
case 4:
case 8:
case 12:
sb.append( " " );
case 1:
case 2:
case 3:
case 5:
case 6:
case 7:
case 9:
case 10:
case 11:
case 13:
case 14:
case 15:
sb.append( Strings.dumpByte( b ) ).append( " " );
}
pos++;
}
LOG_PAGES.debug( sb.toString() );
}
}
/**
* Compute the number of pages needed to store some specific size of data.
*
* @param dataSize The size of the data we want to store in pages
* @return The number of pages needed
*/
private int computeNbPages( int dataSize )
{
if ( dataSize <= 0 )
{
return 0;
}
// Compute the number of pages needed.
// Considering that each page can contain PageSize bytes,
// but that the first 8 bytes are used for links and we
// use 4 bytes to store the data size, the number of needed
// pages is :
// NbPages = ( (dataSize - (PageSize - 8 - 4 )) / (PageSize - 8) ) + 1
// NbPages += ( if (dataSize - (PageSize - 8 - 4 )) % (PageSize - 8) > 0 : 1 : 0 )
int availableSize = ( pageSize - LONG_SIZE );
int nbNeededPages = 1;
// Compute the number of pages that will be full but the first page
if ( dataSize > availableSize - INT_SIZE )
{
int remainingSize = dataSize - ( availableSize - INT_SIZE );
nbNeededPages += remainingSize / availableSize;
int remain = remainingSize % availableSize;
if ( remain > 0 )
{
nbNeededPages++;
}
}
return nbNeededPages;
}
/**
* Get as many pages as needed to store the data of the given size. The returned
* PageIOs are all linked together.
*
* @param dataSize The data size
* @return An array of pages, enough to store the full data
*/
private PageIO[] getFreePageIOs( int dataSize ) throws IOException
{
if ( dataSize == 0 )
{
return new PageIO[]
{};
}
int nbNeededPages = computeNbPages( dataSize );
PageIO[] pageIOs = new PageIO[nbNeededPages];
// The first page : set the size
pageIOs[0] = fetchNewPage();
pageIOs[0].setSize( dataSize );
for ( int i = 1; i < nbNeededPages; i++ )
{
pageIOs[i] = fetchNewPage();
// Create the link
pageIOs[i - 1].setNextPage( pageIOs[i].getOffset() );
}
return pageIOs;
}
/**
* Return a new Page. We take one of the existing free pages, or we create
* a new page at the end of the file.
*
* @return The fetched PageIO
*/
private PageIO fetchNewPage() throws IOException
{
//System.out.println( "Fetching new page" );
if ( firstFreePage == NO_PAGE )
{
nbCreatedPages.incrementAndGet();
// We don't have any free page. Reclaim some new page at the end
// of the file
PageIO newPage = new PageIO( endOfFileOffset );
endOfFileOffset += pageSize;
ByteBuffer data = ByteBuffer.allocateDirect( pageSize );
newPage.setData( data );
newPage.setNextPage( NO_PAGE );
newPage.setSize( 0 );
LOG.debug( "Requiring a new page at offset {}", newPage.getOffset() );
return newPage;
}
else
{
nbReusedPages.incrementAndGet();
freePageLock.lock();
// We have some existing free page. Fetch it from disk
PageIO pageIo = fetchPage( firstFreePage );
// Update the firstFreePage pointer
firstFreePage = pageIo.getNextPage();
freePageLock.unlock();
// overwrite the data of old page
ByteBuffer data = ByteBuffer.allocateDirect( pageSize );
pageIo.setData( data );
pageIo.setNextPage( NO_PAGE );
pageIo.setSize( 0 );
LOG.debug( "Reused page at offset {}", pageIo.getOffset() );
return pageIo;
}
}
/**
* fetch a page from disk, knowing its position in the file.
*
* @param offset The position in the file
* @return The found page
*/
/* no qualifier */PageIO fetchPage( long offset ) throws IOException, EndOfFileExceededException
{
checkOffset( offset );
if ( fileChannel.size() < offset + pageSize )
{
// Error : we are past the end of the file
throw new EndOfFileExceededException( "We are fetching a page on " + offset +
" when the file's size is " + fileChannel.size() );
}
else
{
// Read the page
fileChannel.position( offset );
ByteBuffer data = ByteBuffer.allocate( pageSize );
fileChannel.read( data );
data.rewind();
PageIO readPage = new PageIO( offset );
readPage.setData( data );
return readPage;
}
}
/**
* @return the pageSize
*/
public int getPageSize()
{
return pageSize;
}
/**
* Set the page size, ie the number of bytes a page can store.
*
* @param pageSize The number of bytes for a page
*/
/* no qualifier */void setPageSize( int pageSize )
{
if ( this.pageSize >= 13 )
{
this.pageSize = pageSize;
}
else
{
this.pageSize = DEFAULT_PAGE_SIZE;
}
}
/**
* Close the RecordManager and flush everything on disk
*/
public void close() throws IOException
{
beginTransaction();
// Close all the managed B-trees
for ( BTree<Object, Object> tree : managedBtrees.values() )
{
tree.close();
}
// Close the management B-trees
copiedPageBtree.close();
btreeOfBtrees.close();
managedBtrees.clear();
// Write the data
fileChannel.force( true );
// And close the channel
fileChannel.close();
commit();
}
/** Hex chars */
private static final byte[] HEX_CHAR = new byte[]
{ '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F' };
public static String dump( byte octet )
{
return new String( new byte[]
{ HEX_CHAR[( octet & 0x00F0 ) >> 4], HEX_CHAR[octet & 0x000F] } );
}
/**
* Dump a pageIO
*/
private void dump( PageIO pageIo )
{
ByteBuffer buffer = pageIo.getData();
buffer.mark();
byte[] longBuffer = new byte[LONG_SIZE];
byte[] intBuffer = new byte[INT_SIZE];
// get the next page offset
buffer.get( longBuffer );
long nextOffset = LongSerializer.deserialize( longBuffer );
// Get the data size
buffer.get( intBuffer );
int size = IntSerializer.deserialize( intBuffer );
buffer.reset();
System.out.println( "PageIO[" + Long.toHexString( pageIo.getOffset() ) + "], size = " + size + ", NEXT PageIO:"
+ Long.toHexString( nextOffset ) );
System.out.println( " 0 1 2 3 4 5 6 7 8 9 A B C D E F " );
System.out.println( "+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+" );
for ( int i = 0; i < buffer.limit(); i += 16 )
{
System.out.print( "|" );
for ( int j = 0; j < 16; j++ )
{
System.out.print( dump( buffer.get() ) );
if ( j == 15 )
{
System.out.println( "|" );
}
else
{
System.out.print( " " );
}
}
}
System.out.println( "+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+--+" );
buffer.reset();
}
/**
* Dump the RecordManager file
* @throws IOException
*/
public void dump()
{
System.out.println( "/---------------------------- Dump ----------------------------\\" );
try
{
RandomAccessFile randomFile = new RandomAccessFile( file, "r" );
FileChannel fileChannel = randomFile.getChannel();
ByteBuffer recordManagerHeader = ByteBuffer.allocate( RECORD_MANAGER_HEADER_SIZE );
// load the RecordManager header
fileChannel.read( recordManagerHeader );
recordManagerHeader.rewind();
// The page size
long fileSize = fileChannel.size();
int pageSize = recordManagerHeader.getInt();
long nbPages = fileSize / pageSize;
// The number of managed B-trees
int nbBtree = recordManagerHeader.getInt();
// The first free page
long firstFreePage = recordManagerHeader.getLong();
// The current B-tree of B-trees
long currentBtreeOfBtreesPage = recordManagerHeader.getLong();
// The previous B-tree of B-trees
long previousBtreeOfBtreesPage = recordManagerHeader.getLong();
// The current CopiedPages B-tree
long currentCopiedPagesBtreePage = recordManagerHeader.getLong();
// The previous CopiedPages B-tree
long previousCopiedPagesBtreePage = recordManagerHeader.getLong();
System.out.println( " RecordManager" );
System.out.println( " -------------" );
System.out.println( " Size = 0x" + Long.toHexString( fileSize ) );
System.out.println( " NbPages = " + nbPages );
System.out.println( " Header " );
System.out.println( " page size : " + pageSize );
System.out.println( " nbTree : " + nbBtree );
System.out.println( " firstFreePage : 0x" + Long.toHexString( firstFreePage ) );
System.out.println( " current BOB : 0x" + Long.toHexString( currentBtreeOfBtreesPage ) );
System.out.println( " previous BOB : 0x" + Long.toHexString( previousBtreeOfBtreesPage ) );
System.out.println( " current CopiedPages : 0x" + Long.toHexString( currentCopiedPagesBtreePage ) );
System.out.println( " previous CopiedPages : 0x" + Long.toHexString( previousCopiedPagesBtreePage ) );
// Dump the Free pages list
dumpFreePages( firstFreePage );
// Dump the B-tree of B-trees
dumpBtreeHeader( currentBtreeOfBtreesPage );
// Dump the previous B-tree of B-trees if any
if ( previousBtreeOfBtreesPage != NO_PAGE )
{
dumpBtreeHeader( previousBtreeOfBtreesPage );
}
// Dump the CopiedPages B-tree
dumpBtreeHeader( currentCopiedPagesBtreePage );
// Dump the previous B-tree of B-trees if any
if ( previousCopiedPagesBtreePage != NO_PAGE )
{
dumpBtreeHeader( previousCopiedPagesBtreePage );
}
// Dump all the user's B-tree
randomFile.close();
System.out.println( "\\---------------------------- Dump ----------------------------/" );
}
catch ( IOException ioe )
{
System.out.println( "Exception while dumping the file : " + ioe.getMessage() );
}
}
/**
* Dump the free pages
*/
private void dumpFreePages( long freePageOffset ) throws EndOfFileExceededException, IOException
{
System.out.println( "\n FreePages : " );
int pageNb = 1;
while ( freePageOffset != NO_PAGE )
{
PageIO pageIo = fetchPage( freePageOffset );
System.out.println( " freePage[" + pageNb + "] : 0x" + Long.toHexString( pageIo.getOffset() ) );
freePageOffset = pageIo.getNextPage();
pageNb++;
}
}
/**
* Dump a B-tree Header
*/
private long dumpBtreeHeader( long btreeOffset ) throws EndOfFileExceededException, IOException
{
// First read the B-tree header
PageIO[] pageIos = readPageIOs( btreeOffset, Long.MAX_VALUE );
long dataPos = 0L;
// The B-tree current revision
long revision = readLong( pageIos, dataPos );
dataPos += LONG_SIZE;
// The nb elems in the tree
long nbElems = readLong( pageIos, dataPos );
dataPos += LONG_SIZE;
// The B-tree rootPage offset
long rootPageOffset = readLong( pageIos, dataPos );
dataPos += LONG_SIZE;
// The B-tree page size
int btreePageSize = readInt( pageIos, dataPos );
dataPos += INT_SIZE;
// The tree name
ByteBuffer btreeNameBytes = readBytes( pageIos, dataPos );
dataPos += INT_SIZE + btreeNameBytes.limit();
String btreeName = Strings.utf8ToString( btreeNameBytes );
// The keySerializer FQCN
ByteBuffer keySerializerBytes = readBytes( pageIos, dataPos );
dataPos += INT_SIZE + keySerializerBytes.limit();
String keySerializerFqcn = "";
if ( keySerializerBytes != null )
{
keySerializerFqcn = Strings.utf8ToString( keySerializerBytes );
}
// The valueSerialier FQCN
ByteBuffer valueSerializerBytes = readBytes( pageIos, dataPos );
String valueSerializerFqcn = "";
dataPos += INT_SIZE + valueSerializerBytes.limit();
if ( valueSerializerBytes != null )
{
valueSerializerFqcn = Strings.utf8ToString( valueSerializerBytes );
}
// The B-tree allowDuplicates flag
int allowDuplicates = readInt( pageIos, dataPos );
boolean dupsAllowed = allowDuplicates != 0;
dataPos += INT_SIZE;
// System.out.println( "\n B-Tree " + btreeName );
// System.out.println( " ------------------------- " );
// System.out.println( " nbPageIOs[" + pageIos.length + "] = " + pageIoList );
if ( LOG.isDebugEnabled() )
{
StringBuilder sb = new StringBuilder();
boolean isFirst = true;
for ( PageIO pageIo : pageIos )
{
if ( isFirst )
{
isFirst = false;
}
else
{
sb.append( ", " );
}
sb.append( "0x" ).append( Long.toHexString( pageIo.getOffset() ) );
}
String pageIoList = sb.toString();
LOG.debug( " PageIOs[{}] = {}", pageIos.length, pageIoList );
// System.out.println( " dataSize = "+ pageIos[0].getSize() );
LOG.debug( " dataSize = {}", pageIos[0].getSize() );
LOG.debug( " B-tree '{}'", btreeName );
LOG.debug( " revision : {}", revision );
LOG.debug( " nbElems : {}", nbElems );
LOG.debug( " rootPageOffset : 0x{}", Long.toHexString( rootPageOffset ) );
LOG.debug( " B-tree page size : {}", btreePageSize );
LOG.debug( " keySerializer : '{}'", keySerializerFqcn );
LOG.debug( " valueSerializer : '{}'", valueSerializerFqcn );
LOG.debug( " dups allowed : {}", dupsAllowed );
//
// System.out.println( " B-tree '" + btreeName + "'" );
// System.out.println( " revision : " + revision );
// System.out.println( " nbElems : " + nbElems );
// System.out.println( " rootPageOffset : 0x" + Long.toHexString( rootPageOffset ) );
// System.out.println( " B-tree page size : " + btreePageSize );
// System.out.println( " keySerializer : " + keySerializerFqcn );
// System.out.println( " valueSerializer : " + valueSerializerFqcn );
// System.out.println( " dups allowed : " + dupsAllowed );
}
return rootPageOffset;
}
/**
* Get the number of managed trees. We don't count the CopiedPage B-tree and the B-tree of B-trees
*
* @return The number of managed B-trees
*/
public int getNbManagedTrees()
{
return nbBtree;
}
/**
* Get the managed B-trees. We don't return the CopiedPage B-tree nor the B-tree of B-trees.
*
* @return The managed B-trees
*/
public Set<String> getManagedTrees()
{
Set<String> btrees = new HashSet<String>( managedBtrees.keySet() );
return btrees;
}
/**
* Stores the copied pages into the CopiedPages B-tree
*
* @param name The B-tree name
* @param revision The revision
* @param copiedPages The pages that have been copied while creating this revision
* @throws IOException If we weren't able to store the data on disk
*/
/* No Qualifier */void storeCopiedPages( String name, long revision, long[] copiedPages ) throws IOException
{
RevisionName revisionName = new RevisionName( revision, name );
copiedPageBtree.insert( revisionName, copiedPages );
}
/**
* Store a reference to an old rootPage into the Revision B-tree
*
* @param btree The B-tree we want to keep an old RootPage for
* @param rootPage The old rootPage
* @throws IOException If we have an issue while writing on disk
*/
/* No qualifier */<K, V> void storeRootPage( BTree<K, V> btree, Page<K, V> rootPage ) throws IOException
{
if ( !isKeepRevisions() )
{
return;
}
if ( btree == copiedPageBtree )
{
return;
}
NameRevision nameRevision = new NameRevision( btree.getName(), rootPage.getRevision() );
( ( AbstractBTree<NameRevision, Long> ) btreeOfBtrees ).insert( nameRevision,
( ( AbstractPage<K, V> ) rootPage ).getOffset(), 0 );
if ( LOG_CHECK.isDebugEnabled() )
{
MavibotInspector.check( this );
}
}
/**
* Fetch the rootPage of a given B-tree for a given revision.
*
* @param btree The B-tree we are interested in
* @param revision The revision we want to get back
* @return The rootPage for this B-tree and this revision, if any
* @throws KeyNotFoundException If we can't find the rootPage for this revision and this B-tree
* @throws IOException If we had an ise while accessing the data on disk
*/
/* No qualifier */<K, V> Page<K, V> getRootPage( BTree<K, V> btree, long revision ) throws KeyNotFoundException,
IOException
{
if ( btree.getRevision() == revision )
{
// We are asking for the current revision
return btree.getRootPage();
}
// Get the B-tree header offset
NameRevision nameRevision = new NameRevision( btree.getName(), revision );
long btreeHeaderOffset = btreeOfBtrees.get( nameRevision );
// get the B-tree rootPage
Page<K, V> btreeRoot = readRootPage( btree, btreeHeaderOffset );
return btreeRoot;
}
/**
* Read a root page from the B-tree header offset
*/
private <K, V> Page<K, V> readRootPage( BTree<K, V> btree, long btreeHeaderOffset )
throws EndOfFileExceededException, IOException
{
// Read the B-tree header pages on disk
PageIO[] btreeHeaderPageIos = readPageIOs( btreeHeaderOffset, Long.MAX_VALUE );
long dataPos = LONG_SIZE + LONG_SIZE;
// The B-tree rootPage offset
long rootPageOffset = readLong( btreeHeaderPageIos, dataPos );
// Read the rootPage pages on disk
PageIO[] rootPageIos = readPageIOs( rootPageOffset, Long.MAX_VALUE );
// Now, convert it to a Page
Page<K, V> btreeRoot = readPage( btree, rootPageIos );
return btreeRoot;
}
/**
* Get one managed trees, knowing its name.
*
* @param name The B-tree name we are looking for
* @return The managed B-trees
*/
public <K, V> BTree<K, V> getManagedTree( String name )
{
return ( BTree<K, V> ) managedBtrees.get( name );
}
/**
* Move a list of pages to the free page list. A logical page is associated with one
* or more physical PageIOs, which are on the disk. We have to move all those PagIO instances
* to the free list, and do the same in memory (we try to keep a reference to a set of
* free pages.
*
* @param btree The B-tree which were owning the pages
* @param revision The current revision
* @param pages The pages to free
* @throws IOException If we had a problem while updating the file
* @throws EndOfFileExceededException If we tried to write after the end of the file
*/
/* Package protected */<K, V> void freePages( BTree<K, V> btree, long revision, List<Page<K, V>> pages )
throws EndOfFileExceededException, IOException
{
if ( ( pages == null ) || pages.isEmpty() )
{
return;
}
if ( !keepRevisions )
{
// if the B-tree doesn't keep revisions, we can safely move
// the pages to the freed page list.
if ( LOG.isDebugEnabled() )
{
LOG.debug( "Freeing the following pages :" );
for ( Page<K, V> page : pages )
{
LOG.debug( " {}", page );
}
}
for ( Page<K, V> page : pages )
{
long pageOffset = ( ( AbstractPage<K, V> ) page ).getOffset();
PageIO[] pageIos = readPageIOs( pageOffset, Long.MAX_VALUE );
for ( PageIO pageIo : pageIos )
{
freedPages.add( pageIo );
}
}
}
else
{
// We are keeping revisions of standard B-trees, so we move the pages to the CopiedPages B-tree
// but only for non managed B-trees
if ( LOG.isDebugEnabled() )
{
LOG.debug( "Moving the following pages to the CopiedBtree :" );
for ( Page<K, V> page : pages )
{
LOG.debug( " {}", page );
}
}
long[] pageOffsets = new long[pages.size()];
int pos = 0;
for ( Page<K, V> page : pages )
{
pageOffsets[pos++] = ( ( AbstractPage<K, V> ) page ).offset;
}
if ( ( btree.getType() != BTreeTypeEnum.BTREE_OF_BTREES )
&& ( btree.getType() != BTreeTypeEnum.COPIED_PAGES_BTREE ) )
{
// Deal with standard B-trees
RevisionName revisionName = new RevisionName( revision, btree.getName() );
copiedPageBtree.insert( revisionName, pageOffsets );
// Update the RecordManager Copiedpage Offset
currentCopiedPagesBtreeOffset = ( ( PersistedBTree<RevisionName, long[]> ) copiedPageBtree ).getBtreeOffset();
}
else
{
// Managed B-trees : we simply free the copied pages
for ( long pageOffset : pageOffsets )
{
PageIO[] pageIos = readPageIOs( pageOffset, Long.MAX_VALUE );
for ( PageIO pageIo : pageIos )
{
freedPages.add( pageIo );
}
}
}
}
}
/**
* Add a PageIO to the list of free PageIOs
*
* @param pageIo The page to free
* @throws IOException If we weren't capable of updating the file
*/
/* no qualifier */ void free( PageIO pageIo ) throws IOException
{
freePageLock.lock();
// We add the Page's PageIOs before the
// existing free pages.
// Link it to the first free page
pageIo.setNextPage( firstFreePage );
LOG.debug( "Flushing the first free page" );
// And flush it to disk
//FIXME can be flushed last after releasing the lock
flushPages( pageIo );
// We can update the firstFreePage offset
firstFreePage = pageIo.getOffset();
freePageLock.unlock();
}
/**
* Add an array of PageIOs to the list of free PageIOs
*
* @param offsets The offsets of the pages whose associated PageIOs will be fetched and freed.
* @throws IOException If we weren't capable of updating the file
*/
/*no qualifier*/ void free( long... offsets ) throws IOException
{
freePageLock.lock();
List<PageIO> pageIos = new ArrayList<PageIO>();
int pageIndex = 0;
for ( int i = 0; i < offsets.length; i++ )
{
PageIO[] ios = readPageIOs( offsets[i], Long.MAX_VALUE );
for ( PageIO io : ios )
{
pageIos.add( io );
if ( pageIndex > 0 )
{
pageIos.get( pageIndex - 1 ).setNextPage( io.getOffset() );
}
pageIndex++;
}
}
// We add the Page's PageIOs before the
// existing free pages.
// Link it to the first free page
pageIos.get( pageIndex - 1 ).setNextPage( firstFreePage );
LOG.debug( "Flushing the first free page" );
// And flush it to disk
//FIXME can be flushed last after releasing the lock
flushPages( pageIos.toArray( new PageIO[0] ) );
// We can update the firstFreePage offset
firstFreePage = pageIos.get( 0 ).getOffset();
freePageLock.unlock();
}
/**
* @return the keepRevisions flag
*/
public boolean isKeepRevisions()
{
return keepRevisions;
}
/**
* @param keepRevisions the keepRevisions flag to set
*/
public void setKeepRevisions( boolean keepRevisions )
{
this.keepRevisions = keepRevisions;
}
/**
* Creates a B-tree and automatically adds it to the list of managed btrees
*
* @param name the name of the B-tree
* @param keySerializer key serializer
* @param valueSerializer value serializer
* @param allowDuplicates flag for allowing duplicate keys
* @return a managed B-tree
* @throws IOException If we weren't able to update the file on disk
* @throws BTreeAlreadyManagedException If the B-tree is already managed
*/
@SuppressWarnings("all")
public <K, V> BTree<K, V> addBTree( String name, ElementSerializer<K> keySerializer,
ElementSerializer<V> valueSerializer, boolean allowDuplicates )
throws IOException, BTreeAlreadyManagedException
{
PersistedBTreeConfiguration config = new PersistedBTreeConfiguration();
config.setName( name );
config.setKeySerializer( keySerializer );
config.setValueSerializer( valueSerializer );
config.setAllowDuplicates( allowDuplicates );
BTree btree = new PersistedBTree( config );
manage( btree );
if ( LOG_CHECK.isDebugEnabled() )
{
MavibotInspector.check( this );
}
return btree;
}
/**
* Add a newly closd transaction into the closed transaction queue
*/
/* no qualifier */<K, V> void releaseTransaction( ReadTransaction<K, V> readTransaction )
{
RevisionName revisionName = new RevisionName(
readTransaction.getRevision(),
readTransaction.getBtreeHeader().getBtree().getName() );
//closedTransactionsQueue.add( revisionName );
}
/**
* Get the current BTreeHeader for a given Btree. It might not exist
*/
public BTreeHeader getBTreeHeader( String name )
{
// Get a lock
btreeHeadersLock.readLock().lock();
// get the current BTree Header for this BTree and revision
BTreeHeader<?, ?> btreeHeader = currentBTreeHeaders.get( name );
// And unlock
btreeHeadersLock.readLock().unlock();
return btreeHeader;
}
/**
* Get the new BTreeHeader for a given Btree. It might not exist
*/
public BTreeHeader getNewBTreeHeader( String name )
{
// get the current BTree Header for this BTree and revision
BTreeHeader<?, ?> btreeHeader = newBTreeHeaders.get( name );
return btreeHeader;
}
/**
* {@inheritDoc}
*/
public void updateNewBTreeHeaders( BTreeHeader btreeHeader )
{
newBTreeHeaders.put( btreeHeader.getBtree().getName(), btreeHeader );
}
/**
* Swap the current BtreeHeader map with the new one. This method will only
* be called in a single trhead, when the current transaction will be committed.
*/
private void swapCurrentBtreeHeaders()
{
// Copy the reference to the current BtreeHeader Map
Map<String, BTreeHeader<?, ?>> tmp = currentBTreeHeaders;
// Get a write lock
btreeHeadersLock.writeLock().lock();
// Swap the new BTreeHeader Map
currentBTreeHeaders = newBTreeHeaders;
// And unlock
btreeHeadersLock.writeLock().unlock();
// Last, not least, clear the Map and reinject the latest revision in it
tmp.clear();
tmp.putAll( currentBTreeHeaders );
// And update the new BTreeHeader map
newBTreeHeaders = tmp;
}
/**
* revert the new BTreeHeaders Map to the current BTreeHeader Map. This method
* is called when we have to rollback a transaction.
*/
private void revertBtreeHeaders()
{
// Clean up teh new BTreeHeaders Map
newBTreeHeaders.clear();
// Reinject the latest revision in it
newBTreeHeaders.putAll( currentBTreeHeaders );
}
/**
* Loads a B-tree holding the values of a duplicate key
* This tree is also called as dups tree or sub tree
*
* @param offset the offset of the B-tree header
* @return the deserialized B-tree
*/
/* No qualifier */<K, V> BTree<V, V> loadDupsBtree( long btreeHeaderOffset, BTree<K, V> parentBtree )
{
PageIO[] pageIos = null;
try
{
pageIos = readPageIOs( btreeHeaderOffset, Long.MAX_VALUE );
BTree<V, V> subBtree = BTreeFactory.<V, V> createPersistedBTree( BTreeTypeEnum.PERSISTED_SUB );
loadBtree( pageIos, subBtree, parentBtree );
return subBtree;
}
catch ( Exception e )
{
// should not happen
throw new BTreeCreationException( e );
}
}
private void checkFreePages() throws EndOfFileExceededException, IOException
{
//System.out.println( "Checking the free pages, starting from " + Long.toHexString( firstFreePage ) );
// read all the free pages, add them into a set, to be sure we don't have a cycle
Set<Long> freePageOffsets = new HashSet<Long>();
long currentFreePageOffset = firstFreePage;
while ( currentFreePageOffset != NO_PAGE )
{
//System.out.println( "Next page offset :" + Long.toHexString( currentFreePageOffset ) );
if ( ( currentFreePageOffset % pageSize ) != 0 )
{
throw new InvalidOffsetException( "Wrong offset : " + Long.toHexString( currentFreePageOffset ) );
}
if ( freePageOffsets.contains( currentFreePageOffset ) )
{
throw new InvalidOffsetException( "Offset : " + Long.toHexString( currentFreePageOffset )
+ " already read, there is a cycle" );
}
freePageOffsets.add( currentFreePageOffset );
PageIO pageIO = fetchPage( currentFreePageOffset );
currentFreePageOffset = pageIO.getNextPage();
}
return;
}
/**
* sets the threshold of the number of commits to be performed before
* reclaiming the free pages.
*
* @param pageReclaimerThreshold the number of commits before the reclaimer runs
*/
/* no qualifier */ void setPageReclaimerThreshold( int pageReclaimerThreshold )
{
this.pageReclaimerThreshold = pageReclaimerThreshold;
}
/* no qualifier */ void _disableReclaimer( boolean toggle )
{
this.disableReclaimer = toggle;
}
/**
* @see Object#toString()
*/
public String toString()
{
StringBuilder sb = new StringBuilder();
sb.append( "RM free pages : [" );
if ( firstFreePage != NO_PAGE )
{
long current = firstFreePage;
boolean isFirst = true;
while ( current != NO_PAGE )
{
if ( isFirst )
{
isFirst = false;
}
else
{
sb.append( ", " );
}
PageIO pageIo;
try
{
pageIo = fetchPage( current );
sb.append( pageIo.getOffset() );
current = pageIo.getNextPage();
}
catch ( EndOfFileExceededException e )
{
e.printStackTrace();
}
catch ( IOException e )
{
e.printStackTrace();
}
}
}
sb.append( "]" );
return sb.toString();
}
}