*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Verify;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
+import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.StampedLock;
-import javax.annotation.concurrent.GuardedBy;
+import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.checkerframework.checker.lock.qual.Holding;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
import org.opendaylight.controller.cluster.access.client.ConnectedClientConnection;
import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.Response;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.mdsal.common.api.TransactionChainClosedException;
+import org.opendaylight.mdsal.dom.api.DOMTransactionChainClosedException;
import org.opendaylight.yangtools.concepts.Identifiable;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.slf4j.Logger;
*
* @author Robert Varga
*/
-abstract class AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> {
+public abstract class AbstractClientHistory extends LocalAbortable implements Identifiable<LocalHistoryIdentifier> {
enum State {
IDLE,
TX_OPEN,
private final Map<Long, ProxyHistory> histories = new ConcurrentHashMap<>();
private final StampedLock lock = new StampedLock();
- private final AbstractDataStoreClientBehavior client;
- private final LocalHistoryIdentifier identifier;
+ private final @NonNull AbstractDataStoreClientBehavior client;
+ private final @NonNull LocalHistoryIdentifier identifier;
// Used via NEXT_TX_UPDATER
@SuppressWarnings("unused")
private volatile State state = State.IDLE;
AbstractClientHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier identifier) {
- this.client = Preconditions.checkNotNull(client);
- this.identifier = Preconditions.checkNotNull(identifier);
- Preconditions.checkArgument(identifier.getCookie() == 0);
+ this.client = requireNonNull(client);
+ this.identifier = requireNonNull(identifier);
+ checkArgument(identifier.getCookie() == 0);
}
final State state() {
final void updateState(final State expected, final State next) {
final boolean success = STATE_UPDATER.compareAndSet(this, expected, next);
- Preconditions.checkState(success, "Race condition detected, state changed from %s to %s", expected, state);
+ checkState(success, "Race condition detected, state changed from %s to %s", expected, state);
LOG.debug("Client history {} changed state from {} to {}", this, expected, next);
}
+ final synchronized void doClose() {
+ final State local = state;
+ if (local != State.CLOSED) {
+ checkState(local == State.IDLE, "Local history %s has an open transaction", this);
+ histories.values().forEach(ProxyHistory::close);
+ updateState(local, State.CLOSED);
+ }
+ }
+
+ final synchronized void onProxyDestroyed(final ProxyHistory proxyHistory) {
+ histories.remove(proxyHistory.getIdentifier().getCookie());
+ LOG.debug("{}: removed destroyed proxy {}", this, proxyHistory);
+ }
+
@Override
public final LocalHistoryIdentifier getIdentifier() {
return identifier;
/**
* Create a new history proxy for a given shard.
*
+ * @param shard Shard cookie
* @throws InversibleLockException if the shard is being reconnected
*/
- @GuardedBy("lock")
+ @Holding("lock")
private ProxyHistory createHistoryProxy(final Long shard) {
final AbstractClientConnection<ShardBackendInfo> connection = client.getConnection(shard);
final LocalHistoryIdentifier proxyId = new LocalHistoryIdentifier(identifier.getClientId(),
return ret;
}
- abstract ProxyHistory createHistoryProxy(final LocalHistoryIdentifier historyId,
- final AbstractClientConnection<ShardBackendInfo> connection);
+ abstract ProxyHistory createHistoryProxy(LocalHistoryIdentifier historyId,
+ AbstractClientConnection<ShardBackendInfo> connection);
private void createHistoryCallback(final Response<?, ?> response) {
LOG.debug("Create history response {}", response);
}
- private ProxyHistory ensureHistoryProxy(final TransactionIdentifier transactionId, final Long shard) {
+ private @NonNull ProxyHistory ensureHistoryProxy(final TransactionIdentifier transactionId, final Long shard) {
while (true) {
try {
// Short-lived lock to ensure exclusion of createHistoryProxy and the lookup phase in startReconnect,
}
}
- final AbstractProxyTransaction createSnapshotProxy(final TransactionIdentifier transactionId, final Long shard) {
+ final @NonNull AbstractProxyTransaction createSnapshotProxy(final TransactionIdentifier transactionId,
+ final Long shard) {
return ensureHistoryProxy(transactionId, shard).createTransactionProxy(transactionId, true);
}
- final AbstractProxyTransaction createTransactionProxy(final TransactionIdentifier transactionId, final Long shard) {
+ final @NonNull AbstractProxyTransaction createTransactionProxy(final TransactionIdentifier transactionId,
+ final Long shard) {
return ensureHistoryProxy(transactionId, shard).createTransactionProxy(transactionId, false);
}
private void checkNotClosed() {
if (state == State.CLOSED) {
- throw new TransactionChainClosedException(String.format("Local history %s is closed", identifier));
+ throw new DOMTransactionChainClosedException(String.format("Local history %s is closed", identifier));
}
}
* Allocate a new {@link ClientTransaction}.
*
* @return A new {@link ClientTransaction}
- * @throws TransactionChainClosedException if this history is closed
+ * @throws DOMTransactionChainClosedException if this history is closed
* @throws IllegalStateException if a previous dependent transaction has not been closed
*/
- public final ClientTransaction createTransaction() {
+ // Non-final for mocking
+ public @NonNull ClientTransaction createTransaction() {
checkNotClosed();
synchronized (this) {
* Create a new {@link ClientSnapshot}.
*
* @return A new {@link ClientSnapshot}
- * @throws TransactionChainClosedException if this history is closed
+ * @throws DOMTransactionChainClosedException if this history is closed
* @throws IllegalStateException if a previous dependent transaction has not been closed
*/
- public final ClientSnapshot takeSnapshot() {
+ // Non-final for mocking
+ public ClientSnapshot takeSnapshot() {
checkNotClosed();
synchronized (this) {
}
}
- @GuardedBy("this")
+ @Holding("this")
abstract ClientSnapshot doCreateSnapshot();
- @GuardedBy("this")
+ @Holding("this")
abstract ClientTransaction doCreateTransaction();
/**
- * Callback invoked from {@link ClientTransaction} when a child transaction readied for submission.
+ * Callback invoked from {@link AbstractClientHandle}'s lifecycle to inform that a particular transaction is
+ * completing with a set of participating shards.
*
* @param txId Transaction identifier
+ * @param participatingShards Participating shard cookies
+ */
+ final void onTransactionShardsBound(final TransactionIdentifier txId, final Set<Long> participatingShards) {
+ // Guard against startReconnect() kicking in. It is okay to connect new participants concurrently, as those
+ // will not see the holes caused by this.
+ final long stamp = lock.readLock();
+ try {
+ for (var entry : histories.entrySet()) {
+ if (!participatingShards.contains(entry.getKey())) {
+ entry.getValue().skipTransaction(txId);
+ }
+ }
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ /**
+ * Callback invoked from {@link ClientTransaction} when a child transaction readied for submission.
+ *
+ * @param tx Client transaction
* @param cohort Transaction commit cohort
*/
synchronized AbstractTransactionCommitCohort onTransactionReady(final ClientTransaction tx,
}
final AbstractTransactionCommitCohort previous = readyTransactions.putIfAbsent(txId, cohort);
- Preconditions.checkState(previous == null, "Duplicate cohort %s for transaction %s, already have %s",
- cohort, txId, previous);
+ checkState(previous == null, "Duplicate cohort %s for transaction %s, already have %s", cohort, txId, previous);
LOG.debug("Local history {} readied transaction {}", this, txId);
return cohort;
*
* @param txId transaction identifier
*/
+ // Non-final for mocking
synchronized void onTransactionComplete(final TransactionIdentifier txId) {
if (readyTransactions.remove(txId) == null) {
LOG.warn("Could not find completed transaction {}", txId);
}
}
- HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) {
+ final HistoryReconnectCohort startReconnect(final ConnectedClientConnection<ShardBackendInfo> newConn) {
/*
* This looks ugly and unusual and there is a reason for that, as the locking involved is in multiple places.
*
return null;
}
- final ProxyReconnectCohort proxy = Verify.verifyNotNull(oldProxy.startReconnect(newConn));
+ final ProxyReconnectCohort proxy = verifyNotNull(oldProxy.startReconnect(newConn));
return new HistoryReconnectCohort() {
@Override
ProxyReconnectCohort getProxy() {
}
@Override
- void replaySuccessfulRequests(final Iterable<ConnectionEntry> previousEntries) {
- proxy.replaySuccessfulRequests(previousEntries);
+ void replayRequests(final Collection<ConnectionEntry> previousEntries) {
+ proxy.replayRequests(previousEntries);
}
@Override
Code Review / controller.git / blobdiff