diff --git a/CHANGES.txt b/CHANGES.txt index 04122b93f5..985b803e74 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -56,6 +56,7 @@ * Allow preparing timestamp, ttl and limit in CQL3 queries (CASSANDRA-4450) * Support native link w/o JNA in Java7 (CASSANDRA-3734) * Use SASL authentication in binary protocol v2 (CASSANDRA-5545) + * Replace Thrift HsHa with LMAX Disruptor based implementation (CASSANDRA-5582) 1.2.6 * Reduce SSTableLoader memory usage (CASSANDRA-5555) diff --git a/build.xml b/build.xml index 7e97e6603b..f5a21be39c 100644 --- a/build.xml +++ b/build.xml @@ -353,6 +353,7 @@ + @@ -459,7 +460,8 @@ - + + diff --git a/lib/disruptor-3.0.1.jar b/lib/disruptor-3.0.1.jar new file mode 100644 index 0000000000..1899ed0651 Binary files /dev/null and b/lib/disruptor-3.0.1.jar differ diff --git a/lib/thrift-server-0.1.jar b/lib/thrift-server-0.1.jar new file mode 100644 index 0000000000..2c595a0512 Binary files /dev/null and b/lib/thrift-server-0.1.jar differ diff --git a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java b/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java deleted file mode 100644 index ca838b1797..0000000000 --- a/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java +++ /dev/null @@ -1,103 +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.cassandra.thrift; - -import java.net.InetSocketAddress; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.TimeUnit; - -import org.apache.thrift.server.TThreadedSelectorServer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; -import org.apache.cassandra.concurrent.NamedThreadFactory; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.thrift.server.TServer; -import org.apache.thrift.transport.TNonblockingServerTransport; -import org.apache.thrift.transport.TNonblockingSocket; -import org.apache.thrift.transport.TTransportException; - -/** - * This is a interim solution till THRIFT-1167 gets committed... - * - * The idea here is to avoid sticking to one CPU for IO's. For better throughput - * it is spread across multiple threads. Number of selector thread can be the - * number of CPU available. - */ -public class CustomTHsHaServer extends TThreadedSelectorServer -{ - private static final Logger LOGGER = LoggerFactory.getLogger(CustomTHsHaServer.class.getName()); - - /** - * All the arguments to Non Blocking Server will apply here. In addition, - * executor pool will be responsible for creating the internal threads which - * will process the data. threads for selection usually are equal to the - * number of cpu's - */ - public CustomTHsHaServer(Args args) - { - super(args); - } - - protected boolean requestInvoke(FrameBuffer frameBuffer) - { - TNonblockingSocket socket = (TNonblockingSocket) frameBuffer.trans_; - ThriftSessionManager.instance.setCurrentSocket(socket.getSocketChannel().socket().getRemoteSocketAddress()); - frameBuffer.invoke(); - return true; - } - - public static class Factory implements TServerFactory - { - public TServer buildTServer(Args args) - { - if (DatabaseDescriptor.getClientEncryptionOptions().enabled) - throw new RuntimeException("Client SSL is not supported for non-blocking sockets (hsha). Please remove client ssl from the configuration."); - - final InetSocketAddress addr = args.addr; - TNonblockingServerTransport serverTransport; - try - { - serverTransport = new TCustomNonblockingServerSocket(addr, args.keepAlive, args.sendBufferSize, args.recvBufferSize); - } - catch (TTransportException e) - { - throw new RuntimeException(String.format("Unable to create thrift socket to %s:%s", addr.getAddress(), addr.getPort()), e); - } - - // This is NIO selector service but the invocation will be Multi-Threaded with the Executor service. - ExecutorService executorService = new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getRpcMinThreads(), - DatabaseDescriptor.getRpcMaxThreads(), - 60L, - TimeUnit.SECONDS, - new SynchronousQueue(), - new NamedThreadFactory("RPC-Thread"), "RPC-THREAD-POOL"); - TThreadedSelectorServer.Args serverArgs = new TThreadedSelectorServer.Args(serverTransport).inputTransportFactory(args.inTransportFactory) - .outputTransportFactory(args.outTransportFactory) - .inputProtocolFactory(args.tProtocolFactory) - .outputProtocolFactory(args.tProtocolFactory) - .processor(args.processor) - .selectorThreads(Runtime.getRuntime().availableProcessors()) - .executorService(executorService); - // Check for available processors in the system which will be equal to the IO Threads. - return new CustomTHsHaServer(serverArgs); - } - } -} diff --git a/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java b/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java new file mode 100644 index 0000000000..a757315867 --- /dev/null +++ b/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java @@ -0,0 +1,86 @@ +/* + * 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.cassandra.thrift; + +import java.net.InetSocketAddress; + +import com.thinkaurelius.thrift.Message; +import com.thinkaurelius.thrift.TDisruptorServer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.thrift.server.TServer; +import org.apache.thrift.transport.TNonblockingServerTransport; +import org.apache.thrift.transport.TNonblockingSocket; +import org.apache.thrift.transport.TTransportException; + +public class THsHaDisruptorServer extends TDisruptorServer +{ + private static final Logger logger = LoggerFactory.getLogger(THsHaDisruptorServer.class.getName()); + + /** + * All the arguments to Non Blocking Server will apply here. In addition, + * executor pool will be responsible for creating the internal threads which + * will process the data. threads for selection usually are equal to the + * number of cpu's + */ + public THsHaDisruptorServer(Args args) + { + super(args); + logger.info("Starting up {}", this); + } + + @Override + protected void beforeInvoke(Message buffer) + { + TNonblockingSocket socket = (TNonblockingSocket) buffer.transport; + ThriftSessionManager.instance.setCurrentSocket(socket.getSocketChannel().socket().getRemoteSocketAddress()); + } + + public static class Factory implements TServerFactory + { + public TServer buildTServer(Args args) + { + if (DatabaseDescriptor.getClientEncryptionOptions().enabled) + throw new RuntimeException("Client SSL is not supported for non-blocking sockets (hsha). Please remove client ssl from the configuration."); + + final InetSocketAddress addr = args.addr; + TNonblockingServerTransport serverTransport; + try + { + serverTransport = new TCustomNonblockingServerSocket(addr, args.keepAlive, args.sendBufferSize, args.recvBufferSize); + } + catch (TTransportException e) + { + throw new RuntimeException(String.format("Unable to create thrift socket to %s:%s", addr.getAddress(), addr.getPort()), e); + } + + com.thinkaurelius.thrift.util.TBinaryProtocol.Factory protocolFactory = new com.thinkaurelius.thrift.util.TBinaryProtocol.Factory(true, true); + + TDisruptorServer.Args serverArgs = new TDisruptorServer.Args(serverTransport).inputTransportFactory(args.inTransportFactory) + .outputTransportFactory(args.outTransportFactory) + .inputProtocolFactory(protocolFactory) + .outputProtocolFactory(protocolFactory) + .processor(args.processor); + + return new THsHaDisruptorServer(serverArgs); + } + } +} diff --git a/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java b/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java index 208f664012..4bf0acd097 100644 --- a/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java +++ b/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java @@ -21,6 +21,7 @@ package org.apache.cassandra.thrift; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.thinkaurelius.thrift.TDisruptorServer; import org.apache.thrift.server.TServer; /** @@ -53,7 +54,7 @@ public class TServerCustomFactory implements TServerFactory } else if(ThriftServer.HSHA.equalsIgnoreCase(serverType)) { - server = new CustomTHsHaServer.Factory().buildTServer(args); + server = new THsHaDisruptorServer.Factory().buildTServer(args); logger.info(String.format("Using custom half-sync/half-async thrift server on %s : %s", args.addr.getHostName(), args.addr.getPort())); } else