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