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

   1 /*
   2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.datastore;
   9
  10 import static com.google.common.base.Verify.verify;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.base.MoreObjects;
  14 import com.google.common.base.MoreObjects.ToStringHelper;
  15 import com.google.common.collect.Collections2;
  16 import com.google.common.collect.ImmutableList;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import org.eclipse.jdt.annotation.NonNull;
  20 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  21 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  22 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  23 import org.opendaylight.controller.cluster.datastore.persisted.FrontendClientMetadata;
  24 import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata;
  25 import org.opendaylight.controller.cluster.datastore.utils.ImmutableUnsignedLongSet;
  26 import org.opendaylight.controller.cluster.datastore.utils.MutableUnsignedLongSet;
  27 import org.opendaylight.yangtools.concepts.Identifiable;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 /**
  32  * This class is NOT thread-safe.
  33  */
  34 abstract sealed class FrontendClientMetadataBuilder implements Identifiable<ClientIdentifier> {
  35     static final class Disabled extends FrontendClientMetadataBuilder {
  36         Disabled(final String shardName, final ClientIdentifier identifier) {
  37             super(shardName, identifier);
  38         }
  39
  40         @Override
  41         FrontendClientMetadata build() {
  42             return new FrontendClientMetadata(getIdentifier(), ImmutableUnsignedLongSet.of(), ImmutableList.of());
  43         }
  44
  45         @Override
  46         void onHistoryCreated(final LocalHistoryIdentifier historyId) {
  47             // No-op
  48         }
  49
  50         @Override
  51         void onHistoryClosed(final LocalHistoryIdentifier historyId) {
  52             // No-op
  53         }
  54
  55         @Override
  56         void onHistoryPurged(final LocalHistoryIdentifier historyId) {
  57             // No-op
  58         }
  59
  60         @Override
  61         void onTransactionAborted(final TransactionIdentifier txId) {
  62             // No-op
  63         }
  64
  65         @Override
  66         void onTransactionCommitted(final TransactionIdentifier txId) {
  67             // No-op
  68         }
  69
  70         @Override
  71         void onTransactionPurged(final TransactionIdentifier txId) {
  72             // No-op
  73         }
  74
  75         @Override
  76         void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
  77             // No-op
  78         }
  79
  80         @Override
  81         LeaderFrontendState toLeaderState(final Shard shard) {
  82             return new LeaderFrontendState.Disabled(shard.persistenceId(), getIdentifier(), shard.getDataStore());
  83         }
  84     }
  85
  86     static final class Enabled extends FrontendClientMetadataBuilder {
  87         private final Map<LocalHistoryIdentifier, FrontendHistoryMetadataBuilder> currentHistories = new HashMap<>();
  88         private final MutableUnsignedLongSet purgedHistories;
  89         private final LocalHistoryIdentifier standaloneId;
  90
  91         Enabled(final String shardName, final ClientIdentifier identifier) {
  92             super(shardName, identifier);
  93
  94             purgedHistories = MutableUnsignedLongSet.of();
  95
  96             // History for stand-alone transactions is always present
  97             standaloneId = standaloneHistoryId();
  98             currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId));
  99         }
 100
 101         Enabled(final String shardName, final FrontendClientMetadata meta) {
 102             super(shardName, meta.getIdentifier());
 103
 104             purgedHistories = meta.getPurgedHistories().mutableCopy();
 105             for (FrontendHistoryMetadata h : meta.getCurrentHistories()) {
 106                 final FrontendHistoryMetadataBuilder b = new FrontendHistoryMetadataBuilder(getIdentifier(), h);
 107                 currentHistories.put(b.getIdentifier(), b);
 108             }
 109
 110             // Sanity check and recovery
 111             standaloneId = standaloneHistoryId();
 112             if (!currentHistories.containsKey(standaloneId)) {
 113                 LOG.warn("{}: Client {} recovered histories {} do not contain stand-alone history, attempting recovery",
 114                     shardName, getIdentifier(), currentHistories);
 115                 currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId));
 116             }
 117         }
 118
 119         @Override
 120         FrontendClientMetadata build() {
 121             return new FrontendClientMetadata(getIdentifier(), purgedHistories.immutableCopy(),
 122                 Collections2.transform(currentHistories.values(), FrontendHistoryMetadataBuilder::build));
 123         }
 124
 125         @Override
 126         void onHistoryCreated(final LocalHistoryIdentifier historyId) {
 127             final FrontendHistoryMetadataBuilder newMeta = new FrontendHistoryMetadataBuilder(historyId);
 128             final FrontendHistoryMetadataBuilder oldMeta = currentHistories.putIfAbsent(historyId, newMeta);
 129             if (oldMeta != null) {
 130                 // This should not be happening, warn about it
 131                 LOG.warn("{}: Reused local history {}", shardName(), historyId);
 132             } else {
 133                 LOG.debug("{}: Created local history {}", shardName(), historyId);
 134             }
 135         }
 136
 137         @Override
 138         void onHistoryClosed(final LocalHistoryIdentifier historyId) {
 139             final FrontendHistoryMetadataBuilder builder = currentHistories.get(historyId);
 140             if (builder != null) {
 141                 builder.onHistoryClosed();
 142                 LOG.debug("{}: Closed history {}", shardName(), historyId);
 143             } else {
 144                 LOG.warn("{}: Closed unknown history {}, ignoring", shardName(), historyId);
 145             }
 146         }
 147
 148         @Override
 149         void onHistoryPurged(final LocalHistoryIdentifier historyId) {
 150             final FrontendHistoryMetadataBuilder history = currentHistories.remove(historyId);
 151             final long historyBits = historyId.getHistoryId();
 152             if (history == null) {
 153                 if (!purgedHistories.contains(historyBits)) {
 154                     purgedHistories.add(historyBits);
 155                     LOG.warn("{}: Purging unknown history {}", shardName(), historyId);
 156                 } else {
 157                     LOG.warn("{}: Duplicate purge of history {}", shardName(), historyId);
 158                 }
 159             } else {
 160                 purgedHistories.add(historyBits);
 161                 LOG.debug("{}: Purged history {}", shardName(), historyId);
 162             }
 163         }
 164
 165         @Override
 166         void onTransactionAborted(final TransactionIdentifier txId) {
 167             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 168             if (history != null) {
 169                 history.onTransactionAborted(txId);
 170                 LOG.debug("{}: Aborted transaction {}", shardName(), txId);
 171             } else {
 172                 LOG.warn("{}: Unknown history for aborted transaction {}, ignoring", shardName(), txId);
 173             }
 174         }
 175
 176         @Override
 177         void onTransactionCommitted(final TransactionIdentifier txId) {
 178             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 179             if (history != null) {
 180                 history.onTransactionCommitted(txId);
 181                 LOG.debug("{}: Committed transaction {}", shardName(), txId);
 182             } else {
 183                 LOG.warn("{}: Unknown history for commited transaction {}, ignoring", shardName(), txId);
 184             }
 185         }
 186
 187         @Override
 188         void onTransactionPurged(final TransactionIdentifier txId) {
 189             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 190             if (history != null) {
 191                 history.onTransactionPurged(txId);
 192                 LOG.debug("{}: Purged transaction {}", shardName(), txId);
 193             } else {
 194                 LOG.warn("{}: Unknown history for purged transaction {}, ignoring", shardName(), txId);
 195             }
 196         }
 197
 198         @Override
 199         void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
 200             final FrontendHistoryMetadataBuilder history = getHistory(historyId);
 201             if (history != null) {
 202                 history.onTransactionsSkipped(txIds);
 203                 LOG.debug("{}: History {} skipped transactions {}", shardName(), historyId, txIds);
 204             } else {
 205                 LOG.warn("{}: Unknown history {} for skipped transactions, ignoring", shardName(), historyId);
 206             }
 207         }
 208
 209         @Override
 210         LeaderFrontendState toLeaderState(final Shard shard) {
 211             // Note: we have to make sure to *copy* all current state and not leak any views, otherwise leader/follower
 212             //       interactions would get intertwined leading to inconsistencies.
 213             final Map<LocalHistoryIdentifier, LocalFrontendHistory> histories = new HashMap<>();
 214             for (FrontendHistoryMetadataBuilder e : currentHistories.values()) {
 215                 if (e.getIdentifier().getHistoryId() != 0) {
 216                     final AbstractFrontendHistory state = e.toLeaderState(shard);
 217                     verify(state instanceof LocalFrontendHistory, "Unexpected state %s", state);
 218                     histories.put(e.getIdentifier(), (LocalFrontendHistory) state);
 219                 }
 220             }
 221
 222             final AbstractFrontendHistory singleHistory;
 223             final FrontendHistoryMetadataBuilder singleHistoryMeta = currentHistories.get(
 224                 new LocalHistoryIdentifier(getIdentifier(), 0));
 225             if (singleHistoryMeta == null) {
 226                 final ShardDataTree tree = shard.getDataStore();
 227                 singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), getIdentifier(), tree);
 228             } else {
 229                 singleHistory = singleHistoryMeta.toLeaderState(shard);
 230             }
 231
 232             return new LeaderFrontendState.Enabled(shard.persistenceId(), getIdentifier(), shard.getDataStore(),
 233                 purgedHistories.mutableCopy(), singleHistory, histories);
 234         }
 235
 236         @Override
 237         ToStringHelper addToStringAttributes(final ToStringHelper helper) {
 238             return super.addToStringAttributes(helper).add("current", currentHistories).add("purged", purgedHistories);
 239         }
 240
 241         private FrontendHistoryMetadataBuilder getHistory(final TransactionIdentifier txId) {
 242             return getHistory(txId.getHistoryId());
 243         }
 244
 245         private FrontendHistoryMetadataBuilder getHistory(final LocalHistoryIdentifier historyId) {
 246             final LocalHistoryIdentifier local;
 247             if (historyId.getHistoryId() == 0 && historyId.getCookie() != 0) {
 248                 // We are pre-creating the history for free-standing transactions with a zero cookie, hence our lookup
 249                 // needs to account for that.
 250                 LOG.debug("{}: looking up {} instead of {}", shardName(), standaloneId, historyId);
 251                 local = standaloneId;
 252             } else {
 253                 local = historyId;
 254             }
 255
 256             return currentHistories.get(local);
 257         }
 258
 259         private LocalHistoryIdentifier standaloneHistoryId() {
 260             return new LocalHistoryIdentifier(getIdentifier(), 0);
 261         }
 262     }
 263
 264     private static final Logger LOG = LoggerFactory.getLogger(FrontendClientMetadataBuilder.class);
 265
 266     private final @NonNull ClientIdentifier identifier;
 267     private final @NonNull String shardName;
 268
 269     FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier identifier) {
 270         this.shardName = requireNonNull(shardName);
 271         this.identifier = requireNonNull(identifier);
 272     }
 273
 274     static FrontendClientMetadataBuilder of(final String shardName, final FrontendClientMetadata meta) {
 275         // Completely empty histories imply disabled state, as otherwise we'd have a record of the single history --
 276         // either purged or active
 277         return meta.getCurrentHistories().isEmpty() && meta.getPurgedHistories().isEmpty()
 278             ? new Disabled(shardName, meta.getIdentifier()) : new Enabled(shardName, meta);
 279     }
 280
 281     @Override
 282     public final ClientIdentifier getIdentifier() {
 283         return identifier;
 284     }
 285
 286     final String shardName() {
 287         return shardName;
 288     }
 289
 290     abstract FrontendClientMetadata build();
 291
 292     abstract void onHistoryCreated(LocalHistoryIdentifier historyId);
 293
 294     abstract void onHistoryClosed(LocalHistoryIdentifier historyId);
 295
 296     abstract void onHistoryPurged(LocalHistoryIdentifier historyId);
 297
 298     abstract void onTransactionAborted(TransactionIdentifier txId);
 299
 300     abstract void onTransactionCommitted(TransactionIdentifier txId);
 301
 302     abstract void onTransactionPurged(TransactionIdentifier txId);
 303
 304     abstract void onTransactionsSkipped(LocalHistoryIdentifier historyId, ImmutableUnsignedLongSet txIds);
 305
 306     /**
 307      * Transform frontend metadata for a particular client into its {@link LeaderFrontendState} counterpart.
 308      *
 309      * @param shard parent shard
 310      * @return Leader frontend state
 311      */
 312     abstract @NonNull LeaderFrontendState toLeaderState(@NonNull Shard shard);
 313
 314     @Override
 315     public final String toString() {
 316         return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
 317     }
 318
 319     ToStringHelper addToStringAttributes(final ToStringHelper helper) {
 320         return helper.add("identifier", identifier);
 321     }
 322 }