bucket.asBlocking().tryConsume(t, Duration.ofSeconds(20)) sometimes does not wait

[PeterTeunissen]

I am playing with this library and try to understand how I can use it. I wrote a simple restController in a SpringBoot app (see below) to try the basic operations on a bucket.

I am especially interested in the tryConsume and addTokens combination. I was assuming the bucket.asBlocking().tryConsume(t, Duration.ofSeconds(20)) method would wait at least 20 seconds to see if a token becomes available (either from some process calling addToken, or a refillInterval happening). But I see cases where this method returns immediately with value false.

With my example below, I call my 'consume' method 5 times and on 6th I get false (expected).
Then I call blockConsume and it sometimes blocks correctly but also sometimes returns immediately.

What am I missing in my logic?

@RestController
public class APIController {

  private Bucket b;

  @PostConstruct
  public void init() {
    b = Bucket.builder()
        .addLimit(limit -> limit.capacity(5).refillIntervally(1, Duration.ofSeconds(30))).build();
  }

  @GetMapping("/howmany")
  public String howMany() {
    return "" + b.getAvailableTokens();
  }

  @GetMapping("/returnItem")
  public String returnItem(@RequestParam int t) {
    b.addTokens(t);
    return howMany();
  }


  @GetMapping("/consume")
  public String consume(@RequestParam int t) {
    return "" + b.tryConsumeAndReturnRemaining(t);
  }

  @GetMapping("/blockConsume")
  public String blockConsume(@RequestParam int t) throws InterruptedException {
    boolean s = b.asBlocking().tryConsume(t, Duration.ofSeconds(20));
    return "" + s + " " + howMany();
  }

}

[vladimir-bukhtoyarov]

@PeterTeunissen hello,

I was assuming the bucket.asBlocking().tryConsume(t, Duration.ofSeconds(20)) method would wait at least 20 seconds to see if a token becomes available (either from some process calling addToken, or a refillInterval happening).

As mentioneed in javadocs asBlocking().tryConsume() operates only with refill time requires to close deficit. If it is known that specified waitThreshold is not enough to refill requested amount of tokens then it returns false imediatelly. Expectation that asBlocking().tryConsume interracts with addToken and wise versa, is not true.

Also it needs to understand that if thread is became blocked, then it is guaranteed that this thread will consume requested tokens, no one thread can steal the tokens while thread is being parked, this is achived by reservation - consuming even if available tokens become less than zero. Try to run code bellow:

import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

public class Example {

    public static void main(String[] args) throws InterruptedException {
        Bucket bucket = Bucket.builder()
            .addLimit(limit -> limit.capacity(100).refillGreedy (10, Duration.ofSeconds(1)))
            .build();
        bucket = addListener(bucket);

        while (true) {
            int toConsume = ThreadLocalRandom.current().nextInt(99) + 1;
            if (bucket.asBlocking().tryConsume(toConsume, Duration.ofSeconds(60))) {
                System.out.println("Consumed " + toConsume + " tokens. Available " + bucket.getAvailableTokens() + " tokens.");
            } else {
                break;
            }
        }
    }

    private static Bucket addListener(Bucket bucket) {
        return bucket.toListenable(new SimpleBucketListener() {
            @Override
            public void beforeParking(long nanos) {
                System.out.println("Available tokens before parking " + bucket.getAvailableTokens() + ". Is going to be parked for " + TimeUnit.NANOSECONDS.toMillis(nanos) + " millis");
                super.beforeParking(nanos);
            }
        });
    }

}

According to your code - it should return false in aproximatelly 1/3 (30-20/30) cases when trying in consume when no tokens.

[PeterTeunissen]

I can understand the logic that if I only want to wait 20 second and no new tokens will be available in those 20 seconds, the API returns false immediately, as an optimization.

I am struggling to understand the negative number.

If I have two containers that both call the same rest end-point which is limited at e.g. 1600 calls/minute. Both containers receive load from our application and one container can be more or less busy than the other. So it is possible that one container consumes 1500 calls in a minute which leaves only 100 for the other. How would I use bucket4j in a blocking way such that we don't exceed the combined 1600 calls/minute?

[vladimir-bukhtoyarov]

Negative amount of tokens does not lead to limits violation. This math trick just is used inside BlockingBucket for protection of requested tokens from to be stollen by parallel threads while current thread is being parked while awaiting to refill deficite. This style has zero algoritmic cost and pays nothing for coordination of parallel token's consumers, does not require concurrent or distributed collections.

Imagine bucket with capacity 3 and refill 1 token/1 second, and currently available tokens 1, and 5 parallel requests that come in same moment of time with request bucket.asBlocking().tryConsume(1, Duration.ofSeconds(3))

#	Tokens before execution	Execution result	Tokens after execution	Parking time
1	1	true	0	-
2	0	true	-1	1s
3	-1	true	-2	2s
4	-2	true	-3	3s
5	-3	false	-3	-

As you can see from last column of table, threads are parked for different time that is proportional to time for close deficit, there is no overconsumption.