This repository has been archived by the owner on Nov 9, 2017. It is now read-only.
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
rhbz988202 - new leaky bucket to control logging volumn and refactor …
…to use single processor for all requests
- Loading branch information
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| package org.zanata.limits; | ||
|
|
||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import com.google.common.base.Objects; | ||
| import com.google.common.base.Ticker; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| /** | ||
| * @author Patrick Huang <a | ||
| * href="mailto:pahuang@redhat.com">pahuang@redhat.com</a> | ||
| */ | ||
| @Slf4j | ||
| public class LeakyBucket { | ||
| private final long refillPeriod; | ||
| private final long capacity; | ||
| private final Ticker ticker; | ||
| private volatile long permit; | ||
| private volatile long lastRead; | ||
|
|
||
| /** | ||
| * Simple form leaky bucket. Initialized with a capacity and full. Each | ||
| * #tryAcquire() will deduct 1 permit. Permits is refilled on demand after | ||
| * set time period. | ||
| * | ||
| * @param capacity | ||
| * capacity | ||
| * @param refillDuration | ||
| * refill duration | ||
| * @param refillTimeUnit | ||
| * refill time unit | ||
| */ | ||
| public LeakyBucket(long capacity, int refillDuration, | ||
| TimeUnit refillTimeUnit) { | ||
| this.capacity = capacity; | ||
| permit = capacity; | ||
| ticker = Ticker.systemTicker(); | ||
| refillPeriod = | ||
| TimeUnit.NANOSECONDS.convert(refillDuration, refillTimeUnit); | ||
| } | ||
|
|
||
| public synchronized boolean tryAcquire() { | ||
| onDemandRefill(); | ||
| if (permit > 0) { | ||
| log.debug("deduct 1 permit and try acquire return true"); | ||
| lastRead = ticker.read(); | ||
| permit--; | ||
| return true; | ||
| } else { | ||
| return false; | ||
| } | ||
| } | ||
|
|
||
| private synchronized void onDemandRefill() { | ||
| if (permit == capacity) { | ||
| return; | ||
| } | ||
| long timePassed = ticker.read() - lastRead; | ||
| log.debug("time passed: {}", timePassed); | ||
| long permitsShouldAdd = timePassed / refillPeriod; | ||
| log.debug("permits should add: {}", permitsShouldAdd); | ||
| if (timePassed >= refillPeriod) { | ||
| permit = Math.min(capacity, permit + permitsShouldAdd); | ||
| log.debug("refilled and now with {} permits", permit); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| // @formatter:off | ||
| return Objects.toStringHelper(this) | ||
| .add("refillPeriod", refillPeriod) | ||
| .add("capacity", capacity) | ||
| .add("permit", permit) | ||
| .add("lastRead", lastRead) | ||
| .toString(); | ||
| // @formatter:on | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package org.zanata.limits; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.concurrent.Callable; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| import org.apache.log4j.Level; | ||
| import org.apache.log4j.LogManager; | ||
| import org.hamcrest.Matchers; | ||
| import org.testng.annotations.BeforeMethod; | ||
| import org.testng.annotations.Test; | ||
| import com.google.common.base.Function; | ||
| import com.google.common.base.Throwables; | ||
| import com.google.common.collect.Lists; | ||
| import com.google.common.util.concurrent.Uninterruptibles; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import static org.hamcrest.MatcherAssert.assertThat; | ||
|
|
||
| /** | ||
| * @author Patrick Huang <a | ||
| * href="mailto:pahuang@redhat.com">pahuang@redhat.com</a> | ||
| */ | ||
| @Test(groups = "unit-tests") | ||
| @Slf4j | ||
| public class LeakyBucketTest { | ||
| @BeforeMethod | ||
| public void beforeMethod() { | ||
| // LogManager.getLogger(LeakyBucket.class.getPackage().getName()) | ||
| // .setLevel(Level.DEBUG); | ||
| } | ||
|
|
||
| @Test | ||
| public void willWaitUntilRefill() { | ||
| int refillDuration = 20; | ||
| TimeUnit refillTimeUnit = TimeUnit.MILLISECONDS; | ||
| LeakyBucket bucket = new LeakyBucket(1, refillDuration, refillTimeUnit); | ||
|
|
||
| assertThat(bucket.tryAcquire(), Matchers.is(true)); | ||
| assertThat(bucket.tryAcquire(), Matchers.is(false)); | ||
|
|
||
| Uninterruptibles.sleepUninterruptibly(refillDuration, refillTimeUnit); | ||
|
|
||
| assertThat(bucket.tryAcquire(), Matchers.is(true)); | ||
| } | ||
|
|
||
| @Test | ||
| public void concurrentTest() throws InterruptedException { | ||
| int refillDuration = 10; | ||
| TimeUnit refillTimeUnit = TimeUnit.MILLISECONDS; | ||
| final LeakyBucket bucket = | ||
| new LeakyBucket(1, refillDuration, refillTimeUnit); | ||
| Callable<Boolean> callable = new Callable<Boolean>() { | ||
|
|
||
| @Override | ||
| public Boolean call() throws Exception { | ||
| return bucket.tryAcquire(); | ||
| } | ||
| }; | ||
|
|
||
| int threads = 3; | ||
| List<Callable<Boolean>> callables = | ||
| Collections.nCopies(threads, callable); | ||
| ExecutorService executorService = Executors.newFixedThreadPool(threads); | ||
| List<Future<Boolean>> futures = executorService.invokeAll(callables); | ||
|
|
||
| List<Boolean> result = getFutureResult(futures); | ||
| assertThat(result, Matchers.containsInAnyOrder(true, false, false)); | ||
|
|
||
| // wait enough time and try again | ||
| Uninterruptibles.sleepUninterruptibly(refillDuration, refillTimeUnit); | ||
This comment has been minimized.
Sorry, something went wrong. 
seanf
Contributor
|
||
| List<Future<Boolean>> callAgain = executorService.invokeAll(callables); | ||
| assertThat(getFutureResult(callAgain), | ||
| Matchers.containsInAnyOrder(true, false, false)); | ||
| } | ||
|
|
||
| private static List<Boolean> getFutureResult(List<Future<Boolean>> futures) { | ||
| return Lists.transform(futures, | ||
| new Function<Future<Boolean>, Boolean>() { | ||
| @Override | ||
| public Boolean apply(Future<Boolean> input) { | ||
| try { | ||
| return input.get(); | ||
| } catch (Exception e) { | ||
| throw Throwables.propagate(e); | ||
| } | ||
| } | ||
| }); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
3 comments
on commit 74c811f
There was a problem hiding this comment.
@huangp Why didn't you just bring back https://github.com/bbeck/token-bucket ?
There was a problem hiding this comment.
First I try to avoid adding another dependency if I can. Second the refill in his implementation is not strictly time based(Not that it matters for this particular case since we only have 1 permit max). But I used to have a test for it:
TokenBucket bucket = TokenBuckets
.newFixedIntervalRefill(2, 1, 100, TimeUnit.MILLISECONDS);
bucket.tryConsume(); // true
bucket.tryConsume(); // false
// refill rate is 1 per 100 ms so after 200 ms it should've fill up.
Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
bucket.tryConsume(); // true
bucket.tryConsume(); // false. Refill will give you maximum 1 permit each call.If we have some other needs and needed to customize it, we will have to modify. Since it's a simple class I'd use our own copy of implementation.
There was a problem hiding this comment.
On 2014-03-27 09:25, Patrick Huang wrote:
First I try to avoid adding another dependency if I can. Second the
refill in his implementation is not strictly time based(Not that it
matters for this particular case since we only have 1 permit max). But I
used to have a test for it:
TokenBucket bucket = TokenBuckets
.newFixedIntervalRefill(2, 1, 100, TimeUnit.MILLISECONDS);
bucket.tryConsume(); // true
bucket.tryConsume(); // false
// refill rate is 1 per 100 ms so after 200 ms it should've fill up.
Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
bucket.tryConsume(); // true
bucket.tryConsume(); // false. Refill will give you maximum 1 permit each call.
If we have some other needs and needed to customize it, we will have to
modify. Since it's a simple class I'd use our own copy of implementation.
—
Reply to this email directly or view it on GitHub
74c811f#commitcomment-5812847.
Okay, but please add this as a test case for our token bucket.
##
Sean Flanigan
Senior Software Engineer
Engineering - Internationalisation
Red Hat
As I said for a previous commit, a single leaky bucket for all the log warnings would be fine. A cache keyed by api key would be nice, but this one is not configured to evict anything, ever.
Also, I don't think this needs to be static, now that there is a single RateLimitingProcessor.