From 191244b1a443c841fe215d44765fc499ab31e438 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 27 Jun 2018 11:41:55 +0200 Subject: [PATCH] DATAREDIS-843 - Adapt compare-and-set to changed transaction rollback response. We now consider a transactional rollback that returns an empty EXEC result as rollback for the CAS (compare-and-set) operation. Previously, we checked only that the response of EXEC is not null. A rollback returns an empty list which was previously considered a successful CAS operation. The code for CAS is now extracted to CompareAndSet and is reused from RedisAtomic implementations. Original Pull Request: #349 --- .../redis/support/atomic/CompareAndSet.java | 80 +++++++++++ .../support/atomic/RedisAtomicDouble.java | 25 +--- .../support/atomic/RedisAtomicInteger.java | 24 +--- .../redis/support/atomic/RedisAtomicLong.java | 24 +--- .../atomic/CompareAndSetIntegrationTests.java | 136 ++++++++++++++++++ .../atomic/RedisAtomicDoubleTests.java | 6 - 6 files changed, 219 insertions(+), 76 deletions(-) create mode 100644 src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java create mode 100644 src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java diff --git a/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java b/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java new file mode 100644 index 000000000..0708d09d0 --- /dev/null +++ b/src/main/java/org/springframework/data/redis/support/atomic/CompareAndSet.java @@ -0,0 +1,80 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed 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.springframework.data.redis.support.atomic; + +import lombok.RequiredArgsConstructor; + +import java.util.Collection; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.core.RedisOperations; +import org.springframework.data.redis.core.SessionCallback; +import org.springframework.util.CollectionUtils; + +/** + * Compare-and-set (CAS) operation using Redis Transactions ({@literal WATCH} and {@literal MULTI}) to atomically update + * the value at {@code key}. + *

