CDS: Implement front-end support for local transactions
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / RemoteTransactionContextSupport.java
1 /*
2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
3  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
4  *
5  * This program and the accompanying materials are made available under the
6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
7  * and is available at http://www.eclipse.org/legal/epl-v10.html
8  */
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.ActorSelection;
12 import akka.dispatch.OnComplete;
13 import com.google.common.base.Preconditions;
14 import java.util.concurrent.Semaphore;
15 import java.util.concurrent.TimeUnit;
16 import org.opendaylight.controller.cluster.datastore.compat.PreLithiumTransactionContextImpl;
17 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
18 import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
19 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
20 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
21 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
24 import scala.concurrent.Future;
25 import scala.concurrent.duration.FiniteDuration;
26
27 /**
28  * Handles creation of TransactionContext instances for remote transactions. This class creates
29  * remote transactions, if necessary, by sending CreateTransaction messages with retries, up to a limit,
30  * if the shard doesn't have a leader yet. This is done by scheduling a retry task after a short delay.
31  * <p>
32  * The end result from a completed CreateTransaction message is a TransactionContext that is
33  * used to perform transaction operations. Transaction operations that occur before the
34  * CreateTransaction completes are cache via a TransactionContextWrapper and executed once the
35  * CreateTransaction completes, successfully or not.
36  */
37 final class RemoteTransactionContextSupport {
38     private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContextSupport.class);
39
40     /**
41      * Time interval in between transaction create retries.
42      */
43     private static final FiniteDuration CREATE_TX_TRY_INTERVAL = FiniteDuration.create(1, TimeUnit.SECONDS);
44
45     private final TransactionProxy parent;
46     private final String shardName;
47
48     /**
49      * The target primary shard.
50      */
51     private volatile ActorSelection primaryShard;
52     private volatile int createTxTries;
53
54     private final TransactionContextWrapper transactionContextAdapter;
55
56     RemoteTransactionContextSupport(final TransactionContextWrapper transactionContextAdapter, final TransactionProxy parent,
57             final String shardName) {
58         this.parent = Preconditions.checkNotNull(parent);
59         this.shardName = shardName;
60         this.transactionContextAdapter = transactionContextAdapter;
61         createTxTries = (int) (parent.getActorContext().getDatastoreContext().
62                 getShardLeaderElectionTimeout().duration().toMillis() /
63                 CREATE_TX_TRY_INTERVAL.toMillis());
64     }
65
66     String getShardName() {
67         return shardName;
68     }
69
70     private TransactionType getTransactionType() {
71         return parent.getType();
72     }
73
74     private ActorContext getActorContext() {
75         return parent.getActorContext();
76     }
77
78     private Semaphore getOperationLimiter() {
79         return parent.getLimiter();
80     }
81
82     private TransactionIdentifier getIdentifier() {
83         return parent.getIdentifier();
84     }
85
86     /**
87      * Sets the target primary shard and initiates a CreateTransaction try.
88      */
89     void setPrimaryShard(ActorSelection primaryShard) {
90         this.primaryShard = primaryShard;
91
92         if (getTransactionType() == TransactionType.WRITE_ONLY &&
93                 getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
94             LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
95                 getIdentifier(), primaryShard);
96
97             // For write-only Tx's we prepare the transaction modifications directly on the shard actor
98             // to avoid the overhead of creating a separate transaction actor.
99             // FIXME: can't assume the shard version is LITHIUM_VERSION - need to obtain it somehow.
100             transactionContextAdapter.executePriorTransactionOperations(createValidTransactionContext(this.primaryShard,
101                     this.primaryShard.path().toString(), DataStoreVersions.LITHIUM_VERSION));
102         } else {
103             tryCreateTransaction();
104         }
105     }
106
107     /**
108      * Performs a CreateTransaction try async.
109      */
110     private void tryCreateTransaction() {
111         if(LOG.isDebugEnabled()) {
112             LOG.debug("Tx {} Primary shard {} found - trying create transaction", getIdentifier(), primaryShard);
113         }
114
115         Object serializedCreateMessage = new CreateTransaction(getIdentifier().toString(),
116             getTransactionType().ordinal(), getIdentifier().getChainId()).toSerializable();
117
118         Future<Object> createTxFuture = getActorContext().executeOperationAsync(primaryShard, serializedCreateMessage);
119
120         createTxFuture.onComplete(new OnComplete<Object>() {
121             @Override
122             public void onComplete(Throwable failure, Object response) {
123                 onCreateTransactionComplete(failure, response);
124             }
125         }, getActorContext().getClientDispatcher());
126     }
127
128     private void onCreateTransactionComplete(Throwable failure, Object response) {
129         if(failure instanceof NoShardLeaderException) {
130             // There's no leader for the shard yet - schedule and try again, unless we're out
131             // of retries. Note: createTxTries is volatile as it may be written by different
132             // threads however not concurrently, therefore decrementing it non-atomically here
133             // is ok.
134             if(--createTxTries > 0) {
135                 LOG.debug("Tx {} Shard {} has no leader yet - scheduling create Tx retry",
136                     getIdentifier(), shardName);
137
138                 getActorContext().getActorSystem().scheduler().scheduleOnce(CREATE_TX_TRY_INTERVAL,
139                         new Runnable() {
140                             @Override
141                             public void run() {
142                                 tryCreateTransaction();
143                             }
144                         }, getActorContext().getClientDispatcher());
145                 return;
146             }
147         }
148
149         createTransactionContext(failure, response);
150     }
151
152     private void createTransactionContext(Throwable failure, Object response) {
153         // Create the TransactionContext from the response or failure. Store the new
154         // TransactionContext locally until we've completed invoking the
155         // TransactionOperations. This avoids thread timing issues which could cause
156         // out-of-order TransactionOperations. Eg, on a modification operation, if the
157         // TransactionContext is non-null, then we directly call the TransactionContext.
158         // However, at the same time, the code may be executing the cached
159         // TransactionOperations. So to avoid thus timing, we don't publish the
160         // TransactionContext until after we've executed all cached TransactionOperations.
161         TransactionContext localTransactionContext;
162         if(failure != null) {
163             LOG.debug("Tx {} Creating NoOpTransaction because of error", getIdentifier(), failure);
164
165             localTransactionContext = new NoOpTransactionContext(failure, getIdentifier(), getOperationLimiter());
166         } else if (CreateTransactionReply.SERIALIZABLE_CLASS.equals(response.getClass())) {
167             localTransactionContext = createValidTransactionContext(
168                     CreateTransactionReply.fromSerializable(response));
169         } else {
170             IllegalArgumentException exception = new IllegalArgumentException(String.format(
171                     "Invalid reply type %s for CreateTransaction", response.getClass()));
172
173             localTransactionContext = new NoOpTransactionContext(exception, getIdentifier(), getOperationLimiter());
174         }
175
176         transactionContextAdapter.executePriorTransactionOperations(localTransactionContext);
177     }
178
179     private TransactionContext createValidTransactionContext(CreateTransactionReply reply) {
180         LOG.debug("Tx {} Received {}", getIdentifier(), reply);
181
182         return createValidTransactionContext(getActorContext().actorSelection(reply.getTransactionPath()),
183                 reply.getTransactionPath(), reply.getVersion());
184     }
185
186     private TransactionContext createValidTransactionContext(ActorSelection transactionActor, String transactionPath,
187             short remoteTransactionVersion) {
188         // TxActor is always created where the leader of the shard is.
189         // Check if TxActor is created in the same node
190         boolean isTxActorLocal = getActorContext().isPathLocal(transactionPath);
191         final TransactionContext ret;
192
193         if (remoteTransactionVersion < DataStoreVersions.LITHIUM_VERSION) {
194             ret = new PreLithiumTransactionContextImpl(transactionPath, transactionActor, getIdentifier(),
195                 getActorContext(), isTxActorLocal, remoteTransactionVersion, parent.getCompleter());
196         } else {
197             ret = new RemoteTransactionContext(transactionActor, getIdentifier(), getActorContext(),
198                 isTxActorLocal, remoteTransactionVersion, parent.getCompleter());
199         }
200
201         TransactionContextCleanup.track(this, ret);
202         return ret;
203     }
204 }
205

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.