mirror of https://github.com/apache/cassandra
Replace Thrift HsHa with LMAX Disruptor based implementation
patch by Pavel Yaskevich; reviewed by Aleksey Yeschenko for CASSANDRA-5582
This commit is contained in:
Pavel Yaskevich
parent
47ac188a3f
commit
98eec0a223
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -353,6 +353,7 @@
|
|||
<dependency groupId="com.googlecode.json-simple" artifactId="json-simple" version="1.1"/>
|
||||
<dependency groupId="com.github.stephenc.high-scale-lib" artifactId="high-scale-lib" version="1.1.2"/>
|
||||
<dependency groupId="com.github.stephenc" artifactId="jamm" version="0.2.5"/>
|
||||
<dependency groupId="com.thinkaurelius.thrift" artifactId="thrift-server" version="0.1"/>
|
||||
<dependency groupId="org.yaml" artifactId="snakeyaml" version="1.6"/>
|
||||
<dependency groupId="org.apache.thrift" artifactId="libthrift" version="0.9.0"/>
|
||||
|
||||
|
|
@ -459,7 +460,8 @@
|
|||
<dependency groupId="edu.stanford.ppl" artifactId="snaptree"/>
|
||||
<dependency groupId="org.mindrot" artifactId="jbcrypt"/>
|
||||
<dependency groupId="com.yammer.metrics" artifactId="metrics-core"/>
|
||||
|
||||
<dependency groupId="com.thinkaurelius.thrift" artifactId="thrift-server" version="0.1"/>
|
||||
|
||||
<dependency groupId="log4j" artifactId="log4j"/>
|
||||
<!-- cassandra has a hard dependency on log4j, so force slf4j's log4j provider at runtime -->
|
||||
<dependency groupId="org.slf4j" artifactId="slf4j-log4j12" scope="runtime"/>
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
|
|
@ -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<Runnable>(),
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue