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.FutureCallback;
21 import com.google.common.util.concurrent.Futures;
22 import com.google.common.util.concurrent.ListenableFuture;
23 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
24 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
25 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
26 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
27 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
28 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
29 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
30 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
31 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
32 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
33 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
34 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
35 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
36 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
37 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
38 import org.opendaylight.controller.cluster.datastore.modification.Modification;
39 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
40 import org.opendaylight.controller.cluster.raft.ConfigParams;
41 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
42 import org.opendaylight.controller.cluster.raft.RaftActor;
43 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
44 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
45 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
46 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
47 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
48 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
49 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
50 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
51 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
52 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
53 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
54 import scala.concurrent.duration.FiniteDuration;
56 import java.util.ArrayList;
57 import java.util.Date;
58 import java.util.HashMap;
59 import java.util.List;
61 import java.util.concurrent.ExecutionException;
62 import java.util.concurrent.TimeUnit;
65 * A Shard represents a portion of the logical data tree <br/>
67 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
70 public class Shard extends RaftActor {
72 private static final ConfigParams configParams = new ShardConfigParams();
74 public static final String DEFAULT_NAME = "default";
76 // The state of this Shard
77 private final InMemoryDOMDataStore store;
79 private final Map<Object, DOMStoreThreePhaseCommitCohort>
80 modificationToCohort = new HashMap<>();
82 private final LoggingAdapter LOG =
83 Logging.getLogger(getContext().system(), this);
85 // By default persistent will be true and can be turned off using the system
86 // property shard.persistent
87 private final boolean persistent;
89 /// The name of this shard
90 private final ShardIdentifier name;
92 private volatile SchemaContext schemaContext;
94 private final ShardStats shardMBean;
96 private final List<ActorSelection> dataChangeListeners = new ArrayList<>();
98 private Shard(ShardIdentifier name, Map<ShardIdentifier, String> peerAddresses,
99 InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
100 super(name.toString(), mapPeerAddresses(peerAddresses), Optional.of(configParams));
104 String setting = System.getProperty("shard.persistent");
106 this.persistent = !"false".equals(setting);
108 LOG.info("Shard created : {} persistent : {}", name, persistent);
110 store = InMemoryDOMDataStoreFactory.create(name.toString(), null, dataStoreProperties);
112 shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString());
116 private static Map<String, String> mapPeerAddresses(
117 Map<ShardIdentifier, String> peerAddresses) {
118 Map<String, String> map = new HashMap<>();
120 for (Map.Entry<ShardIdentifier, String> entry : peerAddresses
122 map.put(entry.getKey().toString(), entry.getValue());
131 public static Props props(final ShardIdentifier name,
132 final Map<ShardIdentifier, String> peerAddresses,
133 final InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
134 Preconditions.checkNotNull(name, "name should not be null");
136 .checkNotNull(peerAddresses, "peerAddresses should not be null");
138 return Props.create(new Creator<Shard>() {
141 public Shard create() throws Exception {
142 return new Shard(name, peerAddresses, dataStoreProperties);
149 @Override public void onReceiveCommand(Object message) {
150 LOG.debug("Received message {} from {}", message.getClass().toString(),
153 if (message.getClass()
154 .equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
156 createTransactionChain();
157 } else if (getLeader() != null) {
158 getLeader().forward(message, getContext());
160 } else if (message instanceof RegisterChangeListener) {
161 registerChangeListener((RegisterChangeListener) message);
162 } else if (message instanceof UpdateSchemaContext) {
163 updateSchemaContext((UpdateSchemaContext) message);
164 } else if (message instanceof ForwardedCommitTransaction) {
165 handleForwardedCommit((ForwardedCommitTransaction) message);
166 } else if (message.getClass()
167 .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
169 createTransaction(CreateTransaction.fromSerializable(message));
170 } else if (getLeader() != null) {
171 getLeader().forward(message, getContext());
173 } else if (message instanceof PeerAddressResolved) {
174 PeerAddressResolved resolved = (PeerAddressResolved) message;
175 setPeerAddress(resolved.getPeerId().toString(),
176 resolved.getPeerAddress());
178 super.onReceiveCommand(message);
182 private ActorRef createTypedTransactionActor(
183 CreateTransaction createTransaction,
184 ShardTransactionIdentifier transactionId) {
185 if (createTransaction.getTransactionType()
186 == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
188 shardMBean.incrementReadOnlyTransactionCount();
190 return getContext().actorOf(
192 .props(store.newReadOnlyTransaction(), getSelf(),
193 schemaContext), transactionId.toString());
195 } else if (createTransaction.getTransactionType()
196 == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
198 shardMBean.incrementReadWriteTransactionCount();
200 return getContext().actorOf(
202 .props(store.newReadWriteTransaction(), getSelf(),
203 schemaContext), transactionId.toString());
206 } else if (createTransaction.getTransactionType()
207 == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
209 shardMBean.incrementWriteOnlyTransactionCount();
211 return getContext().actorOf(
213 .props(store.newWriteOnlyTransaction(), getSelf(),
214 schemaContext), transactionId.toString());
216 throw new IllegalArgumentException(
217 "Shard="+name + ":CreateTransaction message has unidentified transaction type="
218 + createTransaction.getTransactionType());
222 private void createTransaction(CreateTransaction createTransaction) {
224 ShardTransactionIdentifier transactionId =
225 ShardTransactionIdentifier.builder()
226 .remoteTransactionId(createTransaction.getTransactionId())
228 LOG.debug("Creating transaction : {} ", transactionId);
229 ActorRef transactionActor =
230 createTypedTransactionActor(createTransaction, transactionId);
233 .tell(new CreateTransactionReply(
234 Serialization.serializedActorPath(transactionActor),
235 createTransaction.getTransactionId()).toSerializable(),
239 private void commit(final ActorRef sender, Object serialized) {
240 Modification modification = MutableCompositeModification
241 .fromSerializable(serialized, schemaContext);
242 DOMStoreThreePhaseCommitCohort cohort =
243 modificationToCohort.remove(serialized);
244 if (cohort == null) {
246 "Could not find cohort for modification : {}. Writing modification using a new transaction",
248 DOMStoreReadWriteTransaction transaction =
249 store.newReadWriteTransaction();
250 modification.apply(transaction);
251 DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
252 ListenableFuture<Void> future =
253 commitCohort.preCommit();
256 future = commitCohort.commit();
258 } catch (InterruptedException | ExecutionException e) {
259 shardMBean.incrementFailedTransactionsCount();
260 LOG.error("Failed to commit", e);
263 //we want to just apply the recovery commit and return
264 shardMBean.incrementCommittedTransactionCount();
268 final ListenableFuture<Void> future = cohort.commit();
269 final ActorRef self = getSelf();
271 Futures.addCallback(future, new FutureCallback<Void>() {
272 public void onSuccess(Void v) {
273 sender.tell(new CommitTransactionReply().toSerializable(),self);
274 shardMBean.incrementCommittedTransactionCount();
275 shardMBean.setLastCommittedTransactionTime(new Date());
278 public void onFailure(Throwable t) {
279 LOG.error(t, "An exception happened during commit");
280 shardMBean.incrementFailedTransactionsCount();
281 sender.tell(new akka.actor.Status.Failure(t), self);
287 private void handleForwardedCommit(ForwardedCommitTransaction message) {
288 Object serializedModification =
289 message.getModification().toSerializable();
292 .put(serializedModification, message.getCohort());
295 this.persistData(getSender(), "identifier",
296 new CompositeModificationPayload(serializedModification));
298 this.commit(getSender(), serializedModification);
302 private void updateSchemaContext(UpdateSchemaContext message) {
303 this.schemaContext = message.getSchemaContext();
304 store.onGlobalContextUpdated(message.getSchemaContext());
307 private void registerChangeListener(
308 RegisterChangeListener registerChangeListener) {
310 LOG.debug("registerDataChangeListener for {}", registerChangeListener
314 ActorSelection dataChangeListenerPath = getContext()
315 .system().actorSelection(
316 registerChangeListener.getDataChangeListenerPath());
319 // Notify the listener if notifications should be enabled or not
320 // If this shard is the leader then it will enable notifications else
322 dataChangeListenerPath
323 .tell(new EnableNotification(isLeader()), getSelf());
325 // Now store a reference to the data change listener so it can be notified
326 // at a later point if notifications should be enabled or disabled
327 dataChangeListeners.add(dataChangeListenerPath);
329 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
331 new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
333 org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
335 store.registerChangeListener(registerChangeListener.getPath(),
336 listener, registerChangeListener.getScope());
337 ActorRef listenerRegistration =
338 getContext().actorOf(
339 DataChangeListenerRegistration.props(registration));
342 "registerDataChangeListener sending reply, listenerRegistrationPath = {} "
343 , listenerRegistration.path().toString());
346 .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
350 private void createTransactionChain() {
351 DOMStoreTransactionChain chain = store.createTransactionChain();
352 ActorRef transactionChain =
353 getContext().actorOf(
354 ShardTransactionChain.props(chain, schemaContext));
356 .tell(new CreateTransactionChainReply(transactionChain.path())
361 @Override protected void applyState(ActorRef clientActor, String identifier,
364 if (data instanceof CompositeModificationPayload) {
365 Object modification =
366 ((CompositeModificationPayload) data).getModification();
368 if (modification != null) {
369 commit(clientActor, modification);
372 "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
373 identifier, clientActor.path().toString());
378 LOG.error("Unknown state received {}", data);
382 ReplicatedLogEntry lastLogEntry = getLastLogEntry();
384 if (lastLogEntry != null) {
385 shardMBean.setLastLogIndex(lastLogEntry.getIndex());
386 shardMBean.setLastLogTerm(lastLogEntry.getTerm());
389 shardMBean.setCommitIndex(getCommitIndex());
390 shardMBean.setLastApplied(getLastApplied());
394 @Override protected Object createSnapshot() {
395 throw new UnsupportedOperationException("createSnapshot");
398 @Override protected void applySnapshot(Object snapshot) {
399 throw new UnsupportedOperationException("applySnapshot");
402 @Override protected void onStateChanged() {
403 for (ActorSelection dataChangeListener : dataChangeListeners) {
405 .tell(new EnableNotification(isLeader()), getSelf());
408 if (getLeaderId() != null) {
409 shardMBean.setLeader(getLeaderId());
412 shardMBean.setRaftState(getRaftState().name());
413 shardMBean.setCurrentTerm(getCurrentTerm());
416 @Override public String persistenceId() {
417 return this.name.toString();
421 private static class ShardConfigParams extends DefaultConfigParamsImpl {
422 public static final FiniteDuration HEART_BEAT_INTERVAL =
423 new FiniteDuration(500, TimeUnit.MILLISECONDS);
425 @Override public FiniteDuration getHeartBeatInterval() {
426 return HEART_BEAT_INTERVAL;