From 39dca76e7c33ccceb65a64f37f3b1eb53838d2c9 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Fri, 3 Jun 2016 13:41:36 +0200 Subject: [PATCH] DATACASS-253 - Add synchronization to CachedPreparedStatementCreator. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CachedPreparedStatementCreator is now thread-safe and synchronizes concurrent calls to statement preparation. The synchronization prevents multiple calls to Session.prepare(…) for the same session, CQL statement and keyspace. --- pom.xml | 8 + spring-cql/pom.xml | 5 + .../core/CachedPreparedStatementCreator.java | 85 ++++++-- ...chedPreparedStatementCreatorUnitTests.java | 187 ++++++++++++++++++ 4 files changed, 264 insertions(+), 21 deletions(-) create mode 100644 spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java diff --git a/pom.xml b/pom.xml index 02877e232..50a87ddc8 100644 --- a/pom.xml +++ b/pom.xml @@ -76,6 +76,7 @@ spring-data-cassandra 1.0 2.16 + 1.01 multi 1.13.0.BUILD-SNAPSHOT @@ -189,6 +190,13 @@ 3.1 test + + + edu.umd.cs.mtc + multithreadedtc + ${multithreadedtc.version} + test + diff --git a/spring-cql/pom.xml b/spring-cql/pom.xml index d37c3e33d..1f09198d8 100644 --- a/spring-cql/pom.xml +++ b/spring-cql/pom.xml @@ -87,6 +87,11 @@ commons-lang3 test + + edu.umd.cs.mtc + multithreadedtc + test + diff --git a/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java b/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java index 5c9019a70..324fbbfbb 100644 --- a/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java +++ b/spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2014 the original author or authors. + * Copyright 2013-2016 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. @@ -27,27 +27,32 @@ import com.datastax.driver.core.Session; import com.datastax.driver.core.exceptions.DriverException; /** - * This Prepared Statement Creator maintains a cache of all prepared statements for the duration of this life of the - * container. When preparing statements with Cassandra, each Statement should be prepared once and only once due to the - * overhead of preparing the statement. + * This {@link PreparedStatementCreator} maintains a static cache of all prepared statements for the duration of the JVM + * runtime, more specific the lifecycle of the associated {@link ClassLoader}. When preparing statements with Cassandra, + * each Statement should be prepared once and only once due to the overhead of preparing the statement. + *

+ * {@link CachedPreparedStatementCreator} is thread-safe and does not require external synchronization when used by + * concurrent threads. * * @author David Webb + * @author Mark Paluch */ public class CachedPreparedStatementCreator implements PreparedStatementCreator { private static final Logger log = LoggerFactory.getLogger(CachedPreparedStatementCreator.class); + private static final Map> CACHE = new ConcurrentHashMap>(); private final String cql; - private static final Map> psMap = new ConcurrentHashMap>(); - /** - * Create a PreparedStatementCreator from the provided CQL. + * Create a {@link PreparedStatementCreator} from the provided CQL. * - * @param cql + * @param cql must not be empty and not {@literal null}. */ public CachedPreparedStatementCreator(String cql) { - Assert.notNull(cql, "CQL is required to create a PreparedStatement"); + + Assert.hasText(cql, "CQL is required to create a PreparedStatement"); + this.cql = cql; } @@ -55,29 +60,67 @@ public class CachedPreparedStatementCreator implements PreparedStatementCreator return this.cql; } + /* (non-Javadoc) + * @see org.springframework.cassandra.core.PreparedStatementCreator#createPreparedStatement(com.datastax.driver.core.Session) + */ @Override public PreparedStatement createPreparedStatement(Session session) throws DriverException { - StringBuilder keyspaceCQLKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql); + StringBuilder cacheKey = new StringBuilder().append(session.getLoggedKeyspace()).append("|").append(this.cql); - log.debug(String.format("Cachable PreparedStatement in Keyspace [%s]", session.getLoggedKeyspace())); + log.debug("Cachable PreparedStatement in Keyspace {}", session.getLoggedKeyspace()); + + Map sessionCache = getOrCreateSessionLocalCache(session); + return getOrPrepareStatement(session, cacheKey.toString(), sessionCache); + } + + private Map getOrCreateSessionLocalCache(Session session) { + + Map sessionMap = CACHE.get(session); - Map sessionMap = psMap.get(session); if (sessionMap == null) { - sessionMap = new ConcurrentHashMap(); - psMap.put(session, sessionMap); + + synchronized (session) { + + if (CACHE.containsKey(session)) { + sessionMap = CACHE.get(session); + } else { + + sessionMap = new ConcurrentHashMap(); + CACHE.put(session, sessionMap); + } + } } - PreparedStatement pstmt = sessionMap.get(keyspaceCQLKey.toString()); + return sessionMap; + } + + private PreparedStatement getOrPrepareStatement(Session session, String cacheKey, + Map sessionCache) { + + PreparedStatement pstmt = sessionCache.get(cacheKey); + if (pstmt == null) { - log.debug("No Cached PreparedStatement found...Creating and Caching"); - pstmt = session.prepare(this.cql); - sessionMap.put(keyspaceCQLKey.toString(), pstmt); - } else { - log.debug("Found cached PreparedStatement"); + + synchronized (sessionCache) { + + if (sessionCache.containsKey(cacheKey)) { + + log.debug("Found cached PreparedStatement"); + return sessionCache.get(cacheKey); + } + + log.debug("No Cached PreparedStatement found...Creating and Caching"); + + pstmt = session.prepare(this.cql); + sessionCache.put(cacheKey, pstmt); + + return pstmt; + } } + log.debug("Found cached PreparedStatement"); + return pstmt; } - } diff --git a/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java b/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java new file mode 100644 index 000000000..64fc9a4c6 --- /dev/null +++ b/spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java @@ -0,0 +1,187 @@ +/* + * Copyright 2016 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.cassandra.core; + +import static edu.umd.cs.mtc.TestFramework.*; +import static org.hamcrest.Matchers.*; +import static org.junit.Assert.*; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.*; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; + +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.Session; + +import edu.umd.cs.mtc.MultithreadedTestCase; + +/** + * Unit tests for {@link CachedPreparedStatementCreator}. + * + * @author Mark Paluch + */ +@RunWith(MockitoJUnitRunner.class) +public class CachedPreparedStatementCreatorUnitTests { + + PreparedStatement preparedStatement; + @Mock Session sessionMock; + + @Before + public void before() throws Exception { + + preparedStatement = newProxy(PreparedStatement.class, new TestInvocationHandler()); + when(sessionMock.prepare(anyString())).thenReturn(preparedStatement); + } + + /** + * @see DATACASS-253 + */ + @Test(expected = IllegalArgumentException.class) + public void shouldRejectEmptyCql() { + new CachedPreparedStatementCreator(""); + } + + /** + * @see DATACASS-253 + */ + @Test(expected = IllegalArgumentException.class) + public void shouldRejectNullCql() { + new CachedPreparedStatementCreator(null); + } + + /** + * @see DATACASS-253 + */ + @Test + public void shouldCreatePreparedStatement() { + + CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql"); + + PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock); + + assertThat(result, is(sameInstance(preparedStatement))); + verify(sessionMock).prepare("my cql"); + } + + /** + * @see DATACASS-253 + */ + @Test + public void shouldCacheCreatePreparedStatement() { + + CachedPreparedStatementCreator cachedPreparedStatementCreator = new CachedPreparedStatementCreator("my cql"); + + cachedPreparedStatementCreator.createPreparedStatement(sessionMock); + cachedPreparedStatementCreator.createPreparedStatement(sessionMock); + PreparedStatement result = cachedPreparedStatementCreator.createPreparedStatement(sessionMock); + + assertThat(result, is(sameInstance(preparedStatement))); + verify(sessionMock, times(1)).prepare("my cql"); + } + + /** + * @see DATACASS-253 + * @throws Throwable + */ + @Test + public void concurrentAccessToCreateStatementShouldBeSynchronized() throws Throwable { + + CreatePreparedStatementIsThreadSafe concurrentPrepareStatement = new CreatePreparedStatementIsThreadSafe( + preparedStatement, new CachedPreparedStatementCreator("my cql")); + + runManyTimes(concurrentPrepareStatement, 5); + } + + @SuppressWarnings("unused") + private static class CreatePreparedStatementIsThreadSafe extends MultithreadedTestCase { + + final CachedPreparedStatementCreator preparedStatementCreator; + final Session session; + final AtomicInteger atomicInteger = new AtomicInteger(); + + public CreatePreparedStatementIsThreadSafe(final PreparedStatement preparedStatement, + CachedPreparedStatementCreator preparedStatementCreator) { + + this.preparedStatementCreator = preparedStatementCreator; + + this.session = newProxy(Session.class, new TestInvocationHandler() { + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + + if (method.getName().equals("prepare") && args.length == 1) { + + waitForTick(2); + atomicInteger.incrementAndGet(); + return preparedStatement; + } + + return super.invoke(proxy, method, args); + } + }); + } + + public void thread1() { + + waitForTick(1); + + preparedStatementCreator.createPreparedStatement(session); + + assertThat(atomicInteger.get(), is(1)); + + } + + public void thread2() { + + waitForTick(1); + + preparedStatementCreator.createPreparedStatement(session); + + assertThat(atomicInteger.get(), is(1)); + } + } + + private static class TestInvocationHandler implements InvocationHandler { + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + + if (method.getName().equals("hashCode")) { + return hashCode(); + } + + if (method.getName().equals("equals") && args.length == 1) { + return equals(args[0]); + } + + return null; + } + } + + private static T newProxy(Class theClass, InvocationHandler invocationHandler) { + return (T) Proxy.newProxyInstance(CachedPreparedStatementCreatorUnitTests.class.getClassLoader(), + new Class[] { theClass }, invocationHandler); + } +}