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;
private final String type;
+ private final TransactionContextFactory txContextFactory;
+
public DistributedDataStore(ActorSystem actorSystem, ClusterWrapper cluster,
Configuration configuration, DatastoreContext datastoreContext) {
Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");
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;
+ this.txContextFactory = TransactionContextFactory.create(actorContext);
datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl(datastoreContext.getDataStoreMXBeanType());
datastoreConfigMXBean.setContext(datastoreContext);
public DistributedDataStore(ActorContext actorContext) {
this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null");
+ this.txContextFactory = TransactionContextFactory.create(actorContext);
this.type = UNKNOWN_TYPE;
this.waitTillReadyTimeInMillis =
actorContext.getDatastoreContext().getShardLeaderElectionTimeout().duration().toMillis() * READY_WAIT_FACTOR;
-
}
public void setCloseable(AutoCloseable closeable) {
@Override
public DOMStoreTransactionChain createTransactionChain() {
- return new TransactionChainProxy(actorContext);
+ return txContextFactory.createTransactionChain();
}
@Override
public DOMStoreReadTransaction newReadOnlyTransaction() {
- return new TransactionProxy(actorContext, TransactionType.READ_ONLY);
+ return new TransactionProxy(txContextFactory, TransactionType.READ_ONLY);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
actorContext.acquireTxCreationPermit();
- return new TransactionProxy(actorContext, TransactionType.WRITE_ONLY);
+ return new TransactionProxy(txContextFactory, TransactionType.WRITE_ONLY);
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
actorContext.acquireTxCreationPermit();
- return new TransactionProxy(actorContext, TransactionType.READ_WRITE);
+ return new TransactionProxy(txContextFactory, TransactionType.READ_WRITE);
}
@Override
datastoreConfigMXBean.unregisterMBean();
datastoreInfoMXBean.unregisterMBean();
- if(closeable != null) {
+ if (closeable != null) {
try {
closeable.close();
} catch (Exception e) {
- LOG.debug("Error closing insance", e);
+ LOG.debug("Error closing instance", e);
}
}
+ txContextFactory.close();
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;