opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/TransactionRateLimiter.java

   1 /*
   2  * Copyright (c) 2015 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 com.codahale.metrics.Snapshot;
  12 import com.codahale.metrics.Timer;
  13 import com.google.common.annotations.VisibleForTesting;
  14 import com.google.common.util.concurrent.RateLimiter;
  15 import java.util.concurrent.TimeUnit;
  16 import java.util.concurrent.atomic.AtomicLong;
  17 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 public class TransactionRateLimiter {
  22     private static final Logger LOG = LoggerFactory.getLogger(TransactionRateLimiter.class);
  23
  24     private final ActorUtils actorUtils;
  25     private final long commitTimeoutInSeconds;
  26     private final String dataStoreName;
  27     private final RateLimiter txRateLimiter;
  28     private final AtomicLong acquireCount = new AtomicLong();
  29
  30     private volatile long pollOnCount = 1;
  31
  32     public TransactionRateLimiter(ActorUtils actorUtils) {
  33         this.actorUtils = actorUtils;
  34         this.commitTimeoutInSeconds = actorUtils.getDatastoreContext().getShardTransactionCommitTimeoutInSeconds();
  35         this.dataStoreName = actorUtils.getDataStoreName();
  36         this.txRateLimiter = RateLimiter.create(actorUtils.getDatastoreContext()
  37                 .getTransactionCreationInitialRateLimit());
  38     }
  39
  40     public void acquire() {
  41         adjustRateLimit();
  42         txRateLimiter.acquire();
  43     }
  44
  45     private void adjustRateLimit() {
  46         final long count = acquireCount.incrementAndGet();
  47         if (count >= pollOnCount) {
  48             final Timer commitTimer = actorUtils.getOperationTimer(ActorUtils.COMMIT);
  49             double newRateLimit = calculateNewRateLimit(commitTimer, commitTimeoutInSeconds);
  50
  51             if (newRateLimit < 1.0) {
  52                 newRateLimit = getRateLimitFromOtherDataStores();
  53             }
  54
  55             if (newRateLimit >= 1.0) {
  56                 txRateLimiter.setRate(newRateLimit);
  57                 pollOnCount = count + (long) newRateLimit / 2;
  58             }
  59         }
  60     }
  61
  62     public double getTxCreationLimit() {
  63         return txRateLimiter.getRate();
  64     }
  65
  66     private double getRateLimitFromOtherDataStores() {
  67         // Since we have no rate data for unused Tx's data store, adjust to the rate from another
  68         // data store that does have rate data.
  69         for (String name: DatastoreContext.getGlobalDatastoreNames()) {
  70             if (name.equals(this.dataStoreName)) {
  71                 continue;
  72             }
  73
  74             double newRateLimit = calculateNewRateLimit(actorUtils.getOperationTimer(name, ActorUtils.COMMIT),
  75                     this.commitTimeoutInSeconds);
  76             if (newRateLimit > 0.0) {
  77                 LOG.debug("On unused Tx - data Store {} commit rateLimit adjusted to {}",
  78                         this.dataStoreName, newRateLimit);
  79
  80                 return newRateLimit;
  81             }
  82         }
  83
  84         return -1.0D;
  85     }
  86
  87     private static double calculateNewRateLimit(Timer commitTimer, long commitTimeoutInSeconds) {
  88         if (commitTimer == null) {
  89             // This can happen in unit tests.
  90             return 0;
  91         }
  92
  93         Snapshot timerSnapshot = commitTimer.getSnapshot();
  94         double newRateLimit = 0;
  95
  96         long commitTimeoutInNanos = TimeUnit.SECONDS.toNanos(commitTimeoutInSeconds);
  97
  98         // Find the time that it takes for transactions to get executed in every 10th percentile
  99         // Compute the rate limit for that percentile and sum it up
 100         for (int i = 1; i <= 10; i++) {
 101             // Get the amount of time transactions take in the i*10th percentile
 102             double percentileTimeInNanos = timerSnapshot.getValue(i * 0.1D);
 103
 104             if (percentileTimeInNanos > 0) {
 105                 // Figure out the rate limit for the i*10th percentile in nanos
 106                 double percentileRateLimit = commitTimeoutInNanos / percentileTimeInNanos;
 107
 108                 // Add the percentileRateLimit to the total rate limit
 109                 newRateLimit += percentileRateLimit;
 110             }
 111         }
 112
 113         // Compute the rate limit per second
 114         return newRateLimit / (commitTimeoutInSeconds * 10);
 115     }
 116
 117     @VisibleForTesting
 118     long getPollOnCount() {
 119         return pollOnCount;
 120     }
 121
 122     @VisibleForTesting
 123     void setPollOnCount(long value) {
 124         pollOnCount = value;
 125     }
 126
 127     @VisibleForTesting
 128     void setAcquireCount(long value) {
 129         acquireCount.set(value);
 130     }
 131 }