From 5ab5e1bbd19cd392901bc6222daaf0180c26cafb Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 30 Jun 2010 05:48:24 +0000 Subject: [PATCH] merge from 0.6 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@959197 13f79535-47bb-0310-9956-ffa450edef68 --- .rat-excludes | 2 + CHANGES.txt | 8 ++- build.xml | 2 +- debian/changelog | 6 ++ .../DebuggableThreadPoolExecutor.java | 4 +- .../concurrent/NamedThreadFactory.java | 12 +++- .../cassandra/db/CompactionManager.java | 4 +- .../cassandra/db/HintedHandOffManager.java | 10 ++- .../gms/GossipDigestAck2VerbHandler.java | 21 +++++++ .../gms/GossipDigestAckVerbHandler.java | 21 +++++++ .../gms/GossipDigestSynVerbHandler.java | 21 +++++++ .../cassandra/service/StorageProxy.java | 62 +++++++++++-------- .../cassandra/io/BloomFilterTrackerTest.java | 21 +++++++ 13 files changed, 161 insertions(+), 33 deletions(-) diff --git a/.rat-excludes b/.rat-excludes index d5c9a7b40b..437f61e21e 100644 --- a/.rat-excludes +++ b/.rat-excludes @@ -12,3 +12,5 @@ src/gen-java/** build/** lib/licenses/*.txt .settings/** +contrib/pig/example-script.pig +contrib/redhat/cassandra diff --git a/CHANGES.txt b/CHANGES.txt index 122a29b1cf..bd0f9d9d8b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -39,9 +39,13 @@ dev * allow multiple repair sessions per node (CASSANDRA-1190) +0.6.4 + * avoid queuing multiple hint deliveries for the same endpoint + (CASSANDRA-1229) + + 0.6.3 * retry to make streaming connections up to 8 times. (CASSANDRA-1019) - * fix potential for duplicate rows seen by Hadoop jobs (CASSANDRA-1042) * reject describe_ring() calls on invalid keyspaces (CASSANDRA-1111) * don't reject reads at CL.ALL (CASSANDRA-1152) * reject deletions to supercolumns in CFs containing only standard @@ -65,6 +69,8 @@ dev * remove opportunistic repairs, when two machines with overlapping replica responsibilities happen to finish major compactions of the same CF near the same time. repairs are now fully manual (CASSANDRA-1190) + * add ability to lower compaction priority (default is no change from 0.6.2) + (CASSANDRA-1181) 0.6.2 diff --git a/build.xml b/build.xml index aec5823879..d510f172d7 100644 --- a/build.xml +++ b/build.xml @@ -45,7 +45,7 @@ - + Fri, 25 Jun 2010 17:18:54 -0500 + cassandra (0.6.2) unstable; urgency=low * New stable point release. diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 087a93442a..11dd6c4452 100644 --- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java @@ -30,9 +30,9 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor { protected static Logger logger = LoggerFactory.getLogger(JMXEnabledThreadPoolExecutor.class); - public DebuggableThreadPoolExecutor(String threadPoolName) + public DebuggableThreadPoolExecutor(String threadPoolName, int priority) { - this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName)); + this(1, 1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue(), new NamedThreadFactory(threadPoolName, priority)); } public DebuggableThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory) diff --git a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java index c53cc48010..76f8223ed0 100644 --- a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java +++ b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java @@ -30,16 +30,26 @@ import java.util.concurrent.atomic.*; public class NamedThreadFactory implements ThreadFactory { protected final String id; + private final int priority; protected final AtomicInteger n = new AtomicInteger(1); public NamedThreadFactory(String id) { + this(id, Thread.NORM_PRIORITY); + } + + public NamedThreadFactory(String id, int priority) + { + this.id = id; + this.priority = priority; } public Thread newThread(Runnable runnable) { String name = id + ":" + n.getAndIncrement(); - return new Thread(runnable, name); + Thread thread = new Thread(runnable, name); + thread.setPriority(priority); + return thread; } } diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 1a432f2dd1..62fdc7eb08 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -619,7 +619,9 @@ public class CompactionManager implements CompactionManagerMBean public CompactionExecutor() { - super("CompactionExecutor"); + super("CompactionExecutor", System.getProperty("cassandra.compaction.priority") == null + ? Thread.NORM_PRIORITY + : Integer.parseInt(System.getProperty("cassandra.compaction.priority"))); } @Override diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 6686616860..2631bd2d6f 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -20,6 +20,7 @@ package org.apache.cassandra.db; import java.net.UnknownHostException; import java.util.Collection; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.TimeoutException; import java.util.concurrent.ExecutorService; import java.io.IOException; @@ -45,6 +46,7 @@ import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.utils.FBUtilities.UTF8; import org.apache.cassandra.utils.WrappedRunnable; +import org.cliffc.high_scale_lib.NonBlockingHashSet; /** @@ -84,6 +86,8 @@ public class HintedHandOffManager private static final int PAGE_SIZE = 10000; private static final String SEPARATOR = "-"; + private final NonBlockingHashSet queuedDeliveries = new NonBlockingHashSet(); + private final ExecutorService executor_ = new JMXEnabledThreadPoolExecutor("HINTED-HANDOFF-POOL"); private static boolean sendMessage(InetAddress endpoint, String tableName, String cfName, byte[] key) throws IOException @@ -170,9 +174,10 @@ public class HintedHandOffManager } - private static void deliverHintsToEndpoint(InetAddress endpoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException + private void deliverHintsToEndpoint(InetAddress endpoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException { logger_.info("Started hinted handoff for endpoint " + endpoint); + queuedDeliveries.remove(endpoint); // 1. Get the key of the endpoint we need to handoff // 2. For each column read the list of rows: subcolumns are KS + SEPARATOR + CF @@ -263,6 +268,9 @@ public class HintedHandOffManager */ public void deliverHints(final InetAddress to) { + if (!queuedDeliveries.add(to)) + return; + Runnable r = new WrappedRunnable() { public void runMayThrow() throws Exception diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java index 46473275a0..d3edae3a9a 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAck2VerbHandler.java @@ -1,4 +1,25 @@ package org.apache.cassandra.gms; +/* + * + * 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. + * + */ + import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java index 13ba8345bc..0b9e730346 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java @@ -1,4 +1,25 @@ package org.apache.cassandra.gms; +/* + * + * 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. + * + */ + import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java index 2b8668fcfe..8cc555684b 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java @@ -1,4 +1,25 @@ package org.apache.cassandra.gms; +/* + * + * 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. + * + */ + import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.net.IVerbHandler; diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index ee04de14f5..ebbde86058 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -501,22 +501,24 @@ public class StorageProxy implements StorageProxyMBean long startTime = System.nanoTime(); final String table = command.keyspace; - List>> ranges = getRestrictedRanges(command.range, command.keyspace); + List ranges = getRestrictedRanges(command.range); // now scan until we have enough results List rows = new ArrayList(command.max_keys); - for (Pair> pair : getRangeIterator(ranges, command.range.left)) + for (AbstractBounds range : getRangeIterator(ranges, command.range.left)) { - AbstractBounds range = pair.left; - List endpoints = pair.right; + List liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(command.keyspace, range.right); + DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints); + RangeSliceCommand c2 = new RangeSliceCommand(command.keyspace, command.column_family, command.super_column, command.predicate, range, command.max_keys); Message message = c2.getMessage(); // collect replies and resolve according to consistency level - RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, endpoints); + RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(table); QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level, table); - for (InetAddress endpoint : endpoints) + // TODO bail early if live endpoints can't satisfy requested consistency level + for (InetAddress endpoint : liveEndpoints) { MessagingService.instance.sendRR(message, endpoint, handler); if (logger.isDebugEnabled()) @@ -621,30 +623,30 @@ public class StorageProxy implements StorageProxyMBean /** * returns an iterator that will return ranges in ring order, starting with the one that contains the start token */ - private static Iterable>> getRangeIterator(final List>> ranges, Token start) + private static Iterable getRangeIterator(final List ranges, Token start) { // find the one to start with int i; for (i = 0; i < ranges.size(); i++) { - AbstractBounds range = ranges.get(i).left; + AbstractBounds range = ranges.get(i); if (range.contains(start) || range.left.equals(start)) break; } - AbstractBounds range = ranges.get(i).left; + AbstractBounds range = ranges.get(i); assert range.contains(start) || range.left.equals(start); // make sure the loop didn't just end b/c ranges were exhausted // return an iterable that starts w/ the correct range and iterates the rest in ring order final int begin = i; - return new Iterable>>() + return new Iterable() { - public Iterator>> iterator() + public Iterator iterator() { - return new AbstractIterator>>() + return new AbstractIterator() { int n = 0; - protected Pair> computeNext() + protected AbstractBounds computeNext() { if (n == ranges.size()) return endOfData(); @@ -669,30 +671,38 @@ public class StorageProxy implements StorageProxyMBean * D, but we don't want any other results from it until after the (D, T] range. Unwrapping so that * the ranges we consider are (D, T], (T, MIN], (MIN, D] fixes this. */ - private static List>> getRestrictedRanges(AbstractBounds queryRange, String keyspace) - throws UnavailableException + private static List getRestrictedRanges(AbstractBounds queryRange) { TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata(); - Iterator iter = TokenMetadata.ringIterator(tokenMetadata.sortedTokens(), queryRange.left); - List>> ranges = new ArrayList>>(); - while (iter.hasNext()) - { - Token nodeToken = iter.next(); - Range nodeRange = new Range(tokenMetadata.getPredecessor(nodeToken), nodeToken); - List endpoints = StorageService.instance.getLiveNaturalEndpoints(keyspace, nodeToken); - DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints); - Set restrictedRanges = queryRange.restrictTo(nodeRange); - for (AbstractBounds range : restrictedRanges) + List ranges = new ArrayList(); + // for each node, compute its intersection with the query range, and add its unwrapped components to our list + for (Token nodeToken : tokenMetadata.sortedTokens()) + { + Range nodeRange = new Range(tokenMetadata.getPredecessor(nodeToken), nodeToken); + for (AbstractBounds range : queryRange.restrictTo(nodeRange)) { for (AbstractBounds unwrapped : range.unwrap()) { if (logger.isDebugEnabled()) logger.debug("Adding to restricted ranges " + unwrapped + " for " + nodeRange); - ranges.add(new Pair>(unwrapped, endpoints)); + ranges.add(unwrapped); } } } + + // re-sort ranges in ring order, post-unwrapping + Comparator comparator = new Comparator() + { + public int compare(AbstractBounds o1, AbstractBounds o2) + { + // no restricted ranges will overlap so we don't need to worry about inclusive vs exclusive left, + // just sort by raw token position. + return o1.left.compareTo(o2.left); + } + }; + Collections.sort(ranges, comparator); + return ranges; } diff --git a/test/unit/org/apache/cassandra/io/BloomFilterTrackerTest.java b/test/unit/org/apache/cassandra/io/BloomFilterTrackerTest.java index b10ec10998..29a1588bdb 100644 --- a/test/unit/org/apache/cassandra/io/BloomFilterTrackerTest.java +++ b/test/unit/org/apache/cassandra/io/BloomFilterTrackerTest.java @@ -1,4 +1,25 @@ package org.apache.cassandra.io; +/* + * + * 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. + * + */ + import org.junit.Test;