natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Repair common subranges of a set of nodes in one session 

reviewed by Stefania Alborghetti for CASSANDRA-5220
This commit is contained in:
Marcus Olsson 2015-08-06 08:23:10 -05:00 committed by Jonathan Ellis
parent bf4740867a
commit 0dd50a6cdc
32 changed files with 1257 additions and 196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.0-beta1
* Repair improvements when using vnodes (CASSANDRA-5220)
* Disable scripted UDFs by default (CASSANDRA-9889)
* Add transparent data encryption core classes (CASSANDRA-9945)
* Bytecode inspection for Java-UDFs (CASSANDRA-9890)

19
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -69,6 +69,8 @@ import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.SSTableRewriter;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.metrics.CompactionMetrics;
import org.apache.cassandra.repair.Validator;
@ -76,7 +78,7 @@ import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.UUIDGen;
@ -1045,7 +1047,7 @@ public class CompactionManager implements CompactionManagerMBean
for (SSTableReader sstable : sstableCandidates.sstables)
{
if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singletonList(validator.desc.range)))
if (new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(validator.desc.ranges))
{
sstablesToValidate.add(sstable);
}
@ -1074,19 +1076,20 @@ public class CompactionManager implements CompactionManagerMBean
gcBefore = getDefaultGcBefore(cfs, nowInSec);
}
// Create Merkle tree suitable to hold estimated partitions for given range.
// We blindly assume that partition is evenly distributed on all sstables for now.
// Create Merkle trees suitable to hold estimated partitions for the given ranges.
// We blindly assume that a partition is evenly distributed on all sstables for now.
long numPartitions = 0;
for (SSTableReader sstable : sstables)
{
numPartitions += sstable.estimatedKeysForRanges(singleton(validator.desc.range));
numPartitions += sstable.estimatedKeysForRanges(validator.desc.ranges);
}
// determine tree depth from number of partitions, but cap at 20 to prevent large tree.
int depth = numPartitions > 0 ? (int) Math.min(Math.floor(Math.log(numPartitions)), 20) : 0;
MerkleTree tree = new MerkleTree(cfs.getPartitioner(), validator.desc.range, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, depth));
MerkleTrees tree = new MerkleTrees(cfs.getPartitioner());
tree.addMerkleTrees((int) Math.pow(2, depth), validator.desc.ranges);
long start = System.nanoTime();
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables, validator.desc.range);
try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables, validator.desc.ranges);
ValidationCompactionController controller = new ValidationCompactionController(cfs, gcBefore);
CompactionIterator ci = new ValidationCompactionIterator(scanners.scanners, controller, nowInSec, metrics))
{
@ -1119,7 +1122,7 @@ public class CompactionManager implements CompactionManagerMBean
duration,
depth,
numPartitions,
MerkleTree.serializer.serializedSize(tree, 0),
MerkleTrees.serializer.serializedSize(tree, 0),
validator.desc);
}
}

24
src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import java.util.concurrent.Callable;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -340,7 +341,7 @@ public class CompactionStrategyManager implements INotificationConsumer
* @return
*/
@SuppressWarnings("resource")
public synchronized AbstractCompactionStrategy.ScannerList getScanners(Collection<SSTableReader> sstables, Range<Token> range)
public synchronized AbstractCompactionStrategy.ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges)
{
List<SSTableReader> repairedSSTables = new ArrayList<>();
List<SSTableReader> unrepairedSSTables = new ArrayList<>();
@ -352,19 +353,26 @@ public class CompactionStrategyManager implements INotificationConsumer
unrepairedSSTables.add(sstable);
}
Set<ISSTableScanner> scanners = new HashSet<>(sstables.size());
AbstractCompactionStrategy.ScannerList repairedScanners = repaired.getScanners(repairedSSTables, range);
AbstractCompactionStrategy.ScannerList unrepairedScanners = unrepaired.getScanners(unrepairedSSTables, range);
for (Range<Token> range : ranges)
{
AbstractCompactionStrategy.ScannerList repairedScanners = repaired.getScanners(repairedSSTables, range);
AbstractCompactionStrategy.ScannerList unrepairedScanners = unrepaired.getScanners(unrepairedSSTables, range);
List<ISSTableScanner> scanners = new ArrayList<>(repairedScanners.scanners.size() + unrepairedScanners.scanners.size());
scanners.addAll(repairedScanners.scanners);
scanners.addAll(unrepairedScanners.scanners);
return new AbstractCompactionStrategy.ScannerList(scanners);
for (ISSTableScanner scanner : Iterables.concat(repairedScanners.scanners, unrepairedScanners.scanners))
{
if (!scanners.add(scanner))
scanner.close();
}
}
return new AbstractCompactionStrategy.ScannerList(new ArrayList<>(scanners));
}
public synchronized AbstractCompactionStrategy.ScannerList getScanners(Collection<SSTableReader> sstables)
{
return getScanners(sstables, null);
return getScanners(sstables, Collections.singleton(null));
}
public Collection<Collection<SSTableReader>> groupSSTablesForAntiCompaction(Collection<SSTableReader> sstablesToGroup)

6
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -1111,6 +1111,12 @@ public final class MessagingService implements MessagingServiceMBean
return StorageService.instance.getTokenMetadata().partitioner;
}
public static void validatePartitioner(Collection<? extends AbstractBounds<?>> allBounds)
{
for (AbstractBounds<?> bounds : allBounds)
validatePartitioner(bounds);
}
public static void validatePartitioner(AbstractBounds<?> bounds)
{
if (globalPartitioner() != bounds.left.getPartitioner())

17
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -48,21 +48,14 @@ public class RepairJob extends AbstractFuture<RepairResult> implements Runnable
*
* @param session RepairSession that this RepairJob belongs
* @param columnFamily name of the ColumnFamily to repair
* @param parallelismDegree how to run repair job in parallel
* @param repairedAt when the repair occurred (millis)
* @param taskExecutor Executor to run various repair tasks
*/
public RepairJob(RepairSession session,
String columnFamily,
RepairParallelism parallelismDegree,
long repairedAt,
ListeningExecutorService taskExecutor)
public RepairJob(RepairSession session, String columnFamily)
{
this.session = session;
this.desc = new RepairJobDesc(session.parentRepairSession, session.getId(), session.keyspace, columnFamily, session.getRange());
this.repairedAt = repairedAt;
this.taskExecutor = taskExecutor;
this.parallelismDegree = parallelismDegree;
this.desc = new RepairJobDesc(session.parentRepairSession, session.getId(), session.keyspace, columnFamily, session.getRanges());
this.repairedAt = session.repairedAt;
this.taskExecutor = session.taskExecutor;
this.parallelismDegree = session.parallelismDegree;
}
/**

41
src/java/org/apache/cassandra/repair/RepairJobDesc.java
View File

@ -18,6 +18,8 @@
package org.apache.cassandra.repair;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.UUID;
import com.google.common.base.Objects;
@ -47,21 +49,21 @@ public class RepairJobDesc
public final String keyspace;
public final String columnFamily;
/** repairing range */
public final Range<Token> range;
public final Collection<Range<Token>> ranges;
public RepairJobDesc(UUID parentSessionId, UUID sessionId, String keyspace, String columnFamily, Range<Token> range)
public RepairJobDesc(UUID parentSessionId, UUID sessionId, String keyspace, String columnFamily, Collection<Range<Token>> ranges)
{
this.parentSessionId = parentSessionId;
this.sessionId = sessionId;
this.keyspace = keyspace;
this.columnFamily = columnFamily;
this.range = range;
this.ranges = ranges;
}
@Override
public String toString()
{
return "[repair #" + sessionId + " on " + keyspace + "/" + columnFamily + ", " + range + "]";
return "[repair #" + sessionId + " on " + keyspace + "/" + columnFamily + ", " + ranges + "]";
}
@Override
@ -74,7 +76,7 @@ public class RepairJobDesc
if (!columnFamily.equals(that.columnFamily)) return false;
if (!keyspace.equals(that.keyspace)) return false;
if (range != null ? !range.equals(that.range) : that.range != null) return false;
if (ranges != null ? that.ranges == null || (ranges.size() != that.ranges.size()) || (ranges.size() == that.ranges.size() && !ranges.containsAll(that.ranges)) : that.ranges != null) return false;
if (!sessionId.equals(that.sessionId)) return false;
if (parentSessionId != null ? !parentSessionId.equals(that.parentSessionId) : that.parentSessionId != null) return false;
@ -84,7 +86,7 @@ public class RepairJobDesc
@Override
public int hashCode()
{
return Objects.hashCode(sessionId, keyspace, columnFamily, range);
return Objects.hashCode(sessionId, keyspace, columnFamily, ranges);
}
private static class RepairJobDescSerializer implements IVersionedSerializer<RepairJobDesc>
@ -100,8 +102,10 @@ public class RepairJobDesc
UUIDSerializer.serializer.serialize(desc.sessionId, out, version);
out.writeUTF(desc.keyspace);
out.writeUTF(desc.columnFamily);
MessagingService.validatePartitioner(desc.range);
AbstractBounds.tokenSerializer.serialize(desc.range, out, version);
MessagingService.validatePartitioner(desc.ranges);
out.writeInt(desc.ranges.size());
for (Range<Token> rt : desc.ranges)
AbstractBounds.tokenSerializer.serialize(rt, out, version);
}
public RepairJobDesc deserialize(DataInputPlus in, int version) throws IOException
@ -115,8 +119,19 @@ public class RepairJobDesc
UUID sessionId = UUIDSerializer.serializer.deserialize(in, version);
String keyspace = in.readUTF();
String columnFamily = in.readUTF();
Range<Token> range = (Range<Token>)AbstractBounds.tokenSerializer.deserialize(in, MessagingService.globalPartitioner(), version);
return new RepairJobDesc(parentSessionId, sessionId, keyspace, columnFamily, range);
int nRanges = in.readInt();
Collection<Range<Token>> ranges = new ArrayList<>();
Range<Token> range;
for (int i = 0; i < nRanges; i++)
{
range = (Range<Token>) AbstractBounds.tokenSerializer.deserialize(in,
MessagingService.globalPartitioner(), version);
ranges.add(range);
}
return new RepairJobDesc(parentSessionId, sessionId, keyspace, columnFamily, ranges);
}
public long serializedSize(RepairJobDesc desc, int version)
@ -131,7 +146,11 @@ public class RepairJobDesc
size += UUIDSerializer.serializer.serializedSize(desc.sessionId, version);
size += TypeSizes.sizeof(desc.keyspace);
size += TypeSizes.sizeof(desc.columnFamily);
size += AbstractBounds.tokenSerializer.serializedSize(desc.range, version);
size += TypeSizes.sizeof(desc.ranges.size());
for (Range<Token> rt : desc.ranges)
{
size += AbstractBounds.tokenSerializer.serializedSize(rt, version);
}
return size;
}
}

