Skip to content

Commit f7f9c02

Browse files
bajajneha27bajajneha27
committed
Refactoring
1 parent 176e6f8 commit f7f9c02

5 files changed

Lines changed: 77 additions & 42 deletions

File tree

‎google/cloud/storage/internal/async/connection_tracing.cc‎

Lines changed: 27 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414

1515
#include "google/cloud/storage/internal/async/connection_tracing.h"
1616
#include "google/cloud/storage/async/writer_connection.h"
17-
#include "google/cloud/storage/internal/async/default_options.h"
1817
#include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h"
1918
#include "google/cloud/storage/internal/async/reader_connection_tracing.h"
2019
#include "google/cloud/storage/internal/async/rewriter_connection_tracing.h"
@@ -42,7 +41,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
4241
public:
4342
explicit AsyncConnectionTracing(
4443
std::shared_ptr<storage::AsyncConnection> impl)
45-
: impl_(std::move(impl)) {}
44+
: impl_(std::move(impl)), cache_(BucketMetadataCache::Singleton()) {}
4645

4746
~AsyncConnectionTracing() override {
4847
for (auto& f : bg_tasks_) {
@@ -256,26 +255,33 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
256255
std::string const& bucket_name) {
257256
CleanupCompletedTasks();
258257

259-
if (!BucketMetadataCache::Singleton().StartFetch(bucket_name)) {
258+
if (!cache_.StartFetch(bucket_name)) {
260259
return;
261260
}
262261

263-
auto f = std::async(std::launch::async, [bucket_name, options]() {
264-
google::cloud::internal::OptionsSpan span(options);
265-
auto conn = MakeStorageConnection(options);
266-
auto const normalized =
267-
BucketMetadataCache::NormalizeBucketName(bucket_name);
268-
storage::internal::GetBucketMetadataRequest request(normalized);
269-
auto metadata = conn->GetBucketMetadata(request);
270-
if (metadata.ok()) {
271-
BucketMetadataCache::Singleton().Put(
272-
bucket_name, BucketCacheEntry::FromMetadata(*metadata));
273-
} else if (metadata.status().code() == StatusCode::kPermissionDenied) {
274-
BucketMetadataCache::Singleton().Put(
275-
bucket_name, {"projects/_/buckets/" + normalized, "global"});
262+
std::shared_ptr<storage::internal::StorageConnection> conn;
263+
{
264+
std::lock_guard<std::mutex> lk(mu_);
265+
if (!sync_conn_) {
266+
sync_conn_ = MakeStorageConnection(impl_->options());
276267
}
277-
BucketMetadataCache::Singleton().EndFetch(bucket_name);
278-
});
268+
conn = sync_conn_;
269+
}
270+
271+
auto f =
272+
std::async(std::launch::async, [bucket_name, options, conn, this]() {
273+
google::cloud::internal::OptionsSpan span(options);
274+
auto const normalized_bucket_name =
275+
BucketMetadataCache::NormalizeBucketName(bucket_name);
276+
storage::internal::GetBucketMetadataRequest request(
277+
normalized_bucket_name);
278+
auto metadata = conn->GetBucketMetadata(request);
279+
auto entry = BucketCacheEntry::Create(bucket_name, metadata);
280+
if (entry) {
281+
cache_.Put(bucket_name, std::move(*entry));
282+
}
283+
cache_.EndFetch(bucket_name);
284+
});
279285

280286
std::unique_lock<std::mutex> lk(mu_);
281287
bg_tasks_.push_back(std::move(f));
@@ -284,7 +290,7 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
284290
void EnrichSpan(opentelemetry::trace::Span& span, Options const& options,
285291
std::string const& bucket_name) {
286292
if (bucket_name.empty()) return;
287-
auto entry = BucketMetadataCache::Singleton().Get(bucket_name);
293+
auto entry = cache_.Get(bucket_name);
288294
if (entry.has_value()) {
289295
span.SetAttribute("gcp.resource.destination.id", entry->id);
290296
span.SetAttribute("gcp.resource.destination.location", entry->location);
@@ -294,6 +300,8 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
294300
}
295301

296302
std::shared_ptr<storage::AsyncConnection> impl_;
303+
BucketMetadataCache& cache_;
304+
std::shared_ptr<storage::internal::StorageConnection> sync_conn_;
297305
std::vector<std::future<void>> bg_tasks_;
298306
std::mutex mu_;
299307
};

‎google/cloud/storage/internal/bucket_metadata_cache.cc‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
#include "google/cloud/storage/internal/bucket_metadata_cache.h"
1616
#include "google/cloud/storage/bucket_metadata.h"
17+
#include "google/cloud/status.h"
1718
#include <mutex>
1819
#include <utility>
1920

@@ -39,6 +40,21 @@ BucketCacheEntry BucketCacheEntry::FromMetadata(
3940
std::move(loc)};
4041
}
4142

43+
absl::optional<BucketCacheEntry> BucketCacheEntry::Create(
44+
std::string const& bucket_name,
45+
StatusOr<storage::BucketMetadata> const& metadata) {
46+
if (metadata.ok()) {
47+
return FromMetadata(*metadata);
48+
}
49+
if (metadata.status().code() == StatusCode::kPermissionDenied) {
50+
return BucketCacheEntry{
51+
"projects/_/buckets/" +
52+
BucketMetadataCache::NormalizeBucketName(bucket_name),
53+
"global"};
54+
}
55+
return absl::nullopt;
56+
}
57+
4258
void BucketMetadataCache::MoveToFront(std::list<std::string>::iterator it) {
4359
list_.splice(list_.begin(), list_, it);
4460
}

‎google/cloud/storage/internal/bucket_metadata_cache.h‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,9 @@
1515
#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
1616
#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_BUCKET_METADATA_CACHE_H
1717

18+
#include "google/cloud/storage/bucket_metadata.h"
1819
#include "google/cloud/storage/version.h"
20+
#include "google/cloud/status_or.h"
1921
#include "absl/strings/match.h"
2022
#include "absl/types/optional.h"
2123
#include <cstddef>
@@ -28,11 +30,6 @@
2830

2931
namespace google {
3032
namespace cloud {
31-
namespace storage {
32-
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
33-
class BucketMetadata;
34-
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
35-
} // namespace storage
3633
namespace storage_internal {
3734
GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
3835

@@ -41,12 +38,15 @@ struct BucketCacheEntry {
4138
std::string location;
4239

4340
static BucketCacheEntry FromMetadata(storage::BucketMetadata const& m);
41+
static absl::optional<BucketCacheEntry> Create(
42+
std::string const& bucket_name,
43+
StatusOr<storage::BucketMetadata> const& metadata);
4444
};
4545

4646
class BucketMetadataCache {
4747
public:
4848
explicit BucketMetadataCache(std::size_t max_size = 10000)
49-
: max_size_(max_size) {}
49+
: max_size_(max_size == 0 ? 1 : max_size) {}
5050

5151
static BucketMetadataCache& Singleton();
5252

‎google/cloud/storage/internal/tracing_connection.cc‎

Lines changed: 13 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -70,20 +70,19 @@ void TracingConnection::MaybeTriggerBackgroundFetch(
7070
}
7171

7272
auto current_options = google::cloud::internal::SaveCurrentOptions();
73-
auto f = std::async(std::launch::async, [this, bucket_name,
74-
current_options]() {
75-
google::cloud::internal::OptionsSpan span(current_options);
76-
storage::internal::GetBucketMetadataRequest request(bucket_name);
77-
auto result = impl_->GetBucketMetadata(request);
78-
79-
if (result.ok()) {
80-
cache().Put(bucket_name, BucketCacheEntry::FromMetadata(*result));
81-
} else if (result.status().code() == StatusCode::kPermissionDenied) {
82-
cache().Put(bucket_name, {"projects/_/buckets/" + bucket_name, "global"});
83-
}
84-
85-
cache().EndFetch(bucket_name);
86-
});
73+
auto f =
74+
std::async(std::launch::async, [this, bucket_name, current_options]() {
75+
google::cloud::internal::OptionsSpan span(current_options);
76+
storage::internal::GetBucketMetadataRequest request(bucket_name);
77+
auto result = impl_->GetBucketMetadata(request);
78+
79+
auto entry = BucketCacheEntry::Create(bucket_name, result);
80+
if (entry) {
81+
cache().Put(bucket_name, std::move(*entry));
82+
}
83+
84+
cache().EndFetch(bucket_name);
85+
});
8786

8887
bg_tasks_.push_back(std::move(f));
8988
}

‎google/cloud/storage/tests/object_plenty_clients_serially_integration_test.cc‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,21 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
6868
EXPECT_EQ(StatusCode::kUnimplemented, num_fds_before_test.status().code());
6969
}
7070
std::size_t delta = 0;
71+
if (track_open_files) {
72+
// Warmup to find the maximum delta, accounting for any asynchronous
73+
// background requests (like IAM AllowedLocations) that might or might
74+
// not have opened their sockets yet when we measure.
75+
for (int i = 0; i != 5; ++i) {
76+
auto read_client = MakeIntegrationTestClient(options);
77+
auto stream = read_client.ReadObject(bucket_name_, object_name);
78+
char c;
79+
stream.read(&c, 1);
80+
auto num_fds_during_test = GetNumOpenFiles();
81+
ASSERT_STATUS_OK(num_fds_during_test);
82+
delta = (std::max)(delta, *num_fds_during_test - *num_fds_before_test);
83+
}
84+
}
85+
7186
for (int i = 0; i != 100; ++i) {
7287
auto read_client = MakeIntegrationTestClient(options);
7388
auto stream = read_client.ReadObject(bucket_name_, object_name);
@@ -76,9 +91,6 @@ TEST_F(ObjectPlentyClientsSeriallyIntegrationTest, PlentyClientsSerially) {
7691
if (track_open_files) {
7792
auto num_fds_during_test = GetNumOpenFiles();
7893
ASSERT_STATUS_OK(num_fds_during_test);
79-
if (delta == 0) {
80-
delta = *num_fds_during_test - *num_fds_before_test;
81-
}
8294
EXPECT_GE(*num_fds_before_test + delta, *num_fds_during_test)
8395
<< "Expect each client to create the same number of file descriptors"
8496
<< ", num_fds_before_test=" << *num_fds_before_test

0 commit comments

Comments
 (0)