c5a3a032d463e99b7744e38fc619c4102f6f9a3a
[controller.git] / opendaylight / md-sal / sal-dummy-distributed-datastore / src / main / java / org / opendaylight / controller / dummy / datastore / DummyShard.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.dummy.datastore;
10
11 import akka.actor.Props;
12 import akka.actor.UntypedActor;
13 import com.google.common.base.Stopwatch;
14 import java.util.concurrent.TimeUnit;
15 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
16 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
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.InstallSnapshot;
20 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
21 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
22 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
23 import org.slf4j.Logger;
24 import org.slf4j.LoggerFactory;
25
26 public class DummyShard extends UntypedActor {
27     private static final Logger LOG = LoggerFactory.getLogger(DummyShard.class);
28
29     private final Configuration configuration;
30     private final String followerId;
31     private long lastMessageIndex  = -1;
32     private long lastMessageSize = 0;
33     private Stopwatch appendEntriesWatch;
34
35     public DummyShard(Configuration configuration, String followerId) {
36         this.configuration = configuration;
37         this.followerId = followerId;
38         LOG.info("Creating : {}", followerId);
39     }
40
41     @Override
42     public void onReceive(Object message) throws Exception {
43         if (message instanceof RequestVote) {
44             RequestVote req = (RequestVote) message;
45             sender().tell(new RequestVoteReply(req.getTerm(), true), self());
46         } else if (message instanceof AppendEntries) {
47             handleAppendEntries((AppendEntries) message);
48         } else if (message instanceof InstallSnapshot) {
49             handleInstallSnapshot((InstallSnapshot) message);
50         } else {
51             LOG.error("Unknown message : {}", message.getClass());
52         }
53     }
54
55     private void handleInstallSnapshot(InstallSnapshot req) {
56         sender().tell(new InstallSnapshotReply(req.getTerm(), followerId, req.getChunkIndex(), true), self());
57     }
58
59     protected void handleAppendEntries(AppendEntries req) throws InterruptedException {
60         LOG.info("{} - Received AppendEntries message : leader term = {}, index = {}, prevLogIndex = {}, size = {}",
61                 followerId, req.getTerm(),req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
62
63         if (appendEntriesWatch != null) {
64             long elapsed = appendEntriesWatch.elapsed(TimeUnit.SECONDS);
65             if (elapsed >= 5) {
66                 LOG.error("More than 5 seconds since last append entry, elapsed Time = {} seconds"
67                                 + ", leaderCommit = {}, prevLogIndex = {}, size = {}",
68                         elapsed, req.getLeaderCommit(), req.getPrevLogIndex(), req.getEntries().size());
69             }
70             appendEntriesWatch.reset().start();
71         } else {
72             appendEntriesWatch = Stopwatch.createStarted();
73         }
74
75         if (lastMessageIndex == req.getLeaderCommit() && req.getEntries().size() > 0 && lastMessageSize > 0) {
76             LOG.error("{} - Duplicate message with leaderCommit = {} prevLogIndex = {} received", followerId,
77                     req.getLeaderCommit(), req.getPrevLogIndex());
78         }
79
80         lastMessageIndex = req.getLeaderCommit();
81         lastMessageSize = req.getEntries().size();
82
83         long lastIndex = req.getLeaderCommit();
84         if (req.getEntries().size() > 0) {
85             for (ReplicatedLogEntry entry : req.getEntries()) {
86                 lastIndex = entry.getIndex();
87             }
88         }
89
90         if (configuration.shouldCauseTrouble() && req.getEntries().size() > 0) {
91             boolean ignore = false;
92
93             if (configuration.shouldDropReplies()) {
94                 ignore = Math.random() > 0.5;
95             }
96
97             long delay = (long) (Math.random() * configuration.getMaxDelayInMillis());
98
99             if (!ignore) {
100                 LOG.info("{} - Randomizing delay : {}", followerId, delay);
101                 Thread.sleep(delay);
102                 sender().tell(new AppendEntriesReply(followerId, req.getTerm(), true, lastIndex, req.getTerm(),
103                         DataStoreVersions.CURRENT_VERSION), self());
104             }
105         } else {
106             sender().tell(new AppendEntriesReply(followerId, req.getTerm(), true, lastIndex, req.getTerm(),
107                     DataStoreVersions.CURRENT_VERSION), self());
108         }
109     }
110
111     public static Props props(Configuration configuration, final String followerId) {
112         return Props.create(DummyShard.class, configuration, followerId);
113     }
114 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.