Jonathan Hart
Committed by Ray Milkey

Don't run anti-entropy when under high load

Change-Id: I9e480708b9eced73da98e5c4cb27a18aeb08f09a
...@@ -19,6 +19,7 @@ import org.apache.commons.lang3.RandomUtils; ...@@ -19,6 +19,7 @@ import org.apache.commons.lang3.RandomUtils;
19 import org.apache.commons.lang3.mutable.MutableBoolean; 19 import org.apache.commons.lang3.mutable.MutableBoolean;
20 import org.apache.commons.lang3.tuple.Pair; 20 import org.apache.commons.lang3.tuple.Pair;
21 import org.onlab.util.KryoNamespace; 21 import org.onlab.util.KryoNamespace;
22 +import org.onlab.util.SlidingWindowCounter;
22 import org.onosproject.cluster.ClusterService; 23 import org.onosproject.cluster.ClusterService;
23 import org.onosproject.cluster.ControllerNode; 24 import org.onosproject.cluster.ControllerNode;
24 import org.onosproject.cluster.NodeId; 25 import org.onosproject.cluster.NodeId;
...@@ -49,6 +50,7 @@ import java.util.concurrent.ExecutorService; ...@@ -49,6 +50,7 @@ import java.util.concurrent.ExecutorService;
49 import java.util.concurrent.Executors; 50 import java.util.concurrent.Executors;
50 import java.util.concurrent.ScheduledExecutorService; 51 import java.util.concurrent.ScheduledExecutorService;
51 import java.util.concurrent.TimeUnit; 52 import java.util.concurrent.TimeUnit;
53 +import java.util.concurrent.atomic.AtomicLong;
52 import java.util.stream.Collectors; 54 import java.util.stream.Collectors;
53 55
54 import static com.google.common.base.Preconditions.checkNotNull; 56 import static com.google.common.base.Preconditions.checkNotNull;
...@@ -100,6 +102,12 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -100,6 +102,12 @@ public class EventuallyConsistentMapImpl<K, V>
100 private long periodSec = 5; 102 private long periodSec = 5;
101 private boolean lightweightAntiEntropy = true; 103 private boolean lightweightAntiEntropy = true;
102 104
105 + private static final int WINDOW_SIZE = 5;
106 + private static final int HIGH_LOAD_THRESHOLD = 0;
107 + private static final int LOAD_WINDOW = 2;
108 + SlidingWindowCounter counter = new SlidingWindowCounter(WINDOW_SIZE);
109 + AtomicLong operations = new AtomicLong();
110 +
103 /** 111 /**
104 * Creates a new eventually consistent map shared amongst multiple instances. 112 * Creates a new eventually consistent map shared amongst multiple instances.
105 * <p> 113 * <p>
...@@ -271,6 +279,7 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -271,6 +279,7 @@ public class EventuallyConsistentMapImpl<K, V>
271 } 279 }
272 280
273 private boolean putInternal(K key, V value, Timestamp timestamp) { 281 private boolean putInternal(K key, V value, Timestamp timestamp) {
282 + counter.incrementCount();
274 Timestamp removed = removedItems.get(key); 283 Timestamp removed = removedItems.get(key);
275 if (removed != null && removed.isNewerThan(timestamp)) { 284 if (removed != null && removed.isNewerThan(timestamp)) {
276 log.debug("ecmap - removed was newer {}", value); 285 log.debug("ecmap - removed was newer {}", value);
...@@ -318,6 +327,7 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -318,6 +327,7 @@ public class EventuallyConsistentMapImpl<K, V>
318 } 327 }
319 328
320 private boolean removeInternal(K key, Timestamp timestamp) { 329 private boolean removeInternal(K key, Timestamp timestamp) {
330 + counter.incrementCount();
321 final MutableBoolean updated = new MutableBoolean(false); 331 final MutableBoolean updated = new MutableBoolean(false);
322 332
323 items.compute(key, (k, existing) -> { 333 items.compute(key, (k, existing) -> {
...@@ -515,6 +525,10 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -515,6 +525,10 @@ public class EventuallyConsistentMapImpl<K, V>
515 clusterCommunicator.unicast(message, peer); 525 clusterCommunicator.unicast(message, peer);
516 } 526 }
517 527
528 + private boolean underHighLoad() {
529 + return counter.get(LOAD_WINDOW) > HIGH_LOAD_THRESHOLD;
530 + }
531 +
518 private final class SendAdvertisementTask implements Runnable { 532 private final class SendAdvertisementTask implements Runnable {
519 @Override 533 @Override
520 public void run() { 534 public void run() {
...@@ -523,6 +537,10 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -523,6 +537,10 @@ public class EventuallyConsistentMapImpl<K, V>
523 return; 537 return;
524 } 538 }
525 539
540 + if (underHighLoad()) {
541 + return;
542 + }
543 +
526 try { 544 try {
527 final NodeId self = clusterService.getLocalNode().id(); 545 final NodeId self = clusterService.getLocalNode().id();
528 Set<ControllerNode> nodes = clusterService.getNodes(); 546 Set<ControllerNode> nodes = clusterService.getNodes();
...@@ -745,7 +763,9 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -745,7 +763,9 @@ public class EventuallyConsistentMapImpl<K, V>
745 message.sender()); 763 message.sender());
746 AntiEntropyAdvertisement<K> advertisement = serializer.decode(message.payload()); 764 AntiEntropyAdvertisement<K> advertisement = serializer.decode(message.payload());
747 try { 765 try {
766 + if (!underHighLoad()) {
748 handleAntiEntropyAdvertisement(advertisement); 767 handleAntiEntropyAdvertisement(advertisement);
768 + }
749 } catch (Exception e) { 769 } catch (Exception e) {
750 log.warn("Exception thrown handling advertisements", e); 770 log.warn("Exception thrown handling advertisements", e);
751 } 771 }
......
1 +/*
2 + * Copyright 2015 Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onlab.util;
17 +
18 +import java.util.ArrayList;
19 +import java.util.Collections;
20 +import java.util.List;
21 +import java.util.concurrent.Executors;
22 +import java.util.concurrent.ScheduledExecutorService;
23 +import java.util.concurrent.TimeUnit;
24 +import java.util.concurrent.atomic.AtomicLong;
25 +import java.util.stream.Collectors;
26 +
27 +import static com.google.common.base.Preconditions.checkArgument;
28 +
29 +/**
30 + * Maintains a sliding window of value counts. The sliding window counter is
31 + * initialized with a number of window slots. Calls to #incrementCount() will
32 + * increment the value in the current window slot. Periodically the window
33 + * slides and the oldest value count is dropped. Calls to #get() will get the
34 + * total count for the last N window slots.
35 + */
36 +public final class SlidingWindowCounter {
37 +
38 + private volatile int headSlot;
39 + private final int windowSlots;
40 +
41 + private final List<AtomicLong> counters;
42 +
43 + private final ScheduledExecutorService background;
44 +
45 + private static final int SLIDE_WINDOW_PERIOD_SECONDS = 1;
46 +
47 + /**
48 + * Creates a new sliding window counter with the given total number of
49 + * window slots.
50 + *
51 + * @param windowSlots total number of window slots
52 + */
53 + public SlidingWindowCounter(int windowSlots) {
54 + checkArgument(windowSlots > 0, "Window size must be a positive integer");
55 +
56 + this.windowSlots = windowSlots;
57 + this.headSlot = 0;
58 +
59 + // Initialize each item in the list to an AtomicLong of 0
60 + this.counters = Collections.nCopies(windowSlots, 0)
61 + .stream()
62 + .map(AtomicLong::new)
63 + .collect(Collectors.toCollection(ArrayList::new));
64 +
65 + background = Executors.newSingleThreadScheduledExecutor();
66 + background.scheduleWithFixedDelay(this::advanceHead, 0,
67 + SLIDE_WINDOW_PERIOD_SECONDS, TimeUnit.SECONDS);
68 + }
69 +
70 + /**
71 + * Releases resources used by the SlidingWindowCounter.
72 + */
73 + public void destroy() {
74 + background.shutdownNow();
75 + }
76 +
77 + /**
78 + * Increments the count of the current window slot by 1.
79 + */
80 + public void incrementCount() {
81 + incrementCount(headSlot, 1);
82 + }
83 +
84 + /**
85 + * Increments the count of the current window slot by the given value.
86 + *
87 + * @param value value to increment by
88 + */
89 + public void incrementCount(long value) {
90 + incrementCount(headSlot, value);
91 + }
92 +
93 + private void incrementCount(int slot, long value) {
94 + counters.get(slot).addAndGet(value);
95 + }
96 +
97 + /**
98 + * Gets the total count for the last N window slots.
99 + *
100 + * @param slots number of slots to include in the count
101 + * @return total count for last N slots
102 + */
103 + public long get(int slots) {
104 + checkArgument(slots <= windowSlots,
105 + "Requested window must be less than the total window slots");
106 +
107 + long sum = 0;
108 +
109 + for (int i = 0; i < slots; i++) {
110 + int currentIndex = headSlot - i;
111 + if (currentIndex < 0) {
112 + currentIndex = counters.size() + currentIndex;
113 + }
114 + sum += counters.get(currentIndex).get();
115 + }
116 +
117 + return sum;
118 + }
119 +
120 + void advanceHead() {
121 + counters.get(slotAfter(headSlot)).set(0);
122 + headSlot = slotAfter(headSlot);
123 + }
124 +
125 + private int slotAfter(int slot) {
126 + return (slot + 1) % windowSlots;
127 + }
128 +
129 +}
1 +/*
2 + * Copyright 2015 Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onlab.util;
17 +
18 +import org.junit.After;
19 +import org.junit.Before;
20 +import org.junit.Test;
21 +
22 +import static junit.framework.TestCase.fail;
23 +import static org.junit.Assert.assertEquals;
24 +import static org.junit.Assert.assertTrue;
25 +
26 +/**
27 + * Unit tests for the sliding window counter.
28 + */
29 +public class SlidingWindowCounterTest {
30 +
31 + private SlidingWindowCounter counter;
32 +
33 + @Before
34 + public void setUp() {
35 + counter = new SlidingWindowCounter(2);
36 + }
37 +
38 + @After
39 + public void tearDown() {
40 + counter.destroy();
41 + }
42 +
43 + @Test
44 + public void testIncrementCount() {
45 + assertEquals(0, counter.get(1));
46 + assertEquals(0, counter.get(2));
47 + counter.incrementCount();
48 + assertEquals(1, counter.get(1));
49 + assertEquals(1, counter.get(2));
50 + counter.incrementCount(2);
51 + assertEquals(3, counter.get(2));
52 + }
53 +
54 + @Test
55 + public void testSlide() {
56 + counter.incrementCount();
57 + counter.advanceHead();
58 + assertEquals(0, counter.get(1));
59 + assertEquals(1, counter.get(2));
60 + counter.incrementCount(2);
61 + assertEquals(2, counter.get(1));
62 + assertEquals(3, counter.get(2));
63 + }
64 +
65 + @Test
66 + public void testWrap() {
67 + counter.incrementCount();
68 + counter.advanceHead();
69 + counter.incrementCount(2);
70 + counter.advanceHead();
71 + assertEquals(0, counter.get(1));
72 + assertEquals(2, counter.get(2));
73 + counter.advanceHead();
74 + assertEquals(0, counter.get(1));
75 + assertEquals(0, counter.get(2));
76 +
77 + }
78 +
79 + @Test
80 + public void testCornerCases() {
81 + try {
82 + counter.get(3);
83 + fail("Exception should have been thrown");
84 + } catch (IllegalArgumentException e) {
85 + assertTrue(true);
86 + }
87 +
88 + try {
89 + new SlidingWindowCounter(0);
90 + fail("Exception should have been thrown");
91 + } catch (IllegalArgumentException e) {
92 + assertTrue(true);
93 + }
94 +
95 + try {
96 + new SlidingWindowCounter(-1);
97 + fail("Exception should have been thrown");
98 + } catch (IllegalArgumentException e) {
99 + assertTrue(true);
100 + }
101 + }
102 +}