Provides a throttling strategy based on the number of pending blocking tasks.

[TheWeaVer]

Motivation:

Give a way to blocking request throttling strategy to prevent the situation that worsens rapidly

Modifications:

• Add new throttling strategy based on the giving limit size

Result:

• Closes Provides a throttling strategy based on the number of pending blocking tasks. #3829

[codecov]

Codecov Report

Attention: Patch coverage is 12.00000% with 44 lines in your changes are missing coverage. Please review.

Project coverage is 74.00%. Comparing base (bc2454a) to head (b94e0f4).
Report is 100 commits behind head on main.

❗ Current head b94e0f4 differs from pull request most recent head b9e34a1. Consider uploading reports for the commit b9e34a1 to get more accurate results

Files	Patch %	Lines
...ommon/util/DefaultLimitedBlockingTaskExecutor.java	0.00%	41 Missing ⚠️
...meria/common/util/LimitedBlockingTaskExecutor.java	0.00%	2 Missing ⚠️
.../armeria/server/throttling/ThrottlingStrategy.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##             main    #4073       +/-   ##
===========================================
+ Coverage        0   74.00%   +74.00%     
- Complexity      0    18099    +18099     
===========================================
  Files           0     1530     +1530     
  Lines           0    67141    +67141     
  Branches        0     8479     +8479     
===========================================
+ Hits            0    49686    +49686     
- Misses          0    13392    +13392     
- Partials        0     4063     +4063     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[trustin]

This PR has an inevitable race condition where the number of pending tasks can exceed the desired limit. How about providing an API that wraps a given Executor, ExecutorService or other well known executor-like interface like the following:

myWrappedExecutor = ThrottlingStrategy.wrap(myExecutor, 10 /* or myLimitSupplier */)

Server
  .builder()
  .blockingTaskExecutor(myWrappedExecutor)
  .decorator(myWrappedExecutor.asDecorator())
  // or
  .decorator(ThrottlingService.newDecorator(myWrappedExecutor.throttlingStrategy()))
  ...

[TheWeaVer]

Let me try! Thank you @trustin

[trustin]

@TheWeaVer Great! Let me stay tuned ❤️

[trustin]

Suggested change

	* {@link SettableIntSupplier} by comparing it to the size of the tasks of the {@link BlockingTaskExecutor}.
	* {@link IntSupplier} by comparing it to the size of the tasks of the {@link BlockingTaskExecutor}.

[trustin]

Could we also add a version that accepts an int instead of IntSupplier for a simpler use case where a user just wants to enforce a static limit?

[TheWeaVer]

I add new method 😄

[jrhee17]

Thanks for the initiative!

I think as it stands, it is difficult to keep sync between the executor state and our DefaultBlockingTaskExecutor.

What do you think of simply following micrometer's approach and try to access the number of queued tasks directly like suggested here: #4073 (comment)

For instance, I think it's fine to watch (ThreadPoolExecutor#getQueue#size, ForkJoinPool::getQueuedTaskCount) and throttle requests based on this value.
If an unsupported ScheduledExecutorService is used to throttle requests, I think we can just throw an exception on initialization.

Let me know what you think @line/dx

#4073 (comment)

Nevermind, I saw this comment late.

[jrhee17]

I think taskCounter will be decremented on each scheduled task invocation, but no increment occurs - this may not correctly reflect the number of pending tasks

[jrhee17]

If future.cancel is called immediately without given a chance to run, I think it is possible that taskCounter is incremented only without decrementing.

I think we may also need to wrap the returned future and check if cancel is called successfully.

[jrhee17]

i.e. We may need something like this:

  class BlockingFuture<T> implements ScheduledFuture<T> {
    private final Future<T> delegate;
    BlockingFuture(Future<T> delegate) {
        this.delegate = delegate;
    }
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        boolean cancelled = delegate.cancel(mayInterruptIfRunning);
        if (cancelled) {
            taskCounter.decrementAndGet();
        }
        return cancelled;
    }

[github-actions]

🔍 Build Scan® (commit: b9e34a1)

Job name	Status	Build Scan®
build-windows-latest-jdk-19	❌ (failure)	https://ge.armeria.dev/s/7pwtyd4uxrywk
build-self-hosted-unsafe-jdk-8	❌ (failure)	https://ge.armeria.dev/s/iqmwbajynllua
build-self-hosted-unsafe-jdk-19-snapshot-blockhound	❌ (failure)	https://ge.armeria.dev/s/jkhenrzvrizh4
build-self-hosted-unsafe-jdk-17-min-java-17-coverage	❌ (failure)	https://ge.armeria.dev/s/2mqoqr26xnpna
build-self-hosted-unsafe-jdk-17-min-java-11	❌ (failure)	https://ge.armeria.dev/s/nsgyrecdv2c5m
build-self-hosted-unsafe-jdk-17-leak	❌ (failure)	https://ge.armeria.dev/s/wgonguq6xsktu
build-self-hosted-unsafe-jdk-11	❌ (failure)	https://ge.armeria.dev/s/r4xn7ibdgctro
build-macos-12-jdk-19	❌ (failure)	https://ge.armeria.dev/s/dssbcihzyedhw

[jrhee17]

Can you also check the CI failures? I think this PR should be almost done once my comments are addressed 🙇