cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jmcken...@apache.org
Subject [2/3] cassandra git commit: Merge branch 'cassandra-2.0' into cassandra-2.1
Date Thu, 05 Mar 2015 17:40:57 GMT
Merge branch 'cassandra-2.0' into cassandra-2.1

Conflicts:
	CHANGES.txt


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

Branch: refs/heads/trunk
Commit: b1c2536ccb244be078cddc0da6c788d2b7c0f4ec
Parents: 7902568 e56d9ef
Author: Joshua McKenzie <jmckenzie@apache.org>
Authored: Thu Mar 5 11:32:41 2015 -0600
Committer: Joshua McKenzie <jmckenzie@apache.org>
Committed: Thu Mar 5 11:32:41 2015 -0600

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 conf/cassandra.yaml                             |   8 ++
 .../org/apache/cassandra/config/Config.java     |   2 +
 .../cassandra/config/DatabaseDescriptor.java    |  19 ++++
 .../apache/cassandra/service/StorageProxy.java  |   8 +-
 .../cassandra/service/StorageProxyMBean.java    |   3 +
 .../transport/ConnectionLimitHandler.java       | 108 +++++++++++++++++++
 .../org/apache/cassandra/transport/Server.java  |   9 ++
 8 files changed, 157 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 137c0f1,4e34c9e..aa2e1f9
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,36 -1,5 +1,37 @@@
 -2.0.13:
 +2.1.4
 + * Make SSTableRewriter.abort() more robust to failure (CASSANDRA-8832)
 + * Remove cold_reads_to_omit from STCS (CASSANDRA-8860)
 + * Make EstimatedHistogram#percentile() use ceil instead of floor (CASSANDRA-8883)
 + * Fix top partitions reporting wrong cardinality (CASSANDRA-8834)
 + * Fix rare NPE in KeyCacheSerializer (CASSANDRA-8067)
 + * Pick sstables for validation as late as possible inc repairs (CASSANDRA-8366)
 + * Fix commitlog getPendingTasks to not increment (CASSANDRA-8856)
 + * Fix parallelism adjustment in range and secondary index queries
 +   when the first fetch does not satisfy the limit (CASSANDRA-8856)
 + * Check if the filtered sstables is non-empty in STCS (CASSANDRA-8843)
 + * Upgrade java-driver used for cassandra-stress (CASSANDRA-8842)
 + * Fix CommitLog.forceRecycleAllSegments() memory access error (CASSANDRA-8812)
 + * Improve assertions in Memory (CASSANDRA-8792)
 + * Fix SSTableRewriter cleanup (CASSANDRA-8802)
 + * Introduce SafeMemory for CompressionMetadata.Writer (CASSANDRA-8758)
 + * 'nodetool info' prints exception against older node (CASSANDRA-8796)
 + * Ensure SSTableReader.last corresponds exactly with the file end (CASSANDRA-8750)
 + * Make SSTableWriter.openEarly more robust and obvious (CASSANDRA-8747)
 + * Enforce SSTableReader.first/last (CASSANDRA-8744)
 + * Cleanup SegmentedFile API (CASSANDRA-8749)
 + * Avoid overlap with early compaction replacement (CASSANDRA-8683)
 + * Safer Resource Management++ (CASSANDRA-8707)
 + * Write partition size estimates into a system table (CASSANDRA-7688)
 + * cqlsh: Fix keys() and full() collection indexes in DESCRIBE output
 +   (CASSANDRA-8154)
 + * Show progress of streaming in nodetool netstats (CASSANDRA-8886)
 + * IndexSummaryBuilder utilises offheap memory, and shares data between
 +   each IndexSummary opened from it (CASSANDRA-8757)
 + * markCompacting only succeeds if the exact SSTableReader instances being 
 +   marked are in the live set (CASSANDRA-8689)
 + * cassandra-stress support for varint (CASSANDRA-8882)
 +Merged from 2.0:
+  * Add ability to limit number of native connections (CASSANDRA-8086)
   * Add offline tool to relevel sstables (CASSANDRA-8301)
   * Preserve stream ID for more protocol errors (CASSANDRA-8848)
   * Fix combining token() function with multi-column relations on

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/conf/cassandra.yaml
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/config/Config.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/service/StorageProxyMBean.java
----------------------------------------------------------------------

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
index 0000000,c45d2cb..7bcf280
mode 000000,100644..100644
--- a/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
+++ b/src/java/org/apache/cassandra/transport/ConnectionLimitHandler.java
@@@ -1,0 -1,111 +1,108 @@@
+ /*
+  * 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.transport;
+ 
+ 
++import io.netty.channel.ChannelHandler;
++import io.netty.channel.ChannelHandlerContext;
++import io.netty.channel.ChannelInboundHandlerAdapter;
+ import org.apache.cassandra.config.DatabaseDescriptor;
 -import org.jboss.netty.channel.ChannelHandler;
 -import org.jboss.netty.channel.ChannelHandlerContext;
 -import org.jboss.netty.channel.ChannelStateEvent;
 -import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
+ import org.slf4j.Logger;
+ import org.slf4j.LoggerFactory;
+ 
+ import java.net.InetAddress;
+ import java.net.InetSocketAddress;
 -import java.util.HashMap;
 -import java.util.Map;
+ import java.util.concurrent.ConcurrentHashMap;
+ import java.util.concurrent.ConcurrentMap;
+ import java.util.concurrent.atomic.AtomicLong;
+ 
+ 
+ /**
 - * {@link SimpleChannelUpstreamHandler} implementation which allows to limit the number
of concurrent
++ * {@link ChannelInboundHandlerAdapter} implementation which allows to limit the number
of concurrent
+  * connections to the Server. Be aware this <strong>MUST</strong> be shared
between all child channels.
+  */
+ @ChannelHandler.Sharable
 -final class ConnectionLimitHandler extends SimpleChannelUpstreamHandler
