Skip to content

feat(storage): implement internal pre-warmed stream dispatch and caching for BiDi reads #16259

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kalragauri wants to merge 3 commits into
base: main
Choose a base branch
from
+261 −28
Open

feat(storage): implement internal pre-warmed stream dispatch and caching for BiDi reads #16259

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9c450d3
feat(storage): implement internal pre-warmed stream dispatch and cach…
kalragauri Jul 15, 2026
ab05549
Address feedback from Gemini code assistant
kalragauri Jul 15, 2026
e944f9d
Merge branch 'main' into feat/fast-open
kalragauri Jul 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions google/cloud/storage/async/client.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,10 @@ future<StatusOr<ObjectDescriptor>> AsyncClient::Open(
future<StatusOr<ObjectDescriptor>> AsyncClient::Open(
google::storage::v2::BidiReadObjectSpec spec, Options opts) {
return connection_
->Open({std::move(spec), google::cloud::internal::MergeOptions(
std::move(opts), connection_->options())})
->Open({std::move(spec),
google::cloud::internal::MergeOptions(std::move(opts),
connection_->options()),
/*initial_read_range=*/absl::nullopt})

@v-pratap v-pratap Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So in all the cases will there be only one read range or we can give multiple read ranges?

@kalragauri kalragauri Jul 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For now, we are only allowing a single range. This optimization targets the 2 round trip penalty of stream setup + read intent. Once the stream is open, subsequent ranges can be multiplexed onto the established stream asynchronously without blocking the initial application threads.

For advanced use cases that require scattered multiplexed reads, ObjectDescriptorImpl already supports assigning multiple concurrent read_ids downstream once the session is opened:

google-cloud-cpp/google/cloud/storage/internal/async/object_descriptor_impl.cc

Lines 206 to 211 in 40d3ab0

auto const id = ++read_id_generator_;
it->active_ranges.emplace(id, range);
auto& read_range = *it->stream->next_request.add_read_ranges();
read_range.set_read_id(id);
read_range.set_read_offset(p.start);
read_range.set_read_length(p.length);

@v-pratap v-pratap Jul 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I understand that we have feature for sending multiple read requests later, but I think in a request we can send 100 read ranges and we are adding this feature to make our api faster. I just want to know whether sending multiple request with Open would make it more faster or sending one read range would be enough to warm up the stream?

.then([](auto f) -> StatusOr<ObjectDescriptor> {
auto connection = f.get();
if (!connection) return std::move(connection).status();
Expand Down
10 changes: 10 additions & 0 deletions google/cloud/storage/async/connection.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,15 @@ class AsyncConnection {
virtual future<StatusOr<google::storage::v2::Object>> InsertObject(
InsertObjectParams p) = 0;

/// A parameter to specify the initial offset and length of an eagerly-started
/// read.
struct InitialReadRange {
/// The starting offset of the read.
std::int64_t offset;
/// The number of bytes to read.
std::int64_t length;
};

/**
* A thin wrapper around the `Open()` parameters.
*
Expand All @@ -85,6 +94,7 @@ class AsyncConnection {
struct OpenParams {
google::storage::v2::BidiReadObjectSpec read_spec;
Options options;
absl::optional<InitialReadRange> initial_read_range;
};

/// Open an object to perform multiple reads.
Expand Down
12 changes: 9 additions & 3 deletions google/cloud/storage/internal/async/connection_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -192,6 +192,12 @@ future<StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>>>
AsyncConnectionImpl::Open(OpenParams p) {
auto initial_request = google::storage::v2::BidiReadObjectRequest{};
*initial_request.mutable_read_object_spec() = p.read_spec;
if (p.initial_read_range) {
auto* range = initial_request.add_read_ranges();
range->set_read_offset(p.initial_read_range->offset);
range->set_read_length(p.initial_read_range->length);
range->set_read_id(1);
}
auto current = internal::MakeImmutableOptions(p.options);
// Get the policy factory and immediately create a policy.
auto resume_policy = current->get<storage::ResumePolicyOption>()();
Expand Down Expand Up @@ -234,8 +240,8 @@ AsyncConnectionImpl::Open(OpenParams p) {
using ReturnType = std::shared_ptr<storage::ObjectDescriptorConnection>;
return pending.then([rp = std::move(resume_policy), fa = std::move(factory),
rs = std::move(p.read_spec),
options = std::move(p.options), refresh = refresh_](
auto f) mutable -> StatusOr<ReturnType> {
options = std::move(p.options), initial_read_range = p.initial_read_range,
refresh = refresh_](auto f) mutable -> StatusOr<ReturnType> {
auto result = f.get();
if (!result) return std::move(result).status();

Expand All @@ -255,7 +261,7 @@ AsyncConnectionImpl::Open(OpenParams p) {
};
auto impl = std::make_shared<ObjectDescriptorImpl>(
std::move(rp), std::move(fa), std::move(rs), std::move(result->stream),
std::move(options), std::move(transport_ok));
initial_read_range, std::move(options), std::move(transport_ok));
impl->Start(std::move(result->first_response));
return ReturnType(impl);
});
Expand Down
12 changes: 8 additions & 4 deletions google/cloud/storage/internal/async/connection_impl_open_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,8 @@ TEST(AsyncConnectionImplTest, OpenSimple) {

auto request = google::storage::v2::BidiReadObjectSpec{};
ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request));
auto pending = connection->Open({std::move(request), connection->options()});
auto pending = connection->Open(
{std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt});

auto next = sequencer.PopFrontWithName();
EXPECT_EQ(next.second, "Start");
Expand Down Expand Up @@ -310,7 +311,8 @@ TEST(AsyncConnectionImplTest, HandleRedirectErrors) {

auto request = google::storage::v2::BidiReadObjectSpec{};
ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest0, &request));
auto pending = connection->Open({std::move(request), connection->options()});
auto pending = connection->Open(
{std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt});

for (int i = 0; i != kRetryAttempts + 1; ++i) {
auto next = sequencer.PopFrontWithName();
Expand Down Expand Up @@ -354,7 +356,8 @@ TEST(AsyncConnectionImplTest, StopOnPermanentError) {

auto request = google::storage::v2::BidiReadObjectSpec{};
ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request));
auto pending = connection->Open({std::move(request), connection->options()});
auto pending = connection->Open(
{std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt});

auto next = sequencer.PopFrontWithName();
EXPECT_EQ(next.second, "Start");
Expand Down Expand Up @@ -390,7 +393,8 @@ TEST(AsyncConnectionImplTest, TooManyTransienErrors) {

auto request = google::storage::v2::BidiReadObjectSpec{};
ASSERT_TRUE(TextFormat::ParseFromString(kExpectedRequest, &request));
auto pending = connection->Open({std::move(request), connection->options()});
auto pending = connection->Open(
{std::move(request), connection->options(), /*initial_read_range=*/absl::nullopt});

for (int i = 0; i != kRetryAttempts + 1; ++i) {
auto next = sequencer.PopFrontWithName();
Expand Down
31 changes: 25 additions & 6 deletions google/cloud/storage/internal/async/object_descriptor_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,11 +39,16 @@ ObjectDescriptorImpl::ObjectDescriptorImpl(
std::unique_ptr<storage::ResumePolicy> resume_policy,
OpenStreamFactory make_stream,
google::storage::v2::BidiReadObjectSpec read_object_spec,
std::shared_ptr<OpenStream> stream, Options options,
std::shared_ptr<OpenStream> stream,
absl::optional<storage::AsyncConnection::InitialReadRange> initial_read_range,
Options options,
std::function<bool()> transport_ok)
: resume_policy_prototype_(std::move(resume_policy)),
make_stream_(std::move(make_stream)),
read_object_spec_(std::move(read_object_spec)),
initial_read_range_(std::move(initial_read_range)),
initial_cache_consumed_(!initial_read_range.has_value()),
read_id_generator_(initial_read_range.has_value() ? 1 : 0),
options_(std::move(options)),
transport_ok_(std::move(transport_ok)) {
stream_manager_ = std::make_unique<StreamManager>(
Expand Down Expand Up @@ -203,12 +208,26 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
}

auto it = stream_manager_->GetLeastBusyStream();
auto const id = ++read_id_generator_;
bool is_cache_hit = false;
if (!initial_cache_consumed_ && initial_read_range_.has_value()) {
auto const& initial = *initial_read_range_;
// Ensure both offset and length match.
if (p.start == initial.offset && limit.value_or(0) == initial.length) {
is_cache_hit = true;
}
initial_cache_consumed_ = true;
}

auto const id = is_cache_hit ? 1 : ++read_id_generator_;
it->active_ranges.emplace(id, range);
auto& read_range = *it->stream->next_request.add_read_ranges();
read_range.set_read_id(id);
read_range.set_read_offset(p.start);
read_range.set_read_length(p.length);

if (!is_cache_hit) {
auto& read_range = *it->stream->next_request.add_read_ranges();
read_range.set_read_id(id);
read_range.set_read_offset(p.start);
read_range.set_read_length(p.length);
}

Flush(std::move(lk), it);

if (!internal::TracingEnabled(options_)) {
Expand Down
17 changes: 12 additions & 5 deletions google/cloud/storage/internal/async/object_descriptor_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
#ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H
#define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_STORAGE_INTERNAL_ASYNC_OBJECT_DESCRIPTOR_IMPL_H

#include "google/cloud/storage/async/connection.h"
#include "google/cloud/storage/async/object_descriptor_connection.h"
#include "google/cloud/storage/async/resume_policy.h"
#include "google/cloud/storage/internal/async/multi_stream_manager.h"
Expand Down Expand Up @@ -57,11 +58,14 @@ class ObjectDescriptorImpl
: public storage::ObjectDescriptorConnection,
public std::enable_shared_from_this<ObjectDescriptorImpl> {
public:
ObjectDescriptorImpl(std::unique_ptr<storage::ResumePolicy> resume_policy,
OpenStreamFactory make_stream,
google::storage::v2::BidiReadObjectSpec read_object_spec,
std::shared_ptr<OpenStream> stream, Options options = {},
std::function<bool()> transport_ok = {});
ObjectDescriptorImpl(
std::unique_ptr<storage::ResumePolicy> resume_policy,
OpenStreamFactory make_stream,
google::storage::v2::BidiReadObjectSpec read_object_spec,
std::shared_ptr<OpenStream> stream,
absl::optional<storage::AsyncConnection::InitialReadRange>
initial_read_range,
Options options = {}, std::function<bool()> transport_ok = {});
~ObjectDescriptorImpl() override;

// Start the read loop.
Expand Down Expand Up @@ -122,6 +126,9 @@ class ObjectDescriptorImpl
mutable std::mutex mu_;
google::storage::v2::BidiReadObjectSpec read_object_spec_;
absl::optional<google::storage::v2::Object> metadata_;
absl::optional<storage::AsyncConnection::InitialReadRange>
initial_read_range_;
bool initial_cache_consumed_ = false;
std::int64_t read_id_generator_ = 0;

Options options_;
Expand Down
Loading
Loading