Add integration, load test testing the concurrent access of an HttpSession backed by Spring Session using Apache Geode with 180 concurrent users (Threads) performing 10,000 concurrent data access opeations on the HttpSession to rule out possible race conditions and lost updates.
This commit is contained in:
John Blum
@@ -0,0 +1,416 @@
|
||||
/*
|
||||
* 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.session.data.gemfire;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.springframework.session.data.gemfire.AbstractGemFireOperationsSessionRepository.DeltaCapableGemFireSession;
|
||||
import static org.springframework.session.data.gemfire.AbstractGemFireOperationsSessionRepository.GemFireSession;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.DataOutput;
|
||||
import java.io.IOException;
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
|
||||
import org.apache.geode.cache.DataPolicy;
|
||||
import org.apache.geode.cache.client.ClientRegionShortcut;
|
||||
|
||||
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.data.gemfire.config.annotation.CacheServerApplication;
|
||||
import org.springframework.data.gemfire.config.annotation.ClientCacheApplication;
|
||||
import org.springframework.data.gemfire.tests.util.IdentityHashCodeComparator;
|
||||
import org.springframework.session.Session;
|
||||
import org.springframework.session.data.gemfire.config.annotation.web.http.EnableGemFireHttpSession;
|
||||
import org.springframework.session.data.gemfire.serialization.data.AbstractDataSerializableSessionSerializer;
|
||||
import org.springframework.session.data.gemfire.serialization.data.provider.DataSerializableSessionAttributesSerializer;
|
||||
import org.springframework.session.data.gemfire.serialization.data.provider.DataSerializableSessionSerializer;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringRunner;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Multi-Threaded, Highly-Concurrent, {@link Session} data access operations integration test.
|
||||
*
|
||||
* @author John Blum
|
||||
* @see java.util.concurrent.ExecutorService
|
||||
* @see org.junit.Test
|
||||
* @see org.springframework.data.gemfire.config.annotation.CacheServerApplication
|
||||
* @see org.springframework.data.gemfire.config.annotation.ClientCacheApplication
|
||||
* @see org.springframework.session.Session
|
||||
* @see org.springframework.session.SessionRepository
|
||||
* @see org.springframework.session.data.gemfire.config.annotation.web.http.EnableGemFireHttpSession
|
||||
* @see org.springframework.session.data.gemfire.config.annotation.web.http.GemFireHttpSessionConfiguration
|
||||
* @see org.springframework.test.context.ContextConfiguration
|
||||
* @see org.springframework.test.context.junit4.SpringRunner
|
||||
* @since 2.1.2
|
||||
*/
|
||||
@RunWith(SpringRunner.class)
|
||||
@ContextConfiguration(
|
||||
classes = MultiThreadedHighlyConcurrentClientServerHttpSessionAccessIntegrationTests.GemFireClientConfiguration.class
|
||||
)
|
||||
@SuppressWarnings("unused")
|
||||
public class MultiThreadedHighlyConcurrentClientServerHttpSessionAccessIntegrationTests
|
||||
extends AbstractGemFireIntegrationTests {
|
||||
|
||||
private static final boolean SESSION_REFERENCE_CHECKING_ENABLED = false;
|
||||
|
||||
private static final int THREAD_COUNT = 180;
|
||||
private static final int WORKLOAD = 10000;
|
||||
|
||||
private static final String GEMFIRE_LOG_LEVEL = "error";
|
||||
|
||||
@BeforeClass
|
||||
public static void startGemFireServer() throws IOException {
|
||||
startGemFireServer(GemFireServerConfiguration.class);
|
||||
}
|
||||
|
||||
private final AtomicInteger sessionReferenceComparisonCounter = new AtomicInteger(0);
|
||||
private final AtomicInteger threadCounter = new AtomicInteger(0);
|
||||
|
||||
private final AtomicReference<String> sessionId = new AtomicReference<>(null);
|
||||
|
||||
private final List<String> sessionAttributeNames = Collections.synchronizedList(new ArrayList<>(WORKLOAD));
|
||||
|
||||
private final Random random = new Random(System.currentTimeMillis());
|
||||
|
||||
private final Set<Session> sessionIdentityHashCodes =
|
||||
Collections.synchronizedSet(new TreeSet<>(IdentityHashCodeComparator.INSTANCE));
|
||||
|
||||
private final Set<Session> sessionReferences =
|
||||
Collections.synchronizedSet(new TreeSet<>((sessionOne, sessionTwo) -> sessionOne == sessionTwo ? 0
|
||||
: sessionReferenceComparisonCounter.incrementAndGet() % 2 == 0 ? -1 : 1));
|
||||
|
||||
@Before
|
||||
public void assertGemFireConfiguration() {
|
||||
|
||||
assertThat(this.gemfireCache.getPdxSerializer())
|
||||
.describedAs("Expected the configured PdxSerializer to be null; but was [%s]",
|
||||
ObjectUtils.nullSafeClassName(this.gemfireCache.getPdxSerializer()))
|
||||
.isNull();
|
||||
|
||||
assertThat(this.sessions.getAttributes().getDataPolicy())
|
||||
.describedAs("Expected Region [%s] DataPolicy of EMPTY; but was %s",
|
||||
this.sessions.getName(), this.sessions.getAttributes().getDataPolicy())
|
||||
.isEqualTo(DataPolicy.EMPTY);
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setupSession() {
|
||||
|
||||
Instant beforeCreationTime = Instant.now();
|
||||
|
||||
Session session = createSession();
|
||||
|
||||
assertThat(session).isNotNull();
|
||||
assertThat(session.getId()).isNotEmpty();
|
||||
assertThat(session.getCreationTime()).isAfterOrEqualTo(beforeCreationTime);
|
||||
assertThat(session.getCreationTime()).isBeforeOrEqualTo(Instant.now());
|
||||
assertThat(session.getLastAccessedTime()).isEqualTo(session.getCreationTime());
|
||||
assertThat(session.isExpired()).isFalse();
|
||||
assertThat(session.getAttributeNames()).isEmpty();
|
||||
|
||||
this.sessionId.set(save(touch(session)).getId());
|
||||
}
|
||||
|
||||
private ExecutorService newSessionAccessBatchWorkloadExecutor() {
|
||||
|
||||
return Executors.newFixedThreadPool(THREAD_COUNT, runnable -> {
|
||||
|
||||
Thread sessionThread = new Thread(runnable);
|
||||
|
||||
sessionThread.setDaemon(true);
|
||||
sessionThread.setName(String.format("Session Thread %d", this.threadCounter.incrementAndGet()));
|
||||
sessionThread.setPriority(Thread.NORM_PRIORITY);
|
||||
|
||||
return sessionThread;
|
||||
});
|
||||
}
|
||||
|
||||
private Collection<Callable<Integer>> newSessionAccessBatchWorkload() {
|
||||
|
||||
Collection<Callable<Integer>> sessionAccessWorkload = new ArrayList<>(WORKLOAD);
|
||||
|
||||
for (int count = 0, readCount = 0; count < WORKLOAD; count++, readCount = 3 * count) {
|
||||
|
||||
sessionAccessWorkload.add(count % 79 != 0
|
||||
? newAddSessionAttributeTask()
|
||||
: readCount % 237 != 0
|
||||
? newRemoveSessionAttributeTask()
|
||||
: newSessionReaderTask());
|
||||
|
||||
//sessionAccessWorkload.add(count % 79 != 0 ? newAddSessionAttributeTask() : newRemoveSessionAttributeTask());
|
||||
//sessionAccessWorkload.add(count % 79 != 0 ? newAddSessionAttributeTask() : newSessionReaderTask());
|
||||
//sessionAccessWorkload.add(newAddSessionAttributeTask());
|
||||
}
|
||||
|
||||
return sessionAccessWorkload;
|
||||
}
|
||||
|
||||
private Callable<Integer> newAddSessionAttributeTask() {
|
||||
|
||||
return () -> {
|
||||
|
||||
Instant beforeLastAccessedTime = Instant.now();
|
||||
|
||||
Session session = get(this.sessionId.get());
|
||||
|
||||
assertThat(session).isNotNull();
|
||||
assertThat(session.getId()).isEqualTo(this.sessionId.get());
|
||||
assertThat(session.getLastAccessedTime()).isAfterOrEqualTo(beforeLastAccessedTime);
|
||||
assertThat(session.isExpired()).isFalse();
|
||||
|
||||
if (SESSION_REFERENCE_CHECKING_ENABLED) {
|
||||
assertThat(this.sessionReferences.add(session))
|
||||
.describedAs("Size [%d]", this.sessionReferences.size())
|
||||
.isTrue();
|
||||
}
|
||||
|
||||
String attributeName = UUID.randomUUID().toString();
|
||||
Object attributeValue = System.currentTimeMillis();
|
||||
|
||||
session.setAttribute(attributeName, attributeValue);
|
||||
|
||||
save(touch(session));
|
||||
|
||||
this.sessionAttributeNames.add(attributeName);
|
||||
|
||||
return 1;
|
||||
};
|
||||
}
|
||||
|
||||
@SuppressWarnings("all")
|
||||
private Callable<Integer> newRemoveSessionAttributeTask() {
|
||||
|
||||
return () -> {
|
||||
|
||||
int returnValue = 0;
|
||||
|
||||
Instant beforeLastAccessedTime = Instant.now();
|
||||
|
||||
Session session = get(this.sessionId.get());
|
||||
|
||||
assertThat(session).isNotNull();
|
||||
assertThat(session.getId()).isEqualTo(this.sessionId.get());
|
||||
assertThat(session.getLastAccessedTime()).isAfterOrEqualTo(beforeLastAccessedTime);
|
||||
assertThat(session.isExpired()).isFalse();
|
||||
|
||||
if (SESSION_REFERENCE_CHECKING_ENABLED) {
|
||||
assertThat(this.sessionReferences.add(session))
|
||||
.describedAs("Size [%d]", this.sessionReferences.size())
|
||||
.isTrue();
|
||||
}
|
||||
|
||||
String attributeName = null;
|
||||
|
||||
synchronized (this.sessionAttributeNames) {
|
||||
|
||||
int size = this.sessionAttributeNames.size();
|
||||
|
||||
if (size > 0) {
|
||||
|
||||
int index = this.random.nextInt(size);
|
||||
|
||||
attributeName = this.sessionAttributeNames.remove(index);
|
||||
}
|
||||
}
|
||||
|
||||
if (session.getAttributeNames().contains(attributeName)) {
|
||||
session.removeAttribute(attributeName);
|
||||
returnValue = -1;
|
||||
}
|
||||
else {
|
||||
Optional.ofNullable(attributeName)
|
||||
.filter(StringUtils::hasText)
|
||||
.ifPresent(this.sessionAttributeNames::add);
|
||||
}
|
||||
|
||||
save(touch(session));
|
||||
|
||||
return returnValue;
|
||||
};
|
||||
}
|
||||
|
||||
private Callable<Integer> newSessionReaderTask() {
|
||||
|
||||
return () -> {
|
||||
|
||||
Instant beforeLastAccessedTime = Instant.now();
|
||||
|
||||
Session session = get(this.sessionId.get());
|
||||
|
||||
assertThat(session).isNotNull();
|
||||
assertThat(session.getId()).isEqualTo(this.sessionId.get());
|
||||
assertThat(session.getLastAccessedTime()).isAfterOrEqualTo(beforeLastAccessedTime);
|
||||
assertThat(session.isExpired()).isFalse();
|
||||
|
||||
if (SESSION_REFERENCE_CHECKING_ENABLED) {
|
||||
assertThat(this.sessionReferences.add(session))
|
||||
.describedAs("Size [%d]", this.sessionReferences.size())
|
||||
.isTrue();
|
||||
}
|
||||
|
||||
save(session);
|
||||
|
||||
return 0;
|
||||
};
|
||||
}
|
||||
|
||||
private <T> T safeFutureGet(Future<T> operation) {
|
||||
|
||||
try {
|
||||
return operation.get();
|
||||
}
|
||||
catch (Exception cause) {
|
||||
throw new RuntimeException("SESSION ACCESS TASK FAILURE", cause);
|
||||
}
|
||||
}
|
||||
|
||||
private int runSessionAccessBatchWorkload() throws InterruptedException {
|
||||
|
||||
ExecutorService sessionAccessExecutor = newSessionAccessBatchWorkloadExecutor();
|
||||
|
||||
try {
|
||||
|
||||
List<Future<Integer>> sessionAccessFutures =
|
||||
sessionAccessExecutor.invokeAll(newSessionAccessBatchWorkload());
|
||||
|
||||
return sessionAccessFutures.stream()
|
||||
.mapToInt(this::safeFutureGet)
|
||||
.sum();
|
||||
}
|
||||
finally {
|
||||
Optional.of(sessionAccessExecutor)
|
||||
.ifPresent(ExecutorService::shutdownNow);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void concurrentSessionAccessIsCorrect() throws InterruptedException {
|
||||
|
||||
int sessionAttributeCount = runSessionAccessBatchWorkload();
|
||||
|
||||
assertThat(sessionAttributeCount).isEqualTo(this.sessionAttributeNames.size());
|
||||
//assertThat(SpyingDataSerializableSessionSerializer.getSerializationCount()).isEqualTo(1);
|
||||
|
||||
Session session = get(this.sessionId.get());
|
||||
|
||||
assertThat(session).isNotNull();
|
||||
assertThat(session.getId()).isEqualTo(this.sessionId.get());
|
||||
assertThat(session.getAttributeNames()).hasSize(sessionAttributeCount);
|
||||
}
|
||||
|
||||
//@ClientCacheApplication(logLevel = GEMFIRE_LOG_LEVEL, subscriptionEnabled = true)
|
||||
@ClientCacheApplication(copyOnRead = true, logLevel = GEMFIRE_LOG_LEVEL, subscriptionEnabled = true)
|
||||
@EnableGemFireHttpSession(
|
||||
clientRegionShortcut = ClientRegionShortcut.PROXY,
|
||||
poolName = "DEFAULT",
|
||||
regionName = "Sessions",
|
||||
sessionSerializerBeanName = "spyingSessionSerializer"
|
||||
)
|
||||
static class GemFireClientConfiguration {
|
||||
|
||||
@Bean
|
||||
SpyingDataSerializableSessionSerializer spyingSessionSerializer() {
|
||||
return new SpyingDataSerializableSessionSerializer();
|
||||
}
|
||||
}
|
||||
|
||||
@CacheServerApplication(
|
||||
name = "MultiThreadedHighlyConcurrentClientServerHttpSessionAccessIntegrationTests",
|
||||
logLevel = GEMFIRE_LOG_LEVEL
|
||||
)
|
||||
//@EnableLogging(logFile = "gemfire.log", logLevel = "debug")
|
||||
@EnableGemFireHttpSession(
|
||||
regionName = "Sessions",
|
||||
sessionSerializerBeanName = "spyingSessionSerializer"
|
||||
)
|
||||
static class GemFireServerConfiguration {
|
||||
|
||||
public static void main(String[] args) {
|
||||
|
||||
AnnotationConfigApplicationContext applicationContext =
|
||||
new AnnotationConfigApplicationContext(GemFireServerConfiguration.class);
|
||||
|
||||
applicationContext.registerShutdownHook();
|
||||
}
|
||||
|
||||
@Bean
|
||||
SpyingDataSerializableSessionSerializer spyingSessionSerializer() {
|
||||
return new SpyingDataSerializableSessionSerializer();
|
||||
}
|
||||
}
|
||||
|
||||
static class SpyingDataSerializableSessionSerializer
|
||||
extends AbstractDataSerializableSessionSerializer<GemFireSession> {
|
||||
|
||||
private static final AtomicInteger serializationCount = new AtomicInteger(0);
|
||||
|
||||
private final DataSerializableSessionSerializer sessionSerializer;
|
||||
|
||||
static int getSerializationCount() {
|
||||
return serializationCount.get();
|
||||
}
|
||||
|
||||
public SpyingDataSerializableSessionSerializer() {
|
||||
this.sessionSerializer = new DataSerializableSessionSerializer();
|
||||
DataSerializableSessionAttributesSerializer.register();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Class<?>[] getSupportedClasses() {
|
||||
return this.sessionSerializer.getSupportedClasses();
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public void serialize(GemFireSession session, DataOutput dataOutput) {
|
||||
|
||||
assertThat(session).isInstanceOf(DeltaCapableGemFireSession.class);
|
||||
//assertThat(session.hasDelta()).isTrue();
|
||||
|
||||
this.sessionSerializer.serialize(session, dataOutput);
|
||||
|
||||
serializationCount.incrementAndGet();
|
||||
}
|
||||
|
||||
@Override
|
||||
public GemFireSession deserialize(DataInput dataInput) {
|
||||
return this.sessionSerializer.deserialize(dataInput);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user