4
src/java/org/apache/cassandra/repair/RepairMessageVerbHandler.java
View File

@ -79,14 +79,14 @@ public class RepairMessageVerbHandler implements IVerbHandler<RepairMessage>
case SNAPSHOT:
logger.debug("Snapshotting {}", desc);
ColumnFamilyStore cfs = Keyspace.open(desc.keyspace).getColumnFamilyStore(desc.columnFamily);
final Range<Token> repairingRange = desc.range;
final Collection<Range<Token>> repairingRange = desc.ranges;
Set<SSTableReader> snapshottedSSSTables = cfs.snapshot(desc.sessionId.toString(), new Predicate<SSTableReader>()
{
public boolean apply(SSTableReader sstable)
{
return sstable != null &&
!sstable.metadata.isIndex() && // exclude SSTables from 2i
new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(Collections.singleton(repairingRange));
new Bounds<>(sstable.first.getToken(), sstable.last.getToken()).intersects(repairingRange);
}
}, true); //ephemeral snapshot, if repair fails, it will be cleaned next startup

45
src/java/org/apache/cassandra/repair/RepairRunnable.java
View File

@ -53,6 +53,7 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.progress.ProgressEvent;
@ -146,17 +147,19 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
final Set<InetAddress> allNeighbors = new HashSet<>();
Map<Range, Set<InetAddress>> rangeToNeighbors = new HashMap<>();
List<Pair<Set<InetAddress>, ? extends Collection<Range<Token>>>> commonRanges = new ArrayList<>();
try
{
for (Range<Token> range : options.getRanges())
{
Set<InetAddress> neighbors = ActiveRepairService.getNeighbors(keyspace, range,
options.getDataCenters(),
options.getHosts());
rangeToNeighbors.put(range, neighbors);
allNeighbors.addAll(neighbors);
Set<InetAddress> neighbors = ActiveRepairService.getNeighbors(keyspace, range,
options.getDataCenters(),
options.getHosts());
addRangeToNeighbors(commonRanges, range, neighbors);
allNeighbors.addAll(neighbors);
}
progress.incrementAndGet();
}
catch (IllegalArgumentException e)
@ -210,13 +213,13 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
"internal"));
List<ListenableFuture<RepairSessionResult>> futures = new ArrayList<>(options.getRanges().size());
for (Range<Token> range : options.getRanges())
for (Pair<Set<InetAddress>, ? extends Collection<Range<Token>>> p : commonRanges)
{
final RepairSession session = ActiveRepairService.instance.submitRepairSession(parentSession,
range,
p.right,
keyspace,
options.getParallelism(),
rangeToNeighbors.get(range),
p.left,
repairedAt,
executor,
cfnames);
@ -228,7 +231,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
public void onSuccess(RepairSessionResult result)
{
String message = String.format("Repair session %s for range %s finished", session.getId(),
session.getRange().toString());
session.getRanges().toString());
logger.info(message);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.PROGRESS,
progress.incrementAndGet(),
@ -239,7 +242,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
public void onFailure(Throwable t)
{
String message = String.format("Repair session %s for range %s failed with error %s",
session.getId(), session.getRange().toString(), t.getMessage());
session.getId(), session.getRanges().toString(), t.getMessage());
logger.error(message, t);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.PROGRESS,
progress.incrementAndGet(),
@ -265,7 +268,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
if (sessionResult != null)
{
successfulRanges.add(sessionResult.range);
successfulRanges.addAll(sessionResult.ranges);
}
else
{
@ -325,6 +328,24 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
});
}
private void addRangeToNeighbors(List<Pair<Set<InetAddress>, ? extends Collection<Range<Token>>>> neighborRangeList, Range<Token> range, Set<InetAddress> neighbors)
{
for (int i = 0; i < neighborRangeList.size(); i++)
{
Pair<Set<InetAddress>, ? extends Collection<Range<Token>>> p = neighborRangeList.get(i);
if (p.left.containsAll(neighbors))
{
p.right.add(range);
return;
}
}
List<Range<Token>> ranges = new ArrayList<>();
ranges.add(range);
neighborRangeList.add(Pair.create(neighbors, ranges));
}
private Thread createQueryThread(final int cmd, final UUID sessionId)
{
return new Thread(new WrappedRunnable()

48
src/java/org/apache/cassandra/repair/RepairSession.java
View File

@ -37,13 +37,13 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.Pair;
/**
* Coordinates the (active) repair of a token range.
* Coordinates the (active) repair of a list of non overlapping token ranges.
*
* A given RepairSession repairs a set of replicas for a given range on a list
* A given RepairSession repairs a set of replicas for a given set of ranges on a list
* of column families. For each of the column family to repair, RepairSession
* creates a {@link RepairJob} that handles the repair of that CF.
*
@ -64,11 +64,11 @@ import org.apache.cassandra.utils.Pair;
* A given session will execute the first phase (validation phase) of each of it's job
* sequentially. In other words, it will start the first job and only start the next one
* once that first job validation phase is complete. This is done so that the replica only
* create one merkle tree at a time, which is our way to ensure that such creation starts
* create one merkle tree per range at a time, which is our way to ensure that such creation starts
* roughly at the same time on every node (see CASSANDRA-2816). However the synchronization
* phases are allowed to run concurrently (with each other and with validation phases).
*
* A given RepairJob has 2 modes: either sequential or not (isSequential flag). If sequential,
* A given RepairJob has 2 modes: either sequential or not (RepairParallelism). If sequential,
* it will requests merkle tree creation from each replica in sequence (though in that case
* we still first send a message to each node to flush and snapshot data so each merkle tree
* creation is still done on similar data, even if the actual creation is not
@ -88,9 +88,9 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
private final String[] cfnames;
public final RepairParallelism parallelismDegree;
/** Range to repair */
public final Range<Token> range;
public final Collection<Range<Token>> ranges;
public final Set<InetAddress> endpoints;
private final long repairedAt;
public final long repairedAt;
// number of validations left to be performed
private final AtomicInteger validationRemaining;
@ -103,7 +103,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
private final ConcurrentMap<Pair<RepairJobDesc, NodePair>, RemoteSyncTask> syncingTasks = new ConcurrentHashMap<>();
// Tasks(snapshot, validate request, differencing, ...) are run on taskExecutor
private final ListeningExecutorService taskExecutor = MoreExecutors.listeningDecorator(DebuggableThreadPoolExecutor.createCachedThreadpoolWithMaxSize("RepairJobTask"));
public final ListeningExecutorService taskExecutor = MoreExecutors.listeningDecorator(DebuggableThreadPoolExecutor.createCachedThreadpoolWithMaxSize("RepairJobTask"));
private volatile boolean terminated = false;
@ -112,7 +112,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
*
* @param parentRepairSession the parent sessions id
* @param id this sessions id
* @param range range to repair
* @param ranges ranges to repair
* @param keyspace name of keyspace
* @param parallelismDegree specifies the degree of parallelism when calculating the merkle trees
* @param endpoints the data centers that should be part of the repair; null for all DCs
@ -121,7 +121,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
*/
public RepairSession(UUID parentRepairSession,
UUID id,
Range<Token> range,
Collection<Range<Token>> ranges,
String keyspace,
RepairParallelism parallelismDegree,
Set<InetAddress> endpoints,
@ -135,7 +135,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
this.parallelismDegree = parallelismDegree;
this.keyspace = keyspace;
this.cfnames = cfnames;
this.range = range;
this.ranges = ranges;
this.endpoints = endpoints;
this.repairedAt = repairedAt;
this.validationRemaining = new AtomicInteger(cfnames.length);
@ -146,9 +146,9 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
return id;
}
public Range<Token> getRange()
public Collection<Range<Token>> getRanges()
{
return range;
return ranges;
}
public void waitForValidation(Pair<RepairJobDesc, InetAddress> key, ValidationTask task)
@ -166,9 +166,9 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
*
* @param desc repair job description
* @param endpoint endpoint that sent merkle tree
* @param tree calculated merkle tree, or null if validation failed
* @param trees calculated merkle trees, or null if validation failed
*/
public void validationComplete(RepairJobDesc desc, InetAddress endpoint, MerkleTree tree)
public void validationComplete(RepairJobDesc desc, InetAddress endpoint, MerkleTrees trees)
{
ValidationTask task = validating.remove(Pair.create(desc, endpoint));
if (task == null)
@ -180,7 +180,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
String message = String.format("Received merkle tree for %s from %s", desc.columnFamily, endpoint);
logger.info("[repair #{}] {}", getId(), message);
Tracing.traceRepair(message);
task.treeReceived(tree);
task.treesReceived(trees);
// Unregister from FailureDetector once we've completed synchronizing Merkle trees.
// After this point, we rely on tcp_keepalive for individual sockets to notify us when a connection is down.
@ -234,15 +234,15 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
if (terminated)
return;
logger.info(String.format("[repair #%s] new session: will sync %s on range %s for %s.%s", getId(), repairedNodes(), range, keyspace, Arrays.toString(cfnames)));
Tracing.traceRepair("Syncing range {}", range);
SystemDistributedKeyspace.startRepairs(getId(), parentRepairSession, keyspace, cfnames, range, endpoints);
logger.info(String.format("[repair #%s] new session: will sync %s on range %s for %s.%s", getId(), repairedNodes(), ranges, keyspace, Arrays.toString(cfnames)));
Tracing.traceRepair("Syncing range {}", ranges);
SystemDistributedKeyspace.startRepairs(getId(), parentRepairSession, keyspace, cfnames, ranges, endpoints);
if (endpoints.isEmpty())
{
logger.info("[repair #{}] {}", getId(), message = String.format("No neighbors to repair with on range %s: session completed", range));
logger.info("[repair #{}] {}", getId(), message = String.format("No neighbors to repair with on range %s: session completed", ranges));
Tracing.traceRepair(message);
set(new RepairSessionResult(id, keyspace, range, Lists.<RepairResult>newArrayList()));
set(new RepairSessionResult(id, keyspace, ranges, Lists.<RepairResult>newArrayList()));
SystemDistributedKeyspace.failRepairs(getId(), keyspace, cfnames, new RuntimeException(message));
return;
}
@ -265,7 +265,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
List<ListenableFuture<RepairResult>> jobs = new ArrayList<>(cfnames.length);
for (String cfname : cfnames)
{
RepairJob job = new RepairJob(this, cfname, parallelismDegree, repairedAt, taskExecutor);
RepairJob job = new RepairJob(this, cfname);
executor.execute(job);
jobs.add(job);
}
@ -277,8 +277,8 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> implement
{
// this repair session is completed
logger.info("[repair #{}] {}", getId(), "Session completed successfully");
Tracing.traceRepair("Completed sync of range {}", range);
set(new RepairSessionResult(id, keyspace, range, results));
Tracing.traceRepair("Completed sync of range {}", ranges);
set(new RepairSessionResult(id, keyspace, ranges, results));
taskExecutor.shutdown();
// mark this session as terminated

6
src/java/org/apache/cassandra/repair/RepairSessionResult.java
View File

@ -30,14 +30,14 @@ public class RepairSessionResult
{
public final UUID sessionId;
public final String keyspace;
public final Range<Token> range;
public final Collection<Range<Token>> ranges;
public final Collection<RepairResult> repairJobResults;
public RepairSessionResult(UUID sessionId, String keyspace, Range<Token> range, Collection<RepairResult> repairJobResults)
public RepairSessionResult(UUID sessionId, String keyspace, Collection<Range<Token>> ranges, Collection<RepairResult> repairJobResults)
{
this.sessionId = sessionId;
this.keyspace = keyspace;
this.range = range;
this.ranges = ranges;
this.repairJobResults = repairJobResults;
}
}

6
src/java/org/apache/cassandra/repair/SyncTask.java
View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.repair;
import java.util.ArrayList;
import java.util.List;
import com.google.common.util.concurrent.AbstractFuture;
@ -27,7 +26,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
/**
* SyncTask will calculate the difference of MerkleTree between two nodes
@ -56,8 +55,7 @@ public abstract class SyncTask extends AbstractFuture<SyncStat> implements Runna
public void run()
{
// compare trees, and collect differences
List<Range<Token>> differences = new ArrayList<>();
differences.addAll(MerkleTree.difference(r1.tree, r2.tree));
List<Range<Token>> differences = MerkleTrees.difference(r1.trees, r2.trees);
stat = new SyncStat(new NodePair(r1.endpoint, r2.endpoint), differences.size());

27
src/java/org/apache/cassandra/repair/SystemDistributedKeyspace.java
View File

@ -130,7 +130,7 @@ public final class SystemDistributedKeyspace
processSilent(fmtQuery);
}
public static void startRepairs(UUID id, UUID parent_id, String keyspaceName, String[] cfnames, Range<Token> range, Iterable<InetAddress> endpoints)
public static void startRepairs(UUID id, UUID parent_id, String keyspaceName, String[] cfnames, Collection<Range<Token>> ranges, Iterable<InetAddress> endpoints)
{
String coordinator = FBUtilities.getBroadcastAddress().getHostAddress();
Set<String> participants = Sets.newHashSet(coordinator);
@ -144,17 +144,20 @@ public final class SystemDistributedKeyspace
for (String cfname : cfnames)
{
String fmtQry = String.format(query, NAME, REPAIR_HISTORY,
keyspaceName,
cfname,
id.toString(),
parent_id.toString(),
range.left.toString(),
range.right.toString(),
coordinator,
Joiner.on("', '").join(participants),
RepairState.STARTED.toString());
processSilent(fmtQry);
for (Range<Token> range : ranges)
{
String fmtQry = String.format(query, NAME, REPAIR_HISTORY,
keyspaceName,
cfname,
id.toString(),
parent_id.toString(),
range.left.toString(),
range.right.toString(),
coordinator,
Joiner.on("', '").join(participants),
RepairState.STARTED.toString());
processSilent(fmtQry);
}
}
}

8
src/java/org/apache/cassandra/repair/TreeResponse.java
View File

@ -19,7 +19,7 @@ package org.apache.cassandra.repair;
import java.net.InetAddress;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
/**
* Merkle tree response sent from given endpoint.
@ -27,11 +27,11 @@ import org.apache.cassandra.utils.MerkleTree;
public class TreeResponse
{
public final InetAddress endpoint;
public final MerkleTree tree;
public final MerkleTrees trees;
public TreeResponse(InetAddress endpoint, MerkleTree tree)
public TreeResponse(InetAddress endpoint, MerkleTrees trees)
{
this.endpoint = endpoint;
this.tree = tree;
this.trees = trees;
}
}

14
src/java/org/apache/cassandra/repair/ValidationTask.java
View File

@ -18,13 +18,17 @@
package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.Map;
import com.google.common.util.concurrent.AbstractFuture;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.exceptions.RepairException;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationRequest;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
/**
* ValidationTask sends {@link ValidationRequest} to a replica.
@ -53,19 +57,19 @@ public class ValidationTask extends AbstractFuture<TreeResponse> implements Runn
}
/**
* Receive MerkleTree from replica node.
* Receive MerkleTrees from replica node.
*
* @param tree MerkleTree that is sent from replica. Null if validation failed on replica node.
* @param trees MerkleTrees that is sent from replica. Null if validation failed on replica node.
*/
public void treeReceived(MerkleTree tree)
public void treesReceived(MerkleTrees trees)
{
if (tree == null)
if (trees == null)
{
setException(new RepairException(desc, "Validation failed in " + endpoint));
}
else
{
set(new TreeResponse(endpoint, tree));
set(new TreeResponse(endpoint, trees));
}
}
}

85
src/java/org/apache/cassandra/repair/Validator.java
View File

@ -33,12 +33,15 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.messages.ValidationComplete;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTree.RowHash;
import org.apache.cassandra.utils.MerkleTrees;
/**
* Handles the building of a merkle tree for a column family.
@ -58,11 +61,11 @@ public class Validator implements Runnable
// null when all rows with the min token have been consumed
private long validated;
private MerkleTree tree;
private MerkleTrees trees;
// current range being updated
private MerkleTree.TreeRange range;
// iterator for iterating sub ranges (MT's leaves)
private MerkleTree.TreeRangeIterator ranges;
private MerkleTrees.TreeRangeIterator ranges;
// last key seen
private DecoratedKey lastKey;
@ -76,9 +79,9 @@ public class Validator implements Runnable
ranges = null;
}
public void prepare(ColumnFamilyStore cfs, MerkleTree tree)
public void prepare(ColumnFamilyStore cfs, MerkleTrees tree)
{
this.tree = tree;
this.trees = tree;
if (!tree.partitioner().preservesOrder())
{
@ -87,32 +90,35 @@ public class Validator implements Runnable
}
else
{
List<DecoratedKey> keys = new ArrayList<>();
for (DecoratedKey sample : cfs.keySamples(desc.range))
for (Range<Token> range : tree.ranges())
{
assert desc.range.contains(sample.getToken()): "Token " + sample.getToken() + " is not within range " + desc.range;
keys.add(sample);
}
if (keys.isEmpty())
{
// use an even tree distribution
tree.init();
}
else
{
int numkeys = keys.size();
Random random = new Random();
// sample the column family using random keys from the index
while (true)
List<DecoratedKey> keys = new ArrayList<>();
for (DecoratedKey sample : cfs.keySamples(range))
{
DecoratedKey dk = keys.get(random.nextInt(numkeys));
if (!tree.split(dk.getToken()))
break;
assert range.contains(sample.getToken()) : "Token " + sample.getToken() + " is not within range " + desc.ranges;
keys.add(sample);
}
if (keys.isEmpty())
{
// use an even tree distribution
tree.init(range);
}
else
{
int numKeys = keys.size();
Random random = new Random();
// sample the column family using random keys from the index
while (true)
{
DecoratedKey dk = keys.get(random.nextInt(numKeys));
if (!tree.split(dk.getToken()))
break;
}
}
}
}
logger.debug("Prepared AEService tree of size {} for {}", tree.size(), desc);
logger.debug("Prepared AEService trees of size {} for {}", trees.size(), desc);
ranges = tree.invalids();
}
@ -124,7 +130,7 @@ public class Validator implements Runnable
*/
public void add(UnfilteredRowIterator partition)
{
assert desc.range.contains(partition.partitionKey().getToken()) : partition.partitionKey().getToken() + " is not contained in " + desc.range;
assert Range.isInRanges(partition.partitionKey().getToken(), desc.ranges) : partition.partitionKey().getToken() + " is not contained in " + desc.ranges;
assert lastKey == null || lastKey.compareTo(partition.partitionKey()) < 0
: "partition " + partition.partitionKey() + " received out of order wrt " + lastKey;
lastKey = partition.partitionKey();
@ -133,13 +139,14 @@ public class Validator implements Runnable
range = ranges.next();
// generate new ranges as long as case 1 is true
while (!range.contains(lastKey.getToken()))
if (!findCorrectRange(lastKey.getToken()))
{
// add the empty hash, and move to the next range
range.ensureHashInitialised();
range = ranges.next();
ranges = trees.invalids();
findCorrectRange(lastKey.getToken());
}
assert range.contains(lastKey.getToken()) : "Token not in MerkleTree: " + lastKey.getToken();
// case 3 must be true: mix in the hashed row
RowHash rowHash = rowHash(partition);
if (rowHash != null)
@ -148,6 +155,16 @@ public class Validator implements Runnable
}
}
public boolean findCorrectRange(Token t)
{
while (!range.contains(t) && ranges.hasNext())
{
range = ranges.next();
}
return range.contains(t);
}
static class CountingDigest extends MessageDigest
{
private long count;
@ -212,9 +229,9 @@ public class Validator implements Runnable
{
// log distribution of rows in tree
logger.debug("Validated {} partitions for {}. Partitions per leaf are:", validated, desc.sessionId);
tree.histogramOfRowCountPerLeaf().log(logger);
trees.logRowCountPerLeaf(logger);
logger.debug("Validated {} partitions for {}. Partition sizes are:", validated, desc.sessionId);
tree.histogramOfRowSizePerLeaf().log(logger);
trees.logRowSizePerLeaf(logger);
}
}
@ -223,8 +240,8 @@ public class Validator implements Runnable
{
assert ranges != null : "Validator was not prepared()";
if (range != null)
range.ensureHashInitialised();
ranges = trees.invalids();
while (ranges.hasNext())
{
range = ranges.next();
@ -255,6 +272,6 @@ public class Validator implements Runnable
logger.info(String.format("[repair #%s] Sending completed merkle tree to %s for %s.%s", desc.sessionId, initiator, desc.keyspace, desc.columnFamily));
Tracing.traceRepair("Sending completed merkle tree to {} for {}.{}", initiator, desc.keyspace, desc.columnFamily);
}
MessagingService.instance().sendOneWay(new ValidationComplete(desc, tree).createMessage(), initiator);
MessagingService.instance().sendOneWay(new ValidationComplete(desc, trees).createMessage(), initiator);
}
}

47
src/java/org/apache/cassandra/repair/messages/ValidationComplete.java
View File

@ -23,7 +23,7 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
/**
* ValidationComplete message is sent when validation compaction completed successfully.
@ -34,24 +34,25 @@ public class ValidationComplete extends RepairMessage
{
public static MessageSerializer serializer = new ValidationCompleteSerializer();
/** true if validation success, false otherwise */
public final boolean success;
/** Merkle hash tree response. Null if validation failed. */
public final MerkleTree tree;
public final MerkleTrees trees;
public ValidationComplete(RepairJobDesc desc)
{
super(Type.VALIDATION_COMPLETE, desc);
this.success = false;
this.tree = null;
trees = null;
}
public ValidationComplete(RepairJobDesc desc, MerkleTree tree)
public ValidationComplete(RepairJobDesc desc, MerkleTrees trees)
{
super(Type.VALIDATION_COMPLETE, desc);
assert tree != null;
this.success = true;
this.tree = tree;
assert trees != null;
this.trees = trees;
}
public boolean success()
{
return trees != null;
}
private static class ValidationCompleteSerializer implements MessageSerializer<ValidationComplete>
@ -59,31 +60,31 @@ public class ValidationComplete extends RepairMessage
public void serialize(ValidationComplete message, DataOutputPlus out, int version) throws IOException
{
RepairJobDesc.serializer.serialize(message.desc, out, version);
out.writeBoolean(message.success);
if (message.success)
MerkleTree.serializer.serialize(message.tree, out, version);
out.writeBoolean(message.success());
if (message.trees != null)
MerkleTrees.serializer.serialize(message.trees, out, version);
}
public ValidationComplete deserialize(DataInputPlus in, int version) throws IOException
{
RepairJobDesc desc = RepairJobDesc.serializer.deserialize(in, version);
if (in.readBoolean())
boolean success = in.readBoolean();
if (success)
{
MerkleTree tree = MerkleTree.serializer.deserialize(in, version);
return new ValidationComplete(desc, tree);
}
else
{
return new ValidationComplete(desc);
MerkleTrees trees = MerkleTrees.serializer.deserialize(in, version);
return new ValidationComplete(desc, trees);
}
return new ValidationComplete(desc);
}
public long serializedSize(ValidationComplete message, int version)
{
long size = RepairJobDesc.serializer.serializedSize(message.desc, version);
size += TypeSizes.sizeof(message.success);
if (message.success)
size += MerkleTree.serializer.serializedSize(message.tree, version);
size += TypeSizes.sizeof(message.success());
if (message.trees != null)
size += MerkleTrees.serializer.serializedSize(message.trees, version);
return size;
}
}

