Vert.x OutOfMemoryError worker thread queue

Question

I'm using Vert.x library to open WebSockets and consume messages off them. The message processing can block so I'm passing these messages to worker threads. I assumed these had some bounded queue with a predefined size but it's just creating a LinkedList as can be seen in TaskQueue.java within vertx core library:

private final LinkedList<Task> tasks = new LinkedList<>();

As a result when my process becomes a slow consumer instead of putting back pressure to the source it's running out of memory - analyzing the heap dump confirmed this. I'm considering not using the worker threads at all, although I'm a bit surprised of this - if a single socket was sending a too much data we may want to put back pressure on that alone, blocking the event loop would penalize other sockets too - or perhaps I'm missing something/not understading how to use the library correctly?

Addition: code snippet Ok I have a class MarshallingStage.java which is registered as textMessageHandler on the webSocket:

MarshallingStage marshallingStage = new MarshallingStage(ctx);
webSocket.textMessageHandler(marshallingStage);

So marshallingStage implements Handler<String>, which unmarshalls json and passes it to the worker thread.

@Override
public void handle(String text) {
    Message message;
    try {
        message = mUnmarshaller.unmarshall(text);
    }
    catch (MarshallingException e) {
        mCtx.disconnect("failed to unmarshall "+ text, e);
        return;
    }
    if (message != null) {
        handleInbound(new MessageEvent(message), mCtx);
    }
    else {
        log.error("{} unmarshalled null message", mCtx.getConnectionId());
    }
}

@Override
public void handleInbound(PipelineEvent event, ProviderContext ctx) {
    Vertx.currentContext().executeBlocking(
            future -> process(event, ctx),
            asyncResult -> {}
    );
}

Given the worker thread has an unbounded list of tasks, if the event-loop thread consumes messages from the tcp buffer faster than the worker thread consumes tasks off that list, the list grows until the process runs out of memory. I really don't see how this could ever work.

DoctorPangloss DoctorPangloss · Accepted Answer · 2018-12-10T05:15:15

You wrote this wrong:

@Override
public void handleInbound(PipelineEvent event, ProviderContext ctx) {
    Vertx.currentContext().executeBlocking(
            future -> process(event, ctx),
            asyncResult -> {}
    );
}

You need to actually call future.handle(Future.succeededFuture()) after the processing is complete.

@Override
public void handleInbound(PipelineEvent event, ProviderContext ctx) {
    Vertx.currentContext().executeBlocking(
            future -> { process(event, ctx); future.handle(Future.succeededFuture()); },
            asyncResult -> {}
    );
}

Otherwise you're stuck waiting to free a worker thread.

Vert.x OutOfMemoryError worker thread queue

1 Answers