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

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