X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fmd%2Fsal%2Fdom%2Fbroker%2Fimpl%2FDOMDataBrokerTransactionChainImpl.java;h=d75ad263f5d1bff02eb5b8e4ed3730ce6d9c48fa;hp=bcefc25ae86c5ec32fce666d8b099c44ac186d62;hb=3ec97cd0a86ad1b79f6854dc6924eb7b06e359a3;hpb=38fad7ae1a91ca3684e0736f94bfae5c11040cb0 diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataBrokerTransactionChainImpl.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataBrokerTransactionChainImpl.java old mode 100644 new mode 100755 index bcefc25ae8..d75ad263f5 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataBrokerTransactionChainImpl.java +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMDataBrokerTransactionChainImpl.java @@ -1,100 +1,144 @@ /* * Copyright (c) 2014 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, * and is available at http://www.eclipse.org/legal/epl-v10.html */ + package org.opendaylight.controller.md.sal.dom.broker.impl; +import com.google.common.base.Preconditions; +import com.google.common.base.Supplier; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.MoreExecutors; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLong; - -import javax.annotation.concurrent.GuardedBy; - -import org.opendaylight.controller.md.sal.common.api.TransactionStatus; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; -import org.opendaylight.controller.md.sal.common.api.data.TransactionChain; import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener; import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction; import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; -import org.opendaylight.yangtools.yang.common.RpcResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; -import com.google.common.collect.ImmutableMap; -import com.google.common.util.concurrent.ListenableFuture; - /** - * NormalizedNode implementation of {@link TransactionChain} which is backed + * NormalizedNode implementation of {@link org.opendaylight.controller.md.sal.common.api.data.TransactionChain} which + * is backed * by several {@link DOMStoreTransactionChain} differentiated by provided - * {@link LogicalDatastoreType} type. - * + * {@link org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType} type. */ -public class DOMDataBrokerTransactionChainImpl extends AbstractDOMForwardedTransactionFactory - implements DOMTransactionChain, DOMDataCommitErrorListener { +@Deprecated +final class DOMDataBrokerTransactionChainImpl extends + AbstractDOMForwardedTransactionFactory implements DOMTransactionChain { + private enum State { + RUNNING, CLOSING, CLOSED, FAILED, + } + private static final AtomicIntegerFieldUpdater COUNTER_UPDATER + = AtomicIntegerFieldUpdater.newUpdater(DOMDataBrokerTransactionChainImpl.class, "counter"); + private static final AtomicReferenceFieldUpdater STATE_UPDATER + = AtomicReferenceFieldUpdater.newUpdater(DOMDataBrokerTransactionChainImpl.class, State.class, "state"); private static final Logger LOG = LoggerFactory.getLogger(DOMDataBrokerTransactionChainImpl.class); - private final DOMDataCommitExecutor coordinator; + private final AtomicLong txNum = new AtomicLong(); + private final AbstractDOMDataBroker broker; private final TransactionChainListener listener; private final long chainId; - private final AtomicLong txNum = new AtomicLong(); - @GuardedBy("this") - private boolean failed = false; + + private volatile State state = State.RUNNING; + private volatile int counter = 0; /** + * DOMDataBrokerTransactionChainImpl constructor. * - * @param chainId - * ID of transaction chain - * @param chains - * Backing {@link DOMStoreTransactionChain}s. - * @param coordinator - * Commit Coordinator which should be used to coordinate commits - * of transaction - * produced by this chain. - * @param listener - * Listener, which listens on transaction chain events. - * @throws NullPointerException - * If any of arguments is null. + * @param chainId ID of transaction chain + * @param chains Backing {@link DOMStoreTransactionChain}s. + * @param broker Commit Coordinator which should be used to coordinate commits + * of transaction + * produced by this chain. + * @param listener Listener, which listens on transaction chain events. + * @throws NullPointerException If any of arguments is null. */ - public DOMDataBrokerTransactionChainImpl(final long chainId, - final ImmutableMap chains, - final DOMDataCommitExecutor coordinator, final TransactionChainListener listener) { + DOMDataBrokerTransactionChainImpl(final long chainId, + final Map chains, + final AbstractDOMDataBroker broker, + final TransactionChainListener listener) { super(chains); this.chainId = chainId; - this.coordinator = Preconditions.checkNotNull(coordinator); + this.broker = Preconditions.checkNotNull(broker); this.listener = Preconditions.checkNotNull(listener); } + private void checkNotFailed() { + Preconditions.checkState(state != State.FAILED, "Transaction chain has failed"); + } + @Override protected Object newTransactionIdentifier() { return "DOM-CHAIN-" + chainId + "-" + txNum.getAndIncrement(); } @Override - public synchronized ListenableFuture> commit( - final DOMDataWriteTransaction transaction, final Iterable cohorts) { - return coordinator.submit(transaction, cohorts, Optional. of(this)); + public ListenableFuture commit(final DOMDataWriteTransaction transaction, + final Collection cohorts, final Supplier futureValueSupplier) { + checkNotFailed(); + checkNotClosed(); + + final ListenableFuture ret = broker.commit(transaction, cohorts, futureValueSupplier); + + COUNTER_UPDATER.incrementAndGet(this); + Futures.addCallback(ret, new FutureCallback() { + @Override + public void onSuccess(final T result) { + transactionCompleted(); + } + + @Override + public void onFailure(final Throwable throwable) { + transactionFailed(transaction, throwable); + } + }, MoreExecutors.directExecutor()); + + return ret; } @Override - public synchronized void close() { + public void close() { + final boolean success = STATE_UPDATER.compareAndSet(this, State.RUNNING, State.CLOSING); + if (!success) { + LOG.debug("Chain {} is no longer running", this); + return; + } + super.close(); for (DOMStoreTransactionChain subChain : getTxFactories().values()) { subChain.close(); } - if (!failed) { - LOG.debug("Transaction chain {} successfully finished.", this); - listener.onTransactionChainSuccessful(this); + if (counter == 0) { + finishClose(); } } - @Override - public synchronized void onCommitFailed(final DOMDataWriteTransaction tx, final Throwable cause) { - failed = true; + private void finishClose() { + state = State.CLOSED; + listener.onTransactionChainSuccessful(this); + } + + private void transactionCompleted() { + if (COUNTER_UPDATER.decrementAndGet(this) == 0 && state == State.CLOSING) { + finishClose(); + } + } + + private void transactionFailed(final DOMDataWriteTransaction tx, final Throwable cause) { + state = State.FAILED; LOG.debug("Transaction chain {} failed.", this, cause); listener.onTransactionChainFailed(this, tx, cause); }