opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DistributedDataStore.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.ActorSystem;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.ListeningExecutorService;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
  17 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
  18 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
  19 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  20 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  21 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
  23 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
  24 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
  25 import org.opendaylight.controller.sal.core.spi.data.DOMStore;
  26 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  28 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  29 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  30 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  31 import org.opendaylight.yangtools.util.PropertyUtils;
  32 import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
  33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  35 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  36 import org.opendaylight.yangtools.yang.model.api.SchemaContextListener;
  37 import org.slf4j.Logger;
  38 import org.slf4j.LoggerFactory;
  39
  40 /**
  41  *
  42  */
  43 public class DistributedDataStore implements DOMStore, SchemaContextListener, AutoCloseable {
  44
  45     private static final Logger
  46         LOG = LoggerFactory.getLogger(DistributedDataStore.class);
  47
  48     private static final String EXECUTOR_MAX_POOL_SIZE_PROP =
  49             "mdsal.dist-datastore-executor-pool.size";
  50     private static final int DEFAULT_EXECUTOR_MAX_POOL_SIZE = 10;
  51
  52     private static final String EXECUTOR_MAX_QUEUE_SIZE_PROP =
  53             "mdsal.dist-datastore-executor-queue.size";
  54     private static final int DEFAULT_EXECUTOR_MAX_QUEUE_SIZE = 5000;
  55
  56     private final ActorContext actorContext;
  57
  58     private SchemaContext schemaContext;
  59
  60     /**
  61      * Executor used to run FutureTask's
  62      *
  63      * This is typically used when we need to make a request to an actor and
  64      * wait for it's response and the consumer needs to be provided a Future.
  65      */
  66     private final ListeningExecutorService executor =
  67             MoreExecutors.listeningDecorator(
  68                     SpecialExecutors.newBlockingBoundedFastThreadPool(
  69                             PropertyUtils.getIntSystemProperty(
  70                                     EXECUTOR_MAX_POOL_SIZE_PROP,
  71                                     DEFAULT_EXECUTOR_MAX_POOL_SIZE),
  72                             PropertyUtils.getIntSystemProperty(
  73                                     EXECUTOR_MAX_QUEUE_SIZE_PROP,
  74                                     DEFAULT_EXECUTOR_MAX_QUEUE_SIZE), "DistDataStore"));
  75
  76     public DistributedDataStore(ActorSystem actorSystem, String type, ClusterWrapper cluster,
  77             Configuration configuration, InMemoryDOMDataStoreConfigProperties dataStoreProperties) {
  78         Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");
  79         Preconditions.checkNotNull(type, "type should not be null");
  80         Preconditions.checkNotNull(cluster, "cluster should not be null");
  81         Preconditions.checkNotNull(configuration, "configuration should not be null");
  82
  83
  84         String shardManagerId = ShardManagerIdentifier.builder().type(type).build().toString();
  85
  86         LOG.info("Creating ShardManager : {}", shardManagerId);
  87
  88         this.actorContext = new ActorContext(actorSystem, actorSystem
  89             .actorOf(ShardManager.props(type, cluster, configuration, dataStoreProperties),
  90                 shardManagerId ), cluster, configuration);
  91     }
  92
  93     public DistributedDataStore(ActorContext actorContext) {
  94         this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null");
  95     }
  96
  97
  98     @Override
  99     public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
 100         YangInstanceIdentifier path, L listener,
 101         AsyncDataBroker.DataChangeScope scope) {
 102
 103         Preconditions.checkNotNull(path, "path should not be null");
 104         Preconditions.checkNotNull(listener, "listener should not be null");
 105
 106
 107         LOG.debug("Registering listener: {} for path: {} scope: {}", listener, path, scope);
 108
 109         ActorRef dataChangeListenerActor = actorContext.getActorSystem().actorOf(
 110             DataChangeListener.props(schemaContext,listener,path ));
 111
 112         String shardName = ShardStrategyFactory.getStrategy(path).findShard(path);
 113
 114         Object result = actorContext.executeLocalShardOperation(shardName,
 115             new RegisterChangeListener(path, dataChangeListenerActor.path(),
 116                 scope),
 117             ActorContext.ASK_DURATION
 118         );
 119
 120         if (result != null) {
 121             RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
 122             return new DataChangeListenerRegistrationProxy(actorContext
 123                 .actorSelection(reply.getListenerRegistrationPath()), listener,
 124                 dataChangeListenerActor);
 125         }
 126
 127         LOG.debug(
 128             "No local shard for shardName {} was found so returning a noop registration",
 129             shardName);
 130         return new NoOpDataChangeListenerRegistration(listener);
 131     }
 132
 133
 134
 135
 136
 137     @Override
 138     public DOMStoreTransactionChain createTransactionChain() {
 139         return new TransactionChainProxy(actorContext, executor, schemaContext);
 140     }
 141
 142     @Override
 143     public DOMStoreReadTransaction newReadOnlyTransaction() {
 144         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
 145             executor, schemaContext);
 146     }
 147
 148     @Override
 149     public DOMStoreWriteTransaction newWriteOnlyTransaction() {
 150         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY,
 151             executor, schemaContext);
 152     }
 153
 154     @Override
 155     public DOMStoreReadWriteTransaction newReadWriteTransaction() {
 156         return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE,
 157             executor, schemaContext);
 158     }
 159
 160     @Override public void onGlobalContextUpdated(SchemaContext schemaContext) {
 161         this.schemaContext = schemaContext;
 162         actorContext.getShardManager().tell(
 163             new UpdateSchemaContext(schemaContext), null);
 164     }
 165
 166     @Override public void close() throws Exception {
 167         actorContext.shutdown();
 168
 169     }
 170 }