Google-cloud-php: Streaming pull for PubSub

[gedimin45]

There seem to be no support for streaming pull for PubSub in the library. Any plans to add it? It really seems like the only way to use PubSub under moderate to high throughput.

[dwsupplee]

This is a good question. As you may already be aware, we do support it in our generated client but not in the veneer.

/cc @michaelbausor @tmatsuo Do you guys have any thoughts as to whether this is something we would like to support? It seems it could be valuable.

[tmatsuo]

It might be useful, although I think it is trickier to write high-performance task workers with PHP than doing the same thing with other languages.

I'm personally interested in implementing such a worker with ReactPHP. It also can be done with forking external processes. Then if the prototype with ReactPHP is performant enough, I would say it's worthwhile to invest on the friendlier hand-written implementation.

[gedimin45]

I used the code similar to this:

use Google\Cloud\PubSub\V1\StreamingPullRequest;
use Google\Cloud\PubSub\V1\SubscriberClient;

$subscriberClient = new SubscriberClient();
try {
    $formattedSubscription = $subscriberClient->subscriptionName('dashboard-tst', 'speed-sub');
    $streamAckDeadlineSeconds = 0;
    $request = new StreamingPullRequest();
    $request->setSubscription($formattedSubscription);
    $request->setStreamAckDeadlineSeconds($streamAckDeadlineSeconds);
    $requests = [$request];

    // Write all requests to the server, then read all responses until the
    // stream is complete
    $stream = $subscriberClient->streamingPull();
    $stream->writeAll($requests);
    foreach ($stream->closeWriteAndReadAll() as $element) {
        // doSomethingWith($element);
    }
} finally {
    $subscriberClient->close();
}

but the code would just block forever at $stream->closeWriteAndReadAll() though there were messages in the subscription available.

Are you sure the underlying BidiStream works as expected? Could it be a gRPC bug?

I think it is trickier to write high-performance task workers with PHP than doing the same thing with other languages

I tend to agree but I can write task workers that are orders of magnitude more performant with Beanstalkd than with PubSub. And the bottleneck is synchronous pull.
What is more, running more instances of the same worker using PubSub does not help much with throughput.

I'm personally interested in implementing such a worker with ReactPHP

I would like to see that. Or maybe using Amphp. But given the code above, is it really necessary? Does that code even make sense?

[dwsupplee]

Are you sure the underlying BidiStream works as expected? Could it be a gRPC bug?

Thanks for raising this issue. Which version of the gRPC extension are you using, and are you also using the protobuf extension?

[gedimin45]

Hey @dwsupplee, I was mistaken, the code I pasted in https://github.com/GoogleCloudPlatform/google-cloud-php/issues/939#issuecomment-375465647 seems to work.

Any chance to get streaming pull into the library any time soon? It appears that PubSub is not usable for anything bigger than moderate workloads without it.

[gedimin45]

Any updates on this? I am amazed how anybody uses PubSub in production with the PHP library.

[tmatsuo]

@ged15

Technically, we can surface the streaming subscriber to the manual layer of the library. We decided however, that we won't proceed that path. Here is the reason.

Regardless of whether you use the streaming pull or not, it is significantly difficult to build a high performant, production level subscriber (mostly worker type) with PHP. We recommend that you use push subscription with a PHP endpoint, or use other language for pull subscriber.

Anyways, thanks for your feedback, and let us know if you think differently. We'll hear you.

[gedimin45]

@tmatsuo,

Thanks for the reply. Correct me if I'm wrong, but push subscriptions are very limited in throughput (I've heard max 1 message per second). Also it puts load on the HTTP load balancer.

This simply makes PubSub not an option for any application written in PHP, and using it to communicate.

As I've already mentioned, we are able to write workers using Beanstalkd that are able to process hundreds of jobs per second without any optimization - yes, that is significantly lower than what we could get from Go or many other languages, but also far from what PubSub can deliver.

The reason I am convinced blocking pull is limiting our worker throughput is that, no matter how many instances of the same worker we spawned, throughput of a single worker doing the same thing written in Go was significantly bigger and we tracked it down that blocking pull just does not return as many messages as it could during single pull.

Therefore I must say that your statement that writing performant workers in PHP is difficult is correct but it seems in my case, PHP is not the problem - rather PubSub. Could you perhaps clarify why you are convinced of that?

I will put it simply - I am fairly convinced that PubSub is the bottleneck for us. Perhaps you can provide benchmarks using the PHP library that proves this wrong?

[tmatsuo]

@ged15

I've heard max 1 message per second

