2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.databroker.actors.dds;
10 import akka.actor.ActorRef;
11 import akka.actor.Status;
12 import com.google.common.base.Throwables;
13 import com.google.common.base.Verify;
15 import java.util.concurrent.ConcurrentHashMap;
16 import java.util.concurrent.atomic.AtomicLong;
17 import java.util.function.Consumer;
18 import org.opendaylight.controller.cluster.access.client.ClientActorBehavior;
19 import org.opendaylight.controller.cluster.access.client.ClientActorContext;
20 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
21 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
22 import org.opendaylight.controller.cluster.access.concepts.Response;
23 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
24 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
29 * {@link ClientActorBehavior} acting as an intermediary between the backend actors and the DistributedDataStore
33 * This class is not visible outside of this package because it breaks the actor containment. Services provided to
34 * Java world outside of actor containment are captured in {@link DistributedDataStoreClient}.
37 * IMPORTANT: this class breaks actor containment via methods implementing {@link DistributedDataStoreClient} contract.
38 * When touching internal state, be mindful of the execution context from which execution context, Actor
39 * or POJO, is the state being accessed or modified.
42 * THREAD SAFETY: this class must always be kept thread-safe, so that both the Actor System thread and the application
43 * threads can run concurrently. All state transitions must be made in a thread-safe manner. When in
44 * doubt, feel free to synchronize on this object.
47 * PERFORMANCE: this class lies in a performance-critical fast path. All code needs to be concise and efficient, but
48 * performance must not come at the price of correctness. Any optimizations need to be carefully analyzed
49 * for correctness and performance impact.
52 * TRADE-OFFS: part of the functionality runs in application threads without switching contexts, which makes it ideal
53 * for performing work and charging applications for it. That has two positive effects:
54 * - CPU usage is distributed across applications, minimizing work done in the actor thread
55 * - CPU usage provides back-pressure towards the application.
57 * @author Robert Varga
59 final class DistributedDataStoreClientBehavior extends ClientActorBehavior implements DistributedDataStoreClient {
60 private static final Logger LOG = LoggerFactory.getLogger(DistributedDataStoreClientBehavior.class);
62 private final Map<TransactionIdentifier, ClientTransaction> transactions = new ConcurrentHashMap<>();
63 private final Map<LocalHistoryIdentifier, ClientLocalHistory> histories = new ConcurrentHashMap<>();
64 private final AtomicLong nextHistoryId = new AtomicLong(1);
65 private final AtomicLong nextTransactionId = new AtomicLong();
66 private final ModuleShardBackendResolver resolver;
67 private final SingleClientHistory singleHistory;
69 private volatile Throwable aborted;
71 DistributedDataStoreClientBehavior(final ClientActorContext context, final ActorContext actorContext) {
73 resolver = new ModuleShardBackendResolver(context.getIdentifier(), actorContext);
74 singleHistory = new SingleClientHistory(this, new LocalHistoryIdentifier(getIdentifier(), 0));
79 // Methods below are invoked from the client actor thread
84 protected void haltClient(final Throwable cause) {
85 // If we have encountered a previous problem there is not cleanup necessary, as we have already cleaned up
86 // Thread safely is not an issue, as both this method and any failures are executed from the same (client actor)
88 if (aborted != null) {
89 abortOperations(cause);
93 private void abortOperations(final Throwable cause) {
94 // This acts as a barrier, application threads check this after they have added an entry in the maps,
95 // and if they observe aborted being non-null, they will perform their cleanup and not return the handle.
98 for (ClientLocalHistory h : histories.values()) {
103 for (ClientTransaction t : transactions.values()) {
106 transactions.clear();
109 private DistributedDataStoreClientBehavior shutdown(final ClientActorBehavior currentBehavior) {
110 abortOperations(new IllegalStateException("Client " + getIdentifier() + " has been shut down"));
115 protected DistributedDataStoreClientBehavior onCommand(final Object command) {
116 if (command instanceof GetClientRequest) {
117 ((GetClientRequest) command).getReplyTo().tell(new Status.Success(this), ActorRef.noSender());
119 LOG.warn("{}: ignoring unhandled command {}", persistenceId(), command);
127 // Methods below are invoked from application threads
131 @SuppressWarnings("checkstyle:IllegalCatch")
132 private static <K, V extends LocalAbortable> V returnIfOperational(final Map<K , V> map, final K key, final V value,
133 final Throwable aborted) {
134 Verify.verify(map.put(key, value) == null);
136 if (aborted != null) {
138 value.localAbort(aborted);
139 } catch (Exception e) {
140 LOG.debug("Close of {} failed", value, e);
142 map.remove(key, value);
143 throw Throwables.propagate(aborted);
150 public ClientLocalHistory createLocalHistory() {
151 final LocalHistoryIdentifier historyId = new LocalHistoryIdentifier(getIdentifier(),
152 nextHistoryId.getAndIncrement());
153 final ClientLocalHistory history = new ClientLocalHistory(this, historyId);
154 LOG.debug("{}: creating a new local history {}", persistenceId(), history);
156 return returnIfOperational(histories, historyId, history, aborted);
160 public ClientTransaction createTransaction() {
161 final TransactionIdentifier txId = new TransactionIdentifier(singleHistory.getIdentifier(),
162 nextTransactionId.getAndIncrement());
163 final ClientTransaction tx = new ClientTransaction(singleHistory, txId);
164 LOG.debug("{}: creating a new transaction {}", persistenceId(), tx);
166 return returnIfOperational(transactions, txId, tx, aborted);
170 public void close() {
171 context().executeInActor(this::shutdown);
175 protected ModuleShardBackendResolver resolver() {
179 void transactionComplete(final ClientTransaction transaction) {
180 transactions.remove(transaction.getIdentifier());
183 void sendRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> completer) {
184 sendRequest(request, response -> {
185 completer.accept(response);