[hornetq-commits] JBoss hornetq SVN: r8070 - in branches/Replication_Clebert: src/main/org/hornetq/core/paging/impl and 8 other directories.

Author: clebert.suconic(a)jboss.com Date: 2009-10-08 17:40:37 -0400 (Thu, 08 Oct 2009) New Revision: 8070 Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java Removed: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java Log: Paging changes Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/PagingStore.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -54,6 +54,8 @@ public boolean readPage() throws Exception; Page getCurrentPage(); + + Page createPage(final int page) throws Exception; /** * Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PageImpl.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -28,6 +28,8 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.Page; import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.utils.SimpleString; /** * @@ -59,18 +61,30 @@ private final SequentialFileFactory fileFactory; private final AtomicInteger size = new AtomicInteger(0); + + private final StorageManager storageManager; + + private final SimpleString storeName; // Static -------------------------------------------------------- // Constructors -------------------------------------------------- - public PageImpl(final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception + public PageImpl(final SimpleString storeName, final StorageManager storageManager, final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception { this.pageId = pageId; this.file = file; - fileFactory = factory; + this.fileFactory = factory; + this.storageManager = storageManager; + this.storeName = storeName; } + + public PageImpl(final SequentialFileFactory factory, final SequentialFile file, final int pageId) throws Exception + { + this(null, null, factory, file, pageId); + } + // Public -------------------------------------------------------- // PagingFile implementation @@ -154,9 +168,11 @@ numberOfMessages.incrementAndGet(); size.addAndGet(buffer.limit()); + storageManager.pageWrite(message, pageId); + if (message.getMessage(null).isLargeMessage()) { - // If we don't sync on large messages we could have the risk of files unnatended files on disk + // If we don't sync on large messages we could have the risk of unattended files on disk sync(); } } @@ -175,6 +191,7 @@ public void close() throws Exception { + storageManager.pageClosed(storeName, pageId); file.close(); } Modified: branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/paging/impl/PagingStoreImpl.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -607,6 +607,35 @@ return currentPage; } + + public Page createPage(final int page) throws Exception + { + String fileName = createFileName(page); + + if (fileFactory == null) + { + fileFactory = storeFactory.newFileFactory(getStoreName()); + } + + SequentialFile file = fileFactory.createSequentialFile(fileName, 1000); + + file.open(); + + long size = file.size(); + + if (fileFactory.isSupportsCallbacks() && size < pageSize) + { + file.fill((int)size, (int)(pageSize - size), (byte)0); + } + + file.position(0); + + file.close(); + + return new PageImpl(this.storeName, storageManager, fileFactory, file, page); + } + + // TestSupportPageStore ------------------------------------------ public void forceAnotherPage() throws Exception @@ -700,36 +729,6 @@ // Protected ----------------------------------------------------- - // In order to test failures, we need to be able to extend this class - // and replace the Page for another Page that will fail before the file is removed - // That's why createPage is not a private method - protected Page createPage(final int page) throws Exception - { - String fileName = createFileName(page); - - if (fileFactory == null) - { - fileFactory = storeFactory.newFileFactory(getStoreName()); - } - - SequentialFile file = fileFactory.createSequentialFile(fileName, 1000); - - file.open(); - - long size = file.size(); - - if (fileFactory.isSupportsCallbacks() && size < pageSize) - { - file.fill((int)size, (int)(pageSize - size), (byte)0); - } - - file.position(0); - - file.close(); - - return new PageImpl(fileFactory, file, page); - } - // Private ------------------------------------------------------- /** @@ -931,8 +930,9 @@ { currentPage.close(); } - + currentPage = createPage(currentPageId); + currentPageSize.set(0); Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/StorageManager.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -18,7 +18,9 @@ import javax.transaction.xa.Xid; +import org.hornetq.core.buffers.ChannelBuffer; import org.hornetq.core.paging.PageTransactionInfo; +import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; import org.hornetq.core.postoffice.Binding; import org.hornetq.core.server.HornetQComponent; @@ -44,6 +46,12 @@ { // Message related operations + void pageClosed(SimpleString storeName, int pageNumber); + + void pageDeleted(SimpleString storeName, int pageNumber); + + void pageWrite(PagedMessage message, int pageNumber); + boolean isReplicated(); void afterReplicated(Runnable run); Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -29,6 +29,7 @@ import javax.transaction.xa.Xid; +import org.hornetq.core.buffers.ChannelBuffer; import org.hornetq.core.buffers.ChannelBuffers; import org.hornetq.core.config.Configuration; import org.hornetq.core.exception.HornetQException; @@ -47,6 +48,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.message.impl.MessageImpl; import org.hornetq.core.paging.PageTransactionInfo; +import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; import org.hornetq.core.paging.impl.PageTransactionInfoImpl; import org.hornetq.core.persistence.QueueBindingInfo; @@ -290,6 +292,47 @@ return replicator != null; } + + // TODO: shouldn't those page methods be on the PageManager? + + /* + * + * (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageClosed(org.hornetq.utils.SimpleString, int) + */ + public void pageClosed(SimpleString storeName, int pageNumber) + { + if (isReplicated()) + { + replicator.pageClosed(storeName, pageNumber); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageDeleted(org.hornetq.utils.SimpleString, int) + */ + public void pageDeleted(SimpleString storeName, int pageNumber) + { + if (isReplicated()) + { + replicator.pageDeleted(storeName, pageNumber); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageWrite(org.hornetq.utils.SimpleString, int, org.hornetq.core.buffers.ChannelBuffer) + */ + public void pageWrite(PagedMessage message, int pageNumber) + { + if (isReplicated()) + { + replicator.pageWrite(message, pageNumber); + } + } + + + // TODO: shouldn't those page methods be on the PageManager? ^^^^ + public void afterReplicated(Runnable run) { if (replicator == null) Modified: branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -21,6 +21,7 @@ import org.hornetq.core.logging.Logger; import org.hornetq.core.paging.PageTransactionInfo; +import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.paging.PagingManager; import org.hornetq.core.persistence.QueueBindingInfo; import org.hornetq.core.persistence.StorageManager; @@ -268,4 +269,25 @@ { } + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageClosed(org.hornetq.utils.SimpleString, int) + */ + public void pageClosed(SimpleString storeName, int pageNumber) + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageDeleted(org.hornetq.utils.SimpleString, int) + */ + public void pageDeleted(SimpleString storeName, int pageNumber) + { + } + + /* (non-Javadoc) + * @see org.hornetq.core.persistence.StorageManager#pageWrite(org.hornetq.core.paging.PagedMessage, int) + */ + public void pageWrite(PagedMessage message, int pageNumber) + { + } + } Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -12,6 +12,9 @@ */ package org.hornetq.core.remoting.impl; + +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_EVENT; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PAGE_WRITE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_PREPARE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_DELETE_TX; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REPLICATION_COMMIT_ROLLBACK; @@ -405,6 +408,16 @@ packet = new ReplicationResponseMessage(); break; } + case REPLICATION_PAGE_WRITE: + { + packet = new ReplicationResponseMessage(); + break; + } + case REPLICATION_PAGE_EVENT: + { + packet = new ReplicationResponseMessage(); + break; + } default: { throw new IllegalArgumentException("Invalid type: " + packetType); Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -155,7 +155,11 @@ public static final byte REPLICATION_COMMIT_ROLLBACK = 86; + public static final byte REPLICATION_PAGE_WRITE = 87; + + public static final byte REPLICATION_PAGE_EVENT = 88; + // Static -------------------------------------------------------- Deleted: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -1,53 +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.remoting.impl.wireformat; - -/** - * A ReplicationPacket - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class ReplicationPacket extends PacketImpl -{ - - /** - * @param type - */ - public ReplicationPacket(byte type) - { - super(type); - // TODO Auto-generated constructor stub - } - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageEventMessage.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -0,0 +1,117 @@ +/* + * 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.remoting.impl.wireformat; + +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.utils.DataConstants; +import org.hornetq.utils.SimpleString; + +/** + * A ReplicationPageWrite + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class ReplicationPageEventMessage extends PacketImpl +{ + + private int pageNumber; + + private SimpleString storeName; + + /** + * True = delete page, False = close page + */ + private boolean isDelete; + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public ReplicationPageEventMessage() + { + super(REPLICATION_PAGE_EVENT); + } + + public ReplicationPageEventMessage(final SimpleString storeName, final int pageNumber, final boolean isDelete) + { + this(); + this.pageNumber = pageNumber; + this.isDelete = isDelete; + this.storeName = storeName; + } + + // Public -------------------------------------------------------- + + @Override + public int getRequiredBufferSize() + { + return BASIC_PACKET_SIZE + DataConstants.SIZE_INT + storeName.sizeof() + DataConstants.SIZE_BOOLEAN; + + } + + @Override + public void encodeBody(final HornetQBuffer buffer) + { + buffer.writeSimpleString(storeName); + buffer.writeInt(pageNumber); + buffer.writeBoolean(isDelete); + } + + @Override + public void decodeBody(final HornetQBuffer buffer) + { + storeName = buffer.readSimpleString(); + pageNumber = buffer.readInt(); + isDelete = buffer.readBoolean(); + } + + /** + * @return the pageNumber + */ + public int getPageNumber() + { + return pageNumber; + } + + /** + * @return the storeName + */ + public SimpleString getStoreName() + { + return storeName; + } + + /** + * @return the isDelete + */ + public boolean isDelete() + { + return isDelete; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPageWriteMessage.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -0,0 +1,107 @@ +/* + * 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.remoting.impl.wireformat; + +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.impl.PagedMessageImpl; +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.utils.DataConstants; + +/** + * A ReplicationPageWrite + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class ReplicationPageWriteMessage extends PacketImpl +{ + + int pageNumber; + + PagedMessage pagedMessage; + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public ReplicationPageWriteMessage() + { + super(REPLICATION_PAGE_WRITE); + } + + public ReplicationPageWriteMessage(final PagedMessage pagedMessage, final int pageNumber) + { + this(); + this.pageNumber = pageNumber; + this.pagedMessage = pagedMessage; + } + + // Public -------------------------------------------------------- + + public int getRequiredBufferSize() + { + return BASIC_PACKET_SIZE + + DataConstants.SIZE_INT + + pagedMessage.getEncodeSize(); + + } + + @Override + public void encodeBody(final HornetQBuffer buffer) + { + buffer.writeInt(pageNumber); + pagedMessage.encode(buffer); + } + + @Override + public void decodeBody(final HornetQBuffer buffer) + { + this.pageNumber = buffer.readInt(); + pagedMessage = new PagedMessageImpl(); + pagedMessage.decode(buffer); + } + + /** + * @return the pageNumber + */ + public int getPageNumber() + { + return pageNumber; + } + + /** + * @return the pagedMessage + */ + public PagedMessage getPagedMessage() + { + return pagedMessage; + } + + + + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -16,7 +16,9 @@ import java.util.Set; import org.hornetq.core.journal.EncodingSupport; +import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.server.HornetQComponent; +import org.hornetq.utils.SimpleString; /** * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> @@ -53,4 +55,23 @@ /** A list of tokens that are still waiting for replications to be completed */ Set<ReplicationToken> getActiveTokens(); + /** + * @param storeName + * @param pageNumber + */ + void pageClosed(SimpleString storeName, int pageNumber); + + /** + * @param storeName + * @param pageNumber + */ + void pageDeleted(SimpleString storeName, int pageNumber); + + + /** + * @param storeName + * @param pageNumber + */ + void pageWrite(PagedMessage message, int pageNumber); + } Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -13,9 +13,17 @@ package org.hornetq.core.replication.impl; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + import org.hornetq.core.config.Configuration; import org.hornetq.core.journal.Journal; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.Page; +import org.hornetq.core.paging.PagedMessage; +import org.hornetq.core.paging.PagingManager; +import org.hornetq.core.paging.impl.PagingManagerImpl; +import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; import org.hornetq.core.persistence.impl.journal.JournalStorageManager; import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.Packet; @@ -25,10 +33,14 @@ import org.hornetq.core.remoting.impl.wireformat.ReplicationCommitMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteTXMessage; +import org.hornetq.core.remoting.impl.wireformat.ReplicationPageEventMessage; +import org.hornetq.core.remoting.impl.wireformat.ReplicationPageWriteMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationPrepareMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationResponseMessage; import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.server.HornetQServer; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.utils.SimpleString; /** * @@ -57,6 +69,10 @@ private JournalStorageManager storage; + private PagingManager pageManager; + + private final ConcurrentMap<SimpleString, ConcurrentMap<Integer, Page>> pageIndex = new ConcurrentHashMap<SimpleString, ConcurrentMap<Integer, Page>>(); + // Constructors -------------------------------------------------- public ReplicationEndpointImpl(final HornetQServer server) { @@ -74,28 +90,37 @@ { if (packet.getType() == PacketImpl.REPLICATION_APPEND) { - handleAppendAddRecord(packet); + handleAppendAddRecord((ReplicationAddMessage)packet); } else if (packet.getType() == PacketImpl.REPLICATION_APPEND_TX) { - handleAppendAddTXRecord(packet); + handleAppendAddTXRecord((ReplicationAddTXMessage)packet); } else if (packet.getType() == PacketImpl.REPLICATION_DELETE) { - handleAppendDelete(packet); + handleAppendDelete((ReplicationDeleteMessage)packet); } else if (packet.getType() == PacketImpl.REPLICATION_DELETE_TX) { - handleAppendDeleteTX(packet); + handleAppendDeleteTX((ReplicationDeleteTXMessage)packet); } else if (packet.getType() == PacketImpl.REPLICATION_PREPARE) { - handlePrepare(packet); + handlePrepare((ReplicationPrepareMessage)packet); } else if (packet.getType() == PacketImpl.REPLICATION_COMMIT_ROLLBACK) { - handleCommitRollback(packet); + handleCommitRollback((ReplicationCommitMessage)packet); } + else if (packet.getType() == PacketImpl.REPLICATION_PAGE_WRITE) + { + handlePageWrite((ReplicationPageWriteMessage)packet); + } + else if (packet.getType() == PacketImpl.REPLICATION_PAGE_EVENT) + { + handlePageEvent((ReplicationPageEventMessage)packet); + } + } catch (Exception e) { @@ -120,8 +145,7 @@ { Configuration config = server.getConfiguration(); - // TODO: this needs an executor - storage = new JournalStorageManager(config, null); + storage = new JournalStorageManager(config, server.getExecutorFactory().getExecutor()); storage.start(); bindingsJournal = storage.getBindingsJournal(); @@ -129,6 +153,15 @@ // We only need to load internal structures on the backup... storage.loadInternalOnly(); + + pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(config.getPagingDirectory(), + server.getExecutorFactory()), + storage, + server.getAddressSettingsRepository(), + false); + + pageManager.start(); + } /* (non-Javadoc) @@ -165,82 +198,70 @@ /** * @param packet */ - private void handleCommitRollback(final Packet packet) throws Exception + private void handleCommitRollback(final ReplicationCommitMessage packet) throws Exception { - ReplicationCommitMessage commitMessage = (ReplicationCommitMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(commitMessage.getJournalID()); - - if (commitMessage.isRollback()) + if (packet.isRollback()) { - journalToUse.appendRollbackRecord(commitMessage.getTxId(), false); + journalToUse.appendRollbackRecord(packet.getTxId(), false); } else { - journalToUse.appendCommitRecord(commitMessage.getTxId(), false); + journalToUse.appendCommitRecord(packet.getTxId(), false); } } /** * @param packet */ - private void handlePrepare(final Packet packet) throws Exception + private void handlePrepare(final ReplicationPrepareMessage packet) throws Exception { - ReplicationPrepareMessage prepareMessage = (ReplicationPrepareMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(prepareMessage.getJournalID()); - - journalToUse.appendPrepareRecord(prepareMessage.getTxId(), prepareMessage.getRecordData(), false); + journalToUse.appendPrepareRecord(packet.getTxId(), packet.getRecordData(), false); } /** * @param packet */ - private void handleAppendDeleteTX(final Packet packet) throws Exception + private void handleAppendDeleteTX(final ReplicationDeleteTXMessage packet) throws Exception { - ReplicationDeleteTXMessage deleteMessage = (ReplicationDeleteTXMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(deleteMessage.getJournalID()); - - journalToUse.appendDeleteRecordTransactional(deleteMessage.getTxId(), - deleteMessage.getId(), - deleteMessage.getRecordData()); + journalToUse.appendDeleteRecordTransactional(packet.getTxId(), packet.getId(), packet.getRecordData()); } /** * @param packet */ - private void handleAppendDelete(final Packet packet) throws Exception + private void handleAppendDelete(final ReplicationDeleteMessage packet) throws Exception { - ReplicationDeleteMessage deleteMessage = (ReplicationDeleteMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(deleteMessage.getJournalID()); - - journalToUse.appendDeleteRecord(deleteMessage.getId(), false); + journalToUse.appendDeleteRecord(packet.getId(), false); } /** * @param packet */ - private void handleAppendAddTXRecord(final Packet packet) throws Exception + private void handleAppendAddTXRecord(final ReplicationAddTXMessage packet) throws Exception { - ReplicationAddTXMessage addMessage = (ReplicationAddTXMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(addMessage.getJournalID()); - - if (addMessage.isUpdate()) + if (packet.isUpdate()) { - journalToUse.appendUpdateRecordTransactional(addMessage.getTxId(), - addMessage.getId(), - addMessage.getRecordType(), - addMessage.getRecordData()); + journalToUse.appendUpdateRecordTransactional(packet.getTxId(), + packet.getId(), + packet.getRecordType(), + packet.getRecordData()); } else { - journalToUse.appendAddRecordTransactional(addMessage.getTxId(), - addMessage.getId(), - addMessage.getRecordType(), - addMessage.getRecordData()); + journalToUse.appendAddRecordTransactional(packet.getTxId(), + packet.getId(), + packet.getRecordType(), + packet.getRecordData()); } } @@ -248,34 +269,109 @@ * @param packet * @throws Exception */ - private void handleAppendAddRecord(final Packet packet) throws Exception + private void handleAppendAddRecord(final ReplicationAddMessage packet) throws Exception { - ReplicationAddMessage addMessage = (ReplicationAddMessage)packet; + Journal journalToUse = getJournal(packet.getJournalID()); - Journal journalToUse = getJournal(addMessage.getJournalID()); - - if (addMessage.isUpdate()) + if (packet.isUpdate()) { if (trace) { - System.out.println("Endpoint appendUpdate id = " + addMessage.getId()); + System.out.println("Endpoint appendUpdate id = " + packet.getId()); } - journalToUse.appendUpdateRecord(addMessage.getId(), - addMessage.getRecordType(), - addMessage.getRecordData(), - false); + journalToUse.appendUpdateRecord(packet.getId(), packet.getRecordType(), packet.getRecordData(), false); } else { if (trace) { - System.out.println("Endpoint append id = " + addMessage.getId()); + System.out.println("Endpoint append id = " + packet.getId()); } - journalToUse.appendAddRecord(addMessage.getId(), addMessage.getRecordType(), addMessage.getRecordData(), false); + journalToUse.appendAddRecord(packet.getId(), packet.getRecordType(), packet.getRecordData(), false); } } /** + * @param packet + */ + private void handlePageEvent(final ReplicationPageEventMessage packet) throws Exception + { + ConcurrentMap<Integer, Page> pages = getPageMap(packet.getStoreName()); + + Page page = pages.remove(packet.getPageNumber()); + + if (page != null) + { + if (packet.isDelete()) + { + page.delete(); + } + else + { + page.close(); + } + } + + } + + /** + * @param packet + */ + private void handlePageWrite(final ReplicationPageWriteMessage packet) throws Exception + { + PagedMessage pgdMessage = packet.getPagedMessage(); + ServerMessage msg = pgdMessage.getMessage(storage); + Page page = getPage(msg.getDestination(), packet.getPageNumber()); + page.write(pgdMessage); + } + + private ConcurrentMap<Integer, Page> getPageMap(final SimpleString storeName) + { + ConcurrentMap<Integer, Page> resultIndex = pageIndex.get(storeName); + + if (resultIndex == null) + { + resultIndex = pageIndex.putIfAbsent(storeName, new ConcurrentHashMap<Integer, Page>()); + } + + return resultIndex; + } + + private Page getPage(final SimpleString storeName, final int pageId) throws Exception + { + ConcurrentMap<Integer, Page> map = getPageMap(storeName); + + Page page = map.get(pageId); + + if (page == null) + { + page = newPage(pageId, storeName, map); + } + + return page; + } + + /** + * @param pageId + * @param map + * @return + */ + private synchronized Page newPage(final int pageId, + final SimpleString storeName, + final ConcurrentMap<Integer, Page> map) throws Exception + { + Page page = map.get(pageId); + + if (page == null) + { + page = pageManager.getPageStore(storeName).createPage(pageId); + map.put(pageId, page); + } + + return page; + } + + /** * @param journalID * @return */ Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -21,6 +21,7 @@ import org.hornetq.core.client.impl.ConnectionManager; import org.hornetq.core.journal.EncodingSupport; import org.hornetq.core.logging.Logger; +import org.hornetq.core.paging.PagedMessage; import org.hornetq.core.remoting.Channel; import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.Packet; @@ -32,11 +33,14 @@ import org.hornetq.core.remoting.impl.wireformat.ReplicationCommitMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationDeleteTXMessage; +import org.hornetq.core.remoting.impl.wireformat.ReplicationPageEventMessage; +import org.hornetq.core.remoting.impl.wireformat.ReplicationPageWriteMessage; import org.hornetq.core.remoting.impl.wireformat.ReplicationPrepareMessage; import org.hornetq.core.remoting.spi.HornetQBuffer; import org.hornetq.core.replication.ReplicationManager; import org.hornetq.core.replication.ReplicationToken; import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.SimpleString; /** * A RepplicationManagerImpl @@ -217,6 +221,39 @@ } /* (non-Javadoc) + * @see org.hornetq.core.replication.ReplicationManager#pageClosed(org.hornetq.utils.SimpleString, int) + */ + public void pageClosed(final SimpleString storeName, final int pageNumber) + { + if (enabled) + { + sendReplicatePacket(new ReplicationPageEventMessage(storeName, pageNumber, false)); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.replication.ReplicationManager#pageDeleted(org.hornetq.utils.SimpleString, int) + */ + public void pageDeleted(final SimpleString storeName, final int pageNumber) + { + if (enabled) + { + sendReplicatePacket(new ReplicationPageEventMessage(storeName, pageNumber, true)); + } + } + + /* (non-Javadoc) + * @see org.hornetq.core.replication.ReplicationManager#pageWrite(org.hornetq.utils.SimpleString, int) + */ + public void pageWrite(final PagedMessage message, final int pageNumber) + { + if (enabled) + { + sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber)); + } + } + + /* (non-Javadoc) * @see org.hornetq.core.server.HornetQComponent#isStarted() */ public synchronized boolean isStarted() Modified: branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java =================================================================== --- branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-10-08 17:09:12 UTC (rev 8069) +++ branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/paging/PageCrashTest.java 2009-10-08 21:40:37 UTC (rev 8070) @@ -310,7 +310,7 @@ } @Override - protected Page createPage(final int page) throws Exception + public Page createPage(final int page) throws Exception { Page originalPage = super.createPage(page);