Repository: cassandra
Updated Branches:
  refs/heads/trunk 560faba2f -> d43b9ce50


http://git-wip-us.apache.org/repos/asf/cassandra/blob/d43b9ce5/test/unit/org/apache/cassandra/net/RateBasedBackPressureTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/net/RateBasedBackPressureTest.java 
b/test/unit/org/apache/cassandra/net/RateBasedBackPressureTest.java
new file mode 100644
index 0000000..b94b6ee
--- /dev/null
+++ b/test/unit/org/apache/cassandra/net/RateBasedBackPressureTest.java
@@ -0,0 +1,409 @@
+/*
+* 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.net;
+
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.RateLimiter;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.cassandra.utils.TestTimeSource;
+import org.apache.cassandra.utils.TimeSource;
+
+import static org.apache.cassandra.net.RateBasedBackPressure.FACTOR;
+import static org.apache.cassandra.net.RateBasedBackPressure.FLOW;
+import static org.apache.cassandra.net.RateBasedBackPressure.HIGH_RATIO;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class RateBasedBackPressureTest
+{
+    @Test(expected = IllegalArgumentException.class)
+    public void testAcceptsNoLessThanThreeArguments() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "1"), new 
TestTimeSource(), 10);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testHighRatioMustBeBiggerThanZero() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0", FACTOR, 
"2", FLOW, "FAST"), new TestTimeSource(), 10);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testHighRatioMustBeSmallerEqualThanOne() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "2", FACTOR, 
"2", FLOW, "FAST"), new TestTimeSource(), 10);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testFactorMustBeBiggerEqualThanOne() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, 
"0", FLOW, "FAST"), new TestTimeSource(), 10);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testWindowSizeMustBeBiggerEqualThanTen() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, 
"5", FLOW, "FAST"), new TestTimeSource(), 1);
+    }
+
+    @Test
+    public void testFlowMustBeEitherFASTorSLOW() throws Exception
+    {
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, 
"1", FLOW, "FAST"), new TestTimeSource(), 10);
+        new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, 
"1", FLOW, "SLOW"), new TestTimeSource(), 10);
+        try
+        {
+            new RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", 
FACTOR, "1", FLOW, "WRONG"), new TestTimeSource(), 10);
+            fail("Expected to fail with wrong flow type.");
+        }
+        catch (Exception ex)
+        {
+        }
+    }
+
+    @Test
+    public void testBackPressureStateUpdates()
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        RateBasedBackPressure strategy = new 
RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+
+        RateBasedBackPressureState state = 
strategy.newState(InetAddress.getLoopbackAddress());
+        state.onMessageSent(null);
+        assertEquals(0, state.incomingRate.size());
+        assertEquals(0, state.outgoingRate.size());
+
+        state = strategy.newState(InetAddress.getLoopbackAddress());
+        state.onResponseReceived();
+        assertEquals(1, state.incomingRate.size());
+        assertEquals(1, state.outgoingRate.size());
+
+        state = strategy.newState(InetAddress.getLoopbackAddress());
+        state.onResponseTimeout();
+        assertEquals(0, state.incomingRate.size());
+        assertEquals(1, state.outgoingRate.size());
+    }
+
+    @Test
+    public void testBackPressureIsNotUpdatedBeyondInfinity() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        RateBasedBackPressure strategy = new 
RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+        RateBasedBackPressureState state = 
strategy.newState(InetAddress.getLoopbackAddress());
+
+        // Get initial rate:
+        double initialRate = state.rateLimiter.getRate();
+        assertEquals(Double.POSITIVE_INFINITY, initialRate, 0.0);
+
+        // Update incoming and outgoing rate equally:
+        state.incomingRate.update(1);
+        state.outgoingRate.update(1);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the rate doesn't change because already at infinity:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        assertEquals(initialRate, state.rateLimiter.getRate(), 0.0);
+    }
+
+    @Test
+    public void testBackPressureIsUpdatedOncePerWindowSize() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        RateBasedBackPressure strategy = new 
RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+        RateBasedBackPressureState state = 
strategy.newState(InetAddress.getLoopbackAddress());
+
+        // Get initial time:
+        long current = state.getLastIntervalAcquire();
+        assertEquals(0, current);
+
+        // Update incoming and outgoing rate:
+        state.incomingRate.update(1);
+        state.outgoingRate.update(1);
+
+        // Move time ahead by window size:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the timestamp changed:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        current = state.getLastIntervalAcquire();
+        assertEquals(timeSource.currentTimeMillis(), current);
+
+        // Move time ahead by less than interval:
+        long previous = current;
+        timeSource.sleep(windowSize / 2, TimeUnit.MILLISECONDS);
+
+        // Verify the last timestamp didn't change because below the window 
size:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        current = state.getLastIntervalAcquire();
+        assertEquals(previous, current);
+    }
+
+    @Test
+    public void testBackPressureWhenBelowHighRatio() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        RateBasedBackPressure strategy = new 
RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+        RateBasedBackPressureState state = 
strategy.newState(InetAddress.getLoopbackAddress());
+
+        // Update incoming and outgoing rate so that the ratio is 0.5:
+        state.incomingRate.update(50);
+        state.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the rate is decreased by factor:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        assertEquals(7.4, state.rateLimiter.getRate(), 0.1);
+    }
+
+    @Test
+    public void 
testBackPressureRateLimiterIsIncreasedAfterGoingAgainAboveHighRatio() throws 
Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        RateBasedBackPressure strategy = new 
RateBasedBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+        RateBasedBackPressureState state = 
strategy.newState(InetAddress.getLoopbackAddress());
+
+        // Update incoming and outgoing rate so that the ratio is 0.5:
+        state.incomingRate.update(50);
+        state.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the rate decreased:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        assertEquals(7.4, state.rateLimiter.getRate(), 0.1);
+
+        // Update incoming and outgoing rate back above high rate:
+        state.incomingRate.update(50);
+        state.outgoingRate.update(50);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify rate limiter is increased by factor:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        assertEquals(8.25, state.rateLimiter.getRate(), 0.1);
+
+        // Update incoming and outgoing rate to keep it below the limiter rate:
+        state.incomingRate.update(1);
+        state.outgoingRate.update(1);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify rate limiter is not increased as already higher than the 
actual rate:
+        strategy.apply(Sets.newHashSet(state), 1, TimeUnit.SECONDS);
+        assertEquals(8.25, state.rateLimiter.getRate(), 0.1);
+    }
+
+    @Test
+    public void testBackPressureFastFlow() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        TestableBackPressure strategy = new 
TestableBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"FAST"), timeSource, windowSize);
+        RateBasedBackPressureState state1 = 
strategy.newState(InetAddress.getByName("127.0.0.1"));
+        RateBasedBackPressureState state2 = 
strategy.newState(InetAddress.getByName("127.0.0.2"));
+        RateBasedBackPressureState state3 = 
strategy.newState(InetAddress.getByName("127.0.0.3"));
+
+        // Update incoming and outgoing rates:
+        state1.incomingRate.update(50);
+        state1.outgoingRate.update(100);
+        state2.incomingRate.update(80); // fast
+        state2.outgoingRate.update(100);
+        state3.incomingRate.update(20);
+        state3.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the fast replica rate limiting has been applied:
+        Set<RateBasedBackPressureState> replicaGroup = Sets.newHashSet(state1, 
state2, state3);
+        strategy.apply(replicaGroup, 1, TimeUnit.SECONDS);
+        assertTrue(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(12.0, 
strategy.getRateLimiterForReplicaGroup(replicaGroup).getRate(), 0.1);
+    }
+
+    @Test
+    public void testBackPressureSlowFlow() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        TestableBackPressure strategy = new 
TestableBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"SLOW"), timeSource, windowSize);
+        RateBasedBackPressureState state1 = 
strategy.newState(InetAddress.getByName("127.0.0.1"));
+        RateBasedBackPressureState state2 = 
strategy.newState(InetAddress.getByName("127.0.0.2"));
+        RateBasedBackPressureState state3 = 
strategy.newState(InetAddress.getByName("127.0.0.3"));
+
+        // Update incoming and outgoing rates:
+        state1.incomingRate.update(50);
+        state1.outgoingRate.update(100);
+        state2.incomingRate.update(100);
+        state2.outgoingRate.update(100);
+        state3.incomingRate.update(20); // slow
+        state3.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the slow replica rate limiting has been applied:
+        Set<RateBasedBackPressureState> replicaGroup = Sets.newHashSet(state1, 
state2, state3);
+        strategy.apply(replicaGroup, 1, TimeUnit.SECONDS);
+        assertTrue(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(3.0, 
strategy.getRateLimiterForReplicaGroup(replicaGroup).getRate(), 0.1);
+    }
+
+    @Test
+    public void testBackPressureWithDifferentGroups() throws Exception
+    {
+        long windowSize = 6000;
+        TestTimeSource timeSource = new TestTimeSource();
+        TestableBackPressure strategy = new 
TestableBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"SLOW"), timeSource, windowSize);
+        RateBasedBackPressureState state1 = 
strategy.newState(InetAddress.getByName("127.0.0.1"));
+        RateBasedBackPressureState state2 = 
strategy.newState(InetAddress.getByName("127.0.0.2"));
+        RateBasedBackPressureState state3 = 
strategy.newState(InetAddress.getByName("127.0.0.3"));
+        RateBasedBackPressureState state4 = 
strategy.newState(InetAddress.getByName("127.0.0.4"));
+
+        // Update incoming and outgoing rates:
+        state1.incomingRate.update(50); // this
+        state1.outgoingRate.update(100);
+        state2.incomingRate.update(100);
+        state2.outgoingRate.update(100);
+        state3.incomingRate.update(20); // this
+        state3.outgoingRate.update(100);
+        state4.incomingRate.update(80);
+        state4.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the first group:
+        Set<RateBasedBackPressureState> replicaGroup = Sets.newHashSet(state1, 
state2);
+        strategy.apply(replicaGroup, 1, TimeUnit.SECONDS);
+        assertTrue(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(7.4, 
strategy.getRateLimiterForReplicaGroup(replicaGroup).getRate(), 0.1);
+
+        // Verify the second group:
+        replicaGroup = Sets.newHashSet(state3, state4);
+        strategy.apply(replicaGroup, 1, TimeUnit.SECONDS);
+        assertTrue(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(3.0, 
strategy.getRateLimiterForReplicaGroup(replicaGroup).getRate(), 0.1);
+    }
+
+    @Test
+    public void testBackPressurePastTimeout() throws Exception
+    {
+        long windowSize = 10000;
+        TestTimeSource timeSource = new TestTimeSource();
+        TestableBackPressure strategy = new 
TestableBackPressure(ImmutableMap.of(HIGH_RATIO, "0.9", FACTOR, "10", FLOW, 
"SLOW"), timeSource, windowSize);
+        RateBasedBackPressureState state1 = 
strategy.newState(InetAddress.getByName("127.0.0.1"));
+        RateBasedBackPressureState state2 = 
strategy.newState(InetAddress.getByName("127.0.0.2"));
+        RateBasedBackPressureState state3 = 
strategy.newState(InetAddress.getByName("127.0.0.3"));
+
+        // Update incoming and outgoing rates:
+        state1.incomingRate.update(5); // slow
+        state1.outgoingRate.update(100);
+        state2.incomingRate.update(100);
+        state2.outgoingRate.update(100);
+        state3.incomingRate.update(100);
+        state3.outgoingRate.update(100);
+
+        // Move time ahead:
+        timeSource.sleep(windowSize, TimeUnit.MILLISECONDS);
+
+        // Verify the slow replica rate limiting has been applied:
+        Set<RateBasedBackPressureState> replicaGroup = Sets.newHashSet(state1, 
state2, state3);
+        strategy.apply(replicaGroup, 4, TimeUnit.SECONDS);
+        assertTrue(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(0.5, 
strategy.getRateLimiterForReplicaGroup(replicaGroup).getRate(), 0.1);
+
+        // Make one more apply call to saturate the rate limit timeout (0.5 
requests per second means 2 requests span
+        // 4 seconds, but we can only make one as we have to subtract the 
incoming response time):
+        strategy.apply(replicaGroup, 4, TimeUnit.SECONDS);
+
+        // Now verify another call to apply doesn't acquire the rate limit 
because of the max timeout of 4 seconds minus
+        // 2 seconds of response time, so the time source itself sleeps two 
second:
+        long start = timeSource.currentTimeMillis();
+        strategy.apply(replicaGroup, 4, TimeUnit.SECONDS);
+        assertFalse(strategy.checkAcquired());
+        assertTrue(strategy.checkApplied());
+        assertEquals(TimeUnit.NANOSECONDS.convert(2, TimeUnit.SECONDS),
+                     strategy.timeout);
+        assertEquals(strategy.timeout,
+                     
TimeUnit.NANOSECONDS.convert(timeSource.currentTimeMillis() - start, 
TimeUnit.MILLISECONDS));
+    }
+
+    public static class TestableBackPressure extends RateBasedBackPressure
+    {
+        public volatile boolean acquired = false;
+        public volatile boolean applied = false;
+        public volatile long timeout;
+
+        public TestableBackPressure(Map<String, Object> args, TimeSource 
timeSource, long windowSize)
+        {
+            super(args, timeSource, windowSize);
+        }
+
+        @Override
+        public boolean doRateLimit(RateLimiter rateLimiter, long 
timeoutInNanos)
+        {
+            acquired = super.doRateLimit(rateLimiter, timeoutInNanos);
+            applied = true;
+            timeout = timeoutInNanos;
+            return acquired;
+        }
+
+        public boolean checkAcquired()
+        {
+            boolean checked = acquired;
+            acquired = false;
+            return checked;
+        }
+
+        public boolean checkApplied()
+        {
+            boolean checked = applied;
+            applied = false;
+            return checked;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d43b9ce5/test/unit/org/apache/cassandra/utils/SlidingTimeRateTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/SlidingTimeRateTest.java 
b/test/unit/org/apache/cassandra/utils/SlidingTimeRateTest.java
new file mode 100644
index 0000000..8c11f9d
--- /dev/null
+++ b/test/unit/org/apache/cassandra/utils/SlidingTimeRateTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SlidingTimeRateTest
+{
+    @Test
+    public void testUpdateAndGet()
+    {
+        SlidingTimeRate rate = new SlidingTimeRate(new TestTimeSource(), 10, 
1, TimeUnit.SECONDS);
+        int updates = 100;
+        for (int i = 0; i < updates; i++)
+        {
+            rate.update(1);
+        }
+        Assert.assertEquals(updates, rate.get(TimeUnit.SECONDS), 0.0);
+    }
+
+    @Test
+    public void testUpdateAndGetBetweenWindows() throws InterruptedException
+    {
+        TestTimeSource time = new TestTimeSource();
+        SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+        int updates = 100;
+        for (int i = 0; i < updates; i++)
+        {
+            rate.update(1);
+            time.sleep(100, TimeUnit.MILLISECONDS);
+        }
+        Assert.assertEquals(10, rate.get(TimeUnit.SECONDS), 0.0);
+    }
+
+    @Test
+    public void testUpdateAndGetPastWindowSize() throws InterruptedException
+    {
+        TestTimeSource time = new TestTimeSource();
+        SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+        int updates = 100;
+        for (int i = 0; i < updates; i++)
+        {
+            rate.update(1);
+        }
+
+        time.sleep(6, TimeUnit.SECONDS);
+
+        Assert.assertEquals(0, rate.get(TimeUnit.SECONDS), 0.0);
+    }
+
+    @Test
+    public void testUpdateAndGetToPointInTime() throws InterruptedException
+    {
+        TestTimeSource time = new TestTimeSource();
+        SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+        int updates = 10;
+        for (int i = 0; i < updates; i++)
+        {
+            rate.update(1);
+            time.sleep(100, TimeUnit.MILLISECONDS);
+        }
+
+        time.sleep(1, TimeUnit.SECONDS);
+
+        Assert.assertEquals(5, rate.get(TimeUnit.SECONDS), 0.0);
+        Assert.assertEquals(10, rate.get(1, TimeUnit.SECONDS), 0.0);
+    }
+
+    @Test
+    public void testDecay() throws InterruptedException
+    {
+        TestTimeSource time = new TestTimeSource();
+        SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+        int updates = 10;
+        for (int i = 0; i < updates; i++)
+        {
+            rate.update(1);
+            time.sleep(100, TimeUnit.MILLISECONDS);
+        }
+        Assert.assertEquals(10, rate.get(TimeUnit.SECONDS), 0.0);
+
+        time.sleep(1, TimeUnit.SECONDS);
+
+        Assert.assertEquals(5, rate.get(TimeUnit.SECONDS), 0.0);
+
+        time.sleep(2, TimeUnit.SECONDS);
+
+        Assert.assertEquals(2.5, rate.get(TimeUnit.SECONDS), 0.0);
+    }
+
+    @Test
+    public void testPruning() throws InterruptedException
+    {
+        TestTimeSource time = new TestTimeSource();
+        SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+
+        rate.update(1);
+        Assert.assertEquals(1, rate.size());
+
+        time.sleep(6, TimeUnit.SECONDS);
+
+        rate.prune();
+        Assert.assertEquals(0, rate.size());
+    }
+
+    @Test
+    public void testConcurrentUpdateAndGet() throws InterruptedException
+    {
+        final ExecutorService executor = 
Executors.newFixedThreadPool(FBUtilities.getAvailableProcessors());
+        final TestTimeSource time = new TestTimeSource();
+        final SlidingTimeRate rate = new SlidingTimeRate(time, 5, 1, 
TimeUnit.SECONDS);
+        int updates = 100000;
+        for (int i = 0; i < updates; i++)
+        {
+            executor.submit(() -> {
+                time.sleep(1, TimeUnit.MILLISECONDS);
+                rate.update(1);
+            });
+        }
+
+        executor.shutdown();
+
+        Assert.assertTrue(executor.awaitTermination(1, TimeUnit.MINUTES));
+        Assert.assertEquals(1000, rate.get(TimeUnit.SECONDS), 100.0);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d43b9ce5/test/unit/org/apache/cassandra/utils/TestTimeSource.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/utils/TestTimeSource.java 
b/test/unit/org/apache/cassandra/utils/TestTimeSource.java
new file mode 100644
index 0000000..4ecd086
--- /dev/null
+++ b/test/unit/org/apache/cassandra/utils/TestTimeSource.java
@@ -0,0 +1,72 @@
+/*
+ * 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.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class TestTimeSource implements TimeSource
+{
+    private final AtomicLong timeInMillis = new 
AtomicLong(System.currentTimeMillis());
+
+    @Override
+    public long currentTimeMillis()
+    {
+        return timeInMillis.get();
+    }
+
+    @Override
+    public long nanoTime()
+    {
+        return timeInMillis.get() * 1_000_000;
+    }
+
+    @Override
+    public TimeSource sleep(long sleepFor, TimeUnit unit)
+    {
+        long current = timeInMillis.get();
+        long sleepInMillis = TimeUnit.MILLISECONDS.convert(sleepFor, unit);
+        boolean elapsed;
+        do
+        {
+            long newTime = current + sleepInMillis;
+            elapsed = timeInMillis.compareAndSet(current, newTime);
+            if (!elapsed)
+            {
+                long updated = timeInMillis.get();
+                if (updated - current >= sleepInMillis)
+                {
+                    elapsed = true;
+                }
+                else
+                {
+                    sleepInMillis -= updated - current;
+                    current = updated;
+                }
+            }
+        }
+        while (!elapsed);
+        return this;
+    }
+
+    @Override
+    public TimeSource sleepUninterruptibly(long sleepFor, TimeUnit unit)
+    {
+        return sleep(sleepFor, unit);
+    }
+}

Reply via email to