Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Clebert Suconic 2020-03-24 18:17:27 -04:00
parent 31c945f8b0
commit a4489a322e
4 changed files with 35 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessage.java
View File

@ -19,6 +19,7 @@ package org.apache.activemq.artemis.protocol.amqp.broker;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import io.netty.buffer.ByteBuf;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException; import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ICoreMessage; 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; 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. * We control durability on a separate property here, as we need to know if it's durable ahead of the header parsing.

22
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPLargeMessagePersister.java
View File

@ -34,11 +34,6 @@ import static org.apache.activemq.artemis.core.persistence.PersisterIDs.AMQPLarg
public class AMQPLargeMessagePersister extends MessagePersister { public class AMQPLargeMessagePersister extends MessagePersister {
private static final Logger log = Logger.getLogger(AMQPLargeMessagePersister.class); 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 final byte ID = AMQPLargeMessagePersister_ID;
public static AMQPLargeMessagePersister theInstance; public static AMQPLargeMessagePersister theInstance;
@ -72,14 +67,12 @@ public class AMQPLargeMessagePersister extends MessagePersister {
} }
private ByteBuf getSavedEncodeBuffer(Message record) { private ByteBuf getSavedEncodeBuffer(Message record) {
ByteBuf buf = savedBuffer.get(); AMQPLargeMessage largeMessage = (AMQPLargeMessage)record;
if (buf == null) { if (largeMessage.temporaryBuffer == null) {
AMQPLargeMessage largeMessage = (AMQPLargeMessage)record; largeMessage.temporaryBuffer = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode());
buf = PooledByteBufAllocator.DEFAULT.buffer(largeMessage.getEstimateSavedEncode()); largeMessage.saveEncoding(largeMessage.temporaryBuffer);
largeMessage.saveEncoding(buf);
savedBuffer.set(buf);
} }
return buf; return largeMessage.temporaryBuffer;
} }
/** /**
@ -89,7 +82,7 @@ public class AMQPLargeMessagePersister extends MessagePersister {
public void encode(ActiveMQBuffer buffer, Message record) { public void encode(ActiveMQBuffer buffer, Message record) {
super.encode(buffer, record); super.encode(buffer, record);
AMQPMessage msgEncode = (AMQPMessage) record; AMQPLargeMessage msgEncode = (AMQPLargeMessage) record;
buffer.writeLong(record.getMessageID()); buffer.writeLong(record.getMessageID());
buffer.writeBoolean(record.isDurable()); buffer.writeBoolean(record.isDurable());
@ -106,8 +99,7 @@ public class AMQPLargeMessagePersister extends MessagePersister {
ByteBuf savedEncodeBuffer = getSavedEncodeBuffer(record); ByteBuf savedEncodeBuffer = getSavedEncodeBuffer(record);
buffer.writeBytes(savedEncodeBuffer, 0, savedEncodeBuffer.writerIndex()); buffer.writeBytes(savedEncodeBuffer, 0, savedEncodeBuffer.writerIndex());
savedEncodeBuffer.release(); savedEncodeBuffer.release();
msgEncode.temporaryBuffer = null;
savedBuffer.set(null);
} }
@Override @Override

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/LargeServerMessageImpl.java
View File

@ -255,8 +255,8 @@ public final class LargeServerMessageImpl extends CoreMessage implements CoreLar
@Override @Override
public Message copy() { public Message copy() {
SequentialFile newfile = storageManager.createFileForLargeMessage(messageID, durable); SequentialFile newfile = storageManager.createFileForLargeMessage(messageID, durable);
LargeServerMessageImpl newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID);
Message newMessage = new LargeServerMessageImpl(this, properties, newfile, messageID); newMessage.setParentRef(this);
return newMessage; return newMessage;
} }

38
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/largemessages/SimpleStreamingLargeMessageTest.java
View File

@ -53,28 +53,30 @@ import org.junit.runners.Parameterized;
*/ */
@RunWith(value = Parameterized.class) @RunWith(value = Parameterized.class)
public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport { public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
private String smallFrameAcceptor = new String("tcp://localhost:" + (AMQP_PORT + 8)); private String smallFrameAcceptor = new String("tcp://localhost:" + (AMQP_PORT + 8));
int frameSize; int frameSize;
int minLargeMessageSize;
@Parameterized.Parameters(name = "frameSize = {0}, minLargeMessage = {1}")
public static Iterable<? extends Object> testParameters() {
@Parameterized.Parameters(name = "frameSize = {0}")
public static Iterable<? extends Object> persistenceEnabled() {
// The reason I use two frames sizes here // The reason I use two frames sizes here
// is because a message that wasn't broken into frames // is because a message that wasn't broken into frames
// but still beyond 50K, should still be considered large when storing // 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.frameSize = frameSize;
this.minLargeMessageSize = minLargeMessageSize;
} }
@Override @Override
protected void addAdditionalAcceptors(ActiveMQServer server) throws Exception { 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) @Test(timeout = 60000)
@ -143,10 +145,10 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS); AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS);
Assert.assertNotNull(msgReceived); Assert.assertNotNull(msgReceived);
Data body = (Data)msgReceived.getWrappedMessage().getBody(); Data body = (Data) msgReceived.getWrappedMessage().getBody();
byte[] bodyArray = body.getValue().getArray(); byte[] bodyArray = body.getValue().getArray();
for (int bI = 0; bI < size; bI++) { for (int bI = 0; bI < size; bI++) {
Assert.assertEquals((byte)'z', bodyArray[bI]); Assert.assertEquals((byte) 'z', bodyArray[bI]);
} }
msgReceived.accept(true); msgReceived.accept(true);
} }
@ -154,7 +156,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
receiver.flow(1); receiver.flow(1);
Assert.assertNull(receiver.receiveNoWait()); Assert.assertNull(receiver.receiveNoWait());
receiver.close(); receiver.close();
connection.close(); connection.close();
@ -181,7 +182,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
@Test @Test
public void testSendWithPropertiesNonPersistent() throws Exception { public void testSendWithPropertiesNonPersistent() throws Exception {
testSendWithPropertiesAndFilter(true, true); testSendWithPropertiesAndFilter(false, false);
} }
@ -209,6 +210,8 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
AmqpMessage message = new AmqpMessage(); AmqpMessage message = new AmqpMessage();
message.setDurable(persistent); message.setDurable(persistent);
boolean odd = (m % 2 == 0); boolean odd = (m % 2 == 0);
message.setApplicationProperty("i", m);
message.setApplicationProperty("oddString", odd ? "odd" : "even");
message.setApplicationProperty("odd", odd); message.setApplicationProperty("odd", odd);
if (odd) { if (odd) {
message.setApplicationProperty("oddID", oddID++); message.setApplicationProperty("oddID", oddID++);
@ -245,14 +248,14 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
AmqpReceiver receiver = session.createReceiver(getQueueName(), "odd=true"); AmqpReceiver receiver = session.createReceiver(getQueueName(), "odd=true");
receiver.flow(10); receiver.flow(10);
for (int i = 0; i < 5; i++) { for (int i = 0; i < 5; i++) {
System.out.println("Receiving " + i);
AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS); AmqpMessage msgReceived = receiver.receive(10, TimeUnit.SECONDS);
Assert.assertNotNull(msgReceived); Assert.assertNotNull(msgReceived);
System.out.println("Received " + msgReceived); Assert.assertTrue((boolean)msgReceived.getApplicationProperty("odd"));
Data body = (Data)msgReceived.getWrappedMessage().getBody(); Assert.assertEquals(i, (int)msgReceived.getApplicationProperty("oddID"));
Data body = (Data) msgReceived.getWrappedMessage().getBody();
byte[] bodyArray = body.getValue().getArray(); byte[] bodyArray = body.getValue().getArray();
for (int bI = 0; bI < size; bI++) { for (int bI = 0; bI < size; bI++) {
Assert.assertEquals((byte)'z', bodyArray[bI]); Assert.assertEquals((byte) 'z', bodyArray[bI]);
} }
msgReceived.accept(true); msgReceived.accept(true);
} }
@ -260,9 +263,7 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
receiver.flow(1); receiver.flow(1);
Assert.assertNull(receiver.receiveNoWait()); Assert.assertNull(receiver.receiveNoWait());
receiver.close(); receiver.close();
connection.close(); connection.close();
validateNoFilesOnLargeDir(getLargeMessagesDir(), 5); validateNoFilesOnLargeDir(getLargeMessagesDir(), 5);
@ -273,7 +274,6 @@ public class SimpleStreamingLargeMessageTest extends AmqpClientTestSupport {
} }
@Test @Test
public void testJMSPersistentTX() throws Exception { public void testJMSPersistentTX() throws Exception {