From be99b60febdec85f1fd0bc8ce0393af1a51483c7 Mon Sep 17 00:00:00 2001 From: Guogang Li Date: Fri, 31 Mar 2023 13:22:23 -0700 Subject: [PATCH] Applied platform thread to network lib PiperOrigin-RevId: 520998502 --- fastpair/internal/test/BUILD | 3 + .../test/fast_pair_fake_http_client.h | 6 + internal/network/BUILD | 5 +- internal/network/http_client.h | 7 +- internal/network/http_client_impl.cc | 184 ++++++++++-------- internal/network/http_client_impl.h | 25 +-- internal/network/http_client_impl_test.cc | 78 ++++++-- internal/platform/BUILD | 1 + 8 files changed, 198 insertions(+), 111 deletions(-) diff --git a/fastpair/internal/test/BUILD b/fastpair/internal/test/BUILD index c7413fad..3ea07c91 100644 --- a/fastpair/internal/test/BUILD +++ b/fastpair/internal/test/BUILD @@ -14,6 +14,8 @@ cc_library( deps = [ "//internal/network:types", "@com_google_absl//absl/functional:any_invocable", + "@com_google_absl//absl/status", + "@com_google_absl//absl/status:statusor", ], ) @@ -30,6 +32,7 @@ cc_test( deps = [ ":nearby_fastpair_test", "//internal/network:types", + "//internal/platform/implementation/g3", "@com_github_protobuf_matchers//protobuf-matchers", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/functional:any_invocable", diff --git a/fastpair/internal/test/fast_pair_fake_http_client.h b/fastpair/internal/test/fast_pair_fake_http_client.h index 84824a3a..c5baccb6 100644 --- a/fastpair/internal/test/fast_pair_fake_http_client.h +++ b/fastpair/internal/test/fast_pair_fake_http_client.h @@ -21,6 +21,7 @@ #include #include +#include "absl/status/statusor.h" #include "internal/network/http_client.h" namespace nearby { @@ -51,6 +52,11 @@ class FastPairFakeHttpClient : public HttpClient { request_infos_.push_back(std::move(request_info)); } + absl::StatusOr GetResponse( + const HttpRequest& request) override { + return absl::UnimplementedError("unimplemented"); + } + // Mock methods void CompleteRequest(const absl::StatusOr& response, size_t pos = 0) { diff --git a/internal/network/BUILD b/internal/network/BUILD index e8a818a1..937c7a83 100644 --- a/internal/network/BUILD +++ b/internal/network/BUILD @@ -31,6 +31,7 @@ cc_library( "//presence:__subpackages__", ], deps = [ + "//internal/platform:types", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/status", "@com_google_absl//absl/status:statusor", @@ -63,11 +64,11 @@ cc_library( deps = [ ":types", "//internal/platform:logging", + "//internal/platform:types", "//internal/platform/implementation:platform", "@com_google_absl//absl/base:core_headers", - "@com_google_absl//absl/container:flat_hash_map", + "@com_google_absl//absl/status:statusor", "@com_google_absl//absl/strings:str_format", - "@com_google_absl//absl/synchronization", ], ) diff --git a/internal/network/http_client.h b/internal/network/http_client.h index a99362c3..ac8e2bff 100644 --- a/internal/network/http_client.h +++ b/internal/network/http_client.h @@ -1,4 +1,4 @@ -// Copyright 2021 Google LLC +// Copyright 2021-2023 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -28,10 +28,15 @@ class HttpClient { public: virtual ~HttpClient() = default; + // Starts HTTP request in asynchronization mode. virtual void StartRequest( const HttpRequest& request, std::function&)> callback) = 0; + // Gets HTTP response in synchronization mode. + virtual absl::StatusOr GetResponse( + const HttpRequest& request) = 0; + // The error may be corrected if retried at a later time. static bool IsRetryableHttpError(absl::Status status) { return absl::IsUnavailable(status) || absl::IsFailedPrecondition(status); diff --git a/internal/network/http_client_impl.cc b/internal/network/http_client_impl.cc index 3f7b4a77..ed59f5ee 100644 --- a/internal/network/http_client_impl.cc +++ b/internal/network/http_client_impl.cc @@ -1,4 +1,4 @@ -// Copyright 2021 Google LLC +// Copyright 2021-2023 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -14,107 +14,135 @@ #include "internal/network/http_client_impl.h" -#include // NOLINT #include -#include // NOLINT +#include #include #include #include +#include "absl/status/statusor.h" #include "internal/network/debug.h" #include "internal/platform/implementation/platform.h" #include "internal/platform/logging.h" +#include "internal/platform/mutex_lock.h" namespace nearby { namespace network { +namespace { + +// In nearby SDK, allowed maximum thread count. +constexpr int kMaxNetworkThreadCount = 3; + +} // namespace + +NearbyHttpClient::NearbyHttpClient() { + network_executor_ = + std::make_unique(kMaxNetworkThreadCount); +} void NearbyHttpClient::StartRequest( const HttpRequest& request, std::function&)> callback) { - absl::MutexLock lock(&mutex_); - CleanThreads(); + MutexLock lock(&mutex_); + NEARBY_LOGS(INFO) << __func__ << ": Start async request to url=" + << request.GetUrl().GetUrlPath(); + if (network_executor_ == nullptr) { + callback(absl::ResourceExhaustedError("no available thread")); + return; + } - std::future http_thread = std::async(std::launch::async, [&, request, - callback]() { - api::WebRequest web_request; - web_request.url = request.GetUrl().GetUrlPath(); - web_request.method = absl::StrCat(request.GetMethodString()); - for (const auto& header : request.GetAllHeaders()) { - for (const auto& value : header.second) { - web_request.headers.emplace(header.first, value); - } - } - web_request.body = absl::StrCat(request.GetBody().GetRawData()); - - if (debug::kRequestEnabled) { - std::stringstream request_stream; - request_stream << "HTTP REQUEST====>" << std::endl; - request_stream << web_request.method << " " << web_request.url - << std::endl; - for (const auto& header : web_request.headers) { - request_stream << header.first << ": " << header.second << std::endl; - } - request_stream << std::endl; - request_stream << "body size: " << request.GetBody().GetRawData().size() - << std::endl; - NEARBY_LOGS(VERBOSE) << request_stream.str(); + network_executor_->Execute([&, request, callback]() { + absl::StatusOr response = InternalGetResponse(request); + if (response.ok()) { + NEARBY_LOGS(INFO) << __func__ << ": Got response from url=" + << request.GetUrl().GetUrlPath(); + } else { + NEARBY_LOGS(ERROR) << __func__ << ": Failed to get response from url=" + << request.GetUrl().GetUrlPath() << ", status" + << response.status(); } - absl::StatusOr web_response = - api::ImplementationPlatform::SendRequest(web_request); - - if (!web_response.ok()) { - if (callback != nullptr) { - callback(web_response.status()); - } - return; - } - - if (debug::kResponseEnabled) { - std::stringstream response_stream; - response_stream << "HTTP RESPONSE====>" << std::endl; - response_stream << "url: " << web_request.url << std::endl; - response_stream << web_response->status_code << " " - << web_response->status_text << std::endl; - for (const auto& header : web_response->headers) { - response_stream << header.first << ": " << header.second << std::endl; - } - response_stream << std::endl; - response_stream << "body size: " << web_response->body.size() - << std::endl; - NEARBY_LOGS(VERBOSE) << response_stream.str(); - } - - HttpResponse response; - - response.SetStatusCode( - static_cast(web_response->status_code)); - response.SetReasonPhrase(web_response->status_text); - for (const auto& header : web_response->headers) { - response.AddHeader(header.first, header.second); - } - response.SetBody(web_response->body); - - if (callback != nullptr) { - callback(response); - } + callback(response); + NEARBY_LOGS(INFO) << __func__ << ": Completed request to url=" + << request.GetUrl().GetUrlPath(); }); - - http_threads_.push_back(std::move(http_thread)); } -void NearbyHttpClient::CleanThreads() { - auto it = http_threads_.begin(); +absl::StatusOr NearbyHttpClient::GetResponse( + const HttpRequest& request) { + NEARBY_LOGS(INFO) << __func__ << ": Start request to url=" + << request.GetUrl().GetUrlPath(); - while (it != http_threads_.end()) { - // Delete the thread if it is ready - auto status = it->wait_for(std::chrono::seconds(0)); - if (status == std::future_status::ready) { - it = http_threads_.erase(it); - } else { - ++it; + absl::StatusOr response = InternalGetResponse(request); + if (response.ok()) { + NEARBY_LOGS(INFO) << __func__ << ": Got response from url=" + << request.GetUrl().GetUrlPath(); + } else { + NEARBY_LOGS(ERROR) << __func__ << ": Failed to get response from url=" + << request.GetUrl().GetUrlPath() << ", status" + << response.status(); + } + + return response; +} + +absl::StatusOr NearbyHttpClient::InternalGetResponse( + const HttpRequest& request) { + api::WebRequest web_request; + web_request.url = request.GetUrl().GetUrlPath(); + web_request.method = absl::StrCat(request.GetMethodString()); + for (const auto& header : request.GetAllHeaders()) { + for (const auto& value : header.second) { + web_request.headers.emplace(header.first, value); } } + web_request.body = absl::StrCat(request.GetBody().GetRawData()); + + if (debug::kRequestEnabled) { + std::stringstream request_stream; + request_stream << "HTTP REQUEST====>" << std::endl; + request_stream << web_request.method << " " << web_request.url << std::endl; + for (const auto& header : web_request.headers) { + request_stream << header.first << ": " << header.second << std::endl; + } + request_stream << std::endl; + request_stream << "body size: " << request.GetBody().GetRawData().size() + << std::endl; + NEARBY_LOGS(VERBOSE) << request_stream.str(); + } + + absl::StatusOr web_response = + api::ImplementationPlatform::SendRequest(web_request); + + if (!web_response.ok()) { + return web_response.status(); + } + + if (debug::kResponseEnabled) { + std::stringstream response_stream; + response_stream << "HTTP RESPONSE====>" << std::endl; + response_stream << "url: " << web_request.url << std::endl; + response_stream << web_response->status_code << " " + << web_response->status_text << std::endl; + for (const auto& header : web_response->headers) { + response_stream << header.first << ": " << header.second << std::endl; + } + response_stream << std::endl; + response_stream << "body size: " << web_response->body.size() << std::endl; + NEARBY_LOGS(VERBOSE) << response_stream.str(); + } + + HttpResponse response; + + response.SetStatusCode( + static_cast(web_response->status_code)); + response.SetReasonPhrase(web_response->status_text); + for (const auto& header : web_response->headers) { + response.AddHeader(header.first, header.second); + } + response.SetBody(web_response->body); + + return response; } } // namespace network diff --git a/internal/network/http_client_impl.h b/internal/network/http_client_impl.h index d0e0646c..6a350927 100644 --- a/internal/network/http_client_impl.h +++ b/internal/network/http_client_impl.h @@ -1,4 +1,4 @@ -// Copyright 2022 Google LLC +// Copyright 2022-2023 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -16,21 +16,20 @@ #define THIRD_PARTY_NEARBY_INTERNAL_NETWORK_HTTP_CLIENT_IMPL_H_ #include -#include // NOLINT -#include // NOLINT -#include -#include +#include #include "absl/base/thread_annotations.h" -#include "absl/synchronization/mutex.h" +#include "absl/status/statusor.h" #include "internal/network/http_client.h" +#include "internal/platform/multi_thread_executor.h" +#include "internal/platform/mutex.h" namespace nearby { namespace network { class NearbyHttpClient : public HttpClient { public: - NearbyHttpClient() = default; + NearbyHttpClient(); ~NearbyHttpClient() override = default; NearbyHttpClient(const NearbyHttpClient&) = default; @@ -38,15 +37,19 @@ class NearbyHttpClient : public HttpClient { NearbyHttpClient(NearbyHttpClient&&) = default; NearbyHttpClient& operator=(NearbyHttpClient&&) = default; + // Starts HTTP request in asynchronization mode. void StartRequest(const HttpRequest& request, std::function&)> callback) override ABSL_LOCKS_EXCLUDED(mutex_); - private: - void CleanThreads() ABSL_SHARED_LOCKS_REQUIRED(mutex_); + // Gets HTTP response in synchronization mode. + absl::StatusOr GetResponse(const HttpRequest& request) override; - absl::Mutex mutex_; - std::vector> http_threads_ ABSL_GUARDED_BY(mutex_); + private: + absl::StatusOr InternalGetResponse(const HttpRequest& request); + + Mutex mutex_; + std::unique_ptr network_executor_ = nullptr; }; } // namespace network diff --git a/internal/network/http_client_impl_test.cc b/internal/network/http_client_impl_test.cc index 2f24e8ad..887e2265 100644 --- a/internal/network/http_client_impl_test.cc +++ b/internal/network/http_client_impl_test.cc @@ -1,4 +1,4 @@ -// Copyright 2021 Google LLC +// Copyright 2021-2023 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -89,7 +89,7 @@ class NearbyHttpClientTest : public ::testing::Test { api::WebRequest GetWebRequest() { return api::GetContext()->web_request; } - absl::StatusOr GetResponse( + absl::StatusOr GetResponseAsync( absl::string_view url, HttpRequestMethod method, const std::multimap& headers, absl::string_view body) { @@ -121,6 +121,28 @@ class NearbyHttpClientTest : public ::testing::Test { return result; } + absl::StatusOr GetResponse( + absl::string_view url, HttpRequestMethod method, + const std::multimap& headers, + absl::string_view body) { + absl::StatusOr result; + absl::StatusOr request_url = Url::Create(url); + if (!request_url.ok()) { + return request_url.status(); + } + + HttpRequest request{request_url.value()}; + auto it = headers.begin(); + while (it != headers.end()) { + request.AddHeader(it->first, it->second); + ++it; + } + request.SetMethod(method); + request.SetBody(body); + + return client_.GetResponse(request); + } + void CheckHeader(const std::multimap& headers, absl::string_view key, absl::string_view expected_value) { auto it = headers.find(std::string(key)); @@ -146,8 +168,8 @@ namespace { TEST_F(NearbyHttpClientTest, TestGet) { MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}}, "web content"); - auto result = - GetResponse("http://www.google.com", HttpRequestMethod::kGet, {}, ""); + auto result = GetResponseAsync("http://www.google.com", + HttpRequestMethod::kGet, {}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -165,8 +187,8 @@ TEST_F(NearbyHttpClientTest, TestGet) { TEST_F(NearbyHttpClientTest, TestGetWithQuery) { MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}}, "web content"); - auto result = GetResponse("http://www.google.com?name=name1&age=36", - HttpRequestMethod::kGet, {}, ""); + auto result = GetResponseAsync("http://www.google.com?name=name1&age=36", + HttpRequestMethod::kGet, {}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -180,8 +202,8 @@ TEST_F(NearbyHttpClientTest, TestGetWithQuery) { TEST_F(NearbyHttpClientTest, TestGetWithErrorResult) { MockFailedResponse(absl::InternalError("no connection.")); - auto result = GetResponse("http://www.google.com?name=name1&age=36", - HttpRequestMethod::kGet, {}, ""); + auto result = GetResponseAsync("http://www.google.com?name=name1&age=36", + HttpRequestMethod::kGet, {}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -192,6 +214,24 @@ TEST_F(NearbyHttpClientTest, TestGetWithErrorResult) { EXPECT_FALSE(result.ok()); } +TEST_F(NearbyHttpClientTest, TestPostAsync) { + MockResponse(HttpStatusCode::kHttpNoContent, "OK", + {{"Content_Type", "text/html"}}, ""); + auto result = GetResponseAsync("http://www.google.com", + HttpRequestMethod::kPost, {}, ""); + + // Checks request. + api::WebRequest web_request = GetWebRequest(); + EXPECT_EQ(web_request.url, "http://www.google.com"); + EXPECT_EQ(web_request.method, "POST"); + + // Checks response. + ASSERT_TRUE(result.ok()); + EXPECT_EQ(result->GetStatusCode(), HttpStatusCode::kHttpNoContent); + HttpResponseBody body = result->GetBody(); + EXPECT_TRUE(body.empty()); +} + TEST_F(NearbyHttpClientTest, TestPost) { MockResponse(HttpStatusCode::kHttpNoContent, "OK", {{"Content_Type", "text/html"}}, ""); @@ -210,12 +250,12 @@ TEST_F(NearbyHttpClientTest, TestPost) { EXPECT_TRUE(body.empty()); } -TEST_F(NearbyHttpClientTest, TestPostWithHeader) { +TEST_F(NearbyHttpClientTest, TestPostWithHeaderAsync) { MockResponse(HttpStatusCode::kHttpNoContent, "OK", {{"Content_Type", "text/html"}}, ""); auto result = - GetResponse("http://www.google.com", HttpRequestMethod::kPost, - {{"Content_Type", "text/json"}, {"size", "596"}}, ""); + GetResponseAsync("http://www.google.com", HttpRequestMethod::kPost, + {{"Content_Type", "text/json"}, {"size", "596"}}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -231,11 +271,11 @@ TEST_F(NearbyHttpClientTest, TestPostWithHeader) { EXPECT_EQ(result->GetBody().GetRawData(), ""); } -TEST_F(NearbyHttpClientTest, TestPostWithErrorResult) { +TEST_F(NearbyHttpClientTest, TestPostWithErrorResultAsync) { MockFailedResponse(absl::UnauthenticatedError("no user.")); auto result = - GetResponse("http://www.google.com", HttpRequestMethod::kPost, - {{"Content_Type", "text/json"}, {"size", "596"}}, ""); + GetResponseAsync("http://www.google.com", HttpRequestMethod::kPost, + {{"Content_Type", "text/json"}, {"size", "596"}}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -249,11 +289,11 @@ TEST_F(NearbyHttpClientTest, TestPostWithErrorResult) { ASSERT_FALSE(result.ok()); } -TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreads) { +TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreadsAsync) { MockResponse(HttpStatusCode::kHttpOk, "OK", {{"Content_Type", "text/html"}}, "web content"); - auto result = - GetResponse("http://www.google.com", HttpRequestMethod::kGet, {}, ""); + auto result = GetResponseAsync("http://www.google.com", + HttpRequestMethod::kGet, {}, ""); // Checks request. api::WebRequest web_request = GetWebRequest(); @@ -263,8 +303,8 @@ TEST_F(NearbyHttpClientTest, TestRequestWithCleanThreads) { // Checks response. ASSERT_TRUE(result.ok()); - result = - GetResponse("http://www.youtube.com", HttpRequestMethod::kGet, {}, ""); + result = GetResponseAsync("http://www.youtube.com", HttpRequestMethod::kGet, + {}, ""); ASSERT_TRUE(result.ok()); } diff --git a/internal/platform/BUILD b/internal/platform/BUILD index c3d09329..1128b4b7 100644 --- a/internal/platform/BUILD +++ b/internal/platform/BUILD @@ -342,6 +342,7 @@ cc_library( "//fastpair:__subpackages__", "//internal/base:__subpackages__", "//internal/flags:__subpackages__", + "//internal/network:__subpackages__", "//internal/platform/implementation/windows:__subpackages__", "//internal/test:__subpackages__", "//internal/weave:__pkg__",