Skip to content

Commit 74b793a

Browse files
zhongkechenzhongkechen
authored
add test cases for virtual thread pool (#324)
1 parent b351a08 commit 74b793a

12 files changed

Lines changed: 296 additions & 53 deletions

File tree

‎.github/workflows/e2e-tests.yml‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,12 @@ jobs:
5757
role-to-assume: "${{ secrets.ACTIONS_INTEGRATION_ROLE_NAME }}"
5858
role-session-name: java-language-sdk-test
5959
aws-region: ${{ env.AWS_REGION }}
60+
- name: Setup Java ${{ matrix.java }}
61+
uses: actions/setup-java@v5
62+
with:
63+
distribution: corretto
64+
java-version: ${{ matrix.java }}
65+
cache: maven
6066
- name: Build locally
6167
run: mvn -B -q -Dmaven.test.skip=true install --file pom.xml
6268
- name: sam build
@@ -70,14 +76,6 @@ jobs:
7076
--resolve-image-repos --resolve-s3 --capabilities CAPABILITY_IAM --parameter-overrides \
7177
'ParameterKey=Architecture,ParameterValue=x86_64 ParameterKey=JavaVersion,ParameterValue=java${{ matrix.java }}'
7278
working-directory: ./examples
73-
- name: Setup Java ${{ matrix.java }}
74-
uses: actions/setup-java@v5
75-
with:
76-
distribution: corretto
77-
java-version: ${{ matrix.java }}
78-
cache: maven
79-
- name: Build locally
80-
run: mvn -B -q -Dmaven.test.skip=true install --file pom.xml
8179
- name: Cloud Based Integration Tests
8280
run: mvn clean test -B -Dtest.cloud.enabled=true -Dtest=CloudBasedIntegrationTest -Dtest.function.name.suffix='-java${{ matrix.java }}-runtime'
8381
working-directory: ./examples

‎examples/pom.xml‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,4 +140,38 @@
140140
</plugin>
141141
</plugins>
142142
</build>
143+
144+
<profiles>
145+
<profile>
146+
<id>exclude-virtual-threads</id>
147+
<activation>
148+
<jdk>[,21)</jdk>
149+
</activation>
150+
<build>
151+
<plugins>
152+
<plugin>
153+
<groupId>org.apache.maven.plugins</groupId>
154+
<artifactId>maven-compiler-plugin</artifactId>
155+
<configuration>
156+
<excludes>
157+
<exclude>**/vt/ManyAsyncStepsVirtualThreadPoolExample.java</exclude>
158+
</excludes>
159+
<testExcludes>
160+
<testExclude>**/vt/ManyAsyncStepsVirtualThreadPoolExampleTest.java</testExclude>
161+
</testExcludes>
162+
</configuration>
163+
</plugin>
164+
<plugin>
165+
<groupId>org.apache.maven.plugins</groupId>
166+
<artifactId>maven-surefire-plugin</artifactId>
167+
<configuration>
168+
<excludes>
169+
<exclude>**/vt/ManyAsyncStepsVirtualThreadPoolExampleTest.java</exclude>
170+
</excludes>
171+
</configuration>
172+
</plugin>
173+
</plugins>
174+
</build>
175+
</profile>
176+
</profiles>
143177
</project>

‎examples/src/main/java/software/amazon/lambda/durable/examples/child/ManyAsyncChildContextExample.java‎

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import software.amazon.lambda.durable.DurableContext;
1010
import software.amazon.lambda.durable.DurableFuture;
1111
import software.amazon.lambda.durable.DurableHandler;
12+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsInput;
13+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsOutput;
1214

1315
/**
1416
* Performance test example demonstrating concurrent async child contexts.
@@ -21,15 +23,10 @@
2123
* <li>All results are collected using {@link DurableFuture#allOf}
2224
* </ul>
2325
*/
24-
public class ManyAsyncChildContextExample
25-
extends DurableHandler<ManyAsyncChildContextExample.Input, ManyAsyncChildContextExample.Output> {
26-
27-
public record Input(int multiplier, int steps) {}
28-
29-
public record Output(long result, long executionTimeMs, long replayTimeMs) {}
26+
public class ManyAsyncChildContextExample extends DurableHandler<ManyAsyncStepsInput, ManyAsyncStepsOutput> {
3027

3128
@Override
32-
public Output handleRequest(Input input, DurableContext context) {
29+
public ManyAsyncStepsOutput handleRequest(ManyAsyncStepsInput input, DurableContext context) {
3330
var startTime = System.nanoTime();
3431
var multiplier = input.multiplier();
3532
var steps = input.steps();
@@ -65,7 +62,7 @@ public Output handleRequest(Input input, DurableContext context) {
6562

6663
var replayTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
6764

68-
return new Output(totalSum, executionTimeMs, replayTimeMs);
65+
return new ManyAsyncStepsOutput(totalSum, executionTimeMs, replayTimeMs);
6966
}
7067

7168
@Override

‎examples/src/main/java/software/amazon/lambda/durable/examples/step/ManyAsyncStepsExample.java‎

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import software.amazon.lambda.durable.DurableContext;
1010
import software.amazon.lambda.durable.DurableFuture;
1111
import software.amazon.lambda.durable.DurableHandler;
12+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsInput;
13+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsOutput;
1214

1315
/**
1416
* Performance test example demonstrating concurrent async steps.
@@ -21,14 +23,10 @@
2123
* <li>All results are collected using {@link DurableFuture#allOf}
2224
* </ul>
2325
*/
24-
public class ManyAsyncStepsExample extends DurableHandler<ManyAsyncStepsExample.Input, ManyAsyncStepsExample.Output> {
25-
26-
public record Input(int multiplier, int steps) {}
27-
28-
public record Output(long result, long executionTimeMs, long replayTimeMs) {}
26+
public class ManyAsyncStepsExample extends DurableHandler<ManyAsyncStepsInput, ManyAsyncStepsOutput> {
2927

3028
@Override
31-
public Output handleRequest(Input input, DurableContext context) {
29+
public ManyAsyncStepsOutput handleRequest(ManyAsyncStepsInput input, DurableContext context) {
3230
var startTime = System.nanoTime();
3331
var multiplier = input.multiplier();
3432
var steps = input.steps();
@@ -60,7 +58,7 @@ public Output handleRequest(Input input, DurableContext context) {
6058

6159
var replayTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
6260

63-
return new Output(totalSum, executionTimeMs, replayTimeMs);
61+
return new ManyAsyncStepsOutput(totalSum, executionTimeMs, replayTimeMs);
6462
}
6563

6664
@Override

‎examples/src/main/java/software/amazon/lambda/durable/examples/types/ManyAsyncStepsInput.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable.examples.types;
4+
5+
public record ManyAsyncStepsInput(int multiplier, int steps) {}

‎examples/src/main/java/software/amazon/lambda/durable/examples/types/ManyAsyncStepsOutput.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable.examples.types;
4+
5+
public record ManyAsyncStepsOutput(long result, long executionTimeMs, long replayTimeMs) {}

‎examples/src/main/java/software/amazon/lambda/durable/examples/vt/ManyAsyncStepsVirtualThreadPoolExample.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: Apache-2.0
3+
package software.amazon.lambda.durable.examples.vt;
4+
5+
import java.time.Duration;
6+
import java.util.ArrayList;
7+
import java.util.concurrent.Executors;
8+
import java.util.concurrent.TimeUnit;
9+
import software.amazon.lambda.durable.DurableConfig;
10+
import software.amazon.lambda.durable.DurableContext;
11+
import software.amazon.lambda.durable.DurableFuture;
12+
import software.amazon.lambda.durable.DurableHandler;
13+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsInput;
14+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsOutput;
15+
16+
/**
17+
* Performance test example demonstrating concurrent async steps.
18+
*
19+
* <p>This example tests the SDK's ability to handle many concurrent operations:
20+
*
21+
* <ul>
22+
* <li>Creates async steps in a loop
23+
* <li>Each step performs a simple computation
24+
* <li>All results are collected using {@link DurableFuture#allOf}
25+
* </ul>
26+
*/
27+
public class ManyAsyncStepsVirtualThreadPoolExample extends DurableHandler<ManyAsyncStepsInput, ManyAsyncStepsOutput> {
28+
29+
@Override
30+
public ManyAsyncStepsOutput handleRequest(ManyAsyncStepsInput input, DurableContext context) {
31+
var startTime = System.nanoTime();
32+
var multiplier = input.multiplier();
33+
var steps = input.steps();
34+
var logger = context.getLogger();
35+
36+
logger.info("Starting {} async steps with multiplier {}", steps, multiplier);
37+
38+
// Create async steps
39+
var futures = new ArrayList<DurableFuture<Integer>>(steps);
40+
for (var i = 0; i < steps; i++) {
41+
var index = i;
42+
var future = context.stepAsync("compute-" + i, Integer.class, stepCtx -> index * multiplier);
43+
futures.add(future);
44+
}
45+
46+
logger.info("All {} async steps created, collecting results", steps);
47+
48+
// Collect all results using allOf
49+
var results = DurableFuture.allOf(futures);
50+
var totalSum = results.stream().mapToInt(Integer::intValue).sum();
51+
52+
// checkpoint the executionTime so that we can have the same value when replay
53+
var executionTimeMs = context.step(
54+
"execution-time", Long.class, stepCtx -> TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime));
55+
logger.info("Completed {} steps, total sum: {}, execution time: {}ms", steps, totalSum, executionTimeMs);
56+
57+
// Wait 2 seconds to test replay
58+
context.wait("post-compute-wait", Duration.ofSeconds(2));
59+
60+
var replayTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
61+
62+
return new ManyAsyncStepsOutput(totalSum, executionTimeMs, replayTimeMs);
63+
}
64+
65+
@Override
66+
protected DurableConfig createConfiguration() {
67+
// Add a small checkpoint delay to help batch the checkpoint requests and reduce the overall latencies
68+
// when the function has many concurrent operations
69+
return DurableConfig.builder()
70+
.withCheckpointDelay(Duration.ofMillis(10))
71+
.withExecutorService(Executors.newVirtualThreadPerTaskExecutor())
72+
.build();
73+
}
74+
}

‎examples/src/test/java/software/amazon/lambda/durable/examples/CloudBasedIntegrationTest.java‎

Lines changed: 48 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@
1212
import java.util.concurrent.atomic.AtomicInteger;
1313
import org.junit.jupiter.api.BeforeAll;
1414
import org.junit.jupiter.api.Test;
15+
import org.junit.jupiter.api.condition.EnabledForJreRange;
1516
import org.junit.jupiter.api.condition.EnabledIf;
17+
import org.junit.jupiter.api.condition.JRE;
1618
import org.junit.jupiter.params.ParameterizedTest;
1719
import org.junit.jupiter.params.provider.CsvSource;
1820
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
@@ -22,11 +24,11 @@
2224
import software.amazon.awssdk.services.lambda.model.OperationStatus;
2325
import software.amazon.awssdk.services.sts.StsClient;
2426
import software.amazon.lambda.durable.TypeToken;
25-
import software.amazon.lambda.durable.examples.child.ManyAsyncChildContextExample;
2627
import software.amazon.lambda.durable.examples.general.GenericTypesExample;
27-
import software.amazon.lambda.durable.examples.step.ManyAsyncStepsExample;
2828
import software.amazon.lambda.durable.examples.types.ApprovalRequest;
2929
import software.amazon.lambda.durable.examples.types.GreetingRequest;
30+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsInput;
31+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsOutput;
3032
import software.amazon.lambda.durable.examples.wait.ConcurrentWaitForConditionExample;
3133
import software.amazon.lambda.durable.model.ExecutionStatus;
3234
import software.amazon.lambda.durable.serde.JacksonSerDes;
@@ -536,10 +538,10 @@ void testManyAsyncStepsExample(int steps, long maxExecutionTime, long maxReplayT
536538
for (var i = 0; i < PERFORMANCE_TEST_REPEAT; i++) {
537539
var runner = CloudDurableTestRunner.create(
538540
arn("many-async-steps-example"),
539-
ManyAsyncStepsExample.Input.class,
540-
ManyAsyncStepsExample.Output.class,
541+
ManyAsyncStepsInput.class,
542+
ManyAsyncStepsOutput.class,
541543
lambdaClient);
542-
var result = runner.run(new ManyAsyncStepsExample.Input(2, steps));
544+
var result = runner.run(new ManyAsyncStepsInput(2, steps));
543545

544546
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
545547

@@ -565,6 +567,44 @@ void testManyAsyncStepsExample(int steps, long maxExecutionTime, long maxReplayT
565567
assertTrue(minimalExecutionTimeMs < maxExecutionTime);
566568
}
567569

570+
@EnabledForJreRange(min = JRE.JAVA_21)
571+
@ParameterizedTest
572+
@CsvSource({"100, 1000, 20", "500, 2000, 30", "1000, 3000, 50"})
573+
void testManyAsyncStepsVirtualThreadExample(int steps, long maxExecutionTime, long maxReplayTime) {
574+
long minimalExecutionTimeMs = Long.MAX_VALUE;
575+
long minimalReplayTimeMs = Long.MAX_VALUE;
576+
for (var i = 0; i < PERFORMANCE_TEST_REPEAT; i++) {
577+
var runner = CloudDurableTestRunner.create(
578+
arn("many-async-steps-virtual-thread-pool-example"),
579+
ManyAsyncStepsInput.class,
580+
ManyAsyncStepsOutput.class,
581+
lambdaClient);
582+
var result = runner.run(new ManyAsyncStepsInput(2, steps));
583+
584+
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
585+
586+
var finalResult = result.getResult();
587+
System.out.printf("ManyAsyncStepsVirtualThreadPoolExample result (%d steps): %s\n", steps, finalResult);
588+
assertNotNull(finalResult);
589+
assertEquals((long) steps * (steps - 1), finalResult.result()); // Sum of 0..steps * 2
590+
591+
// Verify some operations are tracked
592+
assertNotNull(runner.getOperation("compute-0"));
593+
assertNotNull(runner.getOperation("compute-" + (steps - 1)));
594+
595+
if (finalResult.executionTimeMs() < minimalExecutionTimeMs) {
596+
minimalExecutionTimeMs = finalResult.executionTimeMs();
597+
}
598+
599+
if (finalResult.replayTimeMs() < minimalReplayTimeMs) {
600+
minimalReplayTimeMs = finalResult.replayTimeMs();
601+
}
602+
}
603+
604+
assertTrue(minimalReplayTimeMs < maxReplayTime);
605+
assertTrue(minimalExecutionTimeMs < maxExecutionTime);
606+
}
607+
568608
@ParameterizedTest
569609
// OOM if it creates 1000 child contexts
570610
@CsvSource({"100, 1500, 10", "500, 3000, 20"})
@@ -574,10 +614,10 @@ void testManyAsyncChildContextExample(int steps, long maxExecutionTime, long max
574614
for (var i = 0; i < PERFORMANCE_TEST_REPEAT; i++) {
575615
var runner = CloudDurableTestRunner.create(
576616
arn("many-async-child-context-example"),
577-
ManyAsyncChildContextExample.Input.class,
578-
ManyAsyncChildContextExample.Output.class,
617+
ManyAsyncStepsInput.class,
618+
ManyAsyncStepsOutput.class,
579619
lambdaClient);
580-
var result = runner.run(new ManyAsyncChildContextExample.Input(2, steps));
620+
var result = runner.run(new ManyAsyncStepsInput(2, steps));
581621

582622
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
583623

‎examples/src/test/java/software/amazon/lambda/durable/examples/child/ManyAsyncChildContextExampleTest.java‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,8 @@
66
import static org.junit.jupiter.api.Assertions.assertNotNull;
77

88
import org.junit.jupiter.api.Test;
9-
import software.amazon.lambda.durable.examples.step.ManyAsyncStepsExample;
9+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsInput;
10+
import software.amazon.lambda.durable.examples.types.ManyAsyncStepsOutput;
1011
import software.amazon.lambda.durable.model.ExecutionStatus;
1112
import software.amazon.lambda.durable.testing.LocalDurableTestRunner;
1213

@@ -15,14 +16,14 @@ class ManyAsyncChildContextExampleTest {
1516
@Test
1617
void testManyAsyncSteps() {
1718
var handler = new ManyAsyncChildContextExample();
18-
var runner = LocalDurableTestRunner.create(ManyAsyncChildContextExample.Input.class, handler);
19+
var runner = LocalDurableTestRunner.create(ManyAsyncStepsInput.class, handler);
1920

20-
var input = new ManyAsyncChildContextExample.Input(2, 500);
21+
var input = new ManyAsyncStepsInput(2, 500);
2122
var result = runner.runUntilComplete(input);
2223

2324
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
2425

25-
var output = result.getResult(ManyAsyncStepsExample.Output.class);
26+
var output = result.getResult(ManyAsyncStepsOutput.class);
2627
assertNotNull(output);
2728

2829
// Sum of 0..499 * 2 = 499 * 500 / 2 * 2 = 249500
@@ -32,25 +33,23 @@ void testManyAsyncSteps() {
3233
@Test
3334
void testManyAsyncStepsWithDefaultMultiplier() {
3435
var handler = new ManyAsyncChildContextExample();
35-
var runner = LocalDurableTestRunner.create(ManyAsyncChildContextExample.Input.class, handler);
36+
var runner = LocalDurableTestRunner.create(ManyAsyncStepsInput.class, handler);
3637

37-
var input = new ManyAsyncChildContextExample.Input(1, 500);
38+
var input = new ManyAsyncStepsInput(1, 500);
3839
var result = runner.runUntilComplete(input);
3940

4041
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
4142

4243
// Sum of 0..499 = 499 * 500 / 2 = 124750
43-
assertEquals(
44-
124750,
45-
result.getResult(ManyAsyncChildContextExample.Output.class).result());
44+
assertEquals(124750, result.getResult(ManyAsyncStepsOutput.class).result());
4645
}
4746

4847
@Test
4948
void testOperationsAreTracked() {
5049
var handler = new ManyAsyncChildContextExample();
51-
var runner = LocalDurableTestRunner.create(ManyAsyncChildContextExample.Input.class, handler);
50+
var runner = LocalDurableTestRunner.create(ManyAsyncStepsInput.class, handler);
5251

53-
var result = runner.runUntilComplete(new ManyAsyncChildContextExample.Input(1, 500));
52+
var result = runner.runUntilComplete(new ManyAsyncStepsInput(1, 500));
5453

5554
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
5655

0 commit comments

Comments
 (0)