Simplify code using Java 8 features
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / sharding / DistributedShardedDOMDataTree.java
1 /*
2  * Copyright (c) 2016, 2017 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.sharding;
10
11 import static akka.actor.ActorRef.noSender;
12
13 import akka.actor.ActorRef;
14 import akka.actor.ActorSystem;
15 import akka.actor.PoisonPill;
16 import akka.actor.Props;
17 import akka.dispatch.Mapper;
18 import akka.dispatch.OnComplete;
19 import akka.pattern.Patterns;
20 import akka.util.Timeout;
21 import com.google.common.base.Optional;
22 import com.google.common.base.Preconditions;
23 import com.google.common.base.Throwables;
24 import com.google.common.collect.ClassToInstanceMap;
25 import com.google.common.collect.ForwardingObject;
26 import com.google.common.collect.ImmutableClassToInstanceMap;
27 import com.google.common.util.concurrent.FutureCallback;
28 import com.google.common.util.concurrent.Futures;
29 import com.google.common.util.concurrent.ListenableFuture;
30 import com.google.common.util.concurrent.MoreExecutors;
31 import com.google.common.util.concurrent.SettableFuture;
32 import com.google.common.util.concurrent.Uninterruptibles;
33 import java.util.AbstractMap.SimpleEntry;
34 import java.util.Collection;
35 import java.util.Collections;
36 import java.util.Comparator;
37 import java.util.EnumMap;
38 import java.util.HashMap;
39 import java.util.List;
40 import java.util.Map;
41 import java.util.Map.Entry;
42 import java.util.Set;
43 import java.util.concurrent.CompletionStage;
44 import java.util.concurrent.ExecutionException;
45 import java.util.concurrent.TimeUnit;
46 import java.util.concurrent.TimeoutException;
47 import javax.annotation.Nonnull;
48 import javax.annotation.Nullable;
49 import javax.annotation.concurrent.GuardedBy;
50 import org.opendaylight.controller.cluster.ActorSystemProvider;
51 import org.opendaylight.controller.cluster.access.concepts.MemberName;
52 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
53 import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
54 import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
55 import org.opendaylight.controller.cluster.datastore.Shard;
56 import org.opendaylight.controller.cluster.datastore.config.Configuration;
57 import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
58 import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
59 import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
60 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
61 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
62 import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
63 import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
64 import org.opendaylight.controller.cluster.sharding.ShardedDataTreeActor.ShardedDataTreeActorCreator;
65 import org.opendaylight.controller.cluster.sharding.messages.InitConfigListener;
66 import org.opendaylight.controller.cluster.sharding.messages.LookupPrefixShard;
67 import org.opendaylight.controller.cluster.sharding.messages.PrefixShardRemovalLookup;
68 import org.opendaylight.controller.cluster.sharding.messages.ProducerCreated;
69 import org.opendaylight.controller.cluster.sharding.messages.ProducerRemoved;
70 import org.opendaylight.controller.cluster.sharding.messages.StartConfigShardLookup;
71 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
72 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
73 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
74 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
75 import org.opendaylight.mdsal.dom.api.DOMDataTreeLoopException;
76 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
77 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducerException;
78 import org.opendaylight.mdsal.dom.api.DOMDataTreeService;
79 import org.opendaylight.mdsal.dom.api.DOMDataTreeServiceExtension;
80 import org.opendaylight.mdsal.dom.api.DOMDataTreeShard;
81 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
82 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingService;
83 import org.opendaylight.mdsal.dom.broker.DOMDataTreeShardRegistration;
84 import org.opendaylight.mdsal.dom.broker.ShardedDOMDataTree;
85 import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTable;
86 import org.opendaylight.mdsal.dom.spi.DOMDataTreePrefixTableEntry;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.clustering.prefix.shard.configuration.rev170110.PrefixShards;
88 import org.opendaylight.yangtools.concepts.ListenerRegistration;
89 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
90 import org.slf4j.Logger;
91 import org.slf4j.LoggerFactory;
92 import scala.compat.java8.FutureConverters;
93 import scala.concurrent.Future;
94 import scala.concurrent.Promise;
95 import scala.concurrent.duration.FiniteDuration;
96
97 /**
98  * A layer on top of DOMDataTreeService that distributes producer/shard registrations to remote nodes via
99  * {@link ShardedDataTreeActor}. Also provides QoL method for addition of prefix based clustered shard into the system.
100  */
101 public class DistributedShardedDOMDataTree implements DOMDataTreeService, DOMDataTreeShardingService,
102         DistributedShardFactory {
103
104     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTree.class);
105
106     private static final int MAX_ACTOR_CREATION_RETRIES = 100;
107     private static final int ACTOR_RETRY_DELAY = 100;
108     private static final TimeUnit ACTOR_RETRY_TIME_UNIT = TimeUnit.MILLISECONDS;
109     private static final int LOOKUP_TASK_MAX_RETRIES = 100;
110     static final FiniteDuration SHARD_FUTURE_TIMEOUT_DURATION =
111             new FiniteDuration(LOOKUP_TASK_MAX_RETRIES * LOOKUP_TASK_MAX_RETRIES * 3, TimeUnit.SECONDS);
112     static final Timeout SHARD_FUTURE_TIMEOUT = new Timeout(SHARD_FUTURE_TIMEOUT_DURATION);
113
114     static final String ACTOR_ID = "ShardedDOMDataTreeFrontend";
115
116     private final ShardedDOMDataTree shardedDOMDataTree;
117     private final ActorSystem actorSystem;
118     private final AbstractDataStore distributedOperDatastore;
119     private final AbstractDataStore distributedConfigDatastore;
120
121     private final ActorRef shardedDataTreeActor;
122     private final MemberName memberName;
123
124     @GuardedBy("shards")
125     private final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shards =
126             DOMDataTreePrefixTable.create();
127
128     private final EnumMap<LogicalDatastoreType, Entry<DataStoreClient, ActorRef>> configurationShardMap =
129             new EnumMap<>(LogicalDatastoreType.class);
130
131     private final EnumMap<LogicalDatastoreType, PrefixedShardConfigWriter> writerMap =
132             new EnumMap<>(LogicalDatastoreType.class);
133
134     private final PrefixedShardConfigUpdateHandler updateHandler;
135
136     public DistributedShardedDOMDataTree(final ActorSystemProvider actorSystemProvider,
137                                          final AbstractDataStore distributedOperDatastore,
138                                          final AbstractDataStore distributedConfigDatastore) {
139         this.actorSystem = Preconditions.checkNotNull(actorSystemProvider).getActorSystem();
140         this.distributedOperDatastore = Preconditions.checkNotNull(distributedOperDatastore);
141         this.distributedConfigDatastore = Preconditions.checkNotNull(distributedConfigDatastore);
142         shardedDOMDataTree = new ShardedDOMDataTree();
143
144         shardedDataTreeActor = createShardedDataTreeActor(actorSystem,
145                 new ShardedDataTreeActorCreator()
146                         .setShardingService(this)
147                         .setActorSystem(actorSystem)
148                         .setClusterWrapper(distributedConfigDatastore.getActorContext().getClusterWrapper())
149                         .setDistributedConfigDatastore(distributedConfigDatastore)
150                         .setDistributedOperDatastore(distributedOperDatastore)
151                         .setLookupTaskMaxRetries(LOOKUP_TASK_MAX_RETRIES),
152                 ACTOR_ID);
153
154         this.memberName = distributedConfigDatastore.getActorContext().getCurrentMemberName();
155
156         updateHandler = new PrefixedShardConfigUpdateHandler(shardedDataTreeActor,
157                 distributedConfigDatastore.getActorContext().getCurrentMemberName());
158
159         LOG.debug("{} - Starting prefix configuration shards", memberName);
160         createPrefixConfigShard(distributedConfigDatastore);
161         createPrefixConfigShard(distributedOperDatastore);
162     }
163
164     private static void createPrefixConfigShard(final AbstractDataStore dataStore) {
165         Configuration configuration = dataStore.getActorContext().getConfiguration();
166         Collection<MemberName> memberNames = configuration.getUniqueMemberNamesForAllShards();
167         CreateShard createShardMessage =
168                 new CreateShard(new ModuleShardConfiguration(PrefixShards.QNAME.getNamespace(),
169                         "prefix-shard-configuration", ClusterUtils.PREFIX_CONFIG_SHARD_ID, ModuleShardStrategy.NAME,
170                         memberNames),
171                         Shard.builder(), dataStore.getActorContext().getDatastoreContext());
172
173         dataStore.getActorContext().getShardManager().tell(createShardMessage, noSender());
174     }
175
176     /**
177      * This will try to initialize prefix configuration shards upon their
178      * successful start. We need to create writers to these shards, so we can
179      * satisfy future {@link #createDistributedShard} and
180      * {@link #resolveShardAdditions} requests and update prefix configuration
181      * shards accordingly.
182      *
183      * <p>
184      * We also need to initialize listeners on these shards, so we can react
185      * on changes made on them by other cluster members or even by ourselves.
186      *
187      * <p>
188      * Finally, we need to be sure that default shards for both operational and
189      * configuration data stores are up and running and we have distributed
190      * shards frontend created for them.
191      *
192      * <p>
193      * This is intended to be invoked by blueprint as initialization method.
194      */
195     public void init() {
196         // create our writers to the configuration
197         try {
198             LOG.debug("{} - starting config shard lookup.", memberName);
199
200             // We have to wait for prefix config shards to be up and running
201             // so we can create datastore clients for them
202             handleConfigShardLookup().get(SHARD_FUTURE_TIMEOUT_DURATION.length(), SHARD_FUTURE_TIMEOUT_DURATION.unit());
203         } catch (InterruptedException | ExecutionException | TimeoutException e) {
204             throw new IllegalStateException("Prefix config shards not found", e);
205         }
206
207         try {
208             LOG.debug("{}: Prefix configuration shards ready - creating clients", memberName);
209             configurationShardMap.put(LogicalDatastoreType.CONFIGURATION,
210                     createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
211                             distributedConfigDatastore.getActorContext()));
212         } catch (final DOMDataTreeShardCreationFailedException e) {
213             throw new IllegalStateException(
214                     "Unable to create datastoreClient for config DS prefix configuration shard.", e);
215         }
216
217         try {
218             configurationShardMap.put(LogicalDatastoreType.OPERATIONAL,
219                     createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
220                             distributedOperDatastore.getActorContext()));
221
222         } catch (final DOMDataTreeShardCreationFailedException e) {
223             throw new IllegalStateException(
224                         "Unable to create datastoreClient for oper DS prefix configuration shard.", e);
225         }
226
227         writerMap.put(LogicalDatastoreType.CONFIGURATION, new PrefixedShardConfigWriter(
228                 configurationShardMap.get(LogicalDatastoreType.CONFIGURATION).getKey()));
229
230         writerMap.put(LogicalDatastoreType.OPERATIONAL, new PrefixedShardConfigWriter(
231                 configurationShardMap.get(LogicalDatastoreType.OPERATIONAL).getKey()));
232
233         updateHandler.initListener(distributedConfigDatastore, LogicalDatastoreType.CONFIGURATION);
234         updateHandler.initListener(distributedOperDatastore, LogicalDatastoreType.OPERATIONAL);
235
236         distributedConfigDatastore.getActorContext().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
237         distributedOperDatastore.getActorContext().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
238
239
240         //create shard registration for DEFAULT_SHARD
241         initDefaultShard(LogicalDatastoreType.CONFIGURATION);
242         initDefaultShard(LogicalDatastoreType.OPERATIONAL);
243     }
244
245     private ListenableFuture<List<Void>> handleConfigShardLookup() {
246
247         final ListenableFuture<Void> configFuture = lookupConfigShard(LogicalDatastoreType.CONFIGURATION);
248         final ListenableFuture<Void> operFuture = lookupConfigShard(LogicalDatastoreType.OPERATIONAL);
249
250         return Futures.allAsList(configFuture, operFuture);
251     }
252
253     private ListenableFuture<Void> lookupConfigShard(final LogicalDatastoreType type) {
254         final SettableFuture<Void> future = SettableFuture.create();
255
256         final Future<Object> ask =
257                 Patterns.ask(shardedDataTreeActor, new StartConfigShardLookup(type), SHARD_FUTURE_TIMEOUT);
258
259         ask.onComplete(new OnComplete<Object>() {
260             @Override
261             public void onComplete(final Throwable throwable, final Object result) {
262                 if (throwable != null) {
263                     future.setException(throwable);
264                 } else {
265                     future.set(null);
266                 }
267             }
268         }, actorSystem.dispatcher());
269
270         return future;
271     }
272
273     @Nonnull
274     @Override
275     public <T extends DOMDataTreeListener> ListenerRegistration<T> registerListener(
276             final T listener, final Collection<DOMDataTreeIdentifier> subtrees,
277             final boolean allowRxMerges, final Collection<DOMDataTreeProducer> producers)
278             throws DOMDataTreeLoopException {
279         return shardedDOMDataTree.registerListener(listener, subtrees, allowRxMerges, producers);
280     }
281
282     @Override
283     public ClassToInstanceMap<DOMDataTreeServiceExtension> getExtensions() {
284         return ImmutableClassToInstanceMap.of();
285     }
286
287     @Nonnull
288     @Override
289     public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
290         LOG.debug("{} - Creating producer for {}", memberName, subtrees);
291         final DOMDataTreeProducer producer = shardedDOMDataTree.createProducer(subtrees);
292
293         final Object response = distributedConfigDatastore.getActorContext()
294                 .executeOperation(shardedDataTreeActor, new ProducerCreated(subtrees));
295         if (response == null) {
296             LOG.debug("{} - Received success from remote nodes, creating producer:{}", memberName, subtrees);
297             return new ProxyProducer(producer, subtrees, shardedDataTreeActor,
298                     distributedConfigDatastore.getActorContext(), shards);
299         }
300
301         closeProducer(producer);
302
303         if (response instanceof Throwable) {
304             Throwables.throwIfUnchecked((Throwable) response);
305             throw new RuntimeException((Throwable) response);
306         }
307         throw new RuntimeException("Unexpected response to create producer received." + response);
308     }
309
310     @Override
311     public CompletionStage<DistributedShardRegistration> createDistributedShard(
312             final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers)
313             throws DOMDataTreeShardingConflictException {
314
315         synchronized (shards) {
316             final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
317                     shards.lookup(prefix);
318             if (lookup != null && lookup.getValue().getPrefix().equals(prefix)) {
319                 throw new DOMDataTreeShardingConflictException(
320                         "Prefix " + prefix + " is already occupied by another shard.");
321             }
322         }
323
324         final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType());
325
326         final ListenableFuture<Void> writeFuture =
327                 writer.writeConfig(prefix.getRootIdentifier(), replicaMembers);
328
329         final Promise<DistributedShardRegistration> shardRegistrationPromise = akka.dispatch.Futures.promise();
330         Futures.addCallback(writeFuture, new FutureCallback<Void>() {
331             @Override
332             public void onSuccess(@Nullable final Void result) {
333
334                 final Future<Object> ask =
335                         Patterns.ask(shardedDataTreeActor, new LookupPrefixShard(prefix), SHARD_FUTURE_TIMEOUT);
336
337                 shardRegistrationPromise.completeWith(ask.transform(
338                         new Mapper<Object, DistributedShardRegistration>() {
339                             @Override
340                             public DistributedShardRegistration apply(final Object parameter) {
341                                 return new DistributedShardRegistrationImpl(
342                                         prefix, shardedDataTreeActor, DistributedShardedDOMDataTree.this);
343                             }
344                         },
345                         new Mapper<Throwable, Throwable>() {
346                             @Override
347                             public Throwable apply(final Throwable throwable) {
348                                 return new DOMDataTreeShardCreationFailedException(
349                                         "Unable to create a cds shard.", throwable);
350                             }
351                         }, actorSystem.dispatcher()));
352             }
353
354             @Override
355             public void onFailure(final Throwable throwable) {
356                 shardRegistrationPromise.failure(
357                         new DOMDataTreeShardCreationFailedException("Unable to create a cds shard.", throwable));
358             }
359         }, MoreExecutors.directExecutor());
360
361         return FutureConverters.toJava(shardRegistrationPromise.future());
362     }
363
364     void resolveShardAdditions(final Set<DOMDataTreeIdentifier> additions) {
365         LOG.debug("{}: Resolving additions : {}", memberName, additions);
366         // we need to register the shards from top to bottom, so we need to atleast make sure the ordering reflects that
367         additions
368             .stream()
369             .sorted(Comparator.comparingInt(o -> o.getRootIdentifier().getPathArguments().size()))
370             .forEachOrdered(this::createShardFrontend);
371     }
372
373     void resolveShardRemovals(final Set<DOMDataTreeIdentifier> removals) {
374         LOG.debug("{}: Resolving removals : {}", memberName, removals);
375
376         // do we need to go from bottom to top?
377         removals.forEach(this::despawnShardFrontend);
378     }
379
380     private void createShardFrontend(final DOMDataTreeIdentifier prefix) {
381         LOG.debug("{}: Creating CDS shard for prefix: {}", memberName, prefix);
382         final String shardName = ClusterUtils.getCleanShardName(prefix.getRootIdentifier());
383         final AbstractDataStore distributedDataStore =
384                 prefix.getDatastoreType().equals(org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION)
385                         ? distributedConfigDatastore : distributedOperDatastore;
386
387         try (DOMDataTreeProducer producer = localCreateProducer(Collections.singletonList(prefix))) {
388             final Entry<DataStoreClient, ActorRef> entry =
389                     createDatastoreClient(shardName, distributedDataStore.getActorContext());
390
391             final DistributedShardFrontend shard =
392                     new DistributedShardFrontend(distributedDataStore, entry.getKey(), prefix);
393
394             final DOMDataTreeShardRegistration<DOMDataTreeShard> reg =
395                     shardedDOMDataTree.registerDataTreeShard(prefix, shard, producer);
396
397             synchronized (shards) {
398                 shards.store(prefix, reg);
399             }
400
401         } catch (final DOMDataTreeShardingConflictException e) {
402             LOG.error("{}: Prefix {} is already occupied by another shard",
403                     distributedConfigDatastore.getActorContext().getClusterWrapper().getCurrentMemberName(), prefix, e);
404         } catch (DOMDataTreeProducerException e) {
405             LOG.error("Unable to close producer", e);
406         } catch (DOMDataTreeShardCreationFailedException e) {
407             LOG.error("Unable to create datastore client for shard {}", prefix, e);
408         }
409     }
410
411     private void despawnShardFrontend(final DOMDataTreeIdentifier prefix) {
412         LOG.debug("{}: Removing CDS shard for prefix: {}", memberName, prefix);
413         final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup;
414         synchronized (shards) {
415             lookup = shards.lookup(prefix);
416         }
417
418         if (lookup == null || !lookup.getValue().getPrefix().equals(prefix)) {
419             LOG.debug("{}: Received despawn for non-existing CDS shard frontend, prefix: {}, ignoring..",
420                     memberName, prefix);
421             return;
422         }
423
424         lookup.getValue().close();
425         // need to remove from our local table thats used for tracking
426         synchronized (shards) {
427             shards.remove(prefix);
428         }
429
430         final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType());
431         final ListenableFuture<Void> future = writer.removeConfig(prefix.getRootIdentifier());
432
433         Futures.addCallback(future, new FutureCallback<Void>() {
434             @Override
435             public void onSuccess(@Nullable final Void result) {
436                 LOG.debug("{} - Succesfuly removed shard for {}", memberName, prefix);
437             }
438
439             @Override
440             public void onFailure(final Throwable throwable) {
441                 LOG.error("Removal of shard {} from configuration failed.", prefix, throwable);
442             }
443         }, MoreExecutors.directExecutor());
444     }
445
446     DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookupShardFrontend(
447             final DOMDataTreeIdentifier prefix) {
448         synchronized (shards) {
449             return shards.lookup(prefix);
450         }
451     }
452
453     DOMDataTreeProducer localCreateProducer(final Collection<DOMDataTreeIdentifier> prefix) {
454         return shardedDOMDataTree.createProducer(prefix);
455     }
456
457     @Nonnull
458     @Override
459     public <T extends DOMDataTreeShard> ListenerRegistration<T> registerDataTreeShard(
460             @Nonnull final DOMDataTreeIdentifier prefix,
461             @Nonnull final T shard,
462             @Nonnull final DOMDataTreeProducer producer)
463             throws DOMDataTreeShardingConflictException {
464
465         LOG.debug("Registering shard[{}] at prefix: {}", shard, prefix);
466
467         if (producer instanceof ProxyProducer) {
468             return shardedDOMDataTree.registerDataTreeShard(prefix, shard, ((ProxyProducer) producer).delegate());
469         }
470
471         return shardedDOMDataTree.registerDataTreeShard(prefix, shard, producer);
472     }
473
474     @SuppressWarnings("checkstyle:IllegalCatch")
475     private Entry<DataStoreClient, ActorRef> createDatastoreClient(
476             final String shardName, final ActorContext actorContext)
477             throws DOMDataTreeShardCreationFailedException {
478
479         LOG.debug("{}: Creating distributed datastore client for shard {}", memberName, shardName);
480         final Props distributedDataStoreClientProps =
481                 SimpleDataStoreClientActor.props(memberName, "Shard-" + shardName, actorContext, shardName);
482
483         final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
484         try {
485             return new SimpleEntry<>(SimpleDataStoreClientActor
486                     .getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS), clientActor);
487         } catch (final Exception e) {
488             LOG.error("{}: Failed to get actor for {}", distributedDataStoreClientProps, memberName, e);
489             clientActor.tell(PoisonPill.getInstance(), noSender());
490             throw new DOMDataTreeShardCreationFailedException(
491                     "Unable to create datastore client for shard{" + shardName + "}", e);
492         }
493     }
494
495     @SuppressWarnings("checkstyle:IllegalCatch")
496     private void initDefaultShard(final LogicalDatastoreType logicalDatastoreType) {
497
498         final PrefixedShardConfigWriter writer = writerMap.get(logicalDatastoreType);
499
500         if (writer.checkDefaultIsPresent()) {
501             LOG.debug("{}: Default shard for {} is already present in the config. Possibly saved in snapshot.",
502                     memberName, logicalDatastoreType);
503         } else {
504             try {
505                 // Currently the default shard configuration is present in the out-of-box modules.conf and is
506                 // expected to be present. So look up the local default shard here and create the frontend.
507
508                 // TODO we don't have to do it for config and operational default shard separately. Just one of them
509                 // should be enough
510                 final ActorContext actorContext = logicalDatastoreType == LogicalDatastoreType.CONFIGURATION
511                         ? distributedConfigDatastore.getActorContext() : distributedOperDatastore.getActorContext();
512
513                 final Optional<ActorRef> defaultLocalShardOptional =
514                         actorContext.findLocalShard(ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
515
516                 if (defaultLocalShardOptional.isPresent()) {
517                     LOG.debug("{}: Default shard for {} is already started, creating just frontend", memberName,
518                             logicalDatastoreType);
519                     createShardFrontend(new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY));
520                 }
521
522                 // The local shard isn't present - we assume that means the local member isn't in the replica list
523                 // and will be dynamically created later via an explicit add-shard-replica request. This is the
524                 // bootstrapping mechanism to add a new node into an existing cluster. The following code to create
525                 // the default shard as a prefix shard is problematic in this scenario so it is commented out. Since
526                 // the default shard is a module-based shard by default, it makes sense to always treat it as such,
527                 // ie bootstrap it in the same manner as the special prefix-configuration and EOS shards.
528 //                final Collection<MemberName> names = distributedConfigDatastore.getActorContext().getConfiguration()
529 //                        .getUniqueMemberNamesForAllShards();
530 //                Await.result(FutureConverters.toScala(createDistributedShard(
531 //                        new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY), names)),
532 //                        SHARD_FUTURE_TIMEOUT_DURATION);
533 //            } catch (DOMDataTreeShardingConflictException e) {
534 //                LOG.debug("{}: Default shard for {} already registered, possibly due to other node doing it faster",
535 //                        memberName, logicalDatastoreType);
536             } catch (Exception e) {
537                 LOG.error("{}: Default shard initialization for {} failed", memberName, logicalDatastoreType, e);
538                 throw new RuntimeException(e);
539             }
540         }
541     }
542
543     private static void closeProducer(final DOMDataTreeProducer producer) {
544         try {
545             producer.close();
546         } catch (final DOMDataTreeProducerException e) {
547             LOG.error("Unable to close producer", e);
548         }
549     }
550
551     @SuppressWarnings("checkstyle:IllegalCatch")
552     private static ActorRef createShardedDataTreeActor(final ActorSystem actorSystem,
553                                                        final ShardedDataTreeActorCreator creator,
554                                                        final String shardDataTreeActorId) {
555         Exception lastException = null;
556
557         for (int i = 0; i < MAX_ACTOR_CREATION_RETRIES; i++) {
558             try {
559                 return actorSystem.actorOf(creator.props(), shardDataTreeActorId);
560             } catch (final Exception e) {
561                 lastException = e;
562                 Uninterruptibles.sleepUninterruptibly(ACTOR_RETRY_DELAY, ACTOR_RETRY_TIME_UNIT);
563                 LOG.debug("Could not create actor {} because of {} -"
564                                 + " waiting for sometime before retrying (retry count = {})",
565                         shardDataTreeActorId, e.getMessage(), i);
566             }
567         }
568
569         throw new IllegalStateException("Failed to create actor for ShardedDOMDataTree", lastException);
570     }
571
572     private class DistributedShardRegistrationImpl implements DistributedShardRegistration {
573
574         private final DOMDataTreeIdentifier prefix;
575         private final ActorRef shardedDataTreeActor;
576         private final DistributedShardedDOMDataTree distributedShardedDOMDataTree;
577
578         DistributedShardRegistrationImpl(final DOMDataTreeIdentifier prefix,
579                                          final ActorRef shardedDataTreeActor,
580                                          final DistributedShardedDOMDataTree distributedShardedDOMDataTree) {
581             this.prefix = prefix;
582             this.shardedDataTreeActor = shardedDataTreeActor;
583             this.distributedShardedDOMDataTree = distributedShardedDOMDataTree;
584         }
585
586         @Override
587         public CompletionStage<Void> close() {
588             // first despawn on the local node
589             distributedShardedDOMDataTree.despawnShardFrontend(prefix);
590             // update the config so the remote nodes are updated
591             final Future<Object> ask =
592                     Patterns.ask(shardedDataTreeActor, new PrefixShardRemovalLookup(prefix), SHARD_FUTURE_TIMEOUT);
593
594             final Future<Void> closeFuture = ask.transform(
595                     new Mapper<Object, Void>() {
596                         @Override
597                         public Void apply(final Object parameter) {
598                             return null;
599                         }
600                     },
601                     new Mapper<Throwable, Throwable>() {
602                         @Override
603                         public Throwable apply(final Throwable throwable) {
604                             return throwable;
605                         }
606                     }, actorSystem.dispatcher());
607
608             return FutureConverters.toJava(closeFuture);
609         }
610     }
611
612     // TODO what about producers created by this producer?
613     // They should also be CDSProducers
614     private static final class ProxyProducer extends ForwardingObject implements CDSDataTreeProducer {
615
616         private final DOMDataTreeProducer delegate;
617         private final Collection<DOMDataTreeIdentifier> subtrees;
618         private final ActorRef shardDataTreeActor;
619         private final ActorContext actorContext;
620         @GuardedBy("shardAccessMap")
621         private final Map<DOMDataTreeIdentifier, CDSShardAccessImpl> shardAccessMap = new HashMap<>();
622
623         // We don't have to guard access to shardTable in ProxyProducer.
624         // ShardTable's entries relevant to this ProxyProducer shouldn't
625         // change during producer's lifetime.
626         private final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardTable;
627
628         ProxyProducer(final DOMDataTreeProducer delegate,
629                       final Collection<DOMDataTreeIdentifier> subtrees,
630                       final ActorRef shardDataTreeActor,
631                       final ActorContext actorContext,
632                       final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardLayout) {
633             this.delegate = Preconditions.checkNotNull(delegate);
634             this.subtrees = Preconditions.checkNotNull(subtrees);
635             this.shardDataTreeActor = Preconditions.checkNotNull(shardDataTreeActor);
636             this.actorContext = Preconditions.checkNotNull(actorContext);
637             this.shardTable = Preconditions.checkNotNull(shardLayout);
638         }
639
640         @Nonnull
641         @Override
642         public DOMDataTreeCursorAwareTransaction createTransaction(final boolean isolated) {
643             return delegate.createTransaction(isolated);
644         }
645
646         @Nonnull
647         @Override
648         @SuppressWarnings("checkstyle:hiddenField")
649         public DOMDataTreeProducer createProducer(@Nonnull final Collection<DOMDataTreeIdentifier> subtrees) {
650             // TODO we probably don't need to distribute this on the remote nodes since once we have this producer
651             // open we surely have the rights to all the subtrees.
652             return delegate.createProducer(subtrees);
653         }
654
655         @Override
656         @SuppressWarnings("checkstyle:IllegalCatch")
657         public void close() throws DOMDataTreeProducerException {
658             delegate.close();
659
660             synchronized (shardAccessMap) {
661                 shardAccessMap.values().forEach(CDSShardAccessImpl::close);
662             }
663
664             final Object o = actorContext.executeOperation(shardDataTreeActor, new ProducerRemoved(subtrees));
665             if (o instanceof DOMDataTreeProducerException) {
666                 throw (DOMDataTreeProducerException) o;
667             } else if (o instanceof Throwable) {
668                 throw new DOMDataTreeProducerException("Unable to close producer", (Throwable) o);
669             }
670         }
671
672         @Override
673         protected DOMDataTreeProducer delegate() {
674             return delegate;
675         }
676
677         @Nonnull
678         @Override
679         public CDSShardAccess getShardAccess(@Nonnull final DOMDataTreeIdentifier subtree) {
680             Preconditions.checkArgument(
681                     subtrees.stream().anyMatch(dataTreeIdentifier -> dataTreeIdentifier.contains(subtree)),
682                     "Subtree %s is not controlled by this producer %s", subtree, this);
683
684             final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
685                     shardTable.lookup(subtree);
686             Preconditions.checkState(lookup != null, "Subtree %s is not contained in any registered shard.", subtree);
687
688             final DOMDataTreeIdentifier lookupId = lookup.getValue().getPrefix();
689
690             synchronized (shardAccessMap) {
691                 if (shardAccessMap.get(lookupId) != null) {
692                     return shardAccessMap.get(lookupId);
693                 }
694
695                 // TODO Maybe we can have static factory method and return the same instance
696                 // for same subtrees. But maybe it is not needed since there can be only one
697                 // producer attached to some subtree at a time. And also how we can close ShardAccess
698                 // then
699                 final CDSShardAccessImpl shardAccess = new CDSShardAccessImpl(lookupId, actorContext);
700                 shardAccessMap.put(lookupId, shardAccess);
701                 return shardAccess;
702             }
703         }
704     }
705 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.