dschneider-pivotal commented on a change in pull request #7408: URL: https://github.com/apache/geode/pull/7408#discussion_r823022806
########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/commands/executor/list/BLPopExecutor.java ########## @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license + * agreements. See the NOTICE file distributed with this work for additional information regarding + * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. You may obtain a + * copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + */ +package org.apache.geode.redis.internal.commands.executor.list; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.geode.redis.internal.RedisConstants; +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.executor.CommandExecutor; +import org.apache.geode.redis.internal.commands.executor.RedisResponse; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.data.RedisList; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BLPopExecutor implements CommandExecutor { + + @Override + public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) { + List<byte[]> arguments = command.getCommandArguments(); + int keyCount = arguments.size() - 1; + double timeoutMillis; + try { + timeoutMillis = Coder.bytesToDouble(arguments.get(keyCount)) * 1000; + } catch (NumberFormatException e) { + return RedisResponse.error(RedisConstants.ERROR_TIMEOUT_INVALID); + } + + List<RedisKey> keys = new ArrayList<>(keyCount); + for (int i = 0; i < keyCount; i++) { + keys.add(new RedisKey(arguments.get(i))); + } + + List<byte[]> popped = context.lockedExecute(keys.get(0), keys, + () -> RedisList.blpop(context, keys, (int) timeoutMillis)); Review comment: my understanding is that the timeout comes to us in double whose units is seconds. But couldn't then be a fraction of a second (for example 0.005) seconds? If so why do we cast to an "int" here? Also why do we convert it to millis? Shouldn't we do nanos in case they wanted to wait for less than a millisecond? Do we need to be careful here to not allow a non-zero timeout to become zero (which waits forever)? Maybe this should all be done at a lower level (the conversion of time units). ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/data/RedisList.java ########## @@ -154,6 +159,29 @@ public long lpush(List<byte[]> elementsToAdd, Region<RedisKey, RedisData> region return popped; } + public static List<byte[]> blpop(ExecutionHandlerContext context, List<RedisKey> keys, + int timeout) { + Region<RedisKey, RedisData> region = context.getRegion(); + for (RedisKey key : keys) { + if (region.containsKey(key)) { Review comment: instead of interacting with the region twice (containsKey and getTypedRedisData) could you just add a flavor of getTypedRedisData (maybe getExistingTypedRedisData) that just returns null if it does not exist? ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java ########## @@ -232,7 +236,14 @@ public void channelInactive(ChannelHandlerContext ctx) { } } - private void executeCommand(Command command) throws Exception { + public void resubmitCommand(Command command) { + ChannelHandlerContext ctx = + channel.pipeline().context(ByteToCommandDecoder.class.getSimpleName()); + ctx.fireChannelRead(command); Review comment: do you know for sure that fireChannelRead is async? That is, it just hands command off to some queue and does not end up processing command in this thread. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/BlockingCommandListener.java ########## @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license + * agreements. See the NOTICE file distributed with this work for additional information regarding + * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. You may obtain a + * copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + */ + +package org.apache.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BlockingCommandListener implements EventListener { + + private final ExecutionHandlerContext context; + private final RedisCommandType command; + private final List<RedisKey> keys; + private final byte[][] commandOptions; + private final long timeout; + private final long timeAdded; + private Runnable cleanupTask; + + /** + * Constructor to create an instance of a BlockingCommandListener in response to a blocking + * command. When receiving a relevant event, blocking commands simply resubmit the command + * into the Netty pipeline. + * + * @param context the associated ExecutionHandlerContext + * @param command the blocking command + * @param timeout the timeout for the command to block in milliseconds + * @param keys the list of keys the command is interested in + * @param commandOptions any additional options (other than the keys and timeout) required to + * resubmit the command. + */ + public BlockingCommandListener(ExecutionHandlerContext context, RedisCommandType command, + long timeout, List<RedisKey> keys, byte[]... commandOptions) { + this.context = context; + this.command = command; + this.timeout = timeout; + this.keys = Collections.unmodifiableList(keys); + this.commandOptions = commandOptions; + timeAdded = System.currentTimeMillis(); + } + + @Override + public List<RedisKey> keys() { + return keys; + } + + @Override + public EventResponse process(RedisCommandType commandType, RedisKey key) { + if (!keys.contains(key)) { + return EventResponse.CONTINUE; + } + + resubmitCommand(); + return EventResponse.REMOVE_AND_STOP; + } + + @Override + public void resubmitCommand() { + List<byte[]> byteArgs = new ArrayList<>(keys.size() + 1); + byteArgs.add(command.name().getBytes()); + + keys.forEach(x -> byteArgs.add(x.toBytes())); + byteArgs.addAll(Arrays.asList(commandOptions)); + + // Recalculate the timeout since we've already been waiting + long adjustedTimeout = timeout; + if (adjustedTimeout > 0) { + adjustedTimeout = (adjustedTimeout - (System.currentTimeMillis() - timeAdded)) / 1000; + adjustedTimeout = Math.max(1, adjustedTimeout); Review comment: It seems wrong here to say if we are going to wait we need to wait at least a second. Since the timeout is expressed as a double it could be a fraction of a second. It seems like this code should support an adjustedTimeout that is < 1. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventListener.java ########## @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license + * agreements. See the NOTICE file distributed with this work for additional information regarding + * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. You may obtain a + * copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + */ + +package org.apache.geode.redis.internal.eventing; + +import java.util.List; + +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +/** + * Interface intended to be implemented in order to receive Redis events. Specifically this would be + * keyspace events or blocking commands. EventListeners are registered with the + * {@link EventDistributor}. + */ +public interface EventListener { + + /** + * Receive and process an event. This method should execute very quickly. The return value + * determines additional process steps for the given event. + * + * @param commandType the command triggering the event + * @param key the key triggering the event + * @return response determining subsequent processing steps + */ + EventResponse process(RedisCommandType commandType, RedisKey key); + + /** + * Return the list of keys this listener is interested in. + */ + List<RedisKey> keys(); + + /** + * Method to resubmit a command if appropriate. This is only relevant for listeners that process + * events for blocking commands. Listeners that handle keyspace event notification will not use + * this. + */ + default void resubmitCommand() {}; + + /** + * Retrieve the timeout for this listener. The default is no timeout. + */ + default long getTimeout() { + return 0; + } + + /** + * Set a runnable that is responsible for any cleanup when this listener is removed. + */ + default void setCleanupTask(Runnable r) {} Review comment: All of these "defaults" used here seem like a possible indicator that we should have two interfaces. default can be very helpful when trying to maintain backwards compatibility but since this is a brand new interface I'm not sure we should use it ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/EventResponse.java ########## @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license + * agreements. See the NOTICE file distributed with this work for additional information regarding + * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. You may obtain a + * copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + */ + +package org.apache.geode.redis.internal.eventing; + +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; + +/** + * Response returned by {@link EventListener#process(RedisCommandType, RedisKey)} + */ +public enum EventResponse { Review comment: This enum also seems like an indicator that we should have two listener interfaces. It seems odd to have an enum who comments say if you are a keyspace event listener then use CONTINUE; if a blocking event listener then use REMOVE_AND_STOP. ########## File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/eventing/BlockingCommandListener.java ########## @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license + * agreements. See the NOTICE file distributed with this work for additional information regarding + * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. You may obtain a + * copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + */ + +package org.apache.geode.redis.internal.eventing; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.geode.redis.internal.commands.Command; +import org.apache.geode.redis.internal.commands.RedisCommandType; +import org.apache.geode.redis.internal.data.RedisKey; +import org.apache.geode.redis.internal.netty.Coder; +import org.apache.geode.redis.internal.netty.ExecutionHandlerContext; + +public class BlockingCommandListener implements EventListener { + + private final ExecutionHandlerContext context; + private final RedisCommandType command; + private final List<RedisKey> keys; + private final byte[][] commandOptions; + private final long timeout; + private final long timeAdded; + private Runnable cleanupTask; + + /** + * Constructor to create an instance of a BlockingCommandListener in response to a blocking + * command. When receiving a relevant event, blocking commands simply resubmit the command + * into the Netty pipeline. + * + * @param context the associated ExecutionHandlerContext + * @param command the blocking command + * @param timeout the timeout for the command to block in milliseconds + * @param keys the list of keys the command is interested in + * @param commandOptions any additional options (other than the keys and timeout) required to + * resubmit the command. + */ + public BlockingCommandListener(ExecutionHandlerContext context, RedisCommandType command, + long timeout, List<RedisKey> keys, byte[]... commandOptions) { + this.context = context; + this.command = command; + this.timeout = timeout; + this.keys = Collections.unmodifiableList(keys); + this.commandOptions = commandOptions; + timeAdded = System.currentTimeMillis(); + } + + @Override + public List<RedisKey> keys() { + return keys; + } + + @Override + public EventResponse process(RedisCommandType commandType, RedisKey key) { + if (!keys.contains(key)) { + return EventResponse.CONTINUE; + } + + resubmitCommand(); + return EventResponse.REMOVE_AND_STOP; + } + + @Override + public void resubmitCommand() { + List<byte[]> byteArgs = new ArrayList<>(keys.size() + 1); + byteArgs.add(command.name().getBytes()); + + keys.forEach(x -> byteArgs.add(x.toBytes())); + byteArgs.addAll(Arrays.asList(commandOptions)); + + // Recalculate the timeout since we've already been waiting + long adjustedTimeout = timeout; + if (adjustedTimeout > 0) { + adjustedTimeout = (adjustedTimeout - (System.currentTimeMillis() - timeAdded)) / 1000; + adjustedTimeout = Math.max(1, adjustedTimeout); + } + + // The commands we are currently supporting all have the timeout at the end of the argument + // list. Some newer Redis 7 commands (BLMPOP and BZMPOP) have the timeout as the first argument + // after the command. We'll need to adjust this once those commands are supported. + byteArgs.add(Coder.longToBytes(adjustedTimeout)); + + context.resubmitCommand(new Command(command, byteArgs)); Review comment: Couldn't we just replace in place the Command.commandElems item that contains the timeout? I think we can know the index of the timeout and then just stick a different byte[] in the ArrayList at this index. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@geode.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
