opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.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 com.google.common.base.Preconditions;
  13 import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
  14 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  15 import org.opendaylight.controller.cluster.raft.RaftReplicator;
  16 import org.opendaylight.controller.cluster.raft.RaftState;
  17 import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
  18 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  19 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  20 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  21 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  22
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.List;
  26 import java.util.Map;
  27 import java.util.concurrent.atomic.AtomicLong;
  28
  29 /**
  30  * The behavior of a RaftActor when it is in the Leader state
  31  * <p>
  32  * Leaders:
  33  * <ul>
  34  * <li> Upon election: send initial empty AppendEntries RPCs
  35  * (heartbeat) to each server; repeat during idle periods to
  36  * prevent election timeouts (§5.2)
  37  * <li> If command received from client: append entry to local log,
  38  * respond after entry applied to state machine (§5.3)
  39  * <li> If last log index ≥ nextIndex for a follower: send
  40  * AppendEntries RPC with log entries starting at nextIndex
  41  * <ul>
  42  * <li> If successful: update nextIndex and matchIndex for
  43  * follower (§5.3)
  44  * <li> If AppendEntries fails because of log inconsistency:
  45  * decrement nextIndex and retry (§5.3)
  46  * </ul>
  47  * <li> If there exists an N such that N > commitIndex, a majority
  48  * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
  49  * set commitIndex = N (§5.3, §5.4).
  50  */
  51 public class Leader extends AbstractRaftActorBehavior {
  52
  53
  54     private final Map<String, ActorRef> followerToReplicator = new HashMap<>();
  55
  56     public Leader(RaftActorContext context, List<String> followers){
  57         super(context);
  58
  59         for(String follower : followers) {
  60
  61             ActorRef replicator = context.actorOf(
  62                 RaftReplicator.props(
  63                     new FollowerLogInformationImpl(follower,
  64                         new AtomicLong(0),
  65                         new AtomicLong(0)),
  66                     context.getActor()
  67                 )
  68             );
  69
  70             // Create a replicator for each follower
  71             followerToReplicator.put(follower, replicator);
  72
  73         }
  74
  75     }
  76
  77     @Override protected RaftState handleAppendEntries(ActorRef sender,
  78         AppendEntries appendEntries, RaftState suggestedState) {
  79         return suggestedState;
  80     }
  81
  82     @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
  83         AppendEntriesReply appendEntriesReply, RaftState suggestedState) {
  84         return suggestedState;
  85     }
  86
  87     @Override protected RaftState handleRequestVote(ActorRef sender,
  88         RequestVote requestVote, RaftState suggestedState) {
  89         return suggestedState;
  90     }
  91
  92     @Override protected RaftState handleRequestVoteReply(ActorRef sender,
  93         RequestVoteReply requestVoteReply, RaftState suggestedState) {
  94         return suggestedState;
  95     }
  96
  97     @Override protected RaftState state() {
  98         return RaftState.Leader;
  99     }
 100
 101     @Override public RaftState handleMessage(ActorRef sender, Object message) {
 102         Preconditions.checkNotNull(sender, "sender should not be null");
 103
 104         if(message instanceof SendHeartBeat) {
 105             sender.tell(new AppendEntries(
 106                 context.getTermInformation().getCurrentTerm().get() , context.getId(),
 107                 context.getReplicatedLog().last().getIndex(),
 108                 context.getReplicatedLog().last().getTerm(),
 109                 Collections.EMPTY_LIST, context.getCommitIndex().get()), context.getActor());
 110             return state();
 111         }
 112         return super.handleMessage(sender, message);
 113     }
 114 }