From 674b8d5100bfac7947a976a9aa270d774db9b336 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 10 Jul 2025 20:20:49 +0100 Subject: [PATCH] Deduplicate ranges in WatermarkCollector Also: collect ranges before submitting to TopologyManager to avoid quadratic complexity patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20757 --- .../service/accord/WatermarkCollector.java | 140 ++++++++++-------- .../accord/WatermarkCollectorTest.java | 19 +-- 2 files changed, 85 insertions(+), 74 deletions(-) diff --git a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java index 4ae51792ce..64155e0533 100644 --- a/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java +++ b/src/java/org/apache/cassandra/service/accord/WatermarkCollector.java @@ -19,16 +19,19 @@ package org.apache.cassandra.service.accord; import java.io.IOException; -import java.util.HashMap; +import java.util.ArrayList; +import java.util.Comparator; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.function.BiConsumer; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Predicates; import com.google.common.collect.Iterators; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import com.google.common.primitives.Ints; import accord.api.ConfigurationService; import accord.local.Node; @@ -36,8 +39,9 @@ import accord.primitives.Range; import accord.primitives.Ranges; import accord.topology.Topology; import accord.utils.Invariants; +import accord.utils.ReducingRangeMap; import accord.utils.async.AsyncResult; -import org.agrona.collections.Int2ObjectHashMap; +import org.agrona.collections.Long2LongHashMap; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -60,17 +64,21 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retry */ public class WatermarkCollector implements ConfigurationService.Listener { - private static final Logger logger = LoggerFactory.getLogger(WatermarkCollector.class); + private static final Comparator> sortByEpochThenRange = (a, b) -> { + int c = Long.compareUnsigned(a.getValue(), b.getValue()); + if (c == 0) c = a.getKey().compare(b.getKey()); + return c; + }; - final Map closed; - final Map retired; - final Int2ObjectHashMap synced; + ReducingRangeMap closed; + ReducingRangeMap retired; + final Long2LongHashMap synced; WatermarkCollector() { - closed = new HashMap<>(); - retired = new HashMap<>(); - synced = new Int2ObjectHashMap<>(); + closed = new ReducingRangeMap<>(); + retired = new ReducingRangeMap<>(); + synced = new Long2LongHashMap(-1); } @Override public AsyncResult onTopologyUpdate(Topology topology, boolean isLoad, boolean startSync) @@ -79,40 +87,36 @@ public class WatermarkCollector implements ConfigurationService.Listener } @Override - public void onRemoteSyncComplete(Node.Id node, long epoch) + public synchronized void onRemoteSyncComplete(Node.Id node, long epoch) { - synced.compute(node.id, (k, prev) -> prev == null ? epoch : Long.max(prev, epoch)); + synced.compute(node.id, (k, prev) -> prev == -1 ? epoch : Long.max(prev, epoch)); } @Override - public void onEpochClosed(Ranges ranges, long epoch) + public synchronized void onEpochClosed(Ranges ranges, long epoch) { - synchronized (this) - { - for (Range range : ranges) - this.closed.compute(range, (k, prev) -> prev == null ? epoch : Long.max(prev, epoch)); - } + closed = ReducingRangeMap.merge(closed, ReducingRangeMap.create(ranges, epoch), Long::max); } @Override - public void onEpochRetired(Ranges ranges, long epoch) + public synchronized void onEpochRetired(Ranges ranges, long epoch) { - synchronized (this) - { - for (Range range : ranges) - this.retired.compute(range, (k, prev) -> prev == null ? epoch : Long.max(prev, epoch)); - } + retired = ReducingRangeMap.merge(retired, ReducingRangeMap.create(ranges, epoch), Long::max); } public final IVerbHandler handler = new IVerbHandler() { - public void doVerb(Message message) throws IOException + public void doVerb(Message message) { Invariants.require(AccordService.started()); Snapshot snapshot; synchronized (WatermarkCollector.this) { - snapshot = new Snapshot(new HashMap<>(closed), new HashMap<>(retired), new Int2ObjectHashMap<>(synced)); + List> closedSnapshot = closed.foldlWithBounds((epoch, list, start, end) -> { list.add(Map.entry(start.rangeFactory().newRange(start, end), epoch)); return list; }, new ArrayList<>(), Predicates.alwaysFalse()); + List> retiredSnapshot = retired.foldlWithBounds((epoch, list, start, end) -> { list.add(Map.entry(start.rangeFactory().newRange(start, end), epoch)); return list; }, new ArrayList<>(), Predicates.alwaysFalse()); + Long2LongHashMap syncedSnapshot = new Long2LongHashMap(synced.size(), 0.6f, -1); + syncedSnapshot.putAll(synced); + snapshot = new Snapshot(closedSnapshot, retiredSnapshot, syncedSnapshot); } MessagingService.instance().respond(snapshot, message); } @@ -139,32 +143,47 @@ public class WatermarkCollector implements ConfigurationService.Listener Snapshot snapshot = m.payload; long minEpoch = configService.minEpoch(); - for (Map.Entry e : snapshot.closed.entrySet()) + forEachEpoch(configService::receiveClosed, snapshot.closed); + forEachEpoch(configService::receiveRetired, snapshot.retired); + for (Map.Entry e : snapshot.synced.entrySet()) { - Ranges r = Ranges.of(e.getKey()); - configService.receiveClosed(r, e.getValue()); - } - for (Map.Entry e : snapshot.retired.entrySet()) - { - Ranges r = Ranges.of(e.getKey()); - configService.receiveRetired(r, e.getValue()); - } - for (Map.Entry e : snapshot.synced.entrySet()) - { - Node.Id node = new Node.Id(e.getKey()); + Node.Id node = new Node.Id(Ints.saturatedCast(e.getKey())); for (long epoch = minEpoch; epoch <= e.getValue(); epoch++) configService.receiveRemoteSyncComplete(node, epoch); } }); } + private static void forEachEpoch(BiConsumer forEachEpoch, List> rangesAndEpochs) + { + if (rangesAndEpochs.isEmpty()) + return; + + rangesAndEpochs.sort(sortByEpochThenRange); + long collectingEpoch = rangesAndEpochs.get(0).getValue(); + List ranges = new ArrayList<>(); + for (Map.Entry e : rangesAndEpochs) + { + Range range = e.getKey(); + long epoch = e.getValue(); + if (epoch != collectingEpoch) + { + forEachEpoch.accept(Ranges.of(ranges.toArray(Range[]::new)), collectingEpoch); + collectingEpoch = epoch; + ranges.clear(); + } + ranges.add(range); + } + forEachEpoch.accept(Ranges.of(ranges.toArray(Range[]::new)), collectingEpoch); + } + public static class Snapshot { - public final Map closed; - public final Map retired; - public final Int2ObjectHashMap synced; + public final List> closed; + public final List> retired; + public final Long2LongHashMap synced; - public Snapshot(Map closed, Map retired, Int2ObjectHashMap synced) + public Snapshot(List> closed, List> retired, Long2LongHashMap synced) { this.closed = closed; this.retired = retired; @@ -193,21 +212,21 @@ public class WatermarkCollector implements ConfigurationService.Listener public void serialize(Snapshot t, DataOutputPlus out) throws IOException { out.writeUnsignedVInt32(t.closed.size()); - for (Map.Entry e : t.closed.entrySet()) + for (Map.Entry e : t.closed) { TokenRange.serializer.serialize((TokenRange) e.getKey(), out); out.writeUnsignedVInt(e.getValue()); } out.writeUnsignedVInt32(t.retired.size()); - for (Map.Entry e : t.retired.entrySet()) + for (Map.Entry e : t.retired) { TokenRange.serializer.serialize((TokenRange) e.getKey(), out); out.writeUnsignedVInt(e.getValue()); } out.writeUnsignedVInt32(t.synced.size()); - for (Map.Entry e : t.synced.entrySet()) + for (Map.Entry e : t.synced.entrySet()) { - out.writeUnsignedVInt32(e.getKey()); + out.writeUnsignedVInt(e.getKey()); out.writeUnsignedVInt(e.getValue()); } } @@ -217,25 +236,20 @@ public class WatermarkCollector implements ConfigurationService.Listener public Snapshot deserialize(DataInputPlus in) throws IOException { int closedSize = in.readUnsignedVInt32(); - Map closed = new HashMap<>(); + List> closed = new ArrayList<>(); for (int i = 0; i < closedSize; i++) - { - closed.put(TokenRange.serializer.deserialize(in), - in.readUnsignedVInt()); - } + closed.add(Map.entry(TokenRange.serializer.deserialize(in), in.readUnsignedVInt())); + int retiredSize = in.readUnsignedVInt32(); - Map retired = new HashMap<>(); + List> retired = new ArrayList<>(); for (int i = 0; i < retiredSize; i++) - { - retired.put(TokenRange.serializer.deserialize(in), - in.readUnsignedVInt()); - } + retired.add(Map.entry(TokenRange.serializer.deserialize(in), in.readUnsignedVInt())); + int syncedSize = in.readUnsignedVInt32(); - Int2ObjectHashMap synced = new Int2ObjectHashMap<>(); + Long2LongHashMap synced = new Long2LongHashMap(-1); for (int i = 0; i < syncedSize; i++) { - synced.put(in.readUnsignedVInt32(), - (Long) in.readUnsignedVInt()); + synced.put(in.readUnsignedVInt(), in.readUnsignedVInt()); } return new Snapshot(closed, retired, synced); } @@ -245,19 +259,19 @@ public class WatermarkCollector implements ConfigurationService.Listener { int size = 0; size += TypeSizes.sizeofUnsignedVInt(t.closed.size()); - for (Map.Entry e : t.closed.entrySet()) + for (Map.Entry e : t.closed) { size += TokenRange.serializer.serializedSize((TokenRange) e.getKey()); size += TypeSizes.sizeofUnsignedVInt(e.getValue()); } size += TypeSizes.sizeofUnsignedVInt(t.retired.size()); - for (Map.Entry e : t.retired.entrySet()) + for (Map.Entry e : t.retired) { size += TokenRange.serializer.serializedSize((TokenRange) e.getKey()); size += TypeSizes.sizeofUnsignedVInt(e.getValue()); } size += TypeSizes.sizeofUnsignedVInt(t.synced.size()); - for (Map.Entry e : t.synced.entrySet()) + for (Map.Entry e : t.synced.entrySet()) { size += TypeSizes.sizeofUnsignedVInt(e.getKey()); size += TypeSizes.sizeofUnsignedVInt(e.getValue()); diff --git a/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java b/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java index 06af98f5bc..7b7ec0f5e5 100644 --- a/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java @@ -18,10 +18,10 @@ package org.apache.cassandra.service.accord; +import java.util.ArrayList; import java.util.HashMap; import java.util.Map; -import com.google.common.collect.Sets; import org.junit.Test; import accord.local.Node; @@ -31,6 +31,7 @@ import accord.utils.Gen; import accord.utils.Gens; import accord.utils.Invariants; import org.agrona.collections.Int2ObjectHashMap; +import org.agrona.collections.Long2LongHashMap; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.io.Serializers; @@ -58,34 +59,30 @@ public class WatermarkCollectorTest private static void maybeUpdatePartitioner(WatermarkCollector.Snapshot snapshot) { - for (Range range : Sets.union(snapshot.closed.keySet(), snapshot.retired.keySet())) - { - TokenRange tr = (TokenRange) range; - DatabaseDescriptor.setPartitionerUnsafe(tr.start().token().getPartitioner()); - break; - } + if (!snapshot.closed.isEmpty()) DatabaseDescriptor.setPartitionerUnsafe(((TokenRange)snapshot.closed.get(0).getKey()).start().token().getPartitioner()); + else if (!snapshot.retired.isEmpty()) DatabaseDescriptor.setPartitionerUnsafe(((TokenRange)snapshot.retired.get(0).getKey()).start().token().getPartitioner()); } private Gen snapshotGen() { Gen partitionerGen = AccordGenerators.partitioner(); Gen.LongGen epochGen = AccordGens.epochs(); - Gen> syncedGen = syncedGen(); + Gen syncedGen = syncedGen(); return rs -> { IPartitioner partitioner = partitionerGen.next(rs); Gen rangeGen = AccordGenerators.range(partitioner); Gen> mapGen = mapGen(Gens.ints().between(0, 10), rangeGen, epochGen); - return new WatermarkCollector.Snapshot(mapGen.next(rs), mapGen.next(rs), syncedGen.next(rs)); + return new WatermarkCollector.Snapshot(new ArrayList<>(mapGen.next(rs).entrySet()), new ArrayList<>(mapGen.next(rs).entrySet()), syncedGen.next(rs)); }; } - private static Gen> syncedGen() + private static Gen syncedGen() { Gen.IntGen sizeGen = Gens.ints().between(0, 10); Gen idGen = AccordGens.nodes(); Gen.LongGen epochGen = AccordGens.epochs(); return rs -> { - Int2ObjectHashMap map = new Int2ObjectHashMap<>(); + Long2LongHashMap map = new Long2LongHashMap(-1); Gen uniqueIdGen = idGen.filter(id -> !map.containsKey(id.id)); for (int i = 0, size = sizeGen.nextInt(rs); i < size; i++) map.put(uniqueIdGen.next(rs).id, epochGen.next(rs));