Early throttling observed in cluster run with multiple instances of application

[techdeba901]

Hi Vladimir,

I am trying to run bucket4j(with coherence) with intervally aligned bucket configuration where tokens are fetched in batch. Batch value is calculated based on incoming message rate(thread calculating this value also aligned with
second). This is running in cluster environment where 20+ instances are running and processing message with same key.

Bucket configuration
rate: 60k TPS

Observation:
At 30k TPS for this message key, throttling starts ~@1.5k. It is not clear why this throttling is happening at such a low rate compared to configured rate in the bucket.

When I used refill strategy as greedy it's behavior improves a little and throttling starts @43k TPS for that key.

Attached the sample code I am using for this run.

Could you please help in understanding what possibly go wrong here?

Thanks in advance.

//Bucket configurations
public BucketConfiguration buildBucket(int rate, int durationInSec) {
		List<Bandwidth> bandwidths = new ArrayList<>();
		Instant instant = Instant.now().truncatedTo(ChronoUnit.SECONDS).plus(1, ChronoUnit.SECONDS);
		bandwidths.add(Bandwidth.classic(rate,
				Refill.intervallyAligned(rate, Duration.ofSeconds(durationInSec), instant, true)));
		ConfigurationBuilder bucketConfiguration = Bucket4j.configurationBuilder();
		bandwidths.stream().forEach(bw -> bucketConfiguration.addLimit(bw));
		return bucketConfiguration.build();
	}

	
//Cache Initialization	
private ProxyManager<String> bucketProxy;

	@PostConstruct
	private void intializeRateLimitService() {
		NamedCache cache = CacheFactory.getCache("testMap");
		cache.addIndex(new DeserializationAccelerator(IdentityExtractor.INSTANCE()), false, null);
		this.bucketProxy = Bucket4j.extension(Coherence.class).proxyManagerForCache(cache);
	}	


//Token fetch logic
public synchronized Boolean fetchToken(String bucketKey, Supplier<BucketConfiguration> configurationSupplier,
			int batch) {
		Long probe = 0L;

		// Try to access token from local map first
		AtomicInteger token = cacheMap.get(bucketKey);
		if (token != null && token.get() > 0) {
			if (token.get() == 1) {
				cacheMap.get(bucketKey).getAndSet(0);
			} else {
				cacheMap.get(bucketKey).decrementAndGet();
			}
			return true;
		}

		// Lets try to fetch token using bucket4j sync call with Coherence
		boolean tokenObtained = false;
		Bucket bucket = bucketProxy.getProxy(bucketKey, configurationSupplier);
		probe = bucket.tryConsumeAsMuchAsPossible(batch);

		if (probe > 0) {
			if (cacheMap.containsKey(bucketKey)) {
				cacheMap.get(bucketKey).addAndGet(probe.intValue() - 1);
			} else {
				cacheMap.put(bucketKey, new AtomicInteger(1));
			}
			tokenObtained = true;
		}
		return tokenObtained;
	}	
	
libraries used
--------------	
	
    implementation "com.oracle.coherence.ce:coherence:21.06.2"
    implementation "com.github.vladimir-bukhtoyarov:bucket4j-coherence:6.4.1"
    implementation "com.github.vladimir-bukhtoyarov:bucket4j-core:6.4.1"	

[vladimir-bukhtoyarov]

Off-top, I see that you use the highly loaded bucket. Today I have been released the 7.0.0 version that contains significant enhancement for such case #90 . I am very surprised to know that Coherence is able to process 30K TPS on same key, I did not use to believe in such numbers before. You can significantly reduce pressure to the Coherence cluster by migrating to 7.0.0 and migrating the code a bit:

        ...
        @PostConstruct
	private void intializeRateLimitService() {
		NamedCache cache = CacheFactory.getCache("testMap");
		cache.addIndex(new DeserializationAccelerator(IdentityExtractor.INSTANCE()), false, null);
		this.bucketProxy = new CoherenceManager(cache);
		
                // optimized bucket is not lightweith so it can be stored somewhere for reuse
                this.bucket = bucketProxy.builder()
                                          .withOptimization(Optimizations.batching())
                                          .build(bucketKey, configurationSupplier);
	}

[vladimir-bukhtoyarov]

Also this code can be used as example https://github.com/vladimir-bukhtoyarov/bucket4j/blob/master/bucket4j-examples/src/test/java/example/distributed/optimizers/batching/HazelcastWithBatchingPerformanceExample.java

[Buro45]

@vladimir-bukhtoyarov
In this case the bucket is created once in the following function and saved in a MAP (bucketMap which is member of the class) against the bucket identifier. From the incoming message bucket identifier is derived and used as key to get the Bucket object from the MAP.

