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 }