Possibility to pause polling on Pub/sub subscription

[MrMoose]

Hello everyone,

I was wondering if and how I can pause polling on a subscription without actually cancelling it.

I am using a subscriber with concurrency set to 1 and my own completion queue thread so I only ever execute one handler at a time. Normally this means I intend to receive messages as soon as this thread is idle or done executing a handler.

One message at a time, as quick as possible and as soon as one comes in.

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

pubsub::ConnectionOptions connection_options;
connection_options.DisableBackgroundThreads(my_completion_q);

unique_ptr<pubsub::Subscriber> subscriber = make_unique<pubsub::Subscriber>(pubsub::MakeSubscriberConnection(
		sub,
		subscriber_options,
		connection_options));

....
gc::future<gc::Status> > retval = subscriber->Subscribe(my_handler);

...
// wait until I eventually call 
retval.cancel();

So, I know I can call cancel() on the future to stop the subscriber. But what what if I only need to pause it for half a second or so? I suppose it's quite inefficient to cancel the subscriber and recreate it when I need to continue, so I'm looking for a less intrusive method to pause execution. Since the API doesn't seem to have something to that effect I thought perhaps one could inhibit the completion queue somehow and then continue it?
Can you give me any hints what I can try?

Cheers,
Moose

[coryan]

I was wondering if and how I can pause polling on a subscription without actually cancelling it.

Note that strictly speaking this is not polling. The underlying RPC is a streaming RPC, that means the service will push messages before you ask for them, subject to whatever flow control settings you configured. That is important to understand, because pausing on the client side will not stop the message from being assigned to your client.

With that out of the way. I think the key requirement on what you want is

But what what if I only need to pause it for half a second or so?

I realize this may not be what you want: you could delay delivery of the next message by delaying the ack for any previous message. For example, if you set the flow control options to deliver a single message at a time:

google-cloud-cpp/google/cloud/pubsub/options.h

Lines 225 to 227 in c1ce073

	struct MaxOutstandingMessagesOption {
	using Type = std::int64_t;
	};

Then the service will not send more messages until you ack() or nack() the previous message. You could delay the ack() for 1/2 second and "pause" until you are ready.

That allows you to pause, but I suspect you want to pause before the message is sent? I do not think we can do this with the current RPCs used by the library. By the time your client applications decides to "pause" the service may have already sent a message. I think we would need something more like #7187 where we can pull messages one at a time.

I thought maybe changing the flow control settings dynamically, but that is not allowed:

https://github.com/googleapis/googleapis/blob/9ffb8bb216b22a8ddc539303b36c71db2c568c63/google/pubsub/v1/pubsub.proto#L1159-L1161

Sorry I don't have a good answer and thanks for your excellent questions. Do let me know if #7187 would help, customers asking for features is one of the main inputs for sequencing our work.

[MrMoose]

Hi and thanks for your reply,

about the underlying RPC, I figured this would be a problem. But let's see what you suggest.

Then the service will not send more messages until you ack() or nack() the previous message. You could delay the ack() for 1/2 second and "pause" until you are ready.

That is actually what was used before. The client I am doing this change for used to receive a job and ack it only after it is completed. Problem is that the work packages are getting too large and blow the max 600s visibility deadline. Meaning they cannot reliably ack after less than 10 minutes, causing the message to be received by another node. So sadly, this solves the issue but is not possible for me anymore.

The ticket you mention, #7187 sounds exactly like something that would help. This is similar to what I'm using on AWS/SQS. As you may recall, the library I'm working on has Pubsub and SNS/SQS behind a common interface and in the SQS impl I just loop a blocking poll, which I can easily interrupt to realize this requirement. If I would have something similar here that would be perfect.

So, if there is only the chance for this feature in future, what could I do now, in the meantime? I thought about cancelling the subscription and re-establish it but it says in the docs it would "stop eventually", which doesn't sound like "there will be no more handler calls immediately". How exactly would it behave though? Can you take a guess how bad that really would be to cancel() and restart the subscription? Maybe as a dirty workaround.... ;-)

Cheers,
Moose

[coryan]

Then the service will not send more messages until you ack() or nack() the previous message. You could delay the ack() for 1/2 second and "pause" until you are ready.

That is actually what was used before. The client I am doing this change for used to receive a job and ack it only after it is completed. Problem is that the work packages are getting too large and blow the max 600s visibility deadline. Meaning they cannot reliably ack after less than 10 minutes, causing the message to be received by another node. So sadly, this solves the issue but is not possible for me anymore.

