diff --git a/CHANGELOG.md b/CHANGELOG.md index 419fa3b3d..05c08a957 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,10 @@ Increment the: in the value as-is rather than dropping the attribute. [#1536](https://github.com/open-telemetry/opentelemetry-cpp/issues/1536) +* [BUG] Send one request per curl session, rather than replacing the operation + a running request still belongs to + [#4396](https://github.com/open-telemetry/opentelemetry-cpp/issues/4396) + * [BUG] Report one outcome per request when a curl session is cancelled after the response arrives ([#4363](https://github.com/open-telemetry/opentelemetry-cpp/pull/4363)) diff --git a/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h b/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h index 9a09fac9a..02e288ee4 100644 --- a/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h +++ b/ext/include/opentelemetry/ext/http/client/curl/http_client_curl.h @@ -186,10 +186,25 @@ class Session : public opentelemetry::ext::http::client::Session, std::shared_ptr CreateRequest() noexcept override { + if (send_started_.load(std::memory_order_acquire)) + { + // The request that has been sent stays where it is. libcurl does not copy the body or the + // header list, so the transfer reads them for as long as it runs. + return http_request_; + } + http_request_.reset(new Request()); return http_request_; } + /** + * Send the request this session carries. + * + * A session carries one request. A second call reports CreateFailed to its own handler and + * sends nothing: the easy handle of the first request holds this session and the operation + * that owns it, and replacing that operation takes away what libcurl and the client are still + * reading. + */ void SendRequest( std::shared_ptr callback) noexcept override; @@ -235,6 +250,9 @@ class Session : public opentelemetry::ext::http::client::Session, uint64_t session_id_ = 0UL; HttpClient &http_client_; std::atomic is_session_active_{false}; + + // Raised by the first SendRequest and never lowered. A session carries one request. + std::atomic send_started_{false}; }; class HttpClientSync : public opentelemetry::ext::http::client::HttpClientSync diff --git a/ext/src/http/client/curl/http_client_curl.cc b/ext/src/http/client/curl/http_client_curl.cc index 76f67fcd9..a1c50caf8 100644 --- a/ext/src/http/client/curl/http_client_curl.cc +++ b/ext/src/http/client/curl/http_client_curl.cc @@ -141,6 +141,21 @@ static int deflateInPlace(z_stream *strm, unsigned char *buf, uint32_t len, uint void Session::SendRequest( std::shared_ptr callback) noexcept { + if (send_started_.exchange(true, std::memory_order_acq_rel)) + { + // The first request is not finished with this session. Its easy handle names the session in + // CURLOPT_PRIVATE and names its operation in every callback it was given, and the message + // loop resolves that name to whichever operation the session owns, so a second operation + // would be handed the first one's completion. Worse from a handler that sends again from + // OnResponse, where the operation being replaced is the one running that handler. + if (callback) + { + callback->OnEvent(opentelemetry::ext::http::client::SessionState::CreateFailed, + "a session carries one request"); + } + return; + } + is_session_active_.store(true, std::memory_order_release); const auto &url = host_ + http_request_->uri_; auto callback_ptr = callback.get(); diff --git a/ext/test/http/curl_http_test.cc b/ext/test/http/curl_http_test.cc index 90142962d..65f32436c 100644 --- a/ext/test/http/curl_http_test.cc +++ b/ext/test/http/curl_http_test.cc @@ -97,6 +97,37 @@ class CustomEventHandler : public http_client::EventHandler // inside the Response event, which is the one moment both arms of the completion callback are // eligible: DispatchEvent notifies the handler before it stores the new state, and the callback // runs after both, so it sees an aborted operation that also has a response. +class ReentrantSendHandler : public CustomEventHandler +{ +public: + void OnResponse(http_client::Response & /* response */) noexcept override + { + got_response_.store(true, std::memory_order_release); + + if (session_ != nullptr && !resent_.exchange(true, std::memory_order_acq_rel)) + { + auto again = session_->CreateRequest(); + again->SetUri("get/"); + session_->SendRequest(self_.lock()); + } + } + + void OnEvent(http_client::SessionState state, nostd::string_view /* reason */) noexcept override + { + if (state == http_client::SessionState::CreateFailed) + { + create_failed_.fetch_add(1, std::memory_order_release); + } + } + + http_client::Session *session_ = nullptr; + std::weak_ptr self_; + std::atomic create_failed_{0}; + +private: + std::atomic resent_{false}; +}; + class TerminalCountingHandler : public CustomEventHandler { public: @@ -629,6 +660,63 @@ TEST_F(BasicCurlHttpTests, ResetMultiHandleWithASessionDoesNotDeadlock) client->FinishAllSessions(); } +// The reproduction from #4396. A handler that sends again from OnResponse replaces the operation +// whose Cleanup is running that handler, and Cleanup reads that operation again on the way out. +TEST_F(BasicCurlHttpTests, ASecondRequestFromInsideTheResponseIsRefused) +{ + received_requests_.clear(); + auto session_manager = std::make_shared()->Create(); + ASSERT_TRUE(session_manager != nullptr); + + auto session = session_manager->CreateSession("http://127.0.0.1:19000"); + auto request = session->CreateRequest(); + request->SetUri("get/"); + + auto handler = std::make_shared(); + handler->session_ = session.get(); + handler->self_ = handler; + + session->SendRequest(handler); + ASSERT_TRUE(waitForRequests(30, 1)); + session->FinishSession(); + + EXPECT_TRUE(handler->got_response_.load(std::memory_order_acquire)); + EXPECT_EQ(1, handler->create_failed_.load(std::memory_order_acquire)); + + session_manager->FinishAllSessions(); +} + +// The same thing without the re-entrancy, and the request the first send is reading from is not +// replaced under it either. +TEST_F(BasicCurlHttpTests, ASessionSendsOneRequest) +{ + received_requests_.clear(); + auto session_manager = std::make_shared()->Create(); + ASSERT_TRUE(session_manager != nullptr); + + auto session = session_manager->CreateSession("http://127.0.0.1:19000"); + auto request = session->CreateRequest(); + request->SetUri("get/"); + + auto first = std::make_shared(); + session->SendRequest(first); + ASSERT_TRUE(waitForRequests(30, 1)); + session->FinishSession(); + + EXPECT_TRUE(first->got_response_.load(std::memory_order_acquire)); + EXPECT_EQ(0, first->create_failed_.load(std::memory_order_acquire)); + + auto again = session->CreateRequest(); + EXPECT_EQ(request.get(), again.get()); + + auto second = std::make_shared(); + session->SendRequest(second); + EXPECT_EQ(1, second->create_failed_.load(std::memory_order_acquire)); + EXPECT_FALSE(second->got_response_.load(std::memory_order_acquire)); + + session_manager->FinishAllSessions(); +} + // The caller-thread side of the same cancel. The server handler takes mtx_requests before it // answers, so holding it keeps a response from racing the cancel and the abort lands while the // IO thread is still driving the easy handle. That pairing is what #4369 caught.