common: add a thread safe token bucket

[nitgoy]

Commit Message: Add a thread-safe token bucket wrapper class to be shared across threads. The other token bucket implementation was not thread safe which made it unusable for certain cases. This wrapper class calls the thread-unsafe class for the core token bucket functionality.
Additional Description: New class, not used anywhere else yet except UTs.
Risk Level: Low
Testing: Ran existing and newly added UTs.
Docs Changes: None
Release Notes:
Platform Specific Features:

[repokitteh-read-only]

Hi @nitgoy, welcome and thank you for your contribution.

We will try to review your Pull Request as quickly as possible.

In the meantime, please take a look at the contribution guidelines if you have not done so already.

🐱

Caused by: #14827 was opened by nitgoy.

see: more, trace.

[ggreenway]

I think it would be cleaner to add a wrapper class around the existing token bucket class to add locking, instead of inserting optional locking into it.

[yanavlasov]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[nitgoy]

I think it would be cleaner to add a wrapper class around the existing token bucket class to add locking, instead of inserting optional locking into it.

Done.

It looks like ABSL_EXCLUSIVE_LOCKS_REQUIRED is not acquiring lock for me locally but it does in the CodeQL-build. That indicates the exclusive_locks_required ABSL attribute is not set. Shouldn't it be automatically set for the bazel workspace?

[ggreenway]

It looks like ABSL_EXCLUSIVE_LOCKS_REQUIRED is not acquiring lock for me locally but it does in the CodeQL-build. That indicates the exclusive_locks_required ABSL attribute is not set. Shouldn't it be automatically set for the bazel workspace?

That annotation allows the compiler to verify that the correct lock is held everytime you access a field. It does not automatically acquire the lock for you. It's only used for static code analysis.

[nitgoy]

It looks like ABSL_EXCLUSIVE_LOCKS_REQUIRED is not acquiring lock for me locally but it does in the CodeQL-build. That indicates the exclusive_locks_required ABSL attribute is not set. Shouldn't it be automatically set for the bazel workspace?

That annotation allows the compiler to verify that the correct lock is held everytime you access a field. It does not automatically acquire the lock for you. It's only used for static code analysis.

I thought so too but I see below CodeQL-build errors if I keep it along with the actual LockGuard (see previous iteration). The build passes if I remove the macro. Is this the right macro?

source/common/common/shared_token_bucket_impl.cc:18:21: error: acquiring mutex 'mutex_' that is already held [-Werror,-Wthread-safety-analysis]
Thread::LockGuard lock(mutex_);
^
source/common/common/shared_token_bucket_impl.cc:17:5: note: mutex acquired here
ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) {
^
external/com_google_absl/absl/base/thread_annotations.h:143:18: note: expanded from macro 'ABSL_EXCLUSIVE_LOCKS_REQUIRED'
attribute((exclusive_locks_required(VA_ARGS)))

[ggreenway]

Before we go further with this review, can you explain what this code will be used for? Oftentimes it's easier to make changes like this in the context that it will be used, so the requirements/use-case are clearer.

[nitgoy]

Before we go further with this review, can you explain what this code will be used for? Oftentimes it's easier to make changes like this in the context that it will be used, so the requirements/use-case are clearer.

This is the PR for that will use it: #14096. Essentially, I need it to represent token bucket for per filter-chain/listener limit for bandwidth limit filter that will span across multiple http streams.

[ggreenway]

High-level question: is this wrapper needed, or would it be better to just have a mutex and a TokenBucketImpl in the class that needs it? Will the owning class ever need to manipulate the token bucket and some other state atomically under the same lock?

/wait

[nitgoy]

High-level question: is this wrapper needed, or would it be better to just have a mutex and a TokenBucketImpl in the class that needs it? Will the owning class ever need to manipulate the token bucket and some other state atomically under the same lock?

/wait

The interface methods will be called by the common class StreamRateLimiter. Depending upon the use case callers can pass the TokenBucketImpl or the SharedTokenBucketImpl. That way the StreamRateLimiter class can be agnostic whether the token bucket's shared or not. Also, a caller that doesn't need the thread-safety can avoid paying that cost.

[ggreenway]

Looking at this, and how it's used in the stream limit filter, I think you need to change the API to make consume() and nextTokenAvailable() be atomic. With what you have here, consume() could return no tokens, but before nextTokenAvailable() is called some tokens could be added, and thus nextTokenAvailable() will return 0, the timer will not be scheduled, and no data will be written.

I'm not sure that just adding a mutex around each call on the TokenBucket interface is sufficient for what you're trying to accomplish.

/wait-any

[nitgoy]

With what you have here, consume() could return no tokens, but before nextTokenAvailable() is called some tokens could be added, and thus nextTokenAvailable() will return 0, the timer will not be scheduled, and no data will be written.

Thanks for catching this issue. I'll try adding a test around it. Does it make sense to change the check right after nextTokenAvailable() call to if (ms.count() >= 0)? We could use 1ms as minimum for next timer schedule and move the write_data_cb_ call before it, to allow the current thread to finish that operation before next timer wakeup.

The other approach would be to have the mutex in streamratelimiter class as you suggested but has other downsides as I stated before.

[ggreenway]

I think it makes most sense to add a method to the interface to get the two values atomically.

[nitgoy]

I think it makes most sense to add a method to the interface to get the two values atomically.

Made the suggested changes. Can you please take a look? Working on fixing these other issues.

[azure-pipelines]

Commenter does not have sufficient privileges for PR 14827 in repo envoyproxy/envoy

[ggreenway]

/retest

[repokitteh-read-only]

Retrying Azure Pipelines:
Retried failed jobs in: envoy-presubmit

🐱

Caused by: a #14827 (review) was submitted by @ggreenway.

see: more, trace.