[hornetq-commits] JBoss hornetq SVN: r10944 - branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl.
do-not-reply at jboss.org
do-not-reply at jboss.org
Thu Jul 7 06:22:26 EDT 2011
Author: borges
Date: 2011-07-07 06:22:25 -0400 (Thu, 07 Jul 2011)
New Revision: 10944
Modified:
branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
Log:
Use enum for state instead of numeric codes, reduce code duplication.
Modified: branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-07-07 10:21:52 UTC (rev 10943)
+++ branches/HORNETQ-720_Replication/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2011-07-07 10:22:25 UTC (rev 10944)
@@ -65,26 +65,25 @@
import org.hornetq.utils.DataConstants;
/**
- *
+ *
* <p>A circular log implementation.</p
- *
+ *
* <p>Look at {@link JournalImpl#load(LoaderCallback)} for the file layout
- *
+ *
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
- *
+ *
* @author <a href="mailto:clebert.suconic at jboss.com">Clebert Suconic</a>
*
*/
public class JournalImpl implements TestableJournal, JournalRecordProvider
{
+ private enum JournalState
+ {
+ STOPPED, STARTED, LOADED;
+ }
// Constants -----------------------------------------------------
- private static final int STATE_STOPPED = 0;
- private static final int STATE_STARTED = 1;
-
- private static final int STATE_LOADED = 2;
-
public static final int FORMAT_VERSION = 2;
private static final int COMPATIBLE_VERSIONS[] = new int[] { 1 };
@@ -210,7 +209,7 @@
private volatile JournalFile currentFile;
- private volatile int state;
+ private volatile JournalState state = JournalState.STOPPED;
private final Reclaimer reclaimer = new Reclaimer();
@@ -280,7 +279,7 @@
}
else
{
- this.compactPercentage = (float)compactPercentage / 100f;
+ this.compactPercentage = compactPercentage / 100f;
}
this.compactMinFiles = compactMinFiles;
@@ -378,7 +377,7 @@
return compactor;
}
- /** this method is used internally only however tools may use it to maintenance.
+ /** this method is used internally only however tools may use it to maintenance.
* It won't be part of the interface as the tools should be specific to the implementation */
public List<JournalFile> orderFiles() throws Exception
{
@@ -424,7 +423,7 @@
{
final int filesize = (int)file.getFile().size();
- wholeFileBuffer = fileFactory.newBuffer((int)filesize);
+ wholeFileBuffer = fileFactory.newBuffer(filesize);
final int journalFileSize = file.getFile().read(wholeFileBuffer);
@@ -816,10 +815,7 @@
final boolean sync,
final IOCompletion callback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -891,10 +887,7 @@
final boolean sync,
final IOCompletion callback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -967,10 +960,7 @@
public void appendDeleteRecord(final long id, final boolean sync, final IOCompletion callback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
try
@@ -1046,10 +1036,7 @@
final byte recordType,
final EncodingSupport record) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1088,6 +1075,14 @@
}
}
+ private void checkJournalIsLoaded()
+ {
+ if (state != JournalState.LOADED)
+ {
+ throw new IllegalStateException("Journal must be in state=" + JournalState.LOADED + ", was [" + state + "]");
+ }
+ }
+
public void appendUpdateRecordTransactional(final long txID,
final long id,
final byte recordType,
@@ -1101,10 +1096,7 @@
final byte recordType,
final EncodingSupport record) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1150,10 +1142,7 @@
public void appendDeleteRecordTransactional(final long txID, final long id, final EncodingSupport record) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1223,15 +1212,15 @@
}
}
- /**
- *
- * <p>If the system crashed after a prepare was called, it should store information that is required to bring the transaction
+ /**
+ *
+ * <p>If the system crashed after a prepare was called, it should store information that is required to bring the transaction
* back to a state it could be committed. </p>
- *
+ *
* <p> transactionData allows you to store any other supporting user-data related to the transaction</p>
- *
+ *
* <p> This method also uses the same logic applied on {@link JournalImpl#appendCommitRecord(long, boolean)}
- *
+ *
* @param txID
* @param transactionData extra user data for the prepare
* @throws Exception
@@ -1242,10 +1231,7 @@
final IOCompletion callback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1310,30 +1296,27 @@
appendCommitRecord(txID, sync, callback, true);
}
-
+
/**
* <p>A transaction record (Commit or Prepare), will hold the number of elements the transaction has on each file.</p>
- * <p>For example, a transaction was spread along 3 journal files with 10 pendingTransactions on each file.
+ * <p>For example, a transaction was spread along 3 journal files with 10 pendingTransactions on each file.
* (What could happen if there are too many pendingTransactions, or if an user event delayed pendingTransactions to come in time to a single file).</p>
* <p>The element-summary will then have</p>
* <p>FileID1, 10</p>
* <p>FileID2, 10</p>
* <p>FileID3, 10</p>
- *
+ *
* <br>
* <p> During the load, the transaction needs to have 30 pendingTransactions spread across the files as originally written.</p>
* <p> If for any reason there are missing pendingTransactions, that means the transaction was not completed and we should ignore the whole transaction </p>
- * <p> We can't just use a global counter as reclaiming could delete files after the transaction was successfully committed.
+ * <p> We can't just use a global counter as reclaiming could delete files after the transaction was successfully committed.
* That also means not having a whole file on journal-reload doesn't mean we have to invalidate the transaction </p>
*
*/
public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback, boolean lineUpContext) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1392,10 +1375,7 @@
public void appendRollbackRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
- {
- throw new IllegalStateException("Journal must be loaded first");
- }
+ checkJournalIsLoaded();
compactingLock.readLock().lock();
@@ -1566,16 +1546,16 @@
return info;
}
-
-
+
+
public void testCompact() throws Exception
{
final AtomicInteger errors = new AtomicInteger(0);
-
+
final CountDownLatch latch = new CountDownLatch(1);
-
+
compactorRunning.set(true);
-
+
// We can't use the executor for the compacting... or we would dead lock because of file open and creation
// operations (that will use the executor)
compactorExecutor.execute(new Runnable()
@@ -1599,14 +1579,14 @@
}
}
});
-
+
try
{
if (!latch.await(60, TimeUnit.SECONDS))
{
throw new RuntimeException("Didn't finish compact timely");
}
-
+
if (errors.get() > 0)
{
throw new RuntimeException("Error during testCompact, look at the logs");
@@ -1619,12 +1599,12 @@
}
/**
- *
+ *
* Note: This method can't be called from the main executor, as it will invoke other methods depending on it.
- *
+ *
* Note: only synchronized methods on journal are methods responsible for the life-cycle such as stop, start
* records will still come as this is being executed
- *
+ *
*/
protected synchronized void compact() throws Exception
{
@@ -1657,7 +1637,7 @@
compactingLock.writeLock().lock();
try
{
- if (state != JournalImpl.STATE_LOADED)
+ if (state != JournalState.LOADED)
{
return;
}
@@ -1837,9 +1817,9 @@
}
- /**
+ /**
* <p>Load data accordingly to the record layouts</p>
- *
+ *
* <p>Basic record layout:</p>
* <table border=1>
* <tr><td><b>Field Name</b></td><td><b>Size</b></td></tr>
@@ -1853,9 +1833,9 @@
* <tr><td>RecordBody</td><td>Byte Array (size=BodySize)</td></tr>
* <tr><td>Check Size</td><td>Integer (4 bytes)</td></tr>
* </table>
- *
+ *
* <p> The check-size is used to validate if the record is valid and complete </p>
- *
+ *
* <p>Commit/Prepare record layout:</p>
* <table border=1>
* <tr><td><b>Field Name</b></td><td><b>Size</b></td></tr>
@@ -1870,18 +1850,18 @@
* <tr><td>* NumberOfElements(n)</td><td>Integer (4 bytes)</td></tr>
* <tr><td>CheckSize</td><td>Integer (4 bytes)</td</tr>
* </table>
- *
- * <p> * FileID and NumberOfElements are the transaction summary, and they will be repeated (N)umberOfFiles times </p>
- *
+ *
+ * <p> * FileID and NumberOfElements are the transaction summary, and they will be repeated (N)umberOfFiles times </p>
+ *
* */
public JournalLoadInformation load(final LoaderCallback loadManager) throws Exception
{
return load(loadManager, true);
}
-
+
public synchronized JournalLoadInformation load(final LoaderCallback loadManager, boolean fixFailingTransactions) throws Exception
{
- if (state != JournalImpl.STATE_STARTED)
+ if (state != JournalState.STARTED)
{
throw new IllegalStateException("Journal must be in started state");
}
@@ -2197,7 +2177,7 @@
filesRepository.pushOpenedFile();
- state = JournalImpl.STATE_LOADED;
+ state = JournalState.LOADED;
for (TransactionHolder transaction : loadTransactions.values())
{
@@ -2241,7 +2221,7 @@
return new JournalLoadInformation(records.size(), maxID.longValue());
}
- /**
+ /**
* @return true if cleanup was called
*/
public boolean checkReclaimStatus() throws Exception
@@ -2311,7 +2291,7 @@
return;
}
- if (state != JournalImpl.STATE_LOADED)
+ if (state != JournalState.LOADED)
{
return;
}
@@ -2528,12 +2508,12 @@
public synchronized boolean isStarted()
{
- return state != JournalImpl.STATE_STOPPED;
+ return state != JournalState.STOPPED;
}
public synchronized void start()
{
- if (state != JournalImpl.STATE_STOPPED)
+ if (state != JournalState.STOPPED)
{
throw new IllegalStateException("Journal is not stopped");
}
@@ -2560,14 +2540,14 @@
fileFactory.start();
- state = JournalImpl.STATE_STARTED;
+ state = JournalState.STARTED;
}
public synchronized void stop() throws Exception
{
JournalImpl.trace("Stopping the journal");
- if (state == JournalImpl.STATE_STOPPED)
+ if (state == JournalState.STOPPED)
{
throw new IllegalStateException("Journal is already stopped");
}
@@ -2577,7 +2557,7 @@
try
{
- state = JournalImpl.STATE_STOPPED;
+ state = JournalState.STOPPED;
compactorExecutor.shutdown();
@@ -2722,9 +2702,9 @@
/**
* <p> Check for holes on the transaction (a commit written but with an incomplete transaction) </p>
* <p>This method will validate if the transaction (PREPARE/COMMIT) is complete as stated on the COMMIT-RECORD.</p>
- *
- * <p>Look at the javadoc on {@link JournalImpl#appendCommitRecord(long)} about how the transaction-summary is recorded</p>
- *
+ *
+ * <p>Look at the javadoc on {@link JournalImpl#appendCommitRecord(long)} about how the transaction-summary is recorded</p>
+ *
* @param journalTransaction
* @param orderedFiles
* @param recordedSummary
@@ -2895,8 +2875,8 @@
buffer.writeLong(fileID);
}
- /**
- *
+ /**
+ *
* @param completeTransaction If the appendRecord is for a prepare or commit, where we should update the number of pendingTransactions on the current file
* */
private JournalFile appendRecord(final JournalInternalRecord encoder,
@@ -2905,7 +2885,7 @@
final JournalTransaction tx,
final IOAsyncTask parameterCallback) throws Exception
{
- if (state != JournalImpl.STATE_LOADED)
+ if (state != JournalState.LOADED)
{
throw new IllegalStateException("The journal is not loaded " + state);
}
@@ -3006,7 +2986,7 @@
private void scheduleReclaim()
{
- if (state != JournalImpl.STATE_LOADED)
+ if (state != JournalState.LOADED)
{
return;
}
@@ -3239,6 +3219,7 @@
this.pages = pages;
}
+ @Override
public void run()
{
try
More information about the hornetq-commits
mailing list