feat: support cancellation tokens in Curl REST client and payloads - #16381
feat: support cancellation tokens in Curl REST client and payloads#16381ajayky-os wants to merge 1 commit into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces cancellation support for REST/HTTP operations using libcurl by adding a cancellation token to RestContext and propagating it to CurlImpl. This allows in-progress transfers to be aborted via curl_multi_wakeup and transfer info callbacks. The review feedback highlights critical thread-safety issues regarding the std::shared_ptr<std::atomic<bool>> cancellation token. Specifically, concurrent accesses and modifications to the std::shared_ptr itself in Read(), SetCancellationToken(), and Cancel() are not synchronized and can result in data races, so they should be protected by the multi_mu_ mutex.
| if (cancellation_token_) { | ||
| context.set_cancellation_token(cancellation_token_); | ||
| } |
There was a problem hiding this comment.
Since Read() can be called on a different thread (e.g., the user thread reading the payload) than the thread modifying the cancellation token, copying cancellation_token_ here can race with SetCancellationToken(). We should protect this access with multi_mu_.
std::shared_ptr<std::atomic<bool>> token;
{
std::lock_guard<std::mutex> lk(multi_mu_);
token = cancellation_token_;
}
if (token) {
context.set_cancellation_token(std::move(token));
}| void CurlImpl::SetCancellationToken( | ||
| std::shared_ptr<std::atomic<bool>> token) { | ||
| if (token) { | ||
| if (cancellation_token_->load(std::memory_order_relaxed)) { | ||
| token->store(true, std::memory_order_relaxed); | ||
| } | ||
| cancellation_token_ = std::move(token); | ||
| cancellable_ = true; | ||
| } | ||
| } |
There was a problem hiding this comment.
There is a potential data race on the cancellation_token_ std::shared_ptr member variable. While the underlying std::atomic<bool> is thread-safe, std::shared_ptr itself is not thread-safe for concurrent read and write operations.
Since Cancel() can be called concurrently from another thread, and SetCancellationToken() writes to cancellation_token_ (by moving a new std::shared_ptr into it), this can lead to undefined behavior. We should protect accesses and modifications of the cancellation_token_ std::shared_ptr using multi_mu_.
void CurlImpl::SetCancellationToken(
std::shared_ptr<std::atomic<bool>> token) {
if (token) {
std::lock_guard<std::mutex> lk(multi_mu_);
if (cancellation_token_ &&
cancellation_token_->load(std::memory_order_relaxed)) {
token->store(true, std::memory_order_relaxed);
}
cancellation_token_ = std::move(token);
cancellable_ = true;
}
}| void CurlImpl::Cancel() { | ||
| cancellation_token_->store(true, std::memory_order_relaxed); | ||
| #if CURL_AT_LEAST_VERSION(7, 68, 0) | ||
| // The lock keeps `multi_` alive and owned by this request while the wakeup | ||
| // is delivered: without it the transfer thread could concurrently return | ||
| // the handle to the pool, where another request may already be using it. | ||
| std::lock_guard<std::mutex> lk(multi_mu_); | ||
| if (multi_) { | ||
| (void)curl_multi_wakeup(multi_.get()); | ||
| } | ||
| #endif | ||
| } |
There was a problem hiding this comment.
To avoid data races on the cancellation_token_ std::shared_ptr when Cancel() is called concurrently from another thread, we should access cancellation_token_ while holding the multi_mu_ lock.
void CurlImpl::Cancel() {
std::lock_guard<std::mutex> lk(multi_mu_);
if (cancellation_token_) {
cancellation_token_->store(true, std::memory_order_relaxed);
}
#if CURL_AT_LEAST_VERSION(7, 68, 0)
// The lock keeps `multi_` alive and owned by this request while the wakeup
// is delivered: without it the transfer thread could concurrently return
// the handle to the pool, where another request may already be using it.
if (multi_) {
(void)curl_multi_wakeup(multi_.get());
}
#endif
}
3516423 to
2808537
Compare
|
/gcbrun |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #16381 +/- ##
=======================================
Coverage 92.28% 92.28%
=======================================
Files 2244 2244
Lines 211631 211708 +77
=======================================
+ Hits 195294 195373 +79
+ Misses 16337 16335 -2 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
2 participants
Summary
This PR introduces cancellation support to the internal REST transport layer and libcurl client implementation using cooperative cancellation tokens (
std::shared_ptr<std::atomic<bool>>).Problem & Motivation
When performing speculative operations (such as hedging) or handling user-initiated request cancellations, active HTTP streaming transfers continue consuming network bandwidth and socket connections until
the transfer finishes or times out. Libcurl provides transfer progress callbacks (
CURLOPT_XFERINFOFUNCTION/CURLOPT_PROGRESSFUNCTION) that allow transfers to be aborted mid-flight by returning a non-zero exit code (
CURLE_ABORTED_BY_CALLBACK).Changes
RestContext: Added optionalcancellation_tokenfield and getter/setter (set_cancellation_token()/cancellation_token()).CurlImpl: RegisteredCURLOPT_XFERINFOFUNCTION(with fallback toCURLOPT_PROGRESSFUNCTIONon older libcurl versions) that polls the cancellation token during transfers and aborts early ifflagged.
CurlHttpPayload&CurlRestClient: Propagated cancellation tokens from the request context to the active HTTP payload reader.TracingHttpPayload& Mocks: Updated decorators and testing mocks to forward cancellation tokens across wrappers.Verification & Testing
bazel test //google/cloud:internal_curl_impl_test \ //google/cloud:internal_curl_rest_client_test \ //google/cloud:internal_rest_context_test \ //google/cloud:internal_tracing_http_payload_test