++final class ConnectionLimitHandler extends ChannelInboundHandlerAdapter
+ {
+     private static final Logger logger = LoggerFactory.getLogger(ConnectionLimitHandler.class);
+     private final ConcurrentMap<InetAddress, AtomicLong> connectionsPerClient = new
ConcurrentHashMap<>();
+     private final AtomicLong counter = new AtomicLong(0);
+ 
+     @Override
 -    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent event) throws Exception
++    public void channelActive(ChannelHandlerContext ctx) throws Exception
+     {
+         final long count = counter.incrementAndGet();
+         long limit = DatabaseDescriptor.getNativeTransportMaxConcurrentConnections();
+         // Setting the limit to -1 disables it.
+         if(limit < 0)
+         {
+             limit = Long.MAX_VALUE;
+         }
+         if (count > limit)
+         {
+             // The decrement will be done in channelClosed(...)
+             logger.warn("Exceeded maximum native connection limit of {} by using {} connections",
limit, count);
 -            ctx.getChannel().close();
++            ctx.close();
+         }
+         else
+         {
+             long perIpLimit = DatabaseDescriptor.getNativeTransportMaxConcurrentConnectionsPerIp();
+             if (perIpLimit > 0)
+             {
 -                InetAddress address = ((InetSocketAddress) ctx.getChannel().getRemoteAddress()).getAddress();
++                InetAddress address = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
+ 
+                 AtomicLong perIpCount = connectionsPerClient.get(address);
+                 if (perIpCount == null)
+                 {
+                     perIpCount = new AtomicLong(0);
+ 
+                     AtomicLong old = connectionsPerClient.putIfAbsent(address, perIpCount);
+                     if (old != null)
+                     {
+                         perIpCount = old;
+                     }
+                 }
+                 if (perIpCount.incrementAndGet() > perIpLimit)
+                 {
+                     // The decrement will be done in channelClosed(...)
+                     logger.warn("Exceeded maximum native connection limit per ip of {} by
using {} connections", perIpLimit, perIpCount);
 -                    ctx.getChannel().close();
++                    ctx.close();
+                     return;
+                 }
+             }
 -            super.channelOpen(ctx, event);
++            ctx.fireChannelActive();
+         }
+     }
+ 
+     @Override
 -    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent event) throws
Exception
++    public void channelInactive(ChannelHandlerContext ctx) throws Exception
+     {
+         counter.decrementAndGet();
 -        InetAddress address = ((InetSocketAddress) ctx.getChannel().getRemoteAddress()).getAddress();
++        InetAddress address = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress();
+ 
+         AtomicLong count = connectionsPerClient.get(address);
+         if (count != null)
+         {
+             if (count.decrementAndGet() <= 0)
+             {
+                 connectionsPerClient.remove(address);
+             }
+         }
 -        super.channelClosed(ctx, event);
++        ctx.fireChannelInactive();
+     }
+ }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/b1c2536c/src/java/org/apache/cassandra/transport/Server.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/transport/Server.java
index 60d3e70,30b8a9d..f396fd9
--- a/src/java/org/apache/cassandra/transport/Server.java
+++ b/src/java/org/apache/cassandra/transport/Server.java
@@@ -278,10 -249,18 +279,18 @@@ public class Server implements Cassandr
              this.server = server;
          }
  
 -        public ChannelPipeline getPipeline() throws Exception
 +        protected void initChannel(Channel channel) throws Exception
          {
 -            ChannelPipeline pipeline = Channels.pipeline();
 +            ChannelPipeline pipeline = channel.pipeline();
  
+             // Add the ConnectionLimitHandler to the pipeline if configured to do so.
+             if (DatabaseDescriptor.getNativeTransportMaxConcurrentConnections() > 0
+                     || DatabaseDescriptor.getNativeTransportMaxConcurrentConnectionsPerIp()
> 0)
+             {
+                 // Add as first to the pipeline so the limit is enforced as first action.
+                 pipeline.addFirst("connectionLimitHandler", connectionLimitHandler);
+             }
+ 
              //pipeline.addLast("debug", new LoggingHandler());
  
              pipeline.addLast("frameDecoder", new Frame.Decoder(server.connectionFactory));


Mime
View raw message