You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/03/24 22:23:10 UTC
[activemq-artemis] branch master updated: ARTEMIS-1975 Moving
encode cache towards AMQPLargeMessage
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new a4489a3 ARTEMIS-1975 Moving encode cache towards AMQPLargeMessage
a4489a3 is described below
commit a4489a322e8435929f46f18713edd195a7d3f9ac
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Mar 24 18:17:27 2020 -0400
ARTEMIS-1975 Moving encode cache towards AMQPLargeMessage
Using a property on AMQPLargeMessage instead of a ThreadLocal
This was causing issues on the journal as the message may transverse different threads on the journal.
---
.../protocol/amqp/broker/AMQPLargeMessage.java | 7 ++++
.../amqp/broker/AMQPLargeMessagePersister.java | 22 ++++---------
.../impl/journal/LargeServerMessageImpl.java | 4 +--
.../SimpleStreamingLargeMessageTest.java | 38 +++++++++++-----------
4 files changed, 35 insertions(+), 36 deletions(-)
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java
index d6daac6..7937f7f 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java
@@ -19,6 +19,7 @@ package org.apache.activemq.artemis.protocol.amqp.broker;
import java.nio.ByteBuffer;
+import io.netty.buffer.ByteBuf;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
@@ -74,6 +75,12 @@ public class AMQPLargeMessage extends AMQPMessage implements LargeServerMessage
}
}
}
+
+ /**
+ * AMQPLargeMessagePersister will save the buffer here.
+ * */
+ volatile ByteBuf temporaryBuffer;
+
private final LargeBody largeBody;
/**
* We control durability on a separate property here, as we need to know if it's durable ahead of the header parsing.
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java
index bdafcba..cffda5d 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java
@@ -34,11 +34,6 @@ import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPLarg
public class AMQPLargeMessagePersister extends MessagePersister {
private static final Logger log = Logger.getLogger(AMQPLargeMessagePersister.class);
- // We need to save the encoder ahead of time
- // as we need to know the exact size of the Encoding
- // so we store the savedBuffer on the getEncodeSize before we actually store it
- private static final ThreadLocal<ByteBuf> savedBuffer = new ThreadLocal<>();
-
public static final byte ID = AMQPLargeMessagePersister_ID;
public static AMQPLargeMessagePersister theInstance;
@@ -72,14 +67,12 @@ public class AMQPLargeMessagePersister extends MessagePersister {
}
private ByteBuf getSavedEncodeBuffer(Message record) {
- ByteBuf buf = savedBuffer.get();
- if (buf == null) {
- AMQPLargeMessage largeMessage = (AMQPLargeMessage)record;
- buf = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode());
- largeMessage.saveEncoding(buf);
- savedBuffer.set(buf);
+ AMQPLargeMessage largeMessage = (AMQPLargeMessage)record;
+ if (largeMessage.temporaryBuffer == null) {
+ largeMessage.temporaryBuffer = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode());
+ largeMessage.saveEncoding(largeMessage.temporaryBuffer);
}
- return buf;
+ return largeMessage.temporaryBuffer;
}
/**
@@ -89,7 +82,7 @@ public class AMQPLargeMessagePersister extends MessagePersister {
public void encode(ActiveMQBuffer buffer, Message record) {
super.encode(buffer, record);
- AMQPMessage msgEncode = (AMQPMessage) record;
+ AMQPLargeMessage msgEncode = (AMQPLargeMessage) record;
buffer.writeLong(record.getMessageID());
buffer.writeBoolean(record.isDurable());
@@ -106,8 +99,7 @@ public class AMQPLargeMessagePersister extends MessagePersister {
ByteBuf savedEncodeBuffer = getSavedEncodeBuffer(record);
buffer.writeBytes(savedEncodeBuffer, 0, savedEncodeBuffer.writerIndex());
savedEncodeBuffer.release();
-
- savedBuffer.set(null);
+ msgEncode.temporaryBuffer = null;
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
index 3287974..3ab3f21 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
@@ -255,8 +255,8 @@ public final class LargeServerMessageImpl extends CoreMessage implements CoreLar
@Override
public Message copy() {
SequentialFile newfile = storageManager.createFileForLargeMessage(messageID, durable);
-
- Message newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID);
+ LargeServerMessageImpl newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID);
+ newMessage.setParentRef(this);
return newMessage;
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java
index c749326..2539838 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java
@@ -53,28 +53,30 @@ import org.junit.runners.Parameterized;
*/
@RunWith(value = Parameterized.class)
public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
+
private String smallFrameAcceptor = new String("tcp://localhost:" + (AMQP_PORT + 8));
int frameSize;
+ int minLargeMessageSize;
-
-
- @Parameterized.Parameters(name = "frameSize = {0}")
- public static Iterable<? extends Object> persistenceEnabled() {
+ @Parameterized.Parameters(name = "frameSize = {0}, minLargeMessage = {1}")
+ public static Iterable<? extends Object> testParameters() {
// The reason I use two frames sizes here
// is because a message that wasn't broken into frames
// but still beyond 50K, should still be considered large when storing
- return Arrays.asList(new Object[][]{{512}, {1024 * 1024}});
+ return Arrays.asList(new Object[][]{{512, 50000}, {1024 * 1024, 50000},
+ // we disable large message for at least one parameter to compare results between large and non large messages
+ {1024 * 1024, 50000000}});
}
- public SimpleStreamingLargeMessageTest(int frameSize) {
+ public SimpleStreamingLargeMessageTest(int frameSize, int minLargeMessageSize) {
this.frameSize = frameSize;
+ this.minLargeMessageSize = minLargeMessageSize;
}
-
@Override
protected void addAdditionalAcceptors(ActiveMQServer server) throws Exception {
- server.getConfiguration().addAcceptorConfiguration("flow", smallFrameAcceptor + "?protocols=AMQP;useEpoll=false;maxFrameSize=" + frameSize + ";amqpMinLargeMessageSize=50000");
+ server.getConfiguration().addAcceptorConfiguration("flow", smallFrameAcceptor + "?protocols=AMQP;useEpoll=false;maxFrameSize=" + frameSize + ";amqpMinLargeMessageSize=" + minLargeMessageSize);
}
@Test(timeout = 60000)
@@ -143,10 +145,10 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
for (int i = 0; i < 10; i++) {
AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS);
Assert.assertNotNull(msgReceived);
- Data body = (Data)msgReceived.getWrappedMessage().getBody();
+ Data body = (Data) msgReceived.getWrappedMessage().getBody();
byte[] bodyArray = body.getValue().getArray();
for (int bI = 0; bI < size; bI++) {
- Assert.assertEquals((byte)'z', bodyArray[bI]);
+ Assert.assertEquals((byte) 'z', bodyArray[bI]);
}
msgReceived.accept(true);
}
@@ -154,7 +156,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
receiver.flow(1);
Assert.assertNull(receiver.receiveNoWait());
-
receiver.close();
connection.close();
@@ -181,7 +182,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
@Test
public void testSendWithPropertiesNonPersistent() throws Exception {
- testSendWithPropertiesAndFilter(true, true);
+ testSendWithPropertiesAndFilter(false, false);
}
@@ -209,6 +210,8 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
AmqpMessage message = new AmqpMessage();
message.setDurable(persistent);
boolean odd = (m % 2 == 0);
+ message.setApplicationProperty("i", m);
+ message.setApplicationProperty("oddString", odd ? "odd" : "even");
message.setApplicationProperty("odd", odd);
if (odd) {
message.setApplicationProperty("oddID", oddID++);
@@ -245,14 +248,14 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
AmqpReceiver receiver = session.createReceiver(getQueueName(), "odd=true");
receiver.flow(10);
for (int i = 0; i < 5; i++) {
- System.out.println("Receiving " + i);
AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS);
Assert.assertNotNull(msgReceived);
- System.out.println("Received " + msgReceived);
- Data body = (Data)msgReceived.getWrappedMessage().getBody();
+ Assert.assertTrue((boolean)msgReceived.getApplicationProperty("odd"));
+ Assert.assertEquals(i, (int)msgReceived.getApplicationProperty("oddID"));
+ Data body = (Data) msgReceived.getWrappedMessage().getBody();
byte[] bodyArray = body.getValue().getArray();
for (int bI = 0; bI < size; bI++) {
- Assert.assertEquals((byte)'z', bodyArray[bI]);
+ Assert.assertEquals((byte) 'z', bodyArray[bI]);
}
msgReceived.accept(true);
}
@@ -260,9 +263,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
receiver.flow(1);
Assert.assertNull(receiver.receiveNoWait());
-
receiver.close();
-
connection.close();
validateNoFilesOnLargeDir(getLargeMessagesDir(), 5);
@@ -273,7 +274,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
}
-
@Test
public void testJMSPersistentTX() throws Exception {