opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.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.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;
  39
  40 import java.util.ArrayList;
  41 import java.util.HashMap;
  42 import java.util.List;
  43 import java.util.Map;
  44 import java.util.concurrent.Callable;
  45 import java.util.concurrent.ExecutorService;
  46 import java.util.concurrent.atomic.AtomicLong;
  47
  48 /**
  49  * TransactionProxy acts as a proxy for one or more transactions that were created on a remote shard
  50  * <p>
  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
  54  *</p>
  55  * <p>
  56  * The TransactionProxy does not make any guarantees about atomicity or order in which the transactions on the various
  57  * shards will be executed.
  58  * </p>
  59  */
  60 public class TransactionProxy implements DOMStoreReadWriteTransaction {
  61     public enum TransactionType {
  62         READ_ONLY,
  63         WRITE_ONLY,
  64         READ_WRITE
  65     }
  66
  67     private static final AtomicLong counter = new AtomicLong();
  68
  69     private static final Logger
  70         LOG = LoggerFactory.getLogger(TransactionProxy.class);
  71
  72
  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;
  79
  80     public TransactionProxy(
  81         ActorContext actorContext,
  82         TransactionType transactionType,
  83         ExecutorService executor,
  84         SchemaContext schemaContext
  85     ) {
  86
  87         this.identifier = actorContext.getCurrentMemberName() + "-txn-" + counter.getAndIncrement();
  88         this.transactionType = transactionType;
  89         this.actorContext = actorContext;
  90         this.executor = executor;
  91         this.schemaContext = schemaContext;
  92
  93
  94     }
  95
  96     @Override
  97     public ListenableFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
  98
  99         createTransactionIfMissing(actorContext, path);
 100
 101         return transactionContext(path).readData(path);
 102     }
 103
 104     @Override
 105     public void write(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
 106
 107         createTransactionIfMissing(actorContext, path);
 108
 109         transactionContext(path).writeData(path, data);
 110     }
 111
 112     @Override
 113     public void merge(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
 114
 115         createTransactionIfMissing(actorContext, path);
 116
 117         transactionContext(path).mergeData(path, data);
 118     }
 119
 120     @Override
 121     public void delete(YangInstanceIdentifier path) {
 122
 123         createTransactionIfMissing(actorContext, path);
 124
 125         transactionContext(path).deleteData(path);
 126     }
 127
 128     @Override
 129     public DOMStoreThreePhaseCommitCohort ready() {
 130         List<ActorPath> cohortPaths = new ArrayList<>();
 131
 132         for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
 133             Object result = transactionContext.readyTransaction();
 134
 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));
 140             }
 141         }
 142
 143         return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths, identifier, executor);
 144     }
 145
 146     @Override
 147     public Object getIdentifier() {
 148         return this.identifier;
 149     }
 150
 151     @Override
 152     public void close() {
 153         for(TransactionContext transactionContext : remoteTransactionPaths.values()) {
 154             transactionContext.closeTransaction();
 155         }
 156     }
 157
 158     private TransactionContext transactionContext(YangInstanceIdentifier path){
 159         String shardName = shardNameFromIdentifier(path);
 160         return remoteTransactionPaths.get(shardName);
 161     }
 162
 163     private String shardNameFromIdentifier(YangInstanceIdentifier path){
 164         return ShardStrategyFactory.getStrategy(path).findShard(path);
 165     }
 166
 167     private void createTransactionIfMissing(ActorContext actorContext, YangInstanceIdentifier path) {
 168         String shardName = ShardStrategyFactory.getStrategy(path).findShard(path);
 169
 170         TransactionContext transactionContext =
 171             remoteTransactionPaths.get(shardName);
 172
 173         if(transactionContext != null){
 174             // A transaction already exists with that shard
 175             return;
 176         }
 177
 178         try {
 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);
 186
 187                 String transactionPath = reply.getTransactionPath();
 188
 189                 LOG.info("Received transaction path = {}"  , transactionPath );
 190
 191                 ActorSelection transactionActor =
 192                     actorContext.actorSelection(transactionPath);
 193                 transactionContext =
 194                     new TransactionContextImpl(shardName, transactionPath,
 195                         transactionActor);
 196
 197                 remoteTransactionPaths.put(shardName, transactionContext);
 198             }
 199         } catch(TimeoutException e){
 200             remoteTransactionPaths.put(shardName, new NoOpTransactionContext(shardName));
 201         }
 202     }
 203
 204     private interface TransactionContext {
 205         String getShardName();
 206
 207         String getResolvedCohortPath(String cohortPath);
 208
 209         public void closeTransaction();
 210
 211         public Object readyTransaction();
 212
 213         void deleteData(YangInstanceIdentifier path);
 214
 215         void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 216
 217         ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(final YangInstanceIdentifier path);
 218
 219         void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 220     }
 221
 222
 223     private class TransactionContextImpl implements TransactionContext{
 224         private final String shardName;
 225         private final String actorPath;
 226         private final ActorSelection  actor;
 227
 228
 229         private TransactionContextImpl(String shardName, String actorPath,
 230             ActorSelection actor) {
 231             this.shardName = shardName;
 232             this.actorPath = actorPath;
 233             this.actor = actor;
 234         }
 235
 236         @Override public String getShardName() {
 237             return shardName;
 238         }
 239
 240         private ActorSelection getActor() {
 241             return actor;
 242         }
 243
 244         @Override public String getResolvedCohortPath(String cohortPath){
 245             return actorContext.resolvePath(actorPath, cohortPath);
 246         }
 247
 248         @Override public void closeTransaction() {
 249             getActor().tell(
 250                 new CloseTransaction().toSerializable(), null);
 251         }
 252
 253         @Override public Object readyTransaction() {
 254             return actorContext.executeRemoteOperation(getActor(),
 255                 new ReadyTransaction().toSerializable(),
 256                 ActorContext.ASK_DURATION
 257             );
 258
 259         }
 260
 261         @Override public void deleteData(YangInstanceIdentifier path) {
 262             getActor().tell(new DeleteData(path).toSerializable(), null);
 263         }
 264
 265         @Override public void mergeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data){
 266             getActor().tell(new MergeData(path, data, schemaContext).toSerializable(), null);
 267         }
 268
 269         @Override public ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(final YangInstanceIdentifier path) {
 270
 271             Callable<Optional<NormalizedNode<?,?>>> call = new Callable() {
 272
 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();
 281                         }
 282                         //FIXME : A cast should not be required here ???
 283                         return (Optional<NormalizedNode<?, ?>>) Optional.of(reply.getNormalizedNode());
 284                     }
 285
 286                     return Optional.absent();
 287                 }
 288             };
 289
 290             ListenableFutureTask<Optional<NormalizedNode<?, ?>>>
 291                 future = ListenableFutureTask.create(call);
 292
 293             executor.submit(future);
 294
 295             return future;
 296         }
 297
 298         @Override public void writeData(YangInstanceIdentifier path, NormalizedNode<?, ?> data) {
 299             getActor().tell(new WriteData(path, data, schemaContext).toSerializable(), null);
 300         }
 301
 302     }
 303
 304     private class NoOpTransactionContext implements TransactionContext {
 305
 306         private final Logger
 307             LOG = LoggerFactory.getLogger(NoOpTransactionContext.class);
 308
 309         private final String shardName;
 310
 311         private ActorRef cohort;
 312
 313         public NoOpTransactionContext(String shardName){
 314             this.shardName = shardName;
 315         }
 316         @Override public String getShardName() {
 317             return  shardName;
 318
 319         }
 320
 321         @Override public String getResolvedCohortPath(String cohortPath) {
 322             return cohort.path().toString();
 323         }
 324
 325         @Override public void closeTransaction() {
 326             LOG.error("closeTransaction called");
 327         }
 328
 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();
 333         }
 334
 335         @Override public void deleteData(YangInstanceIdentifier path) {
 336             LOG.error("deleteData called path = {}", path);
 337         }
 338
 339         @Override public void mergeData(YangInstanceIdentifier path,
 340             NormalizedNode<?, ?> data) {
 341             LOG.error("mergeData called path = {}", path);
 342         }
 343
 344         @Override
 345         public ListenableFuture<Optional<NormalizedNode<?, ?>>> readData(
 346             YangInstanceIdentifier path) {
 347             LOG.error("readData called path = {}", path);
 348             return Futures.immediateFuture(
 349                 Optional.<NormalizedNode<?, ?>>absent());
 350         }
 351
 352         @Override public void writeData(YangInstanceIdentifier path,
 353             NormalizedNode<?, ?> data) {
 354             LOG.error("writeData called path = {}", path);
 355         }
 356     }
 357
 358
 359
 360 }