mirror of https://github.com/apache/cassandra
Make repair work on a token range instead of the full ring
patch by slebresne; reviewed by stuhood for CASSANDRA-2324 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1090840 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Sylvain Lebresne
parent
798db242c4
commit
dc89826fd8
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -1736,6 +1736,18 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
|
|||
return Iterables.concat(samples);
|
||||
}
|
||||
|
||||
public Iterable<DecoratedKey> keySamples(Range range)
|
||||
{
|
||||
Collection<SSTableReader> sstables = getSSTables();
|
||||
Iterable<DecoratedKey>[] 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
|
||||
|
|
|
|||
|
|
@ -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<SSTableReader> sstables, Range range) throws IOException
|
||||
{
|
||||
CollatingIterator iter = FBUtilities.getCollatingIterator();
|
||||
for (SSTableReader sstable : sstables)
|
||||
{
|
||||
iter.addIterator(sstable.getDirectScanner(FILE_BUFFER_SIZE, range));
|
||||
}
|
||||
return iter;
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -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<Pair<Long, Long>> rangeIterator;
|
||||
private Pair<Long, Long> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -381,6 +381,97 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
|
|||
});
|
||||
}
|
||||
|
||||
private static List<Pair<Integer,Integer>> getSampleIndexesForRanges(List<IndexSummary.KeyPosition> samples, Collection<Range> ranges)
|
||||
{
|
||||
// use the index to determine a minimal section for each range
|
||||
List<Pair<Integer,Integer>> positions = new ArrayList<Pair<Integer,Integer>>();
|
||||
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<DecoratedKey> getKeySamples(final Range range)
|
||||
{
|
||||
final List<IndexSummary.KeyPosition> samples = indexSummary.getIndexPositions();
|
||||
|
||||
final List<Pair<Integer, Integer>> indexRanges = getSampleIndexesForRanges(samples, Collections.singletonList(range));
|
||||
|
||||
if (indexRanges.isEmpty())
|
||||
return Collections.emptyList();
|
||||
|
||||
return new Iterable<DecoratedKey>()
|
||||
{
|
||||
public Iterator<DecoratedKey> iterator()
|
||||
{
|
||||
return new Iterator<DecoratedKey>()
|
||||
{
|
||||
private Iterator<Pair<Integer, Integer>> rangeIter = indexRanges.iterator();
|
||||
private Pair<Integer, Integer> 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<SSTableReader>
|
|||
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);
|
||||
|
|
|
|||
|
|
@ -40,11 +40,11 @@ public class SSTableScanner implements Iterator<IColumnIterator>, 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<IColumnIterator> iterator;
|
||||
protected boolean exhausted = false;
|
||||
protected Iterator<IColumnIterator> iterator;
|
||||
private QueryFilter filter;
|
||||
|
||||
/**
|
||||
|
|
@ -141,9 +141,9 @@ public class SSTableScanner implements Iterator<IColumnIterator>, Closeable
|
|||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
private class KeyScanningIterator implements Iterator<IColumnIterator>
|
||||
protected class KeyScanningIterator implements Iterator<IColumnIterator>
|
||||
{
|
||||
private long finishedAt;
|
||||
protected long finishedAt;
|
||||
|
||||
public boolean hasNext()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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<InetAddress> getNeighbors(String table)
|
||||
static Set<InetAddress> getNeighbors(String table, Range range)
|
||||
{
|
||||
StorageService ss = StorageService.instance;
|
||||
Set<InetAddress> neighbors = new HashSet<InetAddress>();
|
||||
Map<Range, List<InetAddress>> 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<InetAddress> neighbors = new HashSet<InetAddress>(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<MerkleTree.RowHash> 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<MerkleTree.RowHash>();
|
||||
mintoken = null;
|
||||
validated = 0;
|
||||
range = null;
|
||||
ranges = null;
|
||||
}
|
||||
|
||||
|
||||
public void prepare(ColumnFamilyStore cfs)
|
||||
{
|
||||
List<DecoratedKey> keys = new ArrayList<DecoratedKey>();
|
||||
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<Range> differences;
|
||||
public List<Range> differences;
|
||||
|
||||
public Differencer(TreeRequest request, MerkleTree ltree, MerkleTree rtree)
|
||||
{
|
||||
this.request = request;
|
||||
this.ltree = ltree;
|
||||
this.rtree = rtree;
|
||||
differences = new ArrayList<Range>();
|
||||
this.differences = new ArrayList<Range>();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -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<Range> 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<Range> ranges = new ArrayList<Range>(differences);
|
||||
Collection<SSTableReader> 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 "#<TreeRequest " + sessionid + ", " + endpoint + ", " + cf + ">";
|
||||
return "#<TreeRequest " + sessionid + ", " + endpoint + ", " + cf + ", " + range + ">";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<TreeRequest,Object> 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<TreeRequest,Object>();
|
||||
}
|
||||
|
|
@ -767,10 +768,11 @@ public class AntiEntropyService
|
|||
@Override
|
||||
public void run()
|
||||
{
|
||||
Set<InetAddress> endpoints = AntiEntropyService.getNeighbors(tablename);
|
||||
Set<InetAddress> 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();
|
||||
|
|
|
|||
|
|
@ -1440,6 +1440,34 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
|
|||
* @throws IOException
|
||||
*/
|
||||
public void forceTableRepair(final String tableName, final String... columnFamilies) throws IOException
|
||||
{
|
||||
List<AntiEntropyService.RepairSession> sessions = new ArrayList<AntiEntropyService.RepairSession>();
|
||||
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 */
|
||||
|
|
|
|||
|
|
@ -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<TreeRange> 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<TreeRange> diff = new ArrayList<TreeRange>();
|
||||
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<TreeRange> implements Iterable<TreeRange>, PeekingIterator<TreeRange>
|
||||
{
|
||||
// stack of ranges to visit
|
||||
private final ArrayDeque<TreeRange> tovisit;
|
||||
// interesting range
|
||||
private final Range range;
|
||||
private final MerkleTree tree;
|
||||
|
||||
TreeRangeIterator(MerkleTree tree, Range range)
|
||||
{
|
||||
Token mintoken = tree.partitioner().getMinimumToken();
|
||||
tovisit = new ArrayDeque<TreeRange>();
|
||||
tovisit.add(new TreeRange(tree, mintoken, mintoken, (byte)0, tree.root));
|
||||
|
||||
TreeRangeIterator(MerkleTree tree)
|
||||
{
|
||||
tovisit = new ArrayDeque<TreeRange>();
|
||||
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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<RowMutation> rms = new LinkedList<RowMutation>();
|
||||
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<InetAddress> 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<InetAddress> expected = new HashSet<InetAddress>();
|
||||
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<Range> interesting = new HashSet<Range>();
|
||||
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<Range>(diff.differences));
|
||||
}
|
||||
|
||||
Set<InetAddress> addTokens(int max) throws Throwable
|
||||
{
|
||||
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
|
||||
Set<InetAddress> endpoints = new HashSet<InetAddress>();
|
||||
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<Object>()
|
||||
{
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<InetAddress> expected = addTokens(1 + Table.open(tablename).getReplicationStrategy().getReplicationFactor());
|
||||
expected.remove(FBUtilities.getLocalAddress());
|
||||
assertEquals(expected, AntiEntropyService.getNeighbors(tablename));
|
||||
Collection<Range> ranges = StorageService.instance.getLocalRanges(tablename);
|
||||
Set<InetAddress> neighbors = new HashSet<InetAddress>();
|
||||
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<Range> ranges = StorageService.instance.getLocalRanges(tablename);
|
||||
Set<InetAddress> neighbors = new HashSet<InetAddress>();
|
||||
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<Range> interesting = new HashSet<Range>();
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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<Token> tokens = new ArrayList<Token>();
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<TreeRange> 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<TreeRange> 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<TreeRange> ranges = mt.invalids(full);
|
||||
mt.maxsize(maxsize + 2); // give some room for splitting
|
||||
|
||||
// split the leftmost
|
||||
Iterator<TreeRange> 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<TreeRange> 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)));
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Reference in New Issue