X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=d02bd32099adfb0b2d3accd29edd404a7b854cf5;hb=refs%2Fchanges%2F09%2F83009%2F6;hp=c69b839bea8595acc6f63a2ff80b0e1b7cdf80f3;hpb=1cb411af2b8effb0b40b6a97d686d3dc83857dfd;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java index c69b839bea..d02bd32099 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java @@ -5,9 +5,10 @@ * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - package org.opendaylight.controller.cluster.datastore; +import static com.google.common.base.Verify.verify; + import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.Cancellable; @@ -18,7 +19,6 @@ import akka.actor.Status.Failure; import akka.serialization.JavaSerializer; import akka.serialization.Serialization; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Ticker; import com.google.common.base.Verify; @@ -29,10 +29,13 @@ import java.io.IOException; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.Map; +import java.util.Optional; +import java.util.OptionalLong; import java.util.concurrent.TimeUnit; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; +import org.eclipse.jdt.annotation.NonNull; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.controller.cluster.access.ABIVersion; import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest; import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess; @@ -76,14 +79,13 @@ import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree; import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal; import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState; import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved; -import org.opendaylight.controller.cluster.datastore.messages.PersistAbortTransactionPayload; import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; -import org.opendaylight.controller.cluster.datastore.persisted.AbortTransactionPayload; import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot.ShardSnapshot; +import org.opendaylight.controller.cluster.datastore.persisted.DisableTrackingPayload; import org.opendaylight.controller.cluster.messaging.MessageAssembler; import org.opendaylight.controller.cluster.messaging.MessageSlicer; import org.opendaylight.controller.cluster.messaging.SliceOptions; @@ -107,7 +109,6 @@ import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailed import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.opendaylight.yangtools.yang.model.api.SchemaContextProvider; -import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; /** @@ -363,9 +364,6 @@ public class Shard extends RaftActor { } else if (message instanceof DataTreeCohortActorRegistry.CohortRegistryCommand) { store.processCohortRegistryCommand(getSender(), (DataTreeCohortActorRegistry.CohortRegistryCommand) message); - } else if (message instanceof PersistAbortTransactionPayload) { - final TransactionIdentifier txId = ((PersistAbortTransactionPayload) message).getTransactionId(); - persistPayload(txId, AbortTransactionPayload.create(txId), true); } else if (message instanceof MakeLeaderLocal) { onMakeLeaderLocal(); } else if (RESUME_NEXT_PENDING_TRANSACTION.equals(message)) { @@ -416,22 +414,50 @@ public class Shard extends RaftActor { requestMessageAssembler.checkExpiredAssembledMessageState(); } - private Optional updateAccess(final SimpleShardDataTreeCohort cohort) { + private OptionalLong updateAccess(final SimpleShardDataTreeCohort cohort) { final FrontendIdentifier frontend = cohort.getIdentifier().getHistoryId().getClientId().getFrontendId(); final LeaderFrontendState state = knownFrontends.get(frontend); if (state == null) { // Not tell-based protocol, do nothing - return Optional.absent(); + return OptionalLong.empty(); } if (isIsolatedLeader()) { // We are isolated and no new request can come through until we emerge from it. We are still updating // liveness of frontend when we see it attempting to communicate. Use the last access timer. - return Optional.of(state.getLastSeenTicks()); + return OptionalLong.of(state.getLastSeenTicks()); } // If this frontend has freshly connected, give it some time to catch up before killing its transactions. - return Optional.of(state.getLastConnectTicks()); + return OptionalLong.of(state.getLastConnectTicks()); + } + + private void disableTracking(final DisableTrackingPayload payload) { + final ClientIdentifier clientId = payload.getIdentifier(); + LOG.debug("{}: disabling tracking of {}", persistenceId(), clientId); + frontendMetadata.disableTracking(clientId); + + if (isLeader()) { + final FrontendIdentifier frontendId = clientId.getFrontendId(); + final LeaderFrontendState frontend = knownFrontends.get(frontendId); + if (frontend != null) { + if (clientId.equals(frontend.getIdentifier())) { + if (!(frontend instanceof LeaderFrontendState.Disabled)) { + verify(knownFrontends.replace(frontendId, frontend, + new LeaderFrontendState.Disabled(persistenceId(), clientId, store))); + LOG.debug("{}: leader state for {} disabled", persistenceId(), clientId); + } else { + LOG.debug("{}: leader state {} is already disabled", persistenceId(), frontend); + } + } else { + LOG.debug("{}: leader state {} does not match {}", persistenceId(), frontend, clientId); + } + } else { + LOG.debug("{}: leader state for {} not found", persistenceId(), clientId); + knownFrontends.put(frontendId, new LeaderFrontendState.Disabled(persistenceId(), clientId, + getDataStore())); + } + } } private void onMakeLeaderLocal() { @@ -461,8 +487,7 @@ public class Shard extends RaftActor { } // Acquire our frontend tracking handle and verify generation matches - @Nullable - private LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException { + private @Nullable LeaderFrontendState findFrontend(final ClientIdentifier clientId) throws RequestException { final LeaderFrontendState existing = knownFrontends.get(clientId.getFrontendId()); if (existing != null) { final int cmp = Long.compareUnsigned(existing.getIdentifier().getGeneration(), clientId.getGeneration()); @@ -496,8 +521,7 @@ public class Shard extends RaftActor { throw new OutOfSequenceEnvelopeException(0); } - @Nonnull - private static ABIVersion selectVersion(final ConnectClientRequest message) { + private static @NonNull ABIVersion selectVersion(final ConnectClientRequest message) { final Range clientRange = Range.closed(message.getMinVersion(), message.getMaxVersion()); for (ABIVersion v : SUPPORTED_ABIVERSIONS) { if (clientRange.contains(v)) { @@ -529,7 +553,7 @@ public class Shard extends RaftActor { final ABIVersion selectedVersion = selectVersion(message); final LeaderFrontendState frontend; if (existing == null) { - frontend = new LeaderFrontendState(persistenceId(), clientId, store); + frontend = new LeaderFrontendState.Enabled(persistenceId(), clientId, store); knownFrontends.put(clientId.getFrontendId(), frontend); LOG.debug("{}: created state {} for client {}", persistenceId(), frontend, clientId); } else { @@ -545,8 +569,7 @@ public class Shard extends RaftActor { } } - @Nullable - private RequestSuccess handleRequest(final RequestEnvelope envelope, final long now) + private @Nullable RequestSuccess handleRequest(final RequestEnvelope envelope, final long now) throws RequestException { // We are not the leader, hence we want to fail-fast. if (!isLeader() || paused || !isLeaderActive()) { @@ -621,13 +644,14 @@ public class Shard extends RaftActor { } private void handleCommitTransaction(final CommitTransaction commit) { + final TransactionIdentifier txId = commit.getTransactionId(); if (isLeader()) { - commitCoordinator.handleCommit(commit.getTransactionId(), getSender(), this); + askProtocolEncountered(txId); + commitCoordinator.handleCommit(txId, getSender(), this); } else { ActorSelection leader = getLeader(); if (leader == null) { - messageRetrySupport.addMessageToRetry(commit, getSender(), - "Could not commit transaction " + commit.getTransactionId()); + messageRetrySupport.addMessageToRetry(commit, getSender(), "Could not commit transaction " + txId); } else { LOG.debug("{}: Forwarding CommitTransaction to leader {}", persistenceId(), leader); leader.forward(commit, getContext()); @@ -636,15 +660,17 @@ public class Shard extends RaftActor { } private void handleCanCommitTransaction(final CanCommitTransaction canCommit) { - LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionId()); + final TransactionIdentifier txId = canCommit.getTransactionId(); + LOG.debug("{}: Can committing transaction {}", persistenceId(), txId); if (isLeader()) { - commitCoordinator.handleCanCommit(canCommit.getTransactionId(), getSender(), this); + askProtocolEncountered(txId); + commitCoordinator.handleCanCommit(txId, getSender(), this); } else { ActorSelection leader = getLeader(); if (leader == null) { messageRetrySupport.addMessageToRetry(canCommit, getSender(), - "Could not canCommit transaction " + canCommit.getTransactionId()); + "Could not canCommit transaction " + txId); } else { LOG.debug("{}: Forwarding CanCommitTransaction to leader {}", persistenceId(), leader); leader.forward(canCommit, getContext()); @@ -654,6 +680,8 @@ public class Shard extends RaftActor { @SuppressWarnings("checkstyle:IllegalCatch") protected void handleBatchedModificationsLocal(final BatchedModifications batched, final ActorRef sender) { + askProtocolEncountered(batched.getTransactionId()); + try { commitCoordinator.handleBatchedModifications(batched, sender, this); } catch (Exception e) { @@ -750,6 +778,7 @@ public class Shard extends RaftActor { boolean isLeaderActive = isLeaderActive(); if (isLeader() && isLeaderActive) { + askProtocolEncountered(forwardedReady.getTransactionId()); commitCoordinator.handleForwardedReadyTransaction(forwardedReady, getSender(), this); } else { ActorSelection leader = getLeader(); @@ -769,7 +798,9 @@ public class Shard extends RaftActor { } private void handleAbortTransaction(final AbortTransaction abort) { - doAbortTransaction(abort.getTransactionId(), getSender()); + final TransactionIdentifier transactionId = abort.getTransactionId(); + askProtocolEncountered(transactionId); + doAbortTransaction(transactionId, getSender()); } void doAbortTransaction(final Identifier transactionID, final ActorRef sender) { @@ -790,9 +821,8 @@ public class Shard extends RaftActor { private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) { if (isLeader()) { final LocalHistoryIdentifier id = closeTransactionChain.getIdentifier(); - // FIXME: CONTROLLER-1628: stage purge once no transactions are present - store.closeTransactionChain(id, null); - store.purgeTransactionChain(id, null); + askProtocolEncountered(id.getClientId()); + store.closeTransactionChain(id); } else if (getLeader() != null) { getLeader().forward(closeTransactionChain, getContext()); } else { @@ -802,6 +832,8 @@ public class Shard extends RaftActor { @SuppressWarnings("checkstyle:IllegalCatch") private void createTransaction(final CreateTransaction createTransaction) { + askProtocolEncountered(createTransaction.getTransactionId()); + try { if (TransactionType.fromInt(createTransaction.getTransactionType()) != TransactionType.READ_ONLY && failIfIsolatedLeader(getSender())) { @@ -824,6 +856,27 @@ public class Shard extends RaftActor { transactionId); } + // Called on leader only + private void askProtocolEncountered(final TransactionIdentifier transactionId) { + askProtocolEncountered(transactionId.getHistoryId().getClientId()); + } + + // Called on leader only + private void askProtocolEncountered(final ClientIdentifier clientId) { + final FrontendIdentifier frontend = clientId.getFrontendId(); + final LeaderFrontendState state = knownFrontends.get(frontend); + if (!(state instanceof LeaderFrontendState.Disabled)) { + LOG.debug("{}: encountered ask-based client {}, disabling transaction tracking", persistenceId(), clientId); + if (knownFrontends.isEmpty()) { + knownFrontends = new HashMap<>(); + } + knownFrontends.put(frontend, new LeaderFrontendState.Disabled(persistenceId(), clientId, getDataStore())); + + persistPayload(clientId, DisableTrackingPayload.create(clientId, + datastoreContext.getInitialPayloadSerializedBufferCapacity()), false); + } + } + private void updateSchemaContext(final UpdateSchemaContext message) { updateSchemaContext(message.getSchemaContext()); } @@ -845,7 +898,6 @@ public class Shard extends RaftActor { } @Override - @Nonnull protected RaftActorRecoveryCohort getRaftActorRecoveryCohort() { if (restoreFromSnapshot == null) { return ShardRecoveryCoordinator.create(store, persistenceId(), LOG); @@ -865,7 +917,7 @@ public class Shard extends RaftActor { if (txCommitTimeoutCheckSchedule == null) { // Schedule a message to be periodically sent to check if the current in-progress // transaction should be expired and aborted. - FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS); + FiniteDuration period = FiniteDuration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS); txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule( period, period, getSelf(), TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender()); @@ -875,6 +927,11 @@ public class Shard extends RaftActor { @Override protected void applyState(final ActorRef clientActor, final Identifier identifier, final Object data) { if (data instanceof Payload) { + if (data instanceof DisableTrackingPayload) { + disableTracking((DisableTrackingPayload) data); + return; + } + try { store.applyReplicatedPayload(identifier, (Payload)data); } catch (DataValidationFailedException | IOException e) { @@ -973,8 +1030,10 @@ public class Shard extends RaftActor { paused = true; // Tell-based protocol can replay transaction state, so it is safe to blow it up when we are paused. - knownFrontends.values().forEach(LeaderFrontendState::retire); - knownFrontends = ImmutableMap.of(); + if (datastoreContext.isUseTellBasedProtocol()) { + knownFrontends.values().forEach(LeaderFrontendState::retire); + knownFrontends = ImmutableMap.of(); + } store.setRunOnPendingTransactionsComplete(operation); }