Threading model in Pubsub client

[MrMoose]

Hello everyone,

I hope it is OK to ask another question about the threading model in the pubsub client.

I am a bit confused by the explanation of the pubsub::Subscriber behavior. Let me describe what I want to achieve first, this could make it easier:

My client code will be executed in what I will refer to as a "main" thread. I cannot block this for long. Here I want to create a subscription and then hook up to it. Then return. I want exactly one thread spawned that is then executing the lambda for each message that is pulled. Processing the message may take between 1 and 30 seconds, after which it shall be ack'ed or nack'ed and then the next message shall be pulled.
The main thread will then, at some point, come in to shut the pulling down. Until then it shall be out of the loop. Whether or not I provide that background thread myself or have the client do it implicitly is of no importance to me. As long as I control it.

As I understand it, the pubsub client will always create a thread pool in the background which will execute my lambda as a message comes in. I also gather that by using this, I could achieve what I want like this:

pubsub::SubscriberOptions subscriber_options;
	subscriber_options.set_max_concurrency(1);

auto subscriber = pubsub::Subscriber(pubsub::MakeSubscriberConnection(
		subscription.value(),
		subscriber_options);

auto sample = [](pubsub::Subscriber subscriber) {
		return subscriber.Subscribe(
			[&](pubsub::Message const &m, pubsub::AckHandler h) {
				std::cout << "Received message " << m << "\n";
				std::move(h).ack();
				
		});
	};

Is that correct?

I also have a choice of manually spawning a thread and have it execute Run() on a Completion Q. Much like asio would. But this is not necessary when doing it like I describe above, right?

Important is that I cannot have concurrent messages being pulled. Only one at a time. In other impls I use my own thread, synchronously pull in there, process, loop. This is also an option but not one I like to go for if above method works. I was only a bit confused by the docs of set_max_concurrency() stating that I need enough concurrency to execute multiple messages at a time. Which is the opposite of what I need.

Could you please clarify?

Best wishes and thanks for your help!

Moose.

[coryan]

I am a bit confused by the explanation of the pubsub::Subscriber behavior. Let me describe what I want to achieve first, this could make it easier:

Okay.

My client code will be executed in what I will refer to as a "main" thread. I cannot block this for long. Here I want to create a subscription and then hook up to it. Then return. I want exactly one thread spawned that is then executing the lambda for each message that is pulled. Processing the message may take between 1 and 30 seconds, after which it shall be ack'ed or nack'ed and then the next message shall be pulled.

While it is possible to configure the library to behave as-if this was happening, the current implementation uses streaming pulls. The library does not pull one message at a time, it just receives messages. You can describe how many messages the service is allowed to send at a time, but it is not exactly the same thing.

The main thread will then, at some point, come in to shut the pulling down. Until then it shall be out of the loop. Whether or not I provide that background thread myself or have the client do it implicitly is of no importance to me. As long as I control it.

Ack.

Is that correct?

That looks correct.

I also have a choice of manually spawning a thread and have it execute Run() on a Completion Q. Much like asio would. But this is not necessary when doing it like I describe above, right?

Correct.

Important is that I cannot have concurrent messages being pulled. Only one at a time. In other impls I use my own thread, synchronously pull in there, process, loop. This is also an option but not one I like to go for if above method works.

Ack.

I was only a bit confused by the docs of set_max_concurrency() stating that I need enough concurrency to execute multiple messages at a time. Which is the opposite of what I need.

Could you please clarify?

Assume your thread pool has just 1 thread (which is the default), and you set the max concurrency to 8. You won't really get to run 8 concurrent handlers, you need to create a thread pool with at least 8 threads for that to happen.

Does that make sense?

Also, if you need strict ordering guarantees, consider using ordering keys in your messages. The service needs to know or, if you happen to have multiple applications attached to the same subscription, it may send the messages to different applications in unspecified order.

[MrMoose]

Hello Carlos,

thanks again for your reply. Strict ordering is not an issue. I am modelling as a FIFO queue here but whoever gets a message doesn't matter, as long as one gets it.

After what I read I will try without my own thread pool, let the library handle it and set max concurrency to 1. When you say the library may receive more if the Q is configured that way I will try to live with that, as long as there is no concurrent message handlers.

Thanks for the clarification!

Moose