When the 1st request comes bucket will be built for the first time and stored in the map. And then from 2nd request onwards bucket object will be retrieved from the map using bucket identifier.

public synchronized Bucket getBucket(String bucketIdentifier, Supplier<BucketConfiguration> configurationSupplier) {
    // This if check passed only one time
    if (!bucketMap.containsKey(bucketIdentifier)) {
		Bucket bucket = getBucketProxy().builder().withOptimization(Optimizations.batching())
				.build(bucketIdentifier, configurationSupplier);
		bucketMap.put(bucketIdentifier, bucket);
	}
	return bucketMap.get(bucketIdentifier);
}

Is this not correct way of reusing the bucket object ?

[vladimir-bukhtoyarov]

@Buro45 I missed that bucket is cached. It is the right way, I think that independent requests are not grouped into batches because of low request rate. In order to see the advantages of batching try to do the following:

• Make requests from several threads in parallel
• Make requests with the more frequent rate in order to the bring request intersection.

[techdeba901]

@vladimir-bukhtoyarov
I run with synchronous approach with greedy as refill strategy and run traffic in parallel. I had configured rate limit 60K TPS per key .

Synchronous approach:
At 40k TPS for this message key, throttling started to show up with 6-8 TPS.

code snippet :

Bucket bucket = getBucket(bucketIdentifier, configurationSupplier);
probe = bucket.tryConsume(1);

public synchronized Bucket getBucket(String bucketIdentifier, Supplier configurationSupplier) {
if (!bucketMap.containsKey(bucketIdentifier)) {
Bucket bucket = getBucketProxy().builder().withOptimization(Optimizations.batching())
.build(bucketIdentifier, configurationSupplier);
bucketMap.put(bucketIdentifier, bucket);
}
return bucketMap.get(bucketIdentifier);
}

Asynchronous approach:
When I run asynchronous approach with greedy as refill strategy and run traffic in parallel. I had configured rate limit 60K TPS per key.

At 16k TPS for this message key, I am observing memory leak. So traffic is proceeding further.

code snippet :

AsyncBucketProxy asyncBucket = getBucket(bucketIdentifier, configurationSupplier);
CompletableFuture isSuccess = asyncBucket.tryConsume(1);

public synchronized AsyncBucketProxy getBucket(String bucketIdentifier, Supplier configurationSupplier) {
if (!bucketMap.containsKey(bucketIdentifier)) {
AsyncBucketProxy asyncBucket = getBucketProxy().builder().withOptimization(Optimizations.batching())
.build(bucketIdentifier, configurationSupplier.get());
bucketMap.put(bucketIdentifier, asyncBucket);
}
return bucketMap.get(bucketIdentifier);
}

[vladimir-bukhtoyarov]

Regarding asynchronous approach - you did not get a memory leak, it was just OutOfMemoryError. It is not surprising when tasks are issued with rate that system is not ready to process. Look to my code, I used Semaphore to limit the number of incomplete tasks https://github.com/vladimir-bukhtoyarov/bucket4j/blob/master/bucket4j-examples/src/test/java/example/distributed/optimizers/batching/HazelcastWithBatchingPerformanceExample.java#L149

Regarding reproducing steps in synchronous mode: did you reproduce early throttling from a single thread?

[vladimir-bukhtoyarov]

@techdeba901 According to original question:
I want to start from following line of code

bandwidths.add(Bandwidth.classic(rate, Refill.intervallyAligned(rate, Duration.ofSeconds(durationInSec), instant, true)));

You invoke intervallyAligned with useAdaptiveInitialTokens=true. Was the javadocs clear for you? Do you still to view the problem if useAdaptiveInitialTokens=false?

[techdeba901]

Sorry. Updated the corrected file.

[vladimir-bukhtoyarov]

Also the name isTokenAvailable is not clear for me. Name of method tells that method will not modify something, but behind it modifies the state of bucket by calling tryConsumeAsMuchAsPossible. Why is not used estimateAbilityToConsume that does not modify the state of bucket?

[vladimir-bukhtoyarov]

@techdeba901

Updated the corrected file.

But code still containing cacheMap.put(bucketKey, new AtomicInteger(1)); instead of cacheMap.put(bucketKey, new AtomicInteger(tokenConsumed.intValue() - 1));

[techdeba901]

you are right. Updated the files.

[vladimir-bukhtoyarov]

@techdeba901 I still see new AtomicInteger(1)) at the attached file.

Anyway it is time to use Verbose API, to better understand what is going on:

