getting the size of the requests queue

[as...@google.com]

hi, is it possible to query the size of the requests queue?

thanks

[Mark Paluch]

No, it's not possible. In fact, there are two queues: Disconnect queue to buffer commands while disconnected state and the protocol stack (commands sent to Redis awaiting completion).

Care to explain your use-case or what you're trying to achieve?

Cheers, 

Mark

[as...@google.com]

I was referring to the protocol stack.

I'd simply like to put this number on a graph, to see how close my server is getting to the queue limit (which I explicitly set to a number smaller than Integer.MAX_VALUE). 

it's not a must, as I can do this from my server code (increment an AtomitInteger whenever I run a command, and decrement when it ends), it just seems neater to get this from the library itself

[Krishna]

Think you are interested in the QUEUE_SIZE in io.lettuce.core.protocol.DefaultEndpoint class.

Isn't the request queue size related to buffering commands in a bounded queue first before actually writing/flushing to transport via Netty channel which is at the protocol stack? And disconnect buffer only comes into place when channel is disconnected according to io.lettuce.core.protocol.DefaultEndpoint?

[Krishna]

I ended up doing using reflection to get connection metrics as well netty transport level metrics (both should be same) to expose as a JMX metric.

Probably, @Mark could confirm if this is a safe way or if there is an alternative way to get the numbers.

public int getRequestQueueSize() {
    ChannelGroup channels = null;
    try {
        channels = (ChannelGroup) FieldUtils.readField(redisClient, "channels", true);
    } catch (IllegalAccessException e) {
        // e.printStackTrace();
    }

    int requestQueueSize = 0;
    for(Channel channel : channels) {
        CommandHandler commandHandler = null;
        if (channel != null) {
            commandHandler = channel.pipeline().get(CommandHandler.class);
            DefaultEndpoint endpoint = null;
            try {
                endpoint = (DefaultEndpoint) FieldUtils.readField(commandHandler, "endpoint", true);
                requestQueueSize += (Integer) FieldUtils.readField(endpoint, "queueSize", true);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
    }
    return requestQueueSize;
}


public int getPendingTasks() {
    Map<?, EventLoopGroup> eventLoopGroups = null;
    int result = 0;
    try {
        eventLoopGroups = (Map) FieldUtils.readField(redisClient, "eventLoopGroups", true);
    } catch (IllegalAccessException e) {
        return -1;
    }

    int pendingTasks = 0;
    for (EventLoopGroup eventExecutors : eventLoopGroups.values()) {
        NioEventLoopGroup eventLoopGroup = (NioEventLoopGroup) eventExecutors;
        Iterator<EventExecutor> iterator = eventLoopGroup.iterator();
        while (iterator.hasNext()) {
            SingleThreadEventExecutor eventLoop = (SingleThreadEventExecutor) iterator.next();
            pendingTasks += eventLoop.pendingTasks();
        }
    }
    return pendingTasks;
}

[Mark Paluch]

This way, you will obtain the queue sizes of the active connections. Any connection in a disconnected state isn't considered. Alternatively, you can inspect the closeable resources in AbstractRedisClient that contain references to RedisChannelHandler which is the connection facade. The ChannelWriter is typically the Endpoint (or wrapped by CommandExpiryWriter) which also gives you access to the disconnected buffer and command buffer.