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