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.util.concurrent.ListenableFuture;
20 import com.google.common.util.concurrent.ListeningExecutorService;
21 import com.google.common.util.concurrent.MoreExecutors;
22 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
23 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
24 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
25 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
26 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
27 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
28 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
30 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
31 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
32 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
33 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
34 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
35 import org.opendaylight.controller.cluster.datastore.modification.Modification;
36 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
37 import org.opendaylight.controller.cluster.raft.ConfigParams;
38 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
39 import org.opendaylight.controller.cluster.raft.RaftActor;
40 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
41 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
42 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
43 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
44 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
45 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
46 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
47 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
48 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
49 import scala.concurrent.duration.FiniteDuration;
51 import java.util.ArrayList;
52 import java.util.HashMap;
53 import java.util.List;
55 import java.util.concurrent.ExecutionException;
56 import java.util.concurrent.Executors;
57 import java.util.concurrent.TimeUnit;
60 * A Shard represents a portion of the logical data tree <br/>
62 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
65 public class Shard extends RaftActor {
67 private static final ConfigParams configParams = new ShardConfigParams();
69 public static final String DEFAULT_NAME = "default";
71 private final ListeningExecutorService storeExecutor =
72 MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
74 private final InMemoryDOMDataStore store;
76 private final Map<Object, DOMStoreThreePhaseCommitCohort>
77 modificationToCohort = new HashMap<>();
79 private final LoggingAdapter LOG =
80 Logging.getLogger(getContext().system(), this);
82 // By default persistent will be true and can be turned off using the system
83 // property persistent
84 private final boolean persistent;
86 private final String name;
88 private volatile SchemaContext schemaContext;
90 private final ShardStats shardMBean;
92 private final List<ActorSelection> dataChangeListeners = new ArrayList<>();
94 private Shard(String name, Map<String, String> peerAddresses) {
95 super(name, peerAddresses, Optional.of(configParams));
99 String setting = System.getProperty("shard.persistent");
101 this.persistent = !"false".equals(setting);
103 LOG.info("Creating shard : {} persistent : {}", name, persistent);
105 store = new InMemoryDOMDataStore(name, storeExecutor);
107 shardMBean = ShardMBeanFactory.getShardStatsMBean(name);
111 public static Props props(final String name,
112 final Map<String, String> peerAddresses) {
113 return Props.create(new Creator<Shard>() {
116 public Shard create() throws Exception {
117 return new Shard(name, peerAddresses);
124 @Override public void onReceiveCommand(Object message) {
125 LOG.debug("Received message {} from {}", message.getClass().toString(),
128 if (message.getClass()
129 .equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
131 createTransactionChain();
132 } else if (getLeader() != null) {
133 getLeader().forward(message, getContext());
135 } else if (message instanceof RegisterChangeListener) {
136 registerChangeListener((RegisterChangeListener) message);
137 } else if (message instanceof UpdateSchemaContext) {
138 updateSchemaContext((UpdateSchemaContext) message);
139 } else if (message instanceof ForwardedCommitTransaction) {
140 handleForwardedCommit((ForwardedCommitTransaction) message);
141 } else if (message.getClass()
142 .equals(CreateTransaction.SERIALIZABLE_CLASS)) {
144 createTransaction(CreateTransaction.fromSerializable(message));
145 } else if (getLeader() != null) {
146 getLeader().forward(message, getContext());
148 } else if (message instanceof PeerAddressResolved) {
149 PeerAddressResolved resolved = (PeerAddressResolved) message;
150 setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
152 super.onReceiveCommand(message);
156 private ActorRef createTypedTransactionActor(
157 CreateTransaction createTransaction, String transactionId) {
158 if (createTransaction.getTransactionType()
159 == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
160 shardMBean.incrementReadOnlyTransactionCount();
161 return getContext().actorOf(
163 .props(store.newReadOnlyTransaction(), getSelf(),
164 schemaContext), transactionId);
166 } else if (createTransaction.getTransactionType()
167 == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
168 shardMBean.incrementReadWriteTransactionCount();
169 return getContext().actorOf(
171 .props(store.newReadWriteTransaction(), getSelf(),
172 schemaContext), transactionId);
175 } else if (createTransaction.getTransactionType()
176 == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
177 shardMBean.incrementWriteOnlyTransactionCount();
178 return getContext().actorOf(
180 .props(store.newWriteOnlyTransaction(), getSelf(),
181 schemaContext), transactionId);
183 throw new IllegalArgumentException(
184 "CreateTransaction message has unidentified transaction type="
185 + createTransaction.getTransactionType());
189 private void createTransaction(CreateTransaction createTransaction) {
191 String transactionId = "shard-" + createTransaction.getTransactionId();
192 LOG.info("Creating transaction : {} ", transactionId);
193 ActorRef transactionActor =
194 createTypedTransactionActor(createTransaction, transactionId);
197 .tell(new CreateTransactionReply(
198 Serialization.serializedActorPath(transactionActor),
199 createTransaction.getTransactionId()).toSerializable(),
204 private void commit(final ActorRef sender, Object serialized) {
205 Modification modification =
206 MutableCompositeModification.fromSerializable(
207 serialized, schemaContext);
208 DOMStoreThreePhaseCommitCohort cohort =
209 modificationToCohort.remove(serialized);
210 if (cohort == null) {
212 "Could not find cohort for modification : {}", modification);
213 LOG.info("Writing modification using a new transaction");
214 DOMStoreReadWriteTransaction transaction =
215 store.newReadWriteTransaction();
216 modification.apply(transaction);
217 DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
218 ListenableFuture<Void> future =
219 commitCohort.preCommit();
222 future = commitCohort.commit();
224 } catch (InterruptedException e) {
225 LOG.error("Failed to commit", e);
226 } catch (ExecutionException e) {
227 LOG.error("Failed to commit", e);
231 final ListenableFuture<Void> future = cohort.commit();
232 shardMBean.incrementCommittedTransactionCount();
233 final ActorRef self = getSelf();
234 future.addListener(new Runnable() {
240 if (sender != null) {
242 .tell(new CommitTransactionReply().toSerializable(),
245 LOG.error("sender is null ???");
247 } catch (InterruptedException | ExecutionException e) {
248 // FIXME : Handle this properly
249 LOG.error(e, "An exception happened when committing");
252 }, getContext().dispatcher());
255 private void handleForwardedCommit(ForwardedCommitTransaction message) {
256 Object serializedModification =
257 message.getModification().toSerializable();
260 .put(serializedModification, message.getCohort());
263 this.persistData(getSender(), "identifier",
264 new CompositeModificationPayload(serializedModification));
266 this.commit(getSender(), serializedModification);
270 private void updateSchemaContext(UpdateSchemaContext message) {
271 this.schemaContext = message.getSchemaContext();
272 store.onGlobalContextUpdated(message.getSchemaContext());
275 private void registerChangeListener(
276 RegisterChangeListener registerChangeListener) {
278 LOG.debug("registerDataChangeListener for " + registerChangeListener
282 ActorSelection dataChangeListenerPath = getContext()
283 .system().actorSelection(
284 registerChangeListener.getDataChangeListenerPath());
287 // Notify the listener if notifications should be enabled or not
288 // If this shard is the leader then it will enable notifications else
290 dataChangeListenerPath
291 .tell(new EnableNotification(isLeader()), getSelf());
293 // Now store a reference to the data change listener so it can be notified
294 // at a later point if notifications should be enabled or disabled
295 dataChangeListeners.add(dataChangeListenerPath);
297 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
299 new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
301 org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
303 store.registerChangeListener(registerChangeListener.getPath(),
304 listener, registerChangeListener.getScope());
305 ActorRef listenerRegistration =
306 getContext().actorOf(
307 DataChangeListenerRegistration.props(registration));
310 "registerDataChangeListener sending reply, listenerRegistrationPath = "
311 + listenerRegistration.path().toString());
314 .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
318 private void createTransactionChain() {
319 DOMStoreTransactionChain chain = store.createTransactionChain();
320 ActorRef transactionChain =
321 getContext().actorOf(
322 ShardTransactionChain.props(chain, schemaContext));
324 .tell(new CreateTransactionChainReply(transactionChain.path())
330 @Override protected void applyState(ActorRef clientActor, String identifier,
333 if (data instanceof CompositeModificationPayload) {
334 Object modification =
335 ((CompositeModificationPayload) data).getModification();
337 if (modification != null) {
338 commit(clientActor, modification);
340 LOG.error("modification is null - this is very unexpected");
345 LOG.error("Unknown state received {}", data);
348 ReplicatedLogEntry lastLogEntry = getLastLogEntry();
350 if(lastLogEntry != null){
351 shardMBean.setLastLogIndex(lastLogEntry.getIndex());
352 shardMBean.setLastLogTerm(lastLogEntry.getTerm());
355 shardMBean.setCommitIndex(getCommitIndex());
356 shardMBean.setLastApplied(getLastApplied());
360 @Override protected Object createSnapshot() {
361 throw new UnsupportedOperationException("createSnapshot");
364 @Override protected void applySnapshot(Object snapshot) {
365 throw new UnsupportedOperationException("applySnapshot");
368 @Override protected void onStateChanged() {
369 for (ActorSelection dataChangeListener : dataChangeListeners) {
371 .tell(new EnableNotification(isLeader()), getSelf());
374 if (getLeaderId() != null) {
375 shardMBean.setLeader(getLeaderId());
378 shardMBean.setRaftState(getRaftState().name());
379 shardMBean.setCurrentTerm(getCurrentTerm());
382 @Override public String persistenceId() {
387 private static class ShardConfigParams extends DefaultConfigParamsImpl {
388 public static final FiniteDuration HEART_BEAT_INTERVAL =
389 new FiniteDuration(500, TimeUnit.MILLISECONDS);
391 @Override public FiniteDuration getHeartBeatInterval() {
392 return HEART_BEAT_INTERVAL;