You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jb...@apache.org on 2017/03/23 23:20:13 UTC
[1/4] activemq-artemis git commit: This closes #1128
Repository: activemq-artemis
Updated Branches:
refs/heads/master 356a710b7 -> 4203ae89c
This closes #1128
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/4203ae89
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/4203ae89
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/4203ae89
Branch: refs/heads/master
Commit: 4203ae89ca24c50fa90d33e4700e4847fc38d693
Parents: 356a710 096556f
Author: Justin Bertram <jb...@apache.org>
Authored: Thu Mar 23 18:16:48 2017 -0500
Committer: Justin Bertram <jb...@apache.org>
Committed: Thu Mar 23 18:16:48 2017 -0500
----------------------------------------------------------------------
.../artemis/cli/commands/etc/amqp-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/broker.xml | 5 +-
.../cli/commands/etc/hornetq-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/mqtt-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/stomp-acceptor.txt | 2 +-
.../impl/netty/DelegatingEventLoopGroup.java | 198 +++++++++++++++++++
.../remoting/impl/netty/NettyConnection.java | 2 +-
.../remoting/impl/netty/NettyConnector.java | 51 +++--
.../netty/PartialPooledByteBufAllocator.java | 138 -------------
.../impl/netty/SharedEventLoopGroup.java | 112 +++++++++++
.../impl/netty/SharedNioEventLoopGroup.java | 111 -----------
.../remoting/impl/netty/TransportConstants.java | 23 ++-
.../artemis/core/protocol/ProtocolHandler.java | 5 -
.../core/remoting/impl/netty/NettyAcceptor.java | 59 ++++--
.../core/server/ActiveMQServerLogger.java | 4 +-
.../client/transport/NettyTcpTransport.java | 1 -
.../amqp/client/transport/NettyWSTransport.java | 1 -
.../PartialPooledByteBufAllocator.java | 138 -------------
.../NettyConnectorWithHTTPUpgradeTest.java | 3 -
19 files changed, 419 insertions(+), 440 deletions(-)
----------------------------------------------------------------------
[4/4] activemq-artemis git commit: ARTEMIS-1056 Removing
PartialPooledByteBufAllocator
Posted by jb...@apache.org.
ARTEMIS-1056 Removing PartialPooledByteBufAllocator
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/2c9b0280
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/2c9b0280
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/2c9b0280
Branch: refs/heads/master
Commit: 2c9b02806ddc48a03f9c7dc8b3e00c97b1cf8fb0
Parents: 356a710
Author: Clebert Suconic <cl...@apache.org>
Authored: Thu Mar 23 10:31:24 2017 -0400
Committer: Justin Bertram <jb...@apache.org>
Committed: Thu Mar 23 18:16:48 2017 -0500
----------------------------------------------------------------------
.../remoting/impl/netty/NettyConnection.java | 3 +-
.../remoting/impl/netty/NettyConnector.java | 1 -
.../netty/PartialPooledByteBufAllocator.java | 138 -------------------
.../artemis/core/protocol/ProtocolHandler.java | 5 -
.../core/remoting/impl/netty/NettyAcceptor.java | 1 -
.../client/transport/NettyTcpTransport.java | 1 -
.../amqp/client/transport/NettyWSTransport.java | 1 -
.../PartialPooledByteBufAllocator.java | 138 -------------------
.../NettyConnectorWithHTTPUpgradeTest.java | 3 -
9 files changed, 2 insertions(+), 289 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/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 33dbf4b..c3a71c5 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
@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.concurrent.Semaphore;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
@@ -215,7 +216,7 @@ public class NettyConnection implements Connection {
@Override
public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
- return new ChannelBufferWrapper(PartialPooledByteBufAllocator.INSTANCE.directBuffer(size), true);
+ return new ChannelBufferWrapper(PooledByteBufAllocator.DEFAULT.directBuffer(size), true);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
index 38fb326..4a5e9e6 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
@@ -407,7 +407,6 @@ public class NettyConnector extends AbstractConnector {
}
bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
bootstrap.option(ChannelOption.SO_REUSEADDR, true);
- bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
channelGroup = new DefaultChannelGroup("activemq-connector", GlobalEventExecutor.INSTANCE);
final SSLContext context;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/PartialPooledByteBufAllocator.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/PartialPooledByteBufAllocator.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/PartialPooledByteBufAllocator.java
deleted file mode 100644
index 3a192e8..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/PartialPooledByteBufAllocator.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.core.remoting.impl.netty;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.CompositeByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
-
-/**
- * A {@link ByteBufAllocator} which is partial pooled. Which means only direct {@link ByteBuf}s are pooled. The rest
- * is unpooled.
- */
-public class PartialPooledByteBufAllocator implements ByteBufAllocator {
-
- private static final ByteBufAllocator POOLED = PooledByteBufAllocator.DEFAULT;
- private static final ByteBufAllocator UNPOOLED = new UnpooledByteBufAllocator(false);
-
- public static final PartialPooledByteBufAllocator INSTANCE = new PartialPooledByteBufAllocator();
-
- private PartialPooledByteBufAllocator() {
- }
-
- @Override
- public ByteBuf buffer() {
- return UNPOOLED.heapBuffer();
- }
-
- @Override
- public ByteBuf buffer(int initialCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf buffer(int initialCapacity, int maxCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf ioBuffer() {
- return POOLED.directBuffer();
- }
-
- @Override
- public ByteBuf ioBuffer(int initialCapacity) {
- return POOLED.directBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf ioBuffer(int initialCapacity, int maxCapacity) {
- return POOLED.directBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf heapBuffer() {
- return UNPOOLED.heapBuffer();
- }
-
- @Override
- public ByteBuf heapBuffer(int initialCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf heapBuffer(int initialCapacity, int maxCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf directBuffer() {
- return POOLED.directBuffer();
- }
-
- @Override
- public ByteBuf directBuffer(int initialCapacity) {
- return POOLED.directBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
- return POOLED.directBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public CompositeByteBuf compositeBuffer() {
- return UNPOOLED.compositeHeapBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeBuffer(int maxNumComponents) {
- return UNPOOLED.compositeHeapBuffer(maxNumComponents);
- }
-
- @Override
- public CompositeByteBuf compositeHeapBuffer() {
- return UNPOOLED.compositeHeapBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeHeapBuffer(int maxNumComponents) {
- return UNPOOLED.compositeHeapBuffer(maxNumComponents);
- }
-
- @Override
- public CompositeByteBuf compositeDirectBuffer() {
- return POOLED.compositeDirectBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeDirectBuffer(int maxNumComponents) {
- return POOLED.compositeDirectBuffer();
- }
-
- @Override
- public boolean isDirectBufferPooled() {
- return true;
- }
-
- @Override
- public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
- return UNPOOLED.calculateNewCapacity(minNewCapacity, maxCapacity);
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ProtocolHandler.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ProtocolHandler.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ProtocolHandler.java
index 340861b..ca78f29 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ProtocolHandler.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/protocol/ProtocolHandler.java
@@ -44,7 +44,6 @@ import org.apache.activemq.artemis.core.remoting.impl.netty.HttpKeepAliveRunnabl
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnector;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyServerConnection;
-import org.apache.activemq.artemis.core.remoting.impl.netty.PartialPooledByteBufAllocator;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.server.protocol.websocket.WebSocketServerHandler;
import org.apache.activemq.artemis.spi.core.protocol.ProtocolManager;
@@ -185,10 +184,6 @@ public class ProtocolHandler {
protocolManagerToUse.handshake(connection, new ChannelBufferWrapper(in));
pipeline.remove(this);
- // https://issues.apache.org/jira/browse/ARTEMIS-392
- // Application servers or other components may upgrade a regular socket to Netty
- // We need to be able to work normally as with anything else on Artemis
- ctx.channel().config().setAllocator(PartialPooledByteBufAllocator.INSTANCE);
ctx.flush();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
index aaae108..a46684d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
@@ -319,7 +319,6 @@ public class NettyAcceptor extends AbstractAcceptor {
bootstrap.option(ChannelOption.SO_REUSEADDR, true);
bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
- bootstrap.childOption(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
channelGroup = new DefaultChannelGroup("activemq-accepted-channels", GlobalEventExecutor.INSTANCE);
serverChannelGroup = new DefaultChannelGroup("activemq-acceptor-channels", GlobalEventExecutor.INSTANCE);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java
----------------------------------------------------------------------
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java
index d4b9f54..29963a0 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyTcpTransport.java
@@ -268,7 +268,6 @@ public class NettyTcpTransport implements NettyTransport {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, options.getConnectTimeout());
bootstrap.option(ChannelOption.SO_KEEPALIVE, options.isTcpKeepAlive());
bootstrap.option(ChannelOption.SO_LINGER, options.getSoLinger());
- bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
if (options.getSendBufferSize() != -1) {
bootstrap.option(ChannelOption.SO_SNDBUF, options.getSendBufferSize());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java
----------------------------------------------------------------------
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java
index 8a34a4b..f75a52e 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/NettyWSTransport.java
@@ -285,7 +285,6 @@ public class NettyWSTransport implements NettyTransport {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, options.getConnectTimeout());
bootstrap.option(ChannelOption.SO_KEEPALIVE, options.isTcpKeepAlive());
bootstrap.option(ChannelOption.SO_LINGER, options.getSoLinger());
- bootstrap.option(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
if (options.getSendBufferSize() != -1) {
bootstrap.option(ChannelOption.SO_SNDBUF, options.getSendBufferSize());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/PartialPooledByteBufAllocator.java
----------------------------------------------------------------------
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/PartialPooledByteBufAllocator.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/PartialPooledByteBufAllocator.java
deleted file mode 100644
index 17f601a..0000000
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/transport/PartialPooledByteBufAllocator.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.transport.amqp.client.transport;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.CompositeByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
-
-/**
- * A {@link ByteBufAllocator} which is partial pooled. Which means only direct
- * {@link ByteBuf}s are pooled. The rest is unpooled.
- */
-public class PartialPooledByteBufAllocator implements ByteBufAllocator {
-
- private static final ByteBufAllocator POOLED = PooledByteBufAllocator.DEFAULT;
- private static final ByteBufAllocator UNPOOLED = new UnpooledByteBufAllocator(false);
-
- public static final PartialPooledByteBufAllocator INSTANCE = new PartialPooledByteBufAllocator();
-
- private PartialPooledByteBufAllocator() {
- }
-
- @Override
- public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
- return POOLED.calculateNewCapacity(minNewCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf buffer() {
- return UNPOOLED.heapBuffer();
- }
-
- @Override
- public ByteBuf buffer(int initialCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf buffer(int initialCapacity, int maxCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf ioBuffer() {
- return UNPOOLED.heapBuffer();
- }
-
- @Override
- public ByteBuf ioBuffer(int initialCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf ioBuffer(int initialCapacity, int maxCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf heapBuffer() {
- return UNPOOLED.heapBuffer();
- }
-
- @Override
- public ByteBuf heapBuffer(int initialCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf heapBuffer(int initialCapacity, int maxCapacity) {
- return UNPOOLED.heapBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public ByteBuf directBuffer() {
- return POOLED.directBuffer();
- }
-
- @Override
- public ByteBuf directBuffer(int initialCapacity) {
- return POOLED.directBuffer(initialCapacity);
- }
-
- @Override
- public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
- return POOLED.directBuffer(initialCapacity, maxCapacity);
- }
-
- @Override
- public CompositeByteBuf compositeBuffer() {
- return UNPOOLED.compositeHeapBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeBuffer(int maxNumComponents) {
- return UNPOOLED.compositeHeapBuffer(maxNumComponents);
- }
-
- @Override
- public CompositeByteBuf compositeHeapBuffer() {
- return UNPOOLED.compositeHeapBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeHeapBuffer(int maxNumComponents) {
- return UNPOOLED.compositeHeapBuffer(maxNumComponents);
- }
-
- @Override
- public CompositeByteBuf compositeDirectBuffer() {
- return POOLED.compositeDirectBuffer();
- }
-
- @Override
- public CompositeByteBuf compositeDirectBuffer(int maxNumComponents) {
- return POOLED.compositeDirectBuffer();
- }
-
- @Override
- public boolean isDirectBufferPooled() {
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2c9b0280/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java
index 0afd30c..0f08ecd 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/transports/netty/NettyConnectorWithHTTPUpgradeTest.java
@@ -26,7 +26,6 @@ import java.util.Map;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -52,7 +51,6 @@ import org.apache.activemq.artemis.api.core.client.ClientSessionFactory;
import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptor;
-import org.apache.activemq.artemis.core.remoting.impl.netty.PartialPooledByteBufAllocator;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
import org.apache.activemq.artemis.core.remoting.impl.ssl.SSLSupport;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
@@ -216,7 +214,6 @@ public class NettyConnectorWithHTTPUpgradeTest extends ActiveMQTestBase {
} else {
context = null;
}
- b.childOption(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
[3/4] activemq-artemis git commit: ARTEMIS-994 Tidy up on epoll
Posted by jb...@apache.org.
ARTEMIS-994 Tidy up on epoll
- improved logging
- renamed properties and tidy up on configurations
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/096556f3
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/096556f3
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/096556f3
Branch: refs/heads/master
Commit: 096556f346fdaab17b2e829801f7218ccc3dac1b
Parents: a610748
Author: Clebert Suconic <cl...@apache.org>
Authored: Thu Mar 23 16:32:42 2017 -0400
Committer: Justin Bertram <jb...@apache.org>
Committed: Thu Mar 23 18:16:48 2017 -0500
----------------------------------------------------------------------
.../artemis/cli/commands/etc/amqp-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/broker.xml | 5 +-
.../cli/commands/etc/hornetq-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/mqtt-acceptor.txt | 2 +-
.../artemis/cli/commands/etc/stomp-acceptor.txt | 2 +-
.../remoting/impl/netty/NettyConnector.java | 68 +++++++-------------
.../remoting/impl/netty/TransportConstants.java | 26 +++++---
.../core/remoting/impl/netty/NettyAcceptor.java | 68 +++++++++-----------
.../core/server/ActiveMQServerLogger.java | 4 +-
9 files changed, 77 insertions(+), 102 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/amqp-acceptor.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/amqp-acceptor.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/amqp-acceptor.txt
index 0a4b381..5b20b92 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/amqp-acceptor.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/amqp-acceptor.txt
@@ -1,3 +1,3 @@
<!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic.-->
- <acceptor name="amqp">tcp://${host}:${amqp.port}?protocols=AMQP</acceptor>
+ <acceptor name="amqp">tcp://${host}:${amqp.port}?protocols=AMQP;useEpoll=true</acceptor>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
index f1f96e7..5ca8687 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/broker.xml
@@ -61,9 +61,10 @@ ${ping-config.settings}${journal-buffer.settings}${connector-config.settings}
<global-max-size>100Mb</global-max-size>
<acceptors>
- <!-- Acceptor for every supported protocol -->
- <acceptor name="artemis">tcp://${host}:${default.port}?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE</acceptor>
+ <!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
+ <!-- Acceptor for every supported protocol -->
+ <acceptor name="artemis">tcp://${host}:${default.port}?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true</acceptor>
${amqp-acceptor}${stomp-acceptor}${hornetq-acceptor}${mqtt-acceptor}
</acceptors>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/hornetq-acceptor.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/hornetq-acceptor.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/hornetq-acceptor.txt
index 3c0d804..da77ebe 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/hornetq-acceptor.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/hornetq-acceptor.txt
@@ -1,3 +1,3 @@
<!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
- <acceptor name="hornetq">tcp://${host}:${hq.port}?protocols=HORNETQ,STOMP</acceptor>
+ <acceptor name="hornetq">tcp://${host}:${hq.port}?protocols=HORNETQ,STOMP;useEpoll=true</acceptor>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/mqtt-acceptor.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/mqtt-acceptor.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/mqtt-acceptor.txt
index bf383f8..ceaf615 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/mqtt-acceptor.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/mqtt-acceptor.txt
@@ -1,3 +1,3 @@
<!-- MQTT Acceptor -->
- <acceptor name="mqtt">tcp://${host}:${mqtt.port}?protocols=MQTT</acceptor>
+ <acceptor name="mqtt">tcp://${host}:${mqtt.port}?protocols=MQTT;useEpoll=true</acceptor>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/stomp-acceptor.txt
----------------------------------------------------------------------
diff --git a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/stomp-acceptor.txt b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/stomp-acceptor.txt
index 6c79165..bfe4d8b 100644
--- a/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/stomp-acceptor.txt
+++ b/artemis-cli/src/main/resources/org/apache/activemq/artemis/cli/commands/etc/stomp-acceptor.txt
@@ -1,3 +1,3 @@
<!-- STOMP Acceptor. -->
- <acceptor name="stomp">tcp://${host}:${stomp.port}?protocols=STOMP</acceptor>
+ <acceptor name="stomp">tcp://${host}:${stomp.port}?protocols=STOMP;useEpoll=true</acceptor>
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
index cc062d3..4997e0f 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
@@ -217,15 +217,11 @@ public class NettyConnector extends AbstractConnector {
private String servletPath;
- private int nioRemotingThreads;
-
- private boolean useNioGlobalWorkerPool;
-
private boolean useEpoll;
- private int epollRemotingThreads;
+ private int remotingThreads;
- private boolean useEpollGlobalWorkerPool;
+ private boolean useGlobalWorkerPool;
private ScheduledExecutorService scheduledThreadPool;
@@ -293,17 +289,14 @@ public class NettyConnector extends AbstractConnector {
httpUpgradeEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.HTTP_UPGRADE_ENABLED_PROP_NAME, TransportConstants.DEFAULT_HTTP_UPGRADE_ENABLED, configuration);
- nioRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
+ remotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
+ remotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.REMOTING_THREADS_PROPNAME, remotingThreads, configuration);
- useNioGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_NIO_GLOBAL_WORKER_POOL, configuration);
+ useGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_GLOBAL_WORKER_POOL, configuration);
+ useGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_GLOBAL_WORKER_POOL_PROP_NAME, useGlobalWorkerPool, configuration);
useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration);
- epollRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME, -1, configuration);
-
- useEpollGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL_GLOBAL_WORKER_POOL, configuration);
-
-
useServlet = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_SERVLET_PROP_NAME, TransportConstants.DEFAULT_USE_SERVLET, configuration);
host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, TransportConstants.DEFAULT_HOST, configuration);
port = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT, configuration);
@@ -387,46 +380,31 @@ public class NettyConnector extends AbstractConnector {
return;
}
- // Default to number of cores * 3
- int defaultThreadsToUse = Runtime.getRuntime().availableProcessors() * 3;
-
- if (useEpoll) {
- if (Epoll.isAvailable()) {
- int epollThreadsToUse;
- if (epollRemotingThreads == -1) {
- epollThreadsToUse = defaultThreadsToUse;
- } else {
- epollThreadsToUse = this.epollRemotingThreads;
- }
- if (useEpollGlobalWorkerPool) {
- channelClazz = EpollSocketChannel.class;
- group = SharedEventLoopGroup.getInstance((threadFactory -> new EpollEventLoopGroup(epollThreadsToUse, threadFactory)));
- } else {
- channelClazz = EpollSocketChannel.class;
- group = new EpollEventLoopGroup(epollThreadsToUse);
- }
- logger.info("Connector using native epoll");
-
- } else {
- logger.warn("Connector unable to load native epoll, will continue and load nio");
- }
+ if (remotingThreads == -1) {
+ // Default to number of cores * 3
+ remotingThreads = Runtime.getRuntime().availableProcessors() * 3;
}
- if (channelClazz == null || group == null) {
- int nioThreadsToUse;
- if (nioRemotingThreads == -1) {
- nioThreadsToUse = defaultThreadsToUse;
+ if (useEpoll && Epoll.isAvailable()) {
+ if (useGlobalWorkerPool) {
+ group = SharedEventLoopGroup.getInstance((threadFactory -> new EpollEventLoopGroup(remotingThreads, threadFactory)));
} else {
- nioThreadsToUse = this.nioRemotingThreads;
+ group = new EpollEventLoopGroup(remotingThreads);
}
- if (useNioGlobalWorkerPool) {
+
+ channelClazz = EpollSocketChannel.class;
+ logger.info("Connector " + this + " using native epoll");
+ } else {
+ if (useGlobalWorkerPool) {
channelClazz = NioSocketChannel.class;
- group = SharedEventLoopGroup.getInstance((threadFactory -> new NioEventLoopGroup(nioThreadsToUse, threadFactory)));
+ group = SharedEventLoopGroup.getInstance((threadFactory -> new NioEventLoopGroup(remotingThreads, threadFactory)));
} else {
channelClazz = NioSocketChannel.class;
- group = new NioEventLoopGroup(nioThreadsToUse);
+ group = new NioEventLoopGroup(remotingThreads);
}
- logger.info("Connector using nio");
+
+ channelClazz = NioSocketChannel.class;
+ logger.info("Connector + " + this + " using nio");
}
// if we are a servlet wrap the socketChannelFactory
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
index 12840c1..4293000 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
@@ -49,11 +49,15 @@ public class TransportConstants {
public static final String USE_NIO_PROP_NAME = "useNio";
- public static final String USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME = "useNioGlobalWorkerPool";
-
public static final String USE_EPOLL_PROP_NAME = "useEpoll";
- public static final String USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME = "useEpollGlobalWorkerPool";
+ @Deprecated
+ /**
+ * @deprecated Use USE_GLOBAL_WORKER_POOL_PROP_NAME
+ */
+ public static final String USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME = "useNioGlobalWorkerPool";
+
+ public static final String USE_GLOBAL_WORKER_POOL_PROP_NAME = "useGlobalWorkerPool";
public static final String USE_INVM_PROP_NAME = "useInvm";
@@ -115,9 +119,13 @@ public class TransportConstants {
public static final String TCP_RECEIVEBUFFER_SIZE_PROPNAME = "tcpReceiveBufferSize";
+ @Deprecated
+ /**
+ * @deprecated Use REMOTING_THREADS_PROPNAME
+ */
public static final String NIO_REMOTING_THREADS_PROPNAME = "nioRemotingThreads";
- public static final String EPOLL_REMOTING_THREADS_PROPNAME = "epollRemotingThreads";
+ public static final String REMOTING_THREADS_PROPNAME = "RemotingThreads";
public static final String BATCH_DELAY = "batchDelay";
@@ -131,9 +139,7 @@ public class TransportConstants {
public static final boolean DEFAULT_SSL_ENABLED = false;
- public static final boolean DEFAULT_USE_NIO_GLOBAL_WORKER_POOL = true;
-
- public static final boolean DEFAULT_USE_EPOLL_GLOBAL_WORKER_POOL = true;
+ public static final boolean DEFAULT_USE_GLOBAL_WORKER_POOL = true;
public static final boolean DEFAULT_USE_EPOLL = true;
@@ -248,7 +254,7 @@ public class TransportConstants {
allowableAcceptorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME);
- allowableAcceptorKeys.add(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME);
+ allowableAcceptorKeys.add(TransportConstants.REMOTING_THREADS_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.BATCH_DELAY);
allowableAcceptorKeys.add(TransportConstants.DIRECT_DELIVER);
allowableAcceptorKeys.add(TransportConstants.CLUSTER_CONNECTION);
@@ -280,7 +286,7 @@ public class TransportConstants {
allowableConnectorKeys.add(TransportConstants.USE_NIO_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.USE_EPOLL_PROP_NAME);
- allowableConnectorKeys.add(TransportConstants.USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME);
+ allowableConnectorKeys.add(TransportConstants.USE_GLOBAL_WORKER_POOL_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.HOST_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.PORT_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.LOCAL_ADDRESS_PROP_NAME);
@@ -298,7 +304,7 @@ public class TransportConstants {
allowableConnectorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME);
allowableConnectorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME);
allowableConnectorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME);
- allowableConnectorKeys.add(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME);
+ allowableConnectorKeys.add(TransportConstants.REMOTING_THREADS_PROPNAME);
allowableConnectorKeys.add(TransportConstants.BATCH_DELAY);
allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropMaskPassword());
allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropPasswordCodec());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
index 50faa46..3f60a7c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
@@ -89,9 +89,13 @@ import org.jboss.logging.Logger;
*/
public class NettyAcceptor extends AbstractAcceptor {
+ public static String INVM_ACCEPTOR_TYPE = "IN-VM";
+ public static String NIO_ACCEPTOR_TYPE = "NIO";
+ public static String EPOLL_ACCEPTOR_TYPE = "EPOLL";
+
static {
// Disable default Netty leak detection if the Netty leak detection level system properties are not in use
- if ( System.getProperty("io.netty.leakDetectionLevel") == null && System.getProperty("io.netty.leakDetection.level") == null) {
+ if (System.getProperty("io.netty.leakDetectionLevel") == null && System.getProperty("io.netty.leakDetection.level") == null) {
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.DISABLED);
}
}
@@ -157,9 +161,7 @@ public class NettyAcceptor extends AbstractAcceptor {
private final int tcpReceiveBufferSize;
- private final int nioRemotingThreads;
-
- private final int epollRemotingThreads;
+ private int remotingThreads;
private final ConcurrentMap<Object, NettyServerConnection> connections = new ConcurrentHashMap<>();
@@ -208,12 +210,11 @@ public class NettyAcceptor extends AbstractAcceptor {
sslEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.SSL_ENABLED_PROP_NAME, TransportConstants.DEFAULT_SSL_ENABLED, configuration);
- nioRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
+ remotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
+ remotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.REMOTING_THREADS_PROPNAME, remotingThreads, configuration);
- epollRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME, -1, configuration);
useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration);
-
backlog = ConfigurationHelper.getIntProperty(TransportConstants.BACKLOG_PROP_NAME, -1, configuration);
useInvm = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_INVM_PROP_NAME, TransportConstants.DEFAULT_USE_INVM, configuration);
@@ -278,51 +279,40 @@ public class NettyAcceptor extends AbstractAcceptor {
return;
}
+ String acceptorType;
+
if (useInvm) {
+ acceptorType = INVM_ACCEPTOR_TYPE;
channelClazz = LocalServerChannel.class;
eventLoopGroup = new LocalEventLoopGroup();
} else {
- // Default to number of cores * 3
- int defaultThreadsToUse = Runtime.getRuntime().availableProcessors() * 3;
-
- if (useEpoll) {
- if (Epoll.isAvailable()) {
- int epollThreadsToUse;
- if (epollRemotingThreads == -1) {
- epollThreadsToUse = defaultThreadsToUse;
- } else {
- epollThreadsToUse = this.epollRemotingThreads;
- }
- channelClazz = EpollServerSocketChannel.class;
- eventLoopGroup = new EpollEventLoopGroup(epollThreadsToUse, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
- @Override
- public ActiveMQThreadFactory run() {
- return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
- }
- }));
- logger.info("Acceptor using native epoll");
- } else {
- logger.warn("Acceptor unable to load native epoll, will continue and load nio");
- }
+ if (remotingThreads == -1) {
+ // Default to number of cores * 3
+ remotingThreads = Runtime.getRuntime().availableProcessors() * 3;
}
- if (channelClazz == null || eventLoopGroup == null) {
- int nioThreadsToUse;
- if (nioRemotingThreads == -1) {
- nioThreadsToUse = defaultThreadsToUse;
- } else {
- nioThreadsToUse = nioRemotingThreads;
- }
+ if (useEpoll && Epoll.isAvailable()) {
+ channelClazz = EpollServerSocketChannel.class;
+ eventLoopGroup = new EpollEventLoopGroup(remotingThreads, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
+ @Override
+ public ActiveMQThreadFactory run() {
+ return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
+ }
+ }));
+ acceptorType = EPOLL_ACCEPTOR_TYPE;
+ logger.debug("Acceptor using native epoll");
+ } else {
channelClazz = NioServerSocketChannel.class;
- eventLoopGroup = new NioEventLoopGroup(nioThreadsToUse, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
+ eventLoopGroup = new NioEventLoopGroup(remotingThreads, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
@Override
public ActiveMQThreadFactory run() {
return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
}
}));
- logger.info("Acceptor using nio");
+ acceptorType = NIO_ACCEPTOR_TYPE;
+ logger.debug("Acceptor using nio");
}
}
@@ -384,7 +374,7 @@ public class NettyAcceptor extends AbstractAcceptor {
batchFlusherFuture = scheduledThreadPool.scheduleWithFixedDelay(flusher, batchDelay, batchDelay, TimeUnit.MILLISECONDS);
}
- ActiveMQServerLogger.LOGGER.startedAcceptor(host, port, protocolsString);
+ ActiveMQServerLogger.LOGGER.startedAcceptor(acceptorType, host, port, protocolsString);
}
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/096556f3/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
index 9152eb5..46a07df 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
@@ -169,8 +169,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
void journalUnreferencedMessage(Long messageID);
@LogMessage(level = Logger.Level.INFO)
- @Message(id = 221020, value = "Started Acceptor at {0}:{1,number,#} for protocols [{2}]", format = Message.Format.MESSAGE_FORMAT)
- void startedAcceptor(String host, Integer port, String enabledProtocols);
+ @Message(id = 221020, value = "Started {0} Acceptor at {1}:{2,number,#} for protocols [{3}]", format = Message.Format.MESSAGE_FORMAT)
+ void startedAcceptor(String acceptorType, String host, Integer port, String enabledProtocols);
@LogMessage(level = Logger.Level.INFO)
@Message(id = 221021, value = "failed to remove connection", format = Message.Format.MESSAGE_FORMAT)
[2/4] activemq-artemis git commit: ARTEMIS-994 Support Netty Native
Epoll on Linux
Posted by jb...@apache.org.
ARTEMIS-994 Support Netty Native Epoll on Linux
The following changes are made to support Epoll.
Refactored SharedNioEventLoopGroup into renamed SharedEventLoopGroup to be generic (as so we can re-use for both Nio and Epoll)
Add support and toggles for Epoll in NettyAcceptor and NettyConnector (with fall back to NIO if cannot load Epoll)
Removal from code of PartialPooledByteBufAllocator, caused bad address when doing native, and no longer needed - see jira discussion
New Connector Properties:
useEpoll - toggles to use epoll or not, default true (but we failback to nio gracefully)
remotingThreads = same behaviour as nioRemotingThreads. Previous property is depreated.
useGlobalWorkerPool = same behaviour as useNioGlobalWorkerPool. Old property is deprecated.
New Acceptor Properties:
useEpoll - toggles to use epoll or not, default true (but we failback to nio gracefully)
useGlobalWorkerPool = same behaviour as useNioGlobalWorkerPool but for Epoll.
This closes #1093
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/a610748c
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/a610748c
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/a610748c
Branch: refs/heads/master
Commit: a610748c09ef18f0f4f589a9bf6ac2a745b52977
Parents: 2c9b028
Author: Michael Andr� Pearce <mi...@me.com>
Authored: Wed Mar 15 06:06:45 2017 +0000
Committer: Justin Bertram <jb...@apache.org>
Committed: Thu Mar 23 18:16:48 2017 -0500
----------------------------------------------------------------------
.../impl/netty/DelegatingEventLoopGroup.java | 198 +++++++++++++++++++
.../remoting/impl/netty/NettyConnection.java | 3 +-
.../remoting/impl/netty/NettyConnector.java | 66 +++++--
.../impl/netty/SharedEventLoopGroup.java | 112 +++++++++++
.../impl/netty/SharedNioEventLoopGroup.java | 111 -----------
.../remoting/impl/netty/TransportConstants.java | 15 ++
.../core/remoting/impl/netty/NettyAcceptor.java | 64 ++++--
7 files changed, 430 insertions(+), 139 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/DelegatingEventLoopGroup.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/DelegatingEventLoopGroup.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/DelegatingEventLoopGroup.java
new file mode 100644
index 0000000..f682893
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/DelegatingEventLoopGroup.java
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.core.remoting.impl.netty;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Spliterator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.EventLoop;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
+
+public class DelegatingEventLoopGroup implements EventLoopGroup {
+
+ private final EventLoopGroup delegate;
+
+ public DelegatingEventLoopGroup(EventLoopGroup eventLoopGroup) {
+ this.delegate = eventLoopGroup;
+ }
+
+ @Override
+ public EventLoop next() {
+ return delegate.next();
+ }
+
+ @Override
+ public ChannelFuture register(Channel channel) {
+ return delegate.register(channel);
+ }
+
+ @Override
+ public ChannelFuture register(ChannelPromise channelPromise) {
+ return delegate.register(channelPromise);
+ }
+
+ @Override
+ @Deprecated
+ public ChannelFuture register(Channel channel, ChannelPromise channelPromise) {
+ return delegate.register(channel, channelPromise);
+ }
+
+ @Override
+ public boolean isShuttingDown() {
+ return delegate.isShuttingDown();
+ }
+
+ @Override
+ public Future<?> shutdownGracefully() {
+ return delegate.shutdownGracefully();
+ }
+
+ @Override
+ public Future<?> shutdownGracefully(long l, long l1, TimeUnit timeUnit) {
+ return delegate.shutdownGracefully(l, l1, timeUnit);
+ }
+
+ @Override
+ public Future<?> terminationFuture() {
+ return delegate.terminationFuture();
+ }
+
+ @Override
+ @Deprecated
+ public void shutdown() {
+ delegate.shutdown();
+ }
+
+ @Override
+ @Deprecated
+ public List<Runnable> shutdownNow() {
+ return delegate.shutdownNow();
+ }
+
+ @Override
+ public Iterator<EventExecutor> iterator() {
+ return delegate.iterator();
+ }
+
+ @Override
+ public Future<?> submit(Runnable runnable) {
+ return delegate.submit(runnable);
+ }
+
+ @Override
+ public <T> Future<T> submit(Runnable runnable, T t) {
+ return delegate.submit(runnable, t);
+ }
+
+ @Override
+ public <T> Future<T> submit(Callable<T> callable) {
+ return delegate.submit(callable);
+ }
+
+ @Override
+ public io.netty.util.concurrent.ScheduledFuture<?> schedule(Runnable runnable, long l, TimeUnit timeUnit) {
+ return delegate.schedule(runnable, l, timeUnit);
+ }
+
+ @Override
+ public <V> io.netty.util.concurrent.ScheduledFuture<V> schedule(Callable<V> callable, long l, TimeUnit timeUnit) {
+ return delegate.schedule(callable, l, timeUnit);
+ }
+
+ @Override
+ public io.netty.util.concurrent.ScheduledFuture<?> scheduleAtFixedRate(Runnable runnable,
+ long l,
+ long l1,
+ TimeUnit timeUnit) {
+ return delegate.scheduleAtFixedRate(runnable, l, l1, timeUnit);
+ }
+
+ @Override
+ public io.netty.util.concurrent.ScheduledFuture<?> scheduleWithFixedDelay(Runnable runnable,
+ long l,
+ long l1,
+ TimeUnit timeUnit) {
+ return delegate.scheduleWithFixedDelay(runnable, l, l1, timeUnit);
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return delegate.isShutdown();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return delegate.isTerminated();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
+ return delegate.awaitTermination(timeout, unit);
+ }
+
+ @Override
+ public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+ return delegate.invokeAll(tasks);
+ }
+
+ @Override
+ public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
+ long timeout,
+ TimeUnit unit) throws InterruptedException {
+ return delegate.invokeAll(tasks, timeout, unit);
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+ return delegate.invokeAny(tasks);
+ }
+
+ @Override
+ public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
+ long timeout,
+ TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+ return delegate.invokeAny(tasks, timeout, unit);
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ delegate.execute(command);
+ }
+
+ @Override
+ public void forEach(Consumer<? super EventExecutor> action) {
+ delegate.forEach(action);
+ }
+
+ @Override
+ public Spliterator<EventExecutor> spliterator() {
+ return delegate.spliterator();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/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 c3a71c5..2355439 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
@@ -23,7 +23,6 @@ import java.util.Map;
import java.util.concurrent.Semaphore;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
@@ -216,7 +215,7 @@ public class NettyConnection implements Connection {
@Override
public ActiveMQBuffer createTransportBuffer(final int size, boolean pooled) {
- return new ChannelBufferWrapper(PooledByteBufAllocator.DEFAULT.directBuffer(size), true);
+ return new ChannelBufferWrapper(channel.alloc().directBuffer(size), true);
}
@Override
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
index 4a5e9e6..cc062d3 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnector.java
@@ -58,6 +58,9 @@ import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -218,6 +221,12 @@ public class NettyConnector extends AbstractConnector {
private boolean useNioGlobalWorkerPool;
+ private boolean useEpoll;
+
+ private int epollRemotingThreads;
+
+ private boolean useEpollGlobalWorkerPool;
+
private ScheduledExecutorService scheduledThreadPool;
private Executor closeExecutor;
@@ -288,6 +297,13 @@ public class NettyConnector extends AbstractConnector {
useNioGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_NIO_GLOBAL_WORKER_POOL, configuration);
+ useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration);
+
+ epollRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME, -1, configuration);
+
+ useEpollGlobalWorkerPool = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL_GLOBAL_WORKER_POOL, configuration);
+
+
useServlet = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_SERVLET_PROP_NAME, TransportConstants.DEFAULT_USE_SERVLET, configuration);
host = ConfigurationHelper.getStringProperty(TransportConstants.HOST_PROP_NAME, TransportConstants.DEFAULT_HOST, configuration);
port = ConfigurationHelper.getIntProperty(TransportConstants.PORT_PROP_NAME, TransportConstants.DEFAULT_PORT, configuration);
@@ -371,22 +387,46 @@ public class NettyConnector extends AbstractConnector {
return;
}
- int threadsToUse;
+ // Default to number of cores * 3
+ int defaultThreadsToUse = Runtime.getRuntime().availableProcessors() * 3;
- if (nioRemotingThreads == -1) {
- // Default to number of cores * 3
+ if (useEpoll) {
+ if (Epoll.isAvailable()) {
+ int epollThreadsToUse;
+ if (epollRemotingThreads == -1) {
+ epollThreadsToUse = defaultThreadsToUse;
+ } else {
+ epollThreadsToUse = this.epollRemotingThreads;
+ }
+ if (useEpollGlobalWorkerPool) {
+ channelClazz = EpollSocketChannel.class;
+ group = SharedEventLoopGroup.getInstance((threadFactory -> new EpollEventLoopGroup(epollThreadsToUse, threadFactory)));
+ } else {
+ channelClazz = EpollSocketChannel.class;
+ group = new EpollEventLoopGroup(epollThreadsToUse);
+ }
+ logger.info("Connector using native epoll");
- threadsToUse = Runtime.getRuntime().availableProcessors() * 3;
- } else {
- threadsToUse = this.nioRemotingThreads;
+ } else {
+ logger.warn("Connector unable to load native epoll, will continue and load nio");
+ }
}
- if (useNioGlobalWorkerPool) {
- channelClazz = NioSocketChannel.class;
- group = SharedNioEventLoopGroup.getInstance(threadsToUse);
- } else {
- channelClazz = NioSocketChannel.class;
- group = new NioEventLoopGroup(threadsToUse);
+ if (channelClazz == null || group == null) {
+ int nioThreadsToUse;
+ if (nioRemotingThreads == -1) {
+ nioThreadsToUse = defaultThreadsToUse;
+ } else {
+ nioThreadsToUse = this.nioRemotingThreads;
+ }
+ if (useNioGlobalWorkerPool) {
+ channelClazz = NioSocketChannel.class;
+ group = SharedEventLoopGroup.getInstance((threadFactory -> new NioEventLoopGroup(nioThreadsToUse, threadFactory)));
+ } else {
+ channelClazz = NioSocketChannel.class;
+ group = new NioEventLoopGroup(nioThreadsToUse);
+ }
+ logger.info("Connector using nio");
}
// if we are a servlet wrap the socketChannelFactory
@@ -1053,7 +1093,7 @@ public class NettyConnector extends AbstractConnector {
}
public static void clearThreadPools() {
- SharedNioEventLoopGroup.forceShutdown();
+ SharedEventLoopGroup.forceShutdown();
}
private static ClassLoader getThisClassLoader() {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
new file mode 100644
index 0000000..0af54de
--- /dev/null
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedEventLoopGroup.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.core.remoting.impl.netty;
+
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import io.netty.util.concurrent.Promise;
+import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
+
+public class SharedEventLoopGroup extends DelegatingEventLoopGroup {
+
+ private static SharedEventLoopGroup instance;
+
+ private final AtomicReference<ScheduledFuture<?>> shutdown = new AtomicReference<>();
+ private final AtomicLong channelFactoryCount = new AtomicLong();
+ private final Promise<?> terminationPromise = ImmediateEventExecutor.INSTANCE.newPromise();
+
+ private SharedEventLoopGroup(EventLoopGroup eventLoopGroup) {
+ super(eventLoopGroup);
+ }
+
+ public static synchronized void forceShutdown() {
+ if (instance != null) {
+ instance.shutdown();
+ instance.channelFactoryCount.set(0);
+ instance = null;
+ }
+ }
+
+ public static synchronized SharedEventLoopGroup getInstance(Function<ThreadFactory, EventLoopGroup> eventLoopGroupSupplier) {
+ if (instance != null) {
+ ScheduledFuture f = instance.shutdown.getAndSet(null);
+ if (f != null) {
+ f.cancel(false);
+ }
+ } else {
+ instance = new SharedEventLoopGroup(eventLoopGroupSupplier.apply(AccessController.doPrivileged(new PrivilegedAction<ThreadFactory>() {
+ @Override
+ public ThreadFactory run() {
+ return new ActiveMQThreadFactory("ActiveMQ-client-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
+ }
+ })));
+ }
+ instance.channelFactoryCount.incrementAndGet();
+ return instance;
+ }
+
+ @Override
+ public Future<?> terminationFuture() {
+ return terminationPromise;
+ }
+
+ @Override
+ public Future<?> shutdownGracefully() {
+ return shutdownGracefully(100, 3000, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public Future<?> shutdownGracefully(final long l, final long l2, final TimeUnit timeUnit) {
+ if (channelFactoryCount.decrementAndGet() == 0) {
+ shutdown.compareAndSet(null, next().scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ synchronized (SharedEventLoopGroup.class) {
+ if (shutdown.get() != null) {
+ Future<?> future = SharedEventLoopGroup.super.shutdownGracefully(l, l2, timeUnit);
+ future.addListener(new FutureListener<Object>() {
+ @Override
+ public void operationComplete(Future future) throws Exception {
+ if (future.isSuccess()) {
+ terminationPromise.setSuccess(null);
+ } else {
+ terminationPromise.setFailure(future.cause());
+ }
+ }
+ });
+ instance = null;
+ }
+ }
+ }
+
+ }, 10, 10, TimeUnit.SECONDS));
+ }
+ return terminationPromise;
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedNioEventLoopGroup.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedNioEventLoopGroup.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedNioEventLoopGroup.java
deleted file mode 100644
index 0750105..0000000
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/SharedNioEventLoopGroup.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.core.remoting.impl.netty;
-
-import java.security.AccessController;
-import java.security.PrivilegedAction;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
-
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.FutureListener;
-import io.netty.util.concurrent.ImmediateEventExecutor;
-import io.netty.util.concurrent.Promise;
-import org.apache.activemq.artemis.core.client.impl.ClientSessionFactoryImpl;
-import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
-
-public class SharedNioEventLoopGroup extends NioEventLoopGroup {
-
- private static SharedNioEventLoopGroup instance;
-
- private final AtomicReference<ScheduledFuture<?>> shutdown = new AtomicReference<>();
- private final AtomicLong nioChannelFactoryCount = new AtomicLong();
- private final Promise<?> terminationPromise = ImmediateEventExecutor.INSTANCE.newPromise();
-
- private SharedNioEventLoopGroup(int numThreads, ThreadFactory factory) {
- super(numThreads, factory);
- }
-
- public static synchronized void forceShutdown() {
- if (instance != null) {
- instance.shutdown();
- instance.nioChannelFactoryCount.set(0);
- instance = null;
- }
- }
-
- public static synchronized SharedNioEventLoopGroup getInstance(int numThreads) {
- if (instance != null) {
- ScheduledFuture f = instance.shutdown.getAndSet(null);
- if (f != null) {
- f.cancel(false);
- }
- } else {
- instance = new SharedNioEventLoopGroup(numThreads, AccessController.doPrivileged(new PrivilegedAction<ThreadFactory>() {
- @Override
- public ThreadFactory run() {
- return new ActiveMQThreadFactory("ActiveMQ-client-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
- }
- }));
- }
- instance.nioChannelFactoryCount.incrementAndGet();
- return instance;
- }
-
- @Override
- public Future<?> terminationFuture() {
- return terminationPromise;
- }
-
- @Override
- public Future<?> shutdownGracefully() {
- return shutdownGracefully(100, 3000, TimeUnit.MILLISECONDS);
- }
-
- @Override
- public Future<?> shutdownGracefully(final long l, final long l2, final TimeUnit timeUnit) {
- if (nioChannelFactoryCount.decrementAndGet() == 0) {
- shutdown.compareAndSet(null, next().scheduleAtFixedRate(new Runnable() {
- @Override
- public void run() {
- synchronized (SharedNioEventLoopGroup.class) {
- if (shutdown.get() != null) {
- Future<?> future = SharedNioEventLoopGroup.super.shutdownGracefully(l, l2, timeUnit);
- future.addListener(new FutureListener<Object>() {
- @Override
- public void operationComplete(Future future) throws Exception {
- if (future.isSuccess()) {
- terminationPromise.setSuccess(null);
- } else {
- terminationPromise.setFailure(future.cause());
- }
- }
- });
- instance = null;
- }
- }
- }
-
- }, 10, 10, TimeUnit.SECONDS));
- }
- return terminationPromise;
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
index 14efb79..12840c1 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/TransportConstants.java
@@ -51,6 +51,10 @@ public class TransportConstants {
public static final String USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME = "useNioGlobalWorkerPool";
+ public static final String USE_EPOLL_PROP_NAME = "useEpoll";
+
+ public static final String USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME = "useEpollGlobalWorkerPool";
+
public static final String USE_INVM_PROP_NAME = "useInvm";
public static final String ACTIVEMQ_SERVER_NAME = "activemqServerName";
@@ -113,6 +117,8 @@ public class TransportConstants {
public static final String NIO_REMOTING_THREADS_PROPNAME = "nioRemotingThreads";
+ public static final String EPOLL_REMOTING_THREADS_PROPNAME = "epollRemotingThreads";
+
public static final String BATCH_DELAY = "batchDelay";
public static final String DIRECT_DELIVER = "directDeliver";
@@ -127,6 +133,10 @@ public class TransportConstants {
public static final boolean DEFAULT_USE_NIO_GLOBAL_WORKER_POOL = true;
+ public static final boolean DEFAULT_USE_EPOLL_GLOBAL_WORKER_POOL = true;
+
+ public static final boolean DEFAULT_USE_EPOLL = true;
+
public static final boolean DEFAULT_USE_INVM = false;
public static final boolean DEFAULT_USE_SERVLET = false;
@@ -218,6 +228,7 @@ public class TransportConstants {
allowableAcceptorKeys.add(TransportConstants.HTTP_SERVER_SCAN_PERIOD_PROP_NAME);
allowableAcceptorKeys.add(TransportConstants.HTTP_UPGRADE_ENABLED_PROP_NAME);
allowableAcceptorKeys.add(TransportConstants.USE_NIO_PROP_NAME);
+ allowableAcceptorKeys.add(TransportConstants.USE_EPOLL_PROP_NAME);
allowableAcceptorKeys.add(TransportConstants.USE_INVM_PROP_NAME);
allowableAcceptorKeys.add(TransportConstants.PROTOCOL_PROP_NAME);
allowableAcceptorKeys.add(TransportConstants.PROTOCOLS_PROP_NAME);
@@ -237,6 +248,7 @@ public class TransportConstants {
allowableAcceptorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME);
+ allowableAcceptorKeys.add(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME);
allowableAcceptorKeys.add(TransportConstants.BATCH_DELAY);
allowableAcceptorKeys.add(TransportConstants.DIRECT_DELIVER);
allowableAcceptorKeys.add(TransportConstants.CLUSTER_CONNECTION);
@@ -267,6 +279,8 @@ public class TransportConstants {
allowableConnectorKeys.add(TransportConstants.SERVLET_PATH);
allowableConnectorKeys.add(TransportConstants.USE_NIO_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.USE_NIO_GLOBAL_WORKER_POOL_PROP_NAME);
+ allowableConnectorKeys.add(TransportConstants.USE_EPOLL_PROP_NAME);
+ allowableConnectorKeys.add(TransportConstants.USE_EPOLL_GLOBAL_WORKER_POOL_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.HOST_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.PORT_PROP_NAME);
allowableConnectorKeys.add(TransportConstants.LOCAL_ADDRESS_PROP_NAME);
@@ -284,6 +298,7 @@ public class TransportConstants {
allowableConnectorKeys.add(TransportConstants.TCP_SENDBUFFER_SIZE_PROPNAME);
allowableConnectorKeys.add(TransportConstants.TCP_RECEIVEBUFFER_SIZE_PROPNAME);
allowableConnectorKeys.add(TransportConstants.NIO_REMOTING_THREADS_PROPNAME);
+ allowableConnectorKeys.add(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME);
allowableConnectorKeys.add(TransportConstants.BATCH_DELAY);
allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropMaskPassword());
allowableConnectorKeys.add(ActiveMQDefaultConfiguration.getPropPasswordCodec());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/a610748c/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
index a46684d..50faa46 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyAcceptor.java
@@ -44,6 +44,9 @@ import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.ServerChannel;
+import io.netty.channel.epoll.Epoll;
+import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
@@ -117,6 +120,8 @@ public class NettyAcceptor extends AbstractAcceptor {
private final boolean useInvm;
+ private final boolean useEpoll;
+
private final ProtocolHandler protocolHandler;
private final String host;
@@ -154,6 +159,8 @@ public class NettyAcceptor extends AbstractAcceptor {
private final int nioRemotingThreads;
+ private final int epollRemotingThreads;
+
private final ConcurrentMap<Object, NettyServerConnection> connections = new ConcurrentHashMap<>();
private final Map<String, Object> configuration;
@@ -202,6 +209,11 @@ public class NettyAcceptor extends AbstractAcceptor {
sslEnabled = ConfigurationHelper.getBooleanProperty(TransportConstants.SSL_ENABLED_PROP_NAME, TransportConstants.DEFAULT_SSL_ENABLED, configuration);
nioRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.NIO_REMOTING_THREADS_PROPNAME, -1, configuration);
+
+ epollRemotingThreads = ConfigurationHelper.getIntProperty(TransportConstants.EPOLL_REMOTING_THREADS_PROPNAME, -1, configuration);
+ useEpoll = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_EPOLL_PROP_NAME, TransportConstants.DEFAULT_USE_EPOLL, configuration);
+
+
backlog = ConfigurationHelper.getIntProperty(TransportConstants.BACKLOG_PROP_NAME, -1, configuration);
useInvm = ConfigurationHelper.getBooleanProperty(TransportConstants.USE_INVM_PROP_NAME, TransportConstants.DEFAULT_USE_INVM, configuration);
@@ -270,22 +282,48 @@ public class NettyAcceptor extends AbstractAcceptor {
channelClazz = LocalServerChannel.class;
eventLoopGroup = new LocalEventLoopGroup();
} else {
- int threadsToUse;
-
- if (nioRemotingThreads == -1) {
- // Default to number of cores * 3
+ // Default to number of cores * 3
+ int defaultThreadsToUse = Runtime.getRuntime().availableProcessors() * 3;
+
+ if (useEpoll) {
+ if (Epoll.isAvailable()) {
+ int epollThreadsToUse;
+ if (epollRemotingThreads == -1) {
+ epollThreadsToUse = defaultThreadsToUse;
+ } else {
+ epollThreadsToUse = this.epollRemotingThreads;
+ }
- threadsToUse = Runtime.getRuntime().availableProcessors() * 3;
- } else {
- threadsToUse = this.nioRemotingThreads;
+ channelClazz = EpollServerSocketChannel.class;
+ eventLoopGroup = new EpollEventLoopGroup(epollThreadsToUse, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
+ @Override
+ public ActiveMQThreadFactory run() {
+ return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
+ }
+ }));
+ logger.info("Acceptor using native epoll");
+ } else {
+ logger.warn("Acceptor unable to load native epoll, will continue and load nio");
+ }
}
- channelClazz = NioServerSocketChannel.class;
- eventLoopGroup = new NioEventLoopGroup(threadsToUse, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
- @Override
- public ActiveMQThreadFactory run() {
- return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
+
+ if (channelClazz == null || eventLoopGroup == null) {
+ int nioThreadsToUse;
+ if (nioRemotingThreads == -1) {
+ nioThreadsToUse = defaultThreadsToUse;
+ } else {
+ nioThreadsToUse = nioRemotingThreads;
}
- }));
+
+ channelClazz = NioServerSocketChannel.class;
+ eventLoopGroup = new NioEventLoopGroup(nioThreadsToUse, AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
+ @Override
+ public ActiveMQThreadFactory run() {
+ return new ActiveMQThreadFactory("activemq-netty-threads", true, ClientSessionFactoryImpl.class.getClassLoader());
+ }
+ }));
+ logger.info("Acceptor using nio");
+ }
}
bootstrap = new ServerBootstrap();