Collecting commands from redis cluster nodes

[seeday]

I'm attempting to broadcast commands to all the replicas in a Redis Cluster deployment, but I'm seeing some weird behavior. Code is below. Redis 6.0.9 and Lettuce 6.1.5.

fun doThing(): CompletableFuture<List<Any?>?> {
    return lettuce.connectAsync().thenComposeAsync {
        it.async().replicas().commands().slowlogGet(1)
    }.thenApply {
        it
    }
}

   
fun doThing2(): CompletableFuture<List<Any?>?> {
    return lettuce.connectAsync().thenComposeAsync {
        it.async().replicas().commands().dispatch(
            CommandType.SLOWLOG,
            NestedMultiOutput(lettuce.codec),
            CommandArgs(lettuce.codec).add("GET").add(1)
        )
    }.thenApply {
        it
    }
}

fun doThing3(): CompletableFuture<List<Any?>?> {
    return lettuce.connectAsync().thenComposeAsync { conn ->
        val futures = conn.async().replicas().asMap().values.map {
            it.dispatch(
                CommandType.SLOWLOG,
                NestedMultiOutput(lettuce.codec),
                CommandArgs(lettuce.codec).add("GET").add(1)
            )
        }
        CompletableFuture.allOf(*futures.toArray { arrayOfNulls<CompletableFuture<*>>(it) })
            .thenApply { futures }
    }.thenApply {
        it.map { it.get() }
    }
}

doThing and doThing3 work just fine, however I feel like doThing2 should also work, and I don't quite understand why. The results from doThing2 return jumbled up, weirdly nested, and out of order, like there's multiple threads fighting over the single NestedMultiOutput. I've tried several ways of synchronizing a NestedMultiOutput, but neither locks nor threadlocals work properly.

While 3 does work, (specifically for custom commands), I'm curious as to what option 1 is doing to synchronize the responses properly, why option 2 is not equivalent to 1, and if it's doing it faster than my mess of CompletableFutures in option 3.

Thanks!

[mp911de]

Thanks for bringing this up. The NodeSelection API abstracts the selection of nodes and the triggering of individual commands meaning that each connection receives its individual CommandOutput and CommandArgs object. async().replicas().commands().dispatch(…) fails because all connections try to write to NestedMultiOutput (i.e. they share a single output object) and that is actually a problem because each response requires its own output.

I think we need to do something about it, specifically throw an exception that this method cannot be used and we need to document that behavior. Instead, we could introduce a dispatch method that takes a Supplier<? extends CommandOutput> so that each dispatch can obtain their own output instance.

[seeday]

Thanks for clearing that up. I'm happy to make/test the changes if you'd like.

So far I'm thinking it'll need a new AsyncCommand subclass who's getOutput method is powered by the supplier, and then a modification to BaseNodeSelectionAsyncCommands to remove the old behavior, a change AbstractRedisAsyncCommands to produce the new command type, and maybe some other stuff down the way. Does that sound reasonable?

[mp911de]

Command instances should not be shared across connections so the only reasonable thing to do is again a Supplier. Feel free to have a look into the NodeSelection API. Because there is some dynamic invocation routing involved, there's a bit of Java Proxy code in place in the form of a InvocationHandler.