kinda working with iota iterators

Author: scotthart

ci/cloudbuild/builds/integration-production.sh

@@ -27,7 +27,7 @@ export CC=clang
 export CXX=clang++
 
 mapfile -t args < <(bazel::common_args)
-io::run bazel test "${args[@]}" --test_tag_filters=-integration-test "${BAZEL_TARGETS[@]}"
+#io::run bazel test "${args[@]}" --test_tag_filters=-integration-test "${BAZEL_TARGETS[@]}"
 
 excluded_rules=(
   "-//examples:grpc_credential_types"
@@ -41,6 +41,6 @@ excluded_rules=(
 
 io::log_h2 "Running the integration tests against prod"
 mapfile -t integration_args < <(integration::bazel_args)
-io::run bazel test "${args[@]}" "${integration_args[@]}" \
-  --cache_test_results="auto" --test_tag_filters="integration-test,-ud-only" \
-  -- "${BAZEL_TARGETS[@]}" "${excluded_rules[@]}"
+io::run bazel test "${args[@]}" "${integration_args[@]}" --test_output=all \
+  --test_filter="*ReadRowsAllRows*" --test_timeout=30 \
+  //google/cloud/bigtable/tests:data_integration_test

ci/cloudbuild/builds/lib/integration.sh

@@ -51,7 +51,7 @@ function integration::bazel_args() {
 
   # Integration tests are inherently flaky. Make up to three attempts to get the
   # test passing.
-  args+=(--flaky_test_attempts=3)
+  #args+=(--flaky_test_attempts=3)
 
   args+=(
     # Common settings

google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.cc

@@ -105,6 +105,7 @@ std::unique_ptr<google::cloud::internal::StreamingReadRpc<
 BigtableRandomTwoLeastUsed::SampleRowKeys(
     std::shared_ptr<grpc::ClientContext> context, Options const& options,
     google::bigtable::v2::SampleRowKeysRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->SampleRowKeys(std::move(context), options, request);
@@ -121,6 +122,7 @@ StatusOr<google::bigtable::v2::MutateRowResponse>
 BigtableRandomTwoLeastUsed::MutateRow(
     grpc::ClientContext& context, Options const& options,
     google::bigtable::v2::MutateRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->MutateRow(context, options, request);
@@ -150,6 +152,7 @@ StatusOr<google::bigtable::v2::CheckAndMutateRowResponse>
 BigtableRandomTwoLeastUsed::CheckAndMutateRow(
     grpc::ClientContext& context, Options const& options,
     google::bigtable::v2::CheckAndMutateRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->CheckAndMutateRow(context, options, request);
@@ -161,6 +164,7 @@ StatusOr<google::bigtable::v2::PingAndWarmResponse>
 BigtableRandomTwoLeastUsed::PingAndWarm(
     grpc::ClientContext& context, Options const& options,
     google::bigtable::v2::PingAndWarmRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->PingAndWarm(context, options, request);
@@ -172,6 +176,7 @@ StatusOr<google::bigtable::v2::ReadModifyWriteRowResponse>
 BigtableRandomTwoLeastUsed::ReadModifyWriteRow(
     grpc::ClientContext& context, Options const& options,
     google::bigtable::v2::ReadModifyWriteRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->ReadModifyWriteRow(context, options, request);
@@ -183,6 +188,7 @@ StatusOr<google::bigtable::v2::PrepareQueryResponse>
 BigtableRandomTwoLeastUsed::PrepareQuery(
     grpc::ClientContext& context, Options const& options,
     google::bigtable::v2::PrepareQueryRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->PrepareQuery(context, options, request);
@@ -195,6 +201,7 @@ std::unique_ptr<google::cloud::internal::StreamingReadRpc<
 BigtableRandomTwoLeastUsed::ExecuteQuery(
     std::shared_ptr<grpc::ClientContext> context, Options const& options,
     google::bigtable::v2::ExecuteQueryRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->ExecuteQuery(std::move(context), options, request);
@@ -214,6 +221,7 @@ BigtableRandomTwoLeastUsed::AsyncReadRows(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::ReadRowsRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result =
@@ -234,6 +242,7 @@ BigtableRandomTwoLeastUsed::AsyncSampleRowKeys(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::SampleRowKeysRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->AsyncSampleRowKeys(cq, std::move(context),
@@ -253,6 +262,7 @@ BigtableRandomTwoLeastUsed::AsyncMutateRow(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::MutateRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result =
@@ -268,6 +278,7 @@ BigtableRandomTwoLeastUsed::AsyncMutateRows(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::MutateRowsRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->AsyncMutateRows(cq, std::move(context),
@@ -288,6 +299,7 @@ BigtableRandomTwoLeastUsed::AsyncCheckAndMutateRow(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::CheckAndMutateRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->AsyncCheckAndMutateRow(cq, std::move(context),
@@ -302,6 +314,7 @@ BigtableRandomTwoLeastUsed::AsyncReadModifyWriteRow(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::ReadModifyWriteRowRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->AsyncReadModifyWriteRow(cq, std::move(context),
@@ -316,6 +329,7 @@ BigtableRandomTwoLeastUsed::AsyncPrepareQuery(
     std::shared_ptr<grpc::ClientContext> context,
     google::cloud::internal::ImmutableOptions options,
     google::bigtable::v2::PrepareQueryRequest const& request) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   auto child = Child();
   auto stub = child->AcquireStub();
   auto result = stub->AsyncPrepareQuery(cq, std::move(context),
@@ -324,7 +338,7 @@ BigtableRandomTwoLeastUsed::AsyncPrepareQuery(
   return result;
 }
 
-std::shared_ptr<StubUsageWrapper<BigtableStub>>
+std::shared_ptr<ChannelUsageWrapper<BigtableStub>>
 BigtableRandomTwoLeastUsed::Child() {
   std::cout << __PRETTY_FUNCTION__ << std::endl;
   return pool_->GetChannelRandomTwoLeastUsed();

google/cloud/bigtable/internal/bigtable_random_two_least_used_decorator.h

@@ -130,7 +130,7 @@ class BigtableRandomTwoLeastUsed : public BigtableStub {
       google::bigtable::v2::PrepareQueryRequest const& request) override;
 
  private:
-  std::shared_ptr<StubUsageWrapper<BigtableStub>> Child();
+  std::shared_ptr<ChannelUsageWrapper<BigtableStub>> Child();
 
   //  std::mutex mu_;
   std::shared_ptr<DynamicChannelPool<BigtableStub>> pool_;

google/cloud/bigtable/internal/bigtable_stub_factory.cc

@@ -81,13 +81,16 @@ std::shared_ptr<BigtableStub> CreateBigtableStubRandomTwoLeastUsed(
     std::function<std::shared_ptr<BigtableStub>(int)>
         refreshing_channel_stub_factory,
     std::shared_ptr<ConnectionRefreshState> refresh_state) {
+  std::cout << __PRETTY_FUNCTION__ << std::endl;
   std::vector<std::shared_ptr<BigtableStub>> children(
       (std::max)(1, options.get<GrpcNumChannelsOption>()));
   int id = 0;
   std::generate(children.begin(), children.end(),
                 [&id, &refreshing_channel_stub_factory] {
                   return refreshing_channel_stub_factory(id++);
                 });
+  std::cout << __PRETTY_FUNCTION__ << ": children.size()=" << children.size()
+            << std::endl;
   return std::make_shared<BigtableRandomTwoLeastUsed>(
       std::move(cq), std::move(refresh_state), refreshing_channel_stub_factory,
       std::move(children));
@@ -103,9 +106,9 @@ std::shared_ptr<BigtableStub> CreateDecoratedStubs(
       options.get<bigtable::MaxConnectionRefreshOption>());
 
   std::shared_ptr<BigtableStub> stub;
-  if (options.has<ChannelSelectionStrategyOption>() &&
-      options.get<ChannelSelectionStrategyOption>() ==
-          ChannelSelectionStrategy::kRandomTwoLeastUsed) {
+  if (options.has<bigtable::experimental::ChannelPoolTypeOption>() &&
+      options.get<bigtable::experimental::ChannelPoolTypeOption>() ==
+          bigtable::experimental::ChannelPoolType::kDynamic) {
     auto refreshing_channel_stub_factory = [stub_factory, cq_impl, refresh,
                                             &auth, options](int id) {
       auto channel = CreateGrpcChannel(*auth, options, id);

google/cloud/bigtable/internal/bigtable_stub_factory.h

@@ -29,7 +29,7 @@ namespace cloud {
 namespace bigtable_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 
-enum class ChannelSelectionStrategy { kNone, kRoundRobin, kRandomTwoLeastUsed };
+enum class ChannelSelectionStrategy { kRoundRobin, kRandomTwoLeastUsed };
 
 struct ChannelSelectionStrategyOption {
   using Type = ChannelSelectionStrategy;

google/cloud/bigtable/internal/connection_refresh_state.cc

@@ -53,10 +53,20 @@ bool ConnectionRefreshState::enabled() const {
   return max_conn_refresh_period_.count() != 0;
 }
 
+void LogFailedConnectionRefresh(Status const& conn_status) {
+  if (!conn_status.ok()) {
+    GCP_LOG(WARNING) << "Failed to refresh connection. Error: " << conn_status;
+  }
+}
+
 void ScheduleChannelRefresh(
     std::shared_ptr<internal::CompletionQueueImpl> const& cq_impl,
     std::shared_ptr<ConnectionRefreshState> const& state,
-    std::shared_ptr<grpc::Channel> const& channel) {
+    std::shared_ptr<grpc::Channel> const& channel,
+    std::function<void(Status const&)> connection_status_fn) {
+  if (!connection_status_fn) {
+    connection_status_fn = LogFailedConnectionRefresh;
+  }
   // The timers will only hold weak pointers to the channel or to the
   // completion queue, so if either of them are destroyed, the timer chain
   // will simply not continue.
@@ -66,7 +76,9 @@ void ScheduleChannelRefresh(
   using TimerFuture = future<StatusOr<std::chrono::system_clock::time_point>>;
   auto timer_future =
       cq.MakeRelativeTimer(state->RandomizedRefreshDelay())
-          .then([weak_channel, weak_cq_impl, state](TimerFuture fut) {
+          .then([weak_channel, weak_cq_impl, state,
+                 connection_status_fn =
+                     std::move(connection_status_fn)](TimerFuture fut) {
             if (!fut.get()) {
               // Timer cancelled.
               return;
@@ -79,17 +91,18 @@ void ScheduleChannelRefresh(
             cq.AsyncWaitConnectionReady(
                   channel,
                   std::chrono::system_clock::now() + kConnectionReadyTimeout)
-                .then([weak_channel, weak_cq_impl, state](future<Status> fut) {
+                .then([weak_channel, weak_cq_impl, state,
+                       connection_status_fn = std::move(connection_status_fn)](
+                          future<Status> fut) {
                   auto conn_status = fut.get();
-                  if (!conn_status.ok()) {
-                    GCP_LOG(WARNING) << "Failed to refresh connection. Error: "
-                                     << conn_status;
-                  }
+                  if (connection_status_fn) connection_status_fn(conn_status);
+
                   auto channel = weak_channel.lock();
                   if (!channel) return;
                   auto cq_impl = weak_cq_impl.lock();
                   if (!cq_impl) return;
-                  ScheduleChannelRefresh(cq_impl, state, channel);
+                  ScheduleChannelRefresh(cq_impl, state, channel,
+                                         std::move(connection_status_fn));
                 });
           });
   state->timers().RegisterTimer(std::move(timer_future));

google/cloud/bigtable/internal/connection_refresh_state.h

@@ -86,7 +86,8 @@ class ConnectionRefreshState {
 void ScheduleChannelRefresh(
     std::shared_ptr<internal::CompletionQueueImpl> const& cq,
     std::shared_ptr<ConnectionRefreshState> const& state,
-    std::shared_ptr<grpc::Channel> const& channel);
+    std::shared_ptr<grpc::Channel> const& channel,
+    std::function<void(Status const&)> connection_status_fn = {});
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace bigtable_internal