public synchronized Boolean isTokenAvailable(String bucketKey,
                                                 Supplier<BucketConfiguration> configurationSupplier) {
        Long tokenConsumed = 0L;
        // Try to access token from local map first
        AtomicInteger token = cacheMap.get(bucketKey);
        if (token != null && token.get() > 0) {
            if (token.get() == 1) {
                cacheMap.get(bucketKey).getAndSet(0);
            } else {
                cacheMap.get(bucketKey).decrementAndGet();
            }
            return true;
        }
        // Lets try to fetch token using bucket4j sync call with Coherence
        boolean tokenObtained = false;
        try {
            Bucket bucket = bucketProxy.getProxy(bucketKey, configurationSupplier);
            /*
             * Here the token count is passed as 20. But in the original program this token
             * count is decided based on incoming message rate
             */
            Date myTime = new Date();
            VerboseResult<Long> tokenConsumedResult = bucket.asVerbose().tryConsumeAsMuchAsPossible(20);
            tokenConsumed = tokenConsumedResult.getValue();

            if (tokenConsumed > 0) {
                if (cacheMap.containsKey(bucketKey)) {
                    cacheMap.get(bucketKey).addAndGet(tokenConsumed.intValue() - 1);
                } else {
                    cacheMap.put(bucketKey, new AtomicInteger(tokenConsumed.intValue() - 1));
                }
                tokenObtained = true;
            } else {
                Date operationTimestampThatWasActualOnServer = new Date(tokenConsumedResult.getOperationTimeNanos() / 1_000_000);
                long nanosToRefillToOriginalState = tokenConsumedResult.getDiagnostics().calculateFullRefillingTime();
                logger.warn("Tokens was not consumed at server-timestamp " + operationTimestampThatWasActualOnServer + ". It was needed to wait " + nanosToRefillToOriginalState + 
                        " nanoseconds before full bucket refill, my timestamp was " + myTime);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return tokenObtained;
    }

Now you are able to see timestamp on node that holds the state of bucket and timestamp on node that got "0" as result of tryConsumeAsMuchAsPossible operation. Also message in log shows the time in nanoseconds that need to wait before bucket will be fully refilled to original state.

[techdeba901]

we observed from the source code of version 6 that async implementation of bucket depends on the CommonForkJoinPool which can be very less. Is there a way in bucket4j to pass your own executor pool. If not, is there a way to define parallelism for the async methods

[vladimir-bukhtoyarov]

@techdeba901 Bucket4j in the background does not use CommonForkJoinPool as well as any other thread pool in all versions.
However, it can be used on behalf of you when you use methods like thenApply/whenComplete and similar without passing own executor as an explicit parameter. Javadocs for each method of async API contains appropriate notice about that, for example https://github.com/vladimir-bukhtoyarov/bucket4j/blob/6.4/bucket4j-core/src/main/java/io/github/bucket4j/AsyncBucket.java#L53

I have debugged Coherence by itself in order to find where it executes aynch operations result callbacks, it always happens inside own Coherence threads

So if you see an usage of CommonForkJoinPool then double-check what you do with CompleteableFutures.

[techdeba901]

Thanks Vladimir for your inputs.

Tokens are added in 100ms when configured rate is 1000 per second (i.e. 100 tokens in every 100ms) when refill strategy is greedy.
But if consumption of tokens is more than 100 per 100ms will tokens be added more greedily(more frequently).

In tryConsumeAndReturnRemaining, when we use this method it return following 3 variables(remainingTokens , nanosToWaitForRefill, nanosToWaitForReset) in version 7.1.0

What is the difference between nanosToWaitForRefill & nanosToWaitForReset?
Assume the tokens are exhausted @800ms within a second boundary.
In that case, does nanosToWaitForReset will be 200ms?
Could you please provide your input?

[vladimir-bukhtoyarov]

The next refill is allowed to happen after 1ms, because refill rate 1000 tokens/1 second fully equals 1 token/1 millisecond.
The consumption rate is not a the part of the TokenBucket algorithm, if the token is allowed to be refilled according to the refill setting then it will be refilled independently of the current consumption rate.

[techdeba901]

In Greedy, is there a way to define refill interval instead of keeping it 1ms without using more than one BW? For example, I want 5000/sec but refill should happen at every 100 ms

[vladimir-bukhtoyarov]

@techdeba901 you should switch to interval refill in such case. For example:

Bandwidth.classic(5000, Refill.intervally(500, Duration.ofMillis(100)));

[techdeba901]

Considering above example, i.e. refill 500 per 100ms, at any point of time if token get exhausted then next refill will happen at the nearest 100ms timestamp. Hence, nanosToWaitForRefill will represent the time remaining to the next 100ms boundary when next refill will happen. Is this a correct understanding?

[vladimir-bukhtoyarov]

Yes, it is how the intervally refill works.