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;
19 import com.google.common.util.concurrent.ListenableFuture;
20 import com.google.common.util.concurrent.ListeningExecutorService;
21 import com.google.common.util.concurrent.MoreExecutors;
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.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.RaftActor;
38 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
39 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
40 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
41 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
42 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
43 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
44 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
45 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
47 import java.util.HashMap;
49 import java.util.concurrent.ExecutionException;
50 import java.util.concurrent.Executors;
53 * A Shard represents a portion of the logical data tree <br/>
55 * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
58 public class Shard extends RaftActor {
60 public static final String DEFAULT_NAME = "default";
62 private final ListeningExecutorService storeExecutor =
63 MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
65 private final InMemoryDOMDataStore store;
67 private final Map<Object, DOMStoreThreePhaseCommitCohort>
68 modificationToCohort = new HashMap<>();
70 private final LoggingAdapter LOG =
71 Logging.getLogger(getContext().system(), this);
73 // By default persistent will be true and can be turned off using the system
74 // property persistent
75 private final boolean persistent;
77 private final String name;
79 private volatile SchemaContext schemaContext;
81 private final ShardStats shardMBean;
83 private Shard(String name, Map<String, String> peerAddresses) {
84 super(name, peerAddresses);
88 String setting = System.getProperty("shard.persistent");
90 this.persistent = !"false".equals(setting);
92 LOG.info("Creating shard : {} persistent : {}", name, persistent);
94 store = new InMemoryDOMDataStore(name, storeExecutor);
96 shardMBean = ShardMBeanFactory.getShardStatsMBean(name);
100 public static Props props(final String name, final Map<String, String> peerAddresses) {
101 return Props.create(new Creator<Shard>() {
104 public Shard create() throws Exception {
105 return new Shard(name, peerAddresses);
112 @Override public void onReceiveCommand(Object message){
113 LOG.debug("Received message {} from {}", message.getClass().toString(), getSender());
115 if (message.getClass().equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
117 createTransactionChain();
118 } else if(getLeader() != null){
119 getLeader().forward(message, getContext());
121 } else if (message.getClass().equals(RegisterChangeListener.SERIALIZABLE_CLASS)) {
122 registerChangeListener(RegisterChangeListener.fromSerializable(getContext().system(), message));
123 } else if (message instanceof UpdateSchemaContext) {
124 updateSchemaContext((UpdateSchemaContext) message);
125 } else if (message instanceof ForwardedCommitTransaction) {
126 handleForwardedCommit((ForwardedCommitTransaction) message);
127 } else if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
129 createTransaction(CreateTransaction.fromSerializable(message));
130 } else if(getLeader() != null){
131 getLeader().forward(message, getContext());
133 } else if (message instanceof PeerAddressResolved){
134 PeerAddressResolved resolved = (PeerAddressResolved) message;
135 setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
137 super.onReceiveCommand(message);
141 private void createTransaction(CreateTransaction createTransaction) {
142 DOMStoreReadWriteTransaction transaction =
143 store.newReadWriteTransaction();
144 String transactionId = "shard-" + createTransaction.getTransactionId();
145 LOG.info("Creating transaction : {} " , transactionId);
146 ActorRef transactionActor = getContext().actorOf(
147 ShardTransaction.props(transaction, getSelf(), schemaContext), transactionId);
150 .tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), createTransaction.getTransactionId()).toSerializable(),
154 private void commit(final ActorRef sender, Object serialized) {
155 Modification modification = MutableCompositeModification.fromSerializable(serialized, schemaContext);
156 DOMStoreThreePhaseCommitCohort cohort =
157 modificationToCohort.remove(serialized);
158 if (cohort == null) {
160 "Could not find cohort for modification : " + modification);
161 LOG.info("Writing modification using a new transaction");
162 modification.apply(store.newReadWriteTransaction());
166 final ListenableFuture<Void> future = cohort.commit();
167 shardMBean.incrementCommittedTransactionCount();
168 final ActorRef self = getSelf();
169 future.addListener(new Runnable() {
177 .tell(new CommitTransactionReply().toSerializable(),
180 LOG.error("sender is null ???");
182 } catch (InterruptedException | ExecutionException e) {
183 // FIXME : Handle this properly
184 LOG.error(e, "An exception happened when committing");
187 }, getContext().dispatcher());
190 private void handleForwardedCommit(ForwardedCommitTransaction message) {
191 Object serializedModification = message.getModification().toSerializable();
194 .put(serializedModification , message.getCohort());
197 this.persistData(getSender(), "identifier", new CompositeModificationPayload(serializedModification));
199 this.commit(getSender(), serializedModification);
203 private void updateSchemaContext(UpdateSchemaContext message) {
204 this.schemaContext = message.getSchemaContext();
205 store.onGlobalContextUpdated(message.getSchemaContext());
208 private void registerChangeListener(
209 RegisterChangeListener registerChangeListener) {
211 LOG.debug("registerDataChangeListener for " + registerChangeListener.getPath());
214 ActorSelection dataChangeListenerPath = getContext()
215 .system().actorSelection(
216 registerChangeListener.getDataChangeListenerPath());
218 AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
219 listener = new DataChangeListenerProxy(schemaContext,dataChangeListenerPath);
221 org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
223 store.registerChangeListener(registerChangeListener.getPath(),
224 listener, registerChangeListener.getScope());
225 ActorRef listenerRegistration =
226 getContext().actorOf(
227 DataChangeListenerRegistration.props(registration));
229 LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = " + listenerRegistration.path().toString());
232 .tell(new RegisterChangeListenerReply(listenerRegistration.path()).toSerializable(),
236 private void createTransactionChain() {
237 DOMStoreTransactionChain chain = store.createTransactionChain();
238 ActorRef transactionChain =
239 getContext().actorOf(
240 ShardTransactionChain.props(chain, schemaContext));
242 .tell(new CreateTransactionChainReply(transactionChain.path())
247 @Override protected void applyState(ActorRef clientActor, String identifier,
250 if(data instanceof CompositeModificationPayload){
251 Object modification =
252 ((CompositeModificationPayload) data).getModification();
253 commit(clientActor, modification);
255 LOG.error("Unknown state received {}", data);
260 @Override protected Object createSnapshot() {
261 throw new UnsupportedOperationException("createSnapshot");
264 @Override protected void applySnapshot(Object snapshot) {
265 throw new UnsupportedOperationException("applySnapshot");
268 @Override public String persistenceId() {