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.base.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.RaftRPC;
  19 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  20 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  21
  22 import java.util.Set;
  23
  24 /**
  25  * The behavior of a RaftActor when it is in the CandidateState
  26  * <p/>
  27  * Candidates (§5.2):
  28  * <ul>
  29  * <li> On conversion to candidate, start election:
  30  * <ul>
  31  * <li> Increment currentTerm
  32  * <li> Vote for self
  33  * <li> Reset election timer
  34  * <li> Send RequestVote RPCs to all other servers
  35  * </ul>
  36  * <li> If votes received from majority of servers: become leader
  37  * <li> If AppendEntries RPC received from new leader: convert to
  38  * follower
  39  * <li> If election timeout elapses: start new election
  40  * </ul>
  41  */
  42 public class Candidate extends AbstractRaftActorBehavior {
  43
  44     private int voteCount;
  45
  46     private final int votesRequired;
  47
  48     private final Set<String> peers;
  49
  50     public Candidate(RaftActorContext context) {
  51         super(context);
  52
  53         peers = context.getPeerAddresses().keySet();
  54
  55         if(LOG.isDebugEnabled()) {
  56             LOG.debug("Election:Candidate has following peers: {}", peers);
  57         }
  58
  59         votesRequired = getMajorityVoteCount(peers.size());
  60
  61         startNewTerm();
  62         scheduleElection(electionDuration());
  63     }
  64
  65     @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
  66         AppendEntries appendEntries) {
  67
  68         if(LOG.isDebugEnabled()) {
  69             LOG.debug(appendEntries.toString());
  70         }
  71
  72         return this;
  73     }
  74
  75     @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
  76         AppendEntriesReply appendEntriesReply) {
  77
  78         return this;
  79     }
  80
  81     @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
  82         RequestVoteReply requestVoteReply) {
  83
  84         if (requestVoteReply.isVoteGranted()) {
  85             voteCount++;
  86         }
  87
  88         if (voteCount >= votesRequired) {
  89             return switchBehavior(new Leader(context));
  90         }
  91
  92         return this;
  93     }
  94
  95     @Override public RaftState state() {
  96         return RaftState.Candidate;
  97     }
  98
  99     @Override
 100     public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 101
 102         Object message = fromSerializableMessage(originalMessage);
 103
 104         if (message instanceof RaftRPC) {
 105
 106             RaftRPC rpc = (RaftRPC) message;
 107
 108             if(LOG.isDebugEnabled()) {
 109                 LOG.debug("RaftRPC message received {} my term is {}", rpc, context.getTermInformation().getCurrentTerm());
 110             }
 111
 112             // If RPC request or response contains term T > currentTerm:
 113             // set currentTerm = T, convert to follower (§5.1)
 114             // This applies to all RPC messages and responses
 115             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 116                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 117
 118                 return switchBehavior(new Follower(context));
 119             }
 120         }
 121
 122         if (message instanceof ElectionTimeout) {
 123             if (votesRequired == 0) {
 124                 // If there are no peers then we should be a Leader
 125                 // We wait for the election timeout to occur before declare
 126                 // ourselves the leader. This gives enough time for a leader
 127                 // who we do not know about (as a peer)
 128                 // to send a message to the candidate
 129
 130                 return switchBehavior(new Leader(context));
 131             }
 132             startNewTerm();
 133             scheduleElection(electionDuration());
 134             return this;
 135         }
 136
 137         return super.handleMessage(sender, message);
 138     }
 139
 140
 141     private void startNewTerm() {
 142
 143
 144         // set voteCount back to 1 (that is voting for self)
 145         voteCount = 1;
 146
 147         // Increment the election term and vote for self
 148         long currentTerm = context.getTermInformation().getCurrentTerm();
 149         context.getTermInformation().updateAndPersist(currentTerm + 1,
 150             context.getId());
 151
 152         if(LOG.isDebugEnabled()) {
 153             LOG.debug("Starting new term {}", (currentTerm + 1));
 154         }
 155
 156         // Request for a vote
 157         // TODO: Retry request for vote if replies do not arrive in a reasonable
 158         // amount of time TBD
 159         for (String peerId : peers) {
 160             ActorSelection peerActor = context.getPeerActorSelection(peerId);
 161             if(peerActor != null) {
 162                 peerActor.tell(new RequestVote(
 163                         context.getTermInformation().getCurrentTerm(),
 164                         context.getId(),
 165                         context.getReplicatedLog().lastIndex(),
 166                         context.getReplicatedLog().lastTerm()),
 167                     context.getActor()
 168                 );
 169             }
 170         }
 171
 172
 173     }
 174
 175     @Override public void close() throws Exception {
 176         stopElection();
 177     }
 178 }