Do not leak DataTree from backend actor
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / shardmanager / ShardInformation.java
1 /*
2  * Copyright (c) 2016 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.datastore.shardmanager;
9
10 import static java.util.Objects.requireNonNull;
11
12 import akka.actor.ActorRef;
13 import akka.actor.Props;
14 import akka.serialization.Serialization;
15 import com.google.common.base.Preconditions;
16 import com.google.common.base.Strings;
17 import java.util.HashSet;
18 import java.util.Iterator;
19 import java.util.Map;
20 import java.util.Objects;
21 import java.util.Optional;
22 import java.util.Set;
23 import org.eclipse.jdt.annotation.Nullable;
24 import org.opendaylight.controller.cluster.access.concepts.MemberName;
25 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
26 import org.opendaylight.controller.cluster.datastore.Shard;
27 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
28 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
29 import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
30 import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
31 import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManager.OnShardInitialized;
32 import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManager.OnShardReady;
33 import org.opendaylight.controller.cluster.raft.RaftState;
34 import org.opendaylight.yangtools.yang.data.api.schema.tree.ReadOnlyDataTree;
35 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
36 import org.slf4j.Logger;
37 import org.slf4j.LoggerFactory;
38
39 final class ShardInformation {
40     private static final Logger LOG = LoggerFactory.getLogger(ShardInformation.class);
41
42     private final Set<OnShardInitialized> onShardInitializedSet = new HashSet<>();
43     private final Map<String, String> initialPeerAddresses;
44     private final ShardPeerAddressResolver addressResolver;
45     private final ShardIdentifier shardId;
46     private final String shardName;
47
48     // This reference indirection is required to have the ability to update the SchemaContext
49     // inside actor props. Otherwise we would be keeping an old SchemaContext there, preventing
50     // it from becoming garbage.
51     private final AtomicShardContextProvider schemaContextProvider = new AtomicShardContextProvider();
52     private ActorRef actor;
53
54     private Optional<ReadOnlyDataTree> localShardDataTree;
55     private boolean leaderAvailable = false;
56
57     // flag that determines if the actor is ready for business
58     private boolean actorInitialized = false;
59
60     private boolean followerSyncStatus = false;
61
62     private String role ;
63     private String leaderId;
64     private short leaderVersion;
65
66     private DatastoreContext datastoreContext;
67     private Shard.AbstractBuilder<?, ?> builder;
68     private boolean activeMember = true;
69
70     ShardInformation(final String shardName, final ShardIdentifier shardId,
71             final Map<String, String> initialPeerAddresses, final DatastoreContext datastoreContext,
72             final Shard.AbstractBuilder<?, ?> builder, final ShardPeerAddressResolver addressResolver) {
73         this.shardName = shardName;
74         this.shardId = shardId;
75         this.initialPeerAddresses = initialPeerAddresses;
76         this.datastoreContext = datastoreContext;
77         this.builder = builder;
78         this.addressResolver = addressResolver;
79     }
80
81     Props newProps() {
82         Props props = requireNonNull(builder).id(shardId).peerAddresses(initialPeerAddresses)
83                 .datastoreContext(datastoreContext).schemaContextProvider(schemaContextProvider).props();
84         builder = null;
85         return props;
86     }
87
88     String getShardName() {
89         return shardName;
90     }
91
92     @Nullable ActorRef getActor() {
93         return actor;
94     }
95
96     void setActor(final ActorRef actor) {
97         this.actor = actor;
98     }
99
100     ShardIdentifier getShardId() {
101         return shardId;
102     }
103
104     void setLocalDataTree(final Optional<ReadOnlyDataTree> dataTree) {
105         this.localShardDataTree = dataTree;
106     }
107
108     Optional<ReadOnlyDataTree> getLocalShardDataTree() {
109         return localShardDataTree;
110     }
111
112     DatastoreContext getDatastoreContext() {
113         return datastoreContext;
114     }
115
116     void setDatastoreContext(final DatastoreContext newDatastoreContext, final ActorRef sender) {
117         this.datastoreContext = newDatastoreContext;
118         if (actor != null) {
119             LOG.debug("Sending new DatastoreContext to {}", shardId);
120             actor.tell(this.datastoreContext, sender);
121         }
122     }
123
124     void updatePeerAddress(final String peerId, final String peerAddress, final ActorRef sender) {
125         LOG.info("updatePeerAddress for peer {} with address {}", peerId, peerAddress);
126
127         if (actor != null) {
128             LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}", peerId,
129                     peerAddress, actor.path());
130
131             actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
132         }
133
134         notifyOnShardInitializedCallbacks();
135     }
136
137     void peerDown(final MemberName memberName, final String peerId, final ActorRef sender) {
138         if (actor != null) {
139             actor.tell(new PeerDown(memberName, peerId), sender);
140         }
141     }
142
143     void peerUp(final MemberName memberName, final String peerId, final ActorRef sender) {
144         if (actor != null) {
145             actor.tell(new PeerUp(memberName, peerId), sender);
146         }
147     }
148
149     boolean isShardReady() {
150         return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
151     }
152
153     boolean isShardReadyWithLeaderId() {
154         return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role)
155                 && !RaftState.PreLeader.name().equals(role)
156                 && (isLeader() || addressResolver.resolve(leaderId) != null);
157     }
158
159     boolean isShardInitialized() {
160         return getActor() != null && actorInitialized;
161     }
162
163     boolean isLeader() {
164         return Objects.equals(leaderId, shardId.toString());
165     }
166
167     String getSerializedLeaderActor() {
168         if (isLeader()) {
169             return Serialization.serializedActorPath(getActor());
170         } else {
171             return addressResolver.resolve(leaderId);
172         }
173     }
174
175     void setActorInitialized() {
176         LOG.debug("Shard {} is initialized", shardId);
177
178         this.actorInitialized = true;
179
180         notifyOnShardInitializedCallbacks();
181     }
182
183     private void notifyOnShardInitializedCallbacks() {
184         if (onShardInitializedSet.isEmpty()) {
185             return;
186         }
187
188         boolean ready = isShardReadyWithLeaderId();
189
190         LOG.debug("Shard {} is {} - notifying {} OnShardInitialized callbacks", shardId,
191             ready ? "ready" : "initialized", onShardInitializedSet.size());
192
193         Iterator<OnShardInitialized> iter = onShardInitializedSet.iterator();
194         while (iter.hasNext()) {
195             OnShardInitialized onShardInitialized = iter.next();
196             if (!(onShardInitialized instanceof OnShardReady) || ready) {
197                 iter.remove();
198                 onShardInitialized.getTimeoutSchedule().cancel();
199                 onShardInitialized.getReplyRunnable().run();
200             }
201         }
202     }
203
204     void addOnShardInitialized(final OnShardInitialized onShardInitialized) {
205         onShardInitializedSet.add(onShardInitialized);
206     }
207
208     void removeOnShardInitialized(final OnShardInitialized onShardInitialized) {
209         onShardInitializedSet.remove(onShardInitialized);
210     }
211
212     void setRole(final String newRole) {
213         this.role = newRole;
214
215         notifyOnShardInitializedCallbacks();
216     }
217
218     String getRole() {
219         return role;
220     }
221
222     void setFollowerSyncStatus(final boolean syncStatus) {
223         this.followerSyncStatus = syncStatus;
224     }
225
226     boolean isInSync() {
227         if (RaftState.Follower.name().equals(this.role)) {
228             return followerSyncStatus;
229         } else if (RaftState.Leader.name().equals(this.role)) {
230             return true;
231         }
232
233         return false;
234     }
235
236     boolean setLeaderId(final String newLeaderId) {
237         final boolean changed = !Objects.equals(this.leaderId, newLeaderId);
238         this.leaderId = newLeaderId;
239         if (newLeaderId != null) {
240             this.leaderAvailable = true;
241         }
242         notifyOnShardInitializedCallbacks();
243
244         return changed;
245     }
246
247     String getLeaderId() {
248         return leaderId;
249     }
250
251     void setLeaderAvailable(final boolean leaderAvailable) {
252         this.leaderAvailable = leaderAvailable;
253
254         if (leaderAvailable) {
255             notifyOnShardInitializedCallbacks();
256         }
257     }
258
259     short getLeaderVersion() {
260         return leaderVersion;
261     }
262
263     void setLeaderVersion(final short leaderVersion) {
264         this.leaderVersion = leaderVersion;
265     }
266
267     boolean isActiveMember() {
268         return activeMember;
269     }
270
271     void setActiveMember(final boolean isActiveMember) {
272         this.activeMember = isActiveMember;
273     }
274
275     SchemaContext getSchemaContext() {
276         return schemaContextProvider.getSchemaContext();
277     }
278
279     void setSchemaContext(final SchemaContext schemaContext) {
280         schemaContextProvider.set(Preconditions.checkNotNull(schemaContext));
281     }
282
283     @Override
284     public String toString() {
285         return "ShardInformation [shardId=" + shardId + ", leaderAvailable=" + leaderAvailable + ", actorInitialized="
286                 + actorInitialized + ", followerSyncStatus=" + followerSyncStatus + ", role=" + role + ", leaderId="
287                 + leaderId + ", activeMember=" + activeMember + "]";
288     }
289
290
291 }

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