Function worker

populate-bucket/populate_bucket.cc:267–331

Run the worker thread for a GKE batch job.

/// Run the worker thread for a GKE batch job.
void worker(boost::program_options::variables_map const& vm) {
  std::cout << "Running in worker mode" << std::endl;

  if (vm.count("project") == 0 or vm["project"].as<std::string>().empty()) {
    throw std::runtime_error("the `schedule` action requires --subscription");
  }
  if (vm.count("subscription") == 0) {
    throw std::runtime_error("the `schedule` action requires --subscription");
  }
  auto const concurrency = vm["concurrency"].as<int>();
  auto const project_id = vm["project"].as<std::string>();
  auto const subscription_id = vm["subscription"].as<std::string>();

  using namespace std::chrono_literals;
  using std::chrono::duration_cast;
  using std::chrono::milliseconds;
  auto const subscription = pubsub::Subscription(project_id, subscription_id);
  auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
      subscription,
      google::cloud::Options{}
          .set<pubsub::MaxOutstandingMessagesOption>(concurrency)
          .set<pubsub::MaxOutstandingBytesOption>(concurrency * 1024)
          .set<pubsub::MaxConcurrencyOption>(concurrency)
          .set<pubsub::MaxDeadlineTimeOption>(300s)
          .set<google::cloud::GrpcBackgroundThreadPoolSizeOption>(
              concurrency)));

  std::atomic<std::int64_t> latency{0};
  std::atomic<std::int64_t> attempts{0};
  std::atomic<std::int64_t> counter{0};
  auto handler = [&, cl = gcs::Client::CreateDefaultClient().value()](
                     pubsub::Message const& m, pubsub::AckHandler h) {
    auto const start = std::chrono::steady_clock::now();
    process_one_item(cl, m);
    auto const elapsed = std::chrono::steady_clock::now() - start;
    latency.fetch_add(duration_cast<milliseconds>(elapsed).count());
    attempts.fetch_add(h.delivery_attempt());
    ++counter;
    std::move(h).ack();
  };

  using namespace std::chrono_literals;
  using google::cloud::StatusCode;
  auto session = subscriber.Subscribe(handler);
  auto total = counter.exchange(0);
  auto mean = [&total](std::int64_t v) -> std::int64_t {
    if (total == 0) return 0;
    return v / total;
  };
  while (session.wait_for(30s) == std::future_status::timeout) {
    auto last = counter.exchange(0);
    total += last;
    std::cout << "Processed " << last << " work items"
              << ", latency=" << mean(latency.load())
              << ", attempts=" << mean(attempts.load()) << ", count=" << total
              << std::endl;
  }