When to create pubsub client subscription?

[iamsamskup]

Greetings everyone,

I'm looking for clarification on pubsub subscriptions when using the C++ client library. After reading the documentation it seems that two different types of "subscriptions" exist. The first type are the subscriptions that exist on GCP and attach directly to topics. The second exist inside user code and are used to connect to the subscriptions that exist on GCP. I was hoping someone would double check my logic to make sure I'm using the client library correctly.

In my project I need the capability to deploy N number of copies of a particular microservice that all run concurrently. I have a set number of topics which each separate microservice will need to subscribe to. The logic flow is as follows:

1. Because subscriptions to topics in GCP cannot have duplicate names I insert a random string of numbers to differentiate each microservice subscription. Then using the following logic I create the topic subscription in GCP using:

google::cloud::pubsub::SubscriptionAdminClient subscription_admin_client(
      google::cloud::pubsub::MakeSubscriptionAdminConnection());


  for (auto topicSubPair : topicSubscriptionList) { // a pair that maps the topic to the unique subscription id

      auto sub = subscription_admin_client.CreateSubscription(
          google::cloud::pubsub::Topic(project_id, std::move(topicSubPair.first)),
          google::cloud::pubsub::Subscription(project_id, std::move(topicSubPair.second)));
}

I run this code for each topic and subscription that should bind to it.

2. Once that step is complete I want to create subscribers in my code to pull from these newly created topics. To do so it seems like I have to first create a different subscription object using:

    for (auto &subscriptionId : topicSubscriptionList) {
      google::cloud::pubsub::Subscription tmpSub(projectID, subscriptionId.second);
      subscriptions.emplace_back(tmpSub);
    }

and then for each subscription object created above use it to create a subscriber:

    for (auto &subscription : subscriptions) {
      auto subscriber = google::cloud::pubsub::Subscriber(google::cloud::pubsub::MakeSubscriberConnection(subscription));
      auto session = subscriber.Subscribe(
          [&,this](google::cloud::pubsub::Message const &m, google::cloud::pubsub::AckHandler h) {
           // need to route messages according to the subscription id 
            std::string subscriptionTopic{subscription.subscription_id()};
            std::move(h).ack();
           
            //mutex lock here
            // messages is a class member dequeue to hold messages from all topics. Read in a loop in main thread
            messages.emplace_back(std::pair(subscriptionTopic, m.data())); 
          });
    }

Once all the setup is complete I should be able to start pulling messages and reading them from the messages queue I created. Thanks for the help!

[coryan]

In principle this will work. You did not mention the expected message rates over these topics, some optimization may be required if the message rate is sufficiently high (at a guess, anything over 1,000 messages/second).

I expect that fine tuning the implicitly created background thread pools, connection pools, and other low-level details may require some work. Here are some starting ideas:

• Each subscriber in you loop creates a thread pool to service the request. You may need to create your own thread pool using GrpcCompletionQueueOption, or reduce the size of the pool on each subscriber GrpcBackgroundThreadPoolSizeOption.

• Each subscriber creates a pool of gRPC channels, the default count is 4. By default the channels use a (globally) shared set of sockets. So effectively you only create 4 sockets, no matter how many subscribers. Because all the subscribers round-robin independently, all the requests will land on the first socket. You may need to set some low-level gRPC options using GrpcChannelArgumentsNativeOption. The ones I have in mind are:

  • GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL will use a per-channel socket pool (instead of a global pool)
  • GRPC_ARG_CHANNEL_POOL_DOMAIN can be used to force a different global pool on each subscriber (or a subset of subscribers).

• In any case, I would use GrpcNumChannelsOption to reduce the number of channels per subscriber, since you are effectively using only one.

[iamsamskup]

Awesome, thanks for getting back to me so quickly. These microservices are maybe going to get a message every half hour even at scale so I definitely have room to play with in terms of optimization.

[coryan]

FWIW, I just sent #10327 for review. It may help your case as you could write:

    namespace pubsub = google::cloud::pubsub
    auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(pubsub::Subscription("unused", "unused")));
    // Note the subscriber is shared across all sessions.
    for (auto &subscription : subscriptions) {
      // Note: you need to make a copy of `subscription` or "Bad Things"[tm] await you.
      auto callback =  [subscripption,this](google::cloud::pubsub::Message const &m, google::cloud::pubsub::AckHandler h) {
           // need to route messages according to the subscription id 
            std::string subscriptionTopic{subscription.subscription_id()};
           
            //mutex lock here
            // messages is a class member dequeue to hold messages from all topics. Read in a loop in main thread
            messages.emplace_back(std::pair(subscriptionTopic, m.data()));
            // If you care about exception safely, you probably want to ack after the message is sucessfully queued.
            std::move(h).ack();
          };
      auto session = subscriber.Subscribe(callback, google::cloud::Options{}.set<pubsub::SubscriptionOption>(subscription));
    }