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

©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.