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