activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject activemq-artemis git commit: ARTEMIS-928 Changing Netty and InVM to copy buffers, and retain them on the Netty Polls.
Date Wed, 18 Jan 2017 22:06:06 GMT
Repository: activemq-artemis
Updated Branches:
  refs/heads/1.x aaf09262f -> 036933a4a


ARTEMIS-928 Changing Netty and InVM to copy buffers, and retain them on the Netty Polls.

(cherry picked from commit 3347a4fd2716c7c998a0ef68499f4dc3d8106241)


Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/036933a4
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/036933a4
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/036933a4

Branch: refs/heads/1.x
Commit: 036933a4a479c127b89f60f8de9f701c82d3319f
Parents: aaf0926
Author: Will Reichert <willr3@gmail.com>
Authored: Wed Jan 4 10:14:16 2017 -0600
Committer: Clebert Suconic <clebertsuconic@apache.org>
Committed: Wed Jan 18 17:04:32 2017 -0500

----------------------------------------------------------------------
 .../artemis/api/core/ActiveMQBuffer.java        |  6 ++++
 .../artemis/api/core/ActiveMQBuffers.java       |  9 +++++
 .../core/buffers/impl/ChannelBufferWrapper.java | 29 +++++++++++++--
 .../CompressedLargeMessageControllerImpl.java   |  5 +++
 .../client/impl/LargeMessageControllerImpl.java |  5 +++
 .../artemis/core/message/impl/MessageImpl.java  | 37 +++++++++++++++++---
 .../artemis/core/protocol/core/Packet.java      |  9 +++++
 .../core/impl/ActiveMQSessionContext.java       |  2 ++
 .../core/protocol/core/impl/ChannelImpl.java    |  6 ++++
 .../core/protocol/core/impl/PacketImpl.java     |  8 ++++-
 .../impl/wireformat/SessionReceiveMessage.java  |  2 +-
 .../impl/wireformat/SessionSendMessage.java     |  2 +-
 .../remoting/impl/netty/NettyConnection.java    |  5 +++
 .../protocol/AbstractRemotingConnection.java    |  5 +++
 .../spi/core/protocol/RemotingConnection.java   |  2 ++
 .../artemis/spi/core/remoting/Connection.java   |  2 ++
 .../amqp/converter/TestConversions.java         |  5 +++
 .../core/protocol/mqtt/MQTTConnection.java      |  5 +++
 .../core/protocol/stomp/StompConnection.java    |  5 +++
 .../core/impl/wireformat/QuorumVoteMessage.java |  6 ++++
 .../core/remoting/impl/invm/InVMConnection.java | 23 ++++++++++--
 .../unit/core/message/impl/MessageImplTest.java |  7 +++-
 22 files changed, 171 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
