2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore;
10 import akka.actor.ActorSelection;
11 import akka.dispatch.OnComplete;
12 import com.google.common.base.Preconditions;
13 import java.util.Collection;
14 import java.util.concurrent.atomic.AtomicInteger;
15 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
16 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionChainIdentifier;
17 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
18 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
19 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
20 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
21 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
22 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
23 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
24 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
25 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
26 import org.slf4j.Logger;
27 import org.slf4j.LoggerFactory;
28 import scala.concurrent.Future;
29 import scala.concurrent.Promise;
32 * A chain of {@link TransactionProxy}s. It allows a single open transaction to be open
33 * at a time. For remote transactions, it also tracks the outstanding readiness requests
34 * towards the shard and unblocks operations only after all have completed.
36 final class TransactionChainProxy extends AbstractTransactionContextFactory<LocalTransactionChain> implements DOMStoreTransactionChain {
37 private static abstract class State {
39 * Check if it is okay to allocate a new transaction.
40 * @throws IllegalStateException if a transaction may not be allocated.
42 abstract void checkReady();
45 * Return the future which needs to be waited for before shard information
46 * is returned (which unblocks remote transactions).
47 * @return Future to wait for, or null of no wait is necessary
49 abstract Future<?> previousFuture();
52 private static abstract class Pending extends State {
53 private final TransactionIdentifier transaction;
54 private final Future<?> previousFuture;
56 Pending(final TransactionIdentifier transaction, final Future<?> previousFuture) {
57 this.previousFuture = previousFuture;
58 this.transaction = Preconditions.checkNotNull(transaction);
62 final Future<?> previousFuture() {
63 return previousFuture;
66 final TransactionIdentifier getIdentifier() {
71 private static final class Allocated extends Pending {
72 Allocated(final TransactionIdentifier transaction, final Future<?> previousFuture) {
73 super(transaction, previousFuture);
78 throw new IllegalStateException(String.format("Previous transaction %s is not ready yet", getIdentifier()));
82 private static final class Submitted extends Pending {
83 Submitted(final TransactionIdentifier transaction, final Future<?> previousFuture) {
84 super(transaction, previousFuture);
93 private static abstract class DefaultState extends State {
95 final Future<?> previousFuture() {
100 private static final State IDLE_STATE = new DefaultState() {
107 private static final State CLOSED_STATE = new DefaultState() {
110 throw new TransactionChainClosedException("Transaction chain has been closed");
114 private static final Logger LOG = LoggerFactory.getLogger(TransactionChainProxy.class);
115 private static final AtomicInteger CHAIN_COUNTER = new AtomicInteger();
116 private static final AtomicReferenceFieldUpdater<TransactionChainProxy, State> STATE_UPDATER =
117 AtomicReferenceFieldUpdater.newUpdater(TransactionChainProxy.class, State.class, "currentState");
119 private final TransactionChainIdentifier transactionChainId;
120 private final TransactionContextFactory parent;
121 private volatile State currentState = IDLE_STATE;
123 TransactionChainProxy(final TransactionContextFactory parent) {
124 super(parent.getActorContext());
126 transactionChainId = new TransactionChainIdentifier(parent.getActorContext().getCurrentMemberName(), CHAIN_COUNTER.incrementAndGet());
127 this.parent = parent;
130 public String getTransactionChainId() {
131 return transactionChainId.toString();
135 public DOMStoreReadTransaction newReadOnlyTransaction() {
136 currentState.checkReady();
137 return new TransactionProxy(this, TransactionType.READ_ONLY);
141 public DOMStoreReadWriteTransaction newReadWriteTransaction() {
142 getActorContext().acquireTxCreationPermit();
143 return allocateWriteTransaction(TransactionType.READ_WRITE);
147 public DOMStoreWriteTransaction newWriteOnlyTransaction() {
148 getActorContext().acquireTxCreationPermit();
149 return allocateWriteTransaction(TransactionType.WRITE_ONLY);
153 public void close() {
154 currentState = CLOSED_STATE;
156 // Send a close transaction chain request to each and every shard
157 getActorContext().broadcast(new CloseTransactionChain(transactionChainId.toString()).toSerializable());
160 private TransactionProxy allocateWriteTransaction(final TransactionType type) {
161 State localState = currentState;
162 localState.checkReady();
164 final TransactionProxy ret = new TransactionProxy(this, type);
165 currentState = new Allocated(ret.getIdentifier(), localState.previousFuture());
170 protected LocalTransactionChain factoryForShard(final String shardName, final ActorSelection shardLeader, final DataTree dataTree) {
171 final LocalTransactionChain ret = new LocalTransactionChain(this, shardLeader, dataTree);
172 LOG.debug("Allocated transaction chain {} for shard {} leader {}", ret, shardName, shardLeader);
177 * This method is overridden to ensure the previous Tx's ready operations complete
178 * before we initiate the next Tx in the chain to avoid creation failures if the
179 * previous Tx's ready operations haven't completed yet.
182 protected Future<PrimaryShardInfo> findPrimaryShard(final String shardName) {
183 // Read current state atomically
184 final State localState = currentState;
186 // There are no outstanding futures, shortcut
187 final Future<?> previous = localState.previousFuture();
188 if (previous == null) {
189 return parent.findPrimaryShard(shardName);
192 final String previousTransactionId;
194 if(localState instanceof Pending){
195 previousTransactionId = ((Pending) localState).getIdentifier().toString();
196 LOG.debug("Waiting for ready futures with pending Tx {}", previousTransactionId);
198 previousTransactionId = "";
199 LOG.debug("Waiting for ready futures on chain {}", getTransactionChainId());
202 // Add a callback for completion of the combined Futures.
203 final Promise<PrimaryShardInfo> returnPromise = akka.dispatch.Futures.promise();
205 final OnComplete onComplete = new OnComplete() {
207 public void onComplete(final Throwable failure, final Object notUsed) {
208 if (failure != null) {
209 // A Ready Future failed so fail the returned Promise.
210 LOG.error("Ready future failed for Tx {}", previousTransactionId);
211 returnPromise.failure(failure);
213 LOG.debug("Previous Tx {} readied - proceeding to FindPrimaryShard",
214 previousTransactionId);
216 // Send the FindPrimaryShard message and use the resulting Future to complete the
218 returnPromise.completeWith(parent.findPrimaryShard(shardName));
223 previous.onComplete(onComplete, getActorContext().getClientDispatcher());
224 return returnPromise.future();
228 protected <T> void onTransactionReady(final TransactionIdentifier transaction, final Collection<Future<T>> cohortFutures) {
229 final State localState = currentState;
230 Preconditions.checkState(localState instanceof Allocated, "Readying transaction %s while state is %s", transaction, localState);
231 final TransactionIdentifier currentTx = ((Allocated)localState).getIdentifier();
232 Preconditions.checkState(transaction.equals(currentTx), "Readying transaction %s while %s is allocated", transaction, currentTx);
234 // Transaction ready and we are not waiting for futures -- go to idle
235 if (cohortFutures.isEmpty()) {
236 currentState = IDLE_STATE;
240 // Combine the ready Futures into 1
241 final Future<Iterable<T>> combined = akka.dispatch.Futures.sequence(
242 cohortFutures, getActorContext().getClientDispatcher());
244 // Record the we have outstanding futures
245 final State newState = new Submitted(transaction, combined);
246 currentState = newState;
248 // Attach a completion reset, but only if we do not allocate a transaction
250 combined.onComplete(new OnComplete<Iterable<T>>() {
252 public void onComplete(final Throwable arg0, final Iterable<T> arg1) {
253 STATE_UPDATER.compareAndSet(TransactionChainProxy.this, newState, IDLE_STATE);
255 }, getActorContext().getClientDispatcher());
259 protected TransactionIdentifier nextIdentifier() {
260 return transactionChainId.newTransactionIdentifier();