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%2FFrontendTransaction.java;h=de46d9d2ae4e90b766bace83bd5b19d80ea3e2c5;hb=546cd1fd100dbaa36908b22c2f422320dbd8c4b2;hp=c0249fd00089d1bd48abe9793ddb8564fb937a94;hpb=d9e3fa9b7d4f1ad484931151c3b3ef237a674ab2;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendTransaction.java index c0249fd000..de46d9d2ae 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendTransaction.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendTransaction.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved. + * Copyright (c) 2016, 2017 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, @@ -7,14 +7,16 @@ */ package org.opendaylight.controller.cluster.datastore; +import static java.util.Objects.requireNonNull; + import com.google.common.base.MoreObjects; -import com.google.common.base.Preconditions; import com.google.common.base.Verify; import java.util.ArrayDeque; -import java.util.Iterator; +import java.util.Optional; import java.util.Queue; -import javax.annotation.Nullable; -import javax.annotation.concurrent.NotThreadSafe; +import org.eclipse.jdt.annotation.Nullable; +import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest; +import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceSuccess; import org.opendaylight.controller.cluster.access.commands.OutOfOrderRequestException; import org.opendaylight.controller.cluster.access.commands.TransactionRequest; import org.opendaylight.controller.cluster.access.commands.TransactionSuccess; @@ -27,11 +29,10 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Frontend common transaction state as observed by the shard leader. + * Frontend common transaction state as observed by the shard leader. This class is NOT thread-safe. * * @author Robert Varga */ -@NotThreadSafe abstract class FrontendTransaction implements Identifiable { private static final Logger LOG = LoggerFactory.getLogger(FrontendTransaction.class); @@ -51,8 +52,8 @@ abstract class FrontendTransaction implements Identifiable> replaySequence(final long sequence) throws RequestException { + final Optional> replaySequence(final long sequence) throws RequestException { // Fast path check: if the requested sequence is the next request, bail early if (expectedSequence == sequence) { - return java.util.Optional.empty(); + return Optional.empty(); } // Check sequencing: we do not need to bother with future requests @@ -93,23 +94,21 @@ abstract class FrontendTransaction implements Identifiable it = replayQueue.iterator(); - while (it.hasNext()) { - final Object replay = it.next(); + for (Object replay : replayQueue) { if (replaySequence == sequence) { if (replay instanceof RequestException) { throw (RequestException) replay; } Verify.verify(replay instanceof TransactionSuccess); - return java.util.Optional.of((TransactionSuccess) replay); + return Optional.of((TransactionSuccess) replay); } replaySequence++; } // Not found - return java.util.Optional.empty(); + return Optional.empty(); } final void purgeSequencesUpTo(final long sequence) { @@ -122,6 +121,14 @@ abstract class FrontendTransaction implements Identifiable handleRequest(final TransactionRequest request, final RequestEnvelope envelope, final long now) throws RequestException { + if (request instanceof IncrementTransactionSequenceRequest) { + final IncrementTransactionSequenceRequest incr = (IncrementTransactionSequenceRequest) request; + expectedSequence += incr.getIncrement(); + + return recordSuccess(incr.getSequence(), + new IncrementTransactionSequenceSuccess(incr.getTarget(), incr.getSequence())); + } + if (previousFailure != null) { LOG.debug("{}: Rejecting request {} due to previous failure", persistenceId(), request, previousFailure); throw previousFailure; @@ -131,8 +138,10 @@ abstract class FrontendTransaction implements Identifiable doHandleRequest(TransactionRequest request, RequestEnvelope envelope, long now) throws RequestException; + abstract void retire(); + private void recordResponse(final long sequence, final Object response) { if (replayQueue.isEmpty()) { firstReplaySequence = sequence; @@ -179,5 +190,4 @@ abstract class FrontendTransaction implements Identifiable