dfaed9ba38c33fabfde2c2a8b894646ac1eedd2a
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / ReplicatedLogImpl.java
1 /*
2  * Copyright (c) 2015 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 package org.opendaylight.controller.cluster.raft;
9
10 import akka.japi.Procedure;
11 import com.google.common.base.Preconditions;
12 import java.util.Collections;
13 import java.util.List;
14 import org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries;
15
16 /**
17  * Implementation of ReplicatedLog used by the RaftActor.
18  */
19 class ReplicatedLogImpl extends AbstractReplicatedLogImpl {
20     private static final int DATA_SIZE_DIVIDER = 5;
21
22     private final Procedure<DeleteEntries> deleteProcedure = new Procedure<DeleteEntries>() {
23         @Override
24         public void apply(final DeleteEntries notUsed) {
25         }
26     };
27
28     private final RaftActorContext context;
29     private long dataSizeSinceLastSnapshot = 0L;
30
31     private ReplicatedLogImpl(final long snapshotIndex, final long snapshotTerm, final List<ReplicatedLogEntry> unAppliedEntries,
32             final RaftActorContext context) {
33         super(snapshotIndex, snapshotTerm, unAppliedEntries, context.getId());
34         this.context = Preconditions.checkNotNull(context);
35     }
36
37     static ReplicatedLog newInstance(final Snapshot snapshot, final RaftActorContext context) {
38         return new ReplicatedLogImpl(snapshot.getLastAppliedIndex(), snapshot.getLastAppliedTerm(),
39                 snapshot.getUnAppliedEntries(), context);
40     }
41
42     static ReplicatedLog newInstance(final RaftActorContext context) {
43         return new ReplicatedLogImpl(-1L, -1L, Collections.<ReplicatedLogEntry>emptyList(), context);
44     }
45
46     @Override
47     public boolean removeFromAndPersist(final long logEntryIndex) {
48         // FIXME: Maybe this should be done after the command is saved
49         long adjustedIndex = removeFrom(logEntryIndex);
50         if(adjustedIndex >= 0) {
51             context.getPersistenceProvider().persist(new DeleteEntries(adjustedIndex), deleteProcedure);
52             return true;
53         }
54
55         return false;
56     }
57
58     @Override
59     public void appendAndPersist(final ReplicatedLogEntry replicatedLogEntry) {
60         appendAndPersist(replicatedLogEntry, null);
61     }
62
63     @Override
64     public void captureSnapshotIfReady(final ReplicatedLogEntry replicatedLogEntry) {
65         final ConfigParams config = context.getConfigParams();
66         final long journalSize = replicatedLogEntry.getIndex() + 1;
67         final long dataThreshold = context.getTotalMemory() * config.getSnapshotDataThresholdPercentage() / 100;
68
69         if (journalSize % config.getSnapshotBatchCount() == 0 || getDataSizeForSnapshotCheck() > dataThreshold) {
70             boolean started = context.getSnapshotManager().capture(replicatedLogEntry,
71                     context.getCurrentBehavior().getReplicatedToAllIndex());
72             if (started && !context.hasFollowers()) {
73                 dataSizeSinceLastSnapshot = 0;
74             }
75         }
76     }
77
78     private long getDataSizeForSnapshotCheck() {
79         if (!context.hasFollowers()) {
80             // When we do not have followers we do not maintain an in-memory log
81             // due to this the journalSize will never become anything close to the
82             // snapshot batch count. In fact will mostly be 1.
83             // Similarly since the journal's dataSize depends on the entries in the
84             // journal the journal's dataSize will never reach a value close to the
85             // memory threshold.
86             // By maintaining the dataSize outside the journal we are tracking essentially
87             // what we have written to the disk however since we no longer are in
88             // need of doing a snapshot just for the sake of freeing up memory we adjust
89             // the real size of data by the DATA_SIZE_DIVIDER so that we do not snapshot as often
90             // as if we were maintaining a real snapshot
91             return dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER;
92         } else {
93             return dataSize();
94         }
95     }
96
97     @Override
98     public void appendAndPersist(final ReplicatedLogEntry replicatedLogEntry,
99             final Procedure<ReplicatedLogEntry> callback)  {
100
101         context.getLogger().debug("{}: Append log entry and persist {} ", context.getId(), replicatedLogEntry);
102
103         // FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
104         if(!append(replicatedLogEntry)) {
105             return;
106         }
107
108         // When persisting events with persist it is guaranteed that the
109         // persistent actor will not receive further commands between the
110         // persist call and the execution(s) of the associated event
111         // handler. This also holds for multiple persist calls in context
112         // of a single command.
113         context.getPersistenceProvider().persist(replicatedLogEntry,
114             new Procedure<ReplicatedLogEntry>() {
115                 @Override
116                 public void apply(final ReplicatedLogEntry param) throws Exception {
117                     context.getLogger().debug("{}: persist complete {}", context.getId(), param);
118
119                     int logEntrySize = param.size();
120                     dataSizeSinceLastSnapshot += logEntrySize;
121
122                     if (callback != null) {
123                         callback.apply(param);
124                     }
125                 }
126             }
127         );
128     }
129 }

©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.