StackOverflowError when chaining multiple Unis

[PyAntony]

Hi,

there seems to be a limit to the amount of Units I can chain. See this code:

        AtomicLong counter = new AtomicLong();
        Uni<Void> uni = Uni.createFrom().voidItem();
        for (int i = 0; i < 1000; i++) {
            uni = uni.chain(() -> Uni.createFrom().item(42).invoke(counter::addAndGet).replaceWithVoid());
        }

        uni.await().indefinitely();
        System.out.println(counter.get());

Chaining about 500 Unis work, but above 600 it throws the StackOverflowError. I found this error when I was using the ReactiveRedisDataSource as client.withConnection; there I am supposed to chain multiple commands (pipelining). I need to send thousands or commands together.

Quarkus version: 2.13

[jponge]

/cc @cescoffier

[jponge]

There's indeed a point where calls can stack up, pretty much like in plain imperative code (but worse because of the design of reactive streams).

Chaining 1000 chain operators that create intermediate Uni objects is big, and if you don't have non-blocking I/O like in the purely CPU-bound snippet above then it materializes as a huge stack of direct method calls which will be bigger than the equivalent imperative code.

I don't know the Redis client well, but isn't there a way to batch operations at the client level rather than doing so at the Mutiny level, and sending many requests rather than properly batched requests?

Ultimately you might look at JVM tuning and -Xss to try to mitigate stack overflow errors, but that will always be fragile.

[cescoffier]

We don't have "good" pipelining support at the moment (for Redis).
What are you trying to achieve? Bulk insert?

[PyAntony]

yes, bulk insert (https://redis.io/docs/manual/pipelining/), but for thousands of commands. @cescoffier I think you answered me when I asked the same here: https://stackoverflow.com/questions/73694703/how-to-use-redis-pipelining. The solution is to use withConnection (the only solution available). But when I run it with thousands of commands (thousands of chain calls) it throws the StackOverflowError exception. Apparently you can do something like this with the VertX client (vert-x3/vertx-redis-client#47):

for(int i = 0; i < 1000; i++) {
  redis.set("mykey", i, res -> {});
}

This does NOT work with the Quarkus client, it uses all connections available in the pool. I think It should be mentioned in the documentation that there is currently not good support for pipelining since it is an essential feature; I really assumed this feature was available.

By the way, there is this config available:

quarkus.redis.max-waiting-handlers:
Sets how many handlers is the client willing to queue. The client will always work on pipeline mode, this means that messages can start queueing. Using this configuration option, you can control how much backlog you’re willing to accept.

I think this is misleading; client only does pipelining when using withConnection method; the client does not work in pipeline mode by default.

[cescoffier]

I meant that we don't have a good pipeline support in the redis data source. The underlying client provides a batch method to achieve what you want: https://vertx.io/docs/apidocs/io/vertx/rxjava/redis/client/Redis.html#batch-java.util.List-

@Inject Redis redis;
// ...
List<Request> requests = ...
redis.batch(requests);
//...

The main issue with the pipelining is collecting the deserializers. We do that for transactions, but for batching thousands of requests, the cost would be too high.

[PyAntony]

Just tested that client (io.vertx.mutiny.redis.client.Redis) and it works well. Documentation only shows these clients though:

@ApplicationScoped
public class RedisExample {
    @Inject ReactiveRedisDataSource reactiveDataSource;
    @Inject RedisDataSource redisDataSource;
    @Inject RedisAPI redisAPI;
    // ...
}

I didn't know I could use this even-lower-level client 😕

UPDATE

So I am using the Redis client with batch (io.vertx.mutiny.redis.client.Redis). It works well, it returns a list of responses in the same order as the commands. The problem comes when an error is introduced, see the following example:

    @Scheduled(every = "5s")
    Uni<Void> run() {
        var requests = Lists.mutable.of(
            Request.cmd(Command.HSET).arg("dc1").arg(UUID.randomUUID().toString()).arg("value1"),
            Request.cmd(Command.HSET).arg("dc1").arg(UUID.randomUUID().toString()).arg("value2"),
            Request.cmd(Command.HSET).arg("dc1").arg(UUID.randomUUID().toString()).arg("value3"),
            Request.cmd(Command.HSET).arg("dc1").arg(UUID.randomUUID().toString()).arg("value4"),
            // incorrect command: running a list command on a hashset
            Request.cmd(Command.LLEN).arg("dc1")
        );

        return redis.batch(requests)
            .onItemOrFailure().transform((responses, e) -> {
                // prints: ERR [4] WRONGTYPE Operation against a key holding the wrong kind of value
                System.out.println(e);
                return new ArrayList();
            })
            .replaceWithVoid();
    }

Once an error is returned from Redis, the error is thrown in the UNI and the other responses are lost. So there is no way to access the other responses? Redis responds with all responses, even if there was an error for any of the commands. For example, here I am passing an incorrect command at the end, but good commands for which I get responses:

antony@pop-os:~$ (printf "PING\r\nHSET dc1 field1 value\r\nLLEN existingHashSet\r\n"; sleep 1) | nc localhost 6379
+PONG
:1
-WRONGTYPE Operation against a key holding the wrong kind of value

[cescoffier]

This API is documented - See https://quarkus.io/guides/redis-reference#apis.

About the failure, this is how the batch support is implemented. Please raise an issue on https://github.com/vert-x3/vertx-redis-client.

[cescoffier]

BTW, not really pipelining, but this works:

rds.withConnection(redis -> {
            for (int i = 0; i < 5000; i++) {
                unis.add(redis.value(Integer.class).set("key-" + i, i));
            }
            return Uni.join().all(unis).andCollectFailures()
                    .replaceWithVoid();
        }).await().indefinitely();

[PyAntony]

This really solves it, thanks!