From a9b9e627b0256a7b55dbfefa6960e1e5b8379e64 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Thu, 9 Jul 2015 13:28:38 +0200 Subject: [PATCH 1/3] Fix comparison contract violation in the dynamic snitch sorting patch by slebresne; reviewed by benedict for CASSANDRA-9519 Conflicts: CHANGES.txt --- CHANGES.txt | 1 + .../locator/DynamicEndpointSnitch.java | 34 +++++++-- .../locator/DynamicEndpointSnitchTest.java | 69 ++++++++++++++++++- 3 files changed, 95 insertions(+), 9 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index a755cb98fe..f20fad8b47 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.0.17 + * Complete CASSANDRA-8448 fix (CASSANDRA-9519) * Don't include auth credentials in debug log (CASSANDRA-9682) * Can't transition from write survey to normal mode (CASSANDRA-9740) * Avoid NPE in AuthSuccess#decode (CASSANDRA-9727) diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 3469847214..f226989864 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -42,9 +42,9 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa private static final double ALPHA = 0.75; // set to 0.75 to make EDS more biased to towards the newer values private static final int WINDOW_SIZE = 100; - private int UPDATE_INTERVAL_IN_MS = DatabaseDescriptor.getDynamicUpdateInterval(); - private int RESET_INTERVAL_IN_MS = DatabaseDescriptor.getDynamicResetInterval(); - private double BADNESS_THRESHOLD = DatabaseDescriptor.getDynamicBadnessThreshold(); + private final int UPDATE_INTERVAL_IN_MS = DatabaseDescriptor.getDynamicUpdateInterval(); + private final int RESET_INTERVAL_IN_MS = DatabaseDescriptor.getDynamicResetInterval(); + private final double BADNESS_THRESHOLD = DatabaseDescriptor.getDynamicBadnessThreshold(); // the score for a merged set of endpoints must be this much worse than the score for separate endpoints to // warrant not merging two ranges into a single range @@ -154,7 +154,18 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa private void sortByProximityWithScore(final InetAddress address, List addresses) { - super.sortByProximity(address, addresses); + // Scores can change concurrently from a call to this method. But Collections.sort() expects + // its comparator to be "stable", that is 2 endpoint should compare the same way for the duration + // of the sort() call. As we copy the scores map on write, it is thus enough to alias the current + // version of it during this call. + final HashMap scores = this.scores; + Collections.sort(addresses, new Comparator() + { + public int compare(InetAddress a1, InetAddress a2) + { + return compareEndpoints(address, a1, a2, scores); + } + }); } private void sortByProximityWithBadness(final InetAddress address, List addresses) @@ -163,6 +174,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa return; subsnitch.sortByProximity(address, addresses); + HashMap scores = this.scores; // Make sure the score don't change in the middle of the loop below + // (which wouldn't really matter here but its cleaner that way). ArrayList subsnitchOrderedScores = new ArrayList<>(addresses.size()); for (InetAddress inet : addresses) { @@ -189,7 +202,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa } } - public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2) + // Compare endpoints given an immutable snapshot of the scores + private int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2, Map scores) { Double scored1 = scores.get(a1); Double scored2 = scores.get(a2); @@ -214,6 +228,14 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa return 1; } + public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2) + { + // That function is fundamentally unsafe because the scores can change at any time and so the result of that + // method is not stable for identical arguments. This is why we don't rely on super.sortByProximity() in + // sortByProximityWithScore(). + throw new UnsupportedOperationException("You shouldn't wrap the DynamicEndpointSnitch (within itself or otherwise)"); + } + public void receiveTiming(InetAddress host, long latency) // this is cheap { ExponentiallyDecayingSample sample = samples.get(host); @@ -263,7 +285,6 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa scores = newScores; } - private void reset() { for (ExponentiallyDecayingSample sample : samples.values()) @@ -287,6 +308,7 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa { return BADNESS_THRESHOLD; } + public String getSubsnitchClassName() { return subsnitch.getClass().getName(); diff --git a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java index e23bcfa312..3f90532e34 100644 --- a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java @@ -21,9 +21,9 @@ package org.apache.cassandra.locator; import java.io.IOException; import java.net.InetAddress; -import java.util.Arrays; -import java.util.List; +import java.util.*; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.service.StorageService; import org.junit.Test; @@ -90,4 +90,67 @@ public class DynamicEndpointSnitchTest order = Arrays.asList(host1, host3, host2); assertEquals(order, dsnitch.getSortedListByProximity(self, Arrays.asList(host1, host2, host3))); } -} \ No newline at end of file + + @Test + public void testConcurrency() throws InterruptedException, IOException, ConfigurationException + { + // The goal of this test is to check for CASSANDRA-8448/CASSANDRA-9519 + double badness = DatabaseDescriptor.getDynamicBadnessThreshold(); + DatabaseDescriptor.setDynamicBadnessThreshold(0.0); + + final int ITERATIONS = 10; + + // do this because SS needs to be initialized before DES can work properly. + StorageService.instance.initClient(0); + SimpleSnitch ss = new SimpleSnitch(); + DynamicEndpointSnitch dsnitch = new DynamicEndpointSnitch(ss, String.valueOf(ss.hashCode())); + InetAddress self = FBUtilities.getBroadcastAddress(); + + List hosts = new ArrayList<>(); + // We want a giant list of hosts so that sorting it takes time, making it much more likely to reproduce the + // problem we're looking for. + for (int i = 0; i < 10; i++) + for (int j = 0; j < 256; j++) + for (int k = 0; k < 256; k++) + hosts.add(InetAddress.getByAddress(new byte[]{127, (byte)i, (byte)j, (byte)k})); + + ScoreUpdater updater = new ScoreUpdater(dsnitch, hosts); + updater.start(); + + List result = null; + for (int i = 0; i < ITERATIONS; i++) + result = dsnitch.getSortedListByProximity(self, hosts); + + updater.stopped = true; + updater.join(); + + DatabaseDescriptor.setDynamicBadnessThreshold(badness); + } + + public static class ScoreUpdater extends Thread + { + private static final int SCORE_RANGE = 100; + + public volatile boolean stopped; + + private final DynamicEndpointSnitch dsnitch; + private final List hosts; + private final Random random = new Random(); + + public ScoreUpdater(DynamicEndpointSnitch dsnitch, List hosts) + { + this.dsnitch = dsnitch; + this.hosts = hosts; + } + + public void run() + { + while (!stopped) + { + InetAddress host = hosts.get(random.nextInt(hosts.size())); + int score = random.nextInt(SCORE_RANGE); + dsnitch.receiveTiming(host, score); + } + } + } +} From 0ef188869049ec6233d115f7a46c25f492e8fa42 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Thu, 16 Jul 2015 15:14:54 +0200 Subject: [PATCH 2/3] Move CASSANDRA-9519 test in long tests (and reduce the size of the list used) --- .../DynamicEndpointSnitchLongTest.java | 104 ++++++++++++++++++ .../locator/DynamicEndpointSnitchTest.java | 64 ----------- 2 files changed, 104 insertions(+), 64 deletions(-) create mode 100644 test/long/org/apache/cassandra/locator/DynamicEndpointSnitchLongTest.java diff --git a/test/long/org/apache/cassandra/locator/DynamicEndpointSnitchLongTest.java b/test/long/org/apache/cassandra/locator/DynamicEndpointSnitchLongTest.java new file mode 100644 index 0000000000..1c628fa52b --- /dev/null +++ b/test/long/org/apache/cassandra/locator/DynamicEndpointSnitchLongTest.java @@ -0,0 +1,104 @@ +/* +* 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.locator; + +import java.io.IOException; +import java.net.InetAddress; +import java.util.*; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.service.StorageService; +import org.junit.Test; + +import org.apache.cassandra.utils.FBUtilities; + +import static org.junit.Assert.assertEquals; + +public class DynamicEndpointSnitchLongTest +{ + @Test + public void testConcurrency() throws InterruptedException, IOException, ConfigurationException + { + // The goal of this test is to check for CASSANDRA-8448/CASSANDRA-9519 + double badness = DatabaseDescriptor.getDynamicBadnessThreshold(); + DatabaseDescriptor.setDynamicBadnessThreshold(0.0); + + try + { + final int ITERATIONS = 10000; + + // do this because SS needs to be initialized before DES can work properly. + StorageService.instance.initClient(0); + SimpleSnitch ss = new SimpleSnitch(); + DynamicEndpointSnitch dsnitch = new DynamicEndpointSnitch(ss, String.valueOf(ss.hashCode())); + InetAddress self = FBUtilities.getBroadcastAddress(); + + List hosts = new ArrayList<>(); + // We want a big list of hosts so sorting takes time, making it much more likely to reproduce the + // problem we're looking for. + for (int i = 0; i < 100; i++) + for (int j = 0; j < 256; j++) + hosts.add(InetAddress.getByAddress(new byte[]{127, 0, (byte)i, (byte)j})); + + ScoreUpdater updater = new ScoreUpdater(dsnitch, hosts); + updater.start(); + + List result = null; + for (int i = 0; i < ITERATIONS; i++) + result = dsnitch.getSortedListByProximity(self, hosts); + + updater.stopped = true; + updater.join(); + } + finally + { + DatabaseDescriptor.setDynamicBadnessThreshold(badness); + } + } + + public static class ScoreUpdater extends Thread + { + private static final int SCORE_RANGE = 100; + + public volatile boolean stopped; + + private final DynamicEndpointSnitch dsnitch; + private final List hosts; + private final Random random = new Random(); + + public ScoreUpdater(DynamicEndpointSnitch dsnitch, List hosts) + { + this.dsnitch = dsnitch; + this.hosts = hosts; + } + + public void run() + { + while (!stopped) + { + InetAddress host = hosts.get(random.nextInt(hosts.size())); + int score = random.nextInt(SCORE_RANGE); + dsnitch.receiveTiming(host, score); + } + } + } +} + diff --git a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java index 3f90532e34..c1928d82b2 100644 --- a/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java +++ b/test/unit/org/apache/cassandra/locator/DynamicEndpointSnitchTest.java @@ -23,7 +23,6 @@ import java.io.IOException; import java.net.InetAddress; import java.util.*; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.service.StorageService; import org.junit.Test; @@ -90,67 +89,4 @@ public class DynamicEndpointSnitchTest order = Arrays.asList(host1, host3, host2); assertEquals(order, dsnitch.getSortedListByProximity(self, Arrays.asList(host1, host2, host3))); } - - @Test - public void testConcurrency() throws InterruptedException, IOException, ConfigurationException - { - // The goal of this test is to check for CASSANDRA-8448/CASSANDRA-9519 - double badness = DatabaseDescriptor.getDynamicBadnessThreshold(); - DatabaseDescriptor.setDynamicBadnessThreshold(0.0); - - final int ITERATIONS = 10; - - // do this because SS needs to be initialized before DES can work properly. - StorageService.instance.initClient(0); - SimpleSnitch ss = new SimpleSnitch(); - DynamicEndpointSnitch dsnitch = new DynamicEndpointSnitch(ss, String.valueOf(ss.hashCode())); - InetAddress self = FBUtilities.getBroadcastAddress(); - - List hosts = new ArrayList<>(); - // We want a giant list of hosts so that sorting it takes time, making it much more likely to reproduce the - // problem we're looking for. - for (int i = 0; i < 10; i++) - for (int j = 0; j < 256; j++) - for (int k = 0; k < 256; k++) - hosts.add(InetAddress.getByAddress(new byte[]{127, (byte)i, (byte)j, (byte)k})); - - ScoreUpdater updater = new ScoreUpdater(dsnitch, hosts); - updater.start(); - - List result = null; - for (int i = 0; i < ITERATIONS; i++) - result = dsnitch.getSortedListByProximity(self, hosts); - - updater.stopped = true; - updater.join(); - - DatabaseDescriptor.setDynamicBadnessThreshold(badness); - } - - public static class ScoreUpdater extends Thread - { - private static final int SCORE_RANGE = 100; - - public volatile boolean stopped; - - private final DynamicEndpointSnitch dsnitch; - private final List hosts; - private final Random random = new Random(); - - public ScoreUpdater(DynamicEndpointSnitch dsnitch, List hosts) - { - this.dsnitch = dsnitch; - this.hosts = hosts; - } - - public void run() - { - while (!stopped) - { - InetAddress host = hosts.get(random.nextInt(hosts.size())); - int score = random.nextInt(SCORE_RANGE); - dsnitch.receiveTiming(host, score); - } - } - } } From f60e4ad4298725dac57c36da8427d992be19eb8a Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Fri, 17 Jul 2015 15:39:32 +0200 Subject: [PATCH 3/3] Don't wrap byte arrays in SequentialWriter patch by slebresne; reviewed by snazy & benedict for CASSANDRA-9797 --- CHANGES.txt | 1 + .../cassandra/io/util/SequentialWriter.java | 22 +++++++++++++++++-- 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 9a262dc2d4..47d1db5f5a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2.0-rc3 + * Don't wrap byte arrays in SequentialWriter (CASSANDRA-9797) * sum() and avg() functions missing for smallint and tinyint types (CASSANDRA-9671) * Revert CASSANDRA-9542 (allow native functions in UDA) (CASSANDRA-9771) Merged from 2.1: diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index f3268a2d6a..915133f2bb 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java @@ -185,12 +185,30 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne public void write(byte[] buffer) throws IOException { - write(ByteBuffer.wrap(buffer, 0, buffer.length)); + write(buffer, 0, buffer.length); } public void write(byte[] data, int offset, int length) throws IOException { - write(ByteBuffer.wrap(data, offset, length)); + if (buffer == null) + throw new ClosedChannelException(); + + int position = offset; + int remaining = length; + while (remaining > 0) + { + if (!buffer.hasRemaining()) + reBuffer(); + + int toCopy = Math.min(remaining, buffer.remaining()); + buffer.put(data, position, toCopy); + + remaining -= toCopy; + position += toCopy; + + isDirty = true; + syncNeeded = true; + } } public int write(ByteBuffer src) throws IOException