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.ActorSelection;
13 import com.google.common.base.Optional;
14 import com.google.common.util.concurrent.ListenableFuture;
15 import com.google.common.util.concurrent.ListenableFutureTask;
16 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
17 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
18 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
19 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
20 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
21 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
22 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
23 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
24 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
25 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
26 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
27 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
30 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
31 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
32 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
34 import java.util.ArrayList;
35 import java.util.HashMap;
36 import java.util.List;
38 import java.util.concurrent.Callable;
39 import java.util.concurrent.ExecutorService;
40 import java.util.concurrent.atomic.AtomicLong;
43 * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
45 * Creating a transaction on the consumer side will create one instance of a transaction proxy. If during
46 * the transaction reads and writes are done on data that belongs to different shards then a separate transaction will
47 * be created on each of those shards by the TransactionProxy
50 * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
51 * shards will be executed.
54 public class TransactionProxy implements DOMStoreReadWriteTransaction {
55 public enum TransactionType {
61 private static final AtomicLong counter = new AtomicLong();
63 private final TransactionType transactionType;
64 private final ActorContext actorContext;
65 private final Map<String, TransactionContext> remoteTransactionPaths = new HashMap<>();
66 private final String identifier;
67 private final ExecutorService executor;
68 private final SchemaContext schemaContext;
70 public TransactionProxy(
71 ActorContext actorContext,
72 TransactionType transactionType,
73 ExecutorService executor,
74 SchemaContext schemaContext
77 this.identifier = actorContext.getCurrentMemberName() + "-txn-" + counter.getAndIncrement();
78 this.transactionType = transactionType;
79 this.actorContext = actorContext;
80 this.executor = executor;
81 this.schemaContext = schemaContext;
87 public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
89 createTransactionIfMissing(actorContext, path);
91 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
93 Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
95 @Override public Optional<NormalizedNode<?,?>> call() throws Exception {
96 Object response = actorContext
97 .executeRemoteOperation(remoteTransaction, new ReadData(path).toSerializable(),
98 ActorContext.ASK_DURATION);
99 if(response.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)){
100 ReadDataReply reply = ReadDataReply.fromSerializable(schemaContext,path, response);
101 if(reply.getNormalizedNode() == null){
102 return Optional.absent();
104 //FIXME : A cast should not be required here ???
105 return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
108 return Optional.absent();
112 ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
113 future = ListenableFutureTask.create(call);
115 executor.submit(future);
121 public void write(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
123 createTransactionIfMissing(actorContext, path);
125 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
126 remoteTransaction.tell(new WriteData(path, data, schemaContext).toSerializable(), null);
130 public void merge(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
132 createTransactionIfMissing(actorContext, path);
134 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
135 remoteTransaction.tell(new MergeData(path, data, schemaContext).toSerializable(), null);
139 public void delete(YangInstanceIdentifier path) {
141 createTransactionIfMissing(actorContext, path);
143 final ActorSelection remoteTransaction = remoteTransactionFromIdentifier(path);
144 remoteTransaction.tell(new DeleteData(path).toSerializable(), null);
148 public DOMStoreThreePhaseCommitCohort ready() {
149 List<ActorPath> cohortPaths = new ArrayList<>();
151 for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
152 Object result = actorContext.executeRemoteOperation(transactionContext.getActor(),
153 new ReadyTransaction().toSerializable(),
154 ActorContext.ASK_DURATION
157 if(result.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)){
158 ReadyTransactionReply reply = ReadyTransactionReply.fromSerializable(actorContext.getActorSystem(),result);
159 String resolvedCohortPath = transactionContext
160 .getResolvedCohortPath(reply.getCohortPath().toString());
161 cohortPaths.add(actorContext.actorFor(resolvedCohortPath));
165 return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths, identifier, executor);
169 public Object getIdentifier() {
170 return this.identifier;
174 public void close() {
175 for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
176 transactionContext.getActor().tell(
177 new CloseTransaction().toSerializable(), null);
181 private ActorSelection remoteTransactionFromIdentifier(YangInstanceIdentifier path){
182 String shardName = shardNameFromIdentifier(path);
183 return remoteTransactionPaths.get(shardName).getActor();
186 private String shardNameFromIdentifier(YangInstanceIdentifier path){
187 return ShardStrategyFactory.getStrategy(path).findShard(path);
190 private void createTransactionIfMissing(ActorContext actorContext, YangInstanceIdentifier path) {
191 String shardName = ShardStrategyFactory.getStrategy(path).findShard(path);
193 TransactionContext transactionContext =
194 remoteTransactionPaths.get(shardName);
196 if(transactionContext != null){
197 // A transaction already exists with that shard
201 Object response = actorContext.executeShardOperation(shardName, new CreateTransaction(identifier).toSerializable(), ActorContext.ASK_DURATION);
202 if(response.getClass().equals(CreateTransactionReply.SERIALIZABLE_CLASS)){
203 CreateTransactionReply reply = CreateTransactionReply.fromSerializable(response);
204 String transactionPath = actorContext.getRemoteActorPath(shardName, reply.getTransactionPath());
206 ActorSelection transactionActor = actorContext.actorSelection(transactionPath);
207 transactionContext = new TransactionContext(shardName, transactionPath, transactionActor);
209 remoteTransactionPaths.put(shardName, transactionContext);
214 private class TransactionContext {
215 private final String shardName;
216 private final String actorPath;
217 private final ActorSelection actor;
220 private TransactionContext(String shardName, String actorPath,
221 ActorSelection actor) {
222 this.shardName = shardName;
223 this.actorPath = actorPath;
228 public String getShardName() {
232 public String getActorPath() {
236 public ActorSelection getActor() {
240 public String getResolvedCohortPath(String cohortPath){
241 return actorContext.resolvePath(actorPath, cohortPath);