2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSelection;
13 import akka.actor.Props;
14 import akka.event.Logging;
15 import akka.event.LoggingAdapter;
16 import akka.japi.Creator;
17 import akka.serialization.Serialization;
18 import com.google.common.base.Optional;
19 import com.google.common.base.Preconditions;
20 import com.google.common.util.concurrent.ListenableFuture;
21 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
22 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
23 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
24 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
25 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
26 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
27 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
28 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
29 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
30 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
31 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
32 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
33 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
34 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
35 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
36 import org.opendaylight.controller.cluster.datastore.modification.Modification;
37 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
38 import org.opendaylight.controller.cluster.raft.ConfigParams;
39 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
40 import org.opendaylight.controller.cluster.raft.RaftActor;
41 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
42 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
43 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
44 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
45 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
46 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
47 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
48 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
49 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
50 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
51 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
52 import scala.concurrent.duration.FiniteDuration;
54 import java.util.ArrayList;
55 import java.util.Date;
56 import java.util.HashMap;
57 import java.util.List;
59 import java.util.concurrent.ExecutionException;
60 import java.util.concurrent.TimeUnit;
63 * A Shard represents a portion of the logical data tree <br/>
65 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
68 public class Shard extends RaftActor {
70 private static final ConfigParams configParams = new ShardConfigParams();
72 public static final String DEFAULT_NAME = "default";
74 // The state of this Shard
75 private final InMemoryDOMDataStore store;
77 private final Map<Object, DOMStoreThreePhaseCommitCohort>
78 modificationToCohort = new HashMap<>();
80 private final LoggingAdapter LOG =
81 Logging.getLogger(getContext().system(), this);
83 // By default persistent will be true and can be turned off using the system
84 // property shard.persistent
85 private final boolean persistent;
87 /// The name of this shard
88 private final ShardIdentifier name;
90 private volatile SchemaContext schemaContext;
92 private final ShardStats shardMBean;
94 private final List<ActorSelection> dataChangeListeners = new ArrayList<>();
96 private Shard(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
97 InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
98 super(name.toString(), mapPeerAddresses(peerAddresses), Optional.of(configParams));
102 String setting = System.getProperty("shard.persistent");
104 this.persistent = !"false".equals(setting);
106 LOG.info("Shard created : {} persistent : {}", name, persistent);
108 store = InMemoryDOMDataStoreFactory.create(name.toString(), null, dataStoreProperties);
110 shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString());
114 private static Map<String, String> mapPeerAddresses(Map<ShardIdentifier, String> peerAddresses){
115 Map<String , String> map = new HashMap<>();
117 for(Map.Entry<ShardIdentifier, String> entry : peerAddresses.entrySet()){
118 map.put(entry.getKey().toString(), entry.getValue());
124 public static Props props(final ShardIdentifier name,
125 final Map<ShardIdentifier, String> peerAddresses,
126 final InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
127 Preconditions.checkNotNull(name, "name should not be null");
128 Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
130 return Props.create(new Creator<Shard>() {
133 public Shard create() throws Exception {
134 return new Shard(name, peerAddresses, dataStoreProperties);
141 @Override public void onReceiveCommand(Object message) {
142 LOG.debug("Received message {} from {}", message.getClass().toString(),
145 if (message.getClass()
146 .equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
148 createTransactionChain();
149 } else if (getLeader() != null) {
150 getLeader().forward(message, getContext());
152 } else if (message instanceof RegisterChangeListener) {
153 registerChangeListener((RegisterChangeListener) message);
154 } else if (message instanceof UpdateSchemaContext) {
155 updateSchemaContext((UpdateSchemaContext) message);
156 } else if (message instanceof ForwardedCommitTransaction) {
157 handleForwardedCommit((ForwardedCommitTransaction) message);
158 } else if (message.getClass()
159 .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
161 createTransaction(CreateTransaction.fromSerializable(message));
162 } else if (getLeader() != null) {
163 getLeader().forward(message, getContext());
165 } else if (message instanceof PeerAddressResolved) {
166 PeerAddressResolved resolved = (PeerAddressResolved) message;
167 setPeerAddress(resolved.getPeerId().toString(), resolved.getPeerAddress());
169 super.onReceiveCommand(message);
173 private ActorRef createTypedTransactionActor(
174 CreateTransaction createTransaction, ShardTransactionIdentifier transactionId) {
175 if (createTransaction.getTransactionType()
176 == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
178 shardMBean.incrementReadOnlyTransactionCount();
180 return getContext().actorOf(
182 .props(store.newReadOnlyTransaction(), getSelf(),
183 schemaContext), transactionId.toString());
185 } else if (createTransaction.getTransactionType()
186 == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
188 shardMBean.incrementReadWriteTransactionCount();
190 return getContext().actorOf(
192 .props(store.newReadWriteTransaction(), getSelf(),
193 schemaContext), transactionId.toString());
196 } else if (createTransaction.getTransactionType()
197 == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
199 shardMBean.incrementWriteOnlyTransactionCount();
201 return getContext().actorOf(
203 .props(store.newWriteOnlyTransaction(), getSelf(),
204 schemaContext), transactionId.toString());
206 // FIXME: This does not seem right
207 throw new IllegalArgumentException(
208 "CreateTransaction message has unidentified transaction type="
209 + createTransaction.getTransactionType());
213 private void createTransaction(CreateTransaction createTransaction) {
215 ShardTransactionIdentifier transactionId = ShardTransactionIdentifier.builder().remoteTransactionId(createTransaction.getTransactionId()).build();
216 LOG.debug("Creating transaction : {} ", transactionId);
217 ActorRef transactionActor =
218 createTypedTransactionActor(createTransaction, transactionId);
221 .tell(new CreateTransactionReply(
222 Serialization.serializedActorPath(transactionActor),
223 createTransaction.getTransactionId()).toSerializable(),
227 private void commit(final ActorRef sender, Object serialized) {
228 Modification modification = MutableCompositeModification
229 .fromSerializable(serialized, schemaContext);
230 DOMStoreThreePhaseCommitCohort cohort =
231 modificationToCohort.remove(serialized);
232 if (cohort == null) {
234 "Could not find cohort for modification : {}. Writing modification using a new transaction",
236 DOMStoreReadWriteTransaction transaction =
237 store.newReadWriteTransaction();
238 modification.apply(transaction);
239 DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
240 ListenableFuture<Void> future =
241 commitCohort.preCommit();
244 future = commitCohort.commit();
246 } catch (InterruptedException | ExecutionException e) {
247 shardMBean.incrementFailedTransactionsCount();
248 LOG.error("Failed to commit", e);
251 //we want to just apply the recovery commit and return
252 shardMBean.incrementCommittedTransactionCount();
256 final ListenableFuture<Void> future = cohort.commit();
257 final ActorRef self = getSelf();
258 future.addListener(new Runnable() {
264 .tell(new CommitTransactionReply().toSerializable(),
266 shardMBean.incrementCommittedTransactionCount();
267 shardMBean.setLastCommittedTransactionTime(new Date());
268 } catch (InterruptedException | ExecutionException e) {
269 shardMBean.incrementFailedTransactionsCount();
270 sender.tell(new akka.actor.Status.Failure(e),self);
273 }, getContext().dispatcher());
276 private void handleForwardedCommit(ForwardedCommitTransaction message) {
277 Object serializedModification =
278 message.getModification().toSerializable();
281 .put(serializedModification, message.getCohort());
284 this.persistData(getSender(), "identifier",
285 new CompositeModificationPayload(serializedModification));
287 this.commit(getSender(), serializedModification);
291 private void updateSchemaContext(UpdateSchemaContext message) {
292 this.schemaContext = message.getSchemaContext();
293 store.onGlobalContextUpdated(message.getSchemaContext());
296 private void registerChangeListener(
297 RegisterChangeListener registerChangeListener) {
299 LOG.debug("registerDataChangeListener for {}", registerChangeListener
303 ActorSelection dataChangeListenerPath = getContext()
304 .system().actorSelection(
305 registerChangeListener.getDataChangeListenerPath());
308 // Notify the listener if notifications should be enabled or not
309 // If this shard is the leader then it will enable notifications else
311 dataChangeListenerPath
312 .tell(new EnableNotification(isLeader()), getSelf());
314 // Now store a reference to the data change listener so it can be notified
315 // at a later point if notifications should be enabled or disabled
316 dataChangeListeners.add(dataChangeListenerPath);
318 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
320 new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
322 org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
324 store.registerChangeListener(registerChangeListener.getPath(),
325 listener, registerChangeListener.getScope());
326 ActorRef listenerRegistration =
327 getContext().actorOf(
328 DataChangeListenerRegistration.props(registration));
331 "registerDataChangeListener sending reply, listenerRegistrationPath = {} "
332 , listenerRegistration.path().toString());
335 .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
339 private void createTransactionChain() {
340 DOMStoreTransactionChain chain = store.createTransactionChain();
341 ActorRef transactionChain =
342 getContext().actorOf(
343 ShardTransactionChain.props(chain, schemaContext));
345 .tell(new CreateTransactionChainReply(transactionChain.path())
350 @Override protected void applyState(ActorRef clientActor, String identifier,
353 if (data instanceof CompositeModificationPayload) {
354 Object modification =
355 ((CompositeModificationPayload) data).getModification();
357 if (modification != null) {
358 commit(clientActor, modification);
361 "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
362 identifier, clientActor.path().toString());
367 LOG.error("Unknown state received {}", data);
371 ReplicatedLogEntry lastLogEntry = getLastLogEntry();
373 if(lastLogEntry != null){
374 shardMBean.setLastLogIndex(lastLogEntry.getIndex());
375 shardMBean.setLastLogTerm(lastLogEntry.getTerm());
378 shardMBean.setCommitIndex(getCommitIndex());
379 shardMBean.setLastApplied(getLastApplied());
383 @Override protected Object createSnapshot() {
384 throw new UnsupportedOperationException("createSnapshot");
387 @Override protected void applySnapshot(Object snapshot) {
388 throw new UnsupportedOperationException("applySnapshot");
391 @Override protected void onStateChanged() {
392 for (ActorSelection dataChangeListener : dataChangeListeners) {
394 .tell(new EnableNotification(isLeader()), getSelf());
397 if (getLeaderId() != null) {
398 shardMBean.setLeader(getLeaderId());
401 shardMBean.setRaftState(getRaftState().name());
402 shardMBean.setCurrentTerm(getCurrentTerm());
405 @Override public String persistenceId() {
406 return this.name.toString();
410 private static class ShardConfigParams extends DefaultConfigParamsImpl {
411 public static final FiniteDuration HEART_BEAT_INTERVAL =
412 new FiniteDuration(500, TimeUnit.MILLISECONDS);
414 @Override public FiniteDuration getHeartBeatInterval() {
415 return HEART_BEAT_INTERVAL;