[GitHub] [geode] DonalEvans commented on a change in pull request #6862: GEODE-8192: Make Radish MSET command atomic

[GitBox <gi...@apache.org>]

DonalEvans commented on a change in pull request #6862:
URL: https://github.com/apache/geode/pull/6862#discussion_r711245979



##########
File path: geode-apis-compatible-with-redis/src/integrationTest/java/org/apache/geode/redis/internal/executor/common/UnsupportedCommandsIntegrationTest.java
##########
@@ -66,11 +66,11 @@ public void shouldReturnUnknownCommandMessage_givenCallToUnsupportedCommand_when
     final String KEY = "key";
     final String NEW_VALUE = "changed value";
     final String EXPECTED_ERROR_MSG =
-        String.format(ERROR_UNKNOWN_COMMAND, "MSET", "`" + KEY + "`", NEW_VALUE);
+        String.format(ERROR_UNKNOWN_COMMAND, "BITCOUNT", "`" + KEY + "`", NEW_VALUE);

Review comment:
       The first test in this class also needs to be updated to use an unsupported command since MSET is no longer unsupported. Perhaps it would make sense to extract the unsupported command used by both tests to a field so that both can be updated if this happens in future?

##########
File path: geode-apis-compatible-with-redis/src/main/java/org/apache/geode/redis/internal/executor/string/MSetExecutor.java
##########
@@ -32,14 +35,25 @@ public RedisResponse executeCommand(Command command, ExecutionHandlerContext con
     List<byte[]> commandElems = command.getProcessedCommand();
     RedisStringCommands stringCommands = context.getStringCommands();
 
-    // TODO: make this atomic
+    int numElements = (commandElems.size() - 1) / 2;

Review comment:
       Rather than getting all the command elements and then having to skip/ignore the first one, you can use `command.getCommandArguments();` to get only the elements you're actually interested in, which makes iteration and sizing more straightforward.

##########
File path: geode-apis-compatible-with-redis/src/integrationTest/java/org/apache/geode/redis/internal/executor/string/AbstractMSetIntegrationTest.java
##########
@@ -98,76 +97,37 @@ public void testMSet_setsKeysAndReturnsCorrectValues() {
   }
 
   @Test
-  @Ignore("GEODE-8192")
-  public void testMSet_concurrentInstances_mustBeAtomic()
-      throws InterruptedException, ExecutionException {
-    String keyBaseName = "MSETBASE";
-    String val1BaseName = "FIRSTVALBASE";
-    String val2BaseName = "SECONDVALBASE";
-    String[] keysAndVals1 = new String[(ITERATION_COUNT * 2)];
-    String[] keysAndVals2 = new String[(ITERATION_COUNT * 2)];
-    String[] keys = new String[ITERATION_COUNT];
-    String[] vals1 = new String[ITERATION_COUNT];
-    String[] vals2 = new String[ITERATION_COUNT];
-    String[] expectedVals;
-
-    SetUpArraysForConcurrentMSet(keyBaseName,
-        val1BaseName, val2BaseName,
-        keysAndVals1, keysAndVals2,
-        keys,
-        vals1, vals2);
-
-    RunTwoMSetsInParallelThreadsAndVerifyReturnValue(keysAndVals1, keysAndVals2);
-
-    List<String> actualVals = jedis.mget(keys);
-    expectedVals = DetermineWhichMSetWonTheRace(vals1, vals2, actualVals);
-
-    assertThat(actualVals.toArray(new String[] {})).contains(expectedVals);
-  }
+  public void testMSet_concurrentInstances_mustBeAtomic() {
+    int KEY_COUNT = 5000;
+    String[] keys = new String[KEY_COUNT];
 
-  private void SetUpArraysForConcurrentMSet(String keyBaseName, String val1BaseName,
-      String val2BaseName, String[] keysAndVals1,
-      String[] keysAndVals2, String[] keys, String[] vals1,
-      String[] vals2) {
-    for (int i = 0; i < ITERATION_COUNT; i++) {
-      String key = keyBaseName + i + hashTag;
-      String value1 = val1BaseName + i;
-      String value2 = val2BaseName + i;
-      keysAndVals1[2 * i] = key;
-      keysAndVals1[2 * i + 1] = value1;
-      keysAndVals2[2 * i] = key;
-      keysAndVals2[2 * i + 1] = value2;
-      keys[i] = key;
-      vals1[i] = value1;
-      vals2[i] = value2;
+    for (int i = 0; i < keys.length; i++) {
+      keys[i] = HASHTAG + "key" + i;
     }
+    String[] keysAndValues1 = makeKeysAndValues(keys, "valueOne");
+    String[] keysAndValues2 = makeKeysAndValues(keys, "valueTwo");
+
+    new ConcurrentLoopingThreads(1000,
+        i -> jedis.mset(keysAndValues1),
+        i -> jedis.mset(keysAndValues2))
+            .runWithAction(() -> {
+              int count = 0;
+              List<String> values = jedis.mget(keys);
+              for (String v : values) {
+                count += v.startsWith("valueOne") ? 1 : -1;
+              }
+              assertThat(Math.abs(count)).isEqualTo(KEY_COUNT);

Review comment:
       Just personal preference, but I found this a little difficult to understand at first, so a clearer way to assert the same thing could be:
   ```
   assertThat(jedis.mget(keys)).satisfiesAnyOf(
       values -> assertThat(values).allSatisfy(value -> assertThat(value).startsWith("valueOne")),
       values -> assertThat(values).allSatisfy(value -> assertThat(value).startsWith("valueTwo"))
   );
   ```

##########
File path: geode-apis-compatible-with-redis/src/integrationTest/java/org/apache/geode/redis/internal/services/LockingStripedCoordinatorIntegrationTest.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.services;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.geode.redis.ConcurrentLoopingThreads;
+import org.apache.geode.redis.internal.data.RedisKey;
+
+public class LockingStripedCoordinatorIntegrationTest {
+
+  @Test
+  public void concurrentLockRequestsDoNotDeadlock() {
+    int keysToLock = 1000;
+    StripedCoordinator coordinator = new LockingStripedCoordinator();
+    List<RedisKey> keyList = new ArrayList<>(keysToLock);
+    List<RedisKey> reversedKeyList = new ArrayList<>(keysToLock);
+
+    for (int i = 0; i < keysToLock; i++) {
+      RedisKey k = new RedisKey(("" + i).getBytes());
+      keyList.add(k);
+    }
+    for (int i = keysToLock - 1; i >= 0; i--) {
+      reversedKeyList.add(keyList.get(i));
+    }
+
+    new ConcurrentLoopingThreads(10000,
+        i -> {
+          List<RedisKey> keys = new ArrayList<>(keyList);
+          coordinator.execute(keys, () -> "OK");
+        },
+        i -> {
+          List<RedisKey> keys = new ArrayList<>(reversedKeyList);
+          coordinator.execute(keys, () -> "OK");
+        }).run();

Review comment:
       Is it worth changing the runnable here to something that can be used to verify that it's actually executed?

##########
File path: geode-apis-compatible-with-redis/src/main/java/org/apache/geode/redis/internal/executor/string/MSetExecutor.java
##########
@@ -32,14 +35,25 @@ public RedisResponse executeCommand(Command command, ExecutionHandlerContext con
     List<byte[]> commandElems = command.getProcessedCommand();
     RedisStringCommands stringCommands = context.getStringCommands();
 
-    // TODO: make this atomic
+    int numElements = (commandElems.size() - 1) / 2;
+    List<RedisKey> keys = new ArrayList<>(numElements);
+    List<byte[]> values = new ArrayList<>(numElements);
+
+    RedisKey previousKey = null;
     for (int i = 1; i < commandElems.size(); i += 2) {
-      byte[] keyArray = commandElems.get(i);
-      RedisKey key = new RedisKey(keyArray);
-      byte[] valueArray = commandElems.get(i + 1);
-      stringCommands.set(key, valueArray, null);
+      RedisKey key = new RedisKey(commandElems.get(i));
+
+      if (previousKey != null && key.getBucketId() != previousKey.getBucketId()) {
+        return RedisResponse.crossSlot(ERROR_WRONG_SLOT);
+      }
+      keys.add(key);
+      values.add(commandElems.get(i + 1));
+
+      previousKey = key;
     }
 
+    stringCommands.mset(keys, values);
+
     return RedisResponse.string(SUCCESS);

Review comment:
       This can be replaced with `return RedisResponse.ok();` and the constant can be removed.

##########
File path: geode-apis-compatible-with-redis/src/distributedTest/java/org/apache/geode/redis/internal/executor/string/MSetDUnitTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.executor.string;
+
+import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.BIND_ADDRESS;
+import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.REDIS_CLIENT_TIMEOUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.logging.log4j.Logger;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Ignore;
+import org.junit.Test;
+import redis.clients.jedis.HostAndPort;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisCluster;
+
+import org.apache.geode.cache.control.RebalanceFactory;
+import org.apache.geode.cache.control.ResourceManager;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.redis.ConcurrentLoopingThreads;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.dunit.rules.RedisClusterStartupRule;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+
+public class MSetDUnitTest {
+
+  private static final Logger logger = LogService.getLogger();
+
+  @ClassRule
+  public static RedisClusterStartupRule clusterStartUp = new RedisClusterStartupRule(4);
+
+  @ClassRule
+  public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
+  private static final String HASHTAG = "{tag}";
+  private static JedisCluster jedis;
+  private static MemberVM server1;
+  private static int locatorPort;
+
+  private static MemberVM locator;
+
+  @BeforeClass
+  public static void classSetup() {
+    locator = clusterStartUp.startLocatorVM(0);
+    locatorPort = locator.getPort();
+    server1 = clusterStartUp.startRedisVM(1, locatorPort);
+    clusterStartUp.startRedisVM(2, locatorPort);
+    clusterStartUp.startRedisVM(3, locatorPort);
+
+    int redisServerPort1 = clusterStartUp.getRedisPort(1);
+    // Ensure that buckets are created using a connection with a fairly high timeout since
+    // clusterNodes does not get retried.
+    new Jedis(BIND_ADDRESS, redisServerPort1, REDIS_CLIENT_TIMEOUT).clusterNodes();
+
+    jedis = new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), 5000, 20);
+  }
+
+  @After
+  public void after() {
+    clusterStartUp.flushAll();
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    jedis.close();
+  }
+
+  @Test
+  public void testMSet_concurrentInstancesHandleBucketMovement() {
+    int KEY_COUNT = 5000;
+    String[] keys = new String[KEY_COUNT];
+
+    for (int i = 0; i < keys.length; i++) {
+      keys[i] = HASHTAG + "key" + i;
+    }
+    String[] keysAndValues1 = makeKeysAndValues(keys, "valueOne");
+    String[] keysAndValues2 = makeKeysAndValues(keys, "valueTwo");
+
+    new ConcurrentLoopingThreads(100,
+        i -> jedis.mset(keysAndValues1),
+        i -> jedis.mset(keysAndValues2),
+        i -> clusterStartUp.moveBucketForKey(keys[0]))
+            .runWithAction(() -> {
+              int count = 0;
+              List<String> values = jedis.mget(keys);
+              for (String v : values) {
+                if (v == null) {
+                  continue;
+                }
+                count += v.startsWith("valueOne") ? 1 : -1;
+              }
+              assertThat(Math.abs(count)).isEqualTo(KEY_COUNT);

Review comment:
       A similar verification process to the one described in `AbstractMSetIntegrationTest` could be used here for added clarity.




-- 
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-unsubscribe@geode.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org