index c129e3a..f30ef35 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffer.java
@@ -1149,4 +1149,10 @@ public interface ActiveMQBuffer extends DataInput {
     * @return A converted NIO Buffer
     */
    ByteBuffer toByteBuffer(int index, int length);
+
+   /**
+   * Release any underlying resources held by this buffer
+   */
+   void release();
+
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffers.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffers.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffers.java
index 22f0f9a..849c921 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffers.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/api/core/ActiveMQBuffers.java
@@ -18,6 +18,7 @@ package org.apache.activemq.artemis.api.core;
 
 import java.nio.ByteBuffer;
 
+import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
 
@@ -26,6 +27,9 @@ import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
  */
 public final class ActiveMQBuffers {
 
+
+   private static final PooledByteBufAllocator ALLOCATOR = new PooledByteBufAllocator();
+
    /**
     * Creates a <em>self-expanding</em> ActiveMQBuffer with the given initial
size
     *
@@ -36,6 +40,11 @@ public final class ActiveMQBuffers {
       return new ChannelBufferWrapper(Unpooled.buffer(size));
    }
 
+   public static ActiveMQBuffer pooledBuffer(final int size) {
+      return new ChannelBufferWrapper(ALLOCATOR.heapBuffer(size),true, true);
+   }
+
+
    /**
     * Creates a <em>self-expanding</em> ActiveMQBuffer filled with the given
byte array
     *

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
index 60262f8..c75be21 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/core/buffers/impl/ChannelBufferWrapper.java
@@ -31,7 +31,7 @@ public class ChannelBufferWrapper implements ActiveMQBuffer {
 
    protected ByteBuf buffer; // NO_UCD (use final)
    private final boolean releasable;
-
+   private final boolean isPooled;
    public static ByteBuf unwrap(ByteBuf buffer) {
       ByteBuf parent;
       while ((parent = buffer.unwrap()) != null && parent != buffer) { // this last
part is just in case the semantic
@@ -45,14 +45,18 @@ public class ChannelBufferWrapper implements ActiveMQBuffer {
    public ChannelBufferWrapper(final ByteBuf buffer) {
       this(buffer, false);
    }
-
    public ChannelBufferWrapper(final ByteBuf buffer, boolean releasable) {
+      this(buffer, releasable, false);
+   }
+   public ChannelBufferWrapper(final ByteBuf buffer, boolean releasable, boolean pooled)
{
       if (!releasable) {
          this.buffer = Unpooled.unreleasableBuffer(buffer);
       } else {
          this.buffer = buffer;
       }
       this.releasable = releasable;
+      this.isPooled = pooled;
+
    }
 
    @Override
@@ -398,7 +402,19 @@ public class ChannelBufferWrapper implements ActiveMQBuffer {
 
    @Override
    public ActiveMQBuffer readSlice(final int length) {
-      return new ChannelBufferWrapper(buffer.readSlice(length), releasable);
+      if ( isPooled ) {
+         ByteBuf fromBuffer = buffer.readSlice(length);
+         ByteBuf newNettyBuffer = Unpooled.buffer(fromBuffer.capacity());
+         int read = fromBuffer.readerIndex();
+         int writ = fromBuffer.writerIndex();
+         fromBuffer.readerIndex(0);
+         fromBuffer.readBytes(newNettyBuffer,0,writ);
+         newNettyBuffer.setIndex(read,writ);
+         ActiveMQBuffer returnBuffer = new ChannelBufferWrapper(newNettyBuffer,releasable,false);
+         returnBuffer.setIndex(read,writ);
+         return returnBuffer;
+      }
+      return new ChannelBufferWrapper(buffer.readSlice(length), releasable, isPooled);
    }
 
    @Override
@@ -523,6 +539,13 @@ public class ChannelBufferWrapper implements ActiveMQBuffer {
    }
 
    @Override
+   public void release() {
+      if ( this.isPooled ) {
+         buffer.release();
+      }
+   }
+
+   @Override
    public boolean writable() {
       return buffer.isWritable();
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
index 24b7f1e..55f9129 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/CompressedLargeMessageControllerImpl.java
@@ -203,6 +203,11 @@ final class CompressedLargeMessageControllerImpl implements LargeMessageControll
    }
 
    @Override
+   public void release() {
+      //no-op
+   }
+
+   @Override
    public int readerIndex() {
       return 0;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
index 5a27f24..951aea2 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/LargeMessageControllerImpl.java
@@ -525,6 +525,11 @@ public class LargeMessageControllerImpl implements LargeMessageController
{
    }
 
    @Override
+   public void release() {
+      //no-op
+   }
+
+   @Override
    public int readerIndex() {
       return (int) readerIndex;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
index 921f97d..1e19817 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/MessageImpl.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Set;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
 import org.apache.activemq.artemis.api.core.ActiveMQException;
@@ -76,9 +77,9 @@ public abstract class MessageImpl implements MessageInternal {
 
    protected byte priority;
 
-   protected ActiveMQBuffer buffer;
+   protected volatile ActiveMQBuffer buffer;
 
-   protected ResetLimitWrappedActiveMQBuffer bodyBuffer;
+   protected volatile ResetLimitWrappedActiveMQBuffer bodyBuffer;
 
    protected volatile boolean bufferValid;
 
@@ -434,12 +435,16 @@ public abstract class MessageImpl implements MessageInternal {
 
    @Override
    public void decodeFromBuffer(final ActiveMQBuffer buffer) {
-      this.buffer = buffer;
+
+      this.buffer = copyMessageBuffer(buffer);
 
       decode();
 
+      //synchronize indexes
+      buffer.setIndex(this.buffer.readerIndex(),this.buffer.writerIndex());
+
       // Setting up the BodyBuffer based on endOfBodyPosition set from decode
-      ResetLimitWrappedActiveMQBuffer tmpbodyBuffer = new ResetLimitWrappedActiveMQBuffer(BODY_OFFSET,
buffer, null);
+      ResetLimitWrappedActiveMQBuffer tmpbodyBuffer = new ResetLimitWrappedActiveMQBuffer(BODY_OFFSET,
this.buffer, null);
       tmpbodyBuffer.readerIndex(BODY_OFFSET);
       tmpbodyBuffer.writerIndex(endOfBodyPosition);
       // only set this after the writer and reader is set,
@@ -449,6 +454,30 @@ public abstract class MessageImpl implements MessageInternal {
 
    }
 
+   private ActiveMQBuffer copyMessageBuffer(ActiveMQBuffer buffer) {
+      ActiveMQBuffer copiedBuffer;
+
+      ByteBuf newNettyBuffer = Unpooled.buffer( buffer.byteBuf().capacity() );
+
+      int read = buffer.byteBuf().readerIndex();
+      int writ = buffer.byteBuf().writerIndex();
+
+      int readArt = buffer.readerIndex();
+      int writArt = buffer.writerIndex();
+      buffer.byteBuf().readerIndex( 0 );
+
+      buffer.byteBuf().readBytes( newNettyBuffer, 0, buffer.byteBuf().writerIndex() );
+      buffer.byteBuf().setIndex( read, writ );
+      newNettyBuffer.setIndex( read, writ );
+
+      copiedBuffer = new ChannelBufferWrapper( newNettyBuffer );
+
+      buffer.setIndex( readArt, writArt );
+      copiedBuffer.setIndex( readArt, writArt );
+
+      return copiedBuffer;
+   }
+
    @Override
    public void bodyChanged() {
       bufferValid = false;

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
index ddb734e..d7ae5b3 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/Packet.java
@@ -64,6 +64,15 @@ public interface Packet {
    ActiveMQBuffer encode(RemotingConnection connection);
 
    /**
+    * Encodes the packet and returns a {@link ActiveMQBuffer} containing the data
+    *
+    * @param connection the connection
+    * @param usePooled if the returned buffer should be pooled or unpooled
+    * @return the buffer to encode to
+    */
+   ActiveMQBuffer encode(RemotingConnection connection, boolean usePooled);
+
+   /**
     * decodes the buffer into this packet
     *
     * @param buffer the buffer to decode from

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
index 56c7135..b6c0793 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java
@@ -877,6 +877,8 @@ public class ActiveMQSessionContext extends SessionContext {
       ActiveMQBuffer buffer = packet.encode(this.getCoreConnection());
 
       conn.write(buffer, false, false);
+
+      buffer.release();
    }
 
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
index 41be080..bf4dd18 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ChannelImpl.java
@@ -301,6 +301,8 @@ public final class ChannelImpl implements Channel {
          // buffer is full, preventing any incoming buffers being handled and blocking failover
          connection.getTransportConnection().write(buffer, flush, batch);
 
+         buffer.release();
+
          return true;
       }
    }
@@ -408,6 +410,7 @@ public final class ChannelImpl implements Channel {
             }
          } finally {
             lock.unlock();
+            buffer.release();
          }
 
          return response;
@@ -630,6 +633,9 @@ public final class ChannelImpl implements Channel {
       final ActiveMQBuffer buffer = packet.encode(connection);
 
       connection.getTransportConnection().write(buffer, false, false);
+
+      buffer.release();
+
    }
 
    private void addResendPacket(Packet packet) {

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
index 6dddf3b..d373176 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/PacketImpl.java
@@ -274,7 +274,13 @@ public class PacketImpl implements Packet {
 
    @Override
    public ActiveMQBuffer encode(final RemotingConnection connection) {
-      ActiveMQBuffer buffer = connection.createTransportBuffer(PacketImpl.INITIAL_PACKET_SIZE);
+      return encode(connection,true);
+   }
+
+
+   @Override
+   public ActiveMQBuffer encode(final RemotingConnection connection, boolean usePooled) {
+      ActiveMQBuffer buffer = connection.createTransportBuffer(PacketImpl.INITIAL_PACKET_SIZE,
usePooled);
 
       // The standard header fields
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionReceiveMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionReceiveMessage.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionReceiveMessage.java
index 02ed2bf..ce76186 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionReceiveMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionReceiveMessage.java
@@ -56,7 +56,7 @@ public class SessionReceiveMessage extends MessagePacket {
    public ActiveMQBuffer encode(final RemotingConnection connection) {
       ActiveMQBuffer buffer = message.getEncodedBuffer();
 
-      ActiveMQBuffer bufferWrite = connection.createTransportBuffer(buffer.writerIndex());
+      ActiveMQBuffer bufferWrite = connection.createTransportBuffer(buffer.writerIndex(),
true);
       bufferWrite.writeBytes(buffer, 0, bufferWrite.capacity());
       bufferWrite.setIndex(buffer.readerIndex(), buffer.writerIndex());
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
index 91d43a5..c7bb30e 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionSendMessage.java
@@ -68,7 +68,7 @@ public class SessionSendMessage extends MessagePacket {
          // this is for unit tests only
          bufferWrite = buffer.copy(0, buffer.capacity());
       } else {
-         bufferWrite = connection.createTransportBuffer(buffer.writerIndex() + 1); // 1 for
the requireResponse
+         bufferWrite = connection.createTransportBuffer(buffer.writerIndex() + 1, true);
// 1 for the requireResponse
       }
       bufferWrite.writeBytes(buffer, 0, buffer.writerIndex());
       bufferWrite.setIndex(buffer.readerIndex(), buffer.writerIndex());

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java
index 0b307ef..33dbf4b 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java
@@ -210,6 +210,11 @@ public class NettyConnection implements Connection {
 
    @Override
    public ActiveMQBuffer createTransportBuffer(final int size) {
+      return createTransportBuffer(size, false);
+   }
+
+   @Override
+   public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
       return new ChannelBufferWrapper(PartialPooledByteBufAllocator.INSTANCE.directBuffer(size),
true);
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java
index a9e12aa..6884243 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/AbstractRemotingConnection.java
@@ -179,6 +179,11 @@ public abstract class AbstractRemotingConnection implements RemotingConnection
{
    }
 
    @Override
+   public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
+      return transportConnection.createTransportBuffer(size, pooled);
+   }
+
+   @Override
    public Connection getTransportConnection() {
       return transportConnection;
    }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java
index 39ecdf6..a68999b 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/protocol/RemotingConnection.java
@@ -120,6 +120,8 @@ public interface RemotingConnection extends BufferHandler {
     */
    ActiveMQBuffer createTransportBuffer(int size);
 
+   ActiveMQBuffer createTransportBuffer(int size, boolean pooled);
+
    /**
     * called when the underlying connection fails.
     *

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java
b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java
index 7ab0c40..a5fcf87 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/Connection.java
@@ -35,6 +35,8 @@ public interface Connection {
     */
    ActiveMQBuffer createTransportBuffer(int size);
 
+   ActiveMQBuffer createTransportBuffer(int size, boolean pooled);
+
    RemotingConnection getProtocolConnection();
 
    void setProtocolConnection(RemotingConnection connection);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
index 96ce90e..cbe2699 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/protocol/amqp/converter/TestConversions.java
@@ -765,5 +765,10 @@ public class TestConversions extends Assert {
       public ByteBuffer toByteBuffer(int index, int length) {
          return null;
       }
+
+      @Override
+      public void release() {
+         //no-op
+      }
    }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
index 446e362..6143cf7 100644
--- a/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
+++ b/artemis-protocols/artemis-mqtt-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/mqtt/MQTTConnection.java
@@ -132,6 +132,11 @@ public class MQTTConnection implements RemotingConnection {
 
    @Override
    public ActiveMQBuffer createTransportBuffer(int size) {
+      return createTransportBuffer(size, false);
+   }
+
+   @Override
+   public ActiveMQBuffer createTransportBuffer(int size, boolean pooled) {
       return transportConnection.createTransportBuffer(size);
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java
b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java
index 6dcff64..ec010f7 100644
--- a/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java
+++ b/artemis-protocols/artemis-stomp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompConnection.java
@@ -265,6 +265,11 @@ public final class StompConnection implements RemotingConnection {
 
    @Override
    public ActiveMQBuffer createTransportBuffer(int size) {
+      return createTransportBuffer(size, false);
+   }
+
+   @Override
+   public ActiveMQBuffer createTransportBuffer(int size, boolean pooled) {
       return ActiveMQBuffers.dynamicBuffer(size);
    }
 

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/QuorumVoteMessage.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/QuorumVoteMessage.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/QuorumVoteMessage.java
index f808349..78ebcb9 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/QuorumVoteMessage.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/QuorumVoteMessage.java
@@ -21,6 +21,7 @@ import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
 import org.apache.activemq.artemis.core.server.cluster.qourum.QuorumVoteHandler;
 import org.apache.activemq.artemis.core.server.cluster.qourum.Vote;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
 
 public class QuorumVoteMessage extends PacketImpl {
 
@@ -41,6 +42,11 @@ public class QuorumVoteMessage extends PacketImpl {
    }
 
    @Override
+   public ActiveMQBuffer encode(final RemotingConnection connection) {
+      return encode(connection,false);
+   }
+
+   @Override
    public void encodeRest(ActiveMQBuffer buffer) {
       super.encodeRest(buffer);
       buffer.writeSimpleString(handler);

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java
b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java
index 27fc544..24931d3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/invm/InVMConnection.java
@@ -146,7 +146,16 @@ public class InVMConnection implements Connection {
 
    @Override
    public ActiveMQBuffer createTransportBuffer(final int size) {
-      return ActiveMQBuffers.dynamicBuffer(size);
+      return createTransportBuffer(size, false);
+   }
+
+   @Override
+   public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
+      if ( pooled ) {
+         return ActiveMQBuffers.pooledBuffer( size );
+      } else {
+         return ActiveMQBuffers.dynamicBuffer( size );
+      }
    }
 
    @Override
@@ -173,9 +182,13 @@ public class InVMConnection implements Connection {
                      final boolean flush,
                      final boolean batch,
                      final ChannelFutureListener futureListener) {
-      final ActiveMQBuffer copied = buffer.copy(0, buffer.capacity());
 
-      copied.setIndex(buffer.readerIndex(), buffer.writerIndex());
+      final ActiveMQBuffer copied = ActiveMQBuffers.pooledBuffer(buffer.capacity());
+      int read = buffer.readerIndex();
+      int writ = buffer.writerIndex();
+      copied.writeBytes(buffer,read,writ - read);
+      copied.setIndex(read,writ);
+      buffer.setIndex(read,writ);
 
       try {
          executor.execute(new Runnable() {
@@ -201,6 +214,10 @@ public class InVMConnection implements Connection {
                   if (logger.isTraceEnabled()) {
                      logger.trace(InVMConnection.this + "::packet sent done");
                   }
+                  copied.release();
+//                  if ( copied.byteBuf().refCnt() > 0 ) {
+//                     copied.release();
+//                  }
                }
             }
          });

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/036933a4/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/message/impl/MessageImplTest.java
----------------------------------------------------------------------
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/message/impl/MessageImplTest.java
b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/message/impl/MessageImplTest.java
index c559bf9..252b0eb 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/message/impl/MessageImplTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/message/impl/MessageImplTest.java
@@ -286,13 +286,18 @@ public class MessageImplTest extends ActiveMQTestBase {
             }
 
             for (int i = 0; i < RUNS; i++) {
+               ActiveMQBuffer buf = null;
                try {
                   SessionSendMessage ssm = new SessionSendMessage(msg);
-                  ActiveMQBuffer buf = ssm.encode(null);
+                  buf = ssm.encode(null);
                   simulateRead(buf);
                } catch (Throwable e) {
                   e.printStackTrace();
                   errors.incrementAndGet();
+               } finally {
+                  if ( buf != null ) {
+                     buf.release();
+                  }
                }
             }
          }


Mime
View raw message