Allow the Pubsub Publisher to work (in low-volume cases), with fewer than 20 threads

[JoshuaFox]

See discussion with Kir Titievsky , Product Manager for Google PubSub, at GoogleGroup

My application publishes a low volume of messages (1 every few seconds at most), with no other use of PubSub. It does not subscribe. We are using simple Java code as in this sample. The Publisher object is retained for the lifetime of the process, as recommended here.

On first use, PubSub creates 60 threads that stay live permanently, like these:

"grpc-default-worker-ELG-1-1 Id\u003d115 RUNNABLE (in native)...
"grpc-default-worker-ELG-1-10 Id\u003d160 RUNNABLE (in native)":...
....
"Gax-16 Id\u003d141 TIMED_WAITING on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@24f8d334": ...
"Gax-17 Id\u003d142 WAITING on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@24f8d334":....

Sixty is a very high default.

Moreover, if I setExecutorThreadCount to 4 (code following), I still get an extra 26 permanent threads. Setting it to 1 or 2 also gets about 20 extra threads.

ExecutorProvider executorProvider = InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(4).build();
Publisher.Builder builder = Publisher.newBuilder(ProjectTopicName.of(proj, topic)).setExecutorProvider(executorProvider);

Our legacy application is already thread-heavy; and it cannot tolerate more than one or two extra threads, even if these are inactive. Unfortunately, the legacy threading model cannot be changed.

Yet it does seem that a low-volume Publisher really needs only one thread.

Please offer a way to configure PubSub down to 1 or 2 threads.

[ajaaym]

@JoshuaFox You can use below setting to have total control over threads from this library.

   final ExecutorProvider fixedExecutorProvider = FixedExecutorProvider.create(
        new ScheduledThreadPoolExecutor(1)); //adjust the size
    Publisher publisher =
        Publisher.newBuilder("topic_name")
            .setExecutorProvider(fixedExecutorProvider)
            .setChannelProvider(
                PublisherStubSettings.defaultGrpcTransportProviderBuilder()
                    .setExecutorProvider(fixedExecutorProvider)
                    .setChannelConfigurator(
                        new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
                          @Override
                          public ManagedChannelBuilder apply(
                              ManagedChannelBuilder managedChannelBuilder) {
                            NettyChannelBuilder nettyChannelBuilder =
                                (NettyChannelBuilder)
                                    managedChannelBuilder.executor(
                                        fixedExecutorProvider.getExecutor());
                            nettyChannelBuilder.eventLoopGroup(
                                new NioEventLoopGroup(1, fixedExecutorProvider.getExecutor()));
                              nettyChannelBuilder.channelType(
                                  NioSocketChannel.class); // Use EPoll if available, if using EPoll update above line to use EPollEventLoopGroup

                            return nettyChannelBuilder;
                          }
                        })
                    .build())
            .build();
    publisher.publish(PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("testdata")).build()).get();

[JoshuaFox]

Do I understand correctly as follows? -- You use the same single-thread ExecutorProvider shared across Publisher, GrpcTransportProvider, and ManagedChannelBuilder , so that in total only 1 thread is added.

[ajaaym]

thats correct

[moadi]

@JoshuaFox did the code snippet posted by @ajaaym resolve your issue? we are seeing similar problems with PubSub creating too many threads in a single process.

[JoshuaFox]

@moadi . Yes, I tested that it indeed brings down the number of threads.

[wyaeld]

@JoshuaFox are you able to explain the interrelationship between the two numeric 1s, its a little trickier without them being named constants

specifically

final ExecutorProvider fixedExecutorProvider = FixedExecutorProvider.create(
        new ScheduledThreadPoolExecutor(1)); //adjust the size

and

new NioEventLoopGroup(1, fixedExecutorProvider.getExecutor()));

The former obviously appears to be the key value, but if for example I wanted a pool of 10 threads across all my subcribers, do I set both numbers to 10, or just the former?

[JoshuaFox]

@JoshuaFox No. That code is quite low-level, and I do not fully understand the thread design.

@ajaaym who posted it is the one to ask.

This again is a reason that the code setting thread counts should be inboarded to the library and exposed in the library's API.

[gfelbing]

@wyaeld We this issue in our application which maintains ~15 publishers/subscribers suffering from ~1000 waiting GAX threads.
In our case, we only adjusted the value for the FixedExecutorProvider, leaving the NioEventLoopGroup at 1.
As far as I understood the code, it is only distributing the ExecutorProvider which is fine to have only one thread for handling this.

@ajaaym I would still be interested if my interpretation is right

[hannahrogers-google]

I am going to mark this issue resolved, as the solution to this question is provided above. Publisher.Builder.setExecutorProvider can be used to provide a custom executor to the Pub/Sub Publisher. See our docs on publisher concurrency control for examples. A FixedExecutorProvider will return the same executor on each call to .getExecutor(), while an InstantiationExecutorProvider will construct a new executor on each call to .getExecutor(). To decrease gax thread counts, the same single-thread FixedExecutorProvider can be shared across the GrpcTransportProvider and ManagedChannelBuilder.