+ * The CAS block registers a {@literal WATCH} on the key holding the expected value which guarantees that changes after + * watching and comparing the key will rollback the transaction. The {@literal WATCH} is reset if the comparison fails. + * + * @author Mark Paluch + * @since 2.0.8 + * @see RedisAtomicDouble + * @see RedisAtomicInteger + * @see RedisAtomicLong + */ +@RequiredArgsConstructor +class CompareAndSet implements SessionCallback { + + private final Supplier getter; + private final Consumer setter; + private final String key; + private final T expect; + private final T update; + + /* + * (non-Javadoc) + * @see org.springframework.data.redis.core.SessionCallback#execute(org.springframework.data.redis.core.RedisOperations) + */ + @Override + @SuppressWarnings("unchecked") + public Boolean execute(RedisOperations operations) throws DataAccessException { + + RedisOperations ops = (RedisOperations) operations; + + ops.watch(key); + + if (expect.equals(getter.get())) { + + ops.multi(); + setter.accept(update); + + if (updateSuccessful(operations.exec())) { + return true; + } + } + + ops.unwatch(); + return false; + } + + private static boolean updateSuccessful(Collection exec) { + return !CollectionUtils.isEmpty(exec); + } +} diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java index 4d26f9b56..c72db3d54 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicDouble.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,11 +25,9 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundKeyOperations; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.SessionCallback; import org.springframework.data.redis.core.ValueOperations; import org.springframework.data.redis.serializer.GenericToStringSerializer; import org.springframework.data.redis.serializer.RedisSerializer; -import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -190,27 +187,7 @@ public class RedisAtomicDouble extends Number implements Serializable, BoundKeyO * expected value. */ public boolean compareAndSet(final double expect, final double update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @SuppressWarnings({ "unchecked", "rawtypes" }) - public Boolean execute(RedisOperations operations) { - for (;;) { - operations.watch(Collections.singleton(key)); - if (expect == get()) { - generalOps.multi(); - set(update); - if (operations.exec() != null) { - return true; - } - } - { - return false; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java index 8844a68be..c31431c73 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicInteger.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,7 +25,6 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundKeyOperations; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.SessionCallback; import org.springframework.data.redis.core.ValueOperations; import org.springframework.data.redis.serializer.GenericToStringSerializer; import org.springframework.data.redis.serializer.RedisSerializer; @@ -187,27 +185,7 @@ public class RedisAtomicInteger extends Number implements Serializable, BoundKey * expected value. */ public boolean compareAndSet(int expect, int update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @SuppressWarnings("unchecked") - public Boolean execute(RedisOperations operations) { - for (;;) { - operations.watch(Collections.singleton(key)); - if (expect == get()) { - generalOps.multi(); - set(update); - if (operations.exec() != null) { - return true; - } - } - { - return false; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java index e0c9c2d5d..a52069d33 100644 --- a/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java +++ b/src/main/java/org/springframework/data/redis/support/atomic/RedisAtomicLong.java @@ -16,7 +16,6 @@ package org.springframework.data.redis.support.atomic; import java.io.Serializable; -import java.util.Collections; import java.util.Date; import java.util.concurrent.TimeUnit; @@ -26,7 +25,6 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundKeyOperations; import org.springframework.data.redis.core.RedisOperations; import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.SessionCallback; import org.springframework.data.redis.core.ValueOperations; import org.springframework.data.redis.serializer.GenericToStringSerializer; import org.springframework.data.redis.serializer.RedisSerializer; @@ -194,27 +192,7 @@ public class RedisAtomicLong extends Number implements Serializable, BoundKeyOpe * expected value. */ public boolean compareAndSet(long expect, long update) { - - return generalOps.execute(new SessionCallback() { - - @Override - @SuppressWarnings("unchecked") - public Boolean execute(RedisOperations operations) { - for (;;) { - operations.watch(Collections.singleton(key)); - if (expect == get()) { - generalOps.multi(); - set(update); - if (operations.exec() != null) { - return true; - } - } - { - return false; - } - } - } - }); + return generalOps.execute(new CompareAndSet<>(this::get, this::set, key, expect, update)); } /** diff --git a/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java b/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java new file mode 100644 index 000000000..5d9be91dc --- /dev/null +++ b/src/test/java/org/springframework/data/redis/support/atomic/CompareAndSetIntegrationTests.java @@ -0,0 +1,136 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed 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.springframework.data.redis.support.atomic; + +import static org.assertj.core.api.Assertions.*; + +import java.util.Collection; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; +import org.junit.runners.Parameterized.Parameters; +import org.springframework.data.redis.ConnectionFactoryTracker; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ValueOperations; +import org.springframework.data.redis.serializer.GenericToStringSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; + +/** + * Integration tests for {@link CompareAndSet}. + * + * @author Mark Paluch + */ +@RunWith(Parameterized.class) +public class CompareAndSetIntegrationTests { + + private static final String KEY = "key"; + + private final RedisConnectionFactory factory; + private final RedisTemplate template; + private final ValueOperations valueOps; + + public CompareAndSetIntegrationTests(RedisConnectionFactory factory) { + + this.factory = factory; + + this.template = new RedisTemplate<>(); + this.template.setConnectionFactory(factory); + this.template.setKeySerializer(StringRedisSerializer.UTF_8); + this.template.setValueSerializer(new GenericToStringSerializer<>(Long.class)); + this.template.afterPropertiesSet(); + + this.valueOps = this.template.opsForValue(); + + ConnectionFactoryTracker.add(factory); + } + + @After + public void stop() { + RedisConnection connection = factory.getConnection(); + connection.flushDb(); + connection.close(); + } + + @AfterClass + public static void cleanUp() { + ConnectionFactoryTracker.cleanUp(); + } + + @Parameters + public static Collection testParams() { + return AtomicCountersParam.testParams(); + } + + @Test // DATAREDIS-843 + public void shouldUpdateCounter() { + + long expected = 5; + long actual = 5; + long update = 6; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> valueOps.set(KEY, newValue), KEY, expected, + update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isTrue(); + assertThat(valueOps.get(KEY)).isEqualTo(update); + } + + @Test // DATAREDIS-843 + public void expectationNotMet() { + + long expected = 5; + long actual = 7; + long update = 6; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> valueOps.set(KEY, newValue), KEY, expected, + update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isFalse(); + assertThat(valueOps.get(KEY)).isNull(); + } + + @Test // DATAREDIS-843 + public void concurrentUpdate() { + + long expected = 5; + long actual = 5; + long update = 6; + long concurrentlyUpdated = 7; + + CompareAndSet cas = new CompareAndSet<>(() -> actual, newValue -> { + + RedisConnection connection = this.factory.getConnection(); + connection.set(KEY.getBytes(), Long.toString(concurrentlyUpdated).getBytes()); + connection.close(); + + valueOps.set(KEY, newValue); + }, KEY, expected, update); + + Boolean executed = this.template.execute(cas); + + assertThat(executed).isFalse(); + assertThat(valueOps.get(KEY)).isEqualTo(concurrentlyUpdated); + } +} diff --git a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java index b2cf852a8..d5c998dc5 100644 --- a/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java +++ b/src/test/java/org/springframework/data/redis/support/atomic/RedisAtomicDoubleTests.java @@ -69,12 +69,6 @@ public class RedisAtomicDoubleTests extends AbstractRedisAtomicsTests { ConnectionFactoryTracker.add(factory); } - @Before - public void setUp() { - // Most atomic Double ops involve incrByFloat, which is new as of 2.6 - assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "2.6")); - } - @After public void stop() { RedisConnection connection = factory.getConnection();