opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionRateLimitingCallback.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;
  10
  11 import com.codahale.metrics.Timer;
  12 import com.google.common.annotations.VisibleForTesting;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.base.Ticker;
  15 import java.util.concurrent.TimeUnit;
  16 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  17
  18 /**
  19  * TransactionRateLimitingCallback computes the new transaction rate limit on the successful completion of a
  20  * transaction.
  21  */
  22 public class TransactionRateLimitingCallback implements OperationCallback {
  23     private static Ticker TICKER = Ticker.systemTicker();
  24
  25     private enum State {
  26         STOPPED,
  27         RUNNING,
  28         PAUSED
  29     }
  30
  31     private final Timer commitTimer;
  32     private long startTime;
  33     private long elapsedTime;
  34     private volatile State state = State.STOPPED;
  35
  36     TransactionRateLimitingCallback(ActorUtils actorUtils) {
  37         commitTimer = actorUtils.getOperationTimer(ActorUtils.COMMIT);
  38     }
  39
  40     @Override
  41     public void run() {
  42         Preconditions.checkState(state == State.STOPPED, "state is not STOPPED");
  43         resume();
  44     }
  45
  46     @Override
  47     public void pause() {
  48         if (state == State.RUNNING) {
  49             elapsedTime += TICKER.read() - startTime;
  50             state = State.PAUSED;
  51         }
  52     }
  53
  54     @Override
  55     public void resume() {
  56         if (state != State.RUNNING) {
  57             startTime = TICKER.read();
  58             state = State.RUNNING;
  59         }
  60     }
  61
  62     @Override
  63     public void success() {
  64         Preconditions.checkState(state != State.STOPPED, "state is STOPPED");
  65         pause();
  66         commitTimer.update(elapsedTime, TimeUnit.NANOSECONDS);
  67         state = State.STOPPED;
  68     }
  69
  70     @Override
  71     public void failure() {
  72         // This would mean we couldn't get a transaction completed in 30 seconds which is
  73         // the default transaction commit timeout. Using the timeout information to figure out the rate limit is
  74         // not going to be useful - so we leave it as it is
  75     }
  76
  77     @VisibleForTesting
  78     static void setTicker(Ticker ticker) {
  79         TICKER = ticker;
  80     }
  81 }