GreengrassCoreIPCV2 Subscribing creates too many threads, causes systemwide problems.

[MaxProfit]

Hello. We're running a Greengrass custom component that uses Java GreengrassCoreIPCClientV2 and we're subscribing to pubsub, a topic that's getting 1000Hz of messages of a small (100 Bytes or so) size. The way we're processing it is currently needing to be single threaded through a synchronize block due to a 3rd party library not being thread safe. This leads us to have a buildup of what seems to be thousands of threads waiting to process messages by the virtual memory consumption numbers ~37Gb of virtual memory from this component, and each thread has stack size of 1024Kb it seems. Eventually this gets overloaded and causes issues with other processes using fork and pthread_create, and obviously stops delivering the messages.

Is there a way to limit the number of threads being created, and keep the messages not being picked up on local pubsub? We're currently working on workarounds on how to deal with the current bottleneck, which is the root cause of all of this, but it seems like creating thousands of threads and not being able to clamp down on that is an interesting design, is it due to not being able to keep anything on pubsub, where you're forced to take the message or have it be dropped?

[MikeDombo]

Hello Matthew,

You can create the client and provide a specific executor in the builder. You can make an executor which limits the number of threads. https://aws.github.io/aws-iot-device-sdk-java-v2/software/amazon/awssdk/aws/greengrass/GreengrassCoreIPCClientV2.Builder.html#withExecutor(java.util.concurrent.Executor).

However, if you have too much data coming in, you will end up with a ton of memory somewhere. If your application does not process the messages, they will backup in Greengrass Nucleus and cause its memory usage to climb. You ultimately will need to send fewer messages or process them faster.

Perhaps whatever is sending the messages can send fewer, but larger messages.

I can also recommend that you put messages into a limited size queue which a different thread then reads from and processes at its own pace. When the queue fills up, then drop any new messages.

int capacity = 1000;
LinkedBlockingDeque<SomeMessage> queue = new LinkedBlockingDeque<>(capacity);

// callback:
boolean didAdd = queue.offer(message);
if (!didAdd) { /* message couldn't be added, drop it */}


// Read from queue:
while (!Thread.currentThread().interrupted()) {
    SomeMessage message = queue.take();
    // process the message
}

With this solution, you're in charge of your own destiny. You can set whatever capacity you want and handle dropping messages in your own way too.

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.