I don't think it's true (but maybe I'm wrong). Where did you see this?

we tracked it down that blocking pull just does not return as many messages as it could during single pull.

Can you post simplified code for the subscriber?

Perhaps you can provide benchmarks using the PHP library that proves this wrong?

Yeah it is a good practice to do such an experiment and get some numbers.

Maybe we can build some kind of job workers, using reactPHP, both with blocking pull and with streaming pull and compare the result.

@dwsupplee I thought you built something like that. Is it possible to share your code so that I can continue experimenting, or do you want to do the experiment by yourself (which is great too)?

[tmatsuo]

@dwsupplee Never mind. I'll build the experiment by myself.

[dwsupplee]

I have been playing around with using ReactPHP for bidirectional streaming in Speech (gist) and have been having some success. I haven't adapted this for PubSub just yet. Looking forward to trying it out with this as well, however. Hopefully we can collaborate.

[taka-oyama]

We've been having some issues with pubsub as well.
I was planning on replacing rabbitmq with this for message-queueing but concluded that it was too slow.
Hopefully this gets fixed soon.

[ava12]

I've tried synchronous pulls (both gRPC and REST transports) and streaming pulls, using Google server and local emulator. Configuration is:

OS: Ubuntu 18.04.4 LTS
PHP: 7.2.24
PubSub: 1.26.0
gRPC: grpc/grpc 1.27.0
Protobuf: google/protobuf v3.12.4 (no C extension)

I found that all three variants can receive and acknowledge more than 500 1-KB messages per second, which looks good (for PHP). Maybe the problem is not PubSub library, but something else?

@ged15, @taka-oyama, how slow is PubSub in your case? Could you please share your configuration and minimal code causing the problem?

[taka-oyama]

Hi, Thanks for looking into the issue.

I tried the pulls again but it's still really slow.

First run takes about 3 seconds and all preceding runs take about a second to execute.

I tested it in follwing env

OS: CentOS 7.8
PHP: 7.4.7
Pubsub: v1.26.0
gRPC: grpc v1.30.0 (C extension)
Protobuf: (latest C extension)

My code looks like below

<?php

require 'vendor/autoload.php';

use Google\Cloud\PubSub\PubSubClient;
use Symfony\Component\Cache\Adapter\ApcuAdapter;

$config = [
    'authCache' => new ApcuAdapter('queue-pubsub'),
    'keyFilePath' => __DIR__.'/gcp.json',
];

$pubSub = new PubSubClient($config);

// Get an instance of a previously created subscription.
$subscription = $pubSub->subscription('pubsub-test');

$then = microtime(true);

$messages = $subscription->pull([
    'returnImmediately' => true,
    'maxMessages' => 1,
]);

echo (microtime(true) - $then) * 1000;
echo PHP_EOL;

[taka-oyama]

I noticed streaming pull was implemented, I will try the streaming pull once it's released.
When will this be released?

[gedimin45]

The way I was using the library initially was wrong.

I came from using Beanstalkd. The approach when using it, is to pull and acknowledge 1 message at a time. It works well because Beanstalkd provides a low response time when pulling (~1 ms in my experience).

Pulling from GCP PubSub, on the other hand, takes quite some time - ~1 s if I recall correctly. My initial approach was to fire up lots of processes that would pull a single message per iteration, process it and acknowledge. This did not scale well.

The approach that did work in the end, was to pull a number of messages - 500 for example - process them, and batch acknowledge. In case a process needed to shut down before processing of the messages was done, the ack deadline for the unprocessed messages would be set to 0. This would trigger an immediate redelivery of those messages to other instances of a processor.

Some infrastructure might need to be put in place using the above approach. If processing a single message takes a significant amount of time, sequential processing will not work. You could employ a library like amphp/parallel to process messages in parallel. In my case, processing of a single message would take a few miliseconds, so sequential processing worked well.

All in all, the important points when using GCP PubSub are:

• messages should be pulled in batches
• setting ack deadline to 0 can be used to to redeliver unprocessed messages during a restart of the processor

With the above approach, a few PHP workers can achieve throughput of thousands PubSub messages per second. Streaming pull is not needed for that.

[ava12]

@taka-oyama, streaming pull was not implemented, it was just a draft PR to play with. Googlers say they don't want to implement this feature (too much work, no advantages). So it won't be released.

As for the code sample:

• there's no point in pulling only 1 message per request (too much overhead);
• returnImmediately flag is deprecated for both gRPC and REST protocols, and using it is discouraged (again, too much overhead); the server will terminate the request (in about 1 minute) if there're no new messages, so it won't block forever.

[taka-oyama]

Hi, thanks for the replies.

I mis-read the thread.
Thanks for letting me know.

The reason I process it this way is because I was trying to implement a laravel queue driver for pubsub.
The queue worker for laravel is implemented in a way where it only processes 1 job at a time.
In theory, I could just pull in bulk and loop through the messages but I'd have to write alot of code that goes against the grain and that'd a bit too much for me.

[gedimin45]

I built a PoC of pulling and parallel processing using amphp/parallel: https://github.com/gedimin45/amphp-gcp-pubsub.

amphp/parallel is a library that enables parallel processing by using ext-parallel, ext-pthreads or spawning processes.

I was able to get a processing throughput of 800 msg/s on the PubSub emulator and 6 threads on my 2014 MacBook Pro. I do suspect I was maxing the emulator out though. I am quite sure the throughput with actual PubSub would be pretty much unlimited given enough CPUs - in fact I see no reason why it would.

Hope this example helps anyone who ever runs into questions on how to process incoming PubSub jobs.