Jonathan Hart
Committed by Brian O'Connor

Add a configuration option to vary how heavyweight anti-entropy is.

Change-Id: I57cea61182b3d19deb47608ffb7dd617529ae34c
...@@ -98,6 +98,7 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -98,6 +98,7 @@ public class EventuallyConsistentMapImpl<K, V>
98 // TODO: Make these anti-entropy params configurable 98 // TODO: Make these anti-entropy params configurable
99 private long initialDelaySec = 5; 99 private long initialDelaySec = 5;
100 private long periodSec = 5; 100 private long periodSec = 5;
101 + private boolean lightweightAntiEntropy = true;
101 102
102 /** 103 /**
103 * Creates a new eventually consistent map shared amongst multiple instances. 104 * Creates a new eventually consistent map shared amongst multiple instances.
...@@ -567,35 +568,32 @@ public class EventuallyConsistentMapImpl<K, V> ...@@ -567,35 +568,32 @@ public class EventuallyConsistentMapImpl<K, V>
567 568
568 private void handleAntiEntropyAdvertisement(AntiEntropyAdvertisement<K> ad) { 569 private void handleAntiEntropyAdvertisement(AntiEntropyAdvertisement<K> ad) {
569 List<EventuallyConsistentMapEvent<K, V>> externalEvents; 570 List<EventuallyConsistentMapEvent<K, V>> externalEvents;
570 - boolean sync = false;
571 571
572 externalEvents = antiEntropyCheckLocalItems(ad); 572 externalEvents = antiEntropyCheckLocalItems(ad);
573 573
574 antiEntropyCheckLocalRemoved(ad); 574 antiEntropyCheckLocalRemoved(ad);
575 575
576 - externalEvents.addAll(antiEntropyCheckRemoteRemoved(ad)); 576 + if (!lightweightAntiEntropy) {
577 - 577 + externalEvents.addAll(antiEntropyCheckRemoteRemoved(ad));
578 - // if remote ad has something unknown, actively sync 578 +
579 - for (K key : ad.timestamps().keySet()) { 579 + // if remote ad has something unknown, actively sync
580 - if (!items.containsKey(key)) { 580 + for (K key : ad.timestamps().keySet()) {
581 - sync = true; 581 + if (!items.containsKey(key)) {
582 - break; 582 + // Send the advertisement back if this peer is out-of-sync
583 - } 583 + final NodeId sender = ad.sender();
584 - } 584 + AntiEntropyAdvertisement<K> myAd = createAdvertisement();
585 + try {
586 + unicastMessage(sender, antiEntropyAdvertisementSubject, myAd);
587 + } catch (IOException e) {
588 + log.debug(
589 + "Failed to send reactive anti-entropy advertisement to {}",
590 + sender);
591 + }
585 592
586 - // Send the advertisement back if this peer is out-of-sync 593 + break;
587 - if (sync) { 594 + }
588 - final NodeId sender = ad.sender();
589 - AntiEntropyAdvertisement<K> myAd = createAdvertisement();
590 - try {
591 - unicastMessage(sender, antiEntropyAdvertisementSubject, myAd);
592 - } catch (IOException e) {
593 - log.debug(
594 - "Failed to send reactive anti-entropy advertisement to {}",
595 - sender);
596 } 595 }
597 } 596 }
598 -
599 externalEvents.forEach(this::notifyListeners); 597 externalEvents.forEach(this::notifyListeners);
600 } 598 }
601 599
......