09ad00598fedb0730cb520740d4f4588a058446e
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / Shard.java
1 /*
2  * Copyright (c) 2014 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
9 package org.opendaylight.controller.cluster.datastore;
10
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.persistence.Persistent;
18 import akka.persistence.UntypedProcessor;
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.messages.CommitTransactionReply;
23 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
24 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
25 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
26 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
27 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
28 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
29 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
30 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
31 import org.opendaylight.controller.cluster.datastore.modification.Modification;
32 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
33 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
35 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
36 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
37 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
38 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
39
40 import java.util.HashMap;
41 import java.util.Map;
42 import java.util.concurrent.ExecutionException;
43 import java.util.concurrent.Executors;
44
45 /**
46  * A Shard represents a portion of the logical data tree <br/>
47  * <p>
48  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
49  * </p>
50  */
51 public class Shard extends UntypedProcessor {
52
53     public static final String DEFAULT_NAME = "default";
54
55     private final ListeningExecutorService storeExecutor =
56         MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
57
58     private final InMemoryDOMDataStore store;
59
60     private final Map<Modification, DOMStoreThreePhaseCommitCohort>
61         modificationToCohort = new HashMap<>();
62
63     private final LoggingAdapter log =
64         Logging.getLogger(getContext().system(), this);
65
66     private Shard(String name) {
67         log.info("Creating shard : {}", name );
68
69         store = new InMemoryDOMDataStore(name, storeExecutor);
70     }
71
72     public static Props props(final String name) {
73         return Props.create(new Creator<Shard>() {
74
75             @Override
76             public Shard create() throws Exception {
77                 return new Shard(name);
78             }
79
80         });
81     }
82
83     @Override
84     public void onReceive(Object message) throws Exception {
85         log.debug("Received message {}", message);
86
87         if (message instanceof CreateTransactionChain) {
88             createTransactionChain();
89         } else if (message instanceof RegisterChangeListener) {
90             registerChangeListener((RegisterChangeListener) message);
91         } else if (message instanceof UpdateSchemaContext) {
92             updateSchemaContext((UpdateSchemaContext) message);
93         } else if (message instanceof ForwardedCommitTransaction) {
94             handleForwardedCommit((ForwardedCommitTransaction) message);
95         } else if (message instanceof Persistent) {
96             commit((Persistent) message);
97         } else if (message instanceof CreateTransaction) {
98             createTransaction();
99         }
100     }
101
102     private void createTransaction() {
103         DOMStoreReadWriteTransaction transaction =
104             store.newReadWriteTransaction();
105         ActorRef transactionActor = getContext().actorOf(
106             ShardTransaction.props(transaction, getSelf()));
107         getSender()
108             .tell(new CreateTransactionReply(transactionActor.path()),
109                 getSelf());
110     }
111
112     private void commit(Persistent message) {
113         Modification modification = (Modification) message.payload();
114         DOMStoreThreePhaseCommitCohort cohort =
115             modificationToCohort.remove(modification);
116         if (cohort == null) {
117             log.error(
118                 "Could not find cohort for modification : " + modification);
119             return;
120         }
121         final ListenableFuture<Void> future = cohort.commit();
122         final ActorRef sender = getSender();
123         final ActorRef self = getSelf();
124         future.addListener(new Runnable() {
125             @Override
126             public void run() {
127                 try {
128                     future.get();
129                     sender.tell(new CommitTransactionReply(), self);
130                 } catch (InterruptedException | ExecutionException e) {
131                     log.error(e, "An exception happened when committing");
132                 }
133             }
134         }, getContext().dispatcher());
135     }
136
137     private void handleForwardedCommit(ForwardedCommitTransaction message) {
138         log.info("received forwarded transaction");
139         modificationToCohort
140             .put(message.getModification(), message.getCohort());
141         getSelf().forward(Persistent.create(message.getModification()),
142             getContext());
143     }
144
145     private void updateSchemaContext(UpdateSchemaContext message) {
146         store.onGlobalContextUpdated(message.getSchemaContext());
147     }
148
149     private void registerChangeListener(
150         RegisterChangeListener registerChangeListener) {
151
152         ActorSelection dataChangeListenerPath = getContext()
153             .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
154
155         AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
156             listener = new DataChangeListenerProxy(dataChangeListenerPath);
157
158         org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
159             registration =
160             store.registerChangeListener(registerChangeListener.getPath(),
161                 listener, registerChangeListener.getScope());
162         ActorRef listenerRegistration =
163             getContext().actorOf(
164                 DataChangeListenerRegistration.props(registration));
165         getSender()
166             .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
167                 getSelf());
168     }
169
170     private void createTransactionChain() {
171         DOMStoreTransactionChain chain = store.createTransactionChain();
172         ActorRef transactionChain =
173             getContext().actorOf(ShardTransactionChain.props(chain));
174         getSender()
175             .tell(new CreateTransactionChainReply(transactionChain.path()),
176                 getSelf());
177     }
178 }

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