opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/LeaderFrontendState.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 java.util.Objects.requireNonNull;
  11
  12 import com.google.common.base.MoreObjects;
  13 import com.google.common.base.MoreObjects.ToStringHelper;
  14 import java.util.HashMap;
  15 import java.util.Iterator;
  16 import java.util.Map;
  17 import org.eclipse.jdt.annotation.Nullable;
  18 import org.opendaylight.controller.cluster.access.commands.CreateLocalHistoryRequest;
  19 import org.opendaylight.controller.cluster.access.commands.DeadHistoryException;
  20 import org.opendaylight.controller.cluster.access.commands.DestroyLocalHistoryRequest;
  21 import org.opendaylight.controller.cluster.access.commands.LocalHistoryRequest;
  22 import org.opendaylight.controller.cluster.access.commands.LocalHistorySuccess;
  23 import org.opendaylight.controller.cluster.access.commands.OutOfSequenceEnvelopeException;
  24 import org.opendaylight.controller.cluster.access.commands.PurgeLocalHistoryRequest;
  25 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
  26 import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
  27 import org.opendaylight.controller.cluster.access.commands.UnknownHistoryException;
  28 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  29 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  30 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  31 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  32 import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException;
  33 import org.opendaylight.controller.cluster.datastore.ShardDataTreeCohort.State;
  34 import org.opendaylight.controller.cluster.datastore.utils.UnsignedLongRangeSet;
  35 import org.opendaylight.yangtools.concepts.Identifiable;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 /**
  40  * Frontend state as observed by the shard leader. This class is responsible for tracking generations and sequencing
  41  * in the frontend/backend conversation. This class is NOT thread-safe.
  42  *
  43  * @author Robert Varga
  44  */
  45 abstract class LeaderFrontendState implements Identifiable<ClientIdentifier> {
  46     static final class Disabled extends LeaderFrontendState {
  47         Disabled(final String persistenceId, final ClientIdentifier clientId, final ShardDataTree tree) {
  48             super(persistenceId, clientId, tree);
  49         }
  50
  51         @Override
  52         LocalHistorySuccess handleLocalHistoryRequest(final LocalHistoryRequest<?> request,
  53                 final RequestEnvelope envelope, final long now) throws RequestException {
  54             throw new UnsupportedRequestException(request);
  55         }
  56
  57         @Override
  58         TransactionSuccess<?> handleTransactionRequest(final TransactionRequest<?> request,
  59                 final RequestEnvelope envelope, final long now) throws RequestException {
  60             throw new UnsupportedRequestException(request);
  61         }
  62     }
  63
  64     static final class Enabled extends LeaderFrontendState {
  65         // Histories which have not been purged
  66         private final Map<LocalHistoryIdentifier, LocalFrontendHistory> localHistories;
  67
  68         // RangeSet performs automatic merging, hence we keep minimal state tracking information
  69         private final UnsignedLongRangeSet purgedHistories;
  70
  71         // Used for all standalone transactions
  72         private final AbstractFrontendHistory standaloneHistory;
  73
  74         private long expectedTxSequence;
  75         private Long lastSeenHistory = null;
  76
  77         Enabled(final String persistenceId, final ClientIdentifier clientId, final ShardDataTree tree) {
  78             this(persistenceId, clientId, tree, UnsignedLongRangeSet.create(),
  79                 StandaloneFrontendHistory.create(persistenceId, clientId, tree), new HashMap<>());
  80         }
  81
  82         Enabled(final String persistenceId, final ClientIdentifier clientId, final ShardDataTree tree,
  83                 final UnsignedLongRangeSet purgedHistories, final AbstractFrontendHistory standaloneHistory,
  84                 final Map<LocalHistoryIdentifier, LocalFrontendHistory> localHistories) {
  85             super(persistenceId, clientId, tree);
  86             this.purgedHistories = requireNonNull(purgedHistories);
  87             this.standaloneHistory = requireNonNull(standaloneHistory);
  88             this.localHistories = requireNonNull(localHistories);
  89         }
  90
  91         @Override
  92         @Nullable LocalHistorySuccess handleLocalHistoryRequest(final LocalHistoryRequest<?> request,
  93                 final RequestEnvelope envelope, final long now) throws RequestException {
  94             checkRequestSequence(envelope);
  95
  96             try {
  97                 if (request instanceof CreateLocalHistoryRequest) {
  98                     return handleCreateHistory((CreateLocalHistoryRequest) request, envelope, now);
  99                 } else if (request instanceof DestroyLocalHistoryRequest) {
 100                     return handleDestroyHistory((DestroyLocalHistoryRequest) request, envelope, now);
 101                 } else if (request instanceof PurgeLocalHistoryRequest) {
 102                     return handlePurgeHistory((PurgeLocalHistoryRequest)request, envelope, now);
 103                 } else {
 104                     LOG.warn("{}: rejecting unsupported request {}", persistenceId(), request);
 105                     throw new UnsupportedRequestException(request);
 106                 }
 107             } finally {
 108                 expectNextRequest();
 109             }
 110         }
 111
 112         @Override
 113         @Nullable TransactionSuccess<?> handleTransactionRequest(final TransactionRequest<?> request,
 114                 final RequestEnvelope envelope, final long now) throws RequestException {
 115             checkRequestSequence(envelope);
 116
 117             try {
 118                 final LocalHistoryIdentifier lhId = request.getTarget().getHistoryId();
 119                 final AbstractFrontendHistory history;
 120
 121                 if (lhId.getHistoryId() != 0) {
 122                     history = localHistories.get(lhId);
 123                     if (history == null) {
 124                         if (purgedHistories.contains(lhId.getHistoryId())) {
 125                             LOG.warn("{}: rejecting request {} to purged history", persistenceId(), request);
 126                             throw new DeadHistoryException(purgedHistories.toImmutable());
 127                         }
 128
 129                         LOG.warn("{}: rejecting unknown history request {}", persistenceId(), request);
 130                         throw new UnknownHistoryException(lastSeenHistory);
 131                     }
 132                 } else {
 133                     history = standaloneHistory;
 134                 }
 135
 136                 return history.handleTransactionRequest(request, envelope, now);
 137             } finally {
 138                 expectNextRequest();
 139             }
 140         }
 141
 142         @Override
 143         void reconnect() {
 144             expectedTxSequence = 0;
 145             super.reconnect();
 146         }
 147
 148         @Override
 149         void retire() {
 150             super.retire();
 151
 152             // Clear out all transaction chains
 153             localHistories.values().forEach(AbstractFrontendHistory::retire);
 154             localHistories.clear();
 155             standaloneHistory.retire();
 156         }
 157
 158         @Override
 159         ToStringHelper addToStringAttributes(final ToStringHelper helper) {
 160             return super.addToStringAttributes(helper).add("purgedHistories", purgedHistories);
 161         }
 162
 163         private LocalHistorySuccess handleCreateHistory(final CreateLocalHistoryRequest request,
 164                 final RequestEnvelope envelope, final long now) throws RequestException {
 165             final LocalHistoryIdentifier historyId = request.getTarget();
 166             final AbstractFrontendHistory existing = localHistories.get(historyId);
 167             if (existing != null) {
 168                 // History already exists: report success
 169                 LOG.debug("{}: history {} already exists", persistenceId(), historyId);
 170                 return new LocalHistorySuccess(historyId, request.getSequence());
 171             }
 172
 173             // We have not found the history. Before we create it we need to check history ID sequencing so that we do
 174             // not end up resurrecting a purged history.
 175             if (purgedHistories.contains(historyId.getHistoryId())) {
 176                 LOG.debug("{}: rejecting purged request {}", persistenceId(), request);
 177                 throw new DeadHistoryException(purgedHistories.toImmutable());
 178             }
 179
 180             // Update last history we have seen
 181             if (lastSeenHistory == null || Long.compareUnsigned(lastSeenHistory, historyId.getHistoryId()) < 0) {
 182                 lastSeenHistory = historyId.getHistoryId();
 183             }
 184
 185             // We have to send the response only after persistence has completed
 186             final ShardDataTreeTransactionChain chain = tree().ensureTransactionChain(historyId, () -> {
 187                 LOG.debug("{}: persisted history {}", persistenceId(), historyId);
 188                 envelope.sendSuccess(new LocalHistorySuccess(historyId, request.getSequence()),
 189                     tree().readTime() - now);
 190             });
 191
 192             localHistories.put(historyId, LocalFrontendHistory.create(persistenceId(), tree(), chain));
 193             LOG.debug("{}: created history {}", persistenceId(), historyId);
 194             return null;
 195         }
 196
 197         private LocalHistorySuccess handleDestroyHistory(final DestroyLocalHistoryRequest request,
 198                 final RequestEnvelope envelope, final long now) {
 199             final LocalHistoryIdentifier id = request.getTarget();
 200             final LocalFrontendHistory existing = localHistories.get(id);
 201             if (existing == null) {
 202                 // History does not exist: report success
 203                 LOG.debug("{}: history {} does not exist, nothing to destroy", persistenceId(), id);
 204                 return new LocalHistorySuccess(id, request.getSequence());
 205             }
 206
 207             existing.destroy(request.getSequence(), envelope, now);
 208             return null;
 209         }
 210
 211         private LocalHistorySuccess handlePurgeHistory(final PurgeLocalHistoryRequest request,
 212                 final RequestEnvelope envelope, final long now) {
 213             final LocalHistoryIdentifier id = request.getTarget();
 214             final LocalFrontendHistory existing = localHistories.remove(id);
 215             if (existing == null) {
 216                 LOG.debug("{}: history {} has already been purged", persistenceId(), id);
 217                 return new LocalHistorySuccess(id, request.getSequence());
 218             }
 219
 220             LOG.debug("{}: purging history {}", persistenceId(), id);
 221             purgedHistories.add(id.getHistoryId());
 222             existing.purge(request.getSequence(), envelope, now);
 223             return null;
 224         }
 225
 226         private void checkRequestSequence(final RequestEnvelope envelope) throws OutOfSequenceEnvelopeException {
 227             if (expectedTxSequence != envelope.getTxSequence()) {
 228                 throw new OutOfSequenceEnvelopeException(expectedTxSequence);
 229             }
 230         }
 231
 232         private void expectNextRequest() {
 233             expectedTxSequence++;
 234         }
 235     }
 236
 237     private static final Logger LOG = LoggerFactory.getLogger(LeaderFrontendState.class);
 238
 239     private final ShardDataTree tree;
 240     private final ClientIdentifier clientId;
 241     private final String persistenceId;
 242
 243     private long lastConnectTicks;
 244     private long lastSeenTicks;
 245
 246     // TODO: explicit failover notification
 247     //       Record the ActorRef for the originating actor and when we switch to being a leader send a notification
 248     //       to the frontend client -- that way it can immediately start sending requests
 249
 250     // TODO: add statistics:
 251     // - number of requests processed
 252     // - number of histories processed
 253     // - per-RequestException throw counters
 254
 255     LeaderFrontendState(final String persistenceId, final ClientIdentifier clientId, final ShardDataTree tree) {
 256         this.persistenceId = requireNonNull(persistenceId);
 257         this.clientId = requireNonNull(clientId);
 258         this.tree = requireNonNull(tree);
 259         this.lastSeenTicks = tree.readTime();
 260     }
 261
 262     @Override
 263     public final ClientIdentifier getIdentifier() {
 264         return clientId;
 265     }
 266
 267     final String persistenceId() {
 268         return persistenceId;
 269     }
 270
 271     final long getLastConnectTicks() {
 272         return lastConnectTicks;
 273     }
 274
 275     final long getLastSeenTicks() {
 276         return lastSeenTicks;
 277     }
 278
 279     final ShardDataTree tree() {
 280         return tree;
 281     }
 282
 283     final void touch() {
 284         this.lastSeenTicks = tree.readTime();
 285     }
 286
 287     abstract @Nullable LocalHistorySuccess handleLocalHistoryRequest(LocalHistoryRequest<?> request,
 288             RequestEnvelope envelope, long now) throws RequestException;
 289
 290     abstract @Nullable TransactionSuccess<?> handleTransactionRequest(TransactionRequest<?> request,
 291             RequestEnvelope envelope, long now) throws RequestException;
 292
 293     void reconnect() {
 294         lastConnectTicks = tree.readTime();
 295     }
 296
 297     void retire() {
 298         // Hunt down any transactions associated with this frontend
 299         final Iterator<SimpleShardDataTreeCohort> it = tree.cohortIterator();
 300         while (it.hasNext()) {
 301             final SimpleShardDataTreeCohort cohort = it.next();
 302             if (clientId.equals(cohort.getIdentifier().getHistoryId().getClientId())) {
 303                 if (cohort.getState() != State.COMMIT_PENDING) {
 304                     LOG.debug("{}: Retiring transaction {}", persistenceId, cohort.getIdentifier());
 305                     it.remove();
 306                 } else {
 307                     LOG.debug("{}: Transaction {} already committing, not retiring it", persistenceId,
 308                         cohort.getIdentifier());
 309                 }
 310             }
 311         }
 312     }
 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("clientId", clientId).add("nanosAgo", tree.readTime() - lastSeenTicks);
 321     }
 322 }