8:05 p.m.
Author: clebert.suconic(a)jboss.com
Date: 2009-11-12 21:05:29 -0500 (Thu, 12 Nov 2009)
New Revision: 8278
Modified:
trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java
trunk/src/main/org/hornetq/core/journal/impl/TimedBufferObserver.java
trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java
Log:
https://jira.jboss.org/jira/browse/HORNETQ-214 - performance tweaks
Modified: trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2009-11-12
21:08:49 UTC (rev 8277)
+++ trunk/src/main/org/hornetq/core/journal/impl/AbstractSequentialFile.java 2009-11-13
02:05:29 UTC (rev 8278)
@@ -55,9 +55,6 @@
* This is the class returned to the factory when the file is being activated. */
protected final TimedBufferObserver timedBufferObserver = new LocalBufferObserver();
-
-
-
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
@@ -66,7 +63,7 @@
* @param file
* @param directory
*/
- public AbstractSequentialFile(String directory, File file, SequentialFileFactory
factory)
+ public AbstractSequentialFile(final String directory, final File file, final
SequentialFileFactory factory)
{
super();
this.file = file;
@@ -86,7 +83,6 @@
return file.getName();
}
-
public final void delete() throws Exception
{
if (isOpen())
@@ -107,12 +103,10 @@
return position.get();
}
-
public final void renameTo(final String newFileName) throws Exception
{
close();
File newFile = new File(directory + "/" + newFileName);
-
if (!file.equals(newFile))
{
@@ -120,13 +114,12 @@
file = newFile;
}
}
-
- public final boolean fits(int size)
+ public final boolean fits(final int size)
{
if (timedBuffer == null)
{
- return this.position.get() + size <= fileSize;
+ return position.get() + size <= fileSize;
}
else
{
@@ -150,22 +143,22 @@
}
}
- public void setTimedBuffer(TimedBuffer buffer)
+ public void setTimedBuffer(final TimedBuffer buffer)
{
if (timedBuffer != null)
{
timedBuffer.setObserver(null);
}
- this.timedBuffer = buffer;
+ timedBuffer = buffer;
if (buffer != null)
{
- buffer.setObserver(this.timedBufferObserver);
+ buffer.setObserver(timedBufferObserver);
}
}
-
+
public void write(final HornetQBuffer bytes, final boolean sync, final IOCompletion
callback) throws Exception
{
if (timedBuffer != null)
@@ -196,8 +189,6 @@
write(bytes, false, DummyCallback.getInstance());
}
}
-
-
// Package protected ---------------------------------------------
@@ -208,7 +199,6 @@
return file;
}
-
// Private -------------------------------------------------------
// Inner classes -------------------------------------------------
@@ -217,7 +207,7 @@
{
final List<IOCompletion> delegates;
- DelegateCallback(List<IOCompletion> delegates)
+ DelegateCallback(final List<IOCompletion> delegates)
{
this.delegates = delegates;
}
@@ -237,7 +227,7 @@
}
}
- public void onError(int errorCode, String errorMessage)
+ public void onError(final int errorCode, final String errorMessage)
{
for (IOCompletion callback : delegates)
{
@@ -259,7 +249,7 @@
protected class LocalBufferObserver implements TimedBufferObserver
{
- public void flushBuffer(ByteBuffer buffer, List<IOCompletion> callbacks)
+ public void flushBuffer(final ByteBuffer buffer, final boolean requestedSync, final
List<IOCompletion> callbacks)
{
buffer.flip();
@@ -269,7 +259,7 @@
}
else
{
- writeDirect(buffer, true, new DelegateCallback(callbacks));
+ writeDirect(buffer, requestedSync, new DelegateCallback(callbacks));
}
}
@@ -295,6 +285,7 @@
}
}
+ @Override
public String toString()
{
return "TimedBufferObserver on file (" + getFile().getName() +
")";
Modified: trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-11-12 21:08:49 UTC
(rev 8277)
+++ trunk/src/main/org/hornetq/core/journal/impl/TimedBuffer.java 2009-11-13 02:05:29 UTC
(rev 8278)
@@ -241,18 +241,15 @@
if (sync)
{
+ if (!pendingSync)
+ {
+ pendingSync = true;
+ }
+
if (flushOnSync)
{
flush();
}
- else
- {
- // We should flush on the next timeout, no matter what other activity happens
on the buffer
- if (!pendingSync)
- {
- pendingSync = true;
- }
- }
}
if (buffer.writerIndex() == bufferLimit)
@@ -281,7 +278,7 @@
directBuffer.put(buffer.array(), 0, pos);
- bufferObserver.flushBuffer(directBuffer, callbacks);
+ bufferObserver.flushBuffer(directBuffer, pendingSync, callbacks);
callbacks = new ArrayList<IOCompletion>();
Modified: trunk/src/main/org/hornetq/core/journal/impl/TimedBufferObserver.java
===================================================================
--- trunk/src/main/org/hornetq/core/journal/impl/TimedBufferObserver.java 2009-11-12
21:08:49 UTC (rev 8277)
+++ trunk/src/main/org/hornetq/core/journal/impl/TimedBufferObserver.java 2009-11-13
02:05:29 UTC (rev 8278)
@@ -39,7 +39,7 @@
// Public --------------------------------------------------------
- public void flushBuffer(ByteBuffer buffer, List<IOCompletion> callbacks);
+ public void flushBuffer(ByteBuffer buffer, boolean syncRequested,
List<IOCompletion> callbacks);
/** Return the number of remaining bytes that still fit on the observer (file) */
Modified:
trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
---
trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-12
21:08:49 UTC (rev 8277)
+++
trunk/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-11-13
02:05:29 UTC (rev 8278)
@@ -226,7 +226,7 @@
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");
+ 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(),
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-11-12
21:08:49 UTC (rev 8277)
+++
trunk/tests/src/org/hornetq/tests/unit/core/journal/impl/TimedBufferTest.java 2009-11-13
02:05:29 UTC (rev 8278)
@@ -64,7 +64,7 @@
final AtomicInteger flushTimes = new AtomicInteger(0);
class TestObserver implements TimedBufferObserver
{
- public void flushBuffer(final ByteBuffer buffer, final List<IOCompletion>
callbacks)
+ public void flushBuffer(final ByteBuffer buffer, final boolean sync, final
List<IOCompletion> callbacks)
{
buffers.add(buffer);
flushTimes.incrementAndGet();