import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private MerkleTree.RowHash rowHash(AbstractCompactedRow row)
{
    validated++;
    // MerkleTree uses XOR internally, so we want lots of output bits here
    CountingDigest digest = new CountingDigest(FBUtilities.newMessageDigest("SHA-256"));
    row.update(digest);
    // only return new hash for merkle tree in case digest was updated - see CASSANDRA-8979
    if (digest.count > 0)
    {
        return new MerkleTree.RowHash(row.key.getToken(), digest.digest(), digest.count);
    }
    else
    {
        return null;
    }
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
/**
 * Add a new received tree and return the number of remaining tree to
 * be received for the job to be complete.
 *
 * Callers may assume exactly one addTree call will result in zero remaining endpoints.
 *
 * @param endpoint address of the endpoint that sent response
 * @param tree sent Merkle tree or null if validation failed on endpoint
 * @return the number of responses waiting to receive
 */
public synchronized int addTree(InetAddress endpoint, MerkleTree tree)
{
    // Wait for all request to have been performed (see #3400)
    try
    {
        requestsSent.await();
    }
    catch (InterruptedException e)
    {
        throw new AssertionError("Interrupted while waiting for requests to be sent");
    }

    if (tree == null)
        failed = true;
    else
        trees.add(new TreeResponse(endpoint, tree));
    return treeRequests.completed(endpoint);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
/**
 * Compares our trees, and triggers repairs for any ranges that mismatch.
 */
public void run()
{
    // compare trees, and collect differences
    differences.addAll(MerkleTree.difference(r1.tree, r2.tree));

    // choose a repair method based on the significance of the difference
    String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
    if (differences.isEmpty())
    {
        logger.info(String.format(format, "are consistent"));
        // send back sync complete message
        MessagingService.instance().sendOneWay(new SyncComplete(desc, r1.endpoint, r2.endpoint, true).createMessage(), FBUtilities.getLocalAddress());
        return;
    }

    // non-0 difference: perform streaming repair
    logger.info(String.format(format, "have " + differences.size() + " range(s) out of sync"));
    performStreamingRepair();
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private void testValidationCompleteWrite() throws IOException
{
    // empty validation
    Validator v0 = new Validator(DESC, FBUtilities.getBroadcastAddress(),  -1);
    ValidationComplete c0 = new ValidationComplete(DESC, v0.tree);

    // validation with a tree
    IPartitioner p = new RandomPartitioner();
    MerkleTree mt = new MerkleTree(p, FULL_RANGE, MerkleTree.RECOMMENDED_DEPTH, Integer.MAX_VALUE);
    for (int i = 0; i < 10; i++)
        mt.split(p.getRandomToken());
    Validator v1 = new Validator(DESC, FBUtilities.getBroadcastAddress(), mt, -1);
    ValidationComplete c1 = new ValidationComplete(DESC, v1.tree);

    // validation failed
    ValidationComplete c3 = new ValidationComplete(DESC);

    testRepairMessageWrite("service.ValidationComplete.bin", c0, c1, c3);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private void testTreeResponseWrite() throws IOException
{
    // empty validation
    AntiEntropyService.Validator v0 = new AntiEntropyService.Validator(Statics.req);

    // validation with a tree
    IPartitioner p = new RandomPartitioner();
    MerkleTree mt = new MerkleTree(p, FULL_RANGE, MerkleTree.RECOMMENDED_DEPTH, Integer.MAX_VALUE);
    for (int i = 0; i < 10; i++)
        mt.split(p.getRandomToken());
    AntiEntropyService.Validator v1 = new AntiEntropyService.Validator(Statics.req, mt);

    DataOutputStream out = getOutput("service.TreeResponse.bin");
    AntiEntropyService.Validator.serializer.serialize(v0, out, getVersion());
    AntiEntropyService.Validator.serializer.serialize(v1, out, getVersion());
    v0.createMessage().serialize(out, getVersion());
    v1.createMessage().serialize(out, getVersion());
    out.close();

    // test serializedSize
    testSerializedSize(v0, AntiEntropyService.Validator.serializer);
    testSerializedSize(v1, AntiEntropyService.Validator.serializer);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
/**
 * Compares our trees, and triggers repairs for any ranges that mismatch.
 */
public void run()
{
    // compare trees, and collect differences
    differences.addAll(MerkleTree.difference(r1.tree, r2.tree));

    // choose a repair method based on the significance of the difference
    String format = String.format("[repair #%s] Endpoints %s and %s %%s for %s", desc.sessionId, r1.endpoint, r2.endpoint, desc.columnFamily);
    if (differences.isEmpty())
    {
        logger.info(String.format(format, "are consistent"));
        // send back sync complete message
        MessagingService.instance().sendOneWay(new SyncComplete(desc, r1.endpoint, r2.endpoint, true).createMessage(), FBUtilities.getLocalAddress());
        return;
    }

    // non-0 difference: perform streaming repair
    logger.info(format, "have {} range(s) out of sync", differences.size());
    performStreamingRepair();
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private void testTreeResponseWrite() throws IOException
{
    AntiEntropyService.Validator v0 = new AntiEntropyService.Validator(Statics.req);
    IPartitioner part = new RandomPartitioner();
    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++)
    {
        Token t = part.getRandomToken();
        tokens.add(t);
        mt.split(t);
    }
    AntiEntropyService.Validator v1 = new AntiEntropyService.Validator(Statics.req, mt);
    DataOutputStream out = getOutput("service.TreeResponse.bin");
    AntiEntropyService.TreeResponseVerbHandler.SERIALIZER.serialize(v0, out, getVersion());
    AntiEntropyService.TreeResponseVerbHandler.SERIALIZER.serialize(v1, out, getVersion());
    Message.serializer().serialize(AntiEntropyService.TreeResponseVerbHandler.makeVerb(FBUtilities.getLocalAddress(), v0), out, getVersion());
    Message.serializer().serialize(AntiEntropyService.TreeResponseVerbHandler.makeVerb(FBUtilities.getLocalAddress(), v1), out, getVersion());
    out.close();
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
public void prepare(ColumnFamilyStore cfs, MerkleTree tree)
{
    this.tree = tree;

    if (!tree.partitioner().preservesOrder())
    {
        // You can't beat an even tree distribution for md5
        tree.init();
    }
    else
    {
        List<DecoratedKey> keys = new ArrayList<>();
        for (DecoratedKey sample : cfs.keySamples(desc.range))
        {
            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)
            {
                DecoratedKey dk = keys.get(random.nextInt(numkeys));
                if (!tree.split(dk.getToken()))
                    break;
            }
        }
    }
    logger.debug("Prepared AEService tree of size {} for {}", tree.size(), desc);
    ranges = tree.invalids();
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
public ValidationComplete(RepairJobDesc desc, MerkleTree tree)
{
    super(Type.VALIDATION_COMPLETE, desc);
    assert tree != null;
    this.success = true;
    this.tree = tree;
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
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);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
public ValidationComplete deserialize(DataInput in, int version) throws IOException
{
    RepairJobDesc desc = RepairJobDesc.serializer.deserialize(in, version);
    if (in.readBoolean())
    {
        MerkleTree tree = MerkleTree.serializer.deserialize(in, version);
        return new ValidationComplete(desc, tree);
    }
    else
    {
        return new ValidationComplete(desc);
    }
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
public long serializedSize(ValidationComplete message, int version)
{
    long size = RepairJobDesc.serializer.serializedSize(message.desc, version);
    size += TypeSizes.NATIVE.sizeof(message.success);
    if (message.success)
        size += MerkleTree.serializer.serializedSize(message.tree, version);
    return size;
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
@Test
public void testDifference() throws Throwable
{
    Range<Token> range = new Range<>(partirioner.getMinimumToken(), partirioner.getRandomToken());
    UUID parentRepairSession = UUID.randomUUID();
    Keyspace keyspace = Keyspace.open("Keyspace1");
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Standard1");

    ActiveRepairService.instance.registerParentRepairSession(parentRepairSession, Arrays.asList(cfs), Arrays.asList(range));

    RepairJobDesc desc = new RepairJobDesc(parentRepairSession, UUID.randomUUID(), "Keyspace1", "Standard1", range);

    MerkleTree tree1 = createInitialTree(desc);
    MerkleTree tree2 = createInitialTree(desc);

    // change a range in one of the trees
    Token token = partirioner.midpoint(range.left, range.right);
    tree1.invalidate(token);
    MerkleTree.TreeRange changed = tree1.get(token);
    changed.hash("non-empty hash!".getBytes());

    Set<Range<Token>> interesting = new HashSet<>();
    interesting.add(changed);

    // difference the trees
    // note: we reuse the same endpoint which is bogus in theory but fine here
    TreeResponse r1 = new TreeResponse(InetAddress.getByName("127.0.0.1"), tree1);
    TreeResponse r2 = new TreeResponse(InetAddress.getByName("127.0.0.2"), tree2);
    Differencer diff = new Differencer(desc, r1, r2);
    diff.run();

    // ensure that the changed range was recorded
    assertEquals("Wrong differing ranges", interesting, new HashSet<>(diff.differences));
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private MerkleTree createInitialTree(RepairJobDesc desc)
{
    MerkleTree tree = new MerkleTree(partirioner, desc.range, MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15));
    tree.init();
    for (MerkleTree.TreeRange r : tree.invalids())
    {
        r.ensureHashInitialised();
    }
    return tree;
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
/**
 * Create Validator with default size of initial Merkle Tree.
 */
public Validator(RepairJobDesc desc, InetAddress initiator, int gcBefore)
{
    this(desc,
         initiator,
         // TODO: memory usage (maxsize) should either be tunable per
         // CF, globally, or as shared for all CFs in a cluster
         new MerkleTree(DatabaseDescriptor.getPartitioner(), desc.range, MerkleTree.RECOMMENDED_DEPTH, (int)Math.pow(2, 15)),
         gcBefore);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private MerkleTree.RowHash rowHash(AbstractCompactedRow row)
{
    validated++;
    // MerkleTree uses XOR internally, so we want lots of output bits here
    CountingDigest digest = new CountingDigest(FBUtilities.newMessageDigest("SHA-256"));
    row.update(digest);
    return new MerkleTree.RowHash(row.key.token, digest.digest(), digest.count);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
public void serialize(ValidationComplete message, DataOutput 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);
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
@Test
public void testDifference() throws Throwable
{
    Range<Token> range = new Range<>(partirioner.getMinimumToken(), partirioner.getRandomToken());
    RepairJobDesc desc = new RepairJobDesc(UUID.randomUUID(), "Keyspace1", "Standard1", range);

    MerkleTree tree1 = createInitialTree(desc);
    MerkleTree tree2 = createInitialTree(desc);

    // change a range in one of the trees
    Token token = partirioner.midpoint(range.left, range.right);
    tree1.invalidate(token);
    MerkleTree.TreeRange changed = tree1.get(token);
    changed.hash("non-empty hash!".getBytes());

    Set<Range<Token>> interesting = new HashSet<>();
    interesting.add(changed);

    // difference the trees
    // note: we reuse the same endpoint which is bogus in theory but fine here
    TreeResponse r1 = new TreeResponse(InetAddress.getByName("127.0.0.1"), tree1);
    TreeResponse r2 = new TreeResponse(InetAddress.getByName("127.0.0.2"), tree2);
    Differencer diff = new Differencer(desc, r1, r2);
    diff.run();

    // ensure that the changed range was recorded
    assertEquals("Wrong differing ranges", interesting, new HashSet<>(diff.differences));
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
private MerkleTree.RowHash rowHash(UnfilteredRowIterator partition)
{
    validated++;
    // MerkleTree uses XOR internally, so we want lots of output bits here
    CountingDigest digest = new CountingDigest(FBUtilities.newMessageDigest("SHA-256"));
    UnfilteredRowIterators.digest(null, partition, digest, MessagingService.current_version);
    // only return new hash for merkle tree in case digest was updated - see CASSANDRA-8979
    return digest.count > 0
         ? new MerkleTree.RowHash(partition.partitionKey().getToken(), digest.digest(), digest.count)
         : null;
}
 

import org.apache.cassandra.utils.MerkleTree; //导入依赖的package包/类
@Test
public void testDifference() throws Throwable
{
    Range<Token> range = new Range<>(partirioner.getMinimumToken(), partirioner.getRandomToken());
    UUID parentRepairSession = UUID.randomUUID();
    Keyspace keyspace = Keyspace.open(KEYSPACE1);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("Standard1");

    ActiveRepairService.instance.registerParentRepairSession(parentRepairSession,  FBUtilities.getBroadcastAddress(), Arrays.asList(cfs), Arrays.asList(range), false, System.currentTimeMillis(), false);

    RepairJobDesc desc = new RepairJobDesc(parentRepairSession, UUID.randomUUID(), KEYSPACE1, "Standard1", Arrays.asList(range));

    MerkleTrees tree1 = createInitialTree(desc);

    MerkleTrees tree2 = createInitialTree(desc);

    // change a range in one of the trees
    Token token = partirioner.midpoint(range.left, range.right);
    tree1.invalidate(token);
    MerkleTree.TreeRange changed = tree1.get(token);
    changed.hash("non-empty hash!".getBytes());

    Set<Range<Token>> interesting = new HashSet<>();
    interesting.add(changed);

    // difference the trees
    // note: we reuse the same endpoint which is bogus in theory but fine here
    TreeResponse r1 = new TreeResponse(InetAddress.getByName("127.0.0.1"), tree1);
    TreeResponse r2 = new TreeResponse(InetAddress.getByName("127.0.0.2"), tree2);
    LocalSyncTask task = new LocalSyncTask(desc, r1, r2, ActiveRepairService.UNREPAIRED_SSTABLE);
    task.run();

    // ensure that the changed range was recorded
    assertEquals("Wrong differing ranges", interesting.size(), task.getCurrentStat().numberOfDifferences);
}