BUG-5280: handle TransactionPurgeRequest replay
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / LocalReadWriteProxyTransaction.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.function.Consumer;
13 import javax.annotation.Nullable;
14 import javax.annotation.concurrent.NotThreadSafe;
15 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
16 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
17 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
18 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
19 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
20 import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
21 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
22 import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
23 import org.opendaylight.controller.cluster.access.commands.TransactionModification;
24 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
25 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
26 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
27 import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
28 import org.opendaylight.controller.cluster.access.concepts.Response;
29 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
30 import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;
31 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
32 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
33 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
34 import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeModification;
35 import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeSnapshot;
36 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
37 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModificationCursor;
38 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41
42 /**
43  * An {@link AbstractProxyTransaction} for dispatching a transaction towards a shard leader which is co-located with
44  * the client instance.
45  *
46  * <p>
47  * It requires a {@link DataTreeSnapshot}, which is used to instantiated a new {@link DataTreeModification}. Operations
48  * are then performed on this modification and once the transaction is submitted, the modification is sent to the shard
49  * leader.
50  *
51  * <p>
52  * This class is not thread-safe as usual with transactions. Since it does not interact with the backend until the
53  * transaction is submitted, at which point this class gets out of the picture, this is not a cause for concern.
54  *
55  * @author Robert Varga
56  */
57 @NotThreadSafe
58 final class LocalReadWriteProxyTransaction extends LocalProxyTransaction {
59     private static final Logger LOG = LoggerFactory.getLogger(LocalReadWriteProxyTransaction.class);
60
61     private CursorAwareDataTreeModification modification;
62     private CursorAwareDataTreeModification sealedModification;
63
64     LocalReadWriteProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier,
65         final DataTreeSnapshot snapshot) {
66         super(parent, identifier);
67         this.modification = (CursorAwareDataTreeModification) snapshot.newModification();
68     }
69
70     @Override
71     boolean isSnapshotOnly() {
72         return false;
73     }
74
75     @Override
76     CursorAwareDataTreeSnapshot readOnlyView() {
77         return modification;
78     }
79
80     @Override
81     void doDelete(final YangInstanceIdentifier path) {
82         modification.delete(path);
83     }
84
85     @Override
86     void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
87         modification.merge(path, data);
88     }
89
90     @Override
91     void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
92         modification.write(path, data);
93     }
94
95     private RuntimeException abortedException() {
96         return new IllegalStateException("Tracker " + getIdentifier() + " has been aborted");
97     }
98
99     private RuntimeException submittedException() {
100         return new IllegalStateException("Tracker " + getIdentifier() + " has been submitted");
101     }
102
103     @Override
104     CommitLocalTransactionRequest commitRequest(final boolean coordinated) {
105         final CommitLocalTransactionRequest ret = new CommitLocalTransactionRequest(getIdentifier(), nextSequence(),
106             localActor(), modification, coordinated);
107         modification = new FailedDataTreeModification(this::submittedException);
108         return ret;
109     }
110
111     @Override
112     void doSeal() {
113         modification.ready();
114         sealedModification = modification;
115     }
116
117     @Override
118     void flushState(final AbstractProxyTransaction successor) {
119         sealedModification.applyToCursor(new AbstractDataTreeModificationCursor() {
120             @Override
121             public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
122                 successor.write(current().node(child), data);
123             }
124
125             @Override
126             public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
127                 successor.merge(current().node(child), data);
128             }
129
130             @Override
131             public void delete(final PathArgument child) {
132                 successor.delete(current().node(child));
133             }
134         });
135     }
136
137     DataTreeSnapshot getSnapshot() {
138         Preconditions.checkState(sealedModification != null, "Proxy %s is not sealed yet", getIdentifier());
139         return sealedModification;
140     }
141
142     @Override
143     void applyModifyTransactionRequest(final ModifyTransactionRequest request,
144             final @Nullable Consumer<Response<?, ?>> callback) {
145         for (TransactionModification mod : request.getModifications()) {
146             if (mod instanceof TransactionWrite) {
147                 modification.write(mod.getPath(), ((TransactionWrite)mod).getData());
148             } else if (mod instanceof TransactionMerge) {
149                 modification.merge(mod.getPath(), ((TransactionMerge)mod).getData());
150             } else if (mod instanceof TransactionDelete) {
151                 modification.delete(mod.getPath());
152             } else {
153                 throw new IllegalArgumentException("Unsupported modification " + mod);
154             }
155         }
156
157         final java.util.Optional<PersistenceProtocol> maybeProtocol = request.getPersistenceProtocol();
158         if (maybeProtocol.isPresent()) {
159             Verify.verify(callback != null, "Request {} has null callback", request);
160             ensureSealed();
161
162             switch (maybeProtocol.get()) {
163                 case ABORT:
164                     sendAbort(callback);
165                     break;
166                 case READY:
167                     // No-op, as we have already issued a seal()
168                     break;
169                 case SIMPLE:
170                     sendRequest(commitRequest(false), callback);
171                     break;
172                 case THREE_PHASE:
173                     sendRequest(commitRequest(true), callback);
174                     break;
175                 default:
176                     throw new IllegalArgumentException("Unhandled protocol " + maybeProtocol.get());
177             }
178         }
179     }
180
181     @Override
182     void handleForwardedRemoteRequest(final TransactionRequest<?> request,
183             final @Nullable Consumer<Response<?, ?>> callback) {
184         LOG.debug("Applying forwarded request {}", request);
185
186         if (request instanceof TransactionPreCommitRequest) {
187             sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
188         } else if (request instanceof TransactionDoCommitRequest) {
189             sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
190         } else if (request instanceof TransactionAbortRequest) {
191             sendAbort(callback);
192         } else {
193             super.handleForwardedRemoteRequest(request, callback);
194         }
195     }
196
197     @Override
198     void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest<?> request,
199             final Consumer<Response<?, ?>> callback) {
200         if (request instanceof CommitLocalTransactionRequest) {
201             final CommitLocalTransactionRequest req = (CommitLocalTransactionRequest) request;
202             final DataTreeModification mod = req.getModification();
203
204             LOG.debug("Applying modification {} to successor {}", mod, successor);
205             mod.applyToCursor(new AbstractDataTreeModificationCursor() {
206                 @Override
207                 public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
208                     successor.write(current().node(child), data);
209                 }
210
211                 @Override
212                 public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
213                     successor.merge(current().node(child), data);
214                 }
215
216                 @Override
217                 public void delete(final PathArgument child) {
218                     successor.delete(current().node(child));
219                 }
220             });
221
222             successor.ensureSealed();
223
224             final ModifyTransactionRequest successorReq = successor.commitRequest(req.isCoordinated());
225             successor.sendRequest(successorReq, callback);
226         } else if (request instanceof AbortLocalTransactionRequest) {
227             LOG.debug("Forwarding abort {} to successor {}", request, successor);
228             successor.abort();
229         } else if (request instanceof TransactionPurgeRequest) {
230             LOG.debug("Forwarding purge {} to successor {}", request, successor);
231             successor.purge();
232         } else {
233             throw new IllegalArgumentException("Unhandled request" + request);
234         }
235     }
236
237     @Override
238     void forwardToLocal(final LocalProxyTransaction successor, final TransactionRequest<?> request,
239             final Consumer<Response<?, ?>> callback) {
240         if (request instanceof CommitLocalTransactionRequest) {
241             Verify.verify(successor instanceof LocalReadWriteProxyTransaction);
242             ((LocalReadWriteProxyTransaction) successor).sendCommit((CommitLocalTransactionRequest)request, callback);
243             LOG.debug("Forwarded request {} to successor {}", request, successor);
244         } else {
245             super.forwardToLocal(successor, request, callback);
246         }
247     }
248
249     @Override
250     void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
251         super.sendAbort(request, callback);
252         modification = new FailedDataTreeModification(this::abortedException);
253     }
254
255     private void sendCommit(final CommitLocalTransactionRequest request, final Consumer<Response<?, ?>> callback) {
256         // Rebase old modification on new data tree.
257         try (DataTreeModificationCursor cursor = modification.createCursor(YangInstanceIdentifier.EMPTY)) {
258             request.getModification().applyToCursor(cursor);
259         }
260
261         ensureSealed();
262         sendRequest(commitRequest(request.isCoordinated()), callback);
263     }
264 }

©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.