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.ForwardedCommitTransaction;
  27 import org.opendaylight.controller.cluster.datastore.messages.NonPersistent;
  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 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
  40
  41 import java.util.HashMap;
  42 import java.util.Map;
  43 import java.util.concurrent.ExecutionException;
  44 import java.util.concurrent.Executors;
  45
  46 /**
  47  * A Shard represents a portion of the logical data tree <br/>
  48  * <p>
  49  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
  50  * </p>
  51  */
  52 public class Shard extends UntypedProcessor {
  53
  54     public static final String DEFAULT_NAME = "default";
  55
  56     private final ListeningExecutorService storeExecutor =
  57         MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
  58
  59     private final InMemoryDOMDataStore store;
  60
  61     private final Map<Modification, DOMStoreThreePhaseCommitCohort>
  62         modificationToCohort = new HashMap<>();
  63
  64     private final LoggingAdapter log =
  65         Logging.getLogger(getContext().system(), this);
  66
  67     // By default persistent will be true and can be turned off using the system
  68     // property persistent
  69     private final boolean persistent;
  70
  71     private Shard(String name) {
  72
  73         String setting = System.getProperty("shard.persistent");
  74         this.persistent = !"false".equals(setting);
  75
  76         log.info("Creating shard : {} persistent : {}", name , persistent);
  77
  78         store = new InMemoryDOMDataStore(name, storeExecutor);
  79     }
  80
  81     public static Props props(final String name) {
  82         return Props.create(new Creator<Shard>() {
  83
  84             @Override
  85             public Shard create() throws Exception {
  86                 return new Shard(name);
  87             }
  88
  89         });
  90     }
  91
  92
  93     @Override
  94     public void onReceive(Object message) throws Exception {
  95         log.debug("Received message {}", message);
  96
  97         if (message instanceof CreateTransactionChain) {
  98             createTransactionChain();
  99         } else if (message instanceof RegisterChangeListener) {
 100             registerChangeListener((RegisterChangeListener) message);
 101         } else if (message instanceof UpdateSchemaContext) {
 102             updateSchemaContext((UpdateSchemaContext) message);
 103         } else if (message instanceof ForwardedCommitTransaction) {
 104             handleForwardedCommit((ForwardedCommitTransaction) message);
 105         } else if (message instanceof Persistent) {
 106             commit((Modification) ((Persistent) message).payload());
 107         } else if (message instanceof CreateTransaction) {
 108             createTransaction((CreateTransaction) message);
 109         } else if(message instanceof NonPersistent){
 110             commit((Modification) ((NonPersistent) message).payload());
 111         }
 112     }
 113
 114     private void createTransaction(CreateTransaction createTransaction) {
 115         DOMStoreReadWriteTransaction transaction =
 116             store.newReadWriteTransaction();
 117         ActorRef transactionActor = getContext().actorOf(
 118             ShardTransaction.props(transaction, getSelf()), "shard-" + createTransaction.getTransactionId());
 119         getSender()
 120             .tell(CreateTransactionReply.newBuilder()
 121                     .setTransactionActorPath(transactionActor.path().toString())
 122                     .setTransactionId(createTransaction.getTransactionId())
 123                     .build(),
 124                 getSelf());
 125     }
 126
 127     private void commit(Modification modification) {
 128         DOMStoreThreePhaseCommitCohort cohort =
 129             modificationToCohort.remove(modification);
 130         if (cohort == null) {
 131             log.error(
 132                 "Could not find cohort for modification : " + modification);
 133             return;
 134         }
 135         final ListenableFuture<Void> future = cohort.commit();
 136         final ActorRef sender = getSender();
 137         final ActorRef self = getSelf();
 138         future.addListener(new Runnable() {
 139             @Override
 140             public void run() {
 141                 try {
 142                     future.get();
 143                     sender.tell(new CommitTransactionReply(), self);
 144                 } catch (InterruptedException | ExecutionException e) {
 145                     // FIXME : Handle this properly
 146                     log.error(e, "An exception happened when committing");
 147                 }
 148             }
 149         }, getContext().dispatcher());
 150     }
 151
 152     private void handleForwardedCommit(ForwardedCommitTransaction message) {
 153         modificationToCohort
 154             .put(message.getModification(), message.getCohort());
 155         if(persistent) {
 156             getSelf().forward(Persistent.create(message.getModification()),
 157                 getContext());
 158         } else {
 159             getSelf().forward(NonPersistent.create(message.getModification()),
 160                 getContext());
 161         }
 162     }
 163
 164     private void updateSchemaContext(UpdateSchemaContext message) {
 165         store.onGlobalContextUpdated(message.getSchemaContext());
 166     }
 167
 168     private void registerChangeListener(
 169         RegisterChangeListener registerChangeListener) {
 170
 171         ActorSelection dataChangeListenerPath = getContext()
 172             .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
 173
 174         AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
 175             listener = new DataChangeListenerProxy(dataChangeListenerPath);
 176
 177         org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
 178             registration =
 179             store.registerChangeListener(registerChangeListener.getPath(),
 180                 listener, registerChangeListener.getScope());
 181         ActorRef listenerRegistration =
 182             getContext().actorOf(
 183                 DataChangeListenerRegistration.props(registration));
 184         getSender()
 185             .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
 186                 getSelf());
 187     }
 188
 189     private void createTransactionChain() {
 190         DOMStoreTransactionChain chain = store.createTransactionChain();
 191         ActorRef transactionChain =
 192             getContext().actorOf(ShardTransactionChain.props(chain));
 193         getSender()
 194             .tell(new CreateTransactionChainReply(transactionChain.path()),
 195                 getSelf());
 196     }
 197 }