package org.opendaylight.controller.cluster.datastore;
+import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Uninterruptibles;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
String shardDispatcher =
new Dispatchers(actorSystem.dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
- actorContext = new ActorContext(actorSystem, actorSystem.actorOf(
- ShardManager.props(cluster, configuration, datastoreContext, waitTillReadyCountDownLatch)
- .withDispatcher(shardDispatcher).withMailbox(ActorContext.MAILBOX), shardManagerId ),
- cluster, configuration, datastoreContext);
+ actorContext = new ActorContext(actorSystem, createShardManager(actorSystem, cluster, configuration,
+ datastoreContext, shardDispatcher, shardManagerId ), cluster, configuration, datastoreContext);
this.waitTillReadyTimeInMillis =
actorContext.getDatastoreContext().getShardLeaderElectionTimeout().duration().toMillis() * READY_WAIT_FACTOR;
try {
closeable.close();
} catch (Exception e) {
- LOG.debug("Error closing insance", e);
+ LOG.debug("Error closing instance", e);
}
}
actorContext.shutdown();
+ DistributedDataStoreFactory.destroyInstance(this);
}
@VisibleForTesting
}
}
+ private ActorRef createShardManager(ActorSystem actorSystem, ClusterWrapper cluster, Configuration configuration,
+ DatastoreContext datastoreContext, String shardDispatcher, String shardManagerId){
+ Exception lastException = null;
+
+ for(int i=0;i<100;i++) {
+ try {
+ return actorSystem.actorOf(
+ ShardManager.props(cluster, configuration, datastoreContext, waitTillReadyCountDownLatch)
+ .withDispatcher(shardDispatcher).withMailbox(ActorContext.MAILBOX), shardManagerId);
+ } catch (Exception e){
+ lastException = e;
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+ LOG.debug(String.format("Could not create actor %s because of %s - waiting for sometime before retrying (retry count = %d)", shardManagerId, e.getMessage(), i));
+ }
+ }
+
+ throw new IllegalStateException("Failed to create Shard Manager", lastException);
+ }
+
@VisibleForTesting
public CountDownLatch getWaitTillReadyCountDownLatch() {
return waitTillReadyCountDownLatch;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.osgi.BundleDelegatingClassLoader;
+import com.google.common.base.Preconditions;
import com.typesafe.config.ConfigFactory;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.datastore.config.ConfigurationReader;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.sal.core.api.model.SchemaService;
import org.osgi.framework.BundleContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.duration.Duration;
public class DistributedDataStoreFactory {
private static final String ACTOR_SYSTEM_NAME = "opendaylight-cluster-data";
private static final String CONFIGURATION_NAME = "odl-cluster-data";
+ private static ActorSystem actorSystem = null;
+ private static final Set<DistributedDataStore> createdInstances = new HashSet<>(2);
+ private static final Logger LOG = LoggerFactory.getLogger(DistributedDataStoreFactory.class);
- private static volatile ActorSystem persistentActorSystem = null;
-
- public static DistributedDataStore createInstance(SchemaService schemaService,
+ public static synchronized DistributedDataStore createInstance(SchemaService schemaService,
DatastoreContext datastoreContext, BundleContext bundleContext) {
+ LOG.info("Create data store instance of type : {}", datastoreContext.getDataStoreType());
+
DatastoreContextIntrospector introspector = new DatastoreContextIntrospector(datastoreContext);
DatastoreContextConfigAdminOverlay overlay = new DatastoreContextConfigAdminOverlay(
introspector, bundleContext);
- ActorSystem actorSystem = getOrCreateInstance(bundleContext, datastoreContext.getConfigurationReader());
+ ActorSystem actorSystem = getActorSystem(bundleContext, datastoreContext.getConfigurationReader());
Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
final DistributedDataStore dataStore = new DistributedDataStore(actorSystem,
new ClusterWrapperImpl(actorSystem), config, introspector.getContext());
dataStore.setCloseable(overlay);
dataStore.waitTillReady();
+
+ createdInstances.add(dataStore);
return dataStore;
}
- private static final ActorSystem getOrCreateInstance(final BundleContext bundleContext, ConfigurationReader configurationReader) {
- ActorSystem ret = persistentActorSystem;
- if (ret == null) {
- synchronized (DistributedDataStoreFactory.class) {
- ret = persistentActorSystem;
- if (ret == null) {
- // Create an OSGi bundle classloader for actor system
- BundleDelegatingClassLoader classLoader = new BundleDelegatingClassLoader(bundleContext.getBundle(),
- Thread.currentThread().getContextClassLoader());
-
- ret = ActorSystem.create(ACTOR_SYSTEM_NAME,
- ConfigFactory.load(configurationReader.read()).getConfig(CONFIGURATION_NAME), classLoader);
- ret.actorOf(Props.create(TerminationMonitor.class), "termination-monitor");
-
- persistentActorSystem = ret;
+ private static synchronized final ActorSystem getActorSystem(final BundleContext bundleContext,
+ ConfigurationReader configurationReader) {
+ if (actorSystem == null) {
+ // Create an OSGi bundle classloader for actor system
+ BundleDelegatingClassLoader classLoader = new BundleDelegatingClassLoader(bundleContext.getBundle(),
+ Thread.currentThread().getContextClassLoader());
+
+ actorSystem = ActorSystem.create(ACTOR_SYSTEM_NAME,
+ ConfigFactory.load(configurationReader.read()).getConfig(CONFIGURATION_NAME), classLoader);
+ actorSystem.actorOf(Props.create(TerminationMonitor.class), "termination-monitor");
+ }
+
+ return actorSystem;
+ }
+
+ public static synchronized void destroyInstance(DistributedDataStore dataStore){
+ Preconditions.checkNotNull(dataStore, "dataStore should not be null");
+
+ LOG.info("Destroy data store instance of type : {}", dataStore.getActorContext().getDataStoreType());
+
+ if(createdInstances.remove(dataStore)){
+ if(createdInstances.size() == 0){
+ if(actorSystem != null) {
+ actorSystem.shutdown();
+ try {
+ actorSystem.awaitTermination(Duration.create(10, TimeUnit.SECONDS));
+ } catch (Exception e) {
+ LOG.warn("Error awaiting actor termination", e);
+ }
+ actorSystem = null;
}
}
}
-
- return ret;
}
+
}
Code Review / controller.git / commitdiff