[
https://issues.apache.org/jira/browse/GEODE-8864?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17291296#comment-17291296
]
ASF GitHub Bot commented on GEODE-8864:
---------------------------------------
sabbey37 commented on a change in pull request #5954:
URL: https://github.com/apache/geode/pull/5954#discussion_r583268501
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/executor/hash/HScanExecutor.java
##########
@@ -44,6 +45,9 @@
@Override
public RedisResponse executeCommand(Command command,
ExecutionHandlerContext context) {
+
+ final UUID CLIENT_ID = context.getClientUUID();
+
List<byte[]> commandElems = command.getProcessedCommand();
Review comment:
This is my fault... but if we want to match native Redis, the `count`
parameter should actually be a long... and only return the `ERR value is not an
integer or out of range` message if it exceeds the maximum long capacity
(9223372036854775807)... we'll have to change this in all the SCANs (if we
decide to).
##########
File path:
geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/executor/hash/AbstractHScanIntegrationTest.java
##########
@@ -336,48 +441,136 @@ public void
givenMultipleCountsAndMatches_returnsEntriesMatchingLastMatchParamet
}
@Test
- public void givenNegativeCursor_returnsEntriesUsingAbsoluteValueOfCursor() {
- Map<String, String> entryMap = new HashMap<>();
- entryMap.put("1", "yellow");
- entryMap.put("2", "green");
- entryMap.put("3", "orange");
- jedis.hmset("colors", entryMap);
+ public void should_notReturnValue_givenValueWasRemovedBeforeHSCANISCalled() {
- String cursor = "-100";
- ScanResult<Map.Entry<String, String>> result;
- List<Map.Entry<String, String>> allEntries = new ArrayList<>();
+ Map<String, String> data = new HashMap<>();
+ data.put("field_1", "yellow");
+ data.put("field_2", "green");
+ data.put("field_3", "grey");
+ jedis.hmset("colors", data);
- do {
- result = jedis.hscan("colors", cursor);
- allEntries.addAll(result.getResult());
- cursor = result.getCursor();
- } while (!result.isCompleteIteration());
+ jedis.hdel("colors", "field_3");
+ data.remove("field_3");
- assertThat(allEntries).hasSize(3);
- assertThat(new HashSet<>(allEntries)).isEqualTo(entryMap.entrySet());
+ GeodeAwaitility.await().untilAsserted(
+ () -> assertThat(jedis.hget("colors", "field_3")).isNull());
+
+ ScanResult<Map.Entry<String, String>> result = jedis.hscan("colors", "0");
+
+ assertThat(new HashSet<>(result.getResult()))
+ .containsExactlyInAnyOrderElementsOf(data.entrySet());
}
@Test
- public void givenCursorGreaterThanUnsignedLongCapacity_returnsCursorError() {
- assertThatThrownBy(() -> jedis.hscan("a", "18446744073709551616"))
- .hasMessageContaining(ERROR_CURSOR);
+ public void should_retun_not_error_given_non_zero_cursor_on_first_call() {
+
+ Map<String, String> data = new HashMap<>();
+ data.put("field_1", "yellow");
+ data.put("field_2", "green");
+ data.put("field_3", "grey");
+ jedis.hmset("colors", data);
+
+
+ ScanResult<Map.Entry<String, String>> result = jedis.hscan("colors", "5");
+
+ assertThat(new HashSet<>(result.getResult()))
+ .containsExactlyInAnyOrderElementsOf(data.entrySet());
}
+ /**** Concurrency ***/
+
+ private final int SIZE_OF_INITIAL_HASH_DATA = 100;
+ final String HASH_KEY = "key";
+ final String BASE_FIELD = "baseField_";
+
@Test
- public void
givenNegativeCursorGreaterThanUnsignedLongCapacity_returnsCursorError() {
- assertThatThrownBy(() -> jedis.hscan("a", "-18446744073709551616"))
- .hasMessageContaining(ERROR_CURSOR);
+ public void
should_notLoseFields_givenConcurrentThreadsDoingHScansAndChangingValues() {
+ final Map<String, String> INITIAL_HASH_DATA =
makeEntrySet(SIZE_OF_INITIAL_HASH_DATA);
+ jedis.hset(HASH_KEY, INITIAL_HASH_DATA);
+ final int ITERATION_COUNT = 500;
+
+ new ConcurrentLoopingThreads(ITERATION_COUNT,
+ (i) -> multipleHScanAndAssertOnSizeOfResultSet(jedis,
INITIAL_HASH_DATA),
+ (i) -> multipleHScanAndAssertOnSizeOfResultSet(jedis2,
INITIAL_HASH_DATA),
+ (i) -> {
+ int fieldSuffix = i % SIZE_OF_INITIAL_HASH_DATA;
+ jedis3.hset(HASH_KEY, BASE_FIELD + fieldSuffix, "new_value_" + i);
+ }).run();
+
}
@Test
- public void givenInvalidRegexSyntax_returnsEmptyArray() {
- jedis.hset("a", "1", "green");
- ScanParams scanParams = new ScanParams();
- scanParams.count(1);
- scanParams.match("\\p");
+ public void
should_notLoseKeysForConsistentlyPresentFields_givenConcurrentThreadsAddingAndRemovingFields()
{
+ final Map<String, String> INITIAL_HASH_DATA =
makeEntrySet(SIZE_OF_INITIAL_HASH_DATA);
+ jedis.hset(HASH_KEY, INITIAL_HASH_DATA);
+ final int ITERATION_COUNT = 500;
+
+ new ConcurrentLoopingThreads(ITERATION_COUNT,
+ (i) -> multipleHScanAndAssertOnContentOfResultSet(jedis,
INITIAL_HASH_DATA),
+ (i) -> multipleHScanAndAssertOnContentOfResultSet(jedis2,
INITIAL_HASH_DATA),
+ (i) -> {
+ String field = "new_" + BASE_FIELD + i;
+ jedis3.hset(HASH_KEY, field, "whatever");
+ jedis3.hdel(HASH_KEY, field);
+ }).run();
- ScanResult<Map.Entry<String, String>> result = jedis.hscan("a", "0",
scanParams);
+ }
- assertThat(result.getResult()).isEmpty();
+ @Test
+ public void should_notAlterUnderlyingData_givenMultipleConcurrentHscans() {
+ final Map<String, String> INITIAL_HASH_DATA =
makeEntrySet(SIZE_OF_INITIAL_HASH_DATA);
+ jedis.hset(HASH_KEY, INITIAL_HASH_DATA);
+ final int ITERATION_COUNT = 500;
+
+ new ConcurrentLoopingThreads(ITERATION_COUNT,
+ (i) -> multipleHScanAndAssertOnContentOfResultSet(jedis,
INITIAL_HASH_DATA),
+ (i) -> multipleHScanAndAssertOnContentOfResultSet(jedis2,
INITIAL_HASH_DATA));
+
+ INITIAL_HASH_DATA.forEach((field, value) -> {
+ assertThat(jedis3.hget(HASH_KEY, field).equals(value));
+ });
+
+ }
+
+ private void multipleHScanAndAssertOnContentOfResultSet(Jedis jedis,
+ final Map<String, String> intialHashData) {
Review comment:
Since we're fixing other things, could we correct the spelling of
`initial`?
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/data/RedisHash.java
##########
@@ -203,49 +279,122 @@ public int hstrlen(ByteArrayWrapper field) {
return new ArrayList<>(hash.keySet());
}
- public Pair<BigInteger, List<Object>> hscan(Pattern matchPattern, int count,
BigInteger cursor) {
- List<Object> returnList = new ArrayList<Object>();
- int size = hash.size();
- BigInteger beforeCursor = new BigInteger("0");
- int numElements = 0;
- int i = -1;
- for (Map.Entry<ByteArrayWrapper, ByteArrayWrapper> entry :
hash.entrySet()) {
- ByteArrayWrapper key = entry.getKey();
- ByteArrayWrapper value = entry.getValue();
- i++;
- if (beforeCursor.compareTo(cursor) < 0) {
- beforeCursor = beforeCursor.add(new BigInteger("1"));
+ public ImmutablePair<BigInteger, List<Object>> hscan(UUID clientID, Pattern
matchPattern,
+ int count,
+ BigInteger cursorParameter) {
+
+ int startCursor = cursorParameter.intValue();
+
+ List<ByteArrayWrapper> keysToScan = getSnapShotOfKeySet(clientID);
+
+ Pair<Integer, List<Object>> resultsPair =
+ getResultsPair(keysToScan, startCursor, count, matchPattern);
+
+ List<Object> resultList = resultsPair.getRight();
+
+ Integer numberOfIterationsCompleted = resultsPair.getLeft();
+
+ int returnCursorValueAsInt =
+ getCursorValueToReturn(startCursor, numberOfIterationsCompleted,
keysToScan);
+
+ if (returnCursorValueAsInt == 0) {
+ removeHSCANSnapshot(clientID);
+ }
+
+ return new ImmutablePair<>(BigInteger.valueOf(returnCursorValueAsInt),
+ resultList);
+ }
+
+ private void removeHSCANSnapshot(UUID clientID) {
+ this.hScanSnapShots.remove(clientID);
+ this.hScanSnapShotCreationTimes.remove(clientID);
+
+ if (this.hScanSnapShots.isEmpty()) {
+ shutDownHscanSnapshotScheduledRemoval();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private Pair<Integer, List<Object>> getResultsPair(List<ByteArrayWrapper>
keysSnapShot,
+ int startCursor,
+ int count,
+ Pattern matchPattern) {
+
+ int indexOfKeys = startCursor;
+
+ List<ByteArrayWrapper> resultList = new ArrayList<>();
+
+ for (int index = startCursor; index < keysSnapShot.size(); index++) {
+
+ if ((index - startCursor) == count) {
+ break;
+ }
+
+ ByteArrayWrapper key = keysSnapShot.get(index);
+ indexOfKeys++;
+
+ ByteArrayWrapper value = hash.get(key);
+ if (value == null) {
continue;
}
if (matchPattern != null) {
if (matchPattern.matcher(key.toString()).matches()) {
- returnList.add(key);
- returnList.add(value);
- numElements++;
+ resultList.add(key);
+ resultList.add(value);
}
} else {
- returnList.add(key);
- returnList.add(value);
- numElements++;
+ resultList.add(key);
+ resultList.add(value);
}
+ }
- if (numElements == count) {
- break;
- }
+ Integer numberOfIterationsCompleted = indexOfKeys - startCursor;
+
+ Pair resultsPair = new ImmutablePair(numberOfIterationsCompleted,
resultList);
+
+ return resultsPair;
+ }
+
+ private int getCursorValueToReturn(int startCursor,
+ int numberOfIterationsCompleted,
+ List<ByteArrayWrapper> keySnapshot) {
+
+ if (startCursor + numberOfIterationsCompleted >= keySnapshot.size()) {
+ return 0;
}
- Pair<BigInteger, List<Object>> scanResult;
- if (i >= size - 1) {
- scanResult = new ImmutablePair<>(new BigInteger("0"), returnList);
- } else {
- scanResult = new ImmutablePair<>(new BigInteger(String.valueOf(i + 1)),
returnList);
+ return (startCursor + numberOfIterationsCompleted);
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<ByteArrayWrapper> getSnapShotOfKeySet(UUID clientID) {
+ List<ByteArrayWrapper> keySnapShot = this.hScanSnapShots.get(clientID);
+
+ if (keySnapShot == null) {
+ if (this.hScanSnapShots.isEmpty()) {
+ startHscanSnapshotScheduledRemoval();
+ }
+ keySnapShot = createKeySnapShot(clientID);
}
- return scanResult;
+ return keySnapShot;
+ }
+
+
+ private List<ByteArrayWrapper> createKeySnapShot(UUID clientID) {
+ List<ByteArrayWrapper> keySnapShot = new ArrayList<>();
+ keySnapShot.addAll(hash.keySet());
Review comment:
Via IntelliJ's suggestions, you could combine these two lines to be:
`List<ByteArrayWrapper> keySnapShot = new ArrayList<>(hash.keySet());`
##########
File path:
geode-redis/src/test/java/org/apache/geode/redis/internal/data/RedisHashTest.java
##########
@@ -161,4 +163,109 @@ public void
setExpirationTimestamp_stores_delta_that_is_stable() throws IOExcept
o2.fromDelta(in);
assertThat(o2).isEqualTo(o1);
}
+
+ @Test
+ public void hscanSnaphots_shouldBeEmpty_givenHscanHasNotBeenCalled() {
+ RedisHash subject = createRedisHash(100);
+ assertThat(subject.getHscanSnapShots()).isEmpty();
+ }
+
+ @Test
+ public void hscanSnaphots_shouldContainSnapshot_givenHscanHasBeenCalled() {
+
+ final List<ByteArrayWrapper> FIELDS_AND_VALUES_FOR_HASH =
createListOfDataElements(100);
+ RedisHash subject = new RedisHash(FIELDS_AND_VALUES_FOR_HASH);
+ UUID clientID = UUID.randomUUID();
+
+ subject.hscan(clientID, null, 10, BigInteger.valueOf(0));
+
+ ConcurrentHashMap<UUID, List<ByteArrayWrapper>> hscanSnapShotMap =
subject.getHscanSnapShots();
+
+ assertThat(hscanSnapShotMap.containsKey(clientID)).isTrue();
+
+ List<ByteArrayWrapper> keyList = hscanSnapShotMap.get(clientID);
+ assertThat(keyList).isNotEmpty();
+
+ FIELDS_AND_VALUES_FOR_HASH.forEach((entry) -> {
+ if (entry.toString().contains("field")) {
+ assertThat(keyList.contains(entry)).isTrue();
+ } else if (entry.toString().contains("value")) {
+ assertThat(keyList.contains(entry)).isFalse();
Review comment:
you could do the following:
```
if (entry.toString().contains("field")) {
assertThat(keyList).contains(entry);
} else if (entry.toString().contains("value")) {
assertThat(keyList).doesNotContain(entry);
}
```
this will result in clearer failure messages.
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/data/RedisHash.java
##########
@@ -203,49 +279,122 @@ public int hstrlen(ByteArrayWrapper field) {
return new ArrayList<>(hash.keySet());
}
- public Pair<BigInteger, List<Object>> hscan(Pattern matchPattern, int count,
BigInteger cursor) {
- List<Object> returnList = new ArrayList<Object>();
- int size = hash.size();
- BigInteger beforeCursor = new BigInteger("0");
- int numElements = 0;
- int i = -1;
- for (Map.Entry<ByteArrayWrapper, ByteArrayWrapper> entry :
hash.entrySet()) {
- ByteArrayWrapper key = entry.getKey();
- ByteArrayWrapper value = entry.getValue();
- i++;
- if (beforeCursor.compareTo(cursor) < 0) {
- beforeCursor = beforeCursor.add(new BigInteger("1"));
+ public ImmutablePair<BigInteger, List<Object>> hscan(UUID clientID, Pattern
matchPattern,
+ int count,
+ BigInteger cursorParameter) {
+
+ int startCursor = cursorParameter.intValue();
Review comment:
I'm kinda torn about this... I think it makes sense for the cursor value
to be an `int` since the maximum list size in Java is also an `int`... and
`2,147,483,647` entries is quite a lot... but in Redis the cursor value max/min
capacity is the same as an unsigned long (18446744073709551615 or
-18446744073709551615).
So does it make more sense for us to differ from Redis and error if the
cursor exceeds the max int capacity or continue to error if the cursor exceeds
the max long capacity, but convert it to an `int` behind the scenes?
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/data/RedisHash.java
##########
@@ -46,19 +53,88 @@
public class RedisHash extends AbstractRedisData {
public static final RedisHash NULL_REDIS_HASH = new NullRedisHash();
private HashMap<ByteArrayWrapper, ByteArrayWrapper> hash;
+ private ConcurrentHashMap<UUID, List<ByteArrayWrapper>> hScanSnapShots;
+ private ConcurrentHashMap<UUID, Long> hScanSnapShotCreationTimes;
+ private ScheduledExecutorService HSCANSnapshotExpirationExecutor = null;
+
+ private static int default_hscan_snapshots_expire_check_frequency =
+ Integer.getInteger("redis.hscan-snapshot-cleanup-interval", 30000);
+
+ private static int default_hscan_snapshots_milliseconds_to_live =
+ Integer.getInteger("redis.hscan-snapshot-expiry", 30000);
+
+ private int HSCAN_SNAPSHOTS_EXPIRE_CHECK_FREQUENCY_MILLISECONDS;
+ private int MINIMUM_MILLISECONDS_FOR_HSCAN_SNAPSHOTS_TO_LIVE;
+
+ @VisibleForTesting
+ public RedisHash(List<ByteArrayWrapper> fieldsToSet, int
hscanSnapShotExpirationCheckFrequency,
+ int minimumLifeForHscanSnaphot) {
+ this();
+
+ this.HSCAN_SNAPSHOTS_EXPIRE_CHECK_FREQUENCY_MILLISECONDS =
+ hscanSnapShotExpirationCheckFrequency;
+ this.MINIMUM_MILLISECONDS_FOR_HSCAN_SNAPSHOTS_TO_LIVE =
minimumLifeForHscanSnaphot;
- public RedisHash(List<ByteArrayWrapper> fieldsToSet) {
- hash = new HashMap<>();
Iterator<ByteArrayWrapper> iterator = fieldsToSet.iterator();
while (iterator.hasNext()) {
hashPut(iterator.next(), iterator.next());
}
}
+ public RedisHash(List<ByteArrayWrapper> fieldsToSet) {
+ this(fieldsToSet,
+ default_hscan_snapshots_expire_check_frequency,
+ default_hscan_snapshots_milliseconds_to_live);
+ }
+
+ // for serialization
public RedisHash() {
- // for serialization
+ this.hash = new HashMap<>();
+ this.hScanSnapShots = new ConcurrentHashMap<>();
+ this.hScanSnapShotCreationTimes = new ConcurrentHashMap<>();
+
+ this.HSCAN_SNAPSHOTS_EXPIRE_CHECK_FREQUENCY_MILLISECONDS =
+ this.default_hscan_snapshots_expire_check_frequency;
+
+ this.MINIMUM_MILLISECONDS_FOR_HSCAN_SNAPSHOTS_TO_LIVE =
+ this.default_hscan_snapshots_milliseconds_to_live;
}
+
+ private void expireHScanSnapshots() {
+
+ this.hScanSnapShotCreationTimes.entrySet().forEach(entry -> {
Review comment:
You could actually do `this.hScanSnapShotCreationTimes.forEach((entry,
creationTime) -> {...}`, that way you wouldn't have to do the
`entry.getValue();` on the next line. Not necessary, just something cool
IntelliJ suggested.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
> finish implementation of Redis HScan Command
> --------------------------------------------
>
> Key: GEODE-8864
> URL: https://issues.apache.org/jira/browse/GEODE-8864
> Project: Geode
> Issue Type: New Feature
> Components: redis
> Reporter: John Hutchison
> Priority: Major
> Labels: pull-request-available
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
