[hornetq-commits] JBoss hornetq SVN: r8483 - in trunk: src/config/common/schema and 25 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Tue Dec 1 15:18:49 EST 2009
Author: timfox
Date: 2009-12-01 15:18:47 -0500 (Tue, 01 Dec 2009)
New Revision: 8483
Added:
trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFileFactory.java
Removed:
trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java
Modified:
trunk/build-hornetq.xml
trunk/src/config/common/schema/hornetq-configuration.xsd
trunk/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java
trunk/src/main/org/hornetq/core/config/Configuration.java
trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java
trunk/src/main/org/hornetq/core/journal/IOCompletion.java
trunk/src/main/org/hornetq/core/journal/Journal.java
trunk/src/main/org/hornetq/core/journal/SequentialFile.java
trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java
trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java
trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java
trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
trunk/src/main/org/hornetq/core/journal/impl/DummyCallback.java
trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java
trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java
trunk/src/main/org/hornetq/core/journal/impl/SimpleWaitIOCallback.java
trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java
trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java
trunk/src/main/org/hornetq/core/management/HornetQServerControl.java
trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java
trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java
trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
trunk/src/main/org/hornetq/utils/TokenBucketLimiterImpl.java
trunk/tests/config/ConfigurationTest-full-config.xml
trunk/tests/src/org/hornetq/tests/integration/client/BlockingSendTest.java
trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverReplicationTest.java
trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java
trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java
trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalCompactTest.java
trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalImplTest.java
trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java
trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java
trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java
trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java
trunk/tests/src/org/hornetq/tests/opt/SendTest.java
trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java
trunk/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java
trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java
trunk/tests/src/org/hornetq/tests/unit/core/deployers/impl/QueueDeployerTest.java
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java
trunk/tests/src/org/hornetq/tests/util/ListJournal.java
Log:
Mainly tuning of journal/timed buffer and various other small changes
Modified: trunk/build-hornetq.xml
===================================================================
--- trunk/build-hornetq.xml 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/build-hornetq.xml 2009-12-01 20:18:47 UTC (rev 8483)
@@ -1226,7 +1226,7 @@
<fileset dir="${test.classes.dir}">
<!-- exlcuded because of https://jira.jboss.org/jira/browse/HORNETQ-65 -->
<exclude name="**/cluster/failover/*StaticClusterWithBackupFailoverTest.class" />
- <exclude name="**/cluster/failover/*WithBackupFailoverTest.class" />
+ <exclude name="**/cluster/failover/*WithBackupFailoverTest.class" />
<include name="${tests.param}"/>
</fileset>
</batchtest>
Modified: trunk/src/config/common/schema/hornetq-configuration.xsd
===================================================================
--- trunk/src/config/common/schema/hornetq-configuration.xsd 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/config/common/schema/hornetq-configuration.xsd 2009-12-01 20:18:47 UTC (rev 8483)
@@ -151,9 +151,7 @@
<xsd:element maxOccurs="1" minOccurs="0" name="journal-buffer-timeout" type="xsd:long">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="journal-buffer-size" type="xsd:long">
- </xsd:element>
- <xsd:element maxOccurs="1" minOccurs="0" name="journal-flush-on-sync" type="xsd:boolean">
- </xsd:element>
+ </xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="journal-sync-transactional" type="xsd:boolean">
</xsd:element>
<xsd:element maxOccurs="1" minOccurs="0" name="journal-sync-non-transactional" type="xsd:boolean">
Modified: trunk/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/asyncio/impl/AsynchronousFileImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -290,12 +290,12 @@
if (writeExecutor != null)
{
+ maxIOSemaphore.acquireUninterruptibly();
+
writeExecutor.execute(new Runnable()
{
public void run()
{
- maxIOSemaphore.acquireUninterruptibly();
-
long sequence = nextWritingSequence.getAndIncrement();
try
@@ -445,7 +445,7 @@
private void callbackDone(final AIOCallback callback, final long sequence, final ByteBuffer buffer)
{
maxIOSemaphore.release();
-
+
pendingWrites.down();
callbackLock.lock();
Modified: trunk/src/main/org/hornetq/core/config/Configuration.java
===================================================================
--- trunk/src/main/org/hornetq/core/config/Configuration.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/config/Configuration.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -224,22 +224,34 @@
void setJournalMinFiles(int files);
- int getJournalMaxAIO();
+ //AIO and NIO need different values for these params
+
+ int getJournalMaxIO_AIO();
- void setJournalMaxAIO(int maxAIO);
+ void setJournalMaxIO_AIO(int journalMaxIO);
- void setJournalBufferSize(int size);
+ int getJournalBufferTimeout_AIO();
+
+ void setJournalBufferTimeout_AIO(int journalBufferTimeout);
+
+ int getJournalBufferSize_AIO();
+
+ void setJournalBufferSize_AIO(int journalBufferSize);
- int getJournalBufferSize();
- void setJournalBufferTimeout(int timeout);
+ int getJournalMaxIO_NIO();
+
+ void setJournalMaxIO_NIO(int journalMaxIO);
+
+ int getJournalBufferTimeout_NIO();
+
+ void setJournalBufferTimeout_NIO(int journalBufferTimeout);
+
+ int getJournalBufferSize_NIO();
+
+ void setJournalBufferSize_NIO(int journalBufferSize);
- int getJournalBufferTimeout();
-
- void setJournalFlushOnSync(boolean flush);
-
- boolean isJournalFlushOnSync();
-
+
boolean isCreateBindingsDir();
void setCreateBindingsDir(boolean create);
Modified: trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -100,14 +100,22 @@
public static final int DEFAULT_JOURNAL_MIN_FILES = 2;
- public static final int DEFAULT_JOURNAL_MAX_AIO = 500;
+ // AIO and NIO need to have different defaults for some values
- public static final boolean DEFAULT_JOURNAL_FLUSH_SYNC = false;
+ public static final int DEFAULT_JOURNAL_MAX_IO_AIO = 500;
- public static final int DEFAULT_JOURNAL_BUFFER_TIMEOUT = 20000;
+ public static final int DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO = (int)(1000000000d / 2000);
- public static final int DEFAULT_JOURNAL_BUFFER_SIZE = 128 * 1024;
+ public static final int DEFAULT_JOURNAL_BUFFER_SIZE_AIO = 490 * 1024;
+ public static final int DEFAULT_JOURNAL_MAX_IO_NIO = 1;
+
+ public static final int DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO = (int)(1000000000d / 300);
+
+ public static final int DEFAULT_JOURNAL_BUFFER_SIZE_NIO = 490 * 1024;
+
+
+
public static final boolean DEFAULT_JOURNAL_LOG_WRITE_RATE = false;
public static final int DEFAULT_JOURNAL_PERF_BLAST_PAGES = -1;
@@ -266,14 +274,23 @@
protected int journalMinFiles = DEFAULT_JOURNAL_MIN_FILES;
- protected int journalMaxAIO = DEFAULT_JOURNAL_MAX_AIO;
+
+ //AIO and NIO need different values for these attributes
+
+ protected int journalMaxIO_AIO = DEFAULT_JOURNAL_MAX_IO_AIO;
- protected boolean journalFlushSync = DEFAULT_JOURNAL_FLUSH_SYNC;
+ protected int journalBufferTimeout_AIO = DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO;
- protected int journalBufferTimeout = DEFAULT_JOURNAL_BUFFER_TIMEOUT;
+ protected int journalBufferSize_AIO = DEFAULT_JOURNAL_BUFFER_SIZE_AIO;
+
+ protected int journalMaxIO_NIO = DEFAULT_JOURNAL_MAX_IO_NIO;
- protected int journalBufferSize = DEFAULT_JOURNAL_BUFFER_SIZE;
+ protected int journalBufferTimeout_NIO = DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO;
+ protected int journalBufferSize_NIO = DEFAULT_JOURNAL_BUFFER_SIZE_NIO;
+
+
+
protected boolean logJournalWriteRate = DEFAULT_JOURNAL_LOG_WRITE_RATE;
protected int journalPerfBlastPages = DEFAULT_JOURNAL_PERF_BLAST_PAGES;
@@ -665,16 +682,6 @@
journalFileSize = size;
}
- public int getJournalMaxAIO()
- {
- return journalMaxAIO;
- }
-
- public void setJournalMaxAIO(final int maxAIO)
- {
- journalMaxAIO = maxAIO;
- }
-
public int getJournalMinFiles()
{
return journalMinFiles;
@@ -815,36 +822,6 @@
jmxDomain = domain;
}
- public void setJournalBufferTimeout(int timeout)
- {
- this.journalBufferTimeout = timeout;
- }
-
- public int getJournalBufferTimeout()
- {
- return journalBufferTimeout;
- }
-
- public void setJournalFlushOnSync(boolean flush)
- {
- journalFlushSync = flush;
- }
-
- public boolean isJournalFlushOnSync()
- {
- return journalFlushSync;
- }
-
- public int getJournalBufferSize()
- {
- return journalBufferSize;
- }
-
- public void setJournalBufferSize(int size)
- {
- this.journalBufferSize = size;
- }
-
public String getLargeMessagesDirectory()
{
return largeMessagesDirectory;
@@ -930,6 +907,131 @@
this.managementRequestTimeout = managementRequestTimeout;
}
+ public int getJournalCompactMinFiles()
+ {
+ return journalCompactMinFiles;
+ }
+
+ public int getJournalCompactPercentage()
+ {
+ return journalCompactPercentage;
+ }
+
+ public void setJournalCompactMinFiles(int minFiles)
+ {
+ this.journalCompactMinFiles = minFiles;
+ }
+
+ public void setJournalCompactPercentage(int percentage)
+ {
+ this.journalCompactPercentage = percentage;
+ }
+
+ public long getServerDumpInterval()
+ {
+ return serverDumpInterval;
+ }
+
+ public void setServerDumpInterval(long intervalInMilliseconds)
+ {
+ this.serverDumpInterval = intervalInMilliseconds;
+ }
+
+ public int getMemoryWarningThreshold()
+ {
+ return memoryWarningThreshold;
+ }
+
+ public void setMemoryWarningThreshold(int memoryWarningThreshold)
+ {
+ this.memoryWarningThreshold = memoryWarningThreshold;
+ }
+
+ public long getMemoryMeasureInterval()
+ {
+ return memoryMeasureInterval;
+ }
+
+ public void setMemoryMeasureInterval(long memoryMeasureInterval)
+ {
+ this.memoryMeasureInterval = memoryMeasureInterval;
+ }
+
+ public String getLogDelegateFactoryClassName()
+ {
+ return logDelegateFactoryClassName;
+ }
+
+ public void setLogDelegateFactoryClassName(String className)
+ {
+ this.logDelegateFactoryClassName = className;
+ }
+
+
+
+ public int getJournalMaxIO_AIO()
+ {
+ return journalMaxIO_AIO;
+ }
+
+ public void setJournalMaxIO_AIO(int journalMaxIO)
+ {
+ this.journalMaxIO_AIO = journalMaxIO;
+ }
+
+ public int getJournalBufferTimeout_AIO()
+ {
+ return journalBufferTimeout_AIO;
+ }
+
+ public void setJournalBufferTimeout_AIO(int journalBufferTimeout)
+ {
+ this.journalBufferTimeout_AIO = journalBufferTimeout;
+ }
+
+ public int getJournalBufferSize_AIO()
+ {
+ return journalBufferSize_AIO;
+ }
+
+ public void setJournalBufferSize_AIO(int journalBufferSize)
+ {
+ this.journalBufferSize_AIO = journalBufferSize;
+ }
+
+
+ public int getJournalMaxIO_NIO()
+ {
+ return journalMaxIO_NIO;
+ }
+
+ public void setJournalMaxIO_NIO(int journalMaxIO)
+ {
+ this.journalMaxIO_NIO = journalMaxIO;
+ }
+
+ public int getJournalBufferTimeout_NIO()
+ {
+ return journalBufferTimeout_NIO;
+ }
+
+ public void setJournalBufferTimeout_NIO(int journalBufferTimeout)
+ {
+ this.journalBufferTimeout_NIO = journalBufferTimeout;
+ }
+
+ public int getJournalBufferSize_NIO()
+ {
+ return journalBufferSize_NIO;
+ }
+
+ public void setJournalBufferSize_NIO(int journalBufferSize)
+ {
+ this.journalBufferSize_NIO = journalBufferSize;
+ }
+
+
+
@Override
public boolean equals(Object obj)
{
@@ -965,7 +1067,7 @@
{
return false;
}
-
+
if (clustered != other.clustered)
return false;
if (connectionTTLOverride != other.connectionTTLOverride)
@@ -983,12 +1085,18 @@
return false;
if (jmxManagementEnabled != other.jmxManagementEnabled)
return false;
- if (journalBufferSize != other.journalBufferSize)
+ if (this.journalBufferSize_AIO != other.journalBufferSize_AIO)
return false;
- if (journalBufferTimeout != other.journalBufferTimeout)
+ if (journalBufferTimeout_AIO != other.journalBufferTimeout_AIO)
return false;
- if (journalFlushSync != other.journalFlushSync)
+ if (journalMaxIO_AIO != other.journalMaxIO_AIO)
+ return false;
+ if (this.journalBufferSize_NIO != other.journalBufferSize_NIO)
return false;
+ if (journalBufferTimeout_NIO != other.journalBufferTimeout_NIO)
+ return false;
+ if (journalMaxIO_NIO != other.journalMaxIO_NIO)
+ return false;
if (journalCompactMinFiles != other.journalCompactMinFiles)
return false;
if (journalCompactPercentage != other.journalCompactPercentage)
@@ -1002,8 +1110,7 @@
return false;
if (journalFileSize != other.journalFileSize)
return false;
- if (journalMaxAIO != other.journalMaxAIO)
- return false;
+
if (journalMinFiles != other.journalMinFiles)
return false;
if (journalPerfBlastPages != other.journalPerfBlastPages)
@@ -1100,64 +1207,4 @@
return true;
}
- public int getJournalCompactMinFiles()
- {
- return journalCompactMinFiles;
- }
-
- public int getJournalCompactPercentage()
- {
- return journalCompactPercentage;
- }
-
- public void setJournalCompactMinFiles(int minFiles)
- {
- this.journalCompactMinFiles = minFiles;
- }
-
- public void setJournalCompactPercentage(int percentage)
- {
- this.journalCompactPercentage = percentage;
- }
-
- public long getServerDumpInterval()
- {
- return serverDumpInterval;
- }
-
- public void setServerDumpInterval(long intervalInMilliseconds)
- {
- this.serverDumpInterval = intervalInMilliseconds;
- }
-
- public int getMemoryWarningThreshold()
- {
- return memoryWarningThreshold;
- }
-
- public void setMemoryWarningThreshold(int memoryWarningThreshold)
- {
- this.memoryWarningThreshold = memoryWarningThreshold;
- }
-
- public long getMemoryMeasureInterval()
- {
- return memoryMeasureInterval;
- }
-
- public void setMemoryMeasureInterval(long memoryMeasureInterval)
- {
- this.memoryMeasureInterval = memoryMeasureInterval;
- }
-
- public String getLogDelegateFactoryClassName()
- {
- return logDelegateFactoryClassName;
- }
-
- public void setLogDelegateFactoryClassName(String className)
- {
- this.logDelegateFactoryClassName = className;
- }
-
}
Modified: trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java
===================================================================
--- trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -73,9 +73,9 @@
private static final String DEFAULT_CONFIGURATION_URL = "hornetq-configuration.xml";
private static final String CONFIGURATION_SCHEMA_URL = "schema/hornetq-configuration.xsd";
-
- //For a bridge confirmations must be activated or send acknowledgements won't return
-
+
+ // For a bridge confirmations must be activated or send acknowledgements won't return
+
public static final int DEFAULT_CONFIRMATION_WINDOW_SIZE = 1024 * 1024;
// Static --------------------------------------------------------------------------
@@ -335,11 +335,36 @@
journalFileSize = getInteger(e, "journal-file-size", journalFileSize, GT_ZERO);
- journalFlushSync = getBoolean(e, "journal-flush-on-sync", DEFAULT_JOURNAL_FLUSH_SYNC);
+ int journalBufferTimeout = getInteger(e,
+ "journal-buffer-timeout",
+ journalType == JournalType.ASYNCIO ? DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO
+ : DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO,
+ GT_ZERO);
- journalBufferTimeout = getInteger(e, "journal-buffer-timeout", DEFAULT_JOURNAL_BUFFER_TIMEOUT, GT_ZERO);
+ int journalBufferSize = getInteger(e,
+ "journal-buffer-size",
+ journalType == JournalType.ASYNCIO ? DEFAULT_JOURNAL_BUFFER_SIZE_AIO
+ : DEFAULT_JOURNAL_BUFFER_SIZE_NIO,
+ GT_ZERO);
- journalBufferSize = getInteger(e, "journal-buffer-size", DEFAULT_JOURNAL_BUFFER_SIZE, GT_ZERO);
+ int journalMaxIO = getInteger(e,
+ "journal-max-aio",
+ journalType == JournalType.ASYNCIO ? DEFAULT_JOURNAL_MAX_IO_AIO
+ : DEFAULT_JOURNAL_MAX_IO_NIO,
+ GT_ZERO);
+
+ if (journalType == JournalType.ASYNCIO)
+ {
+ journalBufferTimeout_AIO = journalBufferTimeout;
+ journalBufferSize_AIO = journalBufferSize;
+ journalMaxIO_AIO = journalMaxIO;
+ }
+ else
+ {
+ journalBufferTimeout_NIO = journalBufferTimeout;
+ journalBufferSize_NIO = journalBufferSize;
+ journalMaxIO_NIO = journalMaxIO;
+ }
journalMinFiles = getInteger(e, "journal-min-files", journalMinFiles, GT_ZERO);
@@ -347,8 +372,6 @@
journalCompactPercentage = getInteger(e, "journal-compact-percentage", journalCompactPercentage, PERCENTAGE);
- journalMaxAIO = getInteger(e, "journal-max-aio", journalMaxAIO, GT_ZERO);
-
logJournalWriteRate = getBoolean(e, "log-journal-write-rate", DEFAULT_JOURNAL_LOG_WRITE_RATE);
journalPerfBlastPages = getInteger(e, "perf-blast-pages", DEFAULT_JOURNAL_PERF_BLAST_PAGES, MINUS_ONE_OR_GT_ZERO);
@@ -367,12 +390,12 @@
GT_ZERO);
serverDumpInterval = getLong(e, "server-dump-interval", serverDumpInterval, MINUS_ONE_OR_GT_ZERO); // in
- // milliseconds
+ // milliseconds
memoryWarningThreshold = getInteger(e, "memory-warning-threshold", memoryWarningThreshold, PERCENTAGE);
memoryMeasureInterval = getLong(e, "memory-measure-interval", memoryMeasureInterval, MINUS_ONE_OR_GT_ZERO); // in
- // milliseconds
+ // milliseconds
backupWindowSize = getInteger(e, "backup-window-size", DEFAULT_BACKUP_WINDOW_SIZE, MINUS_ONE_OR_GT_ZERO);
@@ -523,7 +546,7 @@
long retryInterval = getLong(e, "retry-interval", DEFAULT_CLUSTER_RETRY_INTERVAL, GT_ZERO);
int confirmationWindowSize = getInteger(e, "confirmation-window-size", DEFAULT_CONFIRMATION_WINDOW_SIZE, GT_ZERO);
-
+
String discoveryGroupName = null;
List<Pair<String, String>> connectorPairs = new ArrayList<Pair<String, String>>();
@@ -609,10 +632,13 @@
String transformerClassName = getString(brNode, "transformer-class-name", null, NO_CHECK);
long retryInterval = getLong(brNode, "retry-interval", DEFAULT_RETRY_INTERVAL, GT_ZERO);
-
- //Default bridge conf
- int confirmationWindowSize = getInteger(brNode, "confirmation-window-size", DEFAULT_CONFIRMATION_WINDOW_SIZE, GT_ZERO);
-
+
+ // Default bridge conf
+ int confirmationWindowSize = getInteger(brNode,
+ "confirmation-window-size",
+ DEFAULT_CONFIRMATION_WINDOW_SIZE,
+ GT_ZERO);
+
double retryIntervalMultiplier = getDouble(brNode,
"retry-interval-multiplier",
DEFAULT_RETRY_INTERVAL_MULTIPLIER,
Modified: trunk/src/main/org/hornetq/core/journal/IOCompletion.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/IOCompletion.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/IOCompletion.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -22,5 +22,5 @@
*/
public interface IOCompletion extends IOAsyncTask
{
- void lineUp();
+ void storeLineUp();
}
Modified: trunk/src/main/org/hornetq/core/journal/Journal.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/Journal.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/Journal.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -112,5 +112,6 @@
void perfBlast(int pages) throws Exception;
+ void runDirectJournalBlast() throws Exception;
}
Modified: trunk/src/main/org/hornetq/core/journal/SequentialFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/SequentialFile.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/SequentialFile.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -88,10 +88,6 @@
void renameTo(String newFileName) throws Exception;
- void disableAutoFlush();
-
- void enableAutoFlush();
-
SequentialFile copy();
void setTimedBuffer(TimedBuffer buffer);
Modified: trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -36,9 +36,6 @@
void releaseBuffer(ByteBuffer buffer);
- /** The factory may need to do some initialization before the file is activated.
- * this was added as a hook for AIO to initialize the Observer on TimedBuffer.
- * It could be eventually done the same on NIO if we implement TimedBuffer on NIO */
void activateBuffer(SequentialFile file);
void deactivateBuffer();
@@ -61,7 +58,5 @@
*/
void createDirs() throws Exception;
- void flush();
-
-
+ void flush();
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFile.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -46,7 +46,6 @@
/** The pool for Thread pollers */
private final Executor pollerExecutor;
-
public AIOSequentialFile(final SequentialFileFactory factory,
final int bufferSize,
@@ -87,18 +86,27 @@
public SequentialFile copy()
{
- return new AIOSequentialFile(factory, -1, -1, getFile().getParent(), getFileName(), maxIO, bufferCallback, writerExecutor, pollerExecutor);
+ return new AIOSequentialFile(factory,
+ -1,
+ -1,
+ getFile().getParent(),
+ getFileName(),
+ maxIO,
+ bufferCallback,
+ writerExecutor,
+ pollerExecutor);
}
+ @Override
public synchronized void close() throws Exception
{
if (!opened)
{
return;
}
-
+
super.close();
-
+
opened = false;
timedBuffer = null;
@@ -106,7 +114,7 @@
aioFile.close();
aioFile = null;
- this.notifyAll();
+ notifyAll();
}
/* (non-Javadoc)
@@ -165,7 +173,7 @@
aioFile.fill(filePosition, blocks, blockSize, fillCharacter);
- this.fileSize = aioFile.size();
+ fileSize = aioFile.size();
}
public void open() throws Exception
@@ -176,16 +184,16 @@
public synchronized void open(final int maxIO) throws Exception
{
opened = true;
-
+
aioFile = new AsynchronousFileImpl(writerExecutor, pollerExecutor);
-
+
aioFile.open(getFile().getAbsolutePath(), maxIO);
-
+
position.set(0);
-
+
aioFile.setBufferCallback(bufferCallback);
-
- this.fileSize = aioFile.size();
+
+ fileSize = aioFile.size();
}
public void setBufferCallback(final BufferCallback callback)
@@ -216,7 +224,6 @@
return bytesRead;
}
-
public void sync() throws Exception
{
@@ -244,18 +251,14 @@
// Public methods
// -----------------------------------------------------------------------------------------------------
- // Protected methods
- // -----------------------------------------------------------------------------------------------------
-
-
public void writeDirect(final ByteBuffer bytes, final boolean sync) throws Exception
{
if (sync)
{
SimpleWaitIOCallback completion = new SimpleWaitIOCallback();
-
+
writeDirect(bytes, true, completion);
-
+
completion.waitCompletion();
}
else
@@ -264,12 +267,11 @@
}
}
-
/**
*
* @param sync Not used on AIO
* */
- public void writeDirect(final ByteBuffer bytes, final boolean sync, IOAsyncTask callback)
+ public void writeDirect(final ByteBuffer bytes, final boolean sync, final IOAsyncTask callback)
{
final int bytesToWrite = factory.calculateBlockSize(bytes.limit());
@@ -278,7 +280,20 @@
aioFile.write(positionToWrite, bytesToWrite, bytes, callback);
}
+ // Protected methods
+ // -----------------------------------------------------------------------------------------------------
+ @Override
+ protected ByteBuffer newBuffer(int size, int limit)
+ {
+ size = factory.calculateBlockSize(size);
+ limit = factory.calculateBlockSize(limit);
+
+ ByteBuffer buffer = factory.newBuffer(size);
+ buffer.limit(limit);
+ return buffer;
+ }
+
// Private methods
// -----------------------------------------------------------------------------------------------------
Modified: trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -33,7 +33,7 @@
* @author clebert.suconic at jboss.com
*
*/
-public class AIOSequentialFileFactory extends AbstractSequentialFactory
+public class AIOSequentialFileFactory extends AbstractSequentialFileFactory
{
private static final Logger log = Logger.getLogger(AIOSequentialFileFactory.class);
@@ -55,19 +55,17 @@
public AIOSequentialFileFactory(final String journalDir)
{
this(journalDir,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT,
- ConfigurationImpl.DEFAULT_JOURNAL_FLUSH_SYNC,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO,
false);
}
public AIOSequentialFileFactory(final String journalDir,
final int bufferSize,
- final long bufferTimeout,
- final boolean flushOnSync,
+ final int bufferTimeout,
final boolean logRates)
{
- super(journalDir, true, bufferSize, bufferTimeout, flushOnSync, logRates);
+ super(journalDir, true, bufferSize, bufferTimeout, logRates);
}
public SequentialFile createSequentialFile(final String fileName, final int maxIO)
@@ -154,18 +152,21 @@
{
buffersControl.stop();
- pollerExecutor.shutdown();
+ if (pollerExecutor != null)
+ {
+ pollerExecutor.shutdown();
- try
- {
- if (!pollerExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS))
+ try
{
- log.warn("Timed out on AIO poller shutdown", new Exception("Timed out on AIO writer shutdown"));
+ if (!pollerExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS))
+ {
+ log.warn("Timed out on AIO poller shutdown", new Exception("Timed out on AIO writer shutdown"));
+ }
}
+ catch (InterruptedException e)
+ {
+ }
}
- catch (InterruptedException e)
- {
- }
super.stop();
}
Deleted: trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -1,200 +0,0 @@
-/*
- * Copyright 2009 Red Hat, Inc.
- * Red Hat 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.hornetq.core.journal.impl;
-
-import java.io.File;
-import java.io.FilenameFilter;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-
-import org.hornetq.core.journal.SequentialFile;
-import org.hornetq.core.journal.SequentialFileFactory;
-import org.hornetq.core.logging.Logger;
-import org.hornetq.utils.HornetQThreadFactory;
-
-/**
- *
- * An abstract SequentialFileFactory containing basic functionality for both AIO and NIO SequentialFactories
- *
- * @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 abstract class AbstractSequentialFactory implements SequentialFileFactory
-{
-
- // Timeout used to wait executors to shutdown
- protected static final int EXECUTOR_TIMEOUT = 60;
-
- private static final Logger log = Logger.getLogger(AbstractSequentialFactory.class);
-
- protected final String journalDir;
-
- protected final TimedBuffer timedBuffer;
-
- protected final int bufferSize;
-
- protected final long bufferTimeout;
-
- /**
- * Asynchronous writes need to be done at another executor.
- * This needs to be done at NIO, or else we would have the callers thread blocking for the return.
- * At AIO this is necessary as context switches on writes would fire flushes at the kernel.
- * */
- protected ExecutorService writeExecutor;
-
- public AbstractSequentialFactory(final String journalDir,
- final boolean buffered,
- final int bufferSize,
- final long bufferTimeout,
- final boolean flushOnSync,
- final boolean logRates)
- {
- this.journalDir = journalDir;
-
- if (buffered)
- {
- timedBuffer = new TimedBuffer(bufferSize, bufferTimeout, flushOnSync, logRates);
- }
- else
- {
- timedBuffer = null;
- }
- this.bufferSize = bufferSize;
- this.bufferTimeout = bufferTimeout;
- }
-
- public void stop()
- {
- if (timedBuffer != null)
- {
- timedBuffer.stop();
- }
-
- if (isSupportsCallbacks())
- {
- writeExecutor.shutdown();
-
- try
- {
- if (!writeExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS))
- {
- log.warn("Timed out on AIO writer shutdown", new Exception("Timed out on AIO writer shutdown"));
- }
- }
- catch (InterruptedException e)
- {
- }
- }
-
-
- }
-
- public void start()
- {
- if (timedBuffer != null)
- {
- timedBuffer.start();
- }
-
- if (isSupportsCallbacks())
- {
- writeExecutor = Executors.newSingleThreadExecutor(new HornetQThreadFactory("HornetQ-Asynchronous-Persistent-Writes" + System.identityHashCode(this),
- true));
- }
-
-
- }
-
- /* (non-Javadoc)
- * @see org.hornetq.core.journal.SequentialFileFactory#activate(org.hornetq.core.journal.SequentialFile)
- */
- public void activateBuffer(final SequentialFile file)
- {
- if (timedBuffer != null)
- {
- timedBuffer.disableAutoFlush();
- try
- {
- file.setTimedBuffer(timedBuffer);
- }
- finally
- {
- file.enableAutoFlush();
- }
- }
- }
-
- public void flush()
- {
- if (timedBuffer != null)
- {
- timedBuffer.flush();
- }
- }
-
- public void deactivateBuffer()
- {
- if (timedBuffer != null)
- {
- timedBuffer.flush();
- timedBuffer.setObserver(null);
- }
- }
-
- public void releaseBuffer(ByteBuffer buffer)
- {
- }
-
- /**
- * Create the directory if it doesn't exist yet
- */
- public void createDirs() throws Exception
- {
- File file = new File(journalDir);
- boolean ok = file.mkdirs();
- if (!ok)
- {
- throw new IOException("Failed to create directory " + journalDir);
- }
- }
-
- public List<String> listFiles(final String extension) throws Exception
- {
- File dir = new File(journalDir);
-
- FilenameFilter fnf = new FilenameFilter()
- {
- public boolean accept(final File file, final String name)
- {
- return name.endsWith("." + extension);
- }
- };
-
- String[] fileNames = dir.list(fnf);
-
- if (fileNames == null)
- {
- throw new IOException("Failed to list: " + journalDir);
- }
-
- return Arrays.asList(fileNames);
- }
-
-}
Modified: trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -21,9 +21,9 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.buffers.HornetQBuffer;
import org.hornetq.core.buffers.HornetQBuffers;
+import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.journal.IOAsyncTask;
import org.hornetq.core.journal.SequentialFile;
import org.hornetq.core.journal.SequentialFileFactory;
@@ -162,22 +162,6 @@
}
}
- public final void disableAutoFlush()
- {
- if (timedBuffer != null)
- {
- timedBuffer.disableAutoFlush();
- }
- }
-
- public final void enableAutoFlush()
- {
- if (timedBuffer != null)
- {
- timedBuffer.enableAutoFlush();
- }
- }
-
public void setTimedBuffer(final TimedBuffer buffer)
{
if (timedBuffer != null)
@@ -315,7 +299,17 @@
}
}
}
+
+ protected ByteBuffer newBuffer(int size, int limit)
+ {
+ size = factory.calculateBlockSize(size);
+ limit = factory.calculateBlockSize(limit);
+ ByteBuffer buffer = factory.newBuffer(size);
+ buffer.limit(limit);
+ return buffer;
+ }
+
protected class LocalBufferObserver implements TimedBufferObserver
{
public void flushBuffer(final ByteBuffer buffer, final boolean requestedSync, final List<IOAsyncTask> callbacks)
@@ -334,12 +328,7 @@
public ByteBuffer newBuffer(int size, int limit)
{
- size = factory.calculateBlockSize(size);
- limit = factory.calculateBlockSize(limit);
-
- ByteBuffer buffer = factory.newBuffer(size);
- buffer.limit(limit);
- return buffer;
+ return AbstractSequentialFile.this.newBuffer(size, limit);
}
public int getRemainingBytes()
Copied: trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFileFactory.java (from rev 8467, trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFactory.java)
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFileFactory.java (rev 0)
+++ trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFileFactory.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -0,0 +1,187 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat 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.hornetq.core.journal.impl;
+
+import java.io.File;
+import java.io.FilenameFilter;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.hornetq.core.journal.SequentialFile;
+import org.hornetq.core.journal.SequentialFileFactory;
+import org.hornetq.core.logging.Logger;
+import org.hornetq.utils.HornetQThreadFactory;
+
+/**
+ *
+ * An abstract SequentialFileFactory containing basic functionality for both AIO and NIO SequentialFactories
+ *
+ * @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 abstract class AbstractSequentialFileFactory implements SequentialFileFactory
+{
+
+ // Timeout used to wait executors to shutdown
+ protected static final int EXECUTOR_TIMEOUT = 60;
+
+ private static final Logger log = Logger.getLogger(AbstractSequentialFileFactory.class);
+
+ protected final String journalDir;
+
+ protected final TimedBuffer timedBuffer;
+
+ protected final int bufferSize;
+
+ protected final long bufferTimeout;
+
+ /**
+ * Asynchronous writes need to be done at another executor.
+ * This needs to be done at NIO, or else we would have the callers thread blocking for the return.
+ * At AIO this is necessary as context switches on writes would fire flushes at the kernel.
+ * */
+ protected ExecutorService writeExecutor;
+
+ public AbstractSequentialFileFactory(final String journalDir,
+ final boolean buffered,
+ final int bufferSize,
+ final int bufferTimeout,
+ final boolean logRates)
+ {
+ this.journalDir = journalDir;
+
+ if (buffered)
+ {
+ timedBuffer = new TimedBuffer(bufferSize, bufferTimeout, logRates);
+ }
+ else
+ {
+ timedBuffer = null;
+ }
+ this.bufferSize = bufferSize;
+ this.bufferTimeout = bufferTimeout;
+ }
+
+ public void stop()
+ {
+ if (timedBuffer != null)
+ {
+ timedBuffer.stop();
+ }
+
+ if (isSupportsCallbacks() && writeExecutor != null)
+ {
+ writeExecutor.shutdown();
+
+ try
+ {
+ if (!writeExecutor.awaitTermination(EXECUTOR_TIMEOUT, TimeUnit.SECONDS))
+ {
+ log.warn("Timed out on AIO writer shutdown", new Exception("Timed out on AIO writer shutdown"));
+ }
+ }
+ catch (InterruptedException e)
+ {
+ }
+ }
+ }
+
+ public void start()
+ {
+ if (timedBuffer != null)
+ {
+ timedBuffer.start();
+ }
+
+ if (isSupportsCallbacks())
+ {
+ writeExecutor = Executors.newSingleThreadExecutor(new HornetQThreadFactory("HornetQ-Asynchronous-Persistent-Writes" + System.identityHashCode(this),
+ true));
+ }
+
+ }
+
+ /* (non-Javadoc)
+ * @see org.hornetq.core.journal.SequentialFileFactory#activate(org.hornetq.core.journal.SequentialFile)
+ */
+ public void activateBuffer(final SequentialFile file)
+ {
+ if (timedBuffer != null)
+ {
+ file.setTimedBuffer(timedBuffer);
+ }
+ }
+
+ public void flush()
+ {
+ if (timedBuffer != null)
+ {
+ timedBuffer.flush();
+ }
+ }
+
+ public void deactivateBuffer()
+ {
+ if (timedBuffer != null)
+ {
+ timedBuffer.flush();
+ timedBuffer.setObserver(null);
+ }
+ }
+
+ public void releaseBuffer(final ByteBuffer buffer)
+ {
+ }
+
+ /**
+ * Create the directory if it doesn't exist yet
+ */
+ public void createDirs() throws Exception
+ {
+ File file = new File(journalDir);
+ boolean ok = file.mkdirs();
+ if (!ok)
+ {
+ throw new IOException("Failed to create directory " + journalDir);
+ }
+ }
+
+ public List<String> listFiles(final String extension) throws Exception
+ {
+ File dir = new File(journalDir);
+
+ FilenameFilter fnf = new FilenameFilter()
+ {
+ public boolean accept(final File file, final String name)
+ {
+ return name.endsWith("." + extension);
+ }
+ };
+
+ String[] fileNames = dir.list(fnf);
+
+ if (fileNames == null)
+ {
+ throw new IOException("Failed to list: " + journalDir);
+ }
+
+ return Arrays.asList(fileNames);
+ }
+}
Modified: trunk/src/main/org/hornetq/core/journal/impl/DummyCallback.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/DummyCallback.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/DummyCallback.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -50,7 +50,7 @@
/* (non-Javadoc)
* @see org.hornetq.core.journal.IOCompletion#linedUp()
*/
- public void lineUp()
+ public void storeLineUp()
{
}
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/JournalImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -103,7 +103,6 @@
private static final void trace(final String message)
{
log.trace(message);
- //System.out.println("JournalImpl::" + message);
}
// The sizes of primitive types
@@ -166,7 +165,6 @@
private final AtomicInteger nextFileID = new AtomicInteger(0);
- // used for Asynchronous IO only (ignored on NIO).
private final int maxAIO;
private final int fileSize;
@@ -219,9 +217,70 @@
private volatile int state;
private final Reclaimer reclaimer = new Reclaimer();
-
+
// Constructors --------------------------------------------------
+ public void runDirectJournalBlast() throws Exception
+ {
+ final int numIts = 100000000;
+
+ 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(int errorCode, 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(HornetQBuffer buffer)
+ {
+ }
+
+ public void encode(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 JournalImpl(final int fileSize,
final int minFiles,
final int compactMinFiles,
@@ -229,7 +288,7 @@
final SequentialFileFactory fileFactory,
final String filePrefix,
final String fileExtension,
- final int maxAIO)
+ final int maxIO)
{
if (fileFactory == null)
{
@@ -257,7 +316,7 @@
{
throw new NullPointerException("fileExtension is null");
}
- if (maxAIO <= 0)
+ if (maxIO <= 0)
{
throw new IllegalStateException("maxAIO should aways be a positive number");
}
@@ -274,7 +333,9 @@
else
{
this.compactPercentage = (float)compactPercentage / 100f;
- }
+ }
+
+ log.info("creating journal with max io " + maxIO);
this.compactMinFiles = compactMinFiles;
@@ -288,7 +349,7 @@
this.fileExtension = fileExtension;
- this.maxAIO = maxAIO;
+ this.maxAIO = maxIO;
}
public Map<Long, JournalRecord> getRecords()
@@ -533,13 +594,13 @@
switch (recordType)
{
case ADD_RECORD:
- {
+ {
reader.onReadAddRecord(new RecordInfo(recordID, userRecordType, record, false));
break;
}
case UPDATE_RECORD:
- {
+ {
reader.onReadUpdateRecord(new RecordInfo(recordID, userRecordType, record, true));
break;
}
@@ -641,25 +702,33 @@
{
appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync);
}
-
- public void appendAddRecord(final long id, final byte recordType, final byte[] record, final boolean sync, final IOCompletion callback) throws Exception
+
+ public void appendAddRecord(final long id,
+ final byte recordType,
+ final byte[] record,
+ final boolean sync,
+ final IOCompletion callback) throws Exception
{
appendAddRecord(id, recordType, new ByteArrayEncoding(record), sync, callback);
}
-
+
public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) throws Exception
{
SyncIOCompletion callback = getSyncCallback(sync);
-
+
appendAddRecord(id, recordType, record, sync, callback);
-
+
if (callback != null)
{
callback.waitCompletion();
}
}
- public void appendAddRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync, final IOCompletion callback) throws Exception
+ public void appendAddRecord(final long id,
+ final byte recordType,
+ final EncodingSupport record,
+ final boolean sync,
+ final IOCompletion callback) throws Exception
{
if (LOAD_TRACE)
{
@@ -669,16 +738,16 @@
{
throw new IllegalStateException("Journal must be loaded first");
}
-
+
compactingLock.readLock().lock();
try
- {
+ {
JournalInternalRecord addRecord = new JournalAddRecord(true, id, recordType, record);
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
lockAppend.lock();
@@ -704,7 +773,11 @@
appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync);
}
- public void appendUpdateRecord(final long id, final byte recordType, final byte[] record, final boolean sync, final IOCompletion callback) throws Exception
+ public void appendUpdateRecord(final long id,
+ final byte recordType,
+ final byte[] record,
+ final boolean sync,
+ final IOCompletion callback) throws Exception
{
appendUpdateRecord(id, recordType, new ByteArrayEncoding(record), sync, callback);
}
@@ -712,16 +785,20 @@
public void appendUpdateRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync) throws Exception
{
SyncIOCompletion callback = getSyncCallback(sync);
-
+
appendUpdateRecord(id, recordType, record, sync, callback);
-
+
if (callback != null)
{
callback.waitCompletion();
}
}
-
- public void appendUpdateRecord(final long id, final byte recordType, final EncodingSupport record, final boolean sync, final IOCompletion callback) throws Exception
+
+ public void appendUpdateRecord(final long id,
+ final byte recordType,
+ final EncodingSupport record,
+ final boolean sync,
+ final IOCompletion callback) throws Exception
{
if (LOAD_TRACE)
{
@@ -731,7 +808,7 @@
{
throw new IllegalStateException("Journal must be loaded first");
}
-
+
compactingLock.readLock().lock();
try
@@ -747,12 +824,12 @@
}
JournalInternalRecord updateRecord = new JournalAddRecord(false, id, recordType, record);
-
+
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
-
+
lockAppend.lock();
try
{
@@ -780,19 +857,18 @@
}
}
-
public void appendDeleteRecord(final long id, final boolean sync) throws Exception
{
SyncIOCompletion callback = getSyncCallback(sync);
-
+
appendDeleteRecord(id, sync, callback);
-
+
if (callback != null)
{
callback.waitCompletion();
}
}
-
+
public void appendDeleteRecord(final long id, final boolean sync, final IOCompletion callback) throws Exception
{
if (LOAD_TRACE)
@@ -803,7 +879,7 @@
{
throw new IllegalStateException("Journal must be loaded first");
}
-
+
compactingLock.readLock().lock();
try
@@ -818,12 +894,12 @@
throw new IllegalStateException("Cannot find add info " + id);
}
}
-
+
JournalInternalRecord deleteRecord = new JournalDeleteRecord(id);
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
lockAppend.lock();
@@ -997,8 +1073,7 @@
{
appendDeleteRecordTransactional(txID, id, NullEncoding.instance);
}
-
-
+
public void appendPrepareRecord(long txID, byte[] transactionData, boolean sync, IOCompletion completion) throws Exception
{
appendPrepareRecord(txID, new ByteArrayEncoding(transactionData), sync, completion);
@@ -1015,9 +1090,9 @@
public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync) throws Exception
{
SyncIOCompletion syncCompletion = getSyncCallback(sync);
-
+
appendPrepareRecord(txID, transactionData, sync, syncCompletion);
-
+
if (syncCompletion != null)
{
syncCompletion.waitCompletion();
@@ -1037,7 +1112,10 @@
* @param transactionData - extra user data for the prepare
* @throws Exception
*/
- public void appendPrepareRecord(final long txID, final EncodingSupport transactionData, final boolean sync, IOCompletion callback) throws Exception
+ public void appendPrepareRecord(final long txID,
+ final EncodingSupport transactionData,
+ final boolean sync,
+ IOCompletion callback) throws Exception
{
if (LOAD_TRACE)
{
@@ -1060,7 +1138,7 @@
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
lockAppend.lock();
@@ -1081,22 +1159,19 @@
compactingLock.readLock().unlock();
}
}
-
-
-
+
public void appendCommitRecord(final long txID, final boolean sync) throws Exception
{
SyncIOCompletion syncCompletion = getSyncCallback(sync);
-
+
appendCommitRecord(txID, sync, syncCompletion);
-
+
if (syncCompletion != null)
{
syncCompletion.waitCompletion();
}
}
-
/**
* <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.
@@ -1114,7 +1189,6 @@
*
* @see JournalImpl#writeTransaction(byte, long, org.hornetq.core.journal.impl.JournalImpl.JournalTransaction, EncodingSupport)
*/
-
public void appendCommitRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception
{
@@ -1130,7 +1204,6 @@
try
{
-
if (tx == null)
{
throw new IllegalStateException("Cannot find tx with id " + txID);
@@ -1140,7 +1213,7 @@
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
lockAppend.lock();
@@ -1162,20 +1235,19 @@
}
}
-
public void appendRollbackRecord(final long txID, final boolean sync) throws Exception
{
SyncIOCompletion syncCompletion = getSyncCallback(sync);
-
+
appendRollbackRecord(txID, sync, syncCompletion);
-
+
if (syncCompletion != null)
{
syncCompletion.waitCompletion();
}
}
-
+
public void appendRollbackRecord(final long txID, final boolean sync, final IOCompletion callback) throws Exception
{
if (state != STATE_LOADED)
@@ -1195,12 +1267,12 @@
{
throw new IllegalStateException("Cannot find tx with id " + txID);
}
-
+
JournalInternalRecord rollbackRecord = new JournalRollbackRecordTX(txID);
if (callback != null)
{
- callback.lineUp();
+ callback.storeLineUp();
}
lockAppend.lock();
@@ -1276,7 +1348,7 @@
}
public void addRecord(final RecordInfo info)
- {
+ {
records.add(info);
}
@@ -1288,7 +1360,7 @@
public void deleteRecord(final long id)
{
recordsToDelete.add(id);
-
+
// Clean up when the list is too large, or it won't be possible to load large sets of files
// Done as part of JBMESSAGING-1678
if (recordsToDelete.size() == DELETE_FLUSH)
@@ -1324,7 +1396,7 @@
committedRecords.add(record);
}
}
-
+
return info;
}
@@ -2695,7 +2767,6 @@
}
/**
- * Note: You should aways guarantee locking the semaphore lock.
*
* @param completeTransaction If the appendRecord is for a prepare or commit, where we should update the number of pendingTransactions on the current file
* */
@@ -2705,109 +2776,91 @@
final JournalTransaction tx,
final IOAsyncTask parameterCallback) throws Exception
{
- try
+ if (state != STATE_LOADED)
{
- if (state != STATE_LOADED)
- {
- throw new IllegalStateException("The journal is not loaded " + state);
- }
-
- final IOAsyncTask callback;
+ throw new IllegalStateException("The journal is not loaded " + state);
+ }
- int size = encoder.getEncodeSize();
+ final IOAsyncTask callback;
- // We take into account the fileID used on the Header
- if (size > fileSize - currentFile.getFile().calculateBlockStart(SIZE_HEADER))
- {
- throw new IllegalArgumentException("Record is too large to store " + size);
- }
+ int size = encoder.getEncodeSize();
- // Disable auto flush on the timer. The Timer should'nt flush anything
- currentFile.getFile().disableAutoFlush();
+ // We take into account the fileID used on the Header
+ if (size > fileSize - currentFile.getFile().calculateBlockStart(SIZE_HEADER))
+ {
+ throw new IllegalArgumentException("Record is too large to store " + size);
+ }
+ if (!currentFile.getFile().fits(size))
+ {
+ moveNextFile(false);
+
+ // The same check needs to be done at the new file also
if (!currentFile.getFile().fits(size))
{
- currentFile.getFile().enableAutoFlush();
-
- moveNextFile(false);
-
- currentFile.getFile().disableAutoFlush();
-
- // The same check needs to be done at the new file also
- if (!currentFile.getFile().fits(size))
- {
- // Sanity check, this should never happen
- throw new IllegalStateException("Invalid logic on buffer allocation");
- }
+ // Sanity check, this should never happen
+ throw new IllegalStateException("Invalid logic on buffer allocation");
}
+ }
- if (currentFile == null)
- {
- throw new NullPointerException("Current file = null");
- }
+ if (currentFile == null)
+ {
+ throw new NullPointerException("Current file = null");
+ }
- if (tx != null)
+ if (tx != null)
+ {
+ // The callback of a transaction has to be taken inside the lock,
+ // when we guarantee the currentFile will not be changed,
+ // since we individualize the callback per file
+ if (fileFactory.isSupportsCallbacks())
{
- // The callback of a transaction has to be taken inside the lock,
- // when we guarantee the currentFile will not be changed,
- // since we individualize the callback per file
- if (fileFactory.isSupportsCallbacks())
+ // Set the delegated callback as a parameter
+ TransactionCallback txcallback = tx.getCallback(currentFile);
+ if (parameterCallback != null)
{
- // Set the delegated callback as a parameter
- TransactionCallback txcallback = tx.getCallback(currentFile);
- if (parameterCallback != null)
- {
- txcallback.setDelegateCompletion(parameterCallback);
- }
- callback = txcallback;
+ txcallback.setDelegateCompletion(parameterCallback);
}
- else
- {
- callback = null;
- }
-
- if (sync)
- {
- // In an edge case the transaction could still have pending data from previous files.
- // This shouldn't cause any blocking issues, as this is here to guarantee we cover all possibilities
- // on guaranteeing the data is on the disk
- tx.syncPreviousFiles(fileFactory.isSupportsCallbacks(), currentFile);
- }
-
- // We need to add the number of records on currentFile if prepare or commit
- if (completeTransaction)
- {
- // Filling the number of pendingTransactions at the current file
- tx.fillNumberOfRecords(currentFile, encoder);
- }
+ callback = txcallback;
}
else
{
- callback = parameterCallback;
+ callback = null;
}
- // Adding fileID
- encoder.setFileID(currentFile.getFileID());
-
- if (callback != null)
+ if (sync)
{
- currentFile.getFile().write(encoder, sync, callback);
+ // In an edge case the transaction could still have pending data from previous files.
+ // This shouldn't cause any blocking issues, as this is here to guarantee we cover all possibilities
+ // on guaranteeing the data is on the disk
+ tx.syncPreviousFiles(fileFactory.isSupportsCallbacks(), currentFile);
}
- else
+
+ // We need to add the number of records on currentFile if prepare or commit
+ if (completeTransaction)
{
- currentFile.getFile().write(encoder, sync);
+ // Filling the number of pendingTransactions at the current file
+ tx.fillNumberOfRecords(currentFile, encoder);
}
+ }
+ else
+ {
+ callback = parameterCallback;
+ }
- return currentFile;
+ // Adding fileID
+ encoder.setFileID(currentFile.getFileID());
+
+ if (callback != null)
+ {
+ currentFile.getFile().write(encoder, sync, callback);
}
- finally
+ else
{
- if (currentFile != null)
- {
- currentFile.getFile().enableAutoFlush();
- }
+ currentFile.getFile().write(encoder, sync);
}
+ return currentFile;
}
/** Get the ID part of the name */
@@ -3352,7 +3405,7 @@
return id1 < id2 ? -1 : id1 == id2 ? 0 : 1;
}
}
-
+
private class PerfBlast extends Thread
{
private final int pages;
@@ -3371,7 +3424,7 @@
lockAppend.lock();
final ByteArrayEncoding byteEncoder = new ByteArrayEncoding(new byte[128 * 1024]);
-
+
JournalInternalRecord blastRecord = new JournalInternalRecord()
{
@@ -3386,7 +3439,6 @@
byteEncoder.encode(buffer);
}
};
-
for (int i = 0; i < pages; i++)
{
@@ -3401,11 +3453,5 @@
}
}
}
-
-
-
-
-
-
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFile.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -46,21 +46,28 @@
/** The write semaphore here is only used when writing asynchronously */
private Semaphore maxIOSemaphore;
-
+
private final int defaultMaxIO;
-
+
private int maxIO;
- public NIOSequentialFile(final SequentialFileFactory factory, final String directory, final String fileName, final int maxIO, final Executor writerExecutor)
+ public NIOSequentialFile(final SequentialFileFactory factory,
+ final String directory,
+ final String fileName,
+ final int maxIO,
+ final Executor writerExecutor)
{
super(directory, new File(directory + "/" + fileName), factory, writerExecutor);
- this.defaultMaxIO = maxIO;
+ defaultMaxIO = maxIO;
}
- public NIOSequentialFile(final SequentialFileFactory factory, final File file, final int maxIO, final Executor writerExecutor)
+ public NIOSequentialFile(final SequentialFileFactory factory,
+ final File file,
+ final int maxIO,
+ final Executor writerExecutor)
{
super(file.getParent(), new File(file.getPath()), factory, writerExecutor);
- this.defaultMaxIO = maxIO;
+ defaultMaxIO = maxIO;
}
public int getAlignment()
@@ -82,7 +89,7 @@
* Some operations while initializing files on the journal may require a different maxIO */
public synchronized void open() throws Exception
{
- open(this.defaultMaxIO);
+ open(defaultMaxIO);
}
public void open(final int maxIO) throws Exception
@@ -92,10 +99,10 @@
channel = rfile.getChannel();
fileSize = channel.size();
-
+
if (writerExecutor != null)
{
- this.maxIOSemaphore = new Semaphore(maxIO);
+ maxIOSemaphore = new Semaphore(maxIO);
this.maxIO = maxIO;
}
}
@@ -130,18 +137,19 @@
}
}
+ @Override
public synchronized void close() throws Exception
{
super.close();
-
+
if (maxIOSemaphore != null)
{
while (!maxIOSemaphore.tryAcquire(maxIO, 60, TimeUnit.SECONDS))
{
- log.warn("Couldn't get lock after 60 seconds on closing AsynchronousFileImpl::" + this.getFileName());
+ log.warn("Couldn't get lock after 60 seconds on closing AsynchronousFileImpl::" + getFileName());
}
}
-
+
maxIOSemaphore = null;
if (channel != null)
@@ -213,6 +221,7 @@
}
}
+ @Override
public void position(final long pos) throws Exception
{
super.position(pos);
@@ -252,6 +261,16 @@
internalWrite(bytes, sync, null);
}
+ @Override
+ protected ByteBuffer newBuffer(int size, final int limit)
+ {
+ // For NIO, we don't need to allocate a buffer the entire size of the timed buffer, unlike AIO
+
+ size = limit;
+
+ return super.newBuffer(size, limit);
+ }
+
private void internalWrite(final ByteBuffer bytes, final boolean sync, final IOAsyncTask callback) throws Exception
{
if (!isOpen())
@@ -266,7 +285,7 @@
}
return;
}
-
+
if (writerExecutor == null)
{
doInternalWrite(bytes, sync, callback);
@@ -275,7 +294,7 @@
{
// This is a flow control on writing, just like maxAIO on libaio
maxIOSemaphore.acquire();
-
+
writerExecutor.execute(new Runnable()
{
public void run()
@@ -309,7 +328,7 @@
* @throws Exception
*/
private void doInternalWrite(final ByteBuffer bytes, final boolean sync, final IOAsyncTask callback) throws Exception
- {
+ {
position.addAndGet(bytes.limit());
channel.write(bytes);
Modified: trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -28,7 +28,7 @@
* @author <a href="mailto:clebert.suconic at jboss.com">Clebert Suconic</a>
*
*/
-public class NIOSequentialFileFactory extends AbstractSequentialFactory implements SequentialFileFactory
+public class NIOSequentialFileFactory extends AbstractSequentialFileFactory implements SequentialFileFactory
{
private static final Logger log = Logger.getLogger(NIOSequentialFileFactory.class);
@@ -37,9 +37,8 @@
{
this(journalDir,
false,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT,
- ConfigurationImpl.DEFAULT_JOURNAL_FLUSH_SYNC,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_NIO,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO,
false);
}
@@ -47,26 +46,23 @@
{
this(journalDir,
buffered,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT,
- ConfigurationImpl.DEFAULT_JOURNAL_FLUSH_SYNC,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_NIO,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO,
false);
}
public NIOSequentialFileFactory(final String journalDir,
final boolean buffered,
final int bufferSize,
- final long bufferTimeout,
- final boolean flushOnSync,
+ final int bufferTimeout,
final boolean logRates)
{
- super(journalDir, buffered, bufferSize, bufferTimeout, flushOnSync, logRates);
+ super(journalDir, buffered, bufferSize, bufferTimeout, logRates);
}
- // maxIO is ignored on NIO
public SequentialFile createSequentialFile(final String fileName, int maxIO)
{
- if (maxIO < 0)
+ if (maxIO < 1)
{
// A single threaded IO
maxIO = 1;
Modified: trunk/src/main/org/hornetq/core/journal/impl/SimpleWaitIOCallback.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/SimpleWaitIOCallback.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/SimpleWaitIOCallback.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -71,7 +71,7 @@
/* (non-Javadoc)
* @see org.hornetq.core.journal.IOCompletion#linedUp()
*/
- public void lineUp()
+ public void storeLineUp()
{
}
}
Modified: trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/SyncSpeedTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -10,15 +10,20 @@
* implied. See the License for the specific language governing
* permissions and limitations under the License.
*/
-
package org.hornetq.core.journal.impl;
import java.io.File;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
-import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import org.hornetq.core.journal.IOAsyncTask;
+import org.hornetq.core.journal.SequentialFile;
+import org.hornetq.core.journal.SequentialFileFactory;
import org.hornetq.core.logging.Logger;
/**
@@ -34,139 +39,303 @@
public class SyncSpeedTest
{
private static final Logger log = Logger.getLogger(SyncSpeedTest.class);
-
+
public static void main(final String[] args)
{
try
{
- new SyncSpeedTest().run();
+ new SyncSpeedTest().testScaleAIO();
}
catch (Exception e)
{
e.printStackTrace();
}
}
-
+
+ protected SequentialFileFactory fileFactory;
+
+ public boolean AIO = true;
+
+ protected void setupFactory()
+ {
+ if (AIO)
+ {
+ fileFactory = new AIOSequentialFileFactory(".", 0, 0, false);
+ }
+ else
+ {
+ fileFactory = new NIOSequentialFileFactory(".", false, 0, 0, false);
+ }
+ }
+
+ protected SequentialFile createSequentialFile(String fileName)
+ {
+ if (AIO)
+ {
+ return new AIOSequentialFile(fileFactory,
+ 0,
+ 0,
+ ".",
+ fileName,
+ 100000,
+ null,
+ null,
+ Executors.newSingleThreadExecutor());
+ }
+ else
+ {
+ return new NIOSequentialFile(fileFactory, new File(fileName), 1000, null);
+ }
+ }
+
+ public void run2() throws Exception
+ {
+ setupFactory();
+
+ int recordSize = 128 * 1024;
+
+ while (true)
+ {
+ System.out.println("** record size is " + recordSize);
+
+ int warmup = 500;
+
+ int its = 500;
+
+ int fileSize = (its + warmup) * recordSize;
+
+ SequentialFile file = createSequentialFile("sync-speed-test.dat");
+
+ if (file.exists())
+ {
+ file.delete();
+ }
+
+ file.open();
+
+ file.fill(0, fileSize, (byte)'X');
+
+ if (!AIO)
+ {
+ file.sync();
+ }
+
+ ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h');
+
+ long start = 0;
+
+ for (int i = 0; i < its + warmup; i++)
+ {
+ if (i == warmup)
+ {
+ start = System.currentTimeMillis();
+ }
+
+ bb1.rewind();
+
+ file.writeDirect(bb1, true);
+ }
+
+ long end = System.currentTimeMillis();
+
+ double rate = 1000 * ((double)its) / (end - start);
+
+ double throughput = recordSize * rate;
+
+ System.out.println("Rate of " + rate + " syncs per sec");
+ System.out.println("Throughput " + throughput + " bytes per sec");
+ System.out.println("*************");
+
+ recordSize *= 2;
+ }
+ }
+
public void run() throws Exception
{
- int fileSize = 1024 * 1024 * 100;
-
- int recordSize = 1024;
-
- int its = 10 * 1024;
-
- File file = new File("sync-speed-test.dat");
-
- if (file.exists())
+ int recordSize = 256;
+
+ while (true)
{
- file.delete();
+ System.out.println("** record size is " + recordSize);
+
+ int warmup = 500;
+
+ int its = 500;
+
+ int fileSize = (its + warmup) * recordSize;
+
+ File file = new File("sync-speed-test.dat");
+
+ if (file.exists())
+ {
+ file.delete();
+ }
+
+ file.createNewFile();
+
+ RandomAccessFile rfile = new RandomAccessFile(file, "rw");
+
+ FileChannel channel = rfile.getChannel();
+
+ ByteBuffer bb = generateBuffer(fileSize, (byte)'x');
+
+ write(bb, channel, fileSize);
+
+ channel.force(true);
+
+ channel.position(0);
+
+ ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h');
+
+ long start = 0;
+
+ for (int i = 0; i < its + warmup; i++)
+ {
+ if (i == warmup)
+ {
+ start = System.currentTimeMillis();
+ }
+
+ bb1.flip();
+ channel.write(bb1);
+ channel.force(false);
+ }
+
+ long end = System.currentTimeMillis();
+
+ double rate = 1000 * ((double)its) / (end - start);
+
+ double throughput = recordSize * rate;
+
+ System.out.println("Rate of " + rate + " syncs per sec");
+ System.out.println("Throughput " + throughput + " bytes per sec");
+
+ recordSize *= 2;
}
-
- RandomAccessFile rfile = new RandomAccessFile(file, "rw");
-
- FileChannel channel = rfile.getChannel();
-
- ByteBuffer bb = generateBuffer(fileSize, (byte)'x');
-
- write(bb, channel, fileSize);
-
- channel.force(false);
-
- channel.position(0);
-
- MappedByteBuffer mappedBB = channel.map(FileChannel.MapMode.READ_WRITE, 0, fileSize);
-
- mappedBB.load();
- // mappedBB.order(java.nio.ByteOrder.LITTLE_ENDIAN);
- System.out.println("isLoaded=" + mappedBB.isLoaded() + "; isDirect=" + mappedBB.isDirect() + "; byteOrder=" + mappedBB.order());
-
- ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h');
-
- System.out.println("Measuring");
-
- long start = System.currentTimeMillis();
-
- for (int i = 0; i < its; i++)
+ }
+
+ public void testScaleAIO() throws Exception
+ {
+ setupFactory();
+
+ final int recordSize = 1024;
+
+ System.out.println("** record size is " + recordSize);
+
+ final int its = 10;
+
+ for (int numThreads = 1; numThreads <= 10; numThreads++)
{
- bb1.flip();
- mappedBB.position(0);
- mappedBB.put(bb1);
- mappedBB.force();
-
- //write(bb1, channel, recordSize);
- // channel.force(false);
+
+ int fileSize = its * recordSize * numThreads;
+
+ final SequentialFile file = createSequentialFile("sync-speed-test.dat");
+
+ if (file.exists())
+ {
+ file.delete();
+ }
+
+ file.open();
+
+ file.fill(0, fileSize, (byte)'X');
+
+ if (!AIO)
+ {
+ file.sync();
+ }
+
+ final CountDownLatch latch = new CountDownLatch(its * numThreads);
+
+ class MyIOAsyncTask implements IOAsyncTask
+ {
+ public void done()
+ {
+ latch.countDown();
+ }
+
+ public void onError(int errorCode, String errorMessage)
+ {
+
+ }
+ }
+
+ final MyIOAsyncTask task = new MyIOAsyncTask();
+
+ class MyRunner implements Runnable
+ {
+ private ByteBuffer bb1;
+
+ MyRunner()
+ {
+ bb1 = generateBuffer(recordSize, (byte)'h');
+ }
+
+ public void run()
+ {
+ for (int i = 0; i < its; i++)
+ {
+ bb1.rewind();
+
+ file.writeDirect(bb1, true, task);
+// try
+// {
+// file.writeDirect(bb1, true);
+// }
+// catch (Exception e)
+// {
+// e.printStackTrace();
+// }
+ }
+ }
+ }
+
+ Set<Thread> threads = new HashSet<Thread>();
+
+ for (int i = 0; i < numThreads; i++)
+ {
+ MyRunner runner = new MyRunner();
+
+ Thread t = new Thread(runner);
+
+ threads.add(t);
+ }
+
+ long start = System.currentTimeMillis();
+
+ for (Thread t : threads)
+ {
+ log.info("starting thread");
+ t.start();
+ }
+
+ for (Thread t : threads)
+ {
+ t.join();
+ }
+
+ latch.await();
+
+ long end = System.currentTimeMillis();
+
+ double rate = 1000 * ((double)its * numThreads) / (end - start);
+
+ double throughput = recordSize * rate;
+
+ System.out.println("For " + numThreads + " threads:");
+ System.out.println("Rate of " + rate + " records per sec");
+ System.out.println("Throughput " + throughput + " bytes per sec");
+ System.out.println("*************");
}
-
- long end = System.currentTimeMillis();
-
- double rate = 1000 * ((double)its) / (end - start);
-
- System.out.println("Rate of " + rate + " syncs per sec");
- file.delete();
}
-
-// public void run() throws Exception
-// {
-// log.info("******* Starting file sync speed test *******");
-//
-// int fileSize = 1024 * 1024 * 10;
-//
-// int recordSize = 1024;
-//
-// int its = 10 * 1024;
-//
-// File file = new File("sync-speed-test.dat");
-//
-// if (file.exists())
-// {
-// file.delete();
-// }
-//
-// RandomAccessFile rfile = new RandomAccessFile(file, "rw");
-//
-// FileChannel channel = rfile.getChannel();
-//
-// ByteBuffer bb = generateBuffer(fileSize, (byte)'x');
-//
-// write(bb, channel, fileSize);
-//
-// channel.force(false);
-//
-// channel.position(0);
-//
-// ByteBuffer bb1 = generateBuffer(recordSize, (byte)'h');
-//
-// log.info("Measuring");
-//
-// long start = System.currentTimeMillis();
-//
-// for (int i = 0; i < its; i++)
-// {
-// write(bb1, channel, recordSize);
-//
-// channel.force(false);
-// }
-//
-// long end = System.currentTimeMillis();
-//
-// double rate = 1000 * ((double)its) / (end - start);
-//
-// log.info("Rate of " + rate + " syncs per sec");
-//
-// rfile.close();
-//
-// file.delete();
-//
-// log.info("****** test complete *****");
-// }
-
+
private void write(final ByteBuffer buffer, final FileChannel channel, final int size) throws Exception
{
buffer.flip();
-
+
channel.write(buffer);
}
-
+
private ByteBuffer generateBuffer(final int size, final byte ch)
{
ByteBuffer bb = ByteBuffer.allocateDirect(size);
@@ -175,7 +344,7 @@
{
bb.put(ch);
}
-
+
return bb;
- }
-}
+ }
+}
\ No newline at end of file
Modified: trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -19,17 +19,15 @@
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicLong;
import org.hornetq.core.buffers.HornetQBuffer;
import org.hornetq.core.buffers.HornetQBuffers;
import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.journal.IOAsyncTask;
-import org.hornetq.core.journal.IOCompletion;
import org.hornetq.core.journal.impl.dataformat.ByteArrayEncoding;
import org.hornetq.core.logging.Logger;
-import org.hornetq.utils.VariableLatch;
/**
* A TimedBuffer
@@ -48,9 +46,10 @@
private TimedBufferObserver bufferObserver;
- // This is used to pause and resume the timer
- // This is a reusable Latch, that uses java.util.concurrent base classes
- private final VariableLatch latchTimer = new VariableLatch();
+ // If the TimedBuffer is idle - i.e. no records are being added, then it's pointless the timer flush thread
+ // in spinning and checking the time - and using up CPU in the process - this semaphore is used to
+ // prevent that
+ private final Semaphore spinLimiter = new Semaphore(1);
private CheckTimer timerRunnable = new CheckTimer();
@@ -62,13 +61,8 @@
private List<IOAsyncTask> callbacks;
- private final Lock lock = new ReentrantReadWriteLock().writeLock();
+ private volatile int timeout;
- // used to measure inactivity. This buffer will be automatically flushed when more than timeout inactive
- private volatile boolean active = false;
-
- private final long timeout;
-
// used to measure sync requests. When a sync is requested, it shouldn't take more than timeout to happen
private volatile boolean pendingSync = false;
@@ -76,21 +70,24 @@
private volatile boolean started;
- private final boolean flushOnSync;
+ // We use this flag to prevent flush occuring between calling checkSize and addBytes
+ // CheckSize must always be followed by it's corresponding addBytes otherwise the buffer
+ // can get in an inconsistent state
+ private boolean delayFlush;
// for logging write rates
private final boolean logRates;
- private volatile long bytesFlushed;
+ private final AtomicLong bytesFlushed = new AtomicLong(0);
- private volatile long flushesDone;
+ private final AtomicLong flushesDone = new AtomicLong(0);
private Timer logRatesTimer;
private TimerTask logRatesTimerTask;
- private long lastExecution;
+ private final AtomicLong lastFlushTime = new AtomicLong(0);
// Static --------------------------------------------------------
@@ -98,23 +95,29 @@
// Public --------------------------------------------------------
- public TimedBuffer(final int size, final long timeout, final boolean flushOnSync, final boolean logRates)
+ public TimedBuffer(final int size, final int timeout, final boolean logRates)
{
- this.bufferSize = size;
+ log.info("timed buffer size " + size);
+ log.info("timed buffer timeout " + timeout);
+
+ bufferSize = size;
+
this.logRates = logRates;
+
if (logRates)
{
- this.logRatesTimer = new Timer(true);
+ logRatesTimer = new Timer(true);
}
// Setting the interval for nano-sleeps
buffer = HornetQBuffers.fixedBuffer(bufferSize);
+
buffer.clear();
+
bufferLimit = 0;
callbacks = new ArrayList<IOAsyncTask>();
- this.flushOnSync = flushOnSync;
- latchTimer.up();
+
this.timeout = timeout;
}
@@ -127,7 +130,7 @@
timerRunnable = new CheckTimer();
- timerThread = new Thread(timerRunnable, "hornetq-async-buffer");
+ timerThread = new Thread(timerRunnable, "hornetq-buffer-timeout");
timerThread.start();
@@ -148,11 +151,11 @@
return;
}
- this.flush();
+ flush();
- this.bufferObserver = null;
+ bufferObserver = null;
- latchTimer.down();
+ spinLimiter.release();
timerRunnable.close();
@@ -175,26 +178,16 @@
started = false;
}
- public synchronized void setObserver(TimedBufferObserver observer)
+ public synchronized void setObserver(final TimedBufferObserver observer)
{
- if (this.bufferObserver != null)
+ if (bufferObserver != null)
{
flush();
}
- this.bufferObserver = observer;
+ bufferObserver = observer;
}
- public void disableAutoFlush()
- {
- lock.lock();
- }
-
- public void enableAutoFlush()
- {
- lock.unlock();
- }
-
/**
* Verify if the size fits the buffer
* @param sizeChecked
@@ -210,23 +203,36 @@
if (bufferLimit == 0 || buffer.writerIndex() + sizeChecked > bufferLimit)
{
+ // Either there is not enough space left in the buffer for the sized record
+ // Or a flush has just been performed and we need to re-calcualate bufferLimit
+
flush();
- final int remaining = bufferObserver.getRemainingBytes();
+ delayFlush = true;
- if (sizeChecked > remaining)
+ final int remainingInFile = bufferObserver.getRemainingBytes();
+
+ if (sizeChecked > remainingInFile)
{
+ // Need to move to a new file -not enough space in file for this size
+
return false;
}
else
{
- buffer.clear();
- bufferLimit = Math.min(remaining, bufferSize);
+ // There is enough space in the file for this size
+
+ // Need to re-calculate buffer limit
+
+ bufferLimit = Math.min(remainingInFile, bufferSize);
+
return true;
}
}
else
{
+ delayFlush = true;
+
return true;
}
}
@@ -238,87 +244,84 @@
public synchronized void addBytes(final EncodingSupport bytes, final boolean sync, final IOAsyncTask callback)
{
+ delayFlush = false;
+
if (buffer.writerIndex() == 0)
{
- // Resume latch
- latchTimer.down();
+ // More bytes have been added so the timer flush thread can resume
+
+ spinLimiter.release();
}
bytes.encode(buffer);
callbacks.add(callback);
- active = true;
-
if (sync)
{
- if (!pendingSync)
- {
- pendingSync = true;
- }
+ pendingSync = true;
- if (flushOnSync)
- {
- flush();
- }
+// if (System.nanoTime() - lastFlushTime.get() > timeout)
+// {
+// // This might happen if there is low activity in the buffer - the timer hasn't fired because no sync records
+// // have been recently added, and suddenly a sync record is added
+// // In this case we do a flush immediately, which can reduce latency in this case
+//
+// flush();
+// }
}
- if (buffer.writerIndex() == bufferLimit)
- {
- flush();
- }
}
public void flush()
{
- ByteBuffer bufferToFlush = null;
-
- boolean useSync = false;
-
- List<IOAsyncTask> callbacksToCall = null;
-
synchronized (this)
{
- if (buffer.writerIndex() > 0)
+ if (!delayFlush && buffer.writerIndex() > 0)
{
- latchTimer.up();
-
int pos = buffer.writerIndex();
if (logRates)
{
- bytesFlushed += pos;
+ bytesFlushed.addAndGet(pos);
}
- bufferToFlush = bufferObserver.newBuffer(bufferSize, pos);
+ ByteBuffer bufferToFlush = bufferObserver.newBuffer(bufferSize, pos);
// Putting a byteArray on a native buffer is much faster, since it will do in a single native call.
// Using bufferToFlush.put(buffer) would make several append calls for each byte
bufferToFlush.put(buffer.toByteBuffer().array(), 0, pos);
- callbacksToCall = callbacks;
+ if (bufferToFlush != null)
+ {
+ bufferObserver.flushBuffer(bufferToFlush, pendingSync, callbacks);
+ }
- callbacks = new LinkedList<IOAsyncTask>();
+ try
+ {
+ // We acquire the spinLimiter semaphore - this prevents the timer flush thread unnecessarily spinning
+ // when the buffer is inactive
+ spinLimiter.acquire();
+ }
+ catch (InterruptedException e)
+ {
+ // Ignore
+ }
- useSync = pendingSync;
+ lastFlushTime.set(System.nanoTime());
- active = false;
pendingSync = false;
+ callbacks = new LinkedList<IOAsyncTask>();
+
buffer.clear();
+
bufferLimit = 0;
- flushesDone++;
+ flushesDone.incrementAndGet();
}
}
-
- // Execute the flush outside of the lock
- // This is important for NIO performance while we are using NIO Callbacks
- if (bufferToFlush != null)
- {
- bufferObserver.flushBuffer(bufferToFlush, useSync, callbacksToCall);
- }
}
// Package protected ---------------------------------------------
@@ -327,36 +330,18 @@
// Private -------------------------------------------------------
- private void checkTimer()
- {
- // if inactive for more than the timeout
- // of if a sync happened at more than the the timeout ago
- if (!active || pendingSync)
- {
- lock.lock();
- try
- {
- if (bufferObserver != null)
- {
- flush();
- }
- }
- finally
- {
- lock.unlock();
- }
- }
-
- // Set the buffer as inactive.. we will flush the buffer next tick if nothing change this
- active = false;
- }
-
// Inner classes -------------------------------------------------
private class LogRatesTimerTask extends TimerTask
{
private boolean closed;
+ private long lastExecution;
+
+ private long lastBytesFlushed;
+
+ private long lastFlushesDone;
+
@Override
public synchronized void run()
{
@@ -364,22 +349,26 @@
{
long now = System.currentTimeMillis();
+ long bytesF = bytesFlushed.get();
+ long flushesD = flushesDone.get();
+
if (lastExecution != 0)
{
- double rate = 1000 * ((double)bytesFlushed) / (now - lastExecution);
+ double rate = 1000 * (double)(bytesF - lastBytesFlushed) / (now - lastExecution);
log.info("Write rate = " + rate + " bytes / sec or " + (long)(rate / (1024 * 1024)) + " MiB / sec");
- double flushRate = 1000 * ((double)flushesDone) / (now - lastExecution);
+ double flushRate = 1000 * (double)(flushesD - lastFlushesDone) / (now - lastExecution);
log.info("Flush rate = " + flushRate + " flushes / sec");
}
lastExecution = now;
- bytesFlushed = 0;
+ lastBytesFlushed = bytesF;
- flushesDone = 0;
+ lastFlushesDone = flushesD;
}
}
+ @Override
public synchronized boolean cancel()
{
closed = true;
@@ -396,33 +385,29 @@
{
while (!closed)
{
- try
+ // We flush on the timer if there are pending syncs there and we've waited waited at least one
+ // timeout since the time of the last flush
+ // Effectively flushing "resets" the timer
+
+ if (pendingSync && bufferObserver != null && (System.nanoTime() > lastFlushTime.get() + timeout))
{
- latchTimer.waitCompletion();
+ flush();
}
- catch (InterruptedException ignored)
+
+ try
{
- }
+ spinLimiter.acquire();
- sleep();
+ Thread.yield();
- checkTimer();
-
+ spinLimiter.release();
+ }
+ catch (InterruptedException ignore)
+ {
+ }
}
}
- /**
- *
- */
- private void sleep()
- {
- long time = System.nanoTime() + timeout;
- while (time > System.nanoTime())
- {
- Thread.yield();
- }
- }
-
public void close()
{
closed = true;
Modified: trunk/src/main/org/hornetq/core/management/HornetQServerControl.java
===================================================================
--- trunk/src/main/org/hornetq/core/management/HornetQServerControl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/management/HornetQServerControl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -59,8 +59,12 @@
int getJournalMinFiles();
- int getJournalMaxAIO();
+ int getJournalMaxIO();
+ int getJournalBufferSize();
+
+ int getJournalBufferTimeout();
+
int getJournalCompactMinFiles();
int getJournalCompactPercentage();
@@ -87,10 +91,6 @@
boolean isSharedStore();
- int getAIOBufferSize();
-
- int getAIOBufferTimeout();
-
String getPagingDirectory();
boolean isPersistDeliveryCountBeforeDelivery();
Modified: trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -47,6 +47,7 @@
import org.hornetq.core.remoting.RemotingConnection;
import org.hornetq.core.remoting.server.RemotingService;
import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.server.JournalType;
import org.hornetq.core.server.ServerSession;
import org.hornetq.core.transaction.ResourceManager;
import org.hornetq.core.transaction.Transaction;
@@ -88,12 +89,12 @@
// Constructors --------------------------------------------------
public HornetQServerControlImpl(final PostOffice postOffice,
- final Configuration configuration,
- final ResourceManager resourceManager,
- final RemotingService remotingService,
- final HornetQServer messagingServer,
- final MessageCounterManager messageCounterManager,
- final NotificationBroadcasterSupport broadcaster) throws Exception
+ final Configuration configuration,
+ final ResourceManager resourceManager,
+ final RemotingService remotingService,
+ final HornetQServer messagingServer,
+ final MessageCounterManager messageCounterManager,
+ final NotificationBroadcasterSupport broadcaster) throws Exception
{
super(HornetQServerControl.class);
this.postOffice = postOffice;
@@ -128,7 +129,7 @@
{
return configuration.isBackup();
}
-
+
public boolean isSharedStore()
{
return configuration.isSharedStore();
@@ -146,19 +147,28 @@
public String[] getInterceptorClassNames()
{
- return configuration.getInterceptorClassNames().toArray(new String[configuration.getInterceptorClassNames().size()]);
+ return configuration.getInterceptorClassNames().toArray(new String[configuration.getInterceptorClassNames()
+ .size()]);
}
- public int getAIOBufferSize()
+ public int getJournalBufferSize()
{
- return configuration.getJournalBufferSize();
+ return configuration.getJournalType() == JournalType.ASYNCIO ? configuration.getJournalBufferSize_AIO()
+ : configuration.getJournalBufferSize_NIO();
}
-
- public int getAIOBufferTimeout()
+
+ public int getJournalBufferTimeout()
{
- return configuration.getJournalBufferTimeout();
+ return configuration.getJournalType() == JournalType.ASYNCIO ? configuration.getJournalBufferTimeout_AIO()
+ : configuration.getJournalBufferTimeout_NIO();
}
-
+
+ public int getJournalMaxIO()
+ {
+ return configuration.getJournalType() == JournalType.ASYNCIO ? configuration.getJournalMaxIO_AIO()
+ : configuration.getJournalMaxIO_NIO();
+ }
+
public String getJournalDirectory()
{
return configuration.getJournalDirectory();
@@ -169,16 +179,11 @@
return configuration.getJournalFileSize();
}
- public int getJournalMaxAIO()
- {
- return configuration.getJournalMaxAIO();
- }
-
public int getJournalMinFiles()
{
return configuration.getJournalMinFiles();
}
-
+
public int getJournalCompactMinFiles()
{
return configuration.getJournalCompactMinFiles();
@@ -208,7 +213,7 @@
{
return configuration.getScheduledThreadPoolMaxSize();
}
-
+
public int getThreadPoolMaxSize()
{
return configuration.getThreadPoolMaxSize();
@@ -265,7 +270,7 @@
{
server.createQueue(new SimpleString(address), new SimpleString(name), null, true, false);
}
-
+
public void createQueue(final String address, final String name, final boolean durable) throws Exception
{
server.createQueue(new SimpleString(address), new SimpleString(name), null, durable, false);
@@ -291,7 +296,7 @@
QueueControl queue = (QueueControl)queues[i];
names[i] = queue.getName();
}
-
+
return names;
}
@@ -304,7 +309,7 @@
AddressControl address = (AddressControl)addresses[i];
names[i] = address.getAddress();
}
-
+
return names;
}
@@ -319,7 +324,7 @@
{
return server.getConnectionCount();
}
-
+
public void enableMessageCounters()
{
setMessageCounterEnabled(true);
@@ -382,7 +387,6 @@
{
DateFormat dateFormat = DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.MEDIUM);
-
Map<Xid, Long> xids = resourceManager.getPreparedTransactionsWithCreationTime();
ArrayList<Entry<Xid, Long>> xidsSortedByCreationTime = new ArrayList<Map.Entry<Xid, Long>>(xids.entrySet());
Collections.sort(xidsSortedByCreationTime, new Comparator<Entry<Xid, Long>>()
@@ -403,7 +407,7 @@
}
return s;
}
-
+
public String[] listHeuristicCommittedTransactions()
{
List<Xid> xids = resourceManager.getHeuristicCommittedTransactions();
@@ -415,7 +419,7 @@
}
return s;
}
-
+
public String[] listHeuristicRolledBackTransactions()
{
List<Xid> xids = resourceManager.getHeuristicRolledbackTransactions();
@@ -505,7 +509,7 @@
{
remotingService.removeConnection(connection.getID());
connection.fail(new HornetQException(HornetQException.INTERNAL_ERROR, "connections for " + ipAddress +
- " closed by management"));
+ " closed by management"));
closed = true;
}
}
@@ -540,33 +544,33 @@
public Object[] getConnectors() throws Exception
{
Collection<TransportConfiguration> connectorConfigurations = configuration.getConnectorConfigurations().values();
-
+
Object[] ret = new Object[connectorConfigurations.size()];
-
+
int i = 0;
- for (TransportConfiguration config: connectorConfigurations)
+ for (TransportConfiguration config : connectorConfigurations)
{
Object[] tc = new Object[3];
-
+
tc[0] = config.getName();
tc[1] = config.getFactoryClassName();
tc[2] = config.getParams();
-
+
ret[i++] = tc;
}
-
+
return ret;
}
-
+
public String getConnectorsAsJSON() throws Exception
{
JSONArray array = new JSONArray();
-
- for (TransportConfiguration config: configuration.getConnectorConfigurations().values())
+
+ for (TransportConfiguration config : configuration.getConnectorConfigurations().values())
{
array.put(new JSONObject(config));
}
-
+
return array.toString();
}
Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -93,7 +93,6 @@
*/
public class JournalStorageManager implements StorageManager
{
-
private static final Logger log = Logger.getLogger(JournalStorageManager.class);
private static final long CHECKPOINT_BATCH_SIZE = Integer.MAX_VALUE;
@@ -145,7 +144,7 @@
private final SequentialFileFactory largeMessagesFactory;
private volatile boolean started;
-
+
/** Used to create Operation Contexts */
private final ExecutorFactory executorFactory;
@@ -172,7 +171,9 @@
this(config, executorFactory, null);
}
- public JournalStorageManager(final Configuration config, final ExecutorFactory executorFactory, final ReplicationManager replicator)
+ public JournalStorageManager(final Configuration config,
+ final ExecutorFactory executorFactory,
+ final ReplicationManager replicator)
{
this.executorFactory = executorFactory;
@@ -229,37 +230,32 @@
SequentialFileFactory journalFF = null;
- if (config.getJournalType() == JournalType.ASYNCIO)
+ JournalType journalTypeToUse = config.getJournalType();
+
+ if (config.getJournalType() == JournalType.ASYNCIO && !AIOSequentialFileFactory.isSupported())
{
+ log.warn("AIO wasn't located on this platform, it will fall back to using pure Java NIO. If your platform is Linux, install LibAIO to enable the AIO journal");
+
+ journalTypeToUse = JournalType.NIO;
+ }
+
+ if (journalTypeToUse == JournalType.ASYNCIO)
+ {
log.info("AIO journal selected");
- if (!AIOSequentialFileFactory.isSupported())
- {
- log.warn("AIO wasn't located on this platform, it will fall back to using pure Java NIO. If your platform is Linux, install LibAIO to enable the AIO journal");
- journalFF = new NIOSequentialFileFactory(journalDir,
- true,
- config.getJournalBufferSize(),
- config.getJournalBufferTimeout(),
- config.isJournalFlushOnSync(),
- config.isLogJournalWriteRate());
- }
- else
- {
- journalFF = new AIOSequentialFileFactory(journalDir,
- config.getJournalBufferSize(),
- config.getJournalBufferTimeout(),
- config.isJournalFlushOnSync(),
- config.isLogJournalWriteRate());
- log.info("AIO loaded successfully");
- }
+
+ journalFF = new AIOSequentialFileFactory(journalDir,
+ config.getJournalBufferSize_AIO(),
+ config.getJournalBufferTimeout_AIO(),
+ config.isLogJournalWriteRate());
+ log.info("AIO loaded successfully");
}
else if (config.getJournalType() == JournalType.NIO)
{
log.info("NIO Journal selected");
journalFF = new NIOSequentialFileFactory(journalDir,
true,
- config.getJournalBufferSize(),
- config.getJournalBufferTimeout(),
- config.isJournalFlushOnSync(),
+ config.getJournalBufferSize_NIO(),
+ config.getJournalBufferTimeout_NIO(),
config.isLogJournalWriteRate());
}
else
@@ -283,7 +279,8 @@
journalFF,
"hornetq-data",
"hq",
- config.getJournalMaxAIO());
+ config.getJournalType() == JournalType.ASYNCIO ? config.getJournalMaxIO_AIO()
+ : config.getJournalMaxIO_NIO());
if (replicator != null)
{
@@ -308,7 +305,7 @@
{
getContext().complete();
}
-
+
public void clearContext()
{
OperationContextImpl.clearContext();
@@ -319,7 +316,6 @@
return replicator != null;
}
-
public void waitOnOperations() throws Exception
{
waitOnOperations(-1);
@@ -337,8 +333,7 @@
{
waitCallback.waitCompletion();
}
- else
- if (!waitCallback.waitCompletion(timeout))
+ else if (!waitCallback.waitCompletion(timeout))
{
throw new IllegalStateException("no response received from replication");
}
@@ -383,7 +378,6 @@
// TODO: shouldn't those page methods be on the PageManager? ^^^^
-
/* (non-Javadoc)
* @see org.hornetq.core.persistence.StorageManager#getContext()
*/
@@ -391,7 +385,7 @@
{
return OperationContextImpl.getContext(executorFactory);
}
-
+
public void setContext(OperationContext context)
{
OperationContextImpl.setContext(context);
@@ -478,12 +472,12 @@
public void storeMessage(final ServerMessage message) throws Exception
{
- //TODO - how can this be less than zero?
+ // TODO - how can this be less than zero?
if (message.getMessageID() <= 0)
{
throw new HornetQException(HornetQException.ILLEGAL_STATE, "MessageId was not assigned to Message");
}
-
+
// Note that we don't sync, the add reference that comes immediately after will sync if appropriate
if (message.isLargeMessage())
@@ -491,50 +485,64 @@
messageJournal.appendAddRecord(message.getMessageID(),
ADD_LARGE_MESSAGE,
new LargeMessageEncoding((LargeServerMessage)message),
- false, getContext(false));
+ false,
+ getContext(false));
}
else
- {
+ {
messageJournal.appendAddRecord(message.getMessageID(), ADD_MESSAGE, message, false, getContext(false));
}
}
+ public void storeReference(final long queueID, final long messageID, final boolean last) throws Exception
- public void storeReference(final long queueID, final long messageID, final boolean last) throws Exception
- {
- messageJournal.appendUpdateRecord(messageID, ADD_REF, new RefEncoding(queueID), last && syncNonTransactional, getContext(syncNonTransactional));
+ {
+ messageJournal.appendUpdateRecord(messageID,
+ ADD_REF,
+ new RefEncoding(queueID),
+ last && syncNonTransactional,
+ getContext(last && syncNonTransactional));
}
public void storeAcknowledge(final long queueID, final long messageID) throws Exception
- {
- messageJournal.appendUpdateRecord(messageID, ACKNOWLEDGE_REF, new RefEncoding(queueID), syncNonTransactional, getContext(syncNonTransactional));
+ {
+ messageJournal.appendUpdateRecord(messageID,
+ ACKNOWLEDGE_REF,
+ new RefEncoding(queueID),
+ syncNonTransactional,
+ getContext(syncNonTransactional));
}
public void deleteMessage(final long messageID) throws Exception
- {
+ {
messageJournal.appendDeleteRecord(messageID, syncNonTransactional, getContext(syncNonTransactional));
}
public void updateScheduledDeliveryTime(final MessageReference ref) throws Exception
- {
+ {
ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), ref.getQueue()
.getID());
messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(),
SET_SCHEDULED_DELIVERY_TIME,
encoding,
- syncNonTransactional, getContext(syncNonTransactional));
+ syncNonTransactional,
+ getContext(syncNonTransactional));
}
public void storeDuplicateID(final SimpleString address, final byte[] duplID, final long recordID) throws Exception
- {
+ {
DuplicateIDEncoding encoding = new DuplicateIDEncoding(address, duplID);
- messageJournal.appendAddRecord(recordID, DUPLICATE_ID, encoding, syncNonTransactional, getContext(syncNonTransactional));
+ messageJournal.appendAddRecord(recordID,
+ DUPLICATE_ID,
+ encoding,
+ syncNonTransactional,
+ getContext(syncNonTransactional));
}
public void deleteDuplicateID(long recordID) throws Exception
- {
+ {
messageJournal.appendDeleteRecord(recordID, syncNonTransactional, getContext(syncNonTransactional));
}
@@ -591,7 +599,12 @@
public long storeHeuristicCompletion(Xid xid, boolean isCommit) throws Exception
{
long id = generateUniqueID();
- messageJournal.appendAddRecord(id, HEURISTIC_COMPLETION, new HeuristicCompletionEncoding(xid, isCommit), true, getContext(true));
+
+ messageJournal.appendAddRecord(id,
+ HEURISTIC_COMPLETION,
+ new HeuristicCompletionEncoding(xid, isCommit),
+ true,
+ getContext(true));
return id;
}
@@ -668,10 +681,10 @@
DeliveryCountUpdateEncoding updateInfo = new DeliveryCountUpdateEncoding(ref.getQueue().getID(),
ref.getDeliveryCount());
- messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(),
- UPDATE_DELIVERY_COUNT,
- updateInfo,
- syncNonTransactional, getContext(syncNonTransactional));
+ messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), UPDATE_DELIVERY_COUNT, updateInfo,
+
+ syncNonTransactional, getContext(syncNonTransactional));
+
}
private static final class AddMessageRecord
@@ -738,7 +751,7 @@
JournalLoadInformation info = messageJournal.load(records,
preparedTransactions,
new LargeMessageTXFailureCallback(messages));
-
+
ArrayList<LargeServerMessage> largeMessages = new ArrayList<LargeServerMessage>();
Map<Long, Map<Long, AddMessageRecord>> queueMap = new HashMap<Long, Map<Long, AddMessageRecord>>();
@@ -764,7 +777,7 @@
break;
}
case ADD_MESSAGE:
- {
+ {
ServerMessage message = new ServerMessageImpl(record.id, 50);
message.decode(buff);
@@ -967,6 +980,11 @@
messageJournal.perfBlast(perfBlastPages);
}
+ if (System.getProperty("org.hornetq.opt.directblast") != null)
+ {
+ messageJournal.runDirectJournalBlast();
+ }
+
return info;
}
@@ -1355,7 +1373,7 @@
return info;
}
-
+
// Public -----------------------------------------------------------------------------------
public Journal getMessageJournal()
@@ -1416,7 +1434,7 @@
}
// Private ----------------------------------------------------------------------------------
-
+
private void checkAndCreateDir(final String dir, final boolean create)
{
File f = new File(dir);
@@ -1464,18 +1482,15 @@
return DummyOperationContext.getInstance();
}
}
-
-
// Inner Classes
// ----------------------------------------------------------------------------
-
static class DummyOperationContext implements OperationContext
{
-
+
private static DummyOperationContext instance = new DummyOperationContext();
-
+
public static OperationContext getInstance()
{
return instance;
@@ -1512,7 +1527,7 @@
/* (non-Javadoc)
* @see org.hornetq.core.journal.IOCompletion#lineUp()
*/
- public void lineUp()
+ public void storeLineUp()
{
}
@@ -1529,9 +1544,9 @@
public void onError(int errorCode, String errorMessage)
{
}
-
+
}
-
+
private static class XidEncoding implements EncodingSupport
{
final Xid xid;
@@ -2018,5 +2033,4 @@
}
-
}
Modified: trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/persistence/impl/journal/OperationContextImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -20,6 +20,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import org.hornetq.core.journal.IOAsyncTask;
+import org.hornetq.core.logging.Logger;
import org.hornetq.core.persistence.OperationContext;
import org.hornetq.utils.ExecutorFactory;
@@ -38,6 +39,7 @@
*/
public class OperationContextImpl implements OperationContext
{
+ private static final Logger log = Logger.getLogger(OperationContextImpl.class);
private static final ThreadLocal<OperationContext> threadLocalContext = new ThreadLocal<OperationContext>();
@@ -61,8 +63,7 @@
{
threadLocalContext.set(context);
}
-
-
+
private List<TaskHolder> tasks;
private volatile int storeLineUp = 0;
@@ -90,10 +91,9 @@
super();
this.executor = executor;
}
-
- /** To be called by the replication manager, when new replication is added to the queue */
- public void lineUp()
- {
+
+ public void storeLineUp()
+ {
storeLineUp++;
}
@@ -108,7 +108,6 @@
checkTasks();
}
- /** You may have several actions to be done after a replication operation is completed. */
public void executeOnCompletion(final IOAsyncTask completion)
{
if (errorCode != -1)
@@ -159,7 +158,6 @@
}
- /** To be called by the storage manager, when data is confirmed on the channel */
public synchronized void done()
{
stored++;
Modified: trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java
===================================================================
--- trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/replication/impl/ReplicatedJournal.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -519,6 +519,12 @@
return localJournal.getNumberOfRecords();
}
+ public void runDirectJournalBlast() throws Exception
+ {
+ // TODO Auto-generated method stub
+
+ }
+
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
Modified: trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/server/cluster/impl/BridgeImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -695,12 +695,24 @@
}
catch (HornetQException e)
{
+ csf.close();
+
// the session was created while its server was starting, retry it:
if (e.getCode() == HornetQException.SESSION_CREATION_REJECTED)
{
log.warn("Server is starting, retry to create the session for bridge " + name);
+ //Sleep a little to prevent spinning too much
+ try
+ {
+ Thread.sleep(10);
+ }
+ catch (InterruptedException ignore)
+ {
+ }
+
retry = true;
+
continue;
}
else
Modified: trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -22,6 +22,7 @@
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -52,7 +53,10 @@
import org.hornetq.core.exception.HornetQException;
import org.hornetq.core.filter.Filter;
import org.hornetq.core.filter.impl.FilterImpl;
+import org.hornetq.core.journal.EncodingSupport;
import org.hornetq.core.journal.IOAsyncTask;
+import org.hornetq.core.journal.IOCompletion;
+import org.hornetq.core.journal.Journal;
import org.hornetq.core.journal.JournalLoadInformation;
import org.hornetq.core.journal.impl.SyncSpeedTest;
import org.hornetq.core.logging.LogDelegateFactory;
@@ -90,11 +94,16 @@
import org.hornetq.core.security.SecurityStore;
import org.hornetq.core.security.impl.SecurityStoreImpl;
import org.hornetq.core.server.ActivateCallback;
+import org.hornetq.core.server.Consumer;
import org.hornetq.core.server.Divert;
+import org.hornetq.core.server.HandleStatus;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.MemoryManager;
+import org.hornetq.core.server.MessageReference;
import org.hornetq.core.server.Queue;
import org.hornetq.core.server.QueueFactory;
+import org.hornetq.core.server.RoutingContext;
+import org.hornetq.core.server.ServerConsumer;
import org.hornetq.core.server.ServerMessage;
import org.hornetq.core.server.ServerSession;
import org.hornetq.core.server.cluster.ClusterManager;
@@ -630,7 +639,7 @@
throw new HornetQException(HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS,
"Server and client versions incompatible");
}
-
+
if (!checkActivate())
{
// Backup server is not ready to accept connections
@@ -970,7 +979,7 @@
if (replicationEndpoint == null)
{
log.warn("There is no replication endpoint, can't activate this backup server");
-
+
throw new HornetQException(HornetQException.INTERNAL_ERROR, "Can't activate the server");
}
@@ -1189,6 +1198,8 @@
}
initialised = true;
+
+ log.info("********** initialised");
if (System.getProperty("org.hornetq.opt.routeblast") != null)
{
@@ -1482,14 +1493,50 @@
}
}
+
+// private void runRouteBlastNoWait() throws Exception
+// {
+// SimpleString address = new SimpleString("rbnw_address");
+// SimpleString queueName = new SimpleString("rbnw_name");
+//
+// createQueue(address, queueName, null, true, false, true);
+//
+// Queue queue = (Queue)postOffice.getBinding(queueName).getBindable();
+//
+// RBConsumer consumer = new RBConsumer(queue);
+//
+// queue.addConsumer(consumer);
+//
+// final int bodySize = 1024;
+//
+// byte[] body = new byte[bodySize];
+//
+// final int numMessages = 10000000;
+//
+// for (int i = 0; i < numMessages; i++)
+// {
+// final ServerMessage msg = new ServerMessageImpl(storageManager.generateUniqueID(), 1500);
+//
+// msg.getBodyBuffer().writeBytes(body);
+//
+// msg.setDestination(address);
+//
+// msg.setDurable(true);
+//
+// postOffice.route(msg);
+// }
+// }
+
private LinkedBlockingQueue<RouteBlastRunner> available = new LinkedBlockingQueue<RouteBlastRunner>();
+
private void runRouteBlast() throws Exception
{
log.info("*** running route blast");
- final int numThreads = 2;
+
+ final int numThreads = 1;
- final int numClients = 200;
+ final int numClients = 1000;
for (int i = 0; i < numClients; i++)
{
@@ -1499,7 +1546,7 @@
available.add(run);
}
-
+
log.info("setup, now running");
Set<Thread> runners = new HashSet<Thread>();
@@ -1518,43 +1565,39 @@
t.join();
}
}
-
- class Foo implements Runnable
- {
- public void run()
- {
- for (int i = 0; i < 1000000; i++)
- {
- try
- {
- RouteBlastRunner runner = available.take();
-
- runner.run();
- }
- catch (InterruptedException e)
- {
- log.error("Interrupted", e);
- }
- }
- }
- }
-
+
class RouteBlastRunner implements Runnable
{
private SimpleString address;
+ private Set<Consumer> consumers = new HashSet<Consumer>();
+
RouteBlastRunner(SimpleString address)
{
this.address = address;
}
+
+
public void setup() throws Exception
{
final int numQueues = 1;
for (int i = 0; i < numQueues; i++)
{
- createQueue(address, new SimpleString(address + ".hq.route_blast_queue" + i), null, true, false, true);
+ SimpleString queueName = new SimpleString(address + ".hq.route_blast_queue" + i);
+
+ createQueue(address, queueName, null, true, false, true);
+
+ Queue queue = (Queue)postOffice.getBinding(queueName).getBindable();
+
+ RBConsumer consumer = new RBConsumer(queue);
+
+ queue.addConsumer(consumer);
+
+ //log.info("added consumer to queue " + queue);
+
+ consumers.add(consumer);
}
}
@@ -1573,7 +1616,7 @@
msg.setDestination(address);
msg.setDurable(true);
-
+
postOffice.route(msg);
storageManager.afterCompleteOperations(new IOAsyncTask()
@@ -1584,7 +1627,7 @@
}
public void done()
- {
+ {
available.add(RouteBlastRunner.this);
}
});
@@ -1596,7 +1639,59 @@
}
}
+
+
+
+
+ class Foo implements Runnable
+ {
+ public void run()
+ {
+ for (int i = 0; i < 1000000; i++)
+ {
+ try
+ {
+ RouteBlastRunner runner = available.take();
+ runner.run();
+ }
+ catch (InterruptedException e)
+ {
+ log.error("Interrupted", e);
+ }
+ }
+ }
+ }
+
+ private class RBConsumer implements Consumer
+ {
+ private Queue queue;
+
+ RBConsumer(Queue queue)
+ {
+ this.queue = queue;
+ }
+
+ public Filter getFilter()
+ {
+ return null;
+ }
+
+ public HandleStatus handle(MessageReference reference) throws Exception
+ {
+ reference.handled();
+
+ queue.acknowledge(reference);
+
+ //log.info("acking");
+
+ return HandleStatus.HANDLED;
+ }
+
+ }
+
+
+
// Inner classes
// --------------------------------------------------------------------------------
}
Modified: trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/server/impl/QueueImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -1227,7 +1227,7 @@
}
private synchronized boolean directDeliver(final MessageReference reference)
- {
+ {
if (paused || handlers.isEmpty())
{
return false;
@@ -1241,7 +1241,7 @@
MessageHandler handler = getHandlerRoundRobin();
Consumer consumer = handler.getConsumer();
-
+
if (!checkExpired(reference))
{
SimpleString groupID = reference.getMessage().getSimpleStringProperty(MessageImpl.HDR_GROUP_ID);
@@ -1303,7 +1303,7 @@
}
protected synchronized void add(final MessageReference ref, final boolean first)
- {
+ {
if (dontAdd)
{
return;
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerMessageImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -61,7 +61,9 @@
*/
public ServerMessageImpl(final long messageID, final int initialMessageBufferSize)
{
- super(messageID, initialMessageBufferSize);
+ super(initialMessageBufferSize);
+
+ this.messageID = messageID;
}
protected ServerMessageImpl(final int initialMessageBufferSize)
Modified: trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -186,7 +186,7 @@
private final HornetQServer server;
private final SimpleString managementAddress;
-
+
// The current currentLargeMessage being processed
private volatile LargeServerMessage currentLargeMessage;
@@ -241,7 +241,7 @@
this.resourceManager = resourceManager;
this.securityStore = securityStore;
-
+
this.executor = executor;
if (!xa)
@@ -1455,52 +1455,11 @@
sendResponse(packet, null, false, false);
}
}
-
-
-
- public void handleSend2(final ServerMessage message)
- {
- try
- {
- long id = storageManager.generateUniqueID();
- message.setMessageID(id);
- message.encodeMessageIDToBuffer();
-
- if (message.getDestination().equals(managementAddress))
- {
- // It's a management message
-
- handleManagementMessage(message);
- }
- else
- {
- send(message);
- }
- }
- catch (Exception e)
- {
- log.error("Failed to send message", e);
-
- }
- finally
- {
- try
- {
- releaseOutStanding(message, message.getEncodeSize());
- }
- catch (Exception e)
- {
- log.error("Failed to release outstanding credits", e);
- }
- }
-
- }
-
public void handleSend(final SessionSendMessage packet)
{
Packet response = null;
-
+
ServerMessage message = (ServerMessage)packet.getMessage();
try
@@ -1553,8 +1512,8 @@
log.error("Failed to release outstanding credits", e);
}
}
-
- sendResponse(packet, response, false, false);
+
+ sendResponse(packet, response, false, false);
}
public void handleSendContinuations(final SessionSendContinuationMessage packet)
@@ -1638,7 +1597,7 @@
}
}
});
-
+
if (gotCredits > 0)
{
sendProducerCredits(holder, gotCredits, address);
@@ -1752,30 +1711,26 @@
// Private
// ----------------------------------------------------------------------------
- /**
- * Respond to client after replication
- * @param packet
- * @param response
- */
private void sendResponse(final Packet confirmPacket,
final Packet response,
final boolean flush,
final boolean closeChannel)
{
storageManager.afterCompleteOperations(new IOAsyncTask()
- {
+ {
public void onError(int errorCode, String errorMessage)
{
log.warn("Error processing IOCallback code = " + errorCode + " message = " + errorMessage);
- HornetQExceptionMessage exceptionMessage = new HornetQExceptionMessage(new HornetQException(errorCode, errorMessage));
-
- doSendResponse(confirmPacket, exceptionMessage, flush, closeChannel);
+ HornetQExceptionMessage exceptionMessage = new HornetQExceptionMessage(new HornetQException(errorCode,
+ errorMessage));
+
+ doConfirmAndResponse(confirmPacket, exceptionMessage, flush, closeChannel);
}
public void done()
{
- doSendResponse(confirmPacket, response, flush, closeChannel);
+ doConfirmAndResponse(confirmPacket, response, flush, closeChannel);
}
});
}
@@ -1786,11 +1741,11 @@
* @param flush
* @param closeChannel
*/
- private void doSendResponse(final Packet confirmPacket,
- final Packet response,
- final boolean flush,
- final boolean closeChannel)
- {
+ private void doConfirmAndResponse(final Packet confirmPacket,
+ final Packet response,
+ final boolean flush,
+ final boolean closeChannel)
+ {
if (confirmPacket != null)
{
channel.confirm(confirmPacket);
@@ -1802,7 +1757,7 @@
}
if (response != null)
- {
+ {
channel.send(response);
}
Modified: trunk/src/main/org/hornetq/utils/TokenBucketLimiterImpl.java
===================================================================
--- trunk/src/main/org/hornetq/utils/TokenBucketLimiterImpl.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/src/main/org/hornetq/utils/TokenBucketLimiterImpl.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -53,17 +53,21 @@
{
while (!check())
{
- if (!spin)
+ if (spin)
{
- try
- {
- Thread.sleep(1);
- }
- catch (Exception e)
- {
- //Ignore
- }
+ Thread.yield();
}
+ else
+ {
+ try
+ {
+ Thread.sleep(1);
+ }
+ catch (Exception e)
+ {
+ //Ignore
+ }
+ }
}
}
Modified: trunk/tests/config/ConfigurationTest-full-config.xml
===================================================================
--- trunk/tests/config/ConfigurationTest-full-config.xml 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/config/ConfigurationTest-full-config.xml 2009-12-01 20:18:47 UTC (rev 8483)
@@ -38,8 +38,7 @@
<create-journal-dir>false</create-journal-dir>
<journal-type>NIO</journal-type>
<journal-compact-min-files>123</journal-compact-min-files>
- <journal-compact-percentage>33</journal-compact-percentage>
- <journal-flush-on-sync>true</journal-flush-on-sync>
+ <journal-compact-percentage>33</journal-compact-percentage>
<journal-buffer-timeout>1000</journal-buffer-timeout>
<journal-buffer-size>10000</journal-buffer-size>
<journal-sync-transactional>false</journal-sync-transactional>
Modified: trunk/tests/src/org/hornetq/tests/integration/client/BlockingSendTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/client/BlockingSendTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/client/BlockingSendTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -51,7 +51,7 @@
{
server.getConfiguration().setJournalSyncNonTransactional(false);
- server.getConfiguration().setJournalBufferTimeout(15000);
+ server.getConfiguration().setJournalBufferTimeout_AIO(15000);
server.start();
Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/cluster/distribution/ClusterTestBase.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -1138,10 +1138,9 @@
configuration.setSecurityEnabled(false);
configuration.setJournalMinFiles(2);
- configuration.setJournalMaxAIO(1000);
+ configuration.setJournalMaxIO_AIO(1000);
configuration.setJournalFileSize(100 * 1024);
configuration.setJournalType(JournalType.ASYNCIO);
- configuration.setJournalMaxAIO(1000);
configuration.setSharedStore(sharedStorage);
if (sharedStorage)
{
@@ -1254,7 +1253,7 @@
configuration.setJournalDirectory(getJournalDir(node, false));
configuration.setJournalFileSize(100 * 1024);
configuration.setJournalType(JournalType.ASYNCIO);
- configuration.setJournalMaxAIO(1000);
+ configuration.setJournalMaxIO_AIO(1000);
configuration.setPagingDirectory(getPageDir(node, false));
configuration.setLargeMessagesDirectory(getLargeMessagesDir(node, false));
configuration.setClustered(true);
Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverReplicationTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverReplicationTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverReplicationTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -42,11 +42,10 @@
configuration.setSecurityEnabled(false);
configuration.setBindingsDirectory(getBindingsDir(node, false));
configuration.setJournalMinFiles(2);
- configuration.setJournalMaxAIO(1000);
+ configuration.setJournalMaxIO_AIO(1000);
configuration.setJournalDirectory(getJournalDir(node, false));
configuration.setJournalFileSize(100 * 1024);
configuration.setJournalType(JournalType.ASYNCIO);
- configuration.setJournalMaxAIO(1000);
configuration.setPagingDirectory(getPageDir(node, false));
configuration.setLargeMessagesDirectory(getLargeMessagesDir(node, false));
configuration.setClustered(true);
Modified: trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/cluster/failover/GroupingFailoverSharedServerTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -12,18 +12,15 @@
*/
package org.hornetq.tests.integration.cluster.failover;
-import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration;
-import org.hornetq.core.server.JournalType;
-import org.hornetq.core.server.HornetQServer;
-import org.hornetq.core.server.HornetQ;
-import org.hornetq.core.message.impl.MessageImpl;
+import java.util.Map;
+
import org.hornetq.core.config.Configuration;
import org.hornetq.core.config.TransportConfiguration;
import org.hornetq.core.config.impl.ConfigurationImpl;
-import org.hornetq.utils.SimpleString;
+import org.hornetq.core.server.HornetQ;
+import org.hornetq.core.server.HornetQServer;
+import org.hornetq.core.server.JournalType;
-import java.util.Map;
-
/**
* @author <a href="mailto:andy.taylor at jboss.org">Andy Taylor</a>
* Created Oct 26, 2009
@@ -42,11 +39,10 @@
configuration.setSecurityEnabled(false);
configuration.setBindingsDirectory(getBindingsDir(backupNode, false));
configuration.setJournalMinFiles(2);
- configuration.setJournalMaxAIO(1000);
+ configuration.setJournalMaxIO_AIO(1000);
configuration.setJournalDirectory(getJournalDir(backupNode, false));
configuration.setJournalFileSize(100 * 1024);
configuration.setJournalType(JournalType.ASYNCIO);
- configuration.setJournalMaxAIO(1000);
configuration.setPagingDirectory(getPageDir(backupNode, false));
configuration.setLargeMessagesDirectory(getLargeMessagesDir(backupNode, false));
configuration.setClustered(true);
Modified: trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/jms/cluster/JMSFailoverTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -16,6 +16,7 @@
import java.util.HashMap;
import java.util.Map;
+import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.ExceptionListener;
@@ -41,6 +42,7 @@
import org.hornetq.jms.HornetQQueue;
import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.jms.client.HornetQSession;
+import org.hornetq.tests.util.RandomUtil;
import org.hornetq.tests.util.UnitTestCase;
import org.hornetq.utils.SimpleString;
@@ -122,11 +124,15 @@
MessageConsumer consumer = sess.createConsumer(queue);
+ byte[] body = RandomUtil.randomBytes(bodySize);
+
for (int i = 0; i < numMessages; i++)
{
- TextMessage tm = sess.createTextMessage("message" + i);
+ BytesMessage bm = sess.createBytesMessage();
+
+ bm.writeBytes(body);
- producer.send(tm);
+ producer.send(bm);
}
conn.start();
@@ -143,11 +149,11 @@
{
log.info("got message " + i);
- TextMessage tm = (TextMessage)consumer.receive(1000);
+ BytesMessage bm = (BytesMessage)consumer.receive(1000);
- assertNotNull(tm);
+ assertNotNull(bm);
- assertEquals("message" + i, tm.getText());
+ assertEquals(body.length, bm.getBodyLength());
}
TextMessage tm = (TextMessage)consumer.receiveNoWait();
Modified: trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalCompactTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalCompactTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalCompactTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -30,15 +30,12 @@
*
*/
public class AIOJournalCompactTest extends NIOJournalCompactTest
-{
-
-
+{
public static TestSuite suite()
{
return createAIOTestSuite(AIOJournalCompactTest.class);
}
-
// Constants -----------------------------------------------------
// Attributes ----------------------------------------------------
@@ -63,9 +60,8 @@
file.mkdir();
return new AIOSequentialFileFactory(getTestDir(),
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- 1000000,
- true,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO,
+ 1000000,
false
);
}
Modified: trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalImplTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalImplTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/journal/AIOJournalImplTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -70,9 +70,8 @@
file.mkdir();
return new AIOSequentialFileFactory(getTestDir(),
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- 1000000,
- true,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO,
+ 1000000,
false
);
}
Modified: trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/journal/ValidateTransactionHealthTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -388,9 +388,8 @@
if (factoryType.equals("aio"))
{
return new AIOSequentialFileFactory(directory,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE,
- ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT,
- ConfigurationImpl.DEFAULT_JOURNAL_FLUSH_SYNC,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO,
+ ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO,
false);
}
else
Modified: trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -106,9 +106,9 @@
assertEquals(conf.isJournalSyncNonTransactional(), serverControl.isJournalSyncNonTransactional());
assertEquals(conf.getJournalFileSize(), serverControl.getJournalFileSize());
assertEquals(conf.getJournalMinFiles(), serverControl.getJournalMinFiles());
- assertEquals(conf.getJournalMaxAIO(), serverControl.getJournalMaxAIO());
- assertEquals(conf.getJournalBufferSize(), serverControl.getAIOBufferSize());
- assertEquals(conf.getJournalBufferTimeout(), serverControl.getAIOBufferTimeout());
+ assertEquals(conf.getJournalMaxIO_AIO(), serverControl.getJournalMaxIO());
+ assertEquals(conf.getJournalBufferSize_AIO(), serverControl.getJournalBufferSize());
+ assertEquals(conf.getJournalBufferTimeout_AIO(), serverControl.getJournalBufferTimeout());
assertEquals(conf.isCreateBindingsDir(), serverControl.isCreateBindingsDir());
assertEquals(conf.isCreateJournalDir(), serverControl.isCreateJournalDir());
assertEquals(conf.getPagingDirectory(), serverControl.getPagingDirectory());
Modified: trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/management/HornetQServerControlUsingCoreTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -203,9 +203,9 @@
return (Integer)proxy.retrieveAttributeValue("journalFileSize");
}
- public int getJournalMaxAIO()
+ public int getJournalMaxIO()
{
- return (Integer)proxy.retrieveAttributeValue("journalMaxAIO");
+ return (Integer)proxy.retrieveAttributeValue("journalMaxIO");
}
public int getJournalMinFiles()
@@ -423,14 +423,14 @@
proxy.invokeOperation("setMessageCounterSamplePeriod", newPeriod);
}
- public int getAIOBufferSize()
+ public int getJournalBufferSize()
{
- return (Integer)proxy.retrieveAttributeValue("AIOBufferSize");
+ return (Integer)proxy.retrieveAttributeValue("JournalBufferSize");
}
- public int getAIOBufferTimeout()
+ public int getJournalBufferTimeout()
{
- return (Integer)proxy.retrieveAttributeValue("AIOBufferTimeout");
+ return (Integer)proxy.retrieveAttributeValue("JournalBufferTimeout");
}
public int getJournalCompactMinFiles()
Modified: trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -418,7 +418,7 @@
{
OperationContext ctx = OperationContextImpl.getContext(factory);
- ctx.lineUp();
+ ctx.storeLineUp();
String msg = "I'm an exception";
@@ -1076,5 +1076,11 @@
{
}
+ public void runDirectJournalBlast() throws Exception
+ {
+ // TODO Auto-generated method stub
+
+ }
+
}
}
Modified: trunk/tests/src/org/hornetq/tests/opt/SendTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/opt/SendTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/opt/SendTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -74,9 +74,10 @@
{
log.info("*** Starting server");
- System.setProperty("org.hornetq.opt.dontadd", "true");
- System.setProperty("org.hornetq.opt.routeblast", "true");
+ //System.setProperty("org.hornetq.opt.dontadd", "true");
+ // System.setProperty("org.hornetq.opt.routeblast", "true");
//System.setProperty("org.hornetq.opt.generatemessages", "true");
+ System.setProperty("org.hornetq.opt.directblast", "true");
Configuration configuration = new ConfigurationImpl();
configuration.setSecurityEnabled(false);
@@ -91,9 +92,19 @@
configuration.setJournalType(JournalType.ASYNCIO);
+ configuration.setJournalBufferTimeout_NIO(1000000000 / 100); // this is in nanoseconds
+ configuration.setJournalBufferSize_NIO(490 * 1024);
+ configuration.setJournalMaxIO_NIO(1);
+
+ configuration.setJournalBufferTimeout_AIO(1000000000 / 1000); // this is in nanoseconds
+ configuration.setJournalBufferSize_AIO(490 * 1024);
+ configuration.setJournalMaxIO_AIO(500);
+
configuration.setLogJournalWriteRate(true);
- //configuration.setRunSyncSpeedTest(true);
+
+ // configuration.setRunSyncSpeedTest(true);
+
Map<String, Object> params = new HashMap<String, Object>();
params.put(TransportConstants.USE_NIO_PROP_NAME, Boolean.FALSE);
Modified: trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/unit/core/config/impl/ConfigurationImplTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -59,7 +59,8 @@
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_SYNC_NON_TRANSACTIONAL, conf.isJournalSyncNonTransactional());
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_FILE_SIZE, conf.getJournalFileSize());
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MIN_FILES, conf.getJournalMinFiles());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_AIO, conf.getJournalMaxAIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_IO_AIO, conf.getJournalMaxIO_AIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_IO_NIO, conf.getJournalMaxIO_NIO());
assertEquals(ConfigurationImpl.DEFAULT_WILDCARD_ROUTING_ENABLED, conf.isWildcardRoutingEnabled());
assertEquals(ConfigurationImpl.DEFAULT_TRANSACTION_TIMEOUT, conf.getTransactionTimeout());
assertEquals(ConfigurationImpl.DEFAULT_MESSAGE_EXPIRY_SCAN_PERIOD, conf.getMessageExpiryScanPeriod()); // OK
@@ -82,8 +83,10 @@
assertEquals(ConfigurationImpl.DEFAULT_PAGING_DIR, conf.getPagingDirectory());
assertEquals(ConfigurationImpl.DEFAULT_LARGE_MESSAGES_DIR, conf.getLargeMessagesDirectory());
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_COMPACT_PERCENTAGE, conf.getJournalCompactPercentage());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_FLUSH_SYNC, conf.isJournalFlushOnSync());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT, conf.getJournalBufferTimeout());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, conf.getJournalBufferTimeout_AIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, conf.getJournalBufferTimeout_NIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, conf.getJournalBufferSize_AIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, conf.getJournalBufferSize_NIO());
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_LOG_WRITE_RATE, conf.isLogJournalWriteRate());
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_PERF_BLAST_PAGES, conf.getJournalPerfBlastPages());
assertEquals(ConfigurationImpl.DEFAULT_MESSAGE_COUNTER_ENABLED, conf.isMessageCounterEnabled());
@@ -164,8 +167,12 @@
assertEquals(i, conf.getJournalMinFiles());
i = randomInt();
- conf.setJournalMaxAIO(i);
- assertEquals(i, conf.getJournalMaxAIO());
+ conf.setJournalMaxIO_AIO(i);
+ assertEquals(i, conf.getJournalMaxIO_AIO());
+
+ i = randomInt();
+ conf.setJournalMaxIO_NIO(i);
+ assertEquals(i, conf.getJournalMaxIO_NIO());
s = randomString();
conf.setManagementAddress(new SimpleString(s));
@@ -244,16 +251,20 @@
assertEquals(i, conf.getJournalCompactPercentage());
i = randomInt();
- conf.setJournalBufferSize(i);
- assertEquals(i, conf.getJournalBufferSize());
+ conf.setJournalBufferSize_AIO(i);
+ assertEquals(i, conf.getJournalBufferSize_AIO());
i = randomInt();
- conf.setJournalBufferTimeout(i);
- assertEquals(i, conf.getJournalBufferTimeout());
+ conf.setJournalBufferTimeout_AIO(i);
+ assertEquals(i, conf.getJournalBufferTimeout_AIO());
+
+ i = randomInt();
+ conf.setJournalBufferSize_NIO(i);
+ assertEquals(i, conf.getJournalBufferSize_NIO());
- b = randomBoolean();
- conf.setJournalFlushOnSync(b);
- assertEquals(b, conf.isJournalFlushOnSync());
+ i = randomInt();
+ conf.setJournalBufferTimeout_NIO(i);
+ assertEquals(i, conf.getJournalBufferTimeout_NIO());
b = randomBoolean();
conf.setLogJournalWriteRate(b);
@@ -381,8 +392,12 @@
assertEquals(i, conf.getJournalMinFiles());
i = randomInt();
- conf.setJournalMaxAIO(i);
- assertEquals(i, conf.getJournalMaxAIO());
+ conf.setJournalMaxIO_AIO(i);
+ assertEquals(i, conf.getJournalMaxIO_AIO());
+
+ i = randomInt();
+ conf.setJournalMaxIO_NIO(i);
+ assertEquals(i, conf.getJournalMaxIO_NIO());
s = randomString();
conf.setManagementAddress(new SimpleString(s));
@@ -461,16 +476,20 @@
assertEquals(i, conf.getJournalCompactPercentage());
i = randomInt();
- conf.setJournalBufferSize(i);
- assertEquals(i, conf.getJournalBufferSize());
+ conf.setJournalBufferSize_AIO(i);
+ assertEquals(i, conf.getJournalBufferSize_AIO());
i = randomInt();
- conf.setJournalBufferTimeout(i);
- assertEquals(i, conf.getJournalBufferTimeout());
+ conf.setJournalBufferTimeout_AIO(i);
+ assertEquals(i, conf.getJournalBufferTimeout_AIO());
+
+ i = randomInt();
+ conf.setJournalBufferSize_NIO(i);
+ assertEquals(i, conf.getJournalBufferSize_NIO());
- b = randomBoolean();
- conf.setJournalFlushOnSync(b);
- assertEquals(b, conf.isJournalFlushOnSync());
+ i = randomInt();
+ conf.setJournalBufferTimeout_NIO(i);
+ assertEquals(i, conf.getJournalBufferTimeout_NIO());
b = randomBoolean();
conf.setLogJournalWriteRate(b);
Modified: trunk/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/unit/core/config/impl/DefaultsFileConfigurationTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -105,12 +105,18 @@
assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MIN_FILES, conf.getJournalMinFiles());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_AIO, conf.getJournalMaxAIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_IO_AIO, conf.getJournalMaxIO_AIO());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT, conf.getJournalBufferTimeout());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, conf.getJournalBufferTimeout_AIO());
- assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE, conf.getJournalBufferSize());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, conf.getJournalBufferSize_AIO());
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_MAX_IO_NIO, conf.getJournalMaxIO_NIO());
+
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, conf.getJournalBufferTimeout_NIO());
+
+ assertEquals(ConfigurationImpl.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, conf.getJournalBufferSize_NIO());
+
assertEquals(ConfigurationImpl.DEFAULT_CREATE_BINDINGS_DIR, conf.isCreateBindingsDir());
assertEquals(ConfigurationImpl.DEFAULT_CREATE_JOURNAL_DIR, conf.isCreateJournalDir());
Modified: trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -69,17 +69,19 @@
assertEquals(false, conf.isCreateBindingsDir());
assertEquals("somedir2", conf.getJournalDirectory());
assertEquals(false, conf.isCreateJournalDir());
+
assertEquals(JournalType.NIO, conf.getJournalType());
- assertEquals(10000, conf.getJournalBufferSize());
- assertEquals(true, conf.isJournalFlushOnSync());
- assertEquals(1000, conf.getJournalBufferTimeout());
+ assertEquals(10000, conf.getJournalBufferSize_NIO());
+ assertEquals(1000, conf.getJournalBufferTimeout_NIO());
+ assertEquals(56546, conf.getJournalMaxIO_NIO());
+
assertEquals(false, conf.isJournalSyncTransactional());
assertEquals(true, conf.isJournalSyncNonTransactional());
assertEquals(12345678, conf.getJournalFileSize());
assertEquals(100, conf.getJournalMinFiles());
assertEquals(123, conf.getJournalCompactMinFiles());
assertEquals(33, conf.getJournalCompactPercentage());
- assertEquals(56546, conf.getJournalMaxAIO());
+
assertEquals("largemessagesdir", conf.getLargeMessagesDirectory());
assertEquals(95, conf.getMemoryWarningThreshold());
assertEquals(1024, conf.getBackupWindowSize());
Modified: trunk/tests/src/org/hornetq/tests/unit/core/deployers/impl/QueueDeployerTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/deployers/impl/QueueDeployerTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/unit/core/deployers/impl/QueueDeployerTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -275,7 +275,7 @@
return 0;
}
- public int getJournalMaxAIO()
+ public int getJournalMaxIO()
{
return 0;
@@ -526,13 +526,13 @@
}
- public int getAIOBufferSize()
+ public int getJournalBufferSize()
{
return 0;
}
- public int getAIOBufferTimeout()
+ public int getJournalBufferTimeout()
{
return 0;
}
Modified: trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -87,7 +87,7 @@
}
}
- TimedBuffer timedBuffer = new TimedBuffer(100, 3600 * 1000, false, false); // Any big timeout
+ TimedBuffer timedBuffer = new TimedBuffer(100, 3600 * 1000, false); // Any big timeout
timedBuffer.setObserver(new TestObserver());
@@ -106,8 +106,10 @@
timedBuffer.addBytes(buff, false, dummyCallback);
}
+ timedBuffer.checkSize(1);
+
assertEquals(1, flushTimes.get());
-
+
ByteBuffer flushedBuffer = buffers.get(0);
assertEquals(100, flushedBuffer.limit());
Modified: trunk/tests/src/org/hornetq/tests/util/ListJournal.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/util/ListJournal.java 2009-12-01 19:40:40 UTC (rev 8482)
+++ trunk/tests/src/org/hornetq/tests/util/ListJournal.java 2009-12-01 20:18:47 UTC (rev 8483)
@@ -63,7 +63,7 @@
new NIOSequentialFileFactory(fileConf.getJournalDirectory()),
"hornetq-data",
"hq",
- fileConf.getJournalMaxAIO());
+ fileConf.getJournalMaxIO_NIO());
ArrayList<RecordInfo> records = new ArrayList<RecordInfo>();
ArrayList<PreparedTransactionInfo> prepared = new ArrayList<PreparedTransactionInfo>();
More information about the hornetq-commits
mailing list