Author: clebert.suconic(a)jboss.com Date: 2011-12-13 12:49:57 -0500 (Tue, 13 Dec 2011) New Revision: 11904 Modified: branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFileFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java branches/Branch_2_2_EAP/tests/config/ConfigurationTest-full-config.xml branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: JBPAPP-7655 - adding max IO options for paging, changing memory allocation if using AIO (native layer) since the JDK is unreliable on allocating direct buffers (related to when they are GCed) Modified: branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/config/common/schema/hornetq-configuration.xsd 2011-12-13 17:49:57 UTC (rev 11904) @@ -146,6 +146,8 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="create-bindings-dir" type="xsd:boolean"> </xsd:element> + <xsd:element maxOccurs="1" minOccurs="0" name="page-max-concurrent-io" type="xsd:string"> + </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="journal-directory" type="xsd:string"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="create-journal-dir" type="xsd:boolean"> Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/Configuration.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -473,6 +473,15 @@ * Sets the file system directory used to store bindings. */ void setBindingsDirectory(String dir); + + /** The max number of concurrent reads allowed on paging. + * + * Default = 5 */ + int getPageMaxConcurrentIO(); + + /** The max number of concurrent reads allowed on paging. + * Default = 5 */ + void setPageMaxConcurrentIO(int maxIO); /** * Returns the file system directory used to store journal log. Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -89,6 +89,8 @@ public static final String DEFAULT_PAGING_DIR = "data/paging"; public static final String DEFAULT_LARGE_MESSAGES_DIR = "data/largemessages"; + + public static final int DEFAULT_MAX_CONCURRENT_PAGE_IO = 5; public static final boolean DEFAULT_CREATE_JOURNAL_DIR = true; @@ -266,6 +268,8 @@ protected String pagingDirectory = ConfigurationImpl.DEFAULT_PAGING_DIR; // File related attributes ----------------------------------------------------------- + + protected int maxConcurrentPageIO = ConfigurationImpl.DEFAULT_MAX_CONCURRENT_PAGE_IO; protected String largeMessagesDirectory = ConfigurationImpl.DEFAULT_LARGE_MESSAGES_DIR; @@ -624,7 +628,25 @@ { bindingsDirectory = dir; } + + /* (non-Javadoc) + * @see org.hornetq.core.config.Configuration#getPageMaxConcurrentIO() + */ + public int getPageMaxConcurrentIO() + { + return maxConcurrentPageIO; + } + + /* (non-Javadoc) + * @see org.hornetq.core.config.Configuration#setPageMaxConcurrentIO(int) + */ + public void setPageMaxConcurrentIO(int maxIO) + { + this.maxConcurrentPageIO = maxIO; + } + + public String getJournalDirectory() { return journalDirectory; Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -424,6 +424,12 @@ "journal-directory", config.getJournalDirectory(), Validators.NOT_NULL_OR_EMPTY)); + + + config.setPageMaxConcurrentIO(XMLConfigurationUtil.getInteger(e, + "page-max-concurrent-io", + 5, + Validators.MINUS_ONE_OR_GT_ZERO)); config.setPagingDirectory(XMLConfigurationUtil.getString(e, "paging-directory", Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFileFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/SequentialFileFactory.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -35,6 +35,14 @@ /** The SequentialFile will call this method when a disk IO Error happens during the live phase. */ void onIOError(int errorCode, String message, SequentialFile file); + /** used for cases where you need direct buffer outside of the journal context. + * This is because the native layer has a method that can be reused in certain cases like paging */ + ByteBuffer allocateDirectBuffer(int size); + + /** used for cases where you need direct buffer outside of the journal context. + * This is because the native layer has a method that can be reused in certain cases like paging */ + void releaseDirectBuffer(ByteBuffer buffer); + /** * Note: You need to release the buffer if is used for reading operations. * You don't need to do it if using writing operations (AIO Buffer Lister will take of writing operations) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/AIOSequentialFileFactory.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -113,7 +113,29 @@ { return AsynchronousFileImpl.isLoaded(); } + + public ByteBuffer allocateDirectBuffer(final int size) + { + + int blocks = size / 512; + if (size % 512 != 0) + { + blocks ++; + } + + // The buffer on AIO has to be a multiple of 512 + ByteBuffer buffer = AsynchronousFileImpl.newBuffer(blocks * 512); + + buffer.limit(size); + return buffer; + } + + public void releaseDirectBuffer(final ByteBuffer buffer) + { + AsynchronousFileImpl.destroyBuffer(buffer); + } + public ByteBuffer newBuffer(int size) { if (size % 512 != 0) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/journal/impl/NIOSequentialFileFactory.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -99,6 +99,17 @@ { return timedBuffer != null; } + + + public ByteBuffer allocateDirectBuffer(final int size) + { + return ByteBuffer.allocateDirect(size); + } + + public void releaseDirectBuffer(ByteBuffer buffer) + { + // nothing we can do on this case. we can just have good faith on GC + } public ByteBuffer newBuffer(final int size) { Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageCursorProviderImpl.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -183,6 +183,7 @@ { page = pagingStore.createPage((int)pageId); + storageManager.beforePageRead(); page.open(); List<PagedMessage> pgdMessages = page.read(storageManager); @@ -200,6 +201,7 @@ catch (Throwable ignored) { } + storageManager.afterPageRead(); cache.unlock(); } } @@ -455,8 +457,26 @@ // The page is not on cache any more // We need to read the page-file before deleting it // to make sure we remove any large-messages pending - depagedPage.open(); - List<PagedMessage> pgdMessagesList = depagedPage.read(storageManager); + storageManager.beforePageRead(); + + List<PagedMessage> pgdMessagesList = null; + try + { + depagedPage.open(); + pgdMessagesList = depagedPage.read(storageManager); + } + finally + { + try + { + depagedPage.close(); + } + catch (Exception e) + { + } + + storageManager.afterPageRead(); + } depagedPage.close(); pgdMessages = pgdMessagesList.toArray(new PagedMessage[pgdMessagesList.size()]); } Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/paging/cursor/impl/PageSubscriptionImpl.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -1290,7 +1290,11 @@ public void remove() { deliveredCount.incrementAndGet(); - PageSubscriptionImpl.this.getPageInfo(position).remove(position); + PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(position); + if (info != null) + { + info.remove(position); + } } /* (non-Javadoc) Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/StorageManager.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -13,6 +13,7 @@ package org.hornetq.core.persistence; +import java.nio.ByteBuffer; import java.util.List; import java.util.Map; import java.util.Set; @@ -96,6 +97,34 @@ * in case of the pools are full * @throws Exception */ void waitOnOperations() throws Exception; + + /** + * We need a safeguard in place to avoid too much concurrent IO happening on Paging, + * otherwise the system may become irrensponsive if too many destinations are reading all the same time. + * This is called before we read, so we can limit concurrent reads + * @throws Exception + */ + void beforePageRead() throws Exception; + + /** + * We need a safeguard in place to avoid too much concurrent IO happening on Paging, + * otherwise the system may become irrensponsive if too many destinations are reading all the same time. + * This is called after we read, so we can limit concurrent reads + * @throws Exception + */ + void afterPageRead() throws Exception; + + + /** AIO has an optimized buffer which has a method to release it + instead of the way NIO will release data based on GC. + These methods will use that buffer if the inner method supports it */ + ByteBuffer allocateDirectBuffer(int size); + + /** AIO has an optimized buffer which has a method to release it + instead of the way NIO will release data based on GC. + These methods will use that buffer if the inner method supports it */ + void freeDirectuffer(ByteBuffer buffer); + void clearContext(); Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -6,7 +6,9 @@ * 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 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND + private final Semaphore pageMaxConcurrentIO; +, either express or * implied. See the License for the specific language governing * permissions and limitations under the License. */ @@ -31,6 +33,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; import javax.transaction.xa.Xid; @@ -157,6 +160,8 @@ public static final byte PAGE_CURSOR_COUNTER_VALUE = 40; public static final byte PAGE_CURSOR_COUNTER_INC = 41; + + private final Semaphore pageMaxConcurrentIO; private final BatchingIDGenerator idGenerator; @@ -167,6 +172,8 @@ private final Journal bindingsJournal; private final SequentialFileFactory largeMessagesFactory; + + private SequentialFileFactory journalFF = null; private volatile boolean started; @@ -270,8 +277,6 @@ syncTransactional = config.isJournalSyncTransactional(); - SequentialFileFactory journalFF = null; - if (config.getJournalType() == JournalType.ASYNCIO) { JournalStorageManager.log.info("Using AIO Journal"); @@ -329,6 +334,15 @@ largeMessagesFactory = new NIOSequentialFileFactory(largeMessagesDirectory, false, criticalErrorListener); perfBlastPages = config.getJournalPerfBlastPages(); + + if (config.getPageMaxConcurrentIO() != 1) + { + pageMaxConcurrentIO = new Semaphore(config.getPageMaxConcurrentIO()); + } + else + { + pageMaxConcurrentIO = null; + } } public void clearContext() @@ -1598,6 +1612,45 @@ return info; } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#startPageRead() + */ + public void beforePageRead() throws Exception + { + if (pageMaxConcurrentIO != null) + { + pageMaxConcurrentIO.acquire(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#finishPageRead() + */ + public void afterPageRead() throws Exception + { + if (pageMaxConcurrentIO != null) + { + pageMaxConcurrentIO.release(); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#allocateDirectBuffer(long) + */ + public ByteBuffer allocateDirectBuffer(int size) + { + return journalFF.allocateDirectBuffer(size); + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#freeDirectuffer(java.nio.ByteBuffer) + */ + public void freeDirectuffer(ByteBuffer buffer) + { + journalFF.releaseBuffer(buffer); + } + // Public ----------------------------------------------------------------------------------- public Journal getMessageJournal() Modified: branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -13,6 +13,7 @@ package org.hornetq.core.persistence.impl.nullpm; +import java.nio.ByteBuffer; import java.util.Collections; import java.util.List; import java.util.Map; @@ -591,4 +592,34 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#beforePageRead() + */ + public void beforePageRead() throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#afterPageRead() + */ + public void afterPageRead() throws Exception + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#allocateDirectBuffer(int) + */ + public ByteBuffer allocateDirectBuffer(int size) + { + return ByteBuffer.allocateDirect(size); + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#freeDirectuffer(java.nio.ByteBuffer) + */ + public void freeDirectuffer(ByteBuffer buffer) + { + // We can just have hope on GC here :-) + } + } Modified: branches/Branch_2_2_EAP/tests/config/ConfigurationTest-full-config.xml =================================================================== --- branches/Branch_2_2_EAP/tests/config/ConfigurationTest-full-config.xml 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/tests/config/ConfigurationTest-full-config.xml 2011-12-13 17:49:57 UTC (rev 11904) @@ -36,6 +36,7 @@ <create-bindings-dir>false</create-bindings-dir> <journal-directory>somedir2</journal-directory> <create-journal-dir>false</create-journal-dir> + <page-max-concurrent-io>17</page-max-concurrent-io> <journal-type>NIO</journal-type> <journal-compact-min-files>123</journal-compact-min-files> <journal-compact-percentage>33</journal-compact-percentage> Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/config/impl/FileConfigurationTest.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -70,6 +70,8 @@ Assert.assertEquals("pagingdir", conf.getPagingDirectory()); Assert.assertEquals("somedir", conf.getBindingsDirectory()); Assert.assertEquals(false, conf.isCreateBindingsDir()); + + Assert.assertEquals(17, conf.getPageMaxConcurrentIO()); Assert.assertEquals("somedir2", conf.getJournalDirectory()); Assert.assertEquals(false, conf.isCreateJournalDir()); Assert.assertEquals(JournalType.NIO, conf.getJournalType()); Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/journal/impl/fakes/FakeSequentialFileFactory.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -764,4 +764,19 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.journal.SequentialFileFactory#newDirectBuffer(int) + */ + public ByteBuffer allocateDirectBuffer(int size) + { + return ByteBuffer.allocateDirect(size); + } + + /* (non-Javadoc) + * @see org.hornetq.core.journal.SequentialFileFactory#releaseDirectBuffer(java.nio.ByteBuffer) + */ + public void releaseDirectBuffer(ByteBuffer buffer) + { + } + } Modified: branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2011-12-13 17:25:49 UTC (rev 11903) +++ branches/Branch_2_2_EAP/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2011-12-13 17:49:57 UTC (rev 11904) @@ -13,6 +13,7 @@ package org.hornetq.tests.unit.core.paging.impl; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -1706,6 +1707,42 @@ } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#beforePageRead() + */ + public void beforePageRead() throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#afterPageRead() + */ + public void afterPageRead() throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#allocateDirectBuffer(int) + */ + public ByteBuffer allocateDirectBuffer(int size) + { + // TODO Auto-generated method stub + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#freeDirectuffer(java.nio.ByteBuffer) + */ + public void freeDirectuffer(ByteBuffer buffer) + { + // TODO Auto-generated method stub + + } + } class FakeStoreFactory implements PagingStoreFactory @@ -1783,6 +1820,41 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#startPageRead() + */ + public void beforePageRead() throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#finishPageRead() + */ + public void afterPageRead() throws Exception + { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#allocateDirectBuffer(int) + */ + public ByteBuffer allocateDirectBuffer(int size) + { + return ByteBuffer.allocateDirect(size); + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#freeDirectuffer(java.nio.ByteBuffer) + */ + public void freeDirectuffer(ByteBuffer buffer) + { + // TODO Auto-generated method stub + + } + } }