Spring/spring-data-cassandra
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATACASS-253 - Add synchronization to CachedPreparedStatementCreator.

Browse Source 
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.
This commit is contained in:
Mark Paluch
2016-06-03 13:41:36 +02:00
committed by John Blum
parent abcac5f5ee
commit 39dca76e7c
4 changed files with 264 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pom.xml
View File

@@ -76,6 +76,7 @@
<dist.id>spring-data-cassandra</dist.id>
<el.version>1.0</el.version>
<failsafe.version>2.16</failsafe.version>
<multithreadedtc.version>1.01</multithreadedtc.version>
<project.type>multi</project.type>
<springdata.commons>1.13.0.BUILD-SNAPSHOT</springdata.commons>
</properties>
@@ -189,6 +190,13 @@
<version>3.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>edu.umd.cs.mtc</groupId>
<artifactId>multithreadedtc</artifactId>
<version>${multithreadedtc.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</dependencyManagement>

5
spring-cql/pom.xml
View File

@@ -87,6 +87,11 @@
<artifactId>commons-lang3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>edu.umd.cs.mtc</groupId>
<artifactId>multithreadedtc</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>

85
spring-cql/src/main/java/org/springframework/cassandra/core/CachedPreparedStatementCreator.java
View File

@@ -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.
* <p>
* {@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<Session, Map<String, PreparedStatement>> CACHE = new ConcurrentHashMap<Session, Map<String, PreparedStatement>>();
private final String cql;
private static final Map<Session, Map<String, PreparedStatement>> psMap = new ConcurrentHashMap<Session, Map<String, PreparedStatement>>();
/**
* 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<String, PreparedStatement> sessionCache = getOrCreateSessionLocalCache(session);
return getOrPrepareStatement(session, cacheKey.toString(), sessionCache);
}
private Map<String, PreparedStatement> getOrCreateSessionLocalCache(Session session) {
Map<String, PreparedStatement> sessionMap = CACHE.get(session);
Map<String, PreparedStatement> sessionMap = psMap.get(session);
if (sessionMap == null) {
sessionMap = new ConcurrentHashMap<String, PreparedStatement>();
psMap.put(session, sessionMap);
synchronized (session) {
if (CACHE.containsKey(session)) {
sessionMap = CACHE.get(session);
} else {
sessionMap = new ConcurrentHashMap<String, PreparedStatement>();
CACHE.put(session, sessionMap);
}
}
}
PreparedStatement pstmt = sessionMap.get(keyspaceCQLKey.toString());
return sessionMap;
}
private PreparedStatement getOrPrepareStatement(Session session, String cacheKey,
Map<String, PreparedStatement> 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;
}
}

187
spring-cql/src/test/java/org/springframework/cassandra/core/CachedPreparedStatementCreatorUnitTests.java Normal file
View File

@@ -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> T newProxy(Class<T> theClass, InvocationHandler invocationHandler) {
return (T) Proxy.newProxyInstance(CachedPreparedStatementCreatorUnitTests.class.getClassLoader(),
new Class[] { theClass }, invocationHandler);
}
}