b6e9bfabab833e7123583d18b26ea679e23c77c0
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / LocalProxyTransaction.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.Optional;
11 import com.google.common.base.Preconditions;
12 import com.google.common.util.concurrent.CheckedFuture;
13 import com.google.common.util.concurrent.Futures;
14 import java.util.function.Consumer;
15 import javax.annotation.Nonnull;
16 import javax.annotation.Nullable;
17 import javax.annotation.concurrent.NotThreadSafe;
18 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
19 import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
20 import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
21 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
22 import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
23 import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
24 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
25 import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
26 import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;
27 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
28 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
29 import org.opendaylight.controller.cluster.access.concepts.Response;
30 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
31 import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;
32 import org.opendaylight.mdsal.common.api.ReadFailedException;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
36 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
37 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
38 import org.slf4j.Logger;
39 import org.slf4j.LoggerFactory;
40
41 /**
42  * An {@link AbstractProxyTransaction} for dispatching a transaction towards a shard leader which is co-located with
43  * the client instance.
44  *
45  * <p>
46  * It requires a {@link DataTreeSnapshot}, which is used to instantiated a new {@link DataTreeModification}. Operations
47  * are then performed on this modification and once the transaction is submitted, the modification is sent to the shard
48  * leader.
49  *
50  * <p>
51  * This class is not thread-safe as usual with transactions. Since it does not interact with the backend until the
52  * transaction is submitted, at which point this class gets out of the picture, this is not a cause for concern.
53  *
54  * @author Robert Varga
55  */
56 @NotThreadSafe
57 abstract class LocalProxyTransaction extends AbstractProxyTransaction {
58     private static final Logger LOG = LoggerFactory.getLogger(LocalProxyTransaction.class);
59
60     private final TransactionIdentifier identifier;
61
62     LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier, final boolean isDone) {
63         super(parent, isDone);
64         this.identifier = Preconditions.checkNotNull(identifier);
65     }
66
67     @Override
68     public final TransactionIdentifier getIdentifier() {
69         return identifier;
70     }
71
72     @Nonnull
73     abstract DataTreeSnapshot readOnlyView();
74
75     abstract void applyForwardedModifyTransactionRequest(ModifyTransactionRequest request,
76             @Nullable Consumer<Response<?, ?>> callback);
77
78     abstract void replayModifyTransactionRequest(ModifyTransactionRequest request,
79             @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
80
81     @Override
82     final CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
83         return Futures.immediateCheckedFuture(readOnlyView().readNode(path).isPresent());
84     }
85
86     @Override
87     final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
88         return Futures.immediateCheckedFuture(readOnlyView().readNode(path));
89     }
90
91     @Override
92     final AbortLocalTransactionRequest abortRequest() {
93         return new AbortLocalTransactionRequest(identifier, localActor());
94     }
95
96     @Override
97     void handleReplayedLocalRequest(final AbstractLocalTransactionRequest<?> request,
98             final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
99         if (request instanceof AbortLocalTransactionRequest) {
100             enqueueAbort(request, callback, enqueuedTicks);
101         } else {
102             throw new IllegalArgumentException("Unhandled request" + request);
103         }
104     }
105
106     private boolean handleReadRequest(final TransactionRequest<?> request,
107             @Nullable final Consumer<Response<?, ?>> callback) {
108         // Note we delay completion of read requests to limit the scope at which the client can run, as they have
109         // listeners, which we do not want to execute while we are reconnecting.
110         if (request instanceof ReadTransactionRequest) {
111             final YangInstanceIdentifier path = ((ReadTransactionRequest) request).getPath();
112             final Optional<NormalizedNode<?, ?>> result = readOnlyView().readNode(path);
113             if (callback != null) {
114                 // XXX: FB does not see that callback is final, on stack and has be check for non-null.
115                 final Consumer<Response<?, ?>> fbIsStupid = Preconditions.checkNotNull(callback);
116                 executeInActor(() -> fbIsStupid.accept(new ReadTransactionSuccess(request.getTarget(),
117                     request.getSequence(), result)));
118             }
119             return true;
120         } else if (request instanceof ExistsTransactionRequest) {
121             final YangInstanceIdentifier path = ((ExistsTransactionRequest) request).getPath();
122             final boolean result = readOnlyView().readNode(path).isPresent();
123             if (callback != null) {
124                 // XXX: FB does not see that callback is final, on stack and has be check for non-null.
125                 final Consumer<Response<?, ?>> fbIsStupid = Preconditions.checkNotNull(callback);
126                 executeInActor(() -> fbIsStupid.accept(new ExistsTransactionSuccess(request.getTarget(),
127                     request.getSequence(), result)));
128             }
129             return true;
130         } else {
131             return false;
132         }
133     }
134
135     @Override
136     void handleReplayedRemoteRequest(final TransactionRequest<?> request,
137             @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
138         if (request instanceof ModifyTransactionRequest) {
139             replayModifyTransactionRequest((ModifyTransactionRequest) request, callback, enqueuedTicks);
140         } else if (handleReadRequest(request, callback)) {
141             // No-op
142         } else if (request instanceof TransactionPurgeRequest) {
143             enqueuePurge(callback, enqueuedTicks);
144         } else if (request instanceof IncrementTransactionSequenceRequest) {
145             // Local transactions do not have non-replayable requests which would be visible to the backend,
146             // hence we can skip sequence increments.
147             LOG.debug("Not replaying {}", request);
148         } else {
149             throw new IllegalArgumentException("Unhandled request " + request);
150         }
151     }
152
153     /**
154      * Remote-to-local equivalent of {@link #handleReplayedRemoteRequest(TransactionRequest, Consumer, long)},
155      * except it is invoked in the forwarding path from
156      * {@link RemoteProxyTransaction#forwardToLocal(LocalProxyTransaction, TransactionRequest, Consumer)}.
157      *
158      * @param request Forwarded request
159      * @param callback Callback to be invoked once the request completes
160      */
161     void handleForwardedRemoteRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
162         if (request instanceof ModifyTransactionRequest) {
163             applyForwardedModifyTransactionRequest((ModifyTransactionRequest) request, callback);
164         } else if (handleReadRequest(request, callback)) {
165             // No-op
166         } else if (request instanceof TransactionPurgeRequest) {
167             enqueuePurge(callback);
168         } else {
169             throw new IllegalArgumentException("Unhandled request " + request);
170         }
171     }
172
173     @Override
174     final void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest<?> request,
175                          final Consumer<Response<?, ?>> callback) {
176         if (request instanceof CommitLocalTransactionRequest) {
177             final CommitLocalTransactionRequest req = (CommitLocalTransactionRequest) request;
178             final DataTreeModification mod = req.getModification();
179
180             LOG.debug("Applying modification {} to successor {}", mod, successor);
181             mod.applyToCursor(new AbstractDataTreeModificationCursor() {
182                 @Override
183                 public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
184                     successor.write(current().node(child), data);
185                 }
186
187                 @Override
188                 public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
189                     successor.merge(current().node(child), data);
190                 }
191
192                 @Override
193                 public void delete(final PathArgument child) {
194                     successor.delete(current().node(child));
195                 }
196             });
197
198             successor.sealOnly();
199             final ModifyTransactionRequest successorReq = successor.commitRequest(req.isCoordinated());
200             successor.sendRequest(successorReq, callback);
201         } else if (request instanceof AbortLocalTransactionRequest) {
202             LOG.debug("Forwarding abort {} to successor {}", request, successor);
203             successor.abort();
204         } else if (request instanceof TransactionPurgeRequest) {
205             LOG.debug("Forwarding purge {} to successor {}", request, successor);
206             successor.enqueuePurge(callback);
207         } else if (request instanceof ModifyTransactionRequest) {
208             successor.handleForwardedRequest(request, callback);
209         } else {
210             throwUnhandledRequest(request);
211         }
212     }
213
214     @Override
215     void forwardToLocal(final LocalProxyTransaction successor, final TransactionRequest<?> request,
216             final Consumer<Response<?, ?>> callback) {
217         if (request instanceof AbortLocalTransactionRequest) {
218             successor.sendAbort(request, callback);
219         } else if (request instanceof TransactionPurgeRequest) {
220             successor.enqueuePurge(callback);
221         } else {
222             throwUnhandledRequest(request);
223         }
224
225         LOG.debug("Forwarded request {} to successor {}", request, successor);
226     }
227
228     private static void throwUnhandledRequest(final TransactionRequest<?> request) {
229         throw new IllegalArgumentException("Unhandled request" + request);
230     }
231
232     void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
233         sendRequest(request, callback);
234     }
235
236     void enqueueAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
237             final long enqueuedTicks) {
238         enqueueRequest(request, callback, enqueuedTicks);
239     }
240 }

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