홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
Committed by Gerrit Code Review
08960a48 1 parent 723f5531

ONOS-2068 Refresh leaderboard from source every 2s 

Change-Id: I99a6bc6a7bada6147abcab005d86d74204704c21
Inline Side-by-side
Showing 1 changed file with 35 additions and 16 deletions
......@@ -89,7 +89,7 @@ public class DistributedLeadershipManager implements LeadershipService {
private ScheduledExecutorService electionRunner;
private ScheduledExecutorService lockExecutor;
private ScheduledExecutorService staleLeadershipPurgeExecutor;
private ScheduledExecutorService leadershipStatusBroadcaster;
private ScheduledExecutorService leadershipRefresher;
private ConsistentMap<String, NodeId> leaderMap;
private ConsistentMap<String, List<NodeId>> candidateMap;
......@@ -106,7 +106,7 @@ public class DistributedLeadershipManager implements LeadershipService {
// The actual delay is randomly chosen between the interval [0, WAIT_BEFORE_RETRY_MILLIS)
private static final int WAIT_BEFORE_RETRY_MILLIS = 150;
private static final int DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC = 2;
private static final int LEADERSHIP_STATUS_UPDATE_INTERVAL_SEC = 2;
private static final int LEADERSHIP_REFRESH_INTERVAL_SEC = 2;
private static final int DELAY_BETWEEN_STALE_LEADERSHIP_PURGE_ATTEMPTS_SEC = 2;
private final AtomicBoolean staleLeadershipPurgeScheduled = new AtomicBoolean(false);
......@@ -135,8 +135,8 @@ public class DistributedLeadershipManager implements LeadershipService {
4, groupedThreads("onos/store/leadership", "election-thread-%d"));
staleLeadershipPurgeExecutor = Executors.newSingleThreadScheduledExecutor(
groupedThreads("onos/store/leadership", "stale-leadership-evictor"));
leadershipStatusBroadcaster = Executors.newSingleThreadScheduledExecutor(
groupedThreads("onos/store/leadership", "peer-updater"));
leadershipRefresher = Executors.newSingleThreadScheduledExecutor(
groupedThreads("onos/store/leadership", "refresh-thread"));
clusterCommunicator.addSubscriber(
LEADERSHIP_EVENT_MESSAGE_SUBJECT,
SERIALIZER::decode,
......@@ -148,8 +148,8 @@ public class DistributedLeadershipManager implements LeadershipService {
electionRunner.scheduleWithFixedDelay(
this::electLeaders, 0, DELAY_BETWEEN_LEADER_LOCK_ATTEMPTS_SEC, TimeUnit.SECONDS);
leadershipStatusBroadcaster.scheduleWithFixedDelay(
this::sendLeadershipStatus, 0, LEADERSHIP_STATUS_UPDATE_INTERVAL_SEC, TimeUnit.SECONDS);
leadershipRefresher.scheduleWithFixedDelay(
this::refreshLeaderBoard, 0, LEADERSHIP_REFRESH_INTERVAL_SEC, TimeUnit.SECONDS);
listenerRegistry = new ListenerRegistry<>();
eventDispatcher.addSink(LeadershipEvent.class, listenerRegistry);
......@@ -173,7 +173,7 @@ public class DistributedLeadershipManager implements LeadershipService {
messageHandlingExecutor.shutdown();
lockExecutor.shutdown();
staleLeadershipPurgeExecutor.shutdown();
leadershipStatusBroadcaster.shutdown();
leadershipRefresher.shutdown();
log.info("Stopped");
}
......@@ -458,6 +458,7 @@ public class DistributedLeadershipManager implements LeadershipService {
leaderBoard.compute(topic, (k, currentLeadership) -> {
if (currentLeadership == null || currentLeadership.epoch() <= leadershipUpdate.epoch()) {
updateAccepted.set(true);
// FIXME: Removing entries from leaderboard is not safe and should be visited.
return null;
}
return currentLeadership;
......@@ -579,18 +580,36 @@ public class DistributedLeadershipManager implements LeadershipService {
}
}
private void sendLeadershipStatus() {
private void refreshLeaderBoard() {
try {
leaderBoard.forEach((path, leadership) -> {
if (leadership.leader().equals(localNodeId)) {
LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership);
clusterCommunicator.broadcast(event,
LEADERSHIP_EVENT_MESSAGE_SUBJECT,
SERIALIZER::encode);
}
Map<String, Leadership> newLeaderBoard = Maps.newHashMap();
leaderMap.entrySet().forEach(entry -> {
String path = entry.getKey();
Versioned<NodeId> leader = entry.getValue();
Leadership leadership = new Leadership(path,
leader.value(),
leader.version(),
leader.creationTime());
newLeaderBoard.put(path, leadership);
});
// first take snapshot of current leader board.
Map<String, Leadership> currentLeaderBoard = ImmutableMap.copyOf(leaderBoard);
// evict stale leaders
Maps.difference(currentLeaderBoard, newLeaderBoard).entriesOnlyOnLeft().forEach((path, leadership) -> {
log.debug("Evicting {} from leaderboard. It is no longer active leader.", leadership);
onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_BOOTED, leadership));
});
// add missing leaders
Maps.difference(currentLeaderBoard, newLeaderBoard).entriesDiffering().forEach((path, difference) -> {
Leadership leadership = difference.rightValue();
log.debug("Adding {} to leaderboard. It is now the active leader.", leadership);
onLeadershipEvent(new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership));
});
} catch (Exception e) {
log.debug("Failed to send leadership updates", e);
log.debug("Failed to refresh leader board", e);
}
}
......