2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.databroker.actors.dds;
10 import static com.google.common.base.Verify.verifyNotNull;
12 import akka.dispatch.ExecutionContexts;
13 import akka.dispatch.OnComplete;
14 import akka.pattern.Patterns;
15 import akka.util.Timeout;
16 import com.google.common.collect.ImmutableBiMap;
17 import java.util.concurrent.CompletionStage;
18 import java.util.concurrent.ConcurrentHashMap;
19 import java.util.concurrent.ConcurrentMap;
20 import java.util.concurrent.TimeUnit;
21 import java.util.stream.Stream;
22 import org.checkerframework.checker.lock.qual.GuardedBy;
23 import org.eclipse.jdt.annotation.NonNull;
24 import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
25 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
26 import org.opendaylight.controller.cluster.datastore.shardmanager.RegisterForShardAvailabilityChanges;
27 import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
28 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
29 import org.opendaylight.yangtools.concepts.Registration;
30 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
33 import scala.concurrent.Future;
36 * {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
37 * shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
38 * shard leader is resolved via {@link ActorUtils}. The product of resolution is {@link ShardBackendInfo}.
41 * This class is thread-safe.
43 * @author Robert Varga
45 final class ModuleShardBackendResolver extends AbstractShardBackendResolver {
46 private static final Logger LOG = LoggerFactory.getLogger(ModuleShardBackendResolver.class);
48 private final ConcurrentMap<Long, ShardState> backends = new ConcurrentHashMap<>();
50 private final Future<Registration> shardAvailabilityChangesRegFuture;
53 private long nextShard = 1;
55 private volatile ImmutableBiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
57 // FIXME: we really need just ActorContext.findPrimaryShardAsync()
58 ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils) {
59 super(clientId, actorUtils);
61 shardAvailabilityChangesRegFuture = Patterns.ask(actorUtils.getShardManager(),
62 new RegisterForShardAvailabilityChanges(this::onShardAvailabilityChange),
63 Timeout.apply(60, TimeUnit.MINUTES))
64 .map(reply -> (Registration)reply, ExecutionContexts.global());
66 shardAvailabilityChangesRegFuture.onComplete(new OnComplete<Registration>() {
68 public void onComplete(final Throwable failure, final Registration reply) {
69 if (failure != null) {
70 LOG.error("RegisterForShardAvailabilityChanges failed", failure);
73 }, ExecutionContexts.global());
76 private void onShardAvailabilityChange(final String shardName) {
77 LOG.debug("onShardAvailabilityChange for {}", shardName);
79 Long cookie = shards.get(shardName);
81 LOG.debug("No shard cookie found for {}", shardName);
85 notifyStaleBackendInfoCallbacks(cookie);
88 Long resolveShardForPath(final YangInstanceIdentifier path) {
89 return resolveCookie(actorUtils().getShardStrategyFactory().getStrategy(path).findShard(path));
92 Stream<Long> resolveAllShards() {
93 return actorUtils().getConfiguration().getAllShardNames().stream()
95 .map(this::resolveCookie);
98 private @NonNull Long resolveCookie(final String shardName) {
99 final Long cookie = shards.get(shardName);
100 return cookie != null ? cookie : populateShard(shardName);
103 private synchronized @NonNull Long populateShard(final String shardName) {
104 Long cookie = shards.get(shardName);
105 if (cookie == null) {
106 cookie = nextShard++;
107 shards = ImmutableBiMap.<String, Long>builder().putAll(shards).put(shardName, cookie).build();
113 public CompletionStage<ShardBackendInfo> getBackendInfo(final Long cookie) {
115 * We cannot perform a simple computeIfAbsent() here because we need to control sequencing of when the state
116 * is inserted into the map and retired from it (based on the stage result).
118 * We do not want to hook another stage one processing completes and hooking a removal on failure from a compute
119 * method runs the inherent risk of stage completing before the insertion does (i.e. we have a removal of
120 * non-existent element.
122 final ShardState existing = backends.get(cookie);
123 if (existing != null) {
124 return existing.getStage();
127 final String shardName = shards.inverse().get(cookie);
128 if (shardName == null) {
129 LOG.warn("Failing request for non-existent cookie {}", cookie);
130 throw new IllegalArgumentException("Cookie " + cookie + " does not have a shard assigned");
133 LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
134 final ShardState toInsert = resolveBackendInfo(shardName, cookie);
136 final ShardState raced = backends.putIfAbsent(cookie, toInsert);
138 // We have had a concurrent insertion, return that
139 LOG.debug("Race during insertion of state for cookie {} shard {}", cookie, shardName);
140 return raced.getStage();
143 // We have succeeded in populating the map, now we need to take care of pruning the entry if it fails to
145 final CompletionStage<ShardBackendInfo> stage = toInsert.getStage();
146 stage.whenComplete((info, failure) -> {
147 if (failure != null) {
148 LOG.debug("Resolution of cookie {} shard {} failed, removing state", cookie, shardName, failure);
149 backends.remove(cookie, toInsert);
151 // Remove cache state in case someone else forgot to invalidate it
152 flushCache(shardName);
160 public CompletionStage<ShardBackendInfo> refreshBackendInfo(final Long cookie,
161 final ShardBackendInfo staleInfo) {
162 final ShardState existing = backends.get(cookie);
163 if (existing != null) {
164 if (!staleInfo.equals(existing.getResult())) {
165 return existing.getStage();
168 LOG.debug("Invalidating backend information {}", staleInfo);
169 flushCache(staleInfo.getName());
171 LOG.trace("Invalidated cache {}", staleInfo);
172 backends.remove(cookie, existing);
175 return getBackendInfo(cookie);
179 public void close() {
180 shardAvailabilityChangesRegFuture.onComplete(new OnComplete<Registration>() {
182 public void onComplete(final Throwable failure, final Registration reply) {
185 }, ExecutionContexts.global());
189 public String resolveCookieName(final Long cookie) {
190 return verifyNotNull(shards.inverse().get(cookie), "Unexpected null cookie: %s", cookie);