CDS: Implement front-end support for local transactions
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / utils / ActorContext.java
1 /*
2  * Copyright (c) 2014 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.utils;
10
11 import static akka.pattern.Patterns.ask;
12 import akka.actor.ActorPath;
13 import akka.actor.ActorRef;
14 import akka.actor.ActorSelection;
15 import akka.actor.ActorSystem;
16 import akka.actor.Address;
17 import akka.actor.PoisonPill;
18 import akka.dispatch.Futures;
19 import akka.dispatch.Mapper;
20 import akka.dispatch.OnComplete;
21 import akka.pattern.AskTimeoutException;
22 import akka.util.Timeout;
23 import com.codahale.metrics.MetricRegistry;
24 import com.codahale.metrics.Timer;
25 import com.google.common.annotations.VisibleForTesting;
26 import com.google.common.base.Optional;
27 import com.google.common.base.Preconditions;
28 import com.google.common.base.Strings;
29 import com.google.common.cache.Cache;
30 import com.google.common.cache.CacheBuilder;
31 import com.google.common.cache.RemovalListener;
32 import com.google.common.cache.RemovalNotification;
33 import com.google.common.util.concurrent.RateLimiter;
34 import java.util.ArrayList;
35 import java.util.Collection;
36 import java.util.concurrent.TimeUnit;
37 import javax.annotation.concurrent.GuardedBy;
38 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
39 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
40 import org.opendaylight.controller.cluster.datastore.Configuration;
41 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
42 import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
43 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
44 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
45 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
46 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
47 import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
48 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
49 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
50 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
51 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
52 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
53 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
54 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
55 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
56 import org.opendaylight.controller.cluster.reporting.MetricsReporter;
57 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
58 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
59 import org.slf4j.Logger;
60 import org.slf4j.LoggerFactory;
61 import scala.concurrent.Await;
62 import scala.concurrent.ExecutionContext;
63 import scala.concurrent.Future;
64 import scala.concurrent.duration.Duration;
65 import scala.concurrent.duration.FiniteDuration;
66
67 /**
68  * The ActorContext class contains utility methods which could be used by
69  * non-actors (like DistributedDataStore) to work with actors a little more
70  * easily. An ActorContext can be freely passed around to local object instances
71  * but should not be passed to actors especially remote actors
72  */
73 public class ActorContext implements RemovalListener<String, Future<PrimaryShardInfo>> {
74     private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class);
75     private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store";
76     private static final String METRIC_RATE = "rate";
77     private static final Mapper<Throwable, Throwable> FIND_PRIMARY_FAILURE_TRANSFORMER =
78                                                               new Mapper<Throwable, Throwable>() {
79         @Override
80         public Throwable apply(Throwable failure) {
81             Throwable actualFailure = failure;
82             if(failure instanceof AskTimeoutException) {
83                 // A timeout exception most likely means the shard isn't initialized.
84                 actualFailure = new NotInitializedException(
85                         "Timed out trying to find the primary shard. Most likely cause is the " +
86                         "shard is not initialized yet.");
87             }
88
89             return actualFailure;
90         }
91     };
92     public static final String MAILBOX = "bounded-mailbox";
93
94     private final ActorSystem actorSystem;
95     private final ActorRef shardManager;
96     private final ClusterWrapper clusterWrapper;
97     private final Configuration configuration;
98     private DatastoreContext datastoreContext;
99     private FiniteDuration operationDuration;
100     private Timeout operationTimeout;
101     private final String selfAddressHostPort;
102     private RateLimiter txRateLimiter;
103     private final int transactionOutstandingOperationLimit;
104     private Timeout transactionCommitOperationTimeout;
105     private Timeout shardInitializationTimeout;
106     private final Dispatchers dispatchers;
107     private Cache<String, Future<PrimaryShardInfo>> primaryShardInfoCache;
108
109     private volatile SchemaContext schemaContext;
110     private volatile boolean updated;
111     private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry();
112     @GuardedBy("shardInfoListeners")
113     private final Collection<ShardInfoListenerRegistration<?>> shardInfoListeners = new ArrayList<>();
114
115     public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
116             ClusterWrapper clusterWrapper, Configuration configuration) {
117         this(actorSystem, shardManager, clusterWrapper, configuration,
118                 DatastoreContext.newBuilder().build());
119     }
120
121     public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
122             ClusterWrapper clusterWrapper, Configuration configuration,
123             DatastoreContext datastoreContext) {
124         this.actorSystem = actorSystem;
125         this.shardManager = shardManager;
126         this.clusterWrapper = clusterWrapper;
127         this.configuration = configuration;
128         this.datastoreContext = datastoreContext;
129         this.dispatchers = new Dispatchers(actorSystem.dispatchers());
130
131         setCachedProperties();
132
133         Address selfAddress = clusterWrapper.getSelfAddress();
134         if (selfAddress != null && !selfAddress.host().isEmpty()) {
135             selfAddressHostPort = selfAddress.host().get() + ":" + selfAddress.port().get();
136         } else {
137             selfAddressHostPort = null;
138         }
139
140         transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity();
141     }
142
143     private void setCachedProperties() {
144         txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit());
145
146         operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS);
147         operationTimeout = new Timeout(operationDuration);
148
149         transactionCommitOperationTimeout =  new Timeout(Duration.create(
150                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS));
151
152         shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2));
153
154         primaryShardInfoCache = CacheBuilder.newBuilder()
155                 .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
156                 .removalListener(this)
157                 .build();
158     }
159
160     public DatastoreContext getDatastoreContext() {
161         return datastoreContext;
162     }
163
164     public ActorSystem getActorSystem() {
165         return actorSystem;
166     }
167
168     public ActorRef getShardManager() {
169         return shardManager;
170     }
171
172     public ActorSelection actorSelection(String actorPath) {
173         return actorSystem.actorSelection(actorPath);
174     }
175
176     public ActorSelection actorSelection(ActorPath actorPath) {
177         return actorSystem.actorSelection(actorPath);
178     }
179
180     public void setSchemaContext(SchemaContext schemaContext) {
181         this.schemaContext = schemaContext;
182
183         if(shardManager != null) {
184             shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
185         }
186     }
187
188     public void setDatastoreContext(DatastoreContext context) {
189         this.datastoreContext = context;
190         setCachedProperties();
191
192         // We write the 'updated' volatile to trigger a write memory barrier so that the writes above
193         // will be published immediately even though they may not be immediately visible to other
194         // threads due to unsynchronized reads. That's OK though - we're going for eventual
195         // consistency here as immediately visible updates to these members aren't critical. These
196         // members could've been made volatile but wanted to avoid volatile reads as these are
197         // accessed often and updates will be infrequent.
198
199         updated = true;
200
201         if(shardManager != null) {
202             shardManager.tell(context, ActorRef.noSender());
203         }
204     }
205
206     public SchemaContext getSchemaContext() {
207         return schemaContext;
208     }
209
210     public Future<PrimaryShardInfo> findPrimaryShardAsync(final String shardName) {
211         Future<PrimaryShardInfo> ret = primaryShardInfoCache.getIfPresent(shardName);
212         if(ret != null){
213             return ret;
214         }
215         Future<Object> future = executeOperationAsync(shardManager,
216                 new FindPrimary(shardName, true), shardInitializationTimeout);
217
218         return future.transform(new Mapper<Object, PrimaryShardInfo>() {
219             @Override
220             public PrimaryShardInfo checkedApply(Object response) throws Exception {
221                 if(response instanceof RemotePrimaryShardFound) {
222                     LOG.debug("findPrimaryShardAsync received: {}", response);
223                     return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null);
224                 } else if(response instanceof LocalPrimaryShardFound) {
225                     LOG.debug("findPrimaryShardAsync received: {}", response);
226                     LocalPrimaryShardFound found = (LocalPrimaryShardFound)response;
227                     return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getLocalShardDataTree());
228                 } else if(response instanceof NotInitializedException) {
229                     throw (NotInitializedException)response;
230                 } else if(response instanceof PrimaryNotFoundException) {
231                     throw (PrimaryNotFoundException)response;
232                 } else if(response instanceof NoShardLeaderException) {
233                     throw (NoShardLeaderException)response;
234                 }
235
236                 throw new UnknownMessageException(String.format(
237                         "FindPrimary returned unkown response: %s", response));
238             }
239         }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher());
240     }
241
242     private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath,
243             DataTree localShardDataTree) {
244         ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath);
245         PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree));
246         primaryShardInfoCache.put(shardName, Futures.successful(info));
247
248         synchronized (shardInfoListeners) {
249             for (ShardInfoListenerRegistration<?> reg : shardInfoListeners) {
250                 reg.getInstance().onShardInfoUpdated(shardName, info);
251             }
252         }
253         return info;
254     }
255
256     /**
257      * Finds a local shard given its shard name and return it's ActorRef
258      *
259      * @param shardName the name of the local shard that needs to be found
260      * @return a reference to a local shard actor which represents the shard
261      *         specified by the shardName
262      */
263     public Optional<ActorRef> findLocalShard(String shardName) {
264         Object result = executeOperation(shardManager, new FindLocalShard(shardName, false));
265
266         if (result instanceof LocalShardFound) {
267             LocalShardFound found = (LocalShardFound) result;
268             LOG.debug("Local shard found {}", found.getPath());
269             return Optional.of(found.getPath());
270         }
271
272         return Optional.absent();
273     }
274
275     /**
276      * Finds a local shard async given its shard name and return a Future from which to obtain the
277      * ActorRef.
278      *
279      * @param shardName the name of the local shard that needs to be found
280      */
281     public Future<ActorRef> findLocalShardAsync( final String shardName) {
282         Future<Object> future = executeOperationAsync(shardManager,
283                 new FindLocalShard(shardName, true), shardInitializationTimeout);
284
285         return future.map(new Mapper<Object, ActorRef>() {
286             @Override
287             public ActorRef checkedApply(Object response) throws Throwable {
288                 if(response instanceof LocalShardFound) {
289                     LocalShardFound found = (LocalShardFound)response;
290                     LOG.debug("Local shard found {}", found.getPath());
291                     return found.getPath();
292                 } else if(response instanceof NotInitializedException) {
293                     throw (NotInitializedException)response;
294                 } else if(response instanceof LocalShardNotFound) {
295                     throw new LocalShardNotFoundException(
296                             String.format("Local shard for %s does not exist.", shardName));
297                 }
298
299                 throw new UnknownMessageException(String.format(
300                         "FindLocalShard returned unkown response: %s", response));
301             }
302         }, getClientDispatcher());
303     }
304
305     /**
306      * Executes an operation on a local actor and wait for it's response
307      *
308      * @param actor
309      * @param message
310      * @return The response of the operation
311      */
312     public Object executeOperation(ActorRef actor, Object message) {
313         Future<Object> future = executeOperationAsync(actor, message, operationTimeout);
314
315         try {
316             return Await.result(future, operationDuration);
317         } catch (Exception e) {
318             throw new TimeoutException("Sending message " + message.getClass().toString() +
319                     " to actor " + actor.toString() + " failed. Try again later.", e);
320         }
321     }
322
323     public Future<Object> executeOperationAsync(ActorRef actor, Object message, Timeout timeout) {
324         Preconditions.checkArgument(actor != null, "actor must not be null");
325         Preconditions.checkArgument(message != null, "message must not be null");
326
327         LOG.debug("Sending message {} to {}", message.getClass(), actor);
328         return doAsk(actor, message, timeout);
329     }
330
331     /**
332      * Execute an operation on a remote actor and wait for it's response
333      *
334      * @param actor
335      * @param message
336      * @return
337      */
338     public Object executeOperation(ActorSelection actor, Object message) {
339         Future<Object> future = executeOperationAsync(actor, message);
340
341         try {
342             return Await.result(future, operationDuration);
343         } catch (Exception e) {
344             throw new TimeoutException("Sending message " + message.getClass().toString() +
345                     " to actor " + actor.toString() + " failed. Try again later.", e);
346         }
347     }
348
349     /**
350      * Execute an operation on a remote actor asynchronously.
351      *
352      * @param actor the ActorSelection
353      * @param message the message to send
354      * @param timeout the operation timeout
355      * @return a Future containing the eventual result
356      */
357     public Future<Object> executeOperationAsync(ActorSelection actor, Object message,
358             Timeout timeout) {
359         Preconditions.checkArgument(actor != null, "actor must not be null");
360         Preconditions.checkArgument(message != null, "message must not be null");
361
362         LOG.debug("Sending message {} to {}", message.getClass(), actor);
363
364         return doAsk(actor, message, timeout);
365     }
366
367     /**
368      * Execute an operation on a remote actor asynchronously.
369      *
370      * @param actor the ActorSelection
371      * @param message the message to send
372      * @return a Future containing the eventual result
373      */
374     public Future<Object> executeOperationAsync(ActorSelection actor, Object message) {
375         return executeOperationAsync(actor, message, operationTimeout);
376     }
377
378     /**
379      * Sends an operation to be executed by a remote actor asynchronously without waiting for a
380      * reply (essentially set and forget).
381      *
382      * @param actor the ActorSelection
383      * @param message the message to send
384      */
385     public void sendOperationAsync(ActorSelection actor, Object message) {
386         Preconditions.checkArgument(actor != null, "actor must not be null");
387         Preconditions.checkArgument(message != null, "message must not be null");
388
389         LOG.debug("Sending message {} to {}", message.getClass(), actor);
390
391         actor.tell(message, ActorRef.noSender());
392     }
393
394     public void shutdown() {
395         shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender());
396     }
397
398     public ClusterWrapper getClusterWrapper() {
399         return clusterWrapper;
400     }
401
402     public String getCurrentMemberName(){
403         return clusterWrapper.getCurrentMemberName();
404     }
405
406     /**
407      * Send the message to each and every shard
408      *
409      * @param message
410      */
411     public void broadcast(final Object message){
412         for(final String shardName : configuration.getAllShardNames()){
413
414             Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName);
415             primaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
416                 @Override
417                 public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) {
418                     if(failure != null) {
419                         LOG.warn("broadcast failed to send message {} to shard {}:  {}",
420                                 message.getClass().getSimpleName(), shardName, failure);
421                     } else {
422                         primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender());
423                     }
424                 }
425             }, getClientDispatcher());
426         }
427     }
428
429     public FiniteDuration getOperationDuration() {
430         return operationDuration;
431     }
432
433     public boolean isPathLocal(String path) {
434         if (Strings.isNullOrEmpty(path)) {
435             return false;
436         }
437
438         int pathAtIndex = path.indexOf('@');
439         if (pathAtIndex == -1) {
440             //if the path is of local format, then its local and is co-located
441             return true;
442
443         } else if (selfAddressHostPort != null) {
444             // self-address and tx actor path, both are of remote path format
445             int slashIndex = path.indexOf('/', pathAtIndex);
446
447             if (slashIndex == -1) {
448                 return false;
449             }
450
451             String hostPort = path.substring(pathAtIndex + 1, slashIndex);
452             return hostPort.equals(selfAddressHostPort);
453
454         } else {
455             // self address is local format and tx actor path is remote format
456             return false;
457         }
458     }
459
460     /**
461      * @deprecated This method is present only to support backward compatibility with Helium and should not be
462      * used any further
463      *
464      *
465      * @param primaryPath
466      * @param localPathOfRemoteActor
467      * @return
468     */
469     @Deprecated
470     public String resolvePath(final String primaryPath,
471                                             final String localPathOfRemoteActor) {
472         StringBuilder builder = new StringBuilder();
473         String[] primaryPathElements = primaryPath.split("/");
474         builder.append(primaryPathElements[0]).append("//")
475             .append(primaryPathElements[1]).append(primaryPathElements[2]);
476         String[] remotePathElements = localPathOfRemoteActor.split("/");
477         for (int i = 3; i < remotePathElements.length; i++) {
478                 builder.append("/").append(remotePathElements[i]);
479             }
480
481         return builder.toString();
482     }
483
484     /**
485      * Get the maximum number of operations that are to be permitted within a transaction before the transaction
486      * should begin throttling the operations
487      *
488      * Parking reading this configuration here because we need to get to the actor system settings
489      *
490      * @return
491      */
492     public int getTransactionOutstandingOperationLimit(){
493         return transactionOutstandingOperationLimit;
494     }
495
496     /**
497      * This is a utility method that lets us get a Timer object for any operation. This is a little open-ended to allow
498      * us to create a timer for pretty much anything.
499      *
500      * @param operationName
501      * @return
502      */
503     public Timer getOperationTimer(String operationName){
504         return getOperationTimer(datastoreContext.getDataStoreType(), operationName);
505     }
506
507     public Timer getOperationTimer(String dataStoreType, String operationName){
508         final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, dataStoreType,
509                 operationName, METRIC_RATE);
510         return metricRegistry.timer(rate);
511     }
512
513     /**
514      * Get the type of the data store to which this ActorContext belongs
515      *
516      * @return
517      */
518     public String getDataStoreType() {
519         return datastoreContext.getDataStoreType();
520     }
521
522     /**
523      * Set the number of transaction creation permits that are to be allowed
524      *
525      * @param permitsPerSecond
526      */
527     public void setTxCreationLimit(double permitsPerSecond){
528         txRateLimiter.setRate(permitsPerSecond);
529     }
530
531     /**
532      * Get the current transaction creation rate limit
533      * @return
534      */
535     public double getTxCreationLimit(){
536         return txRateLimiter.getRate();
537     }
538
539     /**
540      * Try to acquire a transaction creation permit. Will block if no permits are available.
541      */
542     public void acquireTxCreationPermit(){
543         txRateLimiter.acquire();
544     }
545
546     /**
547      * Return the operation timeout to be used when committing transactions
548      * @return
549      */
550     public Timeout getTransactionCommitOperationTimeout(){
551         return transactionCommitOperationTimeout;
552     }
553
554     /**
555      * An akka dispatcher that is meant to be used when processing ask Futures which were triggered by client
556      * code on the datastore
557      * @return
558      */
559     public ExecutionContext getClientDispatcher() {
560         return this.dispatchers.getDispatcher(Dispatchers.DispatcherType.Client);
561     }
562
563     public String getNotificationDispatcherPath(){
564         return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification);
565     }
566
567     public Configuration getConfiguration() {
568         return configuration;
569     }
570
571     protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout){
572         return ask(actorRef, message, timeout);
573     }
574
575     protected Future<Object> doAsk(ActorSelection actorRef, Object message, Timeout timeout){
576         return ask(actorRef, message, timeout);
577     }
578
579     @VisibleForTesting
580     Cache<String, Future<PrimaryShardInfo>> getPrimaryShardInfoCache() {
581         return primaryShardInfoCache;
582     }
583
584     public <T extends ShardInfoListener> ShardInfoListenerRegistration<T> registerShardInfoListener(final T listener) {
585         final ShardInfoListenerRegistration<T> reg = new ShardInfoListenerRegistration<T>(listener, this);
586
587         synchronized (shardInfoListeners) {
588             shardInfoListeners.add(reg);
589         }
590         return reg;
591     }
592
593     protected void removeShardInfoListener(final ShardInfoListenerRegistration<?> registration) {
594         synchronized (shardInfoListeners) {
595             shardInfoListeners.remove(registration);
596         }
597     }
598
599     @Override
600     public void onRemoval(final RemovalNotification<String, Future<PrimaryShardInfo>> notification) {
601         synchronized (shardInfoListeners) {
602             for (ShardInfoListenerRegistration<?> reg : shardInfoListeners) {
603                 reg.getInstance().onShardInfoUpdated(notification.getKey(), null);
604             }
605         }
606     }
607 }

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