cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xe...@apache.org
Subject git commit: Replace Thrift HsHa with LMAX Disruptor based implementation patch by Pavel Yaskevich; reviewed by Aleksey Yeschenko for CASSANDRA-5582
Date Fri, 07 Jun 2013 22:41:10 GMT
Updated Branches:
  refs/heads/trunk 47ac188a3 -> 98eec0a22


Replace Thrift HsHa with LMAX Disruptor based implementation
patch by Pavel Yaskevich; reviewed by Aleksey Yeschenko for CASSANDRA-5582


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/98eec0a2
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/98eec0a2
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/98eec0a2

Branch: refs/heads/trunk
Commit: 98eec0a223251ecd8fec7ecc9e46b05497d631c6
Parents: 47ac188
Author: Pavel Yaskevich <xedin@apache.org>
Authored: Mon May 20 16:43:14 2013 -0700
Committer: Pavel Yaskevich <xedin@apache.org>
Committed: Fri Jun 7 15:36:48 2013 -0700

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 build.xml                                          |    4 +-
 lib/disruptor-3.0.1.jar                            |  Bin 0 -> 66843 bytes
 lib/thrift-server-0.1.jar                          |  Bin 0 -> 122900 bytes
 .../apache/cassandra/thrift/CustomTHsHaServer.java |  103 ---------------
 .../cassandra/thrift/THsHaDisruptorServer.java     |   86 ++++++++++++
 .../cassandra/thrift/TServerCustomFactory.java     |    3 +-
 7 files changed, 92 insertions(+), 105 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 04122b9..985b803 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)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 7e97e66..f5a21be 100644
--- a/build.xml
+++ b/build.xml
@@ -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"/>

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/lib/disruptor-3.0.1.jar
----------------------------------------------------------------------
diff --git a/lib/disruptor-3.0.1.jar b/lib/disruptor-3.0.1.jar
new file mode 100644
index 0000000..1899ed0
Binary files /dev/null and b/lib/disruptor-3.0.1.jar differ

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/lib/thrift-server-0.1.jar
----------------------------------------------------------------------
diff --git a/lib/thrift-server-0.1.jar b/lib/thrift-server-0.1.jar
new file mode 100644
index 0000000..2c595a0
Binary files /dev/null and b/lib/thrift-server-0.1.jar differ

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/src/java/org/apache/cassandra/thrift/CustomTHsHaServer.java
----------------------------------------------------------------------
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 ca838b1..0000000
--- 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<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);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/src/java/org/apache/cassandra/thrift/THsHaDisruptorServer.java
----------------------------------------------------------------------
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 0000000..a757315
--- /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);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/98eec0a2/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java b/src/java/org/apache/cassandra/thrift/TServerCustomFactory.java
index 208f664..4bf0acd 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


Mime
View raw message