sal-distributed-datastore Checkstyle fixes (for next version)
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / AbstractClientHistory.java
1 /*
2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.controller.cluster.databroker.actors.dds;
9
10 import com.google.common.base.Preconditions;
11 import com.google.common.base.Verify;
12 import java.util.HashMap;
13 import java.util.Map;
14 import java.util.concurrent.ConcurrentHashMap;
15 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
16 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
17 import java.util.concurrent.locks.StampedLock;
18 import javax.annotation.concurrent.GuardedBy;
19 import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
20 import org.opendaylight.controller.cluster.access.client.ConnectedClientConnection;
21 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
22 import org.opendaylight.controller.cluster.access.client.InversibleLockException;
23 import org.opendaylight.controller.cluster.access.commands.CreateLocalHistoryRequest;
24 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
25 import org.opendaylight.controller.cluster.access.concepts.Response;
26 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
27 import org.opendaylight.mdsal.common.api.TransactionChainClosedException;
28 import org.opendaylight.yangtools.concepts.Identifiable;
29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
32
33 /**
34  * Abstract base class for client view of a history. This class has two implementations, one for normal local histories
35  * and the other for single transactions.
36  *
37  * @author Robert Varga
38  */
39 abstract class AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> {
40     enum State {
41         IDLE,
42         TX_OPEN,
43         CLOSED,
44     }
45
46     private static final Logger LOG = LoggerFactory.getLogger(AbstractClientHistory.class);
47     private static final AtomicLongFieldUpdater<AbstractClientHistory> NEXT_TX_UPDATER =
48             AtomicLongFieldUpdater.newUpdater(AbstractClientHistory.class, "nextTx");
49     private static final AtomicReferenceFieldUpdater<AbstractClientHistory, State> STATE_UPDATER =
50             AtomicReferenceFieldUpdater.newUpdater(AbstractClientHistory.class, State.class, "state");
51
52     @GuardedBy("this")
53     private final Map<TransactionIdentifier, AbstractClientHandle<?>> openTransactions = new HashMap<>();
54     @GuardedBy("this")
55     private final Map<TransactionIdentifier, AbstractTransactionCommitCohort> readyTransactions = new HashMap<>();
56
57     @GuardedBy("lock")
58     private final Map<Long, ProxyHistory> histories = new ConcurrentHashMap<>();
59     private final StampedLock lock = new StampedLock();
60
61     private final AbstractDataStoreClientBehavior client;
62     private final LocalHistoryIdentifier identifier;
63
64     // Used via NEXT_TX_UPDATER
65     @SuppressWarnings("unused")
66     private volatile long nextTx = 0;
67
68     private volatile State state = State.IDLE;
69
70     AbstractClientHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier identifier) {
71         this.client = Preconditions.checkNotNull(client);
72         this.identifier = Preconditions.checkNotNull(identifier);
73         Preconditions.checkArgument(identifier.getCookie() == 0);
74     }
75
76     final State state() {
77         return state;
78     }
79
80     final void updateState(final State expected, final State next) {
81         final boolean success = STATE_UPDATER.compareAndSet(this, expected, next);
82         Preconditions.checkState(success, "Race condition detected, state changed from %s to %s", expected, state);
83         LOG.debug("Client history {} changed state from {} to {}", this, expected, next);
84     }
85
86     @Override
87     public final LocalHistoryIdentifier getIdentifier() {
88         return identifier;
89     }
90
91     final long nextTx() {
92         return NEXT_TX_UPDATER.getAndIncrement(this);
93     }
94
95     final Long resolveShardForPath(final YangInstanceIdentifier path) {
96         return client.resolveShardForPath(path);
97     }
98
99     @Override
100     final void localAbort(final Throwable cause) {
101         final State oldState = STATE_UPDATER.getAndSet(this, State.CLOSED);
102         if (oldState != State.CLOSED) {
103             LOG.debug("Force-closing history {}", getIdentifier(), cause);
104
105             synchronized (this) {
106                 for (AbstractClientHandle<?> t : openTransactions.values()) {
107                     t.localAbort(cause);
108                 }
109                 openTransactions.clear();
110                 readyTransactions.clear();
111             }
112         }
113     }
114
115     /**
116      * Create a new history proxy for a given shard.
117      *
118      * @throws InversibleLockException if the shard is being reconnected
119      */
120     @GuardedBy("lock")
121     private ProxyHistory createHistoryProxy(final Long shard) {
122         final AbstractClientConnection<ShardBackendInfo> connection = client.getConnection(shard);
123         final LocalHistoryIdentifier proxyId = new LocalHistoryIdentifier(identifier.getClientId(),
124             identifier.getHistoryId(), shard);
125         LOG.debug("Created proxyId {} for history {} shard {}", proxyId, identifier, shard);
126
127         final ProxyHistory ret = createHistoryProxy(proxyId, connection);
128
129         // Request creation of the history, if it is not the single history
130         if (ret.getIdentifier().getHistoryId() != 0) {
131             connection.sendRequest(new CreateLocalHistoryRequest(ret.getIdentifier(), connection.localActor()),
132                 this::createHistoryCallback);
133         }
134         return ret;
135     }
136
137     abstract ProxyHistory createHistoryProxy(LocalHistoryIdentifier historyId,
138             AbstractClientConnection<ShardBackendInfo> connection);
139
140     private void createHistoryCallback(final Response<?, ?> response) {
141         LOG.debug("Create history response {}", response);
142     }
143
144     private ProxyHistory ensureHistoryProxy(final TransactionIdentifier transactionId, final Long shard) {
145         while (true) {
146             try {
147                 // Short-lived lock to ensure exclusion of createHistoryProxy and the lookup phase in startReconnect,
148                 // see comments in startReconnect() for details.
149                 final long stamp = lock.readLock();
150                 try {
151                     return histories.computeIfAbsent(shard, this::createHistoryProxy);
152                 } finally {
153                     lock.unlockRead(stamp);
154                 }
155             } catch (InversibleLockException e) {
156                 LOG.trace("Waiting for transaction {} shard {} connection to resolve", transactionId, shard);
157                 e.awaitResolution();
158                 LOG.trace("Retrying transaction {} shard {} connection", transactionId, shard);
159             }
160         }
161     }
162
163     final AbstractProxyTransaction createSnapshotProxy(final TransactionIdentifier transactionId, final Long shard) {
164         return ensureHistoryProxy(transactionId, shard).createTransactionProxy(transactionId, true);
165     }
166
167     final AbstractProxyTransaction createTransactionProxy(final TransactionIdentifier transactionId, final Long shard) {
168         return ensureHistoryProxy(transactionId, shard).createTransactionProxy(transactionId, false);
169     }
170
171     private void checkNotClosed() {
172         if (state == State.CLOSED) {
173             throw new TransactionChainClosedException(String.format("Local history %s is closed", identifier));
174         }
175     }
176
177     /**
178      * Allocate a new {@link ClientTransaction}.
179      *
180      * @return A new {@link ClientTransaction}
181      * @throws TransactionChainClosedException if this history is closed
182      * @throws IllegalStateException if a previous dependent transaction has not been closed
183      */
184     public final ClientTransaction createTransaction() {
185         checkNotClosed();
186
187         synchronized (this) {
188             final ClientTransaction ret = doCreateTransaction();
189             openTransactions.put(ret.getIdentifier(), ret);
190             return ret;
191         }
192     }
193
194     /**
195      * Create a new {@link ClientSnapshot}.
196      *
197      * @return A new {@link ClientSnapshot}
198      * @throws TransactionChainClosedException if this history is closed
199      * @throws IllegalStateException if a previous dependent transaction has not been closed
200      */
201     public final ClientSnapshot takeSnapshot() {
202         checkNotClosed();
203
204         synchronized (this) {
205             final ClientSnapshot ret = doCreateSnapshot();
206             openTransactions.put(ret.getIdentifier(), ret);
207             return ret;
208         }
209     }
210
211     @GuardedBy("this")
212     abstract ClientSnapshot doCreateSnapshot();
213
214     @GuardedBy("this")
215     abstract ClientTransaction doCreateTransaction();
216
217     /**
218      * Callback invoked from {@link ClientTransaction} when a child transaction readied for submission.
219      *
220      * @param txId Transaction identifier
221      * @param cohort Transaction commit cohort
222      */
223     synchronized AbstractTransactionCommitCohort onTransactionReady(final ClientTransaction tx,
224             final AbstractTransactionCommitCohort cohort) {
225         final TransactionIdentifier txId = tx.getIdentifier();
226         if (openTransactions.remove(txId) == null) {
227             LOG.warn("Transaction {} not recorded, proceeding with readiness", txId);
228         }
229
230         final AbstractTransactionCommitCohort previous = readyTransactions.putIfAbsent(txId, cohort);
231         Preconditions.checkState(previous == null, "Duplicate cohort %s for transaction %s, already have %s",
232                 cohort, txId, previous);
233
234         LOG.debug("Local history {} readied transaction {}", this, txId);
235         return cohort;
236     }
237
238     /**
239      * Callback invoked from {@link ClientTransaction} when a child transaction has been aborted without touching
240      * backend.
241      *
242      * @param snapshot transaction identifier
243      */
244     synchronized void onTransactionAbort(final AbstractClientHandle<?> snapshot) {
245         if (openTransactions.remove(snapshot.getIdentifier()) == null) {
246             LOG.warn("Could not find aborting transaction {}", snapshot.getIdentifier());
247         }
248     }
249
250     /**
251      * Callback invoked from {@link AbstractTransactionCommitCohort} when a child transaction has been completed
252      * and all its state can be removed.
253      *
254      * @param txId transaction identifier
255      */
256     synchronized void onTransactionComplete(final TransactionIdentifier txId) {
257         if (readyTransactions.remove(txId) == null) {
258             LOG.warn("Could not find completed transaction {}", txId);
259         }
260     }
261
262     HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) {
263         /*
264          * This looks ugly and unusual and there is a reason for that, as the locking involved is in multiple places.
265          *
266          * We need to make sure that a new proxy is not created while we are reconnecting, which is partially satisfied
267          * by client.getConnection() throwing InversibleLockException by the time this method is invoked. That does
268          * not cover the case when createHistoryProxy() has already acquired the connection, but has not yet populated
269          * the history map.
270          *
271          * Hence we need to make sure no potential computation is happening concurrently with us looking at the history
272          * map. Once we have performed that lookup, though, we can release the lock immediately, as all creation
273          * requests are established to happen either before or after the reconnect attempt.
274          */
275         final ProxyHistory oldProxy;
276         final long stamp = lock.writeLock();
277         try {
278             oldProxy = histories.get(newConn.cookie());
279         } finally {
280             lock.unlockWrite(stamp);
281         }
282
283         if (oldProxy == null) {
284             return null;
285         }
286
287         final ProxyReconnectCohort proxy = Verify.verifyNotNull(oldProxy.startReconnect(newConn));
288         return new HistoryReconnectCohort() {
289             @Override
290             ProxyReconnectCohort getProxy() {
291                 return proxy;
292             }
293
294             @Override
295             void replaySuccessfulRequests(final Iterable<ConnectionEntry> previousEntries) {
296                 proxy.replaySuccessfulRequests(previousEntries);
297             }
298
299             @Override
300             public void close() {
301                 LOG.debug("Client history {} finishing reconnect to {}", AbstractClientHistory.this, newConn);
302                 final ProxyHistory newProxy = proxy.finishReconnect();
303                 if (!histories.replace(newConn.cookie(), oldProxy, newProxy)) {
304                     LOG.warn("Failed to replace proxy {} with {} in {}", oldProxy, newProxy,
305                         AbstractClientHistory.this);
306                 }
307             }
308         };
309     }
310 }