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 java.util.Set;
  14 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  15 import org.opendaylight.controller.cluster.raft.RaftState;
  16 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
  17 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  18 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  19 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  20 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  21 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  22
  23 /**
  24  * The behavior of a RaftActor when it is in the CandidateState
  25  * <p/>
  26  * Candidates (§5.2):
  27  * <ul>
  28  * <li> On conversion to candidate, start election:
  29  * <ul>
  30  * <li> Increment currentTerm
  31  * <li> Vote for self
  32  * <li> Reset election timer
  33  * <li> Send RequestVote RPCs to all other servers
  34  * </ul>
  35  * <li> If votes received from majority of servers: become leader
  36  * <li> If AppendEntries RPC received from new leader: convert to
  37  * follower
  38  * <li> If election timeout elapses: start new election
  39  * </ul>
  40  */
  41 public class Candidate extends AbstractRaftActorBehavior {
  42
  43     private int voteCount;
  44
  45     private final int votesRequired;
  46
  47     private final Set<String> peers;
  48
  49     public Candidate(RaftActorContext context) {
  50         super(context);
  51
  52         peers = context.getPeerAddresses().keySet();
  53
  54         if(LOG.isDebugEnabled()) {
  55             LOG.debug("{}: Election: Candidate has following peers: {}", context.getId(), peers);
  56         }
  57
  58         votesRequired = getMajorityVoteCount(peers.size());
  59
  60         startNewTerm();
  61         scheduleElection(electionDuration());
  62     }
  63
  64     @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
  65         AppendEntries appendEntries) {
  66
  67         if(LOG.isDebugEnabled()) {
  68             LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
  69         }
  70
  71         return this;
  72     }
  73
  74     @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
  75         AppendEntriesReply appendEntriesReply) {
  76
  77         return this;
  78     }
  79
  80     @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
  81         RequestVoteReply requestVoteReply) {
  82
  83         if (requestVoteReply.isVoteGranted()) {
  84             voteCount++;
  85         }
  86
  87         if (voteCount >= votesRequired) {
  88             return switchBehavior(new Leader(context));
  89         }
  90
  91         return this;
  92     }
  93
  94     @Override public RaftState state() {
  95         return RaftState.Candidate;
  96     }
  97
  98     @Override
  99     public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 100
 101         Object message = fromSerializableMessage(originalMessage);
 102
 103         if (message instanceof RaftRPC) {
 104
 105             RaftRPC rpc = (RaftRPC) message;
 106
 107             if(LOG.isDebugEnabled()) {
 108                 LOG.debug("{}: RaftRPC message received {} my term is {}", context.getId(), rpc,
 109                         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 {}", context.getId(), (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 }