Maybe we have a bug in the library. You should be able to set the timeout for incoming messages to an arbitrarily large value using:

google-cloud-cpp/google/cloud/pubsub/options.h

Lines 193 to 195 in 2c6dd06

	struct MaxDeadlineTimeOption {
	using Type = std::chrono::seconds;
	};

The leases cannot be extended by more than 600s at a time, and you can control the maximum extension using:

google-cloud-cpp/google/cloud/pubsub/options.h

Lines 208 to 210 in 2c6dd06

	struct MaxDeadlineExtensionOption {
	using Type = std::chrono::seconds;
	};

In other words you can tell the library things like "these messages may take 2 hours to process, but extend their deadline in increments of no more than 5 minutes". When the messages are received the library will remember their receipt time, and continuously extend their lease, by 5 minutes each time, until 2 hours have elapsed.

If that is not the behavior you are observing, I need to go fix the library. If that is the behavior you are observing, but does not solve your problem I need to fix my brainmental model of what customers need.

The ticket you mention, #7187 sounds exactly like something that would help. This is similar to what I'm using on AWS/SQS. As you may recall, the library I'm working on has Pubsub and SNS/SQS behind a common interface and in the SQS impl I just loop a blocking poll, which I can easily interrupt to realize this requirement. If I would have something similar here that would be perfect.

It would help by got getting new messages to the application, but setting MaxOutstandingMessagesOption to 1 should have a similar effect. And (you or the library, we have not defined this API) would need to extend the Ack deadline for this message if it takes more than 600s to process it. Otherwise, don't we have the same problem?

So, if there is only the chance for this feature in future, what could I do now, in the meantime? I thought about cancelling the subscription and re-establish it but it says in the docs it would "stop eventually", which doesn't sound like "there will be no more handler calls immediately".

After .cancel() returns no more messages will be delivered to the handler calls. However, there may be many calls started when you do call .cancel().

How exactly would it behave though? Can you take a guess how bad that really would be to cancel() and restart the subscription? Maybe as a dirty workaround.... ;-)

I expect that shutting down a subscription session can take arbitrarily long. It waits until any handlers provided by the application return. If there are no outstanding calls, then I expect several seconds, there are some timers that are easier to just let expire than try to cancel.

[MrMoose]

In other words you can tell the library things like "these messages may take 2 hours to process, but extend their deadline in increments of no more than 5 minutes". When the messages are received the library will remember their receipt time, and continuously extend their lease, by 5 minutes each time, until 2 hours have elapsed.

OK... I guess I misunderstood the visibility timeout then. When I create a subscription (SDK or Web Console) the maximum value is 600s. I was under the impression that is the absolute maximum a message can be "leased" (aka received and be invisible to other receivers) and after 600s the message will be received by others no matter if I ack or nack it. I can't say there's any bug in the library because I haven't tested that case as I thought this is not supported.
From what I understand now I would only need to set MaxDeadlineTimeOption to what I expect is the maximum time a job can take (which should still be way under 30 minutes) and continue with the practice of sending the ack after the job because now the library will automatically extend the visibility timeout by chunks of 10 minutes. Is that correct? Given MaxOutstandingMessagesOption is already set to 1 this would pretty much solve my issue. I will test that right away.
That would give me plenty of leg room while I can wait for #7187 to come along and change to a different poll strategy then.

I already suspected that a cancel and re-instatement of the subscription would have effects like you describe them and I guess this option is off the table.

Edit:
After some testing I think my assumption above was wrong. I have set the max deadline time to 30 Minutes and created jobs with 11 minutes execution time. After they ack their message the same message comes in again, presumably because the lib did not extend the lease time. Or did I misunderstand or use incorrectly?

Edit2:
After even more testing it looks like it may have been my fault. I was running the completion Q with one thread only to make sure there's no concurrent execution of handlers. Apparently this prevented the library from extending the lease. When I run it with two threads the issue seems to go away. I'll continue testing

Edit3:
I can now confirm that the visibility time is extended OK. As long as there's an additional thread available running the completion q. Meaning at least one more than the one that is waiting for the ack on the message. Settings MaxOutstandingMessages to one eliminates concurrent handler calls.

So my problem is essentially solved. Thanks, that was quite helpful and interesting. I will keep an eye out for that story 7187 nonetheless to see if that would better for the use case.

Thanks,
Moose