4
src/java/org/apache/cassandra/service/ActiveRepairService.java
View File

@ -103,7 +103,7 @@ public class ActiveRepairService
* @return Future for asynchronous call or null if there is no need to repair
*/
public RepairSession submitRepairSession(UUID parentRepairSession,
Range<Token> range,
Collection<Range<Token>> range,
String keyspace,
RepairParallelism parallelismDegree,
Set<InetAddress> endpoints,
@ -383,7 +383,7 @@ public class ActiveRepairService
{
case VALIDATION_COMPLETE:
ValidationComplete validation = (ValidationComplete) message;
session.validationComplete(desc, endpoint, validation.tree);
session.validationComplete(desc, endpoint, validation.trees);
break;
case SYNC_COMPLETE:
// one of replica is synced.

434
src/java/org/apache/cassandra/utils/MerkleTrees.java Normal file
View File

@ -0,0 +1,434 @@
/*
* 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.utils;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.*;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.PeekingIterator;
import org.slf4j.Logger;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
/**
* Wrapper class for handling of multiple MerkleTrees at once.
*
* The MerkleTree's are divided in Ranges of non-overlapping tokens.
*/
public class MerkleTrees implements Iterable<Map.Entry<Range<Token>, MerkleTree>>
{
public static final MerkleTreesSerializer serializer = new MerkleTreesSerializer();
private Map<Range<Token>, MerkleTree> merkleTrees = new TreeMap<>(new TokenRangeComparator());
private IPartitioner partitioner;
/**
* Creates empty MerkleTrees object.
*
* @param partitioner The partitioner to use
*/
public MerkleTrees(IPartitioner partitioner)
{
this(partitioner, new ArrayList<>());
}
private MerkleTrees(IPartitioner partitioner, Collection<MerkleTree> merkleTrees)
{
this.partitioner = partitioner;
addTrees(merkleTrees);
}
/**
* Get the ranges that these merkle trees covers.
*
* @return
*/
public Collection<Range<Token>> ranges()
{
return merkleTrees.keySet();
}
/**
* Get the partitioner in use.
*
* @return
*/
public IPartitioner partitioner()
{
return partitioner;
}
/**
* Add merkle tree's with the defined maxsize and ranges.
*
* @param maxsize
* @param ranges
*/
public void addMerkleTrees(int maxsize, Collection<Range<Token>> ranges)
{
for (Range<Token> range : ranges)
{
addMerkleTree(maxsize, range);
}
}
/**
* Add a MerkleTree with the defined size and range.
*
* @param maxsize
* @param range
* @return The created merkle tree.
*/
public MerkleTree addMerkleTree(int maxsize, Range<Token> range)
{
return addMerkleTree(maxsize, MerkleTree.RECOMMENDED_DEPTH, range);
}
@VisibleForTesting
public MerkleTree addMerkleTree(int maxsize, byte hashdepth, Range<Token> range)
{
MerkleTree tree = new MerkleTree(partitioner, range, hashdepth, maxsize);
addTree(tree);
return tree;
}
/**
* Get the MerkleTree.Range responsible for the given token.
*
* @param t
* @return
*/
public MerkleTree.TreeRange get(Token t)
{
return getMerkleTree(t).get(t);
}
/**
* Init all MerkleTree's with an even tree distribution.
*/
public void init()
{
for (Range<Token> range : merkleTrees.keySet())
{
init(range);
}
}
/**
* Init a selected MerkleTree with an even tree distribution.
*
* @param range
*/
public void init(Range<Token> range)
{
merkleTrees.get(range).init();
}
/**
* Split the MerkleTree responsible for the given token.
*
* @param t
* @return
*/
public boolean split(Token t)
{
return getMerkleTree(t).split(t);
}
/**
* Invalidate the MerkleTree responsible for the given token.
*
* @param t
*/
public void invalidate(Token t)
{
getMerkleTree(t).invalidate(t);
}
/**
* Get the MerkleTree responsible for the given token range.
*
* @param range
* @return
*/
public MerkleTree getMerkleTree(Range<Token> range)
{
return merkleTrees.get(range);
}
public long size()
{
long size = 0;
for (MerkleTree tree : merkleTrees.values())
{
size += tree.size();
}
return size;
}
@VisibleForTesting
public void maxsize(Range<Token> range, int maxsize)
{
getMerkleTree(range).maxsize(maxsize);
}
/**
* Get the MerkleTree responsible for the given token.
*
* @param t
* @return The given MerkleTree or null if none exist.
*/
private MerkleTree getMerkleTree(Token t)
{
for (Range<Token> range : merkleTrees.keySet())
{
if (range.contains(t))
return merkleTrees.get(range);
}
return null;
}
private void addTrees(Collection<MerkleTree> trees)
{
for (MerkleTree tree : trees)
{
addTree(tree);
}
}
private void addTree(MerkleTree tree)
{
assert validateNonOverlapping(tree) : "Range [" + tree.fullRange + "] is intersecting an existing range";
merkleTrees.put(tree.fullRange, tree);
}
private boolean validateNonOverlapping(MerkleTree tree)
{
for (Range<Token> range : merkleTrees.keySet())
{
if (tree.fullRange.intersects(range))
return false;
}
return true;
}
/**
* Get an iterator for all the invalids generated by the MerkleTrees.
*
* @return
*/
public TreeRangeIterator invalids()
{
return new TreeRangeIterator();
}
/**
* Log the row count per leaf for all MerkleTrees.
*
* @param logger
*/
public void logRowCountPerLeaf(Logger logger)
{
for (MerkleTree tree : merkleTrees.values())
{
tree.histogramOfRowCountPerLeaf().log(logger);
}
}
/**
* Log the row size per leaf for all MerkleTrees.
*
* @param logger
*/
public void logRowSizePerLeaf(Logger logger)
{
for (MerkleTree tree : merkleTrees.values())
{
tree.histogramOfRowSizePerLeaf().log(logger);
}
}
@VisibleForTesting
public byte[] hash(Range<Token> range)
{
ByteArrayOutputStream baos = new ByteArrayOutputStream();
boolean hashed = false;
try
{
for (Range<Token> rt : merkleTrees.keySet())
{
if (rt.intersects(range))
{
byte[] bytes = merkleTrees.get(rt).hash(range);
if (bytes != null)
{
baos.write(bytes);
hashed = true;
}
}
}
}
catch (IOException e)
{
throw new RuntimeException("Unable to append merkle tree hash to result");
}
return hashed ? baos.toByteArray() : null;
}
/**
* Get an iterator of all ranges and their MerkleTrees.
*/
public Iterator<Map.Entry<Range<Token>, MerkleTree>> iterator()
{
return merkleTrees.entrySet().iterator();
}
public class TreeRangeIterator extends AbstractIterator<MerkleTree.TreeRange> implements
Iterable<MerkleTree.TreeRange>,
PeekingIterator<MerkleTree.TreeRange>
{
private final Iterator<MerkleTree> it;
private MerkleTree.TreeRangeIterator current = null;
private TreeRangeIterator()
{
it = merkleTrees.values().iterator();
}
public MerkleTree.TreeRange computeNext()
{
if (current == null || !current.hasNext())
return nextIterator();
return current.next();
}
private MerkleTree.TreeRange nextIterator()
{
if (it.hasNext())
{
current = it.next().invalids();
return current.next();
}
return endOfData();
}
public Iterator<MerkleTree.TreeRange> iterator()
{
return this;
}
}
/**
* Get the differences between the two sets of MerkleTrees.
*
* @param ltree
* @param rtree
* @return
*/
public static List<Range<Token>> difference(MerkleTrees ltree, MerkleTrees rtree)
{
List<Range<Token>> differences = new ArrayList<>();
for (MerkleTree tree : ltree.merkleTrees.values())
{
differences.addAll(MerkleTree.difference(tree, rtree.getMerkleTree(tree.fullRange)));
}
return differences;
}
public static class MerkleTreesSerializer implements IVersionedSerializer<MerkleTrees>
{
public void serialize(MerkleTrees trees, DataOutputPlus out, int version) throws IOException
{
out.writeInt(trees.merkleTrees.size());
for (MerkleTree tree : trees.merkleTrees.values())
{
MerkleTree.serializer.serialize(tree, out, version);
}
}
public MerkleTrees deserialize(DataInputPlus in, int version) throws IOException
{
IPartitioner partitioner = null;
int nTrees = in.readInt();
Collection<MerkleTree> trees = new ArrayList<>(nTrees);
if (nTrees > 0)
{
for (int i = 0; i < nTrees; i++)
{
MerkleTree tree = MerkleTree.serializer.deserialize(in, version);
trees.add(tree);
if (partitioner == null)
partitioner = tree.partitioner();
else
assert tree.partitioner() == partitioner;
}
}
return new MerkleTrees(partitioner, trees);
}
public long serializedSize(MerkleTrees trees, int version)
{
assert trees != null;
long size = TypeSizes.sizeof(trees.merkleTrees.size());
for (MerkleTree tree : trees.merkleTrees.values())
{
size += MerkleTree.serializer.serializedSize(tree, version);
}
return size;
}
}
private static class TokenRangeComparator implements Comparator<Range<Token>>
{
@Override
public int compare(Range<Token> rt1, Range<Token> rt2)
{
if (rt1.left.compareTo(rt2.left) == 0)
return 0;
return rt1.compareTo(rt2);
}
}
}

BIN
test/data/serialization/3.0/gms.EndpointState.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/3.0/gms.Gossip.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/3.0/service.SyncComplete.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/3.0/service.SyncRequest.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/3.0/service.ValidationComplete.bin Normal file
View File

Binary file not shown.

BIN
test/data/serialization/3.0/service.ValidationRequest.bin Normal file
View File

Binary file not shown.

3
test/unit/org/apache/cassandra/AbstractSerializationsTester.java
View File

@ -37,7 +37,7 @@ import java.util.Map;
public class AbstractSerializationsTester
{
protected static final String CUR_VER = System.getProperty("cassandra.version", "2.1");
protected static final String CUR_VER = System.getProperty("cassandra.version", "3.0");
protected static final Map<String, Integer> VERSION_MAP = new HashMap<String, Integer> ()
{{
put("0.7", 1);
@ -46,6 +46,7 @@ public class AbstractSerializationsTester
put("2.0", MessagingService.VERSION_20);
put("2.1", MessagingService.VERSION_21);
put("2.2", MessagingService.VERSION_22);
put("3.0", MessagingService.VERSION_30);
}};
protected static final boolean EXECUTE_WRITES = Boolean.getBoolean("cassandra.test-serialization-writes");

2
test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java
View File

@ -173,7 +173,7 @@ public class LeveledCompactionStrategyTest
int gcBefore = keyspace.getColumnFamilyStore(CF_STANDARDDLEVELED).gcBefore(FBUtilities.nowInSeconds());
UUID parentRepSession = UUID.randomUUID();
ActiveRepairService.instance.registerParentRepairSession(parentRepSession, Arrays.asList(cfs), Arrays.asList(range), false);
RepairJobDesc desc = new RepairJobDesc(parentRepSession, UUID.randomUUID(), KEYSPACE1, CF_STANDARDDLEVELED, range);
RepairJobDesc desc = new RepairJobDesc(parentRepSession, UUID.randomUUID(), KEYSPACE1, CF_STANDARDDLEVELED, Arrays.asList(range));
Validator validator = new Validator(desc, FBUtilities.getBroadcastAddress(), gcBefore);
CompactionManager.instance.submitValidation(cfs, validator).get();
}

23
test/unit/org/apache/cassandra/repair/LocalSyncTaskTest.java
View File

@ -20,13 +20,14 @@ package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
@ -37,6 +38,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import static org.junit.Assert.assertEquals;
@ -65,10 +67,11 @@ public class LocalSyncTaskTest extends SchemaLoader
final InetAddress ep2 = InetAddress.getByName("127.0.0.1");
Range<Token> range = new Range<>(partirioner.getMinimumToken(), partirioner.getRandomToken());
RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), KEYSPACE1, "Standard1", range);
RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), KEYSPACE1, "Standard1", Arrays.asList(range));
MerkleTree tree1 = createInitialTree(desc);
MerkleTree tree2 = createInitialTree(desc);
MerkleTrees tree1 = createInitialTree(desc);
MerkleTrees tree2 = createInitialTree(desc);
// difference the trees
// note: we reuse the same endpoint which is bogus in theory but fine here
@ -90,10 +93,11 @@ public class LocalSyncTaskTest extends SchemaLoader
ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, Arrays.asList(cfs), Arrays.asList(range), false);
RepairJobDesc desc = new RepairJobDesc(parentRepairSession, UUID.randomUUID(), KEYSPACE1, "Standard1", range);
RepairJobDesc desc = new RepairJobDesc(parentRepairSession, UUID.randomUUID(), KEYSPACE1, "Standard1", Arrays.asList(range));
MerkleTree tree1 = createInitialTree(desc);
MerkleTree tree2 = createInitialTree(desc);
MerkleTrees tree1 = createInitialTree(desc);
MerkleTrees tree2 = createInitialTree(desc);
// change a range in one of the trees
Token token = partirioner.midpoint(range.left, range.right);
@ -115,9 +119,10 @@ public class LocalSyncTaskTest extends SchemaLoader
assertEquals("Wrong differing ranges", interesting.size(), task.getCurrentStat().numberOfDifferences);
}
private MerkleTree createInitialTree(RepairJobDesc desc)
private MerkleTrees createInitialTree(RepairJobDesc desc)
{
MerkleTree tree = new MerkleTree(partirioner, desc.range, MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15));
MerkleTrees tree = new MerkleTrees(partirioner);
tree.addMerkleTrees((int) Math.pow(2, 15), desc.ranges);
tree.init();
for (MerkleTree.TreeRange r : tree.invalids())
{

3
test/unit/org/apache/cassandra/repair/RepairSessionTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.repair;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
@ -53,7 +54,7 @@ public class RepairSessionTest
IPartitioner p = Murmur3Partitioner.instance;
Range<Token> repairRange = new Range<>(p.getToken(ByteBufferUtil.bytes(0)), p.getToken(ByteBufferUtil.bytes(100)));
Set<InetAddress> endpoints = Sets.newHashSet(remote);
RepairSession session = new RepairSession(parentSessionId, sessionId, repairRange, "Keyspace1", RepairParallelism.SEQUENTIAL, endpoints, ActiveRepairService.UNREPAIRED_SSTABLE, "Standard1");
RepairSession session = new RepairSession(parentSessionId, sessionId, Arrays.asList(repairRange), "Keyspace1", RepairParallelism.SEQUENTIAL, endpoints, ActiveRepairService.UNREPAIRED_SSTABLE, "Standard1");
// perform convict
session.convict(remote, Double.MAX_VALUE);

19
test/unit/org/apache/cassandra/repair/ValidatorTest.java
View File

@ -19,6 +19,9 @@
package org.apache.cassandra.repair;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.junit.After;
@ -43,6 +46,7 @@ import org.apache.cassandra.repair.messages.ValidationComplete;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static org.junit.Assert.assertEquals;
@ -77,7 +81,7 @@ public class ValidatorTest
public void testValidatorComplete() throws Throwable
{
Range<Token> range = new Range<>(partitioner.getMinimumToken(), partitioner.getRandomToken());
final RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), keyspace, columnFamily, range);
final RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), keyspace, columnFamily, Arrays.asList(range));
final SimpleCondition lock = new SimpleCondition();
MessagingService.instance().addMessageSink(new IMessageSink()
@ -91,8 +95,8 @@ public class ValidatorTest
RepairMessage m = (RepairMessage) message.payload;
assertEquals(RepairMessage.Type.VALIDATION_COMPLETE, m.messageType);
assertEquals(desc, m.desc);
assertTrue(((ValidationComplete) m).success);
assertNotNull(((ValidationComplete) m).tree);
assertTrue(((ValidationComplete) m).success());
assertNotNull(((ValidationComplete) m).trees);
}
}
finally
@ -113,7 +117,8 @@ public class ValidatorTest
ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(columnFamily);
Validator validator = new Validator(desc, remote, 0);
MerkleTree tree = new MerkleTree(partitioner, validator.desc.range, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, 15));
MerkleTrees tree = new MerkleTrees(partitioner);
tree.addMerkleTrees((int) Math.pow(2, 15), validator.desc.ranges);
validator.prepare(cfs, tree);
// and confirm that the tree was split
@ -137,7 +142,7 @@ public class ValidatorTest
public void testValidatorFailed() throws Throwable
{
Range<Token> range = new Range<>(partitioner.getMinimumToken(), partitioner.getRandomToken());
final RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), keyspace, columnFamily, range);
final RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), UUID.randomUUID(), keyspace, columnFamily, Arrays.asList(range));
final SimpleCondition lock = new SimpleCondition();
MessagingService.instance().addMessageSink(new IMessageSink()
@ -151,8 +156,8 @@ public class ValidatorTest
RepairMessage m = (RepairMessage) message.payload;
assertEquals(RepairMessage.Type.VALIDATION_COMPLETE, m.messageType);
assertEquals(desc, m.desc);
assertFalse(((ValidationComplete) m).success);
assertNull(((ValidationComplete) m).tree);
assertFalse(((ValidationComplete) m).success());
assertNull(((ValidationComplete) m).trees);
}
}
finally

28
test/unit/org/apache/cassandra/service/SerializationsTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.service;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.UUID;
@ -43,7 +44,7 @@ import org.apache.cassandra.repair.RepairJobDesc;
import org.apache.cassandra.repair.Validator;
import org.apache.cassandra.repair.messages.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MerkleTree;
import org.apache.cassandra.utils.MerkleTrees;
public class SerializationsTest extends AbstractSerializationsTester
{
@ -58,7 +59,7 @@ public class SerializationsTest extends AbstractSerializationsTester
partitionerSwitcher = Util.switchPartitioner(RandomPartitioner.instance);
RANDOM_UUID = UUID.fromString("b5c3d033-75aa-4c2f-a819-947aac7a0c54");
FULL_RANGE = new Range<>(Util.testPartitioner().getMinimumToken(), Util.testPartitioner().getMinimumToken());
DESC = new RepairJobDesc(getVersion() < MessagingService.VERSION_21 ? null : RANDOM_UUID, RANDOM_UUID, "Keyspace1", "Standard1", FULL_RANGE);
DESC = new RepairJobDesc(getVersion() < MessagingService.VERSION_21 ? null : RANDOM_UUID, RANDOM_UUID, "Keyspace1", "Standard1", Arrays.asList(FULL_RANGE));
}
@AfterClass
@ -66,8 +67,7 @@ public class SerializationsTest extends AbstractSerializationsTester
{
partitionerSwitcher.close();
}
private void testRepairMessageWrite(String fileName, RepairMessage... messages) throws IOException
{
try (DataOutputStreamPlus out = getOutput(fileName))
@ -109,13 +109,17 @@ public class SerializationsTest extends AbstractSerializationsTester
private void testValidationCompleteWrite() throws IOException
{
IPartitioner p = RandomPartitioner.instance;
MerkleTrees mt = new MerkleTrees(p);
// empty validation
MerkleTree mt = new MerkleTree(p, FULL_RANGE, MerkleTree.RECOMMENDED_DEPTH, (int) Math.pow(2, 15));
mt.addMerkleTree((int) Math.pow(2, 15), FULL_RANGE);
Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1);
ValidationComplete c0 = new ValidationComplete(DESC, mt);
// validation with a tree
mt = new MerkleTree(p, FULL_RANGE, MerkleTree.RECOMMENDED_DEPTH, Integer.MAX_VALUE);
mt = new MerkleTrees(p);
mt.addMerkleTree(Integer.MAX_VALUE, FULL_RANGE);
for (int i = 0; i < 10; i++)
mt.split(p.getRandomToken());
Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddress(), -1);
@ -140,24 +144,24 @@ public class SerializationsTest extends AbstractSerializationsTester
assert message.messageType == RepairMessage.Type.VALIDATION_COMPLETE;
assert DESC.equals(message.desc);
assert ((ValidationComplete) message).success;
assert ((ValidationComplete) message).tree != null;
assert ((ValidationComplete) message).success();
assert ((ValidationComplete) message).trees != null;
// validation with a tree
message = RepairMessage.serializer.deserialize(in, getVersion());
assert message.messageType == RepairMessage.Type.VALIDATION_COMPLETE;
assert DESC.equals(message.desc);
assert ((ValidationComplete) message).success;
assert ((ValidationComplete) message).tree != null;
assert ((ValidationComplete) message).success();
assert ((ValidationComplete) message).trees != null;
// failed validation
message = RepairMessage.serializer.deserialize(in, getVersion());
assert message.messageType == RepairMessage.Type.VALIDATION_COMPLETE;
assert DESC.equals(message.desc);
assert !((ValidationComplete) message).success;
assert ((ValidationComplete) message).tree == null;
assert !((ValidationComplete) message).success();
assert ((ValidationComplete) message).trees == null;
// MessageOuts
for (int i = 0; i < 3; i++)

1
test/unit/org/apache/cassandra/utils/MerkleTreeTest.java
View File

@ -34,7 +34,6 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.NIODataInputStream;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.MerkleTree.Hashable;
import org.apache.cassandra.utils.MerkleTree.RowHash;

538
test/unit/org/apache/cassandra/utils/MerkleTreesTest.java Normal file
View File

@ -0,0 +1,538 @@
/*
* 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 copyten 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.utils;
import java.math.BigInteger;
import java.util.*;
import com.google.common.collect.AbstractIterator;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.dht.RandomPartitioner.BigIntegerToken;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.MerkleTree.Hashable;
import org.apache.cassandra.utils.MerkleTree.RowHash;
import org.apache.cassandra.utils.MerkleTree.TreeRange;
import org.apache.cassandra.utils.MerkleTrees.TreeRangeIterator;
import static org.junit.Assert.*;
public class MerkleTreesTest
{
public static byte[] DUMMY = "blah".getBytes();
/**
* If a test assumes that the tree is 8 units wide, then it should set this value
* to 8.
*/
public static BigInteger TOKEN_SCALE = new BigInteger("8");
protected static final IPartitioner partitioner = RandomPartitioner.instance;
protected MerkleTrees mts;
private Range<Token> fullRange()
{
return new Range<>(partitioner.getMinimumToken(), partitioner.getMinimumToken());
}
@BeforeClass
public static void setUp()
{
StorageService.instance.setPartitionerUnsafe(partitioner);
}
@Before
public void clear()
{
TOKEN_SCALE = new BigInteger("8");
mts = new MerkleTrees(partitioner);
mts.addMerkleTree(Integer.MAX_VALUE, fullRange());
}
public static void assertHashEquals(final byte[] left, final byte[] right)
{
assertHashEquals("", left, right);
}
public static void assertHashEquals(String message, final byte[] left, final byte[] right)
{
String lstring = left == null ? "null" : Hex.bytesToHex(left);
String rstring = right == null ? "null" : Hex.bytesToHex(right);
assertEquals(message, lstring, rstring);
}
/**
* The value returned by this method is affected by TOKEN_SCALE: setting TOKEN_SCALE
* to 8 means that passing -1 through 8 for this method will return values mapped
* between -1 and Token.MAX_VALUE.
*/
public static Token tok(int i)
{
if (i == -1)
return new BigIntegerToken(new BigInteger("-1"));
BigInteger bint = RandomPartitioner.MAXIMUM.divide(TOKEN_SCALE).multiply(new BigInteger(""+i));
return new BigIntegerToken(bint);
}
@Test
public void testIntersectingRanges()
{
mts = new MerkleTrees(partitioner);
boolean failure = true;
mts.addMerkleTree(1, new Range<>(tok(1), tok(3)));
try
{
mts.addMerkleTree(1, new Range<>(tok(2), tok(4)));
}
catch (AssertionError e)
{
failure = false;
}
assertFalse(failure);
}
@Test
public void testSplit()
{
// split the range (zero, zero] into:
// (zero,four], (four,six], (six,seven] and (seven, zero]
mts.split(tok(4));
mts.split(tok(6));
mts.split(tok(7));
assertEquals(4, mts.size());
assertEquals(new Range<>(tok(7), tok(-1)), mts.get(tok(-1)));
assertEquals(new Range<>(tok(-1), tok(4)), mts.get(tok(3)));
assertEquals(new Range<>(tok(-1), tok(4)), mts.get(tok(4)));
assertEquals(new Range<>(tok(4), tok(6)), mts.get(tok(6)));
assertEquals(new Range<>(tok(6), tok(7)), mts.get(tok(7)));
// check depths
assertEquals((byte) 1, mts.get(tok(4)).depth);
assertEquals((byte) 2, mts.get(tok(6)).depth);
assertEquals((byte) 3, mts.get(tok(7)).depth);
assertEquals((byte) 3, mts.get(tok(-1)).depth);
try
{
mts.split(tok(-1));
fail("Shouldn't be able to split outside the initial range.");
}
catch (AssertionError e)
{
// pass
}
}
@Test
public void testSplitLimitDepth()
{
mts = new MerkleTrees(partitioner);
mts.addMerkleTree(Integer.MAX_VALUE, (byte) 2, fullRange());
assertTrue(mts.split(tok(4)));
assertTrue(mts.split(tok(2)));
assertEquals(3, mts.size());
// should fail to split below hashdepth
assertFalse(mts.split(tok(1)));
assertEquals(3, mts.size());
assertEquals(new Range<>(tok(4), tok(-1)), mts.get(tok(-1)));
assertEquals(new Range<>(tok(-1), tok(2)), mts.get(tok(2)));
assertEquals(new Range<>(tok(2), tok(4)), mts.get(tok(4)));
}
@Test
public void testSplitLimitSize()
{
mts = new MerkleTrees(partitioner);
mts.addMerkleTree(2, fullRange());
assertTrue(mts.split(tok(4)));
assertEquals(2, mts.size());
// should fail to split above maxsize
assertFalse(mts.split(tok(2)));
assertEquals(2, mts.size());
assertEquals(new Range<>(tok(4), tok(-1)), mts.get(tok(-1)));
assertEquals(new Range<>(tok(-1), tok(4)), mts.get(tok(4)));
}
@Test
public void testInvalids()
{
Iterator<TreeRange> ranges;
// (zero, zero]
ranges = mts.invalids();
assertEquals(new Range<>(tok(-1), tok(-1)), ranges.next());
assertFalse(ranges.hasNext());
// all invalid
mts.split(tok(4));
mts.split(tok(2));
mts.split(tok(6));
mts.split(tok(3));
mts.split(tok(5));
ranges = mts.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());
assertEquals(new Range<>(tok(4), tok(5)), ranges.next());
assertEquals(new Range<>(tok(5), tok(6)), ranges.next());
assertEquals(new Range<>(tok(6), tok(-1)), ranges.next());
assertFalse(ranges.hasNext());
}
@Test
public void testHashFull()
{
byte[] val = DUMMY;
Range<Token> range = new Range<>(tok(-1), tok(-1));
// (zero, zero]
assertNull(mts.hash(range));
// validate the range
mts.get(tok(-1)).hash(val);
assertHashEquals(val, mts.hash(range));
}
@Test
public void testHashPartial()
{
byte[] val = DUMMY;
byte[] leftval = hashed(val, 1, 1);
byte[] partialval = hashed(val, 1);
Range<Token> left = new Range<>(tok(-1), tok(4));
Range<Token> partial = new Range<>(tok(2), tok(4));
Range<Token> right = new Range<>(tok(4), tok(-1));
Range<Token> linvalid = new Range<>(tok(1), tok(4));
Range<Token> rinvalid = new Range<>(tok(4), tok(6));
// (zero,two] (two,four] (four, zero]
mts.split(tok(4));
mts.split(tok(2));
assertNull(mts.hash(left));
assertNull(mts.hash(partial));
assertNull(mts.hash(right));
assertNull(mts.hash(linvalid));
assertNull(mts.hash(rinvalid));
// validate the range
mts.get(tok(2)).hash(val);
mts.get(tok(4)).hash(val);
mts.get(tok(-1)).hash(val);
assertHashEquals(leftval, mts.hash(left));
assertHashEquals(partialval, mts.hash(partial));
assertHashEquals(val, mts.hash(right));
assertNull(mts.hash(linvalid));
assertNull(mts.hash(rinvalid));
}
@Test
public void testHashInner()
{
byte[] val = DUMMY;
byte[] lchildval = hashed(val, 3, 3, 2);
byte[] rchildval = hashed(val, 2, 2);
byte[] fullval = hashed(val, 3, 3, 2, 2, 2);
Range<Token> full = new Range<>(tok(-1), tok(-1));
Range<Token> lchild = new Range<>(tok(-1), tok(4));
Range<Token> rchild = new Range<>(tok(4), tok(-1));
Range<Token> invalid = new Range<>(tok(1), tok(-1));
// (zero,one] (one, two] (two,four] (four, six] (six, zero]
mts.split(tok(4));
mts.split(tok(2));
mts.split(tok(6));
mts.split(tok(1));
assertNull(mts.hash(full));
assertNull(mts.hash(lchild));
assertNull(mts.hash(rchild));
assertNull(mts.hash(invalid));
// validate the range
mts.get(tok(1)).hash(val);
mts.get(tok(2)).hash(val);
mts.get(tok(4)).hash(val);
mts.get(tok(6)).hash(val);
mts.get(tok(-1)).hash(val);
assertHashEquals(fullval, mts.hash(full));
assertHashEquals(lchildval, mts.hash(lchild));
assertHashEquals(rchildval, mts.hash(rchild));
assertNull(mts.hash(invalid));
}
@Test
public void testHashDegenerate()
{
TOKEN_SCALE = new BigInteger("32");
byte[] val = DUMMY;
byte[] childfullval = hashed(val, 5, 5, 4);
byte[] fullval = hashed(val, 5, 5, 4, 3, 2, 1);
Range<Token> childfull = new Range<>(tok(-1), tok(4));
Range<Token> full = new Range<>(tok(-1), tok(-1));
Range<Token> invalid = new Range<>(tok(4), tok(-1));
mts.split(tok(16));
mts.split(tok(8));
mts.split(tok(4));
mts.split(tok(2));
mts.split(tok(1));
assertNull(mts.hash(full));
assertNull(mts.hash(childfull));
assertNull(mts.hash(invalid));
// validate the range
mts.get(tok(1)).hash(val);
mts.get(tok(2)).hash(val);
mts.get(tok(4)).hash(val);
mts.get(tok(8)).hash(val);
mts.get(tok(16)).hash(val);
mts.get(tok(-1)).hash(val);
assertHashEquals(fullval, mts.hash(full));
assertHashEquals(childfullval, mts.hash(childfull));
assertNull(mts.hash(invalid));
}
@Test
public void testHashRandom()
{
int max = 1000000;
TOKEN_SCALE = new BigInteger("" + max);
mts = new MerkleTrees(partitioner);
mts.addMerkleTree(32, fullRange());
Random random = new Random();
while (true)
{
if (!mts.split(tok(random.nextInt(max))))
break;
}
// validate the tree
TreeRangeIterator ranges = mts.invalids();
for (TreeRange range : ranges)
range.addHash(new RowHash(range.right, new byte[0], 0));
assert mts.hash(new Range<>(tok(-1), tok(-1))) != null : "Could not hash tree " + mts;
}
/**
* Generate two trees with different splits, but containing the same keys, and
* check that they compare equally.
*
* The set of keys used in this test is: #{2,4,6,8,12,14,0}
*/
@Test
public void testValidateTree()
{
TOKEN_SCALE = new BigInteger("16"); // this test needs slightly more resolution
Range<Token> full = new Range<>(tok(-1), tok(-1));
Iterator<TreeRange> ranges;
MerkleTrees mts2 = new MerkleTrees(partitioner);
mts2.addMerkleTree(Integer.MAX_VALUE, fullRange());
mts.split(tok(8));
mts.split(tok(4));
mts.split(tok(12));
mts.split(tok(6));
mts.split(tok(10));
ranges = mts.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]
ranges.next().addAll(new HIterator(/*empty*/ new int[0])); // (8,10]
ranges.next().addAll(new HIterator(12)); // (10,12]
ranges.next().addAll(new HIterator(14, -1)); // (12,-1]: depth 2
mts2.split(tok(8));
mts2.split(tok(4));
mts2.split(tok(12));
mts2.split(tok(2));
mts2.split(tok(10));
mts2.split(tok(9));
mts2.split(tok(11));
ranges = mts2.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
ranges.next().addAll(new HIterator(/*empty*/ new int[0])); // (8,9]
ranges.next().addAll(new HIterator(/*empty*/ new int[0])); // (9,10]
ranges.next().addAll(new HIterator(/*empty*/ new int[0])); // (10,11]: depth 4
ranges.next().addAll(new HIterator(12)); // (11,12]: depth 4
ranges.next().addAll(new HIterator(14, -1)); // (12,-1]: depth 2
byte[] mthash = mts.hash(full);
byte[] mt2hash = mts2.hash(full);
assertHashEquals("Tree hashes did not match: " + mts + " && " + mts2, mthash, mt2hash);
}
@Test
public void testSerialization() throws Exception
{
Range<Token> first = new Range<>(tok(3), tok(4));
Collection<Range<Token>> ranges = new ArrayList<>();
ranges.add(first);
ranges.add(new Range<Token>(tok(5), tok(2)));
mts = new MerkleTrees(partitioner);
mts.addMerkleTrees(256, ranges);
// populate and validate the tree
mts.init();
for (TreeRange range : mts.invalids())
range.addAll(new HIterator(range.right));
byte[] initialhash = mts.hash(first);
long serializedSize = MerkleTrees.serializer.serializedSize(mts, MessagingService.current_version);
DataOutputBuffer out = new DataOutputBuffer();
MerkleTrees.serializer.serialize(mts, out, MessagingService.current_version);
byte[] serialized = out.toByteArray();
assertEquals(serializedSize, serialized.length);
DataInputBuffer in = new DataInputBuffer(serialized);
MerkleTrees restored = MerkleTrees.serializer.deserialize(in, MessagingService.current_version);
assertHashEquals(initialhash, restored.hash(first));
}
@Test
public void testDifference()
{
int maxsize = 16;
mts = new MerkleTrees(partitioner);
mts.addMerkleTree(32, fullRange());
MerkleTrees mts2 = new MerkleTrees(partitioner);
mts2.addMerkleTree(32, fullRange());
mts.init();
mts2.init();
// add dummy hashes to both trees
for (TreeRange range : mts.invalids())
range.addAll(new HIterator(range.right));
for (TreeRange range : mts2.invalids())
range.addAll(new HIterator(range.right));
TreeRange leftmost = null;
TreeRange middle = null;
mts.maxsize(fullRange(), maxsize + 2); // give some room for splitting
// split the leftmost
Iterator<TreeRange> ranges = mts.invalids();
leftmost = ranges.next();
mts.split(leftmost.right);
// set the hashes for the leaf of the created split
middle = mts.get(leftmost.right);
middle.hash("arbitrary!".getBytes());
mts.get(partitioner.midpoint(leftmost.left, leftmost.right)).hash("even more arbitrary!".getBytes());
// trees should disagree for (leftmost.left, middle.right]
List<Range<Token>> diffs = MerkleTrees.difference(mts, mts2);
assertEquals(diffs + " contains wrong number of differences:", 1, diffs.size());
assertTrue(diffs.contains(new Range<>(leftmost.left, middle.right)));
}
/**
* Return the root hash of a binary tree with leaves at the given depths
* and with the given hash val in each leaf.
*/
byte[] hashed(byte[] val, Integer... depths)
{
ArrayDeque<Integer> dstack = new ArrayDeque<Integer>();
ArrayDeque<byte[]> hstack = new ArrayDeque<byte[]>();
Iterator<Integer> depthiter = Arrays.asList(depths).iterator();
if (depthiter.hasNext())
{
dstack.push(depthiter.next());
hstack.push(val);
}
while (depthiter.hasNext())
{
Integer depth = depthiter.next();
byte[] hash = val;
while (depth.equals(dstack.peek()))
{
// consume the stack
hash = Hashable.binaryHash(hstack.pop(), hash);
depth = dstack.pop()-1;
}
dstack.push(depth);
hstack.push(hash);
}
assert hstack.size() == 1;
return hstack.pop();
}
static class HIterator extends AbstractIterator<RowHash>
{
private Iterator<Token> tokens;
public HIterator(int... tokens)
{
List<Token> tlist = new LinkedList<Token>();
for (int token : tokens)
tlist.add(tok(token));
this.tokens = tlist.iterator();
}
public HIterator(Token... tokens)
{
this.tokens = Arrays.asList(tokens).iterator();
}
public RowHash computeNext()
{
if (tokens.hasNext())
return new RowHash(tokens.next(), DUMMY, DUMMY.length);
return endOfData();
}
}
}