8115473a0565fb2c3865f4bc03e7216a8029a579
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / CompositeDataTreeCohort.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
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.Status;
12 import akka.actor.Status.Failure;
13 import akka.dispatch.ExecutionContexts;
14 import akka.dispatch.Futures;
15 import akka.dispatch.Recover;
16 import akka.japi.Function;
17 import akka.pattern.Patterns;
18 import akka.util.Timeout;
19 import com.google.common.base.Preconditions;
20 import com.google.common.base.Throwables;
21 import com.google.common.collect.Iterables;
22 import java.util.Collection;
23 import java.util.Iterator;
24 import java.util.Optional;
25 import java.util.concurrent.ExecutionException;
26 import java.util.concurrent.TimeoutException;
27 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
28 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.CanCommit;
29 import org.opendaylight.controller.cluster.datastore.DataTreeCohortActor.Success;
30 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
31 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
32 import scala.concurrent.Await;
33 import scala.concurrent.Future;
34
35 /**
36  *
37  * Composite cohort, which coordinates multiple user-provided cohorts as if it was only one cohort.
38  *
39  * It tracks current operation and list of cohorts which successfuly finished previous phase in
40  * case, if abort is necessary to invoke it only on cohort steps which are still active.
41  *
42  */
43 class CompositeDataTreeCohort {
44
45     private enum State {
46         /**
47          * Cohorts are idle, no messages were sent.
48          */
49         IDLE,
50         /**
51          * CanCommit message was sent to all participating cohorts.
52          */
53         CAN_COMMIT_SENT,
54         /**
55          * Successful canCommit responses were received from every participating cohort.
56          */
57         CAN_COMMIT_SUCCESSFUL,
58         /**
59          * PreCommit message was sent to all participating cohorts.
60          */
61         PRE_COMMIT_SENT,
62         /**
63          * Successful preCommit responses were received from every participating cohort.
64          */
65         PRE_COMMIT_SUCCESSFUL,
66         /**
67          * Commit message was send to all participating cohorts.
68          */
69         COMMIT_SENT,
70         /**
71          * Successful commit responses were received from all participating cohorts.
72          */
73         COMMITED,
74         /**
75          * Some of cohorts responsed back with unsuccessful message.
76          *
77          */
78         FAILED,
79         /**
80          *
81          * Abort message was send to all cohorts which responded with success previously.
82          *
83          */
84         ABORTED
85     }
86
87     protected static final Recover<Object> EXCEPTION_TO_MESSAGE = new Recover<Object>() {
88         @Override
89         public Failure recover(final Throwable error) throws Throwable {
90             return new Failure(error);
91         }
92     };
93
94
95     private final DataTreeCohortActorRegistry registry;
96     private final TransactionIdentifier txId;
97     private final SchemaContext schema;
98     private final Timeout timeout;
99     private Iterable<Success> successfulFromPrevious;
100     private State state = State.IDLE;
101
102     CompositeDataTreeCohort(final DataTreeCohortActorRegistry registry, final TransactionIdentifier transactionID,
103         final SchemaContext schema, final Timeout timeout) {
104         this.registry = Preconditions.checkNotNull(registry);
105         this.txId = Preconditions.checkNotNull(transactionID);
106         this.schema = Preconditions.checkNotNull(schema);
107         this.timeout = Preconditions.checkNotNull(timeout);
108     }
109
110     void canCommit(final DataTreeCandidate tip) throws ExecutionException, TimeoutException {
111         Collection<CanCommit> messages = registry.createCanCommitMessages(txId, tip, schema);
112         // FIXME: Optimize empty collection list with pre-created futures, containing success.
113         Future<Iterable<Object>> canCommitsFuture =
114                 Futures.traverse(messages, new Function<CanCommit, Future<Object>>() {
115                     @Override
116                     public Future<Object> apply(final CanCommit input) {
117                         return Patterns.ask(input.getCohort(), input, timeout).recover(EXCEPTION_TO_MESSAGE,
118                                 ExecutionContexts.global());
119                     }
120                 }, ExecutionContexts.global());
121         changeStateFrom(State.IDLE, State.CAN_COMMIT_SENT);
122         processResponses(canCommitsFuture, State.CAN_COMMIT_SENT, State.CAN_COMMIT_SUCCESSFUL);
123     }
124
125     void preCommit() throws ExecutionException, TimeoutException {
126         Preconditions.checkState(successfulFromPrevious != null);
127         Future<Iterable<Object>> preCommitFutures = sendMesageToSuccessful(new DataTreeCohortActor.PreCommit(txId));
128         changeStateFrom(State.CAN_COMMIT_SUCCESSFUL, State.PRE_COMMIT_SENT);
129         processResponses(preCommitFutures, State.PRE_COMMIT_SENT, State.PRE_COMMIT_SUCCESSFUL);
130     }
131
132     void commit() throws ExecutionException, TimeoutException {
133         Preconditions.checkState(successfulFromPrevious != null);
134         Future<Iterable<Object>> commitsFuture = sendMesageToSuccessful(new DataTreeCohortActor.Commit(txId));
135         changeStateFrom(State.PRE_COMMIT_SUCCESSFUL, State.COMMIT_SENT);
136         processResponses(commitsFuture, State.COMMIT_SENT, State.COMMITED);
137     }
138
139     Optional<Future<Iterable<Object>>> abort() {
140         if (successfulFromPrevious != null) {
141             return Optional.of(sendMesageToSuccessful(new DataTreeCohortActor.Abort(txId)));
142         }
143
144         return Optional.empty();
145     }
146
147     private Future<Iterable<Object>> sendMesageToSuccessful(final Object message) {
148         return Futures.traverse(successfulFromPrevious, new Function<DataTreeCohortActor.Success, Future<Object>>() {
149
150             @Override
151             public Future<Object> apply(final DataTreeCohortActor.Success cohortResponse) throws Exception {
152                 return Patterns.ask(cohortResponse.getCohort(), message, timeout);
153             }
154
155         }, ExecutionContexts.global());
156     }
157
158     private void processResponses(final Future<Iterable<Object>> resultsFuture, final State currentState, final State afterState)
159             throws TimeoutException, ExecutionException {
160         final Iterable<Object> results;
161         try {
162             results = Await.result(resultsFuture, timeout.duration());
163         } catch (Exception e) {
164             successfulFromPrevious = null;
165             Throwables.propagateIfInstanceOf(e, TimeoutException.class);
166             throw Throwables.propagate(e);
167         }
168         Iterable<Failure> failed = Iterables.filter(results, Status.Failure.class);
169         Iterable<Success> successful = Iterables.filter(results, DataTreeCohortActor.Success.class);
170         successfulFromPrevious = successful;
171         if (!Iterables.isEmpty(failed)) {
172             changeStateFrom(currentState, State.FAILED);
173             Iterator<Failure> it = failed.iterator();
174             Throwable firstEx = it.next().cause();
175             while (it.hasNext()) {
176                 firstEx.addSuppressed(it.next().cause());
177             }
178             Throwables.propagateIfPossible(firstEx, ExecutionException.class);
179             Throwables.propagateIfPossible(firstEx, TimeoutException.class);
180             throw Throwables.propagate(firstEx);
181         }
182         changeStateFrom(currentState, afterState);
183     }
184
185     void changeStateFrom(final State expected, final State followup) {
186         Preconditions.checkState(state == expected);
187         state = followup;
188     }
189 }

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