opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Candidate.java

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft.behaviors;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSelection;
  13 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  14 import org.opendaylight.controller.cluster.raft.RaftState;
  15 import org.opendaylight.controller.cluster.raft.internal.messages.ElectionTimeout;
  16 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  17 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  18 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  19 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  20
  21 import java.util.Collection;
  22 import java.util.HashMap;
  23 import java.util.Map;
  24
  25 /**
  26  * The behavior of a RaftActor when it is in the CandidateState
  27  * <p/>
  28  * Candidates (§5.2):
  29  * <ul>
  30  * <li> On conversion to candidate, start election:
  31  * <ul>
  32  * <li> Increment currentTerm
  33  * <li> Vote for self
  34  * <li> Reset election timer
  35  * <li> Send RequestVote RPCs to all other servers
  36  * </ul>
  37  * <li> If votes received from majority of servers: become leader
  38  * <li> If AppendEntries RPC received from new leader: convert to
  39  * follower
  40  * <li> If election timeout elapses: start new election
  41  * </ul>
  42  */
  43 public class Candidate extends AbstractRaftActorBehavior {
  44
  45     private final Map<String, ActorSelection> peerToActor = new HashMap<>();
  46
  47     private int voteCount;
  48
  49     private final int votesRequired;
  50
  51     public Candidate(RaftActorContext context) {
  52         super(context);
  53
  54         Collection<String> peerPaths = context.getPeerAddresses().values();
  55
  56         for (String peerPath : peerPaths) {
  57             peerToActor.put(peerPath,
  58                 context.actorSelection(peerPath));
  59         }
  60
  61         context.getLogger().debug("Election:Candidate has following peers:"+peerToActor.keySet());
  62         if(peerPaths.size() > 0) {
  63             // Votes are required from a majority of the peers including self.
  64             // The votesRequired field therefore stores a calculated value
  65             // of the number of votes required for this candidate to win an
  66             // election based on it's known peers.
  67             // If a peer was added during normal operation and raft replicas
  68             // came to know about them then the new peer would also need to be
  69             // taken into consideration when calculating this value.
  70             // Here are some examples for what the votesRequired would be for n
  71             // peers
  72             // 0 peers = 1 votesRequired (0 + 1) / 2 + 1 = 1
  73             // 2 peers = 2 votesRequired (2 + 1) / 2 + 1 = 2
  74             // 4 peers = 3 votesRequired (4 + 1) / 2 + 1 = 3
  75             int noOfPeers = peerPaths.size();
  76             int self = 1;
  77             votesRequired = (noOfPeers + self) / 2 + 1;
  78         } else {
  79             votesRequired = 0;
  80         }
  81
  82         startNewTerm();
  83         scheduleElection(electionDuration());
  84     }
  85
  86     @Override protected RaftState handleAppendEntries(ActorRef sender,
  87         AppendEntries appendEntries, RaftState suggestedState) {
  88
  89         context.getLogger().error("An unexpected AppendEntries received in state " + state());
  90
  91         return suggestedState;
  92     }
  93
  94     @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
  95         AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
  96
  97         // Some peer thinks I was a leader and sent me a reply
  98
  99         return suggestedState;
 100     }
 101
 102     @Override protected RaftState handleRequestVoteReply(ActorRef sender,
 103         RequestVoteReply requestVoteReply, RaftState suggestedState) {
 104         if (suggestedState == RaftState.Follower) {
 105             // If base class thinks I should be follower then I am
 106             return suggestedState;
 107         }
 108
 109         if (requestVoteReply.isVoteGranted()) {
 110             voteCount++;
 111         }
 112
 113         if (voteCount >= votesRequired) {
 114             return RaftState.Leader;
 115         }
 116
 117         return state();
 118     }
 119
 120     @Override public RaftState state() {
 121         return RaftState.Candidate;
 122     }
 123
 124     @Override
 125     public RaftState handleMessage(ActorRef sender, Object message) {
 126         if (message instanceof ElectionTimeout) {
 127             if (votesRequired == 0) {
 128                 // If there are no peers then we should be a Leader
 129                 // We wait for the election timeout to occur before declare
 130                 // ourselves the leader. This gives enough time for a leader
 131                 // who we do not know about (as a peer)
 132                 // to send a message to the candidate
 133                 return RaftState.Leader;
 134             }
 135             startNewTerm();
 136             scheduleElection(electionDuration());
 137             return state();
 138         }
 139         return super.handleMessage(sender, message);
 140     }
 141
 142
 143     private void startNewTerm() {
 144
 145
 146         // set voteCount back to 1 (that is voting for self)
 147         voteCount = 1;
 148
 149         // Increment the election term and vote for self
 150         long currentTerm = context.getTermInformation().getCurrentTerm();
 151         context.getTermInformation().update(currentTerm + 1, context.getId());
 152
 153         context.getLogger().debug("Starting new term " + (currentTerm+1));
 154
 155         // Request for a vote
 156         for (ActorSelection peerActor : peerToActor.values()) {
 157             peerActor.tell(new RequestVote(
 158                     context.getTermInformation().getCurrentTerm(),
 159                     context.getId(),
 160                     context.getReplicatedLog().lastIndex(),
 161                     context.getReplicatedLog().lastTerm()),
 162                 context.getActor()
 163             );
 164         }
 165
 166
 167     }
 168
 169     @Override public void close() throws Exception {
 170         stopElection();
 171     }
 172 }