opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/RootDataTreeChangeListenerProxy.java

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, s.r.o. 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 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static com.google.common.base.Verify.verify;
  11 import static com.google.common.base.Verify.verifyNotNull;
  12 import static java.util.Objects.requireNonNull;
  13
  14 import akka.actor.ActorRef;
  15 import akka.actor.ActorSelection;
  16 import akka.actor.PoisonPill;
  17 import akka.dispatch.OnComplete;
  18 import com.google.common.collect.Maps;
  19 import java.util.ArrayList;
  20 import java.util.HashMap;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Map.Entry;
  24 import java.util.Set;
  25 import org.checkerframework.checker.lock.qual.GuardedBy;
  26 import org.checkerframework.checker.lock.qual.Holding;
  27 import org.eclipse.jdt.annotation.NonNull;
  28 import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration;
  29 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
  30 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
  31 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  32 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
  33 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
  34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 final class RootDataTreeChangeListenerProxy<L extends DOMDataTreeChangeListener> extends AbstractObjectRegistration<L> {
  39     private abstract static class State {
  40
  41     }
  42
  43     private static final class ResolveShards extends State {
  44         final Map<String, Object> localShards = new HashMap<>();
  45         final int shardCount;
  46
  47         ResolveShards(final int shardCount) {
  48             this.shardCount = shardCount;
  49         }
  50     }
  51
  52     private static final class Subscribed extends State {
  53         final List<ActorSelection> subscriptions;
  54         final ActorRef dtclActor;
  55
  56         Subscribed(final ActorRef dtclActor, final int shardCount) {
  57             this.dtclActor = requireNonNull(dtclActor);
  58             subscriptions = new ArrayList<>(shardCount);
  59         }
  60     }
  61
  62     private static final class Terminated extends State {
  63
  64     }
  65
  66     private static final Logger LOG = LoggerFactory.getLogger(RootDataTreeChangeListenerProxy.class);
  67
  68     private final ActorUtils actorUtils;
  69
  70     @GuardedBy("this")
  71     private State state;
  72
  73     RootDataTreeChangeListenerProxy(final ActorUtils actorUtils, final @NonNull L listener,
  74             final Set<String> shardNames) {
  75         super(listener);
  76         this.actorUtils = requireNonNull(actorUtils);
  77         state = new ResolveShards(shardNames.size());
  78
  79         for (String shardName : shardNames) {
  80             actorUtils.findLocalShardAsync(shardName).onComplete(new OnComplete<ActorRef>() {
  81                 @Override
  82                 public void onComplete(final Throwable failure, final ActorRef success) {
  83                     onFindLocalShardComplete(shardName, failure, success);
  84                 }
  85             }, actorUtils.getClientDispatcher());
  86         }
  87     }
  88
  89     @Override
  90     protected synchronized void removeRegistration() {
  91         if (state instanceof Terminated) {
  92             // Trivial case: we have already terminated on a failure, so this is a no-op
  93         } else if (state instanceof ResolveShards) {
  94             // Simple case: just mark the fact we were closed, terminating when resolution finishes
  95             state = new Terminated();
  96         } else if (state instanceof Subscribed subscribed) {
  97             terminate(subscribed);
  98         } else {
  99             throw new IllegalStateException("Unhandled close in state " + state);
 100         }
 101     }
 102
 103     private synchronized void onFindLocalShardComplete(final String shardName, final Throwable failure,
 104             final ActorRef shard) {
 105         if (state instanceof ResolveShards resolveShards) {
 106             localShardsResolved(resolveShards, shardName, failure, shard);
 107         } else {
 108             LOG.debug("{}: lookup for shard {} turned into a noop on state {}", logContext(), shardName, state);
 109         }
 110     }
 111
 112     @Holding("this")
 113     private void localShardsResolved(final ResolveShards current, final String shardName, final Throwable failure,
 114             final ActorRef shard) {
 115         final Object result = failure != null ? failure : verifyNotNull(shard);
 116         LOG.debug("{}: lookup for shard {} resulted in {}", logContext(), shardName, result);
 117         current.localShards.put(shardName, result);
 118
 119         if (current.localShards.size() == current.shardCount) {
 120             // We have all the responses we need
 121             if (current.localShards.values().stream().anyMatch(Throwable.class::isInstance)) {
 122                 reportFailure(current.localShards);
 123             } else {
 124                 subscribeToShards(current.localShards);
 125             }
 126         }
 127     }
 128
 129     @Holding("this")
 130     private void reportFailure(final Map<String, Object> localShards) {
 131         for (Entry<String, Object> entry : Maps.filterValues(localShards, Throwable.class::isInstance).entrySet()) {
 132             final Throwable cause = (Throwable) entry.getValue();
 133             LOG.error("{}: Failed to find local shard {}, cannot register {} at root", logContext(), entry.getKey(),
 134                 getInstance(), cause);
 135         }
 136         state = new Terminated();
 137     }
 138
 139     @Holding("this")
 140     private void subscribeToShards(final Map<String, Object> localShards) {
 141         // Safety check before we start doing anything
 142         for (Entry<String, Object> entry : localShards.entrySet()) {
 143             final Object obj = entry.getValue();
 144             verify(obj instanceof ActorRef, "Unhandled response %s for shard %s", obj, entry.getKey());
 145         }
 146
 147         // Instantiate the DTCL actor and update state
 148         final ActorRef dtclActor = actorUtils.getActorSystem().actorOf(
 149             RootDataTreeChangeListenerActor.props(getInstance(), localShards.size())
 150               .withDispatcher(actorUtils.getNotificationDispatcherPath()));
 151         state = new Subscribed(dtclActor, localShards.size());
 152
 153         // Subscribe to all shards
 154         final RegisterDataTreeChangeListener regMessage = new RegisterDataTreeChangeListener(
 155             YangInstanceIdentifier.of(), dtclActor, true);
 156         for (Entry<String, Object> entry : localShards.entrySet()) {
 157             // Do not retain references to localShards
 158             final String shardName = entry.getKey();
 159             final ActorRef shard = (ActorRef) entry.getValue();
 160
 161             actorUtils.executeOperationAsync(shard, regMessage,
 162                 actorUtils.getDatastoreContext().getShardInitializationTimeout()).onComplete(new OnComplete<>() {
 163                     @Override
 164                     public void onComplete(final Throwable failure, final Object result) {
 165                         onShardSubscribed(shardName, failure, result);
 166                     }
 167                 }, actorUtils.getClientDispatcher());
 168         }
 169     }
 170
 171     private synchronized void onShardSubscribed(final String shardName, final Throwable failure, final Object result) {
 172         if (state instanceof Subscribed current) {
 173             if (failure != null) {
 174                 LOG.error("{}: Shard {} failed to subscribe, terminating listener {}", logContext(),
 175                     shardName,getInstance(), failure);
 176                 terminate(current);
 177             } else {
 178                 onSuccessfulSubscription(current, shardName, (RegisterDataTreeNotificationListenerReply) result);
 179             }
 180         } else {
 181             terminateSubscription(shardName, failure, result);
 182         }
 183     }
 184
 185     @Holding("this")
 186     private void onSuccessfulSubscription(final Subscribed current, final String shardName,
 187             final RegisterDataTreeNotificationListenerReply reply) {
 188         final ActorSelection regActor = actorUtils.actorSelection(reply.getListenerRegistrationPath());
 189         LOG.debug("{}: Shard {} subscribed at {}", logContext(), shardName, regActor);
 190         current.subscriptions.add(regActor);
 191     }
 192
 193     @Holding("this")
 194     private void terminate(final Subscribed current) {
 195         // Terminate the listener
 196         current.dtclActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
 197         // Terminate all subscriptions
 198         for (ActorSelection regActor : current.subscriptions) {
 199             regActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(), ActorRef.noSender());
 200         }
 201         state = new Terminated();
 202     }
 203
 204     // This method should not modify internal state
 205     private void terminateSubscription(final String shardName, final Throwable failure, final Object result) {
 206         if (failure == null) {
 207             final ActorSelection regActor = actorUtils.actorSelection(
 208                 ((RegisterDataTreeNotificationListenerReply) result).getListenerRegistrationPath());
 209             LOG.debug("{}: Shard {} registered late, terminating subscription at {}", logContext(), shardName,
 210                 regActor);
 211             regActor.tell(CloseDataTreeNotificationListenerRegistration.getInstance(), ActorRef.noSender());
 212         } else {
 213             LOG.debug("{}: Shard {} reported late failure", logContext(), shardName, failure);
 214         }
 215     }
 216
 217     private String logContext() {
 218         return actorUtils.getDatastoreContext().getLogicalStoreType().toString();
 219     }
 220 }