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.serialization.Serialization;
  18
  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
  23 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
  24 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
  25 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
  26 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
  27 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
  28 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
  29 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
  30 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
  31 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
  32 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  33 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
  34 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  35 import org.opendaylight.controller.cluster.datastore.modification.Modification;
  36 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
  37 import org.opendaylight.controller.cluster.raft.RaftActor;
  38 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  39 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
  40 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  41 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  42 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  43 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  44 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  45 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  46
  47 import java.util.HashMap;
  48 import java.util.Map;
  49 import java.util.concurrent.ExecutionException;
  50 import java.util.concurrent.Executors;
  51
  52 /**
  53  * A Shard represents a portion of the logical data tree <br/>
  54  * <p>
  55  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
  56  * </p>
  57  */
  58 public class Shard extends RaftActor {
  59
  60     public static final String DEFAULT_NAME = "default";
  61
  62     private final ListeningExecutorService storeExecutor =
  63         MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(2));
  64
  65     private final InMemoryDOMDataStore store;
  66
  67     private final Map<Object, DOMStoreThreePhaseCommitCohort>
  68         modificationToCohort = new HashMap<>();
  69
  70     private final LoggingAdapter LOG =
  71         Logging.getLogger(getContext().system(), this);
  72
  73     // By default persistent will be true and can be turned off using the system
  74     // property persistent
  75     private final boolean persistent;
  76
  77     private final String name;
  78
  79     private volatile SchemaContext schemaContext;
  80
  81     private final ShardStats shardMBean;
  82
  83     private Shard(String name, Map<String, String> peerAddresses) {
  84         super(name, peerAddresses);
  85
  86         this.name = name;
  87
  88         String setting = System.getProperty("shard.persistent");
  89
  90         this.persistent = !"false".equals(setting);
  91
  92         LOG.info("Creating shard : {} persistent : {}", name, persistent);
  93
  94         store = new InMemoryDOMDataStore(name, storeExecutor);
  95
  96         shardMBean = ShardMBeanFactory.getShardStatsMBean(name);
  97
  98     }
  99
 100     public static Props props(final String name, final Map<String, String> peerAddresses) {
 101         return Props.create(new Creator<Shard>() {
 102
 103             @Override
 104             public Shard create() throws Exception {
 105                 return new Shard(name, peerAddresses);
 106             }
 107
 108         });
 109     }
 110
 111
 112     @Override public void onReceiveCommand(Object message){
 113         LOG.debug("Received message {} from {}", message.getClass().toString(), getSender());
 114
 115         if (message.getClass().equals(CreateTransactionChain.SERIALIZABLE_CLASS)) {
 116             if(isLeader()) {
 117                 createTransactionChain();
 118             } else if(getLeader() != null){
 119                 getLeader().forward(message, getContext());
 120             }
 121         } else if (message.getClass().equals(RegisterChangeListener.SERIALIZABLE_CLASS)) {
 122             registerChangeListener(RegisterChangeListener.fromSerializable(getContext().system(), message));
 123         } else if (message instanceof UpdateSchemaContext) {
 124             updateSchemaContext((UpdateSchemaContext) message);
 125         } else if (message instanceof ForwardedCommitTransaction) {
 126             handleForwardedCommit((ForwardedCommitTransaction) message);
 127         } else if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
 128             if(isLeader()) {
 129                 createTransaction(CreateTransaction.fromSerializable(message));
 130             } else if(getLeader() != null){
 131                 getLeader().forward(message, getContext());
 132             }
 133         } else if (message instanceof PeerAddressResolved){
 134             PeerAddressResolved resolved = (PeerAddressResolved) message;
 135             setPeerAddress(resolved.getPeerId(), resolved.getPeerAddress());
 136         } else {
 137           super.onReceiveCommand(message);
 138         }
 139     }
 140
 141     private void createTransaction(CreateTransaction createTransaction) {
 142         DOMStoreReadWriteTransaction transaction =
 143             store.newReadWriteTransaction();
 144         String transactionId = "shard-" + createTransaction.getTransactionId();
 145         LOG.info("Creating transaction : {} " , transactionId);
 146         ActorRef transactionActor = getContext().actorOf(
 147             ShardTransaction.props(transaction, getSelf(), schemaContext), transactionId);
 148
 149         getSender()
 150             .tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor), createTransaction.getTransactionId()).toSerializable(),
 151                 getSelf());
 152     }
 153
 154     private void commit(final ActorRef sender, Object serialized) {
 155         Modification modification = MutableCompositeModification.fromSerializable(serialized, schemaContext);
 156         DOMStoreThreePhaseCommitCohort cohort =
 157             modificationToCohort.remove(serialized);
 158         if (cohort == null) {
 159             LOG.error(
 160                 "Could not find cohort for modification : {}", modification);
 161             LOG.info("Writing modification using a new transaction");
 162             DOMStoreReadWriteTransaction transaction =
 163                 store.newReadWriteTransaction();
 164             modification.apply(transaction);
 165             DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
 166             ListenableFuture<Void> future =
 167                 commitCohort.preCommit();
 168             try {
 169                 future.get();
 170                 future = commitCohort.commit();
 171                 future.get();
 172             } catch (InterruptedException e) {
 173                 LOG.error("Failed to commit", e);
 174             } catch (ExecutionException e) {
 175                 LOG.error("Failed to commit", e);
 176             }
 177         }
 178
 179         final ListenableFuture<Void> future = cohort.commit();
 180         shardMBean.incrementCommittedTransactionCount();
 181         final ActorRef self = getSelf();
 182         future.addListener(new Runnable() {
 183             @Override
 184             public void run() {
 185                 try {
 186                     future.get();
 187
 188                     if(sender != null) {
 189                         sender
 190                             .tell(new CommitTransactionReply().toSerializable(),
 191                                 self);
 192                     } else {
 193                         LOG.error("sender is null ???");
 194                     }
 195                 } catch (InterruptedException | ExecutionException e) {
 196                     // FIXME : Handle this properly
 197                     LOG.error(e, "An exception happened when committing");
 198                 }
 199             }
 200         }, getContext().dispatcher());
 201     }
 202
 203     private void handleForwardedCommit(ForwardedCommitTransaction message) {
 204         Object serializedModification = message.getModification().toSerializable();
 205
 206         modificationToCohort
 207             .put(serializedModification , message.getCohort());
 208
 209         if(persistent) {
 210             this.persistData(getSender(), "identifier", new CompositeModificationPayload(serializedModification));
 211         } else {
 212             this.commit(getSender(), serializedModification);
 213         }
 214     }
 215
 216     private void updateSchemaContext(UpdateSchemaContext message) {
 217         this.schemaContext = message.getSchemaContext();
 218         store.onGlobalContextUpdated(message.getSchemaContext());
 219     }
 220
 221     private void registerChangeListener(
 222         RegisterChangeListener registerChangeListener) {
 223
 224         LOG.debug("registerDataChangeListener for " + registerChangeListener.getPath());
 225
 226
 227         ActorSelection dataChangeListenerPath = getContext()
 228             .system().actorSelection(
 229                 registerChangeListener.getDataChangeListenerPath());
 230
 231         AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>
 232             listener = new DataChangeListenerProxy(schemaContext,dataChangeListenerPath);
 233
 234         org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
 235             registration =
 236             store.registerChangeListener(registerChangeListener.getPath(),
 237                 listener, registerChangeListener.getScope());
 238         ActorRef listenerRegistration =
 239             getContext().actorOf(
 240                 DataChangeListenerRegistration.props(registration));
 241
 242         LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = " + listenerRegistration.path().toString());
 243
 244         getSender()
 245             .tell(new RegisterChangeListenerReply(listenerRegistration.path()).toSerializable(),
 246                 getSelf());
 247     }
 248
 249     private void createTransactionChain() {
 250         DOMStoreTransactionChain chain = store.createTransactionChain();
 251         ActorRef transactionChain =
 252             getContext().actorOf(
 253                 ShardTransactionChain.props(chain, schemaContext));
 254         getSender()
 255             .tell(new CreateTransactionChainReply(transactionChain.path())
 256                 .toSerializable(),
 257                 getSelf());
 258     }
 259
 260     @Override protected void applyState(ActorRef clientActor, String identifier,
 261         Object data) {
 262
 263         if(data instanceof CompositeModificationPayload){
 264             Object modification =
 265                 ((CompositeModificationPayload) data).getModification();
 266
 267             if(modification != null){
 268                 commit(clientActor, modification);
 269             } else {
 270                 LOG.error("modification is null - this is very unexpected");
 271             }
 272
 273
 274         } else {
 275             LOG.error("Unknown state received {}", data);
 276         }
 277
 278     }
 279
 280     @Override protected Object createSnapshot() {
 281         throw new UnsupportedOperationException("createSnapshot");
 282     }
 283
 284     @Override protected void applySnapshot(Object snapshot) {
 285         throw new UnsupportedOperationException("applySnapshot");
 286     }
 287
 288     @Override public String persistenceId() {
 289         return this.name;
 290     }
 291 }