6:01 p.m.
Author: clebert.suconic(a)jboss.com
Date: 2010-04-29 19:01:30 -0400 (Thu, 29 Apr 2010)
New Revision: 9186
Modified:
trunk/src/main/org/hornetq/core/journal/Journal.java
trunk/src/main/org/hornetq/core/journal/impl/AbstractJournalUpdateTask.java
trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java
trunk/src/main/org/hornetq/core/journal/impl/JournalCompactor.java
trunk/src/main/org/hornetq/core/journal/impl/JournalFile.java
trunk/src/main/org/hornetq/core/journal/impl/JournalFileImpl.java
trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java
trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/JournalImplTestUnit.java
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/ReclaimerTest.java
Log:
https://jira.jboss.org/jira/browse/HORNETQ-318 - using a long to the fileID - first
commit
Modified: trunk/src/main/org/hornetq/core/journal/Journal.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/Journal.java 2010-04-29 16:21:28 UTC (rev
9185)
+++ trunk/src/main/org/hornetq/core/journal/Journal.java 2010-04-29 23:01:30 UTC (rev
9186)
@@ -114,6 +114,8 @@
int getAlignment() throws Exception;
int getNumberOfRecords();
+
+ int getUserVersion();
void perfBlast(int pages) throws Exception;
Modified: trunk/src/main/org/hornetq/core/journal/impl/AbstractJournalUpdateTask.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AbstractJournalUpdateTask.java 2010-04-29
16:21:28 UTC (rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/AbstractJournalUpdateTask.java 2010-04-29
23:01:30 UTC (rev 9186)
@@ -55,10 +55,8 @@
protected SequentialFile sequentialFile;
- protected int fileID;
+ protected long nextOrderingID;
- protected int nextOrderingID;
-
private HornetQBuffer writingChannel;
private final Set<Long> recordsSnapshot = new ConcurrentHashSet<Long>();
@@ -72,7 +70,7 @@
protected AbstractJournalUpdateTask(final SequentialFileFactory fileFactory,
final JournalImpl journal,
final Set<Long> recordsSnapshot,
- final int nextOrderingID)
+ final long nextOrderingID)
{
super();
this.journal = journal;
@@ -95,11 +93,8 @@
{
controlFile.open(1, false);
- HornetQBuffer renameBuffer = HornetQBuffers.dynamicBuffer(1);
+ JournalImpl.initFileHeader(fileFactory, controlFile, 0, 0);
- renameBuffer.writeInt(-1);
- renameBuffer.writeInt(-1);
-
HornetQBuffer filesToRename = HornetQBuffers.dynamicBuffer(1);
// DataFiles first
@@ -155,12 +150,16 @@
new
ByteArrayEncoding(filesToRename.toByteBuffer()
.array()));
- controlRecord.setFileID(-1);
+
+ HornetQBuffer renameBuffer =
HornetQBuffers.dynamicBuffer(filesToRename.writerIndex());
+
+ controlRecord.setFileID(0);
+
controlRecord.encode(renameBuffer);
ByteBuffer writeBuffer = fileFactory.newBuffer(renameBuffer.writerIndex());
-
+
writeBuffer.put(renameBuffer.toByteBuffer().array(), 0,
renameBuffer.writerIndex());
writeBuffer.rewind();
@@ -208,16 +207,18 @@
flush();
ByteBuffer bufferWrite = fileFactory.newBuffer(journal.getFileSize());
+
writingChannel = HornetQBuffers.wrappedBuffer(bufferWrite);
currentFile = journal.getFile(false, false, false, true);
+
sequentialFile = currentFile.getFile();
sequentialFile.open(1, false);
- fileID = nextOrderingID++;
- currentFile = new JournalFileImpl(sequentialFile, fileID);
- writingChannel.writeInt(fileID);
+ currentFile = new JournalFileImpl(sequentialFile, nextOrderingID++);
+
+ JournalImpl.writeHeader(writingChannel, journal.getUserVersion(),
currentFile.getFileID());
}
protected void addToRecordsSnaptshot(final long id)
@@ -235,7 +236,7 @@
protected void writeEncoder(final JournalInternalRecord record) throws Exception
{
- record.setFileID(fileID);
+ record.setFileID(currentFile.getRecordID());
record.encode(getWritingChannel());
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java 2010-04-29 16:21:28
UTC (rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalCleaner.java 2010-04-29 23:01:30
UTC (rev 9186)
@@ -55,7 +55,7 @@
protected JournalCleaner(final SequentialFileFactory fileFactory,
final JournalImpl journal,
final Set<Long> recordsSnapshot,
- final int nextOrderingID) throws Exception
+ final long nextOrderingID) throws Exception
{
super(fileFactory, journal, recordsSnapshot, nextOrderingID);
openFile();
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalCompactor.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalCompactor.java 2010-04-29 16:21:28
UTC (rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalCompactor.java 2010-04-29 23:01:30
UTC (rev 9186)
@@ -69,7 +69,7 @@
if (controlFile.exists())
{
- JournalFile file = new JournalFileImpl(controlFile, -1);
+ JournalFile file = new JournalFileImpl(controlFile, 0);
final ArrayList<RecordInfo> records = new ArrayList<RecordInfo>();
@@ -140,7 +140,7 @@
public JournalCompactor(final SequentialFileFactory fileFactory,
final JournalImpl journal,
final Set<Long> recordsSnapshot,
- final int firstFileID)
+ final long firstFileID)
{
super(fileFactory, journal, recordsSnapshot, firstFileID);
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalFile.java 2010-04-29 16:21:28 UTC
(rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalFile.java 2010-04-29 23:01:30 UTC
(rev 9186)
@@ -56,7 +56,11 @@
long getOffset();
- int getFileID();
+ /** This is a field to identify that records on this file actually belong to the
current file.
+ * The possible implementation for this is fileID & Integer.MAX_VALUE */
+ int getRecordID();
+
+ long getFileID();
SequentialFile getFile();
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalFileImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalFileImpl.java 2010-04-29 16:21:28
UTC (rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalFileImpl.java 2010-04-29 23:01:30
UTC (rev 9186)
@@ -35,7 +35,9 @@
private final SequentialFile file;
- private final int fileID;
+ private final long fileID;
+
+ private final int recordID;
private long offset;
@@ -49,11 +51,13 @@
private final Map<JournalFile, AtomicInteger> negCounts = new
ConcurrentHashMap<JournalFile, AtomicInteger>();
- public JournalFileImpl(final SequentialFile file, final int fileID)
+ public JournalFileImpl(final SequentialFile file, final long fileID)
{
this.file = file;
this.fileID = fileID;
+
+ this.recordID = (int)(fileID & (long)Integer.MAX_VALUE);
}
public void clearCounts()
@@ -132,10 +136,15 @@
return offset;
}
- public int getFileID()
+ public long getFileID()
{
return fileID;
}
+
+ public int getRecordID()
+ {
+ return recordID;
+ }
public void setOffset(final long offset)
{
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2010-04-29 16:21:28 UTC
(rev 9185)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2010-04-29 23:01:30 UTC
(rev 9186)
@@ -69,7 +69,7 @@
/**
*
- * <p>A JournalImpl</p
+ * <p>A circular log implementation.</p
*
* <p>Look at {@link JournalImpl#load(LoaderCallback)} for the file layout
*
@@ -88,6 +88,8 @@
private static final int STATE_LOADED = 2;
+ private static final int FORMAT_VERSION = 1;
+
// Static --------------------------------------------------------
private static final Logger log = Logger.getLogger(JournalImpl.class);
@@ -109,7 +111,8 @@
public static final int MIN_FILE_SIZE = 1024;
- public static final int SIZE_HEADER = DataConstants.SIZE_INT;
+ // FileID(Long) + JournalVersion + UserVersion
+ public static final int SIZE_HEADER = DataConstants.SIZE_LONG + DataConstants.SIZE_INT
+ DataConstants.SIZE_INT;
public static final int BASIC_SIZE = DataConstants.SIZE_BYTE + DataConstants.SIZE_INT
+ DataConstants.SIZE_INT;
@@ -163,8 +166,10 @@
private volatile boolean autoReclaim = true;
- private final AtomicInteger nextFileID = new AtomicInteger(0);
+ private final AtomicLong nextFileID = new AtomicLong(0);
+ private final int userVersion;
+
private final int maxAIO;
private final int fileSize;
@@ -220,65 +225,16 @@
// Constructors --------------------------------------------------
- public void runDirectJournalBlast() throws Exception
+ public JournalImpl(final int fileSize,
+ final int minFiles,
+ final int compactMinFiles,
+ final int compactPercentage,
+ final SequentialFileFactory fileFactory,
+ final String filePrefix,
+ final String fileExtension,
+ final int maxAIO)
{
- final int numIts = 100000000;
-
- JournalImpl.log.info("*** running direct journal blast: " + numIts);
-
- final CountDownLatch latch = new CountDownLatch(numIts * 2);
-
- class MyIOAsyncTask implements IOCompletion
- {
- public void done()
- {
- latch.countDown();
- }
-
- public void onError(final int errorCode, final String errorMessage)
- {
-
- }
-
- public void storeLineUp()
- {
- }
- }
-
- final MyIOAsyncTask task = new MyIOAsyncTask();
-
- final int recordSize = 1024;
-
- final byte[] bytes = new byte[recordSize];
-
- class MyRecord implements EncodingSupport
- {
-
- public void decode(final HornetQBuffer buffer)
- {
- }
-
- public void encode(final HornetQBuffer buffer)
- {
- buffer.writeBytes(bytes);
- }
-
- public int getEncodeSize()
- {
- return recordSize;
- }
-
- }
-
- MyRecord record = new MyRecord();
-
- for (int i = 0; i < numIts; i++)
- {
- appendAddRecord(i, (byte)1, record, true, task);
- appendDeleteRecord(i, true, task);
- }
-
- latch.await();
+ this(fileSize, minFiles, compactMinFiles, compactPercentage, fileFactory,
filePrefix, fileExtension, maxAIO, 0);
}
public JournalImpl(final int fileSize,
@@ -288,7 +244,8 @@
final SequentialFileFactory fileFactory,
final String filePrefix,
final String fileExtension,
- final int maxIO)
+ final int maxAIO,
+ final int userVersion)
{
if (fileFactory == null)
{
@@ -316,7 +273,7 @@
{
throw new NullPointerException("fileExtension is null");
}
- if (maxIO <= 0)
+ if (maxAIO <= 0)
{
throw new IllegalStateException("maxAIO should aways be a positive
number");
}
@@ -347,9 +304,72 @@
this.fileExtension = fileExtension;
- maxAIO = maxIO;
+ this.maxAIO = maxAIO;
+
+ this.userVersion = userVersion;
}
+ public void runDirectJournalBlast() throws Exception
+ {
+ final int numIts = 100000000;
+
+ JournalImpl.log.info("*** running direct journal blast: " + numIts);
+
+ final CountDownLatch latch = new CountDownLatch(numIts * 2);
+
+ class MyIOAsyncTask implements IOCompletion
+ {
+ public void done()
+ {
+ latch.countDown();
+ }
+
+ public void onError(final int errorCode, final String errorMessage)
+ {
+
+ }
+
+ public void storeLineUp()
+ {
+ }
+ }
+
+ final MyIOAsyncTask task = new MyIOAsyncTask();
+
+ final int recordSize = 1024;
+
+ final byte[] bytes = new byte[recordSize];
+
+ class MyRecord implements EncodingSupport
+ {
+
+ public void decode(final HornetQBuffer buffer)
+ {
+ }
+
+ public void encode(final HornetQBuffer buffer)
+ {
+ buffer.writeBytes(bytes);
+ }
+
+ public int getEncodeSize()
+ {
+ return recordSize;
+ }
+
+ }
+
+ MyRecord record = new MyRecord();
+
+ for (int i = 0; i < numIts; i++)
+ {
+ appendAddRecord(i, (byte)1, record, true, task);
+ appendDeleteRecord(i, true, task);
+ }
+
+ latch.await();
+ }
+
public Map<Long, JournalRecord> getRecords()
{
return records;
@@ -576,7 +596,7 @@
// This record is from a previous file-usage. The file was
// reused and we need to ignore this record
- if (readFileId != file.getFileID())
+ if (readFileId != file.getRecordID())
{
// If a file has damaged pendingTransactions, we make it a dataFile, and
the
// next reclaiming will fix it
@@ -2442,6 +2462,11 @@
{
return maxAIO;
}
+
+ public int getUserVersion()
+ {
+ return userVersion;
+ }
// In some tests we need to force the journal to move to a next file
public void forceMoveNextFile() throws Exception
@@ -2631,25 +2656,17 @@
// Discard the old JournalFile and set it with a new ID
private JournalFile reinitializeFile(final JournalFile file) throws Exception
{
- int newFileID = generateFileID();
+ long newFileID = generateFileID();
SequentialFile sf = file.getFile();
sf.open(1, false);
- sf.position(0);
+ int position = initFileHeader(this.fileFactory, sf, userVersion, newFileID);
- ByteBuffer bb = fileFactory.newBuffer(JournalImpl.SIZE_HEADER);
-
- bb.putInt(newFileID);
-
- bb.rewind();
-
- sf.writeDirect(bb, true);
-
JournalFile jf = new JournalFileImpl(sf, newFileID);
- sf.position(bb.limit());
+ sf.position(position);
sf.close();
@@ -2747,16 +2764,8 @@
file.open(1, false);
- ByteBuffer bb = fileFactory.newBuffer(JournalImpl.SIZE_HEADER);
+ long fileID = readFileHeader(file);
- file.read(bb);
-
- int fileID = bb.getInt();
-
- fileFactory.releaseBuffer(bb);
-
- bb = null;
-
if (nextFileID.get() < fileID)
{
nextFileID.set(fileID);
@@ -2784,6 +2793,68 @@
return orderedFiles;
}
+ /**
+ * @param file
+ * @return
+ * @throws Exception
+ */
+ private long readFileHeader(SequentialFile file) throws Exception
+ {
+ ByteBuffer bb = fileFactory.newBuffer(JournalImpl.SIZE_HEADER);
+
+ file.read(bb);
+
+ int journalVersion = bb.getInt();
+
+ int userVersion = bb.getInt();
+
+ long fileID = bb.getLong();
+
+ fileFactory.releaseBuffer(bb);
+
+ bb = null;
+ return fileID;
+ }
+
+ /**
+ * @param fileID
+ * @param sequentialFile
+ * @throws Exception
+ */
+ static int initFileHeader(final SequentialFileFactory fileFactory, final
SequentialFile sequentialFile, final int userVersion, final long fileID) throws Exception
+ {
+ // We don't need to release buffers while writing.
+ ByteBuffer bb = fileFactory.newBuffer(JournalImpl.SIZE_HEADER);
+
+ HornetQBuffer buffer = HornetQBuffers.wrappedBuffer(bb);
+
+ writeHeader(buffer, userVersion, fileID);
+
+ bb.rewind();
+
+ int bufferSize = bb.limit();
+
+ sequentialFile.position(0);
+
+ sequentialFile.writeDirect(bb, true);
+
+ return bufferSize;
+ }
+
+ /**
+ * @param buffer
+ * @param userVersion
+ * @param fileID
+ */
+ static void writeHeader(HornetQBuffer buffer, final int userVersion, final long
fileID)
+ {
+ buffer.writeInt(FORMAT_VERSION);
+
+ buffer.writeInt(userVersion);
+
+ 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
@@ -2867,7 +2938,7 @@
}
// Adding fileID
- encoder.setFileID(currentFile.getFileID());
+ encoder.setFileID(currentFile.getRecordID());
if (callback != null)
{
@@ -2903,10 +2974,10 @@
*/
private JournalFile createFile(final boolean keepOpened,
final boolean multiAIO,
- final boolean fill,
+ final boolean init,
final boolean tmpCompact) throws Exception
{
- int fileID = generateFileID();
+ long fileID = generateFileID();
String fileName;
@@ -2930,17 +3001,11 @@
sequentialFile.open(1, false);
- if (fill)
+ if (init)
{
sequentialFile.fill(0, fileSize, JournalImpl.FILL_CHARACTER);
- ByteBuffer bb = fileFactory.newBuffer(JournalImpl.SIZE_HEADER);
-
- bb.putInt(fileID);
-
- bb.rewind();
-
- sequentialFile.writeDirect(bb, true);
+ initFileHeader(this.fileFactory, sequentialFile, userVersion, fileID);
}
long position = sequentialFile.position();
@@ -2979,7 +3044,7 @@
file.getFile().position(file.getFile().calculateBlockStart(JournalImpl.SIZE_HEADER));
}
- private int generateFileID()
+ private long generateFileID()
{
return nextFileID.incrementAndGet();
}
@@ -3098,7 +3163,7 @@
*/
JournalFile getFile(final boolean keepOpened,
final boolean multiAIO,
- final boolean fill,
+ final boolean initFile,
final boolean tmpCompactExtension) throws Exception
{
JournalFile nextOpenedFile = null;
@@ -3117,7 +3182,7 @@
if (nextOpenedFile == null)
{
- nextOpenedFile = createFile(keepOpened, multiAIO, fill, tmpCompactExtension);
+ nextOpenedFile = createFile(keepOpened, multiAIO, initFile,
tmpCompactExtension);
}
else
{
@@ -3435,8 +3500,8 @@
{
public int compare(final JournalFile f1, final JournalFile f2)
{
- int id1 = f1.getFileID();
- int id2 = f2.getFileID();
+ long id1 = f1.getFileID();
+ long id2 = f2.getFileID();
return id1 < id2 ? -1 : id1 == id2 ? 0 : 1;
}
Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java
===================================================================
--- trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java 2010-04-29
16:21:28 UTC (rev 9185)
+++ trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java 2010-04-29
23:01:30 UTC (rev 9186)
@@ -533,8 +533,15 @@
public void runDirectJournalBlast() throws Exception
{
- // TODO Auto-generated method stub
+ localJournal.runDirectJournalBlast();
+ }
+ /* (non-Javadoc)
+ * @see org.hornetq.core.journal.Journal#getUserVersion()
+ */
+ public int getUserVersion()
+ {
+ return localJournal.getUserVersion();
}
// Package protected ---------------------------------------------
Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2010-04-29
16:21:28 UTC (rev 9185)
+++
trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2010-04-29
23:01:30 UTC (rev 9186)
@@ -1078,8 +1078,14 @@
public void runDirectJournalBlast() throws Exception
{
- // TODO Auto-generated method stub
+ }
+ /* (non-Javadoc)
+ * @see org.hornetq.core.journal.Journal#getUserVersion()
+ */
+ public int getUserVersion()
+ {
+ return 0;
}
}
Modified:
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/JournalImplTestUnit.java
===================================================================
---
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/JournalImplTestUnit.java 2010-04-29
16:21:28 UTC (rev 9185)
+++
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/JournalImplTestUnit.java 2010-04-29
23:01:30 UTC (rev 9186)
@@ -662,7 +662,7 @@
public void testReclaimAddUpdateDeleteDifferentFiles1() throws Exception
{
// Make sure there is one record per file
- setup(2, calculateRecordSize(8, getAlignment()) +
calculateRecordSize(JournalImpl.SIZE_ADD_RECORD + recordLength,
+ setup(2, calculateRecordSize(JournalImpl.SIZE_HEADER, getAlignment()) +
calculateRecordSize(JournalImpl.SIZE_ADD_RECORD + recordLength,
getAlignment()), true);
createJournal();
startJournal();
@@ -701,7 +701,7 @@
public void testReclaimAddUpdateDeleteDifferentFiles2() throws Exception
{
// Make sure there is one record per file
- setup(2, calculateRecordSize(8, getAlignment()) +
calculateRecordSize(JournalImpl.SIZE_ADD_RECORD + recordLength,
+ setup(2, calculateRecordSize(JournalImpl.SIZE_HEADER, getAlignment()) +
calculateRecordSize(JournalImpl.SIZE_ADD_RECORD + recordLength,
getAlignment()), true);
createJournal();
Modified: trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/ReclaimerTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/ReclaimerTest.java 2010-04-29
16:21:28 UTC (rev 9185)
+++ trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/ReclaimerTest.java 2010-04-29
23:01:30 UTC (rev 9186)
@@ -792,7 +792,7 @@
return 0;
}
- public int getFileID()
+ public long getFileID()
{
return 0;
}
@@ -977,5 +977,13 @@
this.needCleanup = needCleanup;
}
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.journal.impl.JournalFile#getRecordID()
+ */
+ public int getRecordID()
+ {
+ return 0;
+ }
}
}