From 2f74831f4218142f6e118678a3c74c79c1f7e1ed Mon Sep 17 00:00:00 2001 From: Mahdi Mohammadi Date: Wed, 15 Jun 2016 11:43:27 +0200 Subject: [PATCH] Cache local ranges when calculating repair neighbors patch by Mahdi Mohammadi; reviewed by Paulo Motta for CASSANDRA-11933 --- CHANGES.txt | 1 + .../cassandra/service/ActiveRepairService.java | 8 +++++--- .../apache/cassandra/service/StorageService.java | 6 +++++- .../service/AntiEntropyServiceTestAbstract.java | 14 ++++++++------ 4 files changed, 19 insertions(+), 10 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 7d70902f6c..ec2b48e90b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.1.15 + * Cache local ranges when calculating repair neighbors (CASSANDRA-11933) * Allow LWT operation on static column with only partition keys (CASSANDRA-10532) * Create interval tree over canonical sstables to avoid missing sstables during streaming (CASSANDRA-11886) * cqlsh COPY FROM: shutdown parent cluster after forking, to avoid corrupting SSL connections (CASSANDRA-11749) diff --git a/src/java/org/apache/cassandra/service/ActiveRepairService.java b/src/java/org/apache/cassandra/service/ActiveRepairService.java index f8975f9e1f..4c83c480fa 100644 --- a/src/java/org/apache/cassandra/service/ActiveRepairService.java +++ b/src/java/org/apache/cassandra/service/ActiveRepairService.java @@ -164,7 +164,8 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai RepairFuture submitArtificialRepairSession(RepairJobDesc desc) { Set neighbours = new HashSet<>(); - neighbours.addAll(ActiveRepairService.getNeighbors(desc.keyspace, desc.range, null, null)); + Collection> keyspaceLocalRanges = StorageService.instance.getLocalRanges(desc.keyspace); + neighbours.addAll(ActiveRepairService.getNeighbors(desc.keyspace, keyspaceLocalRanges, desc.range, null, null)); RepairSession session = new RepairSession(desc.parentSessionId, desc.sessionId, desc.range, desc.keyspace, RepairParallelism.PARALLEL, neighbours, new String[]{desc.columnFamily}); sessions.put(session.getId(), session); RepairFuture futureTask = new RepairFuture(session); @@ -176,17 +177,18 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai * Return all of the neighbors with whom we share the provided range. * * @param keyspaceName keyspace to repair + * @param keyspaceLocalRanges local-range for given keyspaceName * @param toRepair token to repair * @param dataCenters the data centers to involve in the repair * * @return neighbors with whom we share the provided range */ - public static Set getNeighbors(String keyspaceName, Range toRepair, Collection dataCenters, Collection hosts) + public static Set getNeighbors(String keyspaceName, Collection> keyspaceLocalRanges, Range toRepair, Collection dataCenters, Collection hosts) { StorageService ss = StorageService.instance; Map, List> replicaSets = ss.getRangeToAddressMap(keyspaceName); Range rangeSuperSet = null; - for (Range range : ss.getLocalRanges(keyspaceName)) + for (Range range : keyspaceLocalRanges) { if (range.contains(toRepair)) { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index eea4556690..27939f9b26 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2978,13 +2978,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return; } + //pre-calculate output of getLocalRanges and pass it to getNeighbors to increase performance and prevent + //calculation multiple times + Collection> keyspaceLocalRanges = getLocalRanges(keyspace); + Set allNeighbors = new HashSet<>(); Map> rangeToNeighbors = new HashMap<>(); for (Range range : ranges) { try { - Set neighbors = ActiveRepairService.getNeighbors(keyspace, range, dataCenters, hosts); + Set neighbors = ActiveRepairService.getNeighbors(keyspace, keyspaceLocalRanges, range, dataCenters, hosts); rangeToNeighbors.put(range, neighbors); allNeighbors.addAll(neighbors); } diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java index ac39de6a63..21eb492f8f 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java @@ -123,7 +123,7 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader Set neighbors = new HashSet(); for (Range range : ranges) { - neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, null, null)); + neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, ranges, range, null, null)); } assertEquals(expected, neighbors); } @@ -146,7 +146,7 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader Set neighbors = new HashSet(); for (Range range : ranges) { - neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, null, null)); + neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, ranges, range, null, null)); } assertEquals(expected, neighbors); } @@ -168,7 +168,7 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader Set neighbors = new HashSet(); for (Range range : ranges) { - neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null)); + neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, ranges, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null)); } assertEquals(expected, neighbors); } @@ -196,7 +196,7 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader Set neighbors = new HashSet(); for (Range range : ranges) { - neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null)); + neighbors.addAll(ActiveRepairService.getNeighbors(keyspaceName, ranges, range, Arrays.asList(DatabaseDescriptor.getLocalDataCenter()), null)); } assertEquals(expected, neighbors); } @@ -218,7 +218,8 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader expected.remove(FBUtilities.getBroadcastAddress()); Collection hosts = Arrays.asList(FBUtilities.getBroadcastAddress().getCanonicalHostName(),expected.get(0).getCanonicalHostName()); - assertEquals(expected.get(0), ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts).iterator().next()); + Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); + assertEquals(expected.get(0), ActiveRepairService.getNeighbors(keyspaceName, ranges, ranges.iterator().next(), null, hosts).iterator().next()); } @Test(expected = IllegalArgumentException.class) @@ -227,7 +228,8 @@ public abstract class AntiEntropyServiceTestAbstract extends SchemaLoader addTokens(2 * Keyspace.open(keyspaceName).getReplicationStrategy().getReplicationFactor()); //Dont give local endpoint Collection hosts = Arrays.asList("127.0.0.3"); - ActiveRepairService.getNeighbors(keyspaceName, StorageService.instance.getLocalRanges(keyspaceName).iterator().next(), null, hosts); + Collection> ranges = StorageService.instance.getLocalRanges(keyspaceName); + ActiveRepairService.getNeighbors(keyspaceName, ranges, ranges.iterator().next(), null, hosts); } Set addTokens(int max) throws Throwable