Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[3873] branches/JOURNAL_HA_BRANCH/bigdata/src
|
From: <tho...@us...> - 2010-11-02 19:00:39
|
Revision: 3873
http://bigdata.svn.sourceforge.net/bigdata/?rev=3873&view=rev
Author: thompsonbry
Date: 2010-11-02 19:00:31 +0000 (Tue, 02 Nov 2010)
Log Message:
-----------
A bunch of edits to clean up RWStrategy and RWStore.
There is a deadlock which can be demonstrated using StressTestConcurrentUnisolatedIndices and which results in a deadlock. It seems like there are probably too many distinct locks -- or that the m_allocationLock needs to be acquired before the m_deferFreeLock (which probably means that the m_deferFreeLock is redundant). Either way, this should be easy enough to clean up.
This thread is holding the Journal's internal lock, holding the RWStore#m_deferFreeLock (in saveDeferrals), and waiting on RWStore#m_allocationLock (in alloc()):
com.bigdata.journal.ConcurrencyManager.writeService14 [WAITING] CPU time: 0:00
java.util.concurrent.locks.ReentrantLock.lock()
com.bigdata.rwstore.RWStore.alloc(int, IAllocationContext)
com.bigdata.rwstore.RWStore.alloc(byte[], int, IAllocationContext)
com.bigdata.rwstore.PSOutputStream.save()
com.bigdata.rwstore.RWStore.saveDeferrals()
com.bigdata.journal.RWStrategy.saveDeleteBlocks()
com.bigdata.journal.DeleteBlockCommitter.handleCommit(long)
com.bigdata.journal.AbstractJournal.notifyCommitters(long)
com.bigdata.journal.AbstractJournal.commitNow(long)
com.bigdata.journal.AbstractJournal.commit()
com.bigdata.journal.WriteExecutorService.commit(boolean)
com.bigdata.journal.WriteExecutorService.groupCommit()
com.bigdata.journal.WriteExecutorService.afterTask(AbstractTask, Throwable)
com.bigdata.journal.AbstractTask.doUnisolatedReadWriteTask()
com.bigdata.journal.AbstractTask.call2()
com.bigdata.journal.AbstractTask.call()
java.util.concurrent.FutureTask.run()
com.bigdata.concurrent.NonBlockingLockManagerWithNewDesign$LockFutureTask.run()
java.lang.Thread.run()
This thread is holding m_allocationLock (in free()) and waiting on m_deferFreeLock:
com.bigdata.journal.ConcurrencyManager.writeService7 [WAITING] CPU time: 0:00
java.util.concurrent.locks.ReentrantLock.lock()
com.bigdata.rwstore.RWStore.deferFree(int, int)
com.bigdata.rwstore.RWStore.free(long, int, IAllocationContext)
com.bigdata.journal.RWStrategy.delete(long, IAllocationContext)
com.bigdata.journal.RWStrategy.delete(long)
com.bigdata.journal.AbstractJournal.delete(long)
com.bigdata.btree.Node.replaceChildRef(long, AbstractNode)
com.bigdata.btree.AbstractNode.copyOnWrite(long)
com.bigdata.btree.AbstractNode.copyOnWrite()
com.bigdata.btree.Leaf.insert(byte[], byte[], boolean, long, Tuple)
com.bigdata.btree.Node.insert(byte[], byte[], boolean, long, Tuple)
com.bigdata.btree.Node.insert(byte[], byte[], boolean, long, Tuple)
com.bigdata.btree.AbstractBTree.insert(byte[], byte[], boolean, long, Tuple)
com.bigdata.btree.AbstractBTree.insert(byte[], byte[])
com.bigdata.journal.StressTestConcurrentUnisolatedIndices$WriteTask.doTask()
com.bigdata.journal.AbstractTask$InnerWriteServiceCallable.call()
com.bigdata.journal.AbstractTask.doUnisolatedReadWriteTask()
com.bigdata.journal.AbstractTask.call2()
com.bigdata.journal.AbstractTask.call()
java.util.concurrent.FutureTask.run()
com.bigdata.concurrent.NonBlockingLockManagerWithNewDesign$LockFutureTask.run()
java.lang.Thread.run()
Modified Paths:
--------------
branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/io/writecache/WriteCacheService.java
branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/IBufferStrategy.java
branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/RWStrategy.java
branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/rwstore/RWStore.java
branches/JOURNAL_HA_BRANCH/bigdata/src/test/com/bigdata/journal/StressTestConcurrentTx.java
Modified: branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/io/writecache/WriteCacheService.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/io/writecache/WriteCacheService.java 2010-11-02 17:46:33 UTC (rev 3872)
+++ branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/io/writecache/WriteCacheService.java 2010-11-02 19:00:31 UTC (rev 3873)
@@ -325,10 +325,10 @@
int addrsUsedCurs = 0;
final char[] addrActions = new char[addrsUsed.length];
final int[] addrLens = new int[addrsUsed.length];
-*/ final long[] addrsUsed = null;
- int addrsUsedCurs = 0;
- final char[] addrActions = null;
- final int[] addrLens = null;
+*/ private final long[] addrsUsed = null;
+ private int addrsUsedCurs = 0;
+ private final char[] addrActions = null;
+ private final int[] addrLens = null;
/**
* The current file extent.
@@ -1989,7 +1989,7 @@
* An array of writeCache actions is maintained that can be used
* to provide a breadcrumb of how that address has been written, saved,
* freed or removed.
- *
+ * <p>
* Write errors often show up as a checksum error, so the length of
* data written to the address cab be crucial information in determining the
* root of any problem.
@@ -2002,9 +2002,9 @@
return "No WriteCache debug info";
}
- StringBuffer ret = new StringBuffer();
- // first see if address was ever written
- boolean written = false;
+ final StringBuffer ret = new StringBuffer();
+// // first see if address was ever written
+// boolean written = false;
for (int i = 0; i < addrsUsed.length; i++) {
if (i == addrsUsedCurs) {
ret.append("|...|");
Modified: branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/IBufferStrategy.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/IBufferStrategy.java 2010-11-02 17:46:33 UTC (rev 3872)
+++ branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/IBufferStrategy.java 2010-11-02 19:00:31 UTC (rev 3873)
@@ -28,7 +28,6 @@
import java.nio.ByteBuffer;
import com.bigdata.counters.CounterSet;
-import com.bigdata.io.writecache.WriteCacheService;
import com.bigdata.rawstore.IAddressManager;
import com.bigdata.rawstore.IMRMW;
import com.bigdata.rawstore.IRawStore;
@@ -204,55 +203,65 @@
public IAddressManager getAddressManager();
/**
- * A method that removes assumptions of how a specific strategy determines whether a transaction commit is required.
+ * A method that removes assumptions of how a specific strategy determines
+ * whether a transaction commit is required.
*
* @param block
- * The root block held by the client, can be checked against the state of the Buffer Strategy
+ * The root block held by the client, can be checked against the
+ * state of the Buffer Strategy
* @return whether any modification has occurred.
*/
- public boolean requiresCommit(IRootBlockView block);
+ public boolean requiresCommit(IRootBlockView block);
/**
- * A method that removes assumptions of how a specific strategy commits data. For most strategies the action is void
- * since the client WORM DISK strategy writes data as allocated. For the Read Write Strategy more data must be managed
- * as part of the protocol outside of the RootBlock, and this is the method that triggers that management.
- * @param abstractJournal
+ * A method that removes assumptions of how a specific strategy commits
+ * data. For most strategies the action is void since the client WORM DISK
+ * strategy writes data as allocated. For the Read Write Strategy more data
+ * must be managed as part of the protocol outside of the RootBlock, and
+ * this is the method that triggers that management.
+ *
+ * @param abstractJournal
*/
- public void commit(IJournal journal);
+ public void commit(IJournal journal);
/**
* A method that requires the implementation to discard its buffered write
* set (if any). The caller is responsible for any necessary synchronization
* as part of the abort protocol.
*/
- public void abort();
-
- /**
- * The RWStrategy requires meta allocation info in the root block, this method is the hook to enable access.
- * The metaStartAddr is the address in the file where the allocation blocks are stored.
- *
- * @return the metaStartAddr for the root block if any
- */
- public long getMetaStartAddr();
- /**
- * The RWStrategy requires meta allocation info in the root block, this method is the hook to enable access.
- * The metaBitsAddr is the address in the file where the metaBits that control the allocation of the allocation
- * blocks themselves is stored.
- *
- * @return the metaBitsAddr for the root block if any
- */
- public long getMetaBitsAddr();
- /**
- * @return the number of bits available in the address to define offset
- */
+ public void abort();
- public int getOffsetBits();
- /**
- * @return the maximum record size supported by this strategy
- */
- public int getMaxRecordSize();
+ /**
+ * The RWStrategy requires meta allocation info in the root block, this
+ * method is the hook to enable access. The metaStartAddr is the address in
+ * the file where the allocation blocks are stored.
+ *
+ * @return the metaStartAddr for the root block if any
+ */
+ public long getMetaStartAddr();
/**
+ * The RWStrategy requires meta allocation info in the root block, this
+ * method is the hook to enable access. The metaBitsAddr is the address in
+ * the file where the metaBits that control the allocation of the allocation
+ * blocks themselves is stored.
+ *
+ * @return the metaBitsAddr for the root block if any
+ */
+ public long getMetaBitsAddr();
+
+ /**
+ * @return the number of bits available in the address to define offset
+ */
+
+ public int getOffsetBits();
+
+ /**
+ * @return the maximum record size supported by this strategy
+ */
+ public int getMaxRecordSize();
+
+ /**
* Return <code>true</code> if the store uses per-record checksums. When
* <code>true</code>, an additional 4 bytes are written after the record on
* the disk. Those bytes contain the checksum of the record.
Modified: branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/RWStrategy.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/RWStrategy.java 2010-11-02 17:46:33 UTC (rev 3872)
+++ branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/journal/RWStrategy.java 2010-11-02 19:00:31 UTC (rev 3873)
@@ -24,15 +24,12 @@
package com.bigdata.journal;
-import java.io.DataInput;
-import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedByInterruptException;
import java.util.UUID;
-import java.util.concurrent.Callable;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.log4j.Logger;
@@ -45,7 +42,6 @@
import com.bigdata.rawstore.IAddressManager;
import com.bigdata.rwstore.IAllocationContext;
import com.bigdata.rwstore.RWStore;
-import com.bigdata.service.AbstractTransactionService;
import com.bigdata.util.ChecksumUtility;
/**
@@ -68,8 +64,9 @@
* @author Martyn Cutcher
*/
public class RWStrategy extends AbstractRawStore implements IBufferStrategy, IHABufferStrategy {
- protected static final Logger log = Logger.getLogger(RWStrategy.class);
+ private static final transient Logger log = Logger.getLogger(RWStrategy.class);
+
final private FileMetadata m_fileMetadata;
final private Quorum<?,?> m_environment;
@@ -88,14 +85,20 @@
private volatile IRootBlockView m_rb;
private volatile IRootBlockView m_rb0;
private volatile IRootBlockView m_rb1;
-
- final ReentrantLock m_commitLock = new ReentrantLock();
-
+
/**
- * Access to the transaction manager of any owning Journal, needed by
- * RWStrategy to manager deleted data.
+ * @todo The use of this lock is suspicious. It is only used by
+ * {@link #commit(IJournal)} and that method is invoked by the
+ * {@link AbstractJournal#commitNow(long)} which is already protected
+ * by a lock.
*/
- private AbstractLocalTransactionManager localTransactionManager = null;
+ private final ReentrantLock m_commitLock = new ReentrantLock();
+
+// /**
+// * Access to the transaction manager of any owning Journal, needed by
+// * RWStrategy to manager deleted data.
+// */
+// private AbstractLocalTransactionManager localTransactionManager = null;
// CounterSet m_counters = new CounterSet();
@@ -125,16 +128,26 @@
}
/**
+ * Return a copy of the current {@link IRootBlockView}.
+ *
* @param rb0
- * @return
+ * When <code>true</code> the view will be flagged as root block
+ * ZERO (0). Otherwise it is flagged as root block ONE (1).
+ *
+ * @return The {@link IRootBlockView}.
*/
- IRootBlockView copyRootBlock(boolean rb0) {
- IRootBlockView rbv = new RootBlockView(rb0, m_rb.getOffsetBits(), m_rb.getNextOffset(), m_rb.getFirstCommitTime(), m_rb.getLastCommitTime(),
- m_rb.getCommitCounter(), m_rb.getCommitRecordAddr(), m_rb.getCommitRecordIndexAddr(), m_fileMetadata.rootBlock.getUUID(),
- m_rb.getQuorumToken(),
- m_rb.getMetaStartAddr(), m_rb.getMetaBitsAddr(), StoreTypeEnum.RW, m_fileMetadata.rootBlock.getCreateTime(), m_rb.getCloseTime(),
- s_ckutil);
+ private IRootBlockView copyRootBlock(final boolean rb0) {
+ final IRootBlockView rbv = new RootBlockView(rb0, m_rb.getOffsetBits(),
+ m_rb.getNextOffset(), m_rb.getFirstCommitTime(), m_rb
+ .getLastCommitTime(), m_rb.getCommitCounter(), m_rb
+ .getCommitRecordAddr(),
+ m_rb.getCommitRecordIndexAddr(), m_fileMetadata.rootBlock
+ .getUUID(), m_rb.getQuorumToken(), m_rb
+ .getMetaStartAddr(), m_rb.getMetaBitsAddr(),
+ StoreTypeEnum.RW, m_fileMetadata.rootBlock.getCreateTime(),
+ m_rb.getCloseTime(), s_ckutil);
+
return rbv;
}
@@ -192,14 +205,19 @@
return m_fileMetadata.raf;
}
- public IRootBlockView newRootBlockView(boolean rootBlock0, int offsetBits, long nextOffset,
- long firstCommitTime, long lastCommitTime, long commitCounter, long commitRecordAddr,
- long commitRecordIndexAddr, long metaStartAddr, long metaBitsAddr, long closeTime) {
+ public IRootBlockView newRootBlockView(boolean rootBlock0,
+ int offsetBits, long nextOffset, long firstCommitTime,
+ long lastCommitTime, long commitCounter, long commitRecordAddr,
+ long commitRecordIndexAddr, long metaStartAddr,
+ long metaBitsAddr, long closeTime) {
- IRootBlockView rbv = new RootBlockView(rootBlock0, offsetBits, nextOffset, firstCommitTime, lastCommitTime,
- commitCounter, commitRecordAddr, commitRecordIndexAddr, m_fileMetadata.rootBlock.getUUID(), -1 /* FIXME: quorumToken */,
- metaStartAddr, metaBitsAddr, StoreTypeEnum.RW, m_fileMetadata.rootBlock.getCreateTime(), closeTime,
- s_ckutil);
+ final IRootBlockView rbv = new RootBlockView(rootBlock0,
+ offsetBits, nextOffset, firstCommitTime, lastCommitTime,
+ commitCounter, commitRecordAddr, commitRecordIndexAddr,
+ m_fileMetadata.rootBlock.getUUID(),
+ -1 /* FIXME: quorumToken */, metaStartAddr, metaBitsAddr,
+ StoreTypeEnum.RW, m_fileMetadata.rootBlock.getCreateTime(),
+ closeTime, s_ckutil);
// writeRootBlock(rbv, ForceEnum.Force); // not sure if this is really needed now!
@@ -211,17 +229,19 @@
}
}
- public ByteBuffer readRootBlock(boolean rootBlock0) {
- checkReopen();
+ public ByteBuffer readRootBlock(final boolean rootBlock0) {
+
+ checkReopen();
IRootBlockView rbv = rootBlock0 ? m_rb0 : m_rb1;
return rbv.asReadOnlyBuffer();
}
- public ByteBuffer read(long addr) {
- checkReopen();
+ public ByteBuffer read(final long addr) {
+ checkReopen();
+
int rwaddr = decodeAddr(addr);
int sze = decodeSize(addr);
@@ -239,12 +259,15 @@
return ByteBuffer.wrap(buf, 0, sze);
}
- public long write(ByteBuffer data) {
- return write(data, null);
+ public long write(final ByteBuffer data) {
+
+ return write(data, null);
+
}
- public long write(ByteBuffer data, IAllocationContext context) {
- checkReopen();
+ public long write(final ByteBuffer data, final IAllocationContext context) {
+
+ checkReopen();
if (data == null) {
throw new IllegalArgumentException();
@@ -291,7 +314,7 @@
// return encodeAddr(m_store.alloc(nbytes), nbytes);
// }
- private long encodeAddr(long alloc, int nbytes) {
+ private long encodeAddr(long alloc, final int nbytes) {
alloc <<= 32;
alloc += nbytes;
@@ -304,52 +327,70 @@
return (int) addr;
}
- private int decodeSize(long addr) {
+ private int decodeSize(final long addr) {
return (int) (addr & 0xFFFFFFFF);
}
- public void delete(long addr) {
- delete(addr, null);
+ public void delete(final long addr) {
+
+ delete(addr, null/* IAllocationContext */);
+
}
/**
* Must check whether there are existing transactions which may access
* this data, and if not free immediately, otherwise defer.
*/
- public void delete(long addr, IAllocationContext context) {
+ public void delete(final long addr, final IAllocationContext context) {
final int rwaddr = decodeAddr(addr);
final int sze = decodeSize(addr);
m_store.free(rwaddr, sze, context);
+
}
- public void detachContext(IAllocationContext context) {
- m_store.detachContext(context);
+ public void detachContext(final IAllocationContext context) {
+
+ m_store.detachContext(context);
+
}
-
+ /*
+ * FIXME Reconcile this class with the methods on the outer class.
+ */
public static class RWAddressManager implements IAddressManager {
- public int getByteCount(long addr) {
- return (int) addr & 0xFFFFFF;
+ public int getByteCount(final long addr) {
+
+ return (int) addr & 0xFFFFFF;
+
}
- public long getOffset(long addr) {
- return -(addr >> 32);
+ public long getOffset(final long addr) {
+
+ return -(addr >> 32);
+
}
- public long toAddr(int nbytes, long offset) {
- offset <<= 32;
+ public long toAddr(final int nbytes, long offset) {
+
+ offset <<= 32;
return offset + nbytes;
+
}
- public String toString(long addr) {
- return "{off="+getOffset(addr)+",len="+getByteCount(addr)+"}";
+ public String toString(final long addr) {
+
+ return "{off=" + getOffset(addr) + ",len=" + getByteCount(addr)
+ + "}";
+
}
+
}
- IAddressManager m_am = new RWAddressManager();
+
+ final private IAddressManager m_am = new RWAddressManager();
public IAddressManager getAddressManager() {
return m_am;
@@ -382,9 +423,9 @@
return 0;
}
- long m_initialExtent = 0;
+ private long m_initialExtent = 0;
- private boolean m_needsReopen = false;
+ private volatile boolean m_needsReopen = false;
public long getInitialExtent() {
return m_initialExtent;
@@ -485,19 +526,33 @@
}
public void deleteResources() {
- if (m_fileMetadata.raf != null && m_fileMetadata.raf.getChannel().isOpen()) {
- throw new IllegalStateException("Backing store is open");
- }
- if (m_fileMetadata.file.exists()) {
- try {
- if (!m_fileMetadata.file.delete()) {
- log.warn("Unable to delete file: " + m_fileMetadata.file);
- }
- } catch (SecurityException e) {
- log.warn("Problem deleting file", e);
- }
- }
+ if (m_fileMetadata.raf != null
+ && m_fileMetadata.raf.getChannel().isOpen()) {
+
+ throw new IllegalStateException("Backing store is open: "
+ + m_fileMetadata.file);
+
+ }
+
+ if (m_fileMetadata.file.exists()) {
+
+ try {
+
+ if (!m_fileMetadata.file.delete()) {
+
+ log.warn("Unable to delete file: " + m_fileMetadata.file);
+
+ }
+
+ } catch (SecurityException e) {
+
+ log.warn("Problem deleting file", e);
+
+ }
+
+ }
+
}
public void destroy() {
@@ -515,10 +570,18 @@
}
public IRootBlockView getRootBlock() {
- return m_fmv.newRootBlockView(! m_rb.isRootBlock0(), m_rb.getOffsetBits(), getNextOffset(),
- m_rb.getFirstCommitTime(), m_rb.getLastCommitTime(), m_rb.getCommitCounter(),
- m_rb.getCommitRecordAddr(), m_rb.getCommitRecordIndexAddr(), getMetaStartAddr(), getMetaBitsAddr(), m_rb.getCloseTime() );
-
+ return m_fmv.newRootBlockView(!m_rb.isRootBlock0(), //
+ m_rb.getOffsetBits(),//
+ getNextOffset(), //
+ m_rb.getFirstCommitTime(),//
+ m_rb.getLastCommitTime(), //
+ m_rb.getCommitCounter(),//
+ m_rb.getCommitRecordAddr(), //
+ m_rb.getCommitRecordIndexAddr(), //
+ getMetaStartAddr(),//
+ getMetaBitsAddr(), //
+ m_rb.getCloseTime()//
+ );
}
/**
@@ -526,7 +589,7 @@
*
* Must pass in earliestTxnTime to commitChanges to enable
*/
- public void commit(IJournal journal) {
+ public void commit(final IJournal journal) {
m_commitLock.lock();
try {
m_store.commitChanges((Journal) journal); // includes a force(false)
@@ -544,7 +607,7 @@
m_store.reset();
}
- public void force(boolean metadata) {
+ public void force(final boolean metadata) {
try {
m_store.flushWrites(metadata);
} catch (ClosedByInterruptException e) {
@@ -572,7 +635,8 @@
log.warn("Request to reopen store after interrupt");
m_store.close();
- m_fileMetadata.raf = new RandomAccessFile(m_fileMetadata.file, m_fileMetadata.fileMode);
+ m_fileMetadata.raf = new RandomAccessFile(m_fileMetadata.file,
+ m_fileMetadata.fileMode);
m_store = new RWStore(m_fmv, false, m_environment); // never read-only for now
m_needsReopen = false;
m_open = true;
@@ -592,11 +656,19 @@
return m_fileMetadata.raf;
}
- public IResourceMetadata getResourceMetadata() {
- // TODO Auto-generated method stub
- return null;
- }
-
+
+ /**
+ * Not supported - this is available on the {@link AbstractJournal}.
+ *
+ * @throws UnsupportedOperationException
+ * always
+ */
+ public IResourceMetadata getResourceMetadata() {
+
+ throw new UnsupportedOperationException();
+
+ }
+
public UUID getUUID() {
return m_fileMetadata.rootBlock.getUUID();
}
@@ -631,25 +703,14 @@
return addr >> 32;
}
- public void packAddr(DataOutput out, long addr) throws IOException {
- // TODO Auto-generated method stub
-
- }
-
- public long toAddr(int nbytes, long offset) {
+ public long toAddr(final int nbytes, final long offset) {
return (offset << 32) + nbytes;
}
- public String toString(long addr) {
- // TODO Auto-generated method stub
- return null;
+ public String toString(final long addr) {
+ return m_am.toString(addr);
}
- public long unpackAddr(DataInput in) throws IOException {
- // TODO Auto-generated method stub
- return 0;
- }
-
/**
* The state of the provided block is not relevant since it does not hold
* information on recent allocations (the meta allocations will only effect the
Modified: branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/rwstore/RWStore.java
===================================================================
--- branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/rwstore/RWStore.java 2010-11-02 17:46:33 UTC (rev 3872)
+++ branches/JOURNAL_HA_BRANCH/bigdata/src/java/com/bigdata/rwstore/RWStore.java 2010-11-02 19:00:31 UTC (rev 3873)
@@ -35,11 +35,10 @@
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -70,128 +69,130 @@
/**
* Storage class
- *
- * Author: Martyn Cutcher
- *
+ * <p>
* Provides an interface to allocating storage within a disk file.
- *
+ * <p>
* Essentially provides a DiskMalloc interface.
- *
+ * <p>
* In addition to the DiskMalloc/ReAlloc mechanism, a single root address can be
* associated. This can be used when opening an existing storage file to
* retrieve some management object - such as an object manager!
- *
+ * <p>
* The allocator also support atomic update via a simple transaction mechanism.
- *
+ * <p>
* Updates are normally committed immediately, but by using startTransaction and
* commitTransaction, the previous state of the store is retained until the
* moment of commitment.
- *
- * It would also be possible to add some journalling/version mechanism, where
+ * <p>
+ * It would also be possible to add some journaling/version mechanism, where
* snapshots of the allocation maps are retained for sometime. For a store which
* was only added to this would not be an unreasonable overhead and would
* support the rolling back of the database weekly or monthly if required.
- *
+ * <p>
* The input/output mechanism uses ByteArray Input and Output Streams.
- *
+ * <p>
* One difference between the disk realloc and in memory realloc is that the
* disk realloc will always return a new address and mark the old address as
* ready to be freed.
- *
+ * <p>
* The method of storing the allocation headers has been changed from always
- * allocting at the end of the file (and moving them on fle extend) to
+ * allocating at the end of the file (and moving them on fle extend) to
* allocation of fixed areas. The meta-allocation data, containing the bitmap
* that controls these allocations, is itself stored in the heap, and is now
* structured to include both the bit data and the list of meta-storage
* addresses.
- *
+ * <p>
* Sizing:
* 256 allocators would reference approximately 2M objects/allocations. At 1K
* per allocator this would require 250K of store. The meta-allocation data
* would therefore need a start address plus 32 bytes (or 8 ints) to represent
- * the meta-allocation bits. An array of such data referencing sequentialy
+ * the meta-allocation bits. An array of such data referencing sequentially
* allocated storage areas completes the meta-allocation requirements.
- *
+ * <p>
* A meta-allocation address can therefore be represented as a single bit offset
* from which the block, providing start address, and bit offset can be
* directly determined.
- *
- * The m_metaBits int array used to be fully used as allocaiton bits, but
+ * <p>
+ * The m_metaBits int array used to be fully used as allocation bits, but
* now stores both the start address plus the 8 ints used to manage that data
* block.
- *
+ * <p>
* Allocation is reduced to sets of allocator objects which have a start address
* and a bitmap of allocated storage maps.
- *
+ * <p>
* Searching thousands of allocation blocks to find storage is not efficient,
- * but by utilising roving pointers and sorting blocks with free space available
+ * but by utilizing roving pointers and sorting blocks with free space available
* this can be made most efficient.
- *
+ * <p>
* In order to provide optimum use of bitmaps, this implementation will NOT use
* the BitSet class.
- *
+ * <p>
* Using the meta-allocation bits, it is straightforward to load ALL the
* allocation headers. A total of (say) 100 allocation headers might provide
- * upto 4000 allocations each -> 400 000 objects, while 1000 headers -> 4m
+ * up to 4000 allocations each -> 400 000 objects, while 1000 headers -> 4m
* objects and 2000 -> 8m objects.
- *
+ * <p>
* The allocators are split into a set of FixedAllocators and then
* BlobAllocation. The FixedAllocators will allocate from 128 to 32K objects,
* with a minimum block allocation of 64K, and a minimum bit number per block of
* 32.
- *
+ * <p>
* Where possible lists and roving pointers will be used to minimise searching
* of the potentially large structures.
- *
+ * <p>
* Since the memory is allocated on (at least) a 128 byte boundary, there is
* some leeway on storing the address. Added to the address is the shift
* required to make to the "standard" 128 byte block, e.g. blocksize = 128 <<
* (addr % 8)
- *
+ * <p>
* NB Useful method on RandomAccessFile.setLength(newLength)
- *
+ * <p>
* When session data is preserved two things must happen - the allocators must
* not reallocate data that has been freed in this session, or more clearly can
* only free data that has been allocated in this session. That should be it.
- *
+ * <p>
* The ALLOC_SIZES table is the fibonacci sequence. We multiply by 64 bytes to
* get actual allocation block sizes. We then allocate bits based on 8K
* allocation rounding and 32 bits at a time allocation. Note that 4181 * 64 =
* 267,584 and 256K is 262,144
- *
+ * <p>
* All data is checksummed, both allocated/saved data and the allocation blocks.
- *
+ * <p>
* BLOB allocation is not handled using chained data buffers but with a blob
* header record. This is indicated with a BlobAllocator that provides indexed
* offsets to the header record (the address encodes the BlobAllocator and the
* offset to the address). The header record stores the number of component
* allocations and the address of each.
- *
+ * <p>
* This approach makes for much more efficient freeing/re-allocation of Blob
* storage, in particular avoiding the need to read in the component blocks
* to determine chained blocks for freeing. This is particularly important
* for larger stores where a disk cache could be flushed through simply freeing
* BLOB allocations.
- *
+ * <h2>
* Deferred Free List
- *
- * The prevous implentation has been amended to associate a single set of
+ * </h2>
+ * <p>
+ * The previous implementation has been amended to associate a single set of
* deferredFree blocks with each CommitRecord. The CommitRecordIndex will
* then provide access to the CommitRecords to support the deferred freeing
* of allocations based on age/earliestTxReleaseTime.
+ * <p>
+ * The last release time processed is held with the MetaAllocation data
*
- * The last release time processed is held with the MetaAllocation data
+ * @author Martyn Cutcher
*/
public class RWStore implements IStore {
- protected static final Logger log = Logger.getLogger(RWStore.class);
+ private static final transient Logger log = Logger.getLogger(RWStore.class);
+
/**
* The sizes of the slots managed by a {@link FixedAllocator} are 64 times
* the values in this array.
*
* @todo good to have 4k and 8k boundaries for better efficiency on SSD.
- * NB A 1K boundry is % 16, so 4K % 64
+ * NB A 1K boundary is % 16, so 4K % 64
* - can still use fibonacci base, but from 4K start
*
* @todo This array should be configurable and must be written into the
@@ -240,8 +241,8 @@
// protected int m_transactionCount;
// private boolean m_committing;
- boolean m_preserveSession = true;
- private boolean m_readOnly;
+ private boolean m_preserveSession = true;
+// private boolean m_readOnly;
/**
* lists of total alloc blocks.
@@ -259,13 +260,13 @@
private final ArrayList<BlobAllocator> m_freeBlobs;
// lists of blocks requiring commitment
- private final ArrayList m_commitList;
+ private final ArrayList<Allocator> m_commitList;
private WriteBlock m_writes;
private final Quorum<?,?> m_quorum;
- private RWWriteCacheService m_writeCache;
+ private final RWWriteCacheService m_writeCache;
- int[] m_allocSizes;
+ private int[] m_allocSizes;
/**
* This lock is used to exclude readers when the extent of the backing file
@@ -310,28 +311,28 @@
/**
* The deferredFreeList is simply an array of releaseTime,freeListAddrs
* stored at commit.
- *
+ * <p>
* Note that when the deferredFreeList is saved, ONLY thefreeListAddrs
* are stored, NOT the releaseTime. This is because on any open of
* the store, all deferredFrees can be released immediately. This
* mechanism may be changed in the future to enable explicit history
* retention, but if so a different header structure would be used since
* it would not be appropriate to retain a simple header linked to
- * thousands if not milions of commit points.
- *
- * If the current txn list exceeds the MAX_DEFERRED_FREE then it is
- * incrementally saved and a new list begun. The master list itself
- * serves as a BLOB header when there is more than a single entry with
- * the same txReleaseTime.
+ * thousands if not millions of commit points.
*/
- private static final int MAX_DEFERRED_FREE = 4094; // fits in 16k block
- volatile long m_lastDeferredReleaseTime = 0L;// = 23; // zero is invalid time
- final ArrayList<Integer> m_currentTxnFreeList = new ArrayList<Integer>();
- final PSOutputStream m_deferredFreeOut;
+// *
+// * If the current txn list exceeds the MAX_DEFERRED_FREE then it is
+// * incrementally saved and a new list begun. The master list itself
+// * serves as a BLOB header when there is more than a single entry with
+// * the same txReleaseTime.
+// private static final int MAX_DEFERRED_FREE = 4094; // fits in 16k block
+ private volatile long m_lastDeferredReleaseTime = 0L;
+// private final ArrayList<Integer> m_currentTxnFreeList = new ArrayList<Integer>();
+ private final PSOutputStream m_deferredFreeOut;
private ReopenFileChannel m_reopener = null;
- BufferedWrite m_bufferedWrite;
+ private BufferedWrite m_bufferedWrite;
class WriteCacheImpl extends WriteCache.FileChannelScatteredWriteCache {
public WriteCacheImpl(final ByteBuffer buf,
@@ -350,14 +351,16 @@
final long firstOffsetignored,
final Map<Long, RecordMetadata> recordMap,
final long nanos) throws InterruptedException, IOException {
- Lock readLock = m_extensionLock.readLock();
- readLock.lock();
- try {
- return super.writeOnChannel(data, firstOffsetignored, recordMap, nanos);
- } finally {
- readLock.unlock();
- }
-
+
+ final Lock readLock = m_extensionLock.readLock();
+ readLock.lock();
+ try {
+ return super.writeOnChannel(data, firstOffsetignored,
+ recordMap, nanos);
+ } finally {
+ readLock.unlock();
+ }
+
}
// Added to enable debug of rare problem
@@ -368,28 +371,28 @@
};
- private String m_filename;
+// private String m_filename;
private final FileMetadataView m_fmv;
private IRootBlockView m_rb;
- volatile private long m_commitCounter;
+// volatile private long m_commitCounter;
volatile private int m_metaBitsAddr;
- /**
- * The ALLOC_SIZES must be initialised from either the file
- * or the properties associaed with the fileMetadataView
- *
- * @param fileMetadataView
- * @param readOnly
- * @param quorum
- * @throws InterruptedException
- */
+ /**
+ * The ALLOC_SIZES must be initialized from either the file or the
+ * properties associated with the fileMetadataView
+ *
+ * @param fileMetadataView
+ * @param readOnly
+ * @param quorum
+ * @throws InterruptedException
+ */
- public RWStore(final FileMetadataView fileMetadataView, final boolean readOnly,
- final Quorum<?,?> quorum) {
+ public RWStore(final FileMetadataView fileMetadataView,
+ final boolean readOnly, final Quorum<?, ?> quorum) {
m_metaBitsSize = cDefaultMetaBitsSize;
@@ -407,12 +410,13 @@
m_rb = m_fmv.getRootBlock();
- m_filename = m_fd.getAbsolutePath();
+// m_filename = m_fd.getAbsolutePath();
- m_commitList = new ArrayList();
+ m_commitList = new ArrayList<Allocator>();
+
m_allocs = new ArrayList<Allocator>();
- m_freeBlobs = new ArrayList();
+ m_freeBlobs = new ArrayList<BlobAllocator>();
try {
m_reopener = new ReopenFileChannel(m_fd, m_raf, "rw");
@@ -426,13 +430,14 @@
m_bufferedWrite = null;
}
- int buffers = m_fmv.getFileMetadata().writeCacheBufferCount;
- log.warn("RWStore using writeCacheService with buffers: " + buffers);
+ final int buffers = m_fmv.getFileMetadata().writeCacheBufferCount;
+ if(log.isInfoEnabled())
+ log.info("RWStore using writeCacheService with buffers: " + buffers);
+
try {
- m_writeCache = new RWWriteCacheService(
- buffers, m_raf
- .length(), m_reopener, m_quorum) {
+ m_writeCache = new RWWriteCacheService(buffers, m_raf.length(),
+ m_reopener, m_quorum) {
public WriteCache newWriteCache(final ByteBuffer buf,
final boolean useChecksum,
@@ -541,24 +546,23 @@
final long metaAddr = rbv.getMetaStartAddr();
final long rawMetaBitsAddr = rbv.getMetaBitsAddr();
if (metaAddr == 0 || rawMetaBitsAddr == 0) {
- log.warn("No meta allocation data included in root block for RWStore"); // possible
- // when
- // rolling
- // back
- // to
- // empty
- // file
+ /*
+ * possible when rolling back to empty file.
+ */
+ log.warn("No meta allocation data included in root block for RWStore");
}
if (log.isTraceEnabled()) {
- int commitRecordAddr = (int) (rbv.getCommitRecordAddr() >> 32);
- log.trace("CommitRecord " + rbv.getCommitRecordAddr() + " at physical address: " + physicalAddress(commitRecordAddr));
- }
+ final int commitRecordAddr = (int) (rbv.getCommitRecordAddr() >> 32);
+ log.trace("CommitRecord " + rbv.getCommitRecordAddr()
+ + " at physical address: "
+ + physicalAddress(commitRecordAddr));
+ }
final long commitCounter = rbv.getCommitCounter();
- final int metaStartAddr = (int) -(metaAddr >> 32); // void
- final int fileSize = (int) -(metaAddr & 0xFFFFFFFF);
+// final int metaStartAddr = (int) -(metaAddr >> 32); // void
+// final int fileSize = (int) -(metaAddr & 0xFFFFFFFF);
if (log.isTraceEnabled())
log.trace("m_allocation: " + nxtalloc + ", m_metaBitsAddr: "
@@ -592,7 +596,7 @@
// m_rb = m_fmv.getRootBlock();
assert(m_rb != null);
- m_commitCounter = m_rb.getCommitCounter();
+// m_commitCounter = m_rb.getCommitCounter();
final long nxtOffset = m_rb.getNextOffset();
m_nextAllocation = -(int) (nxtOffset >> 32);
@@ -607,10 +611,12 @@
m_fileSize = (int) -(metaAddr & 0xFFFFFFFF);
long rawmbaddr = m_rb.getMetaBitsAddr();
- int metaBitsStore = (int) (rawmbaddr & 0xFFFF); // take bottom 16 bits (
- // even 1K of metabits
- // is more than
- // sufficient)
+
+ /*
+ * Take bottom 16 bits (even 1K of metabits is more than sufficient)
+ */
+ final int metaBitsStore = (int) (rawmbaddr & 0xFFFF);
+
if (metaBitsStore > 0) {
rawmbaddr >>= 16;
@@ -623,7 +629,7 @@
m_lastDeferredReleaseTime = strBuf.readLong();
- int allocBlocks = strBuf.readInt();
+ final int allocBlocks = strBuf.readInt();
m_allocSizes = new int[allocBlocks];
for (int i = 0; i < allocBlocks; i++) {
m_allocSizes[i] = strBuf.readInt();
@@ -643,7 +649,7 @@
m_freeFixed = new ArrayList[numFixed];
for (int i = 0; i < numFixed; i++) {
- m_freeFixed[i] = new ArrayList();
+ m_freeFixed[i] = new ArrayList<FixedAllocator>();
}
checkCoreAllocations();
@@ -669,58 +675,58 @@
+ ", " + m_metaBitsAddr);
}
- /*
- * Called when store is opened to make sure any deferred frees are
- * cleared.
- *
- * Stored persistently is only the list of addresses of blocks to be freed,
- * the knowledge of the txn release time does not need to be held persistently,
- * this is only relevant for transient state while the RWStore is open.
- *
- * The deferredCount is the number of entries - integer address and integer
- * count at each address
- */
- private void clearOutstandingDeferrels(final int deferredAddr, final int deferredCount) {
- if (deferredAddr != 0) {
- assert deferredCount != 0;
- final int sze = deferredCount * 8 + 4; // include space for checksum
-
- if (log.isDebugEnabled())
- log.debug("Clearing Outstanding Deferrals: " + deferredCount);
-
- byte[] buf = new byte[sze];
- getData(deferredAddr, buf);
-
- final byte[] blockBuf = new byte[8 * 1024]; // maximum size required
-
- ByteBuffer in = ByteBuffer.wrap(buf);
- for (int i = 0; i < deferredCount; i++) {
- int blockAddr = in.getInt();
- int addrCount = in.getInt();
-
- // now read in this block and free all addresses referenced
- getData(blockAddr, blockBuf, 0, addrCount*4 + 4);
- ByteBuffer inblock = ByteBuffer.wrap(blockBuf);
- for (int b = 0; b < addrCount; b++) {
- final int defAddr = inblock.getInt();
- Allocator alloc = getBlock(defAddr);
- if (alloc instanceof BlobAllocator) {
- b++;
- assert b < addrCount;
- alloc.free(defAddr, inblock.getInt());
- } else {
- alloc.free(defAddr, 0); // size ignored for FreeAllocators
- }
- }
- // once read then free the block allocation
- free(blockAddr, 0);
- }
-
- // lastly free the deferredAddr
- free(deferredAddr, 0);
- }
-
- }
+// /*
+// * Called when store is opened to make sure any deferred frees are
+// * cleared.
+// *
+// * Stored persistently is only the list of addresses of blocks to be freed,
+// * the knowledge of the txn release time does not need to be held persistently,
+// * this is only relevant for transient state while the RWStore is open.
+// *
+// * The deferredCount is the number of entries - integer address and integer
+// * count at each address
+// */
+// private void clearOutstandingDeferrels(final int deferredAddr, final int deferredCount) {
+// if (deferredAddr != 0) {
+// assert deferredCount != 0;
+// final int sze = deferredCount * 8 + 4; // include space for checksum
+//
+// if (log.isDebugEnabled())
+// log.debug("Clearing Outstanding Deferrals: " + deferredCount);
+//
+// byte[] buf = new byte[sze];
+// getData(deferredAddr, buf);
+//
+// final byte[] blockBuf = new byte[8 * 1024]; // maximum size required
+//
+// ByteBuffer in = ByteBuffer.wrap(buf);
+// for (int i = 0; i < deferredCount; i++) {
+// int blockAddr = in.getInt();
+// int addrCount = in.getInt();
+//
+// // now read in this block and free all addresses referenced
+// getData(blockAddr, blockBuf, 0, addrCount*4 + 4);
+// ByteBuffer inblock = ByteBuffer.wrap(blockBuf);
+// for (int b = 0; b < addrCount; b++) {
+// final int defAddr = inblock.getInt();
+// Allocator alloc = getBlock(defAddr);
+// if (alloc instanceof BlobAllocator) {
+// b++;
+// assert b < addrCount;
+// alloc.free(defAddr, inblock.getInt());
+// } else {
+// alloc.free(defAddr, 0); // size ignored for FreeAllocators
+// }
+// }
+// // once read then free the block allocation
+// free(blockAddr, 0);
+// }
+//
+// // lastly free the deferredAddr
+// free(deferredAddr, 0);
+// }
+//
+// }
/*********************************************************************
* make sure resource is closed!
@@ -733,8 +739,9 @@
assert m_allocs.size() == 0;
- System.out.println("readAllocationBlocks, m_metaBits.length: "
- + m_metaBits.length);
+ if (log.isInfoEnabled())
+ log.info("readAllocationBlocks, m_metaBits.length: "
+ + m_metaBits.length);
/**
* Allocators are sorted in StartAddress order (which MUST be the order
@@ -744,15 +751,15 @@
* address (must be two version of same Allocator).
*
* Meta-Allocations stored as {int address; int[8] bits}, so each block
- * holds 8*32=256 allocation slots of 1K totalling 256K.
+ * holds 8*32=256 allocation slots of 1K totaling 256K.
*/
for (int b = 0; b < m_metaBits.length; b += 9) {
- long blockStart = convertAddr(m_metaBits[b]);
- int startBit = (b * 32) + 32;
- int endBit = startBit + (8*32);
+ final long blockStart = convertAddr(m_metaBits[b]);
+ final int startBit = (b * 32) + 32;
+ final int endBit = startBit + (8*32);
for (int i = startBit; i < endBit; i++) {
if (tstBit(m_metaBits, i)) {
- long addr = blockStart + ((i-startBit) * ALLOC_BLOCK_SIZE);
+ final long addr = blockStart + ((i-startBit) * ALLOC_BLOCK_SIZE);
final byte buf[] = new byte[ALLOC_BLOCK_SIZE];
@@ -761,8 +768,8 @@
final DataInputStream strBuf = new DataInputStream(new ByteArrayInputStream(buf));
final int allocSize = strBuf.readInt(); // if Blob < 0
- Allocator allocator = null;
- ArrayList freeList = null;
+ final Allocator allocator;
+ final ArrayList<? extends Allocator> freeList;
if (allocSize > 0) {
int index = 0;
int fixedSize = m_minFixedAlloc;
@@ -811,16 +818,20 @@
* ContextAllocation is released, its allocators will be added to the
* global free lists.
*
- * @param block - the index of the Fixed size alloction
+ * @param block - the index of the Fixed size allocation
* @return the FixedAllocator
*/
- FixedAllocator establishFreeFixedAllocator(int block) {
- ArrayList<FixedAllocator> list = m_freeFixed[block];
+ private FixedAllocator establishFreeFixedAllocator(final int block) {
+
+ final ArrayList<FixedAllocator> list = m_freeFixed[block];
if (list.size() == 0) {
- final int allocSize = 64 * m_allocSizes[block];
+
+ final int allocSize = 64 * m_allocSizes[block];
- FixedAllocator allocator = new FixedAllocator(this, allocSize, m_writeCache);
+ final FixedAllocator allocator = new FixedAllocator(this,
+ allocSize, m_writeCache);
+
allocator.setIndex(m_allocs.size());
m_allocs.add(allocator);
@@ -855,7 +866,7 @@
// }
public long getMaxFileSize() {
- long maxSize = m_maxFileSize;
+ final long maxSize = m_maxFileSize;
return maxSize << 8;
}
@@ -914,12 +925,14 @@
* If it is a BlobAllocation, then the BlobAllocation address points to the address of the BlobHeader
* record.
*/
- public void getData(long addr, byte buf[]) {
+ public void getData(final long addr, final byte buf[]) {
getData(addr, buf, 0, buf.length);
}
- public void getData(long addr, byte buf[], int offset, int length) {
- if (addr == 0) {
+ public void getData(final long addr, final byte buf[], final int offset,
+ final int length) {
+
+ if (addr == 0) {
return;
}
@@ -931,19 +944,24 @@
// length includes space for the checksum
if (length > m_maxFixedAlloc) {
try {
- int alloc = m_maxFixedAlloc-4;
- int nblocks = (alloc - 1 + (length-4))/alloc;
- if (nblocks < 0) throw new IllegalStateException("Allocation error, m_maxFixedAlloc: "+ m_maxFixedAlloc);
-
- byte[] hdrbuf = new byte[4 * (nblocks + 1) + 4]; // plus 4 bytes for checksum
- BlobAllocator ba = (BlobAllocator) getBlock((int) addr);
+ final int alloc = m_maxFixedAlloc-4;
+ final int nblocks = (alloc - 1 + (length-4))/alloc;
+ if (nblocks < 0)
+ throw new IllegalStateException(
+ "Allocation error, m_maxFixedAlloc: "
+ + m_maxFixedAlloc);
+
+ final byte[] hdrbuf = new byte[4 * (nblocks + 1) + 4]; // plus 4 bytes for checksum
+ final BlobAllocator ba = (BlobAllocator) getBlock((int) addr);
getData(ba.getBlobHdrAddress(getOffset((int) addr)), hdrbuf); // read in header
- DataInputStream hdrstr = new DataInputStream(new ByteArrayInputStream(hdrbuf));
- int rhdrs = hdrstr.readInt();
- if (rhdrs != nblocks) {
- throw new IllegalStateException("Incompatible BLOB header record, expected: " + nblocks + ", got: " + rhdrs);
- }
- int[] blobHdr = new int[nblocks];
+ final DataInputStream hdrstr = new DataInputStream(new ByteArrayInputStream(hdrbuf));
+ final int rhdrs = hdrstr.readInt();
+ if (rhdrs != nblocks) {
+ throw new IllegalStateException(
+ "Incompatible BLOB header record, expected: "
+ + nblocks + ", got: " + rhdrs);
+ }
+ final int[] blobHdr = new int[nblocks];
for (int i = 0; i < nblocks; i++) {
blobHdr[i] = hdrstr.readInt();
}
@@ -968,7 +986,7 @@
}
try {
- long paddr = physicalAddress((int) addr);
+ final long paddr = physicalAddress((int) addr);
if (paddr == 0) {
assertAllocators();
@@ -984,19 +1002,25 @@
* value, so the cached data is 4 bytes less than the
* buffer size.
*/
- ByteBuffer bbuf = null;
+ final ByteBuffer bbuf;
try {
bbuf = m_writeCache.read(paddr);
} catch (Throwable t) {
- throw new IllegalStateException("Error reading from WriteCache addr: " + paddr + " length: " + (length-4) + ", writeCacheDebug: " + m_writeCache.addrDebugInfo(paddr), t);
+ throw new IllegalStateException(
+ "Error reading from WriteCache addr: " + paddr
+ + " length: " + (length - 4)
+ + ", writeCacheDebug: "
+ + m_writeCache.addrDebugInfo(paddr), t);
}
if (bbuf != null) {
- byte[] in = bbuf.array(); // reads in with checksum - no need to check if in cache
+ final byte[] in = bbuf.array(); // reads in with checksum - no need to check if in cache
if (in.length != length-4) {
assertAllocators();
-
- throw new IllegalStateException("Incompatible buffer size for addr: " + paddr + ", " + in.length
- + " != " + (length-4) + " writeCacheDebug: " + m_writeCache.addrDebugInfo(paddr));
+ throw new IllegalStateException(
+ "Incompatible buffer size for addr: " + paddr
+ + ", " + in.length + " != "
+ + (length - 4) + " writeCacheDebug: "
+ + m_writeCache.addrDebugInfo(paddr));
}
for (int i = 0; i < length-4; i++) {
buf[offset+i] = in[i];
@@ -1004,21 +1028,23 @@
m_cacheReads++;
} else {
// If checksum is required then the buffer should be sized to include checksum in final 4 bytes
- ByteBuffer bb = ByteBuffer.wrap(buf, offset, length);
+ final ByteBuffer bb = ByteBuffer.wrap(buf, offset, length);
FileChannelUtility.readAll(m_reopener, bb, paddr);
- int chk = ChecksumUtility.getCHK().checksum(buf, offset, length-4); // read checksum
- int tstchk = bb.getInt(offset + length-4);
+ final int chk = ChecksumUtility.getCHK().checksum(buf, offset, length-4); // read checksum
+ final int tstchk = bb.getInt(offset + length-4);
if (chk != tstchk) {
assertAllocators();
- String cacheDebugInfo = m_writeCache.addrDebugInfo(paddr);
+ final String cacheDebugInfo = m_writeCache.addrDebugInfo(paddr);
log.warn("Invalid data checksum for addr: " + paddr
+ ", chk: " + chk + ", tstchk: " + tstchk + ", length: " + length
+ ", first bytes: " + toHexString(buf, 32) + ", successful reads: " + m_diskReads
+ ", at last extend: " + m_readsAtExtend + ", cacheReads: " + m_cacheReads
+ ", writeCacheDebug: " + cacheDebugInfo);
- throw new IllegalStateException("Invalid data checksum from address: " + paddr + ", size: " + (length-4));
+ throw new IllegalStateException(
+ "Invalid data checksum from address: " + paddr
+ + ", size: " + (length - 4));
}
m_diskReads++;
@@ -1042,7 +1068,7 @@
}
}
- static final char[] HEX_CHAR_TABLE = {
+ static private final char[] HEX_CHAR_TABLE = {
'0', '1','2','3',
'4','5','6','7',
'8','9','a','b',
@@ -1050,11 +1076,11 @@
};
// utility to display byte array of maximum i bytes as hexString
- private String toHexString(byte[] buf, int n) {
+ static private String toHexString(final byte[] buf, int n) {
n = n < buf.length ? n : buf.length;
- StringBuffer out = new StringBuffer();
+ final StringBuffer out = new StringBuffer();
for (int i = 0; i < n; i++) {
- int v = buf[i] & 0xFF;
+ final int v = buf[i] & 0xFF;
out.append(HEX_CHAR_TABLE[v >>> 4]);
out.append(HEX_CHAR_TABLE[v &0xF]);
}
@@ -1094,7 +1120,7 @@
* this supports the core functionality of a WormStore, other stores should
* return zero, indicating no previous versions available
**/
- public long getPreviousAddress(long laddr) {
+ public long getPreviousAddress(final long laddr) {
return 0;
}
@@ -1135,7 +1161,7 @@
* is NOT owned, BUT there are active AllocationContexts, in this
* situation, the free must ALWAYS be deferred.
*/
- boolean alwaysDefer = context == null && m_contexts.size() > 0;
+ final boolean alwaysDefer = context == null && m_contexts.size() > 0;
if (alwaysDefer)
if (log.isDebugEnabled())
log.debug("Should defer " + physicalAddress(addr));
@@ -1150,7 +1176,7 @@
}
- private long immediateFreeCount = 0;
+// private long immediateFreeCount = 0;
private void immediateFree(final int addr, final int sze) {
switch (addr) {
@@ -1215,10 +1241,10 @@
* by the finer granularity of the AllocBlocks within a FixedAllocator.
*/
- private volatile long m_maxAllocation = 0;
+// private volatile long m_maxAllocation = 0;
private volatile long m_spareAllocation = 0;
- public int alloc(final int size, IAllocationContext context) {
+ public int alloc(final int size, final IAllocationContext context) {
if (size > m_maxFixedAlloc) {
throw new IllegalArgumentException("Allocation size to big: " + size);
}
@@ -1226,8 +1252,7 @@
m_allocationLock.lock();
try {
try {
- ArrayList list;
- Allocator allocator = null;
+ final Allocator allocator;
final int i = fixedAllocatorIndex(size);
if (context != null) {
allocator = establishContextAllocation(context).getFreeFixed(i);
@@ -1235,7 +1260,7 @@
final int block = 64 * m_allocSizes[i];
m_spareAllocation += (block - size); // Isn't adjusted by frees!
- list = m_freeFixed[i];
+ final ArrayList<FixedAllocator> list = m_freeFixed[i];
if (list.size() == 0) {
allocator = new FixedAllocator(this, block, m_writeCache);
@@ -1250,9 +1275,9 @@
// Verify free list only has allocators with free bits
if (log.isDebugEnabled()){
int tsti = 0;
- Iterator<Allocator> allocs = list.iterator();
+ final Iterator<FixedAllocator> allocs = list.iterator();
while (allocs.hasNext()) {
- Allocator tstAlloc = allocs.next();
+ final Allocator tstAlloc = allocs.next();
if (!tstAlloc.hasFree()) {
throw new IllegalStateException("Free list contains full allocator, " + tsti + " of " + list.size());
}
@@ -1272,10 +1297,11 @@
m_recentAlloc = true;
- long pa = physicalAddress(addr);
- if (pa == 0L) {
- throw new IllegalStateException("No physical address found for " + addr);
- }
+ final long pa = physicalAddress(addr);
+ if (pa == 0L) {
+ throw new IllegalStateException(
+ "No physical address found for " + addr);
+ }
m_allocations++;
m_nativeAllocBytes += size;
@@ -1291,7 +1317,7 @@
}
}
- int fixedAllocatorIndex(int size) {
+ int fixedAllocatorIndex(final int size) {
int i = 0;
int cmp = m_minFixedAlloc;
@@ -1530,12 +1556,24 @@
* simply reset the metaBitsAddr
* @throws IOException
*/
- protected void writeFileSpec() throws IOException {
- m_rb = m_fmv.newRootBlockView(!m_rb.isRootBlock0(), m_rb.getOffsetBits(), getNextOffset(), m_rb
- .getFirstCommitTime(), m_rb.getLastCommitTime(), m_rb.getCommitCounter(), m_rb.getCommitRecordAddr(),
- m_rb.getCommitRecordIndexAddr(), getMetaStartAddr(), getMetaBitsAddr(), m_rb.getLastCommitTime());
-
- m_fmv.getFileMetadata().writeRootBlock(m_rb, ForceEnum.Force);
+ protected void writeFileSpec() throws IOException {
+
+ m_rb = m_fmv.newRootBlockView(//
+ !m_rb.isRootBlock0(), //
+ m_rb.getOffsetBits(), //
+ getNextOffset(), //
+ m_rb.getFirstCommitTime(),//
+ m_rb.getLastCommitTime(), //
+ m_rb.getCommitCounter(), //
+ m_rb.getCommitRecordAddr(),//
+ m_rb.getCommitRecordIndexAddr(), //
+ getMetaStartAddr(),//
+ getMetaBitsAddr(), //
+ m_rb.getLastCommitTime()//
+ );
+
+ m_fmv.getFileMetadata().writeRootBlock(m_rb, ForceEnum.Force);
+
}
// float m_vers = 0.0f;
@@ -1582,8 +1620,8 @@
checkDeferredFrees(true, journal); // free now if possible
// Allocate storage for metaBits
- long oldMetaBits = m_metaBitsAddr;
- int oldMetaBitsSize = (m_metaBits.length + m_allocSizes.length + 1) * 4;
+ final long oldMetaBits = m_metaBitsAddr;
+ final int oldMetaBitsSize = (m_metaBits.length + m_allocSizes.length + 1) * 4;
m_metaBitsAddr = alloc(getRequiredMetaBitsStorage(), null);
// DEBUG SANITY CHECK!
@@ -1596,18 +1634,18 @@
immediateFree((int) oldMetaBits, oldMetaBitsSize);
// save allocation headers
- Iterator iter = m_commitList.iterator();
+ final Iterator<Allocator> iter = m_commitList.iterator();
while (iter.hasNext()) {
- final Allocator allocator = (Allocator) iter.next();
+ final Allocator allocator = iter.next();
final int old = allocator.getDiskAddr();
metaFree(old);
final int naddr = metaAlloc();
allocator.setDiskAddr(naddr);
- if (log.isTraceEnabled())
- log.trace("Update allocator " + allocator.getIndex() + ", old addr: " + old + ", new addr: "
- + naddr);
+ if (log.isTraceEnabled())
+ log.trace("Update allocator " + allocator.getIndex()
+ + ", old addr: " + old + ", new addr: " + naddr);
try {
m_writeCache.write(metaBit2Addr(naddr), ByteBuffer.wrap(allocator.write()), 0, false); // do not use checksum
@@ -1647,16 +1685,21 @@
checkCoreAllocations();
- if (log.isTraceEnabled())
- log.trace("commitChanges for: " + m_nextAllocation + ", " + m_metaBitsAddr + ", active contexts: " + m_contexts.size());
- }
+ if (log.isTraceEnabled())
+ log.trace("commitChanges for: " + m_nextAllocation + ", "
+ + m_metaBitsAddr + ", active contexts: "
+ + m_contexts.size());
+ }
- /**
- * Called prior to commit, so check whether storage can be freed and then
- * whether the deferredheader needs to be saved.
- * <p>
- * Note: The caller MUST be holding the {@link #m_allocationLock}.
- */
+ /**
+ * Called prior to commit, so check whether storage can be freed and then
+ * whether the deferredheader needs to be saved.
+ * <p>
+ * Note: The caller MUST be holding the {@link #m_allocationLock}.
+ * <p>
+ * Note: This method is package private in order to expose it to the unit
+ * tests....
[truncated message content] |
