diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc index 019f78209..48bef3c75 100644 --- a/src/iceberg/catalog/rest/http_client.cc +++ b/src/iceberg/catalog/rest/http_client.cc @@ -20,6 +20,9 @@ #include "iceberg/catalog/rest/http_client.h" #include +#include +#include +#include #include @@ -34,19 +37,25 @@ namespace iceberg::rest { class HttpResponse::Impl { public: - explicit Impl(cpr::Response&& response) : response_(std::move(response)) {} + explicit Impl(cpr::Response response) + : status_code_(static_cast(response.status_code)), + body_(std::move(response.text)), + headers_(response.header.begin(), response.header.end()) {} ~Impl() = default; - int32_t status_code() const { return static_cast(response_.status_code); } + static Result Make(cpr::Response response, + const ErrorHandler& error_handler); - std::string body() const { return response_.text; } + int32_t status_code() const { return status_code_; } - std::unordered_map headers() const { - return {response_.header.begin(), response_.header.end()}; - } + std::string body() const { return body_; } + + std::unordered_map headers() const { return headers_; } private: - cpr::Response response_; + int32_t status_code_; + std::string body_; + std::unordered_map headers_; }; HttpResponse::HttpResponse() = default; @@ -190,6 +199,50 @@ Status HandleFailureResponse(const cpr::Response& response, } // namespace +Result HttpResponse::Impl::Make(cpr::Response response, + const ErrorHandler& error_handler) { + ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); + HttpResponse http_response; + http_response.impl_ = std::make_unique(std::move(response)); + return http_response; +} + +// libcurl does not support using a connection cache from several threads at once +// (CURLSHOPT_SHARE(3)). +class HttpClient::ConnectionPools { + public: + template + Result Run(const Send& send, const ErrorHandler& error_handler) { + std::unique_ptr pool = Acquire(); + // Make() consumes the CPR response, freeing its curl handle before pool reuse. + Result response = HttpResponse::Impl::Make(send(*pool), error_handler); + Release(std::move(pool)); + return response; + } + + private: + std::unique_ptr Acquire() { + { + std::lock_guard lock(mutex_); + if (!idle_.empty()) { + auto pool = std::move(idle_.back()); + idle_.pop_back(); + return pool; + } + } + return std::make_unique(); + } + + void Release(std::unique_ptr pool) { + std::lock_guard lock(mutex_); + idle_.push_back(std::move(pool)); + } + + std::mutex mutex_; + // LIFO: the most recently released pool is the likeliest to have a live connection. + std::vector> idle_; +}; + HttpClient::HttpClient(std::unordered_map default_headers, std::chrono::milliseconds connection_timeout, std::chrono::milliseconds socket_timeout, @@ -198,7 +251,7 @@ HttpClient::HttpClient(std::unordered_map default_head default_headers_{std::move(default_headers)}, connection_timeout_{connection_timeout}, socket_timeout_{socket_timeout}, - connection_pool_{std::make_unique()} { + connection_pools_{std::make_unique()} { // Set default Content-Type for all requests (including GET/HEAD/DELETE). // Many systems require that content type is set regardless, even on an empty body. default_headers_[kHeaderContentType] = kMimeTypeApplicationJson; @@ -234,15 +287,14 @@ Result HttpClient::Get( ICEBERG_ASSIGN_OR_RAISE(auto authenticated, AuthenticateRequest(session, HttpMethod::kGet, std::move(url), MergeHeaders(default_headers_, headers))); - cpr::Response response = - cpr::Get(cpr::Url{authenticated.url}, ToCprHeader(authenticated), - kNoFollowRedirects, *connection_pool_, - cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_}); - - ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); - HttpResponse http_response; - http_response.impl_ = std::make_unique(std::move(response)); - return http_response; + return connection_pools_->Run( + [&](const cpr::ConnectionPool& pool) { + return cpr::Get(cpr::Url{authenticated.url}, ToCprHeader(authenticated), + kNoFollowRedirects, pool, + cpr::ConnectTimeout{connection_timeout_}, + cpr::Timeout{socket_timeout_}); + }, + error_handler); } Result HttpClient::Post( @@ -254,15 +306,14 @@ Result HttpClient::Post( auto authenticated, AuthenticateRequest(session, HttpMethod::kPost, std::move(request_uri), MergeHeaders(default_headers_, headers), body)); - cpr::Response response = - cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body}, - ToCprHeader(authenticated), kNoFollowRedirects, *connection_pool_, - cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_}); - - ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); - HttpResponse http_response; - http_response.impl_ = std::make_unique(std::move(response)); - return http_response; + return connection_pools_->Run( + [&](const cpr::ConnectionPool& pool) { + return cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body}, + ToCprHeader(authenticated), kNoFollowRedirects, pool, + cpr::ConnectTimeout{connection_timeout_}, + cpr::Timeout{socket_timeout_}); + }, + error_handler); } Result HttpClient::PostForm( @@ -286,15 +337,14 @@ Result HttpClient::PostForm( AuthenticateRequest(session, HttpMethod::kPost, std::move(request_uri), MergeHeaders(default_headers_, form_headers), std::move(encoded_body))); - cpr::Response response = - cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body}, - ToCprHeader(authenticated), kNoFollowRedirects, *connection_pool_, - cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_}); - - ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); - HttpResponse http_response; - http_response.impl_ = std::make_unique(std::move(response)); - return http_response; + return connection_pools_->Run( + [&](const cpr::ConnectionPool& pool) { + return cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body}, + ToCprHeader(authenticated), kNoFollowRedirects, pool, + cpr::ConnectTimeout{connection_timeout_}, + cpr::Timeout{socket_timeout_}); + }, + error_handler); } Result HttpClient::Head( @@ -305,15 +355,14 @@ Result HttpClient::Head( auto authenticated, AuthenticateRequest(session, HttpMethod::kHead, std::move(request_uri), MergeHeaders(default_headers_, headers))); - cpr::Response response = - cpr::Head(cpr::Url{authenticated.url}, ToCprHeader(authenticated), - kNoFollowRedirects, *connection_pool_, - cpr::ConnectTimeout{connection_timeout_}, cpr::Timeout{socket_timeout_}); - - ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); - HttpResponse http_response; - http_response.impl_ = std::make_unique(std::move(response)); - return http_response; + return connection_pools_->Run( + [&](const cpr::ConnectionPool& pool) { + return cpr::Head(cpr::Url{authenticated.url}, ToCprHeader(authenticated), + kNoFollowRedirects, pool, + cpr::ConnectTimeout{connection_timeout_}, + cpr::Timeout{socket_timeout_}); + }, + error_handler); } Result HttpClient::Delete( @@ -326,15 +375,14 @@ Result HttpClient::Delete( auto authenticated, AuthenticateRequest(session, HttpMethod::kDelete, std::move(url), MergeHeaders(default_headers_, headers))); - cpr::Response response = cpr::Delete( - cpr::Url{authenticated.url}, ToCprHeader(authenticated), kNoFollowRedirects, - *connection_pool_, cpr::ConnectTimeout{connection_timeout_}, - cpr::Timeout{socket_timeout_}); - - ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler)); - HttpResponse http_response; - http_response.impl_ = std::make_unique(std::move(response)); - return http_response; + return connection_pools_->Run( + [&](const cpr::ConnectionPool& pool) { + return cpr::Delete(cpr::Url{authenticated.url}, ToCprHeader(authenticated), + kNoFollowRedirects, pool, + cpr::ConnectTimeout{connection_timeout_}, + cpr::Timeout{socket_timeout_}); + }, + error_handler); } } // namespace iceberg::rest diff --git a/src/iceberg/catalog/rest/http_client.h b/src/iceberg/catalog/rest/http_client.h index 9a2459387..bc7e60a2a 100644 --- a/src/iceberg/catalog/rest/http_client.h +++ b/src/iceberg/catalog/rest/http_client.h @@ -32,17 +32,9 @@ /// \file iceberg/catalog/rest/http_client.h /// \brief Http client for Iceberg REST API. -namespace cpr { -class ConnectionPool; -} // namespace cpr - namespace iceberg::rest { -/// \brief A simple wrapper for cpr::Response. -/// -/// This class encapsulates the details of the underlying cpr library's response, -/// providing a consistent interface that is independent of the specific network -/// library used. +/// \brief HTTP response status, body, and headers. class ICEBERG_REST_EXPORT HttpResponse { public: HttpResponse(); @@ -70,6 +62,9 @@ class ICEBERG_REST_EXPORT HttpResponse { }; /// \brief HTTP client for making requests to Iceberg REST Catalog API. +/// +/// Thread-safe, provided the AuthSession passed to each request supports +/// concurrent calls to Authenticate(). class ICEBERG_REST_EXPORT HttpClient { public: /// \param connection_timeout,socket_timeout Zero leaves requests unbounded. @@ -116,6 +111,8 @@ class ICEBERG_REST_EXPORT HttpClient { const ErrorHandler& error_handler, auth::AuthSession& session); private: + class ConnectionPools; + /// \brief Build a request URI from this client's base URI and a path. Result RequestUri(const std::string& path) const; @@ -123,7 +120,7 @@ class ICEBERG_REST_EXPORT HttpClient { std::unordered_map default_headers_; std::chrono::milliseconds connection_timeout_; std::chrono::milliseconds socket_timeout_; - std::unique_ptr connection_pool_; + std::unique_ptr connection_pools_; }; } // namespace iceberg::rest diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc index a25d28eb4..c0934a36c 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -584,6 +584,7 @@ RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr f std::string_view RestCatalog::name() const { return name_; } Result> RestCatalog::AsCatalog() { + std::lock_guard lock(default_catalog_mutex_); if (auto catalog = default_catalog_.lock()) { return catalog; } diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h index 561ff3471..51a25c6bf 100644 --- a/src/iceberg/catalog/rest/rest_catalog.h +++ b/src/iceberg/catalog/rest/rest_catalog.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -43,6 +44,10 @@ namespace iceberg::rest { /// \brief Session-aware REST catalog root. +/// +/// Thread-safe, including the catalogs from AsCatalog() and WithContext(), as long as +/// the configured HTTP client, FileIO, metrics reporters, executors, auth manager and +/// its sessions are. class ICEBERG_REST_EXPORT RestCatalog final : public SessionCatalog, public std::enable_shared_from_this { @@ -235,7 +240,8 @@ class ICEBERG_REST_EXPORT RestCatalog final SnapshotMode snapshot_mode_; std::optional client_scan_planning_mode_; SessionContext default_context_; - std::weak_ptr default_catalog_; + std::mutex default_catalog_mutex_; + std::weak_ptr default_catalog_; // Guarded by default_catalog_mutex_. std::shared_ptr reporter_; Executor* metrics_executor_ = nullptr; }; diff --git a/src/iceberg/test/http_client_test.cc b/src/iceberg/test/http_client_test.cc index 6cc653131..37b2ed85e 100644 --- a/src/iceberg/test/http_client_test.cc +++ b/src/iceberg/test/http_client_test.cc @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -301,7 +302,6 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) { ASSERT_NO_FATAL_FAILURE(Listen(listener)); const auto url = std::format("http://127.0.0.1:{}/v1/commit", listener.port); - // Outlives the response, which keeps a curl handle in its connection pool. HttpClient client; auto session = auth::AuthSession::MakeDefault({}); std::thread server([&] { ServeOnce(listener.fd, response_bytes, /*end=*/true); }); @@ -312,6 +312,40 @@ TEST(HttpClientTest, PostReportsTransferFailureAfterSuccessfulStatus) { EXPECT_THAT(response, HasErrorMessage("failed after status 200")); } +TEST(HttpClientTest, ResponseOutlivesLaterRequestsAndClient) { + constexpr std::string_view kBody = R"({"defaults":{}})"; + const auto response_bytes = std::format( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Test: kept\r\n" + "Content-Length: {}\r\nConnection: close\r\n\r\n{}", + kBody.size(), kBody); + Listener listener; + ASSERT_NO_FATAL_FAILURE(Listen(listener)); + const auto url = std::format("http://127.0.0.1:{}/v1/config", listener.port); + + auto client = std::make_unique(); + auto session = auth::AuthSession::MakeDefault({}); + std::thread server([&] { + ServeOnce(listener.fd, response_bytes, /*end=*/true); + ServeOnce(listener.fd, response_bytes, /*end=*/true); + }); + auto first = client->Get(url, /*params=*/{}, /*headers=*/{}, + *DefaultErrorHandler::Instance(), *session); + auto second = client->Get(url, /*params=*/{}, /*headers=*/{}, + *DefaultErrorHandler::Instance(), *session); + server.join(); + client.reset(); + + auto expect_intact = [&](const Result& response) { + ASSERT_THAT(response, IsOk()); + EXPECT_EQ(response->status_code(), 200); + EXPECT_EQ(response->body(), kBody); + EXPECT_THAT(response->headers(), + ::testing::Contains(::testing::Pair("X-Test", "kept"))); + }; + expect_intact(first); + expect_intact(second); +} + #endif class CapturingAuthSession final : public auth::AuthSession { @@ -340,6 +374,22 @@ TEST(HttpClientTest, BuildsRequestUriLikeJavaHttpRequest) { *DefaultErrorHandler::Instance(), session); EXPECT_THAT(absolute, IsError(ErrorKind::kInvalidArgument)); EXPECT_EQ(session.request().url, "https://auth.example/token/"); + + // A different path per method, so a call that fails before authenticating can't + // pass by matching the URL left over from the previous call. + const auto& handler = *DefaultErrorHandler::Instance(); + EXPECT_THAT(client.Get("v1/get", {}, {}, handler, session), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/get"); + EXPECT_THAT(client.Post("v1/post", "{}", {}, handler, session), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/post"); + EXPECT_THAT(client.Head("v1/head", {}, handler, session), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/head"); + EXPECT_THAT(client.Delete("v1/delete", {}, {}, handler, session), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_EQ(session.request().url, "https://catalog.example/api/v1/delete"); } // Redirect responses must be reported without replaying the request at the target. diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index 0e3ba6a3d..eea638b63 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -26,6 +26,7 @@ #include #include #include +#include #include #include @@ -425,6 +426,76 @@ TEST_F(RestCatalogIntegrationTest, FetchServerConfigDirect) { } } +// RestCatalog shares its HttpClient with token refresh and metrics reporting. +TEST_F(RestCatalogIntegrationTest, ConcurrentHttpClientRequests) { + constexpr int kThreads = 32; + constexpr int kRequestsPerThread = 200; + + HttpClient client({}, std::chrono::seconds(30), std::chrono::seconds(30)); + auto noop_session = auth::AuthSession::MakeDefault({}); + const std::string config_url = std::format("{}/v1/config", CatalogUri()); + + std::vector threads; + threads.reserve(kThreads); + for (int i = 0; i < kThreads; ++i) { + threads.emplace_back([&] { + for (int j = 0; j < kRequestsPerThread; ++j) { + ASSERT_THAT(client.Get(config_url, {}, /*headers=*/{}, + *DefaultErrorHandler::Instance(), *noop_session), + IsOk()); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } +} + +TEST_F(RestCatalogIntegrationTest, ConcurrentCatalogUse) { + constexpr int kThreads = 16; + constexpr int kAsCatalogCalls = 10000; + constexpr int kRequests = 50; + + auto config = RestCatalogProperties::default_properties(); + config.Set(RestCatalogProperties::kUri, CatalogUri()) + .Set(RestCatalogProperties::kName, std::string(kCatalogName)) + .Set(RestCatalogProperties::kWarehouse, std::string(kWarehouseName)); + config.mutable_configs()[std::string(RestCatalogProperties::kIOImpl.key())] = + std::string(kStdFileIOImpl); + ICEBERG_UNWRAP_OR_FAIL(auto root, RestCatalog::Make(config)); + + Namespace ns{.levels = {"test_concurrent_catalog_use"}}; + TableIdentifier table_id{.ns = ns, .name = "events"}; + { + ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog()); + ASSERT_THAT(catalog->CreateNamespace(ns, {}), IsOk()); + ASSERT_THAT(CreateDefaultTable(catalog, table_id), IsOk()); + } + + std::vector threads; + threads.reserve(kThreads); + for (int i = 0; i < kThreads; ++i) { + threads.emplace_back([&] { + // Nothing else holds the default catalog, so these calls keep recreating it. + for (int j = 0; j < kAsCatalogCalls; ++j) { + ASSERT_THAT(root->AsCatalog(), IsOk()); + } + ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog()); + for (int j = 0; j < kRequests; ++j) { + ASSERT_THAT(catalog->LoadTable(table_id), IsOk()); + ASSERT_THAT(catalog->ListNamespaces(ns), IsOk()); + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + + ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->AsCatalog()); + ASSERT_THAT(catalog->DropTable(table_id, /*purge=*/false), IsOk()); + ASSERT_THAT(catalog->DropNamespace(ns), IsOk()); +} + // -- Namespace operations -- TEST_F(RestCatalogIntegrationTest, ListNamespaces) { diff --git a/src/iceberg/test/rest_table_scan_test.cc b/src/iceberg/test/rest_table_scan_test.cc index 7610f8249..4877ac255 100644 --- a/src/iceberg/test/rest_table_scan_test.cc +++ b/src/iceberg/test/rest_table_scan_test.cc @@ -105,8 +105,7 @@ MATCHER_P(JsonBodyLacks, key, "") { // -------------------------------------------------------------------------- // Mock HTTP client that overrides the virtual methods of HttpClient. -// The base class constructor creates a cpr::ConnectionPool, which is a -// lightweight allocation (no network connections are opened at construction). +// The base class constructor opens no network connections. // -------------------------------------------------------------------------- class MockHttpClient : public HttpClient { public: