diff --git a/CHANGES.txt b/CHANGES.txt index 9a76134a3d..ed2224ea29 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -15,6 +15,7 @@ * add pluggable SeedProvider (CASSANDRA-1669) * Fix clustertool to not throw exception when calling get_endpoints (CASSANDRA-2437) * upgrade to thrift 0.6 (CASSANDRA-2412) + * repair works on a token range instead of full ring (CASSANDRA-2324) 0.7.5 diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index c176a6b6d0..3e361592f9 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1736,6 +1736,18 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return Iterables.concat(samples); } + public Iterable keySamples(Range range) + { + Collection sstables = getSSTables(); + Iterable[] samples = new Iterable[sstables.size()]; + int i = 0; + for (SSTableReader sstable: sstables) + { + samples[i++] = sstable.getKeySamples(range); + } + return Iterables.concat(samples); + } + /** * For testing. no effort is made to clear historical memtables, nor for * thread safety diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 1ab24b9648..271bba723c 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -34,6 +34,7 @@ import javax.management.MBeanServer; import javax.management.ObjectName; import org.apache.commons.collections.PredicateUtils; +import org.apache.commons.collections.iterators.CollatingIterator; import org.apache.commons.collections.iterators.FilterIterator; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; @@ -51,6 +52,7 @@ import org.apache.cassandra.service.AntiEntropyService; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.OperationType; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.NodeId; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.WrappedRunnable; @@ -825,7 +827,7 @@ public class CompactionManager implements CompactionManagerMBean throw new AssertionError(e); } - CompactionIterator ci = new ValidationCompactionIterator(cfs); + CompactionIterator ci = new ValidationCompactionIterator(cfs, validator.request.range); executor.beginCompaction(cfs.columnFamily, ci); try { @@ -984,9 +986,19 @@ public class CompactionManager implements CompactionManagerMBean private static class ValidationCompactionIterator extends CompactionIterator { - public ValidationCompactionIterator(ColumnFamilyStore cfs) throws IOException + public ValidationCompactionIterator(ColumnFamilyStore cfs, Range range) throws IOException { - super(cfs.getSSTables(), new CompactionController(cfs, cfs.getSSTables(), true, getDefaultGcBefore(cfs), false)); + super(getCollatingIterator(cfs.getSSTables(), range), new CompactionController(cfs, cfs.getSSTables(), true, getDefaultGcBefore(cfs), false)); + } + + protected static CollatingIterator getCollatingIterator(Iterable sstables, Range range) throws IOException + { + CollatingIterator iter = FBUtilities.getCollatingIterator(); + for (SSTableReader sstable : sstables) + { + iter.addIterator(sstable.getDirectScanner(FILE_BUFFER_SIZE, range)); + } + return iter; } @Override diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java new file mode 100644 index 0000000000..69b4e61294 --- /dev/null +++ b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java @@ -0,0 +1,100 @@ +/** + * 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.io.sstable; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; + +import org.apache.cassandra.db.columniterator.IColumnIterator; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.utils.Pair; + +/** + * A SSTableScanner that only reads key in a given range (for validation compaction). + */ +public class SSTableBoundedScanner extends SSTableScanner +{ + private final Iterator> rangeIterator; + private Pair currentRange; + + SSTableBoundedScanner(SSTableReader sstable, int bufferSize, boolean skipCache, Range range) + { + super(sstable, bufferSize, skipCache); + this.rangeIterator = sstable.getPositionsForRanges(Collections.singletonList(range)).iterator(); + if (rangeIterator.hasNext()) + { + currentRange = rangeIterator.next(); + try + { + file.seek(currentRange.left); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + else + { + exhausted = true; + } + } + + @Override + public boolean hasNext() + { + if (iterator == null) + iterator = exhausted ? Arrays.asList(new IColumnIterator[0]).iterator() : new BoundedKeyScanningIterator(); + return iterator.hasNext(); + } + + @Override + public IColumnIterator next() + { + if (iterator == null) + iterator = exhausted ? Arrays.asList(new IColumnIterator[0]).iterator() : new BoundedKeyScanningIterator(); + return iterator.next(); + } + + protected class BoundedKeyScanningIterator extends KeyScanningIterator + { + @Override + public boolean hasNext() + { + if (!super.hasNext()) + return false; + + if (finishedAt < currentRange.right) + return true; + + if (rangeIterator.hasNext()) + { + currentRange = rangeIterator.next(); + finishedAt = currentRange.left; // next() will seek for us + return true; + } + else + { + return false; + } + } + } +} diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 29b52d5fc8..dd2c4df9f1 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -381,6 +381,97 @@ public class SSTableReader extends SSTable implements Comparable }); } + private static List> getSampleIndexesForRanges(List samples, Collection ranges) + { + // use the index to determine a minimal section for each range + List> positions = new ArrayList>(); + if (samples.isEmpty()) + return positions; + + for (AbstractBounds range : AbstractBounds.normalize(ranges)) + { + DecoratedKey leftKey = new DecoratedKey(range.left, null); + DecoratedKey rightKey = new DecoratedKey(range.right, null); + + int left = Collections.binarySearch(samples, new IndexSummary.KeyPosition(leftKey, -1)); + if (left < 0) + left = (left + 1) * -1; + else + // left range are start exclusive + left = left + 1; + if (left == samples.size()) + // left is past the end of the sampling + continue; + + int right = Range.isWrapAround(range.left, range.right) + ? samples.size() - 1 + : Collections.binarySearch(samples, new IndexSummary.KeyPosition(rightKey, -1)); + if (right < 0) + { + // range are end inclusive so we use the previous index from what binarySearch give us + // since that will be the last index we will return + right = (right + 1) * -1; + if (right > 0) + right--; + } + + if (left >= right) + // empty range + continue; + positions.add(new Pair(Integer.valueOf(left), Integer.valueOf(right))); + } + return positions; + } + + public Iterable getKeySamples(final Range range) + { + final List samples = indexSummary.getIndexPositions(); + + final List> indexRanges = getSampleIndexesForRanges(samples, Collections.singletonList(range)); + + if (indexRanges.isEmpty()) + return Collections.emptyList(); + + return new Iterable() + { + public Iterator iterator() + { + return new Iterator() + { + private Iterator> rangeIter = indexRanges.iterator(); + private Pair current; + private int idx; + + public boolean hasNext() + { + if (current == null || idx > current.right) + { + if (rangeIter.hasNext()) + { + current = rangeIter.next(); + idx = current.left; + return true; + } + return false; + } + + return true; + } + + public DecoratedKey next() + { + return samples.get(idx++).key; + } + + public void remove() + { + throw new UnsupportedOperationException(); + } + }; + } + }; + } + /** * Determine the minimal set of sections that can be extracted from this SSTable to cover the given ranges. * @return A sorted list of (offset,end) pairs that cover the given ranges in the datafile for this SSTable. @@ -550,6 +641,17 @@ public class SSTableReader extends SSTable implements Comparable return new SSTableScanner(this, bufferSize, true); } + /** + * Direct I/O SSTableScanner over a defined range of tokens. + * @param bufferSize Buffer size in bytes for this Scanner. + * @param range the range of keys to cover + * @return A Scanner for seeking over the rows of the SSTable. + */ + public SSTableScanner getDirectScanner(int bufferSize, Range range) + { + return new SSTableBoundedScanner(this, bufferSize, true, range); + } + public FileDataInput getFileDataInput(DecoratedKey decoratedKey, int bufferSize) { long position = getPosition(decoratedKey, Operator.EQ); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java index 6cf526dd06..93e07e26c7 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java @@ -40,11 +40,11 @@ public class SSTableScanner implements Iterator, Closeable { private static Logger logger = LoggerFactory.getLogger(SSTableScanner.class); - private final BufferedRandomAccessFile file; + protected final BufferedRandomAccessFile file; public final SSTableReader sstable; private IColumnIterator row; - private boolean exhausted = false; - private Iterator iterator; + protected boolean exhausted = false; + protected Iterator iterator; private QueryFilter filter; /** @@ -141,9 +141,9 @@ public class SSTableScanner implements Iterator, Closeable throw new UnsupportedOperationException(); } - private class KeyScanningIterator implements Iterator + protected class KeyScanningIterator implements Iterator { - private long finishedAt; + protected long finishedAt; public boolean hasNext() { diff --git a/src/java/org/apache/cassandra/service/AntiEntropyService.java b/src/java/org/apache/cassandra/service/AntiEntropyService.java index c23058d9da..7a4d043529 100644 --- a/src/java/org/apache/cassandra/service/AntiEntropyService.java +++ b/src/java/org/apache/cassandra/service/AntiEntropyService.java @@ -38,6 +38,7 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.CompactionManager; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Table; +import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.AbstractCompactedRow; @@ -121,9 +122,9 @@ public class AntiEntropyService * Requests repairs for the given table and column families, and blocks until all repairs have been completed. * TODO: Should add retries: if nodes go offline before they respond to the requests, this could block forever. */ - public RepairSession getRepairSession(String tablename, String... cfnames) + public RepairSession getRepairSession(Range range, String tablename, String... cfnames) { - return new RepairSession(tablename, cfnames); + return new RepairSession(range, tablename, cfnames); } RepairSession getArtificialRepairSession(TreeRequest req, String tablename, String... cfnames) @@ -158,17 +159,24 @@ public class AntiEntropyService /** * Return all of the neighbors with whom we share data. */ - static Set getNeighbors(String table) + static Set getNeighbors(String table, Range range) { StorageService ss = StorageService.instance; - Set neighbors = new HashSet(); Map> replicaSets = ss.getRangeToAddressMap(table); - for (Range range : ss.getLocalRanges(table)) - { - // for every range stored locally (replica or original) collect neighbors storing copies - neighbors.addAll(replicaSets.get(range)); - } + if (!replicaSets.containsKey(range)) + return Collections.emptySet(); + Set neighbors = new HashSet(replicaSets.get(range)); neighbors.remove(FBUtilities.getLocalAddress()); + // Excluding all node with version <= 0.7 since they don't know how to + // create a correct merkle tree (they build it over the full range) + for (InetAddress endpoint : neighbors) + { + if (Gossiper.instance.getVersion(endpoint) <= MessagingService.VERSION_07) + { + logger.info("Excluding " + endpoint + " from repair because it is on version 0.7 or sooner. You should consider updating this node before running repair again."); + neighbors.remove(endpoint); + } + } return neighbors; } @@ -186,9 +194,9 @@ public class AntiEntropyService if (LOCAL.equals(request.endpoint)) { // we're registering a local tree: rendezvous with remote requests for the session - for (InetAddress neighbor : getNeighbors(request.cf.left)) + for (InetAddress neighbor : getNeighbors(request.cf.left, request.range)) { - TreeRequest remotereq = new TreeRequest(request.sessionid, neighbor, request.cf); + TreeRequest remotereq = new TreeRequest(request.sessionid, neighbor, request.range, request.cf); TreePair waiting = ctrees.remove(remotereq); if (waiting != null && waiting.right != null) { @@ -230,9 +238,9 @@ public class AntiEntropyService /** * Requests a tree from the given node, and returns the request that was sent. */ - TreeRequest request(String sessionid, InetAddress remote, String ksname, String cfname) + TreeRequest request(String sessionid, InetAddress remote, Range range, String ksname, String cfname) { - TreeRequest request = new TreeRequest(sessionid, remote, new CFPair(ksname, cfname)); + TreeRequest request = new TreeRequest(sessionid, remote, range, new CFPair(ksname, cfname)); MessagingService.instance().sendOneWay(TreeRequestVerbHandler.makeVerb(request, Gossiper.instance.getVersion(remote)), remote); return request; } @@ -274,7 +282,6 @@ public class AntiEntropyService // the minimum token sorts first, but falls into the last range private transient List minrows; // null when all rows with the min token have been consumed - private transient Token mintoken; private transient long validated; private transient MerkleTree.TreeRange range; private transient MerkleTree.TreeRangeIterator ranges; @@ -286,25 +293,30 @@ public class AntiEntropyService this(request, // TODO: memory usage (maxsize) should either be tunable per // CF, globally, or as shared for all CFs in a cluster - new MerkleTree(DatabaseDescriptor.getPartitioner(), MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15))); + new MerkleTree(DatabaseDescriptor.getPartitioner(), request.range, MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15))); } Validator(TreeRequest request, MerkleTree tree) { this.request = request; this.tree = tree; + // Reestablishing the range because we don't serialize it (for bad + // reason - see MerkleTree for details) + this.tree.fullRange = this.request.range; minrows = new ArrayList(); - mintoken = null; validated = 0; range = null; ranges = null; } - + public void prepare(ColumnFamilyStore cfs) { List keys = new ArrayList(); - for (DecoratedKey sample : cfs.allKeySamples()) + for (DecoratedKey sample : cfs.keySamples(request.range)) + { + assert request.range.contains(sample.token); keys.add(sample); + } if (keys.isEmpty()) { @@ -324,8 +336,7 @@ public class AntiEntropyService } } logger.debug("Prepared AEService tree of size " + tree.size() + " for " + request); - mintoken = tree.partitioner().getMinimumToken(); - ranges = tree.invalids(new Range(mintoken, mintoken)); + ranges = tree.invalids(); } /** @@ -348,19 +359,7 @@ public class AntiEntropyService */ public void add(AbstractCompactedRow row) { - if (mintoken != null) - { - assert ranges != null : "Validator was not prepared()"; - - // check for the minimum token special case - if (row.key.token.compareTo(mintoken) == 0) - { - // and store it to be appended when we complete - minrows.add(rowHash(row)); - return; - } - mintoken = null; - } + assert request.range.contains(row.key.token) : row.key.token + " is not contained in " + request.range; if (range == null) range = ranges.next(); @@ -400,10 +399,6 @@ public class AntiEntropyService range = ranges.next(); range.addHash(EMPTY_ROW); } - // add rows with the minimum token to the final range - if (!minrows.isEmpty()) - for (MerkleTree.RowHash minrow : minrows) - range.addHash(minrow); StageManager.getStage(Stage.ANTI_ENTROPY).execute(this); logger.debug("Validated " + validated + " rows into AEService tree for " + request); @@ -429,14 +424,14 @@ public class AntiEntropyService public final TreeRequest request; public final MerkleTree ltree; public final MerkleTree rtree; - public final List differences; + public List differences; public Differencer(TreeRequest request, MerkleTree ltree, MerkleTree rtree) { this.request = request; this.ltree = ltree; this.rtree = rtree; - differences = new ArrayList(); + this.differences = new ArrayList(); } /** @@ -445,7 +440,6 @@ public class AntiEntropyService public void run() { InetAddress local = FBUtilities.getLocalAddress(); - StorageService ss = StorageService.instance; // restore partitioners (in case we were serialized) if (ltree.partitioner() == null) @@ -453,17 +447,11 @@ public class AntiEntropyService if (rtree.partitioner() == null) rtree.partitioner(StorageService.getPartitioner()); - // determine the ranges where responsibility overlaps - Set interesting = new HashSet(ss.getRangesForEndpoint(request.cf.left, local)); - interesting.retainAll(ss.getRangesForEndpoint(request.cf.left, request.endpoint)); + // compare trees, and collect differences + differences.addAll(MerkleTree.difference(ltree, rtree)); - // compare trees, and collect interesting differences - for (MerkleTree.TreeRange diff : MerkleTree.difference(ltree, rtree)) - for (Range localrange: interesting) - differences.addAll(diff.intersectionWith(localrange)); - // choose a repair method based on the significance of the difference - String format = "Endpoints " + local + " and " + request.endpoint + " %s for " + request.cf; + String format = "Endpoints " + local + " and " + request.endpoint + " %s for " + request.cf + " on " + request.range; if (differences.isEmpty()) { logger.info(String.format(format, "are consistent")); @@ -493,14 +481,13 @@ public class AntiEntropyService ColumnFamilyStore cfstore = Table.open(request.cf.left).getColumnFamilyStore(request.cf.right); try { - List ranges = new ArrayList(differences); Collection sstables = cfstore.getSSTables(); Callback callback = new Callback(); // send ranges to the remote node StreamOutSession outsession = StreamOutSession.create(request.cf.left, request.endpoint, callback); - StreamOut.transferSSTables(outsession, sstables, ranges, OperationType.AES); + StreamOut.transferSSTables(outsession, sstables, differences, OperationType.AES); // request ranges from the remote node - StreamIn.requestRanges(request.endpoint, request.cf.left, ranges, callback, OperationType.AES); + StreamIn.requestRanges(request.endpoint, request.cf.left, differences, callback, OperationType.AES); } catch(Exception e) { @@ -563,13 +550,22 @@ public class AntiEntropyService CompactEndpointSerializationHelper.serialize(request.endpoint, dos); dos.writeUTF(request.cf.left); dos.writeUTF(request.cf.right); + if (version > MessagingService.VERSION_07) + AbstractBounds.serializer().serialize(request.range, dos); } public TreeRequest deserialize(DataInputStream dis, int version) throws IOException { - return new TreeRequest(dis.readUTF(), - CompactEndpointSerializationHelper.deserialize(dis), - new CFPair(dis.readUTF(), dis.readUTF())); + String sessId = dis.readUTF(); + InetAddress endpoint = CompactEndpointSerializationHelper.deserialize(dis); + CFPair cfpair = new CFPair(dis.readUTF(), dis.readUTF()); + Range range; + if (version > MessagingService.VERSION_07) + range = (Range) AbstractBounds.serializer().deserialize(dis); + else + range = new Range(StorageService.getPartitioner().getMinimumToken(), StorageService.getPartitioner().getMinimumToken()); + + return new TreeRequest(sessId, endpoint, range, cfpair); } /** @@ -583,7 +579,7 @@ public class AntiEntropyService try { TreeRequest remotereq = this.deserialize(buffer, message.getVersion()); - TreeRequest request = new TreeRequest(remotereq.sessionid, message.getFrom(), remotereq.cf); + TreeRequest request = new TreeRequest(remotereq.sessionid, message.getFrom(), remotereq.range, remotereq.cf); // trigger readonly-compaction ColumnFamilyStore store = Table.open(request.cf.left).getColumnFamilyStore(request.cf.right); @@ -652,7 +648,7 @@ public class AntiEntropyService { // deserialize the remote tree, and register it Validator response = this.deserialize(buffer, message.getVersion()); - TreeRequest request = new TreeRequest(response.request.sessionid, message.getFrom(), response.request.cf); + TreeRequest request = new TreeRequest(response.request.sessionid, message.getFrom(), response.request.range, response.request.cf); AntiEntropyService.instance.rendezvous(request, response.tree); } catch (IOException e) @@ -686,25 +682,27 @@ public class AntiEntropyService } /** - * A triple of table, cf and address that represents a location we have an outstanding TreeRequest for. + * A tuple of table, cf, address and range that represents a location we have an outstanding TreeRequest for. */ public static class TreeRequest { public final String sessionid; public final InetAddress endpoint; + public final Range range; public final CFPair cf; - public TreeRequest(String sessionid, InetAddress endpoint, CFPair cf) + public TreeRequest(String sessionid, InetAddress endpoint, Range range, CFPair cf) { this.sessionid = sessionid; this.endpoint = endpoint; this.cf = cf; + this.range = range; } @Override public final int hashCode() { - return Objects.hashCode(sessionid, endpoint, cf); + return Objects.hashCode(sessionid, endpoint, cf, range); } @Override @@ -714,19 +712,19 @@ public class AntiEntropyService return false; TreeRequest that = (TreeRequest)o; // handles nulls properly - return Objects.equal(sessionid, that.sessionid) && Objects.equal(endpoint, that.endpoint) && Objects.equal(cf, that.cf); + return Objects.equal(sessionid, that.sessionid) && Objects.equal(endpoint, that.endpoint) && Objects.equal(cf, that.cf) && Objects.equal(range, that.range); } @Override public String toString() { - return "#"; + return "#"; } } /** - * Triggers repairs with all neighbors for the given table and cfs. Typical lifecycle is: start() then join(). - * Executed in client threads. + * Triggers repairs with all neighbors for the given table, cfs and range. + * Typical lifecycle is: start() then join(). Executed in client threads. */ class RepairSession extends Thread { @@ -734,10 +732,12 @@ public class AntiEntropyService private final String[] cfnames; private final SimpleCondition requestsMade; private final ConcurrentHashMap requests; + private final Range range; public RepairSession(TreeRequest req, String tablename, String... cfnames) { super(req.sessionid); + this.range = req.range; this.tablename = tablename; this.cfnames = cfnames; requestsMade = new SimpleCondition(); @@ -747,11 +747,12 @@ public class AntiEntropyService AntiEntropyService.instance.sessions.put(getName(), callback); } - public RepairSession(String tablename, String... cfnames) + public RepairSession(Range range, String tablename, String... cfnames) { super("manual-repair-" + UUID.randomUUID()); this.tablename = tablename; this.cfnames = cfnames; + this.range = range; this.requestsMade = new SimpleCondition(); this.requests = new ConcurrentHashMap(); } @@ -767,10 +768,11 @@ public class AntiEntropyService @Override public void run() { - Set endpoints = AntiEntropyService.getNeighbors(tablename); + Set endpoints = AntiEntropyService.getNeighbors(tablename, range); if (endpoints.isEmpty()) { - logger.info("No neighbors to repair with: " + getName() + " completed."); + requestsMade.signalAll(); + logger.info("No neighbors to repair with for " + tablename + " on " + range + ": " + getName() + " completed."); return; } @@ -784,9 +786,9 @@ public class AntiEntropyService { // send requests to remote nodes and record them for (InetAddress endpoint : endpoints) - requests.put(AntiEntropyService.instance.request(getName(), endpoint, tablename, cfname), this); + requests.put(AntiEntropyService.instance.request(getName(), endpoint, range, tablename, cfname), this); // send but don't record an outstanding request to the local node - AntiEntropyService.instance.request(getName(), FBUtilities.getLocalAddress(), tablename, cfname); + AntiEntropyService.instance.request(getName(), FBUtilities.getLocalAddress(), range, tablename, cfname); } logger.info("Waiting for repair requests: " + requests.keySet()); requestsMade.signalAll(); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 3bb0c40bf0..e77284bb71 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -1440,6 +1440,34 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe * @throws IOException */ public void forceTableRepair(final String tableName, final String... columnFamilies) throws IOException + { + List sessions = new ArrayList(); + for (Range range : getLocalRanges(tableName)) + { + sessions.add(forceTableRepair(range, tableName, columnFamilies)); + } + + boolean failedSession = false; + + // block until all repair sessions have completed + for (AntiEntropyService.RepairSession sess : sessions) + { + try + { + sess.join(); + } + catch (InterruptedException e) + { + logger_.error("Repair session " + sess + " failed.", e); + failedSession = true; + } + } + + if (failedSession) + throw new IOException("Some Repair session(s) failed."); + } + + public AntiEntropyService.RepairSession forceTableRepair(final Range range, final String tableName, final String... columnFamilies) throws IOException { String[] families; if (columnFamilies.length == 0) @@ -1454,18 +1482,9 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe { families = columnFamilies; } - AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getRepairSession(tableName, families); - - try - { - sess.start(); - // block until the repair has completed - sess.join(); - } - catch (InterruptedException e) - { - throw new IOException("Repair session " + sess + " failed.", e); - } + AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getRepairSession(range, tableName, families); + sess.start(); + return sess; } /* End of MBean interface methods */ diff --git a/src/java/org/apache/cassandra/utils/MerkleTree.java b/src/java/org/apache/cassandra/utils/MerkleTree.java index 49ce8c78be..716c87a9e2 100644 --- a/src/java/org/apache/cassandra/utils/MerkleTree.java +++ b/src/java/org/apache/cassandra/utils/MerkleTree.java @@ -68,6 +68,16 @@ public class MerkleTree implements Serializable public final byte hashdepth; + /** + * The top level range that this MerkleTree covers. + * In a perfect world, this should be final and *not* transient. However + * this would break serialization with version >e; 0.7 because it uses + * java serialization. We are moreover always shipping the fullRange will + * the request so we can add it back post-deserialization (as for the + * partitioner). + */ + public transient Range fullRange; + private transient IPartitioner partitioner; private long maxsize; @@ -111,7 +121,7 @@ public class MerkleTree implements Serializable byte hashdepth = dis.readByte(); long maxsize = dis.readLong(); long size = dis.readLong(); - MerkleTree mt = new MerkleTree(null, hashdepth, maxsize); + MerkleTree mt = new MerkleTree(null, null, hashdepth, maxsize); mt.size = size; mt.root = Hashable.serializer.deserialize(dis, version); return mt; @@ -121,13 +131,15 @@ public class MerkleTree implements Serializable /** * @param partitioner The partitioner in use. + * @param range the range this tree covers * @param hashdepth The maximum depth of the tree. 100/(2^depth) is the % * of the key space covered by each subrange of a fully populated tree. * @param maxsize The maximum number of subranges in the tree. */ - public MerkleTree(IPartitioner partitioner, byte hashdepth, long maxsize) + public MerkleTree(IPartitioner partitioner, Range range, byte hashdepth, long maxsize) { assert hashdepth < Byte.MAX_VALUE; + this.fullRange = range; this.partitioner = partitioner; this.hashdepth = hashdepth; this.maxsize = maxsize; @@ -136,6 +148,7 @@ public class MerkleTree implements Serializable root = new Leaf(null); } + static byte inc(byte in) { assert in < Byte.MAX_VALUE; @@ -154,8 +167,7 @@ public class MerkleTree implements Serializable byte sizedepth = (byte)(Math.log10(maxsize) / Math.log10(2)); byte depth = (byte)Math.min(sizedepth, hashdepth); - Token mintoken = partitioner.getMinimumToken(); - root = initHelper(mintoken, mintoken, (byte)0, depth); + root = initHelper(fullRange.left, fullRange.right, (byte)0, depth); size = (long)Math.pow(2, depth); } @@ -165,8 +177,9 @@ public class MerkleTree implements Serializable // we've reached the leaves return new Leaf(); Token midpoint = partitioner.midpoint(left, right); - Hashable lchild = initHelper(left, midpoint, inc(depth), max); - Hashable rchild = initHelper(midpoint, right, inc(depth), max); + + Hashable lchild = midpoint.equals(left) ? new Leaf() : initHelper(left, midpoint, inc(depth), max); + Hashable rchild = midpoint.equals(right) ? new Leaf() : initHelper(midpoint, right, inc(depth), max); return new Inner(midpoint, lchild, rchild); } @@ -214,9 +227,11 @@ public class MerkleTree implements Serializable */ public static List difference(MerkleTree ltree, MerkleTree rtree) { + if (!ltree.fullRange.equals(rtree.fullRange)) + throw new IllegalArgumentException("Difference only make sense on tree covering the same range (but " + ltree.fullRange + " != " + rtree.fullRange + ")"); + List diff = new ArrayList(); - Token mintoken = ltree.partitioner.getMinimumToken(); - TreeRange active = new TreeRange(null, mintoken, mintoken, (byte)0, null); + TreeRange active = new TreeRange(null, ltree.fullRange.left, ltree.fullRange.right, (byte)0, null); byte[] lhash = ltree.hash(active); byte[] rhash = rtree.hash(active); @@ -289,10 +304,9 @@ public class MerkleTree implements Serializable * For testing purposes. * Gets the smallest range containing the token. */ - TreeRange get(Token t) + public TreeRange get(Token t) { - Token mintoken = partitioner.getMinimumToken(); - return getHelper(root, mintoken, mintoken, (byte)0, t); + return getHelper(root, fullRange.left, fullRange.right, (byte)0, t); } TreeRange getHelper(Hashable hashable, Token pleft, Token pright, byte depth, Token t) @@ -314,10 +328,11 @@ public class MerkleTree implements Serializable /** * Invalidates the ranges containing the given token. + * Useful for testing. */ public void invalidate(Token t) { - invalidateHelper(root, partitioner.getMinimumToken(), t); + invalidateHelper(root, fullRange.left, t); } private void invalidateHelper(Hashable hashable, Token pleft, Token t) @@ -326,7 +341,7 @@ public class MerkleTree implements Serializable if (hashable instanceof Leaf) return; // else: node. - + Inner node = (Inner)hashable; if (Range.contains(pleft, node.token, t)) // left child contains token @@ -348,10 +363,9 @@ public class MerkleTree implements Serializable */ public byte[] hash(Range range) { - Token mintoken = partitioner.getMinimumToken(); try { - return hashHelper(root, new Range(mintoken, mintoken), range); + return hashHelper(root, new Range(fullRange.left, fullRange.right), range); } catch (StopRecursion e) { @@ -413,10 +427,9 @@ public class MerkleTree implements Serializable if (!(size < maxsize)) return false; - Token mintoken = partitioner.getMinimumToken(); try { - root = splitHelper(root, mintoken, mintoken, (byte)0, t); + root = splitHelper(root, fullRange.left, fullRange.right, (byte)0, t); } catch (StopRecursion.TooDeep e) { @@ -441,6 +454,13 @@ public class MerkleTree implements Serializable // recurse on the matching child Inner node = (Inner)hashable; + + // FIXME: we are not really 'TooDeep', however we cannot say that the + // split was successfull otherwise we could have a chance of infinite + // loop given how we split. + if (t.equals(node.token) || t.equals(pright)) + throw new StopRecursion.TooDeep(); + if (Range.contains(pleft, node.token, t)) // left child contains token node.lchild(splitHelper(node.lchild, pleft, node.token, inc(depth), t)); @@ -450,52 +470,15 @@ public class MerkleTree implements Serializable return node; } - /** - * Compacts the smallest subranges evenly split by the given token into a - * single range. - * - * Asserts that the given Token falls between two compactable subranges. - */ - public void compact(Token t) - { - root = compactHelper(root, t); - } - - private Hashable compactHelper(Hashable hashable, Token t) - { - // we reached a Leaf without finding an Inner to compact - assert !(hashable instanceof Leaf); - - Inner node = (Inner)hashable; - int comp = t.compareTo(node.token); - if (comp == 0) - { - // this is the node to compact - assert node.lchild() instanceof Leaf && node.rchild() instanceof Leaf : - "Can only compact a subrange evenly split by the given token!"; - - // hash our children together into a new value to replace ourself - size--; - return new Leaf(node.lchild().hash(), node.rchild().hash()); - } - else if (comp < 0) - // recurse to the left - node.lchild(compactHelper(node.lchild(), t)); - else - // recurse to the right - node.rchild(compactHelper(node.rchild(), t)); - return node; - } - /** * Returns a lazy iterator of invalid TreeRanges that need to be filled * in order to make the given Range valid. * * @param range The range to find invalid subranges for. */ - public TreeRangeIterator invalids(Range range) + public TreeRangeIterator invalids() { - return new TreeRangeIterator(this, range); + return new TreeRangeIterator(this); } @Override @@ -569,25 +552,25 @@ public class MerkleTree implements Serializable } /** - * Performs a depth-first, inorder traversal of invalid nodes under the given root - * and intersecting the given range. + * Returns the leaf (range) of a given tree in increasing order. + * If the full range covered by the tree don't wrap, then it will return the + * ranges in increasing order. + * If the full range wrap, the first *and* last range returned by the + * iterator will be the wrapping range. It is the only case where the same + * leaf will be returned twice. */ public static class TreeRangeIterator extends AbstractIterator implements Iterable, PeekingIterator { // stack of ranges to visit private final ArrayDeque tovisit; // interesting range - private final Range range; private final MerkleTree tree; - - TreeRangeIterator(MerkleTree tree, Range range) - { - Token mintoken = tree.partitioner().getMinimumToken(); - tovisit = new ArrayDeque(); - tovisit.add(new TreeRange(tree, mintoken, mintoken, (byte)0, tree.root)); + TreeRangeIterator(MerkleTree tree) + { + tovisit = new ArrayDeque(); + tovisit.add(new TreeRange(tree, tree.fullRange.left, tree.fullRange.right, (byte)0, tree.root)); this.tree = tree; - this.range = range; } /** @@ -601,23 +584,31 @@ public class MerkleTree implements Serializable { TreeRange active = tovisit.pop(); - if (active.hashable.hash() != null) - // skip valid ranges - continue; - if (active.hashable instanceof Leaf) + { // found a leaf invalid range + if (active.isWrapAround() && !tovisit.isEmpty()) + // put to be taken again last + tovisit.addLast(active); return active; + } Inner node = (Inner)active.hashable; - // push intersecting children onto the stack TreeRange left = new TreeRange(tree, active.left, node.token, inc(active.depth), node.lchild); TreeRange right = new TreeRange(tree, node.token, active.right, inc(active.depth), node.rchild); - if (right.intersects(range)) - tovisit.push(right); - if (left.intersects(range)) - tovisit.push(left); - + + if (right.isWrapAround()) + { + // whatever is on the left is 'after' everything we have seen so far (it has greater tokens) + tovisit.addLast(left); + tovisit.addFirst(right); + } + else + { + // do left first then right + tovisit.addFirst(right); + tovisit.addFirst(left); + } } return endOfData(); } diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java deleted file mode 100644 index 9770e37499..0000000000 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ /dev/null @@ -1,258 +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.service; - -import java.net.InetAddress; -import java.util.*; -import java.util.concurrent.Callable; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.ThreadPoolExecutor; - -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; - -import org.apache.cassandra.CleanupHelper; -import org.apache.cassandra.Util; -import org.apache.cassandra.concurrent.Stage; -import org.apache.cassandra.concurrent.StageManager; -import org.apache.cassandra.db.*; -import org.apache.cassandra.db.filter.QueryPath; -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.io.PrecompactedRow; -import org.apache.cassandra.locator.AbstractReplicationStrategy; -import org.apache.cassandra.locator.TokenMetadata; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.MerkleTree; - -import static org.apache.cassandra.service.AntiEntropyService.*; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import org.apache.cassandra.utils.ByteBufferUtil; - -public class AntiEntropyServiceTest extends CleanupHelper -{ - // table and column family to test against - public static AntiEntropyService aes; - - public static String tablename; - public static String cfname; - public static TreeRequest request; - public static ColumnFamilyStore store; - public static InetAddress LOCAL, REMOTE; - - @BeforeClass - public static void prepareClass() throws Exception - { - LOCAL = FBUtilities.getLocalAddress(); - tablename = "Keyspace5"; - StorageService.instance.initServer(); - // generate a fake endpoint for which we can spoof receiving/sending trees - REMOTE = InetAddress.getByName("127.0.0.2"); - cfname = "Standard1"; - store = Table.open(tablename).getColumnFamilyStore(cfname); - } - - @Before - public void prepare() throws Exception - { - aes = AntiEntropyService.instance; - TokenMetadata tmd = StorageService.instance.getTokenMetadata(); - tmd.clearUnsafe(); - StorageService.instance.setToken(StorageService.getPartitioner().getRandomToken()); - tmd.updateNormalToken(StorageService.getPartitioner().getMinimumToken(), REMOTE); - assert tmd.isMember(REMOTE); - - // random session id for each test - request = new TreeRequest(UUID.randomUUID().toString(), LOCAL, new CFPair(tablename, cfname)); - } - - @After - public void teardown() throws Exception - { - flushAES(); - } - - @Test - public void testValidatorPrepare() throws Throwable - { - Validator validator; - - // write - List rms = new LinkedList(); - RowMutation rm; - rm = new RowMutation(tablename, ByteBufferUtil.bytes("key1")); - rm.add(new QueryPath(cfname, null, ByteBufferUtil.bytes("Column1")), ByteBufferUtil.bytes("asdf"), 0); - rms.add(rm); - Util.writeColumnFamily(rms); - - // sample - validator = new Validator(request); - validator.prepare(store); - - // and confirm that the tree was split - assertTrue(validator.tree.size() > 1); - } - - @Test - public void testValidatorComplete() throws Throwable - { - Validator validator = new Validator(request); - validator.prepare(store); - validator.complete(); - - // confirm that the tree was validated - Token min = validator.tree.partitioner().getMinimumToken(); - assert null != validator.tree.hash(new Range(min, min)); - } - - @Test - public void testValidatorAdd() throws Throwable - { - Validator validator = new Validator(request); - IPartitioner part = validator.tree.partitioner(); - Token min = part.getMinimumToken(); - Token mid = part.midpoint(min, min); - validator.prepare(store); - - // add a row with the minimum token - validator.add(new PrecompactedRow(new DecoratedKey(min, ByteBufferUtil.bytes("nonsense!")), null)); - - // and a row after it - validator.add(new PrecompactedRow(new DecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")), null)); - validator.complete(); - - // confirm that the tree was validated - assert null != validator.tree.hash(new Range(min, min)); - } - - @Test - public void testManualRepair() throws Throwable - { - AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getRepairSession(tablename, cfname); - sess.start(); - sess.blockUntilRunning(); - - // ensure that the session doesn't end without a response from REMOTE - sess.join(100); - assert sess.isAlive(); - - // deliver a fake response from REMOTE - AntiEntropyService.instance.completedRequest(new TreeRequest(sess.getName(), REMOTE, request.cf)); - - // block until the repair has completed - sess.join(); - } - - @Test - public void testGetNeighborsPlusOne() throws Throwable - { - // generate rf+1 nodes, and ensure that all nodes are returned - Set expected = addTokens(1 + Table.open(tablename).getReplicationStrategy().getReplicationFactor()); - expected.remove(FBUtilities.getLocalAddress()); - assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); - } - - @Test - public void testGetNeighborsTimesTwo() throws Throwable - { - TokenMetadata tmd = StorageService.instance.getTokenMetadata(); - - // generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned - addTokens(2 * Table.open(tablename).getReplicationStrategy().getReplicationFactor()); - AbstractReplicationStrategy ars = Table.open(tablename).getReplicationStrategy(); - Set expected = new HashSet(); - for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getLocalAddress())) - { - expected.addAll(ars.getRangeAddresses(tmd).get(replicaRange)); - } - expected.remove(FBUtilities.getLocalAddress()); - assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); - } - - @Test - public void testDifferencer() throws Throwable - { - // this next part does some housekeeping so that cleanup in the dfferencer doesn't error out. - AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getArtificialRepairSession(request, tablename, cfname); - - // generate a tree - Validator validator = new Validator(request); - validator.prepare(store); - validator.complete(); - MerkleTree ltree = validator.tree; - - // and a clone - validator = new Validator(request); - validator.prepare(store); - validator.complete(); - MerkleTree rtree = validator.tree; - - // change a range we own in one of the trees - Token ltoken = StorageService.instance.getLocalToken(); - ltree.invalidate(ltoken); - MerkleTree.TreeRange changed = ltree.invalids(StorageService.instance.getLocalPrimaryRange()).next(); - changed.hash("non-empty hash!".getBytes()); - // the changed range has two halves, split on our local token: both will be repaired - // (since this keyspace has RF > N, so every node is responsible for the entire ring) - Set interesting = new HashSet(); - interesting.add(new Range(changed.left, ltoken)); - interesting.add(new Range(ltoken, changed.right)); - - // difference the trees - Differencer diff = new Differencer(request, ltree, rtree); - diff.run(); - - // ensure that the changed range was recorded - assertEquals("Wrong differing ranges", interesting, new HashSet(diff.differences)); - } - - Set addTokens(int max) throws Throwable - { - TokenMetadata tmd = StorageService.instance.getTokenMetadata(); - Set endpoints = new HashSet(); - for (int i = 1; i <= max; i++) - { - InetAddress endpoint = InetAddress.getByName("127.0.0." + i); - tmd.updateNormalToken(StorageService.getPartitioner().getRandomToken(), endpoint); - endpoints.add(endpoint); - } - return endpoints; - } - - void flushAES() throws Exception - { - final ThreadPoolExecutor stage = StageManager.getStage(Stage.ANTI_ENTROPY); - final Callable noop = new Callable() - { - public Boolean call() - { - return true; - } - }; - - // send two tasks through the stage: one to follow existing tasks and a second to follow tasks created by - // those existing tasks: tasks won't recursively create more tasks - stage.submit(noop).get(5000, TimeUnit.MILLISECONDS); - stage.submit(noop).get(5000, TimeUnit.MILLISECONDS); - } -} diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java index 8007c0e663..4045bc9469 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java @@ -62,6 +62,8 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper public ColumnFamilyStore store; public InetAddress LOCAL, REMOTE; + public Range local_range; + private boolean initialized; public abstract void init(); @@ -99,9 +101,11 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper StorageService.instance.setToken(StorageService.getPartitioner().getRandomToken()); tmd.updateNormalToken(StorageService.getPartitioner().getMinimumToken(), REMOTE); assert tmd.isMember(REMOTE); + + local_range = StorageService.instance.getLocalPrimaryRange(); // random session id for each test - request = new TreeRequest(UUID.randomUUID().toString(), LOCAL, new CFPair(tablename, cfname)); + request = new TreeRequest(UUID.randomUUID().toString(), LOCAL, local_range, new CFPair(tablename, cfname)); } @After @@ -143,25 +147,21 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper { Validator validator = new Validator(request); IPartitioner part = validator.tree.partitioner(); - Token min = part.getMinimumToken(); - Token mid = part.midpoint(min, min); + Token mid = part.midpoint(local_range.left, local_range.right); validator.prepare(store); - // add a row with the minimum token - validator.add(new PrecompactedRow(new DecoratedKey(min, ByteBufferUtil.bytes("nonsense!")), null)); - - // and a row after it + // add a row validator.add(new PrecompactedRow(new DecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")), null)); validator.complete(); // confirm that the tree was validated - assert null != validator.tree.hash(new Range(min, min)); + assert null != validator.tree.hash(local_range); } @Test public void testManualRepair() throws Throwable { - AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getRepairSession(tablename, cfname); + AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getRepairSession(local_range, tablename, cfname); sess.start(); sess.blockUntilRunning(); @@ -170,7 +170,7 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper assert sess.isAlive(); // deliver a fake response from REMOTE - AntiEntropyService.instance.completedRequest(new TreeRequest(sess.getName(), REMOTE, request.cf)); + AntiEntropyService.instance.completedRequest(new TreeRequest(sess.getName(), REMOTE, local_range, request.cf)); // block until the repair has completed sess.join(); @@ -182,7 +182,13 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper // generate rf+1 nodes, and ensure that all nodes are returned Set expected = addTokens(1 + Table.open(tablename).getReplicationStrategy().getReplicationFactor()); expected.remove(FBUtilities.getLocalAddress()); - assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); + Collection ranges = StorageService.instance.getLocalRanges(tablename); + Set neighbors = new HashSet(); + for (Range range : ranges) + { + neighbors.addAll(AntiEntropyService.getNeighbors(tablename, range)); + } + assertEquals(expected, neighbors); } @Test @@ -199,14 +205,20 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper expected.addAll(ars.getRangeAddresses(tmd).get(replicaRange)); } expected.remove(FBUtilities.getLocalAddress()); - assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); + Collection ranges = StorageService.instance.getLocalRanges(tablename); + Set neighbors = new HashSet(); + for (Range range : ranges) + { + neighbors.addAll(AntiEntropyService.getNeighbors(tablename, range)); + } + assertEquals(expected, neighbors); } @Test public void testDifferencer() throws Throwable { // this next part does some housekeeping so that cleanup in the differencer doesn't error out. - AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getArtificialRepairSession(request, tablename, cfname); + AntiEntropyService.RepairSession sess = AntiEntropyService.instance.getArtificialRepairSession(request, tablename, cfname); // generate a tree Validator validator = new Validator(request); @@ -220,16 +232,14 @@ public abstract class AntiEntropyServiceTestAbstract extends CleanupHelper validator.complete(); MerkleTree rtree = validator.tree; - // change a range we own in one of the trees - Token ltoken = StorageService.instance.getLocalToken(); + // change a range in one of the trees + Token ltoken = StorageService.getPartitioner().midpoint(local_range.left, local_range.right); ltree.invalidate(ltoken); - MerkleTree.TreeRange changed = ltree.invalids(StorageService.instance.getLocalPrimaryRange()).next(); + MerkleTree.TreeRange changed = ltree.get(ltoken); changed.hash("non-empty hash!".getBytes()); - // the changed range has two halves, split on our local token: both will be repaired - // (since this keyspace has RF > N, so every node is responsible for the entire ring) + Set interesting = new HashSet(); - interesting.add(new Range(changed.left, ltoken)); - interesting.add(new Range(ltoken, changed.right)); + interesting.add(changed); // difference the trees Differencer diff = new Differencer(request, ltree, rtree); diff --git a/test/unit/org/apache/cassandra/service/SerializationsTest.java b/test/unit/org/apache/cassandra/service/SerializationsTest.java index bb000c0c41..496d03e3f1 100644 --- a/test/unit/org/apache/cassandra/service/SerializationsTest.java +++ b/test/unit/org/apache/cassandra/service/SerializationsTest.java @@ -24,6 +24,7 @@ package org.apache.cassandra.service; import org.apache.cassandra.AbstractSerializationsTester; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.RandomPartitioner; +import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; @@ -39,6 +40,8 @@ import java.util.List; public class SerializationsTest extends AbstractSerializationsTester { + public static Range FULL_RANGE = new Range(StorageService.getPartitioner().getMinimumToken(), StorageService.getPartitioner().getMinimumToken()); + private void testTreeRequestWrite() throws IOException { DataOutputStream out = getOutput("service.TreeRequest.bin"); @@ -63,7 +66,7 @@ public class SerializationsTest extends AbstractSerializationsTester { AntiEntropyService.Validator v0 = new AntiEntropyService.Validator(Statics.req); IPartitioner part = new RandomPartitioner(); - MerkleTree mt = new MerkleTree(part, MerkleTree.RECOMMENDED_DEPTH, Integer.MAX_VALUE); + MerkleTree mt = new MerkleTree(part, FULL_RANGE, MerkleTree.RECOMMENDED_DEPTH, Integer.MAX_VALUE); List tokens = new ArrayList(); for (int i = 0; i < 10; i++) { @@ -97,6 +100,6 @@ public class SerializationsTest extends AbstractSerializationsTester private static class Statics { private static final AntiEntropyService.CFPair pair = new AntiEntropyService.CFPair("Keyspace1", "Standard1"); - private static final AntiEntropyService.TreeRequest req = new AntiEntropyService.TreeRequest("sessionId", FBUtilities.getLocalAddress(), pair); + private static final AntiEntropyService.TreeRequest req = new AntiEntropyService.TreeRequest("sessionId", FBUtilities.getLocalAddress(), FULL_RANGE, pair); } } diff --git a/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java b/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java index 7525122f00..a5ad29615c 100644 --- a/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java +++ b/test/unit/org/apache/cassandra/utils/MerkleTreeTest.java @@ -69,12 +69,17 @@ public class MerkleTreeTest protected IPartitioner partitioner; protected MerkleTree mt; + private Range fullRange() + { + return new Range(partitioner.getMinimumToken(), partitioner.getMinimumToken()); + } + @Before public void clear() { TOKEN_SCALE = new BigInteger("8"); partitioner = new RandomPartitioner(); - mt = new MerkleTree(partitioner, RECOMMENDED_DEPTH, Integer.MAX_VALUE); + mt = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, Integer.MAX_VALUE); } public static void assertHashEquals(final byte[] left, final byte[] right) @@ -139,7 +144,7 @@ public class MerkleTreeTest @Test public void testSplitLimitDepth() { - mt = new MerkleTree(partitioner, (byte)2, Integer.MAX_VALUE); + mt = new MerkleTree(partitioner, fullRange(), (byte)2, Integer.MAX_VALUE); assertTrue(mt.split(tok(4))); assertTrue(mt.split(tok(2))); @@ -156,7 +161,7 @@ public class MerkleTreeTest @Test public void testSplitLimitSize() { - mt = new MerkleTree(partitioner, RECOMMENDED_DEPTH, 2); + mt = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, 2); assertTrue(mt.split(tok(4))); assertEquals(2, mt.size()); @@ -168,89 +173,13 @@ public class MerkleTreeTest assertEquals(new Range(tok(-1), tok(4)), mt.get(tok(4))); } - @Test - public void testCompact() - { - // (zero, one], (one,two], ... (seven, zero] - mt.split(tok(4)); - mt.split(tok(2)); - mt.split(tok(6)); - mt.split(tok(1)); - mt.split(tok(3)); - mt.split(tok(5)); - mt.split(tok(7)); - - // compact (zero,two] and then (four,six] - mt.compact(tok(1)); - mt.compact(tok(5)); - assertEquals(6, mt.size()); - assertEquals(new Range(tok(-1), tok(2)), mt.get(tok(2))); - assertEquals(new Range(tok(2), tok(3)), mt.get(tok(3))); - assertEquals(new Range(tok(3), tok(4)), mt.get(tok(4))); - assertEquals(new Range(tok(4), tok(6)), mt.get(tok(5))); - assertEquals(new Range(tok(6), tok(7)), mt.get(tok(7))); - assertEquals(new Range(tok(7), tok(-1)), mt.get(tok(-1))); - // compacted ranges should be at depth 2, and the rest at 3 - for (int i : new int[]{2,6}){ assertEquals((byte)2, mt.get(tok(i)).depth); } - for (int i : new int[]{3,4,7,-1}){ assertEquals((byte)3, mt.get(tok(i)).depth); } - - // compact (two,four] and then (six,zero] - mt.compact(tok(3)); - mt.compact(tok(7)); - assertEquals(4, mt.size()); - assertEquals(new Range(tok(-1), tok(2)), mt.get(tok(2))); - assertEquals(new Range(tok(2), tok(4)), mt.get(tok(4))); - assertEquals(new Range(tok(4), tok(6)), mt.get(tok(5))); - assertEquals(new Range(tok(6), tok(-1)), mt.get(tok(-1))); - for (int i : new int[]{2,4,5,-1}){ assertEquals((byte)2, mt.get(tok(i)).depth); } - - // compact (zero,four] - mt.compact(tok(2)); - assertEquals(3, mt.size()); - assertEquals(new Range(tok(-1), tok(4)), mt.get(tok(2))); - assertEquals(new Range(tok(4), tok(6)), mt.get(tok(6))); - assertEquals(new Range(tok(6), tok(-1)), mt.get(tok(-1))); - - // compact (four, zero] - mt.compact(tok(6)); - assertEquals(2, mt.size()); - assertEquals(new Range(tok(-1), tok(4)), mt.get(tok(2))); - assertEquals(new Range(tok(4), tok(-1)), mt.get(tok(6))); - assertEquals((byte)1, mt.get(tok(2)).depth); - assertEquals((byte)1, mt.get(tok(6)).depth); - - // compact (zero, zero] (the root) - mt.compact(tok(4)); - assertEquals(1, mt.size()); - assertEquals(new Range(tok(-1), tok(-1)), mt.get(tok(-1))); - assertEquals((byte)0, mt.get(tok(-1)).depth); - } - - @Test - public void testCompactHash() - { - byte[] val = DUMMY; - byte[] valXval = hashed(val, 1, 1); - - // (zero, four], (four,zero] - mt.split(tok(4)); - - // validate both ranges - mt.get(tok(4)).hash(val); - mt.get(tok(-1)).hash(val); - - // compact (zero, eight] - mt.compact(tok(4)); - assertHashEquals(valXval, mt.get(tok(-1)).hash()); - } - @Test public void testInvalids() { Iterator ranges; // (zero, zero] - ranges = mt.invalids(new Range(tok(-1), tok(-1))); + ranges = mt.invalids(); assertEquals(new Range(tok(-1), tok(-1)), ranges.next()); assertFalse(ranges.hasNext()); @@ -260,7 +189,8 @@ public class MerkleTreeTest mt.split(tok(6)); mt.split(tok(3)); mt.split(tok(5)); - ranges = mt.invalids(new Range(tok(-1), tok(-1))); + ranges = mt.invalids(); + assertEquals(new Range(tok(6), tok(-1)), ranges.next()); assertEquals(new Range(tok(-1), tok(2)), ranges.next()); assertEquals(new Range(tok(2), tok(3)), ranges.next()); assertEquals(new Range(tok(3), tok(4)), ranges.next()); @@ -268,30 +198,9 @@ public class MerkleTreeTest assertEquals(new Range(tok(5), tok(6)), ranges.next()); assertEquals(new Range(tok(6), tok(-1)), ranges.next()); assertFalse(ranges.hasNext()); - - // some invalid - mt.get(tok(2)).hash("non-null!".getBytes()); - mt.get(tok(4)).hash("non-null!".getBytes()); - mt.get(tok(5)).hash("non-null!".getBytes()); - mt.get(tok(-1)).hash("non-null!".getBytes()); - ranges = mt.invalids(new Range(tok(-1), tok(-1))); - assertEquals(new Range(tok(2), tok(3)), ranges.next()); - assertEquals(new Range(tok(5), tok(6)), ranges.next()); - assertFalse(ranges.hasNext()); - - // some invalid in left subrange - ranges = mt.invalids(new Range(tok(-1), tok(6))); - assertEquals(new Range(tok(2), tok(3)), ranges.next()); - assertEquals(new Range(tok(5), tok(6)), ranges.next()); - assertFalse(ranges.hasNext()); - - // some invalid in right subrange - ranges = mt.invalids(new Range(tok(2), tok(-1))); - assertEquals(new Range(tok(2), tok(3)), ranges.next()); - assertEquals(new Range(tok(5), tok(6)), ranges.next()); - assertFalse(ranges.hasNext()); } + @Test public void testHashFull() { @@ -387,7 +296,7 @@ public class MerkleTreeTest Range full = new Range(tok(-1), tok(-1)); Range invalid = new Range(tok(4), tok(-1)); - mt = new MerkleTree(partitioner, RECOMMENDED_DEPTH, Integer.MAX_VALUE); + mt = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, Integer.MAX_VALUE); mt.split(tok(16)); mt.split(tok(8)); mt.split(tok(4)); @@ -416,7 +325,7 @@ public class MerkleTreeTest int max = 1000000; TOKEN_SCALE = new BigInteger("" + max); - mt = new MerkleTree(partitioner, RECOMMENDED_DEPTH, 32); + mt = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, 32); Random random = new Random(); while (true) { @@ -425,7 +334,7 @@ public class MerkleTreeTest } // validate the tree - TreeRangeIterator ranges = mt.invalids(new Range(tok(-1), tok(-1))); + TreeRangeIterator ranges = mt.invalids(); for (TreeRange range : ranges) range.addHash(new RowHash(range.right, new byte[0])); @@ -446,7 +355,7 @@ public class MerkleTreeTest Range full = new Range(tok(-1), tok(-1)); Iterator ranges; - MerkleTree mt2 = new MerkleTree(partitioner, RECOMMENDED_DEPTH, Integer.MAX_VALUE); + MerkleTree mt2 = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, Integer.MAX_VALUE); mt.split(tok(8)); mt.split(tok(4)); @@ -454,7 +363,7 @@ public class MerkleTreeTest mt.split(tok(6)); mt.split(tok(10)); - ranges = mt.invalids(full); + ranges = mt.invalids(); ranges.next().addAll(new HIterator(2, 4)); // (-1,4]: depth 2 ranges.next().addAll(new HIterator(6)); // (4,6] ranges.next().addAll(new HIterator(8)); // (6,8] @@ -471,7 +380,7 @@ public class MerkleTreeTest mt2.split(tok(9)); mt2.split(tok(11)); - ranges = mt2.invalids(full); + ranges = mt2.invalids(); ranges.next().addAll(new HIterator(2)); // (-1,2] ranges.next().addAll(new HIterator(4)); // (2,4] ranges.next().addAll(new HIterator(6, 8)); // (4,8]: depth 2 @@ -496,7 +405,7 @@ public class MerkleTreeTest // populate and validate the tree mt.maxsize(256); mt.init(); - for (TreeRange range : mt.invalids(full)) + for (TreeRange range : mt.invalids()) range.addAll(new HIterator(range.right)); byte[] initialhash = mt.hash(full); @@ -509,6 +418,7 @@ public class MerkleTreeTest // restore partitioner after serialization restored.partitioner(partitioner); + restored.fullRange = fullRange(); assertHashEquals(initialhash, restored.hash(full)); } @@ -519,40 +429,34 @@ public class MerkleTreeTest Range full = new Range(tok(-1), tok(-1)); int maxsize = 16; mt.maxsize(maxsize); - MerkleTree mt2 = new MerkleTree(partitioner, RECOMMENDED_DEPTH, maxsize); + MerkleTree mt2 = new MerkleTree(partitioner, fullRange(), RECOMMENDED_DEPTH, maxsize); mt.init(); mt2.init(); + // add dummy hashes to both trees + for (TreeRange range : mt.invalids()) + range.addAll(new HIterator(range.right)); + for (TreeRange range : mt2.invalids()) + range.addAll(new HIterator(range.right)); + TreeRange leftmost = null; TreeRange middle = null; - TreeRange rightmost = null; - // compact the leftmost, and split the rightmost - Iterator ranges = mt.invalids(full); + mt.maxsize(maxsize + 2); // give some room for splitting + + // split the leftmost + Iterator ranges = mt.invalids(); leftmost = ranges.next(); - rightmost = null; - while (ranges.hasNext()) - rightmost = ranges.next(); - mt.compact(leftmost.right); - leftmost = mt.get(leftmost.right); // leftmost is now a larger range - mt.split(rightmost.right); + mt.split(leftmost.left); // set the hash for the left neighbor of rightmost - middle = mt.get(rightmost.left); + middle = mt.get(leftmost.right); middle.hash("arbitrary!".getBytes()); - byte depth = middle.depth; - // add dummy hashes to the rest of both trees - for (TreeRange range : mt.invalids(full)) - range.addAll(new HIterator(range.right)); - for (TreeRange range : mt2.invalids(full)) - range.addAll(new HIterator(range.right)); - - // trees should disagree for leftmost, (middle.left, rightmost.right] + // trees should disagree for (middle.left, rightmost.right] List diffs = MerkleTree.difference(mt, mt2); - assertEquals(diffs + " contains wrong number of differences:", 2, diffs.size()); - assertTrue(diffs.contains(leftmost)); - assertTrue(diffs.contains(new Range(middle.left, rightmost.right))); + assertEquals(diffs + " contains wrong number of differences:", 1, diffs.size()); + assertTrue(diffs.contains(new Range(leftmost.left, middle.right))); } /**