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.ActorPath;
12 import akka.actor.ActorRef;
13 import akka.actor.ActorSelection;
14 import akka.actor.Props;
15 import com.google.common.base.Optional;
16 import com.google.common.util.concurrent.Futures;
17 import com.google.common.util.concurrent.ListenableFuture;
18 import com.google.common.util.concurrent.ListenableFutureTask;
19 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
20 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
21 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
22 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
23 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
24 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
25 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
26 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
27 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
28 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
29 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
30 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
31 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
32 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
33 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
36 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
37 import org.slf4j.Logger;
38 import org.slf4j.LoggerFactory;
40 import java.util.ArrayList;
41 import java.util.HashMap;
42 import java.util.List;
44 import java.util.concurrent.Callable;
45 import java.util.concurrent.ExecutorService;
46 import java.util.concurrent.atomic.AtomicLong;
49 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
51 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
52 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
53 * be created on each of those shards by the TransactionProxy
56 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
57 * shards will be executed.
60 public class TransactionProxy implements DOMStoreReadWriteTransaction {
61 public enum TransactionType {
67 private static final AtomicLong counter = new AtomicLong();
69 private static final Logger
70 LOG = LoggerFactory.getLogger(TransactionProxy.class);
73 private final TransactionType transactionType;
74 private final ActorContext actorContext;
75 private final Map<String, TransactionContext> remoteTransactionPaths = new HashMap<>();
76 private final String identifier;
77 private final ExecutorService executor;
78 private final SchemaContext schemaContext;
80 public TransactionProxy(
81 ActorContext actorContext,
82 TransactionType transactionType,
83 ExecutorService executor,
84 SchemaContext schemaContext
87 this.identifier = actorContext.getCurrentMemberName() + "-txn-" + counter.getAndIncrement();
88 this.transactionType = transactionType;
89 this.actorContext = actorContext;
90 this.executor = executor;
91 this.schemaContext = schemaContext;
97 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
99 createTransactionIfMissing(actorContext, path);
101 return transactionContext(path).readData(path);
105 public void write(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
107 createTransactionIfMissing(actorContext, path);
109 transactionContext(path).writeData(path, data);
113 public void merge(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
115 createTransactionIfMissing(actorContext, path);
117 transactionContext(path).mergeData(path, data);
121 public void delete(YangInstanceIdentifier path) {
123 createTransactionIfMissing(actorContext, path);
125 transactionContext(path).deleteData(path);
129 public DOMStoreThreePhaseCommitCohort ready() {
130 List<ActorPath> cohortPaths = new ArrayList<>();
132 for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
133 Object result = transactionContext.readyTransaction();
135 if(result.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)){
136 ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(actorContext.getActorSystem(),result);
137 String resolvedCohortPath = transactionContext
138 .getResolvedCohortPath(reply.getCohortPath().toString());
139 cohortPaths.add(actorContext.actorFor(resolvedCohortPath));
143 return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths, identifier, executor);
147 public Object getIdentifier() {
148 return this.identifier;
152 public void close() {
153 for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
154 transactionContext.closeTransaction();
158 private TransactionContext transactionContext(YangInstanceIdentifier path){
159 String shardName = shardNameFromIdentifier(path);
160 return remoteTransactionPaths.get(shardName);
163 private String shardNameFromIdentifier(YangInstanceIdentifier path){
164 return ShardStrategyFactory.getStrategy(path).findShard(path);
167 private void createTransactionIfMissing(ActorContext actorContext, YangInstanceIdentifier path) {
168 String shardName = ShardStrategyFactory.getStrategy(path).findShard(path);
170 TransactionContext transactionContext =
171 remoteTransactionPaths.get(shardName);
173 if(transactionContext != null){
174 // A transaction already exists with that shard
179 Object response = actorContext.executeShardOperation(shardName,
180 new CreateTransaction(identifier).toSerializable(),
181 ActorContext.ASK_DURATION);
182 if (response.getClass()
183 .equals(CreateTransactionReply.SERIALIZABLE_CLASS)) {
184 CreateTransactionReply reply =
185 CreateTransactionReply.fromSerializable(response);
187 String transactionPath = reply.getTransactionPath();
189 LOG.info("Received transaction path = {}" , transactionPath );
191 ActorSelection transactionActor =
192 actorContext.actorSelection(transactionPath);
194 new TransactionContextImpl(shardName, transactionPath,
197 remoteTransactionPaths.put(shardName, transactionContext);
199 } catch(TimeoutException e){
200 remoteTransactionPaths.put(shardName, new NoOpTransactionContext(shardName));
204 private interface TransactionContext {
205 String getShardName();
207 String getResolvedCohortPath(String cohortPath);
209 public void closeTransaction();
211 public Object readyTransaction();
213 void deleteData(YangInstanceIdentifier path);
215 void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
217 ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(final YangInstanceIdentifier path);
219 void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
223 private class TransactionContextImpl implements TransactionContext{
224 private final String shardName;
225 private final String actorPath;
226 private final ActorSelection actor;
229 private TransactionContextImpl(String shardName, String actorPath,
230 ActorSelection actor) {
231 this.shardName = shardName;
232 this.actorPath = actorPath;
236 @Override public String getShardName() {
240 private ActorSelection getActor() {
244 @Override public String getResolvedCohortPath(String cohortPath){
245 return actorContext.resolvePath(actorPath, cohortPath);
248 @Override public void closeTransaction() {
250 new CloseTransaction().toSerializable(), null);
253 @Override public Object readyTransaction() {
254 return actorContext.executeRemoteOperation(getActor(),
255 new ReadyTransaction().toSerializable(),
256 ActorContext.ASK_DURATION
261 @Override public void deleteData(YangInstanceIdentifier path) {
262 getActor().tell(new DeleteData(path).toSerializable(), null);
265 @Override public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data){
266 getActor().tell(new MergeData(path, data, schemaContext).toSerializable(), null);
269 @Override public ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(final YangInstanceIdentifier path) {
271 Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
273 @Override public Optional<NormalizedNode<?,?>> call() throws Exception {
274 Object response = actorContext
275 .executeRemoteOperation(getActor(), new ReadData(path).toSerializable(),
276 ActorContext.ASK_DURATION);
277 if(response.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)){
278 ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext,path, response);
279 if(reply.getNormalizedNode() == null){
280 return Optional.absent();
282 //FIXME : A cast should not be required here ???
283 return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
286 return Optional.absent();
290 ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
291 future = ListenableFutureTask.create(call);
293 executor.submit(future);
298 @Override public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
299 getActor().tell(new WriteData(path, data, schemaContext).toSerializable(), null);
304 private class NoOpTransactionContext implements TransactionContext {
307 LOG = LoggerFactory.getLogger(NoOpTransactionContext.class);
309 private final String shardName;
311 private ActorRef cohort;
313 public NoOpTransactionContext(String shardName){
314 this.shardName = shardName;
316 @Override public String getShardName() {
321 @Override public String getResolvedCohortPath(String cohortPath) {
322 return cohort.path().toString();
325 @Override public void closeTransaction() {
326 LOG.error("closeTransaction called");
329 @Override public Object readyTransaction() {
330 LOG.error("readyTransaction called");
331 cohort = actorContext.getActorSystem().actorOf(Props.create(NoOpCohort.class));
332 return new ReadyTransactionReply(cohort.path()).toSerializable();
335 @Override public void deleteData(YangInstanceIdentifier path) {
336 LOG.error("deleteData called path = {}", path);
339 @Override public void mergeData(YangInstanceIdentifier path,
340 NormalizedNode<?, ?> data) {
341 LOG.error("mergeData called path = {}", path);
345 public ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(
346 YangInstanceIdentifier path) {
347 LOG.error("readData called path = {}", path);
348 return Futures.immediateFuture(
349 Optional.<NormalizedNode<?, ?>>absent());
352 @Override public void writeData(YangInstanceIdentifier path,
353 NormalizedNode<?, ?> data) {
354 LOG.error("writeData called path = {}", path);