X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=dom%2Fmdsal-dom-inmemory-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fmdsal%2Fdom%2Fstore%2Finmemory%2FInMemoryDOMDataTreeShardProducer.java;h=dcd867bf3c9dc069b15566f8275a1cbc8cf07222;hb=refs%2Fchanges%2F30%2F45830%2F1;hp=89be8f505ab8c7f7dff0e5c825693575058d0b61;hpb=7677854c99074c6c9e760e0c6a78f8b8bcd01bdd;p=mdsal.git diff --git a/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMDataTreeShardProducer.java b/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMDataTreeShardProducer.java index 89be8f505a..dcd867bf3c 100644 --- a/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMDataTreeShardProducer.java +++ b/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMDataTreeShardProducer.java @@ -10,32 +10,203 @@ package org.opendaylight.mdsal.dom.store.inmemory; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableSet; +import java.util.AbstractMap.SimpleEntry; import java.util.Collection; +import java.util.Map.Entry; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; +import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -final class InMemoryDOMDataTreeShardProducer implements DOMDataTreeShardProducer { +class InMemoryDOMDataTreeShardProducer implements DOMDataTreeShardProducer { + + private abstract static class State { + /** + * Allocate a new snapshot. + * + * @return A new snapshot + */ + protected abstract DataTreeSnapshot getSnapshot(Object transactionId); + } + + private static final class Idle extends State { + private final InMemoryDOMDataTreeShardProducer producer; + + Idle(final InMemoryDOMDataTreeShardProducer producer) { + this.producer = Preconditions.checkNotNull(producer); + } + + @Override + protected DataTreeSnapshot getSnapshot(Object transactionId) { + return producer.takeSnapshot(); + } + } + + /** + * We have a transaction out there. + */ + private static final class Allocated extends State { + private static final AtomicReferenceFieldUpdater SNAPSHOT_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(Allocated.class, DataTreeSnapshot.class, "snapshot"); + private final InmemoryDOMDataTreeShardWriteTransaction transaction; + private volatile DataTreeSnapshot snapshot; + + Allocated(final InmemoryDOMDataTreeShardWriteTransaction transaction) { + this.transaction = Preconditions.checkNotNull(transaction); + } + + public InmemoryDOMDataTreeShardWriteTransaction getTransaction() { + return transaction; + } + + @Override + protected DataTreeSnapshot getSnapshot(Object transactionId) { + final DataTreeSnapshot ret = snapshot; + Preconditions.checkState(ret != null, + "Could not get snapshot for transaction %s - previous transaction %s is not ready yet", + transactionId, transaction.getIdentifier()); + return ret; + } + + void setSnapshot(final DataTreeSnapshot snapshot) { + final boolean success = SNAPSHOT_UPDATER.compareAndSet(this, null, snapshot); + Preconditions.checkState(success, "Transaction %s has already been marked as ready", + transaction.getIdentifier()); + } + } + + /** + * Producer is logically shut down, no further allocation allowed. + */ + private static final class Shutdown extends State { + private final String message; + + Shutdown(final String message) { + this.message = Preconditions.checkNotNull(message); + } + + @Override + protected DataTreeSnapshot getSnapshot(Object transactionId) { + throw new IllegalStateException(message); + } + } + + private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMDataTreeShard.class); + private static final AtomicLong COUNTER = new AtomicLong(); private final InMemoryDOMDataTreeShard parentShard; private final Collection prefixes; - private InmemoryDOMDataTreeShardWriteTransaction currentTx; - private InmemoryDOMDataTreeShardWriteTransaction lastSubmittedTx; + private static final AtomicReferenceFieldUpdater STATE_UPDATER = + AtomicReferenceFieldUpdater.newUpdater(InMemoryDOMDataTreeShardProducer.class, State.class, "state"); + private final Idle idleState = new Idle(this); + private volatile State state; InMemoryDOMDataTreeShardProducer(final InMemoryDOMDataTreeShard parentShard, final Collection prefixes) { this.parentShard = Preconditions.checkNotNull(parentShard); this.prefixes = ImmutableSet.copyOf(prefixes); + state = idleState; } @Override - public InmemoryDOMDataTreeShardWriteTransaction createTransaction() { - Preconditions.checkState(currentTx == null); - if (lastSubmittedTx != null) { - currentTx = parentShard.createTransaction(lastSubmittedTx); + public synchronized InmemoryDOMDataTreeShardWriteTransaction createTransaction() { + Entry entry; + InmemoryDOMDataTreeShardWriteTransaction ret; + String transactionId = nextIdentifier(); + + do { + entry = getSnapshot(transactionId); + ret = parentShard.createTransaction(transactionId, this, prefixes, entry.getValue()); + } while (!recordTransaction(entry.getKey(), ret)); + + return ret; + } + + synchronized void transactionReady(final InmemoryDOMDataTreeShardWriteTransaction tx, + final DataTreeModification modification) { + final State localState = state; + LOG.debug("Transaction was readied {}, current state {}", tx.getIdentifier(), localState); + + if (localState instanceof Allocated) { + final Allocated allocated = (Allocated) localState; + final InmemoryDOMDataTreeShardWriteTransaction transaction = allocated.getTransaction(); + Preconditions.checkState(tx.equals(transaction), + "Mis-ordered ready transaction %s last allocated was %s", tx, transaction); + allocated.setSnapshot(modification); } else { - currentTx = parentShard.createTransaction(prefixes); + LOG.debug("Ignoring transaction {} readiness due to state {}", tx, localState); } - return currentTx; + } + + /** + * Notify the base logic that a previously-submitted transaction has been committed successfully. + * + * @param transaction Transaction which completed successfully. + */ + synchronized void onTransactionCommited(final InmemoryDOMDataTreeShardWriteTransaction transaction) { + // If the committed transaction was the one we allocated last, + // we clear it and the ready snapshot, so the next transaction + // allocated refers to the data tree directly. + final State localState = state; + LOG.debug("Transaction {} commit done, current state {}", transaction.getIdentifier(), localState); + + if (!(localState instanceof Allocated)) { + // This can legally happen if the chain is shut down before the transaction was committed + // by the backend. + LOG.debug("Ignoring successful transaction {} in state {}", transaction, localState); + return; + } + + final Allocated allocated = (Allocated) localState; + final InmemoryDOMDataTreeShardWriteTransaction tx = allocated.getTransaction(); + if (!tx.equals(transaction)) { + LOG.debug("Ignoring non-latest successful transaction {} in state {}", transaction, allocated); + return; + } + + if (!STATE_UPDATER.compareAndSet(this, localState, idleState)) { + LOG.debug("Producer {} has already transitioned from {} to {}, not making it idle", this, + localState, state); + } + } + + synchronized void transactionAborted(final InmemoryDOMDataTreeShardWriteTransaction tx) { + final State localState = state; + if (localState instanceof Allocated) { + final Allocated allocated = (Allocated)localState; + if (allocated.getTransaction().equals(tx)) { + final boolean success = STATE_UPDATER.compareAndSet(this, localState, idleState); + if (!success) { + LOG.warn("Transaction {} aborted, but producer {} state already transitioned from {} to {}", + tx, this, localState, state); + } + } + } + } + + + private Entry getSnapshot(String transactionId) { + final State localState = state; + return new SimpleEntry<>(localState, localState.getSnapshot(transactionId)); + } + + private boolean recordTransaction(final State expected, + final InmemoryDOMDataTreeShardWriteTransaction transaction) { + final State state = new Allocated(transaction); + return STATE_UPDATER.compareAndSet(this, expected, state); + } + + private String nextIdentifier() { + return "INMEMORY-SHARD-TX-" + COUNTER.getAndIncrement(); + + } + + DataTreeSnapshot takeSnapshot() { + return parentShard.takeSnapshot(); } @Override