[hornetq-commits] JBoss hornetq SVN: r9680 - in branches/Branch_2_1: src/main/org/hornetq/core/paging/impl and 5 other directories.

Author: clebert.suconic(a)jboss.com Date: 2010-09-13 18:39:14 -0400 (Mon, 13 Sep 2010) New Revision: 9680 Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/PageTransactionInfo.java branches/Branch_2_1/src/main/org/hornetq/core/paging/PagingManager.java branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java branches/Branch_2_1/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Branch_2_1/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java branches/Branch_2_1/tests/src/org/hornetq/tests/integration/client/PagingTest.java branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java Log: Fixing a leak on PageTransaction and adding tests on ordering for paging Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/PageTransactionInfo.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/paging/PageTransactionInfo.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/paging/PageTransactionInfo.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -40,12 +40,12 @@ long getTransactionID(); - void store(StorageManager storageManager,Transaction tx) throws Exception; + void store(StorageManager storageManager, PagingManager pagingManager, Transaction tx) throws Exception; - void storeUpdate(StorageManager storageManager, Transaction tx, int depages) throws Exception; + void storeUpdate(StorageManager storageManager, PagingManager pagingManager, Transaction tx, int depages) throws Exception; // To be used after the update was stored or reload - void update(int update, StorageManager storageManager); + void update(int update, StorageManager storageManager, PagingManager pagingManager); void increment(); Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/PagingManager.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/paging/PagingManager.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/paging/PagingManager.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -13,6 +13,8 @@ package org.hornetq.core.paging; +import java.util.Map; + import org.hornetq.api.core.SimpleString; import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.postoffice.PostOffice; @@ -67,6 +69,8 @@ * @param transactionID */ void removeTransaction(long transactionID); + + Map<Long, PageTransactionInfo> getTransactions(); /** * Reload previously created PagingStores into memory Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PageTransactionInfoImpl.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -20,6 +20,7 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.PageTransactionInfo; +import org.hornetq.core.paging.PagingManager; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.transaction.Transaction; import org.hornetq.core.transaction.TransactionOperation; @@ -56,6 +57,7 @@ public PageTransactionInfoImpl(final long transactionID) { + this(); this.transactionID = transactionID; countDownCompleted = new CountDownLatch(1); } @@ -81,7 +83,7 @@ return transactionID; } - public void update(final int update, final StorageManager storageManager) + public void update(final int update, final StorageManager storageManager, PagingManager pagingManager) { int sizeAfterUpdate = numberOfMessages.addAndGet(-update); if (sizeAfterUpdate == 0 && storageManager != null) @@ -95,6 +97,11 @@ log.warn("Can't delete page transaction id=" + this.recordID); } } + + if (sizeAfterUpdate == 0 && pagingManager != null) + { + pagingManager.removeTransaction(this.transactionID); + } } public void increment() @@ -149,7 +156,7 @@ } } - public void store(final StorageManager storageManager, final Transaction tx) throws Exception + public void store(final StorageManager storageManager, PagingManager pagingManager, final Transaction tx) throws Exception { storageManager.storePageTransaction(tx.getID(), this); } @@ -157,7 +164,7 @@ /* (non-Javadoc) * @see org.hornetq.core.paging.PageTransactionInfo#storeUpdate(org.hornetq.core.persistence.StorageManager, org.hornetq.core.transaction.Transaction, int) */ - public void storeUpdate(final StorageManager storageManager, final Transaction tx, final int depages) throws Exception + public void storeUpdate(final StorageManager storageManager, final PagingManager pagingManager, final Transaction tx, final int depages) throws Exception { storageManager.updatePageTransaction(tx.getID(), this, depages); @@ -187,7 +194,7 @@ public void afterCommit(Transaction tx) { - pgToUpdate.update(depages, storageManager); + pgToUpdate.update(depages, storageManager, pagingManager); } }); } Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingManagerImpl.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -14,6 +14,7 @@ package org.hornetq.core.paging.impl; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -155,7 +156,16 @@ { return transactions.get(id); } + + /* (non-Javadoc) + * @see org.hornetq.core.paging.PagingManager#getTransactions() + */ + public Map<Long, PageTransactionInfo> getTransactions() + { + return transactions; + } + // HornetQComponent implementation // ------------------------------------------------------------------------------------------------ Modified: branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -1059,7 +1059,7 @@ // This will set the journal transaction to commit; depageTransaction.setContainsPersistent(); - entry.getKey().storeUpdate(storageManager, depageTransaction, entry.getValue().intValue()); + entry.getKey().storeUpdate(storageManager, this.pagingManager, depageTransaction, entry.getValue().intValue()); } depageTransaction.commit(); Modified: branches/Branch_2_1/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -913,7 +913,7 @@ PageTransactionInfo pageTX = pagingManager.getTransaction(pageUpdate.pageTX); - pageTX.update(pageUpdate.recods, null); + pageTX.update(pageUpdate.recods, null, null); } else { Modified: branches/Branch_2_1/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java =================================================================== --- branches/Branch_2_1/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/src/main/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -1267,7 +1267,7 @@ store.sync(); } - pageTransaction.store(storageManager, tx); + pageTransaction.store(storageManager, pagingManager, tx); } } } Modified: branches/Branch_2_1/tests/src/org/hornetq/tests/integration/client/PagingTest.java =================================================================== --- branches/Branch_2_1/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/tests/src/org/hornetq/tests/integration/client/PagingTest.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -34,6 +34,7 @@ import org.hornetq.core.config.Configuration; import org.hornetq.core.config.DivertConfiguration; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.impl.PageTransactionInfoImpl; import org.hornetq.core.paging.impl.TestSupportPageStore; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.Queue; @@ -163,6 +164,10 @@ for (int i = 0; i < numberOfMessages; i++) { + if (i % 500 == 0) + { + session.commit(); + } message = session.createMessage(true); HornetQBuffer bodyLocal = message.getBodyBuffer(); @@ -240,6 +245,8 @@ } consumer.close(); + + session.close(); } catch (Throwable e) { @@ -260,6 +267,9 @@ { threads[i].join(); } + + assertEquals(0, server.getPostOffice().getPagingManager().getTransactions().size()); + } finally { @@ -812,6 +822,136 @@ } + + public void testDepageDuringTransaction4() throws Exception + { + clearData(); + + Configuration config = createDefaultConfig(); + + HornetQServer server = createServer(true, + config, + PagingTest.PAGE_SIZE, + PagingTest.PAGE_MAX, + new HashMap<String, AddressSettings>()); + + server.getConfiguration().setJournalSyncNonTransactional(false); + server.getConfiguration().setJournalSyncTransactional(false); + + server.start(); + + final AtomicInteger errors = new AtomicInteger(0); + + final int messageSize = 1024; // 1k + final int numberOfMessages = 10000; + + try + { + final ClientSessionFactory sf = createInVMFactory(); + + + sf.setBlockOnNonDurableSend(true); + sf.setBlockOnDurableSend(true); + sf.setBlockOnAcknowledge(false); + + final byte[] body = new byte[messageSize]; + + + Thread producerThread = new Thread() + { + public void run() + { + ClientSession sessionProducer = null; + try + { + sessionProducer = sf.createSession(false, false); + ClientProducer producer = sessionProducer.createProducer(ADDRESS); + + for (int i = 0 ; i < numberOfMessages; i++) + { + ClientMessage msg = sessionProducer.createMessage(true); + msg.getBodyBuffer().writeBytes(body); + msg.putIntProperty("count", i); + producer.send(msg); + + if (i % 50 == 0 && i != 0) + { + sessionProducer.commit(); + //Thread.sleep(500); + } + } + + sessionProducer.commit(); + + System.out.println("Producer gone"); + + + + } + catch (Throwable e) + { + e.printStackTrace(); // >> junit report + errors.incrementAndGet(); + } + finally + { + try + { + if (sessionProducer != null) + { + sessionProducer.close(); + } + } + catch (Throwable e) + { + e.printStackTrace(); + errors.incrementAndGet(); + } + } + } + }; + + ClientSession session = sf.createSession(true, true, 0); + session.start(); + session.createQueue(PagingTest.ADDRESS, PagingTest.ADDRESS, null, true); + + producerThread.start(); + + ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS); + + + for (int i = 0 ; i < numberOfMessages; i++) + { + ClientMessage msg = consumer.receive(500000); + assertNotNull(msg); + assertEquals(i, msg.getIntProperty("count").intValue()); + msg.acknowledge(); + if (i > 0 && i % 10 == 0) + { + //session.commit(); + } + } + //session.commit(); + + session.close(); + + producerThread.join(); + + assertEquals(0, errors.get()); + } + finally + { + try + { + server.stop(); + } + catch (Throwable ignored) + { + } + } + + } + public void testPageOnSchedulingNoRestart() throws Exception { internalTestPageOnScheduling(false); Modified: branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -688,6 +688,154 @@ Assert.assertEquals(0, storeImpl.getAddressSize()); } + public void testOrderOnPaging() throws Throwable + { + clearData(); + SequentialFileFactory factory = new NIOSequentialFileFactory(this.getPageDir()); + + PagingStoreFactory storeFactory = new FakeStoreFactory(factory); + + final int MAX_SIZE = 1024 * 10; + + AddressSettings settings = new AddressSettings(); + settings.setPageSizeBytes(MAX_SIZE); + settings.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE); + + final TestSupportPageStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, + createMockManager(), + createStorageManagerMock(), + createPostOfficeMock(), + factory, + storeFactory, + new SimpleString("test"), + settings, + executor, + true); + + storeImpl.start(); + + Assert.assertEquals(0, storeImpl.getNumberOfPages()); + + // Marked the store to be paged + storeImpl.startPaging(); + + final CountDownLatch producedLatch = new CountDownLatch(1); + + Assert.assertEquals(1, storeImpl.getNumberOfPages()); + + final SimpleString destination = new SimpleString("test"); + + final long NUMBER_OF_MESSAGES = 100000; + + final List<Throwable> errors = new ArrayList<Throwable>(); + + class WriterThread extends Thread + { + + public WriterThread() + { + super("PageWriter"); + } + + @Override + public void run() + { + + try + { + for (long i = 0; i < NUMBER_OF_MESSAGES; i++) + { + // Each thread will Keep paging until all the messages are depaged. + // This is possible because the depage thread is not actually reading the pages. + // Just using the internal API to remove it from the page file system + ServerMessage msg = createMessage(i, storeImpl, destination, createRandomBuffer(i, 1024)); + msg.putLongProperty("count", i); + while (!storeImpl.page(msg)) + { + storeImpl.startPaging(); + } + + if (i == 0) + { + producedLatch.countDown(); + } + } + } + catch (Throwable e) + { + e.printStackTrace(); + errors.add(e); + } + } + } + + class ReaderThread extends Thread + { + public ReaderThread() + { + super("PageReader"); + } + + @Override + public void run() + { + try + { + + long msgsRead = 0; + + while (msgsRead < NUMBER_OF_MESSAGES) + { + Page page = storeImpl.depage(); + if (page != null) + { + page.open(); + List<PagedMessage> messages = page.read(); + + for (PagedMessage pgmsg : messages) + { + ServerMessage msg = pgmsg.getMessage(null); + + assertEquals(msgsRead++, msg.getMessageID()); + + assertEquals(msg.getMessageID(), msg.getLongProperty("count").longValue()); + } + + page.close(); + page.delete(); + } + else + { + System.out.println("Depaged!!!!"); + Thread.sleep(500); + } + } + + } + catch (Throwable e) + { + e.printStackTrace(); + errors.add(e); + } + } + } + + WriterThread producerThread = new WriterThread(); + producerThread.start(); + ReaderThread consumer = new ReaderThread(); + consumer.start(); + + producerThread.join(); + consumer.join(); + + storeImpl.stop(); + + for (Throwable e: errors) + { + throw e; + } + } + /** * @return */ @@ -878,6 +1026,15 @@ return false; } + /* (non-Javadoc) + * @see org.hornetq.core.paging.PagingManager#getTransactions() + */ + public Map<Long, PageTransactionInfo> getTransactions() + { + // TODO Auto-generated method stub + return null; + } + } class FakeStorageManager implements StorageManager Modified: branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java =================================================================== --- branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java 2010-09-13 21:33:55 UTC (rev 9679) +++ branches/Branch_2_1/tests/src/org/hornetq/tests/unit/core/postoffice/impl/DuplicateDetectionUnitTest.java 2010-09-13 22:39:14 UTC (rev 9680) @@ -17,6 +17,7 @@ import java.util.Collection; import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -35,7 +36,6 @@ import org.hornetq.core.persistence.impl.journal.JournalStorageManager; import org.hornetq.core.postoffice.PostOffice; import org.hornetq.core.postoffice.impl.DuplicateIDCacheImpl; -import org.hornetq.core.server.JournalType; import org.hornetq.core.server.Queue; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.transaction.impl.ResourceManagerImpl; @@ -312,6 +312,14 @@ return false; } + /* (non-Javadoc) + * @see org.hornetq.core.paging.PagingManager#getTransactions() + */ + public Map<Long, PageTransactionInfo> getTransactions() + { + return null; + } + } }