[hornetq-commits] JBoss hornetq SVN: r12012 - in trunk: hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal and 8 other directories.

Author: borges Date: 2012-01-12 07:08:26 -0500 (Thu, 12 Jan 2012) New Revision: 12012 Added: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageInSync.java trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicatedLargeMessage.java Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java trunk/hornetq-core/src/main/java/org/hornetq/core/server/LargeServerMessage.java trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java Log: HORNETQ-720 Fix synchronization issues with Large Messages Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/StorageManager.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -26,6 +26,7 @@ import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.Journal; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -161,6 +162,8 @@ */ LargeServerMessage createLargeMessage(long id, MessageInternal message) throws Exception; + SequentialFile createFileForLargeMessage(final long messageID, String extension); + void prepare(long txID, Xid xid) throws Exception; void commit(long txID) throws Exception; @@ -274,4 +277,11 @@ * Typical scenario is a broken connection. */ void stopReplication(); + + /** + * @param appendFile + * @param messageID + * @param bytes + */ + void addBytesToLargeMessage(SequentialFile appendFile, long messageID, byte[] bytes) throws Exception; } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -381,7 +381,7 @@ messageFiles = prepareJournalForCopy(originalMessageJournal, JournalContent.MESSAGES, nodeID); bindingsFiles = prepareJournalForCopy(originalBindingsJournal, JournalContent.BINDINGS, nodeID); pageFilesToSync = getPageInformationForSync(pagingManager); - largeMessageFilesToSync = getLargeMessageInformation(); + largeMessageFilesToSync = getLargeMessageInformation(); } finally { @@ -469,14 +469,14 @@ private Map<SimpleString, Collection<Integer>> getPageInformationForSync(PagingManager pagingManager) throws Exception { - Map<SimpleString, Collection<Integer>> info = new HashMap<SimpleString, Collection<Integer>>(); - for (SimpleString storeName : pagingManager.getStoreNames()) - { - PagingStore store = pagingManager.getPageStore(storeName); - info.put(storeName, store.getCurrentIds()); - store.forceAnotherPage(); - } - return info; + Map<SimpleString, Collection<Integer>> info = new HashMap<SimpleString, Collection<Integer>>(); + for (SimpleString storeName : pagingManager.getStoreNames()) + { + PagingStore store = pagingManager.getPageStore(storeName); + info.put(storeName, store.getCurrentIds()); + store.forceAnotherPage(); + } + return info; } private void sendLargeMessageFiles(Map<String, Long> largeMessageFilesToSync) throws Exception @@ -509,30 +509,35 @@ */ private Map<String, Long> getLargeMessageInformation() throws Exception { - Map<String, Long> largeMessages = new HashMap<String, Long>(); - List<String> filenames = largeMessagesFactory.listFiles("msg"); - for (String filename : filenames) - { - SequentialFile seqFile = largeMessagesFactory.createSequentialFile(filename, 1); - long size = seqFile.size(); - largeMessages.put(filename, size); - } - return largeMessages; + final String prefix = "msg"; + Map<String, Long> largeMessages = new HashMap<String, Long>(); + List<String> filenames = largeMessagesFactory.listFiles(prefix); + + List<Long> idList = new ArrayList<Long>(); + for (String filename : filenames) + { + idList.add(Long.valueOf(filename.substring(0, filename.length() - (prefix.length() + 1)))); + SequentialFile seqFile = largeMessagesFactory.createSequentialFile(filename, 1); + long size = seqFile.size(); + largeMessages.put(filename, size); + } + replicator.sendLargeMessageIdListMessage(idList); + return largeMessages; } /** * Send an entire journal file to a replicating backup server. */ private void sendJournalFile(JournalFile[] journalFiles, JournalContent type) throws Exception - { - for (JournalFile jf : journalFiles) - { - if (!started) - return; - replicator.syncJournalFile(jf, type); - jf.setCanReclaim(true); - } - } + { + for (JournalFile jf : journalFiles) + { + if (!started) + return; + replicator.syncJournalFile(jf, type); + jf.setCanReclaim(true); + } + } private JournalFile[] prepareJournalForCopy(Journal journal, JournalContent contentType, String nodeID) throws Exception @@ -546,23 +551,23 @@ @Override public void waitOnOperations() throws Exception { - if (!started) - { - JournalStorageManager.log.warn("Server is stopped"); - throw new IllegalStateException("Server is stopped"); - } - waitOnOperations(0); + if (!started) + { + JournalStorageManager.log.warn("Server is stopped"); + throw new IllegalStateException("Server is stopped"); + } + waitOnOperations(0); } @Override public boolean waitOnOperations(final long timeout) throws Exception { - if (!started) - { - JournalStorageManager.log.warn("Server is stopped"); - throw new IllegalStateException("Server is stopped"); - } - return getContext().waitCompletion(timeout); + if (!started) + { + JournalStorageManager.log.warn("Server is stopped"); + throw new IllegalStateException("Server is stopped"); + } + return getContext().waitCompletion(timeout); } @Override @@ -665,7 +670,8 @@ return new LargeServerMessageImpl(this); } - protected final void addBytesToLargeMessage(final SequentialFile file, final long messageId, final byte[] bytes) + public final void + addBytesToLargeMessage(final SequentialFile file, final long messageId, final byte[] bytes) throws Exception { readLock(); @@ -2241,7 +2247,7 @@ * @param messageID * @return */ - SequentialFile createFileForLargeMessage(final long messageID, String extension) + public SequentialFile createFileForLargeMessage(final long messageID, String extension) { return largeMessagesFactory.createSequentialFile(messageID + extension, -1); } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageImpl.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -54,7 +54,6 @@ private long pendingRecordID = -1; private boolean paged; - private boolean replicationSync; // We should only use the NIO implementation on the Journal private SequentialFile file; @@ -107,9 +106,7 @@ paged = true; } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#addBytes(byte[]) - */ + @Override public synchronized void addBytes(final byte[] bytes) throws Exception { validateFile(); @@ -239,6 +236,7 @@ return true; } + @Override public synchronized void deleteFile() throws Exception { validateFile(); @@ -430,14 +428,10 @@ private String getExtension() { - if (replicationSync) - return ".sync"; return durable ? ".msg" : ".tmp"; } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#setLinkedMessage(org.hornetq.core.server.LargeServerMessage) - */ + @Override public void setLinkedMessage(final LargeServerMessage message) { if (file != null) @@ -521,18 +515,10 @@ return bytesRead; } - /* (non-Javadoc) - * @see org.hornetq.core.message.BodyEncoder#getLargeBodySize() - */ + @Override public long getLargeBodySize() { return bodySize; } } - - @Override - public void setReplicationSync(boolean sync) - { - replicationSync = sync; - } } Added: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageInSync.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageInSync.java (rev 0) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/journal/LargeServerMessageInSync.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -0,0 +1,135 @@ +/** + * + */ +package org.hornetq.core.persistence.impl.journal; + +import java.nio.ByteBuffer; + +import org.hornetq.api.core.HornetQException; +import org.hornetq.core.journal.SequentialFile; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.core.replication.ReplicatedLargeMessage; +import org.hornetq.core.server.LargeServerMessage; + +public final class LargeServerMessageInSync implements ReplicatedLargeMessage +{ + private static final Logger log = Logger.getLogger(LargeServerMessageInSync.class); + + private static final String SYNC_EXTENSION = ".sync"; + private final LargeServerMessage mainLM; + private final StorageManager storageManager; + private SequentialFile appendFile; + private boolean syncDone; + + /** + * @param storageManager + */ + public LargeServerMessageInSync(StorageManager storageManager) + { + mainLM = storageManager.createLargeMessage(); + this.storageManager = storageManager; + } + + public synchronized void joinSyncedData(ByteBuffer buffer) throws Exception + { + SequentialFile mainSeqFile = mainLM.getFile(); + if (appendFile != null) + { + for (;;) + { + buffer.rewind(); + int size = appendFile.read(buffer); + mainSeqFile.writeInternal(buffer); + if (size < buffer.capacity()) + { + break; + } + } + deleteAppendFile(); + } + syncDone = true; + } + + public SequentialFile getSyncFile() throws HornetQException + { + return mainLM.getFile(); + } + + @Override + public void setDurable(boolean durable) + { + mainLM.setDurable(durable); + } + + @Override + public synchronized void setMessageID(long id) + { + mainLM.setMessageID(id); + } + + @Override + public synchronized void releaseResources() + { + mainLM.releaseResources(); + if (appendFile != null && appendFile.isOpen()) + { + try + { + appendFile.close(); + } + catch (Exception e) + { + log.error(e.getMessage(), e); + } + } + } + + @Override + public synchronized void deleteFile() throws Exception + { + try + { + mainLM.deleteFile(); + } + finally + { + deleteAppendFile(); + } + } + + /** + * @throws Exception + */ + private void deleteAppendFile() throws Exception + { + if (appendFile != null) + { + if (appendFile.isOpen()) + appendFile.close(); + appendFile.delete(); + } + } + + @Override + public synchronized void addBytes(byte[] bytes) throws Exception + { + if (syncDone) + { + mainLM.addBytes(bytes); + return; + } + + if (appendFile == null) + { + appendFile = storageManager.createFileForLargeMessage(mainLM.getMessageID(), SYNC_EXTENSION); + } + + if (!appendFile.isOpen()) + { + appendFile.open(); + } + storageManager.addBytesToLargeMessage(appendFile, mainLM.getMessageID(), bytes); + } + +} Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageLargeServerMessage.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -138,9 +138,7 @@ } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#incrementDelayDeletionCount() - */ + @Override public void incrementDelayDeletionCount() { @@ -158,51 +156,25 @@ return "LargeServerMessage[messageID=" + messageID + ", durable=" + durable + ", address=" + getAddress() + ",properties=" + properties.toString() + "]"; } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#setPaged() - */ + @Override public void setPaged() { } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#setPendingRecordID(long) - */ + @Override public void setPendingRecordID(long pendingRecordID) { } - /* (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#getPendingRecordID() - */ + @Override public long getPendingRecordID() { return -1; } - /* - * (non-Javadoc) - * @see org.hornetq.core.server.LargeServerMessage#setReplicationSync(boolean) - */ @Override - public void setReplicationSync(boolean sync) - { - // TODO Auto-generated method stub - - } - - @Override public SequentialFile getFile() { return null; } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -28,6 +28,7 @@ import org.hornetq.core.journal.IOAsyncTask; import org.hornetq.core.journal.Journal; import org.hornetq.core.journal.JournalLoadInformation; +import org.hornetq.core.journal.SequentialFile; import org.hornetq.core.message.impl.MessageInternal; import org.hornetq.core.paging.PageTransactionInfo; import org.hornetq.core.paging.PagedMessage; @@ -351,92 +352,67 @@ return true; } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#setReplicator(org.hornetq.core.replication.ReplicationManager) - */ public void setReplicator(final ReplicationManager replicator) { throw new IllegalStateException("Null Persistence should never be used as replicated"); } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#afterCompleteOperations(org.hornetq.core.journal.IOCompletion) - */ public void afterCompleteOperations(final IOAsyncTask run) { run.done(); } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#waitOnOperations() - */ + @Override public void waitOnOperations() throws Exception { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#getContext() - */ + @Override public OperationContext getContext() { return dummyContext; } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#newContext() - */ + @Override public OperationContext newContext(final Executor executor) { return dummyContext; } - + @Override public OperationContext newSingleThreadContext() { return dummyContext; } - - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#setContext(org.hornetq.core.persistence.OperationContext) - */ + @Override public void setContext(final OperationContext context) { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#clearContext() - */ + @Override public void clearContext() { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#recoverAddressSettings() - */ + @Override public List<PersistedAddressSetting> recoverAddressSettings() throws Exception { return Collections.emptyList(); } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#storeAddressSetting(org.hornetq.core.persistconfig.PersistedAddressSetting) - */ + @Override public void storeAddressSetting(PersistedAddressSetting addressSetting) throws Exception { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#recoverPersistedRoles() - */ + @Override public List<PersistedRoles> recoverPersistedRoles() throws Exception { return Collections.emptyList(); } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#storeSecurityRoles(org.hornetq.core.persistconfig.PersistedRoles) - */ + @Override public void storeSecurityRoles(PersistedRoles persistedRoles) throws Exception { } @@ -512,54 +488,44 @@ { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#deletePageCounter(long, long) - */ + @Override public void deletePageCounter(long txID, long recordID) throws Exception { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#storePageCounterInc(long, long, int) - */ + @Override public long storePageCounterInc(long txID, long queueID, int add) throws Exception { return 0; } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#storePageCounterInc(long, int) - */ + @Override public long storePageCounterInc(long queueID, int add) throws Exception { return 0; } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#commit(long, boolean) - */ + @Override public void commit(long txID, boolean lineUpContext) throws Exception { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#lineUpContext() - */ + @Override public void lineUpContext() { } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#confirmPendingLargeMessageTX(org.hornetq.core.transaction.Transaction, long, long) - */ + @Override public void confirmPendingLargeMessageTX(Transaction transaction, long messageID, long recordID) throws Exception { } + @Override public void confirmPendingLargeMessage(long recordID) throws Exception { } + @Override public void stop(boolean ioCriticalError) throws Exception { } @@ -598,4 +564,16 @@ { // no-op } + + @Override + public SequentialFile createFileForLargeMessage(long messageID, String extension) + { + throw new UnsupportedOperationException(); + } + + @Override + public void addBytesToLargeMessage(SequentialFile appendFile, long messageID, byte[] bytes) throws Exception + { + // no-op + } } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/wireformat/ReplicationStartSyncMessage.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -1,5 +1,9 @@ package org.hornetq.core.protocol.core.impl.wireformat; +import java.security.InvalidParameterException; +import java.util.Arrays; +import java.util.List; + import org.hornetq.api.core.HornetQBuffer; import org.hornetq.core.journal.impl.JournalFile; import org.hornetq.core.persistence.impl.journal.JournalStorageManager.JournalContent; @@ -14,15 +18,57 @@ public class ReplicationStartSyncMessage extends PacketImpl { private long[] ids; - private JournalContent journalType; + private SyncDataType dataType; private boolean synchronizationIsFinished; private String nodeID; + public enum SyncDataType + { + JournalBindings(JournalContent.BINDINGS.typeByte), + JournalMessages(JournalContent.MESSAGES.typeByte), + LargeMessages((byte)2); + + private byte code; + + private SyncDataType(byte code) + { + this.code = code; + } + + public static JournalContent getJournalContentType(SyncDataType dataType) + { + return JournalContent.getType(dataType.code); + } + + public static SyncDataType getDataType(byte code) + { + if (code == JournalBindings.code) + return JournalBindings; + if (code == JournalMessages.code) + return JournalMessages; + if (code == LargeMessages.code) + return LargeMessages; + throw new InvalidParameterException("invalid byte: " + code); + } + } + public ReplicationStartSyncMessage() { super(REPLICATION_START_FINISH_SYNC); } + public ReplicationStartSyncMessage(List<Long> filenames) + { + this(); + ids = new long[filenames.size()]; + for (int i = 0; i < filenames.size(); i++) + { + ids[i] = filenames.get(i); + } + dataType = SyncDataType.LargeMessages; + nodeID = ""; // this value will be ignored + } + public ReplicationStartSyncMessage(String nodeID) { this(); @@ -40,7 +86,17 @@ { ids[i] = datafiles[i].getFileID(); } - journalType = contentType; + switch (contentType) + { + case MESSAGES: + dataType = SyncDataType.JournalMessages; + break; + case BINDINGS: + dataType = SyncDataType.JournalBindings; + break; + default: + throw new IllegalArgumentException(); + } } @Override @@ -50,7 +106,7 @@ buffer.writeString(nodeID); if (synchronizationIsFinished) return; - buffer.writeByte(journalType.typeByte); + buffer.writeByte(dataType.code); buffer.writeInt(ids.length); for (long id : ids) { @@ -67,7 +123,7 @@ { return; } - journalType = JournalContent.getType(buffer.readByte()); + dataType = SyncDataType.getDataType(buffer.readByte()); int length = buffer.readInt(); ids = new long[length]; for (int i = 0; i < length; i++) @@ -85,9 +141,9 @@ return synchronizationIsFinished; } - public JournalContent getJournalContentType() + public SyncDataType getDataType() { - return journalType; + return dataType; } public long[] getFileIds() @@ -99,4 +155,58 @@ { return nodeID; } + + @Override + public int hashCode() + { + final int prime = 31; + int result = super.hashCode(); + result = prime * result + Arrays.hashCode(ids); + result = prime * result + ((dataType == null) ? 0 : dataType.hashCode()); + result = prime * result + ((nodeID == null) ? 0 : nodeID.hashCode()); + result = prime * result + (synchronizationIsFinished ? 1231 : 1237); + return result; + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) + { + return true; + } + if (!super.equals(obj)) + { + return false; + } + if (!(obj instanceof ReplicationStartSyncMessage)) + { + return false; + } + ReplicationStartSyncMessage other = (ReplicationStartSyncMessage)obj; + if (!Arrays.equals(ids, other.ids)) + { + return false; + } + if (dataType != other.dataType) + { + return false; + } + if (nodeID == null) + { + if (other.nodeID != null) + { + return false; + } + } + else if (!nodeID.equals(other.nodeID)) + { + return false; + } + if (synchronizationIsFinished != other.synchronizationIsFinished) + { + return false; + } + return true; + } } Added: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicatedLargeMessage.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicatedLargeMessage.java (rev 0) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicatedLargeMessage.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -0,0 +1,38 @@ +/** + * + */ +package org.hornetq.core.replication; + +import org.hornetq.core.server.LargeServerMessage; + +/** + * + */ +public interface ReplicatedLargeMessage +{ + /** + * @see LargeServerMessage#setDurable(boolean) + */ + void setDurable(boolean b); + + /** + * @see LargeServerMessage#setMessageID(long) + */ + void setMessageID(long id); + + /** + * @see LargeServerMessage#releaseResources() + */ + void releaseResources(); + + /** + * @see LargeServerMessage#deleteFile() + */ + void deleteFile() throws Exception; + + /** + * @see LargeServerMessage#addBytes(byte[]) + */ + void addBytes(byte[] body) throws Exception; + +} Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationEndpoint.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -44,6 +44,7 @@ import org.hornetq.core.paging.impl.PagingStoreFactoryNIO; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.persistence.impl.journal.JournalStorageManager.JournalContent; +import org.hornetq.core.persistence.impl.journal.LargeServerMessageInSync; import org.hornetq.core.protocol.core.Channel; import org.hornetq.core.protocol.core.ChannelHandler; import org.hornetq.core.protocol.core.Packet; @@ -64,9 +65,9 @@ import org.hornetq.core.protocol.core.impl.wireformat.ReplicationPrepareMessage; import org.hornetq.core.protocol.core.impl.wireformat.ReplicationResponseMessage; import org.hornetq.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage; +import org.hornetq.core.protocol.core.impl.wireformat.ReplicationStartSyncMessage.SyncDataType; import org.hornetq.core.protocol.core.impl.wireformat.ReplicationSyncFileMessage; import org.hornetq.core.server.HornetQComponent; -import org.hornetq.core.server.LargeServerMessage; import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.impl.HornetQServerImpl; import org.hornetq.core.server.impl.QuorumManager; @@ -97,7 +98,6 @@ /** Files reserved in each journal for synchronization of existing data from the 'live' server. */ private final Map<JournalContent, Map<Long, JournalSyncFile>> filesReservedForSync = new HashMap<JournalContent, Map<Long, JournalSyncFile>>(); - private final Map<Long, LargeServerMessage> largeMessagesOnSync = new HashMap<Long, LargeServerMessage>(); /** * Used to hold the real Journals before the backup is synchronized. This field should be @@ -111,8 +111,8 @@ private final ConcurrentMap<SimpleString, ConcurrentMap<Integer, Page>> pageIndex = new ConcurrentHashMap<SimpleString, ConcurrentMap<Integer, Page>>(); - private final ConcurrentMap<Long, LargeServerMessage> largeMessages = - new ConcurrentHashMap<Long, LargeServerMessage>(); + private final ConcurrentMap<Long, ReplicatedLargeMessage> largeMessages = + new ConcurrentHashMap<Long, ReplicatedLargeMessage>(); // Used on tests, to simulate failures on delete pages private boolean deletePages = true; @@ -285,8 +285,7 @@ return; } - // This could be null if the backup server is being - // shut down without any live server connecting here + // Channel may be null if there isn't a connection to a live server if (channel != null) { channel.close(); @@ -309,18 +308,12 @@ pageIndex.clear(); - for (LargeServerMessage largeMessage : largeMessages.values()) + for (ReplicatedLargeMessage largeMessage : largeMessages.values()) { largeMessage.releaseResources(); } largeMessages.clear(); - for (LargeServerMessage largeMessage : largeMessagesOnSync.values()) - { - largeMessage.releaseResources(); - } - largeMessagesOnSync.clear(); - for (Entry<JournalContent, Map<Long, JournalSyncFile>> entry : filesReservedForSync.entrySet()) { for (JournalSyncFile filesReserved : entry.getValue().values()) @@ -444,42 +437,17 @@ journal.synchronizationUnlock(); } } - synchronized (largeMessagesOnSync) + ByteBuffer buffer = ByteBuffer.allocate(4 * 1024); + for (Entry<Long, ReplicatedLargeMessage> entry : largeMessages.entrySet()) { - synchronized (largeMessages) + ReplicatedLargeMessage lm = entry.getValue(); + if (lm instanceof LargeServerMessageInSync) { - ByteBuffer buffer = ByteBuffer.allocate(10 * 1024); - for (Entry<Long, LargeServerMessage> entry : largeMessages.entrySet()) - { - Long id = entry.getKey(); - LargeServerMessage lm = entry.getValue(); - if (largeMessagesOnSync.containsKey(id)) - { - SequentialFile sq = lm.getFile(); - LargeServerMessage mainLM = largeMessagesOnSync.get(id); - SequentialFile mainSeqFile = mainLM.getFile(); - for (;;) - { - buffer.rewind(); - int size = sq.read(buffer); - mainSeqFile.writeInternal(buffer); - if (size < buffer.capacity()) - { - break; - } - } - } - else - { - // these are large-messages created after sync started - largeMessagesOnSync.put(id, lm); - } - } - largeMessages.clear(); - largeMessages.putAll(largeMessagesOnSync); - largeMessagesOnSync.clear(); + LargeServerMessageInSync lmSync = (LargeServerMessageInSync)lm; + lmSync.joinSyncedData(buffer); } } + journalsHolder = null; quorumManager.setLiveID(liveID); server.setRemoteBackupUpToDate(liveID); @@ -487,7 +455,7 @@ return; } - private void handleReplicationSynchronization(ReplicationSyncFileMessage msg) throws Exception + private synchronized void handleReplicationSynchronization(ReplicationSyncFileMessage msg) throws Exception { Long id = Long.valueOf(msg.getId()); byte[] data = msg.getData(); @@ -496,24 +464,19 @@ { case LARGE_MESSAGE: { - synchronized (largeMessagesOnSync) + ReplicatedLargeMessage largeMessage = lookupLargeMessage(id, false); + if (!(largeMessage instanceof LargeServerMessageInSync)) { - LargeServerMessage largeMessage = largeMessagesOnSync.get(id); - if (largeMessage == null) - { - largeMessage = storage.createLargeMessage(); - largeMessage.setDurable(true); - largeMessage.setMessageID(id); - largeMessagesOnSync.put(id, largeMessage); - } - channel = largeMessage.getFile(); + log.error("large message sync: largeMessage instance is incompatible with it, ignoring data"); + return; } + LargeServerMessageInSync largeMessageInSync=(LargeServerMessageInSync)largeMessage; + channel = largeMessageInSync.getSyncFile(); break; } case PAGE: { Page page = getPage(msg.getPageStore(), (int)msg.getId()); - channel = page.getFile(); break; } @@ -565,31 +528,51 @@ return; } - final Journal journal = journalsHolder.get(packet.getJournalContentType()); + synchronized (this) { if (!started) return; - if (packet.getNodeID() != null) + switch (packet.getDataType()) { - quorumManager.setLiveID(packet.getNodeID()); - } - Map<Long, JournalSyncFile> mapToFill = filesReservedForSync.get(packet.getJournalContentType()); - log.info("Journal " + packet.getJournalContentType() + ". Reserving fileIDs for synchronization: " + - Arrays.toString(packet.getFileIds())); + case LargeMessages: + for (long msgID : packet.getFileIds()) + { + createLargeMessage(msgID, true); + } + break; + case JournalBindings: + case JournalMessages: - for (Entry<Long, JournalFile> entry : journal.createFilesForBackupSync(packet.getFileIds()).entrySet()) - { - mapToFill.put(entry.getKey(), new JournalSyncFile(entry.getValue())); + final JournalContent journalContent = SyncDataType.getJournalContentType(packet.getDataType()); + final Journal journal = journalsHolder.get(journalContent); + + if (packet.getNodeID() != null) + { + // At the start of replication, we still do not know which is the nodeID that the live uses. + // This is the point where the backup gets this information. + quorumManager.setLiveID(packet.getNodeID()); + } + Map<Long, JournalSyncFile> mapToFill = filesReservedForSync.get(journalContent); + log.info("Journal " + journalContent + ". Reserving fileIDs for synchronization: " + + Arrays.toString(packet.getFileIds())); + + for (Entry<Long, JournalFile> entry : journal.createFilesForBackupSync(packet.getFileIds()).entrySet()) + { + mapToFill.put(entry.getKey(), new JournalSyncFile(entry.getValue())); + } + FileWrapperJournal syncJournal = new FileWrapperJournal(journal); + registerJournal(journalContent.typeByte, syncJournal); + break; + default: + throw new HornetQException(HornetQException.INTERNAL_ERROR, "unhandled data type!"); } - FileWrapperJournal syncJournal = new FileWrapperJournal(journal); - registerJournal(packet.getJournalContentType().typeByte, syncJournal); } } private void handleLargeMessageEnd(final ReplicationLargeMessageEndMessage packet) { - LargeServerMessage message = lookupLargeMessage(packet.getMessageId(), true); + ReplicatedLargeMessage message = lookupLargeMessage(packet.getMessageId(), true); if (message != null) { @@ -609,7 +592,7 @@ */ private void handleLargeMessageWrite(final ReplicationLargeMessageWriteMessage packet) throws Exception { - LargeServerMessage message = lookupLargeMessage(packet.getMessageId(), false); + ReplicatedLargeMessage message = lookupLargeMessage(packet.getMessageId(), false); if (message != null) { message.addBytes(packet.getBody()); @@ -624,9 +607,9 @@ compareJournalInformation(request.getJournalInformation()); } - private LargeServerMessage lookupLargeMessage(final long messageId, final boolean delete) + private ReplicatedLargeMessage lookupLargeMessage(final long messageId, final boolean delete) { - LargeServerMessage message; + ReplicatedLargeMessage message; if (delete) { @@ -635,18 +618,6 @@ else { message = largeMessages.get(messageId); - if (message == null) - { - synchronized (largeMessages) - { - if (!server.isRemoteBackupUpToDate()) - { - // in case we need to append data to a file while still sync'ing the backup - createLargeMessage(messageId, true); - message = largeMessages.get(messageId); - } - } - } } if (message == null) @@ -671,10 +642,14 @@ private void createLargeMessage(final long id, boolean sync) { - LargeServerMessage msg = storage.createLargeMessage(); + ReplicatedLargeMessage msg; + if (sync) + msg = new LargeServerMessageInSync(storage); + else + msg = storage.createLargeMessage(); + msg.setDurable(true); msg.setMessageID(id); - msg.setReplicationSync(sync); largeMessages.put(id, msg); } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/ReplicationManager.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -13,6 +13,7 @@ package org.hornetq.core.replication; +import java.util.List; import java.util.Set; import org.hornetq.api.core.HornetQException; @@ -125,4 +126,13 @@ * @throws Exception */ void syncPages(SequentialFile file, long id, SimpleString pageStore) throws Exception; + + /** + * Reserves several LargeMessage IDs in the backup. + * <p> + * Doing this before hand removes the need of synchronizing large-message deletes with the + * largeMessageSyncList. + * @param largeMessageIDs + */ + void sendLargeMessageIdListMessage(List<Long> largeMessageIDs); } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -17,6 +17,7 @@ import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.util.LinkedHashSet; +import java.util.List; import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; @@ -251,9 +252,7 @@ } } - /* (non-Javadoc) - * @see org.hornetq.core.replication.ReplicationManager#largeMessageBegin(byte[]) - */ + @Override public void largeMessageBegin(final long messageId) { if (enabled) @@ -270,9 +269,7 @@ } } - /* (non-Javadoc) - * @see org.hornetq.core.replication.ReplicationManager#largeMessageWrite(long, byte[]) - */ + @Override public void largeMessageWrite(final long messageId, final byte[] body) { if (enabled) @@ -596,4 +593,12 @@ if (enabled) sendReplicatePacket(new ReplicationStartSyncMessage(nodeID)); } + + @Override + public void sendLargeMessageIdListMessage(List<Long> largeMessageIDs) + { + if (enabled) + sendReplicatePacket(new ReplicationStartSyncMessage(largeMessageIDs)); + + } } Modified: trunk/hornetq-core/src/main/java/org/hornetq/core/server/LargeServerMessage.java =================================================================== --- trunk/hornetq-core/src/main/java/org/hornetq/core/server/LargeServerMessage.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-core/src/main/java/org/hornetq/core/server/LargeServerMessage.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -15,21 +15,22 @@ import org.hornetq.api.core.HornetQException; import org.hornetq.core.journal.SequentialFile; +import org.hornetq.core.replication.ReplicatedLargeMessage; /** * A LargeMessage * * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> */ -public interface LargeServerMessage extends ServerMessage +public interface LargeServerMessage extends ServerMessage, ReplicatedLargeMessage { void addBytes(byte[] bytes) throws Exception; /** When a large message is copied (e.g. ExpiryQueue) instead of copying the file, we specify a link between the messages */ void setLinkedMessage(LargeServerMessage message); - + void setPendingRecordID(long pendingRecordID); - + long getPendingRecordID(); boolean isFileExists() throws Exception; @@ -50,13 +51,6 @@ void decrementDelayDeletionCount() throws Exception; /** - * This method only has relevance in a backup server. - * @param sync {@code true} if this file is meant for appends of a message that needs to be - * sync'ed with the live. - */ - void setReplicationSync(boolean sync); - - /** * @return * @throws HornetQException */ Modified: trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java =================================================================== --- trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/hornetq-journal/src/main/java/org/hornetq/core/journal/impl/JournalImpl.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -2725,12 +2725,13 @@ /** * @param completeTransaction If the appendRecord is for a prepare or commit, where we should * update the number of pendingTransactions on the current file + * @throws Exception */ private JournalFile appendRecord(final JournalInternalRecord encoder, final boolean completeTransaction, final boolean sync, final JournalTransaction tx, - final IOAsyncTask parameterCallback) throws Exception + final IOAsyncTask parameterCallback) throws Exception { checkJournalIsLoaded(); Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java =================================================================== --- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncJournalTest.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -24,7 +24,7 @@ private static final int BACKUP_WAIT_TIME = 20; private ServerLocatorInternal locator; - private ClientSessionFactoryInternal sessionFactory; + protected ClientSessionFactoryInternal sessionFactory; private ClientSession session; private ClientProducer producer; private BackupSyncDelay syncDelay; Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java =================================================================== --- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/cluster/failover/BackupSyncLargeMessageTest.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -40,12 +40,27 @@ getAllMessageFileIds(dir).size()); createProducerSendSomeMessages(); startBackupFinishSyncing(); - // File dir = new - // File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory()); receiveMsgsInRange(0, n_msgs / 2); assertEquals("we really ought to delete these after delivery", n_msgs / 2, getAllMessageFileIds(dir).size()); } + public void testDeleteLargeMessagesDuringSync() throws Exception + { + File dir = new File(backupServer.getServer().getConfiguration().getLargeMessagesDirectory()); + assertEquals("Should not have any large messages... previous test failed to clean up?", 0, + getAllMessageFileIds(dir).size()); + createProducerSendSomeMessages(); + + backupServer.start(); + waitForComponent(backupServer.getServer(), 5); + receiveMsgsInRange(0, n_msgs / 2); + + startBackupFinishSyncing(); + backupServer.stop(); + + assertEquals("we really ought to delete these after delivery", n_msgs / 2, getAllMessageFileIds(dir).size()); + } + private Set<Long> getAllMessageFileIds(File dir) { Set<Long> idsOnBkp = new HashSet<Long>(); Modified: trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java =================================================================== --- trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java 2012-01-12 12:07:32 UTC (rev 12011) +++ trunk/tests/integration-tests/src/test/java/org/hornetq/tests/integration/persistence/DeleteMessagesOnStartupTest.java 2012-01-12 12:08:26 UTC (rev 12012) @@ -101,14 +101,4 @@ }; } - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - - }