Skip to content

exposes number of repetitions for race tests as environment variable #1015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 1, 2021
Merged

exposes number of repetitions for race tests as environment variable #1015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions rsocket-core/src/test/java/io/rsocket/RaceTestConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
package io.rsocket;

public class RaceTestConstants {
public static final int REPEATS =
Integer.parseInt(System.getProperty("rsocket.test.race.repeats", "1000"));
}
7 changes: 4 additions & 3 deletions rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import io.netty.util.ReferenceCounted;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RaceTestConstants;
import io.rsocket.TestScheduler;
import io.rsocket.exceptions.ApplicationErrorException;
import io.rsocket.exceptions.CustomRSocketException;
Expand Down Expand Up @@ -404,7 +405,7 @@ static Stream<BiFunction<RSocket, Payload, Publisher<?>>> prepareCalls() {
public void checkNoLeaksOnRacing(
Function<ClientSocketRule, Publisher<Payload>> initiator,
BiConsumer<AssertSubscriber<Payload>, ClientSocketRule> runner) {
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
ClientSocketRule clientSocketRule = new ClientSocketRule();
try {
clientSocketRule
Expand Down Expand Up @@ -987,7 +988,7 @@ public void ensuresCorrectOrderOfStreamIdIssuingInCaseOfRacing(
FrameType interactionType2) {
Assumptions.assumeThat(interactionType1).isNotEqualTo(METADATA_PUSH);
Assumptions.assumeThat(interactionType2).isNotEqualTo(METADATA_PUSH);
for (int i = 1; i < 10000; i += 4) {
for (int i = 1; i < RaceTestConstants.REPEATS; i += 4) {
Payload payload = DefaultPayload.create("test", "test");
Publisher<?> publisher1 = interaction1.apply(rule, payload);
Publisher<?> publisher2 = interaction2.apply(rule, payload);
Expand Down Expand Up @@ -1072,7 +1073,7 @@ public void shouldTerminateAllStreamsIfThereRacingBetweenDisposeAndRequests(
BiFunction<ClientSocketRule, Payload, Publisher<?>> interaction2,
FrameType interactionType1,
FrameType interactionType2) {
for (int i = 1; i < 10000; i++) {
for (int i = 1; i < RaceTestConstants.REPEATS; i++) {
Payload payload1 = ByteBufPayload.create("test", "test");
Payload payload2 = ByteBufPayload.create("test", "test");
AssertSubscriber assertSubscriber1 = AssertSubscriber.create();
Expand Down
13 changes: 7 additions & 6 deletions rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import io.netty.util.ReferenceCounted;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RaceTestConstants;
import io.rsocket.frame.CancelFrameCodec;
import io.rsocket.frame.ErrorFrameCodec;
import io.rsocket.frame.FrameHeaderCodec;
Expand Down Expand Up @@ -247,7 +248,7 @@ protected void hookOnSubscribe(Subscription subscription) {
@Test
public void checkNoLeaksOnRacingCancelFromRequestChannelAndNextFromUpstream() {
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create();

rule.setAcceptingSocket(
Expand Down Expand Up @@ -301,7 +302,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
public void checkNoLeaksOnRacingBetweenDownstreamCancelAndOnNextFromRequestChannelTest() {
Hooks.onErrorDropped((e) -> {});
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create();

FluxSink<Payload>[] sinks = new FluxSink[1];
Expand Down Expand Up @@ -340,7 +341,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
public void checkNoLeaksOnRacingBetweenDownstreamCancelAndOnNextFromRequestChannelTest1() {
Hooks.onErrorDropped((e) -> {});
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create();

FluxSink<Payload>[] sinks = new FluxSink[1];
Expand Down Expand Up @@ -382,7 +383,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
checkNoLeaksOnRacingBetweenDownstreamCancelAndOnNextFromUpstreamOnErrorFromRequestChannelTest1() {
Hooks.onErrorDropped((e) -> {});
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
FluxSink<Payload>[] sinks = new FluxSink[1];
AssertSubscriber<Payload> assertSubscriber = AssertSubscriber.create();
rule.setAcceptingSocket(
Expand Down Expand Up @@ -474,7 +475,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
public void checkNoLeaksOnRacingBetweenDownstreamCancelAndOnNextFromRequestStreamTest1() {
Hooks.onErrorDropped((e) -> {});
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
FluxSink<Payload>[] sinks = new FluxSink[1];

rule.setAcceptingSocket(
Expand Down Expand Up @@ -509,7 +510,7 @@ public Flux<Payload> requestStream(Payload payload) {
public void checkNoLeaksOnRacingBetweenDownstreamCancelAndOnNextFromRequestResponseTest1() {
Hooks.onErrorDropped((e) -> {});
ByteBufAllocator allocator = rule.alloc();
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
Operators.MonoSubscriber<Payload, Payload>[] sources = new Operators.MonoSubscriber[1];

rule.setAcceptingSocket(
Expand Down
37 changes: 20 additions & 17 deletions rsocket-core/src/test/java/io/rsocket/core/ReconnectMonoTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import static org.junit.Assert.assertEquals;

import io.rsocket.RaceTestConstants;
import io.rsocket.internal.subscriber.AssertSubscriber;
import java.io.IOException;
import java.time.Duration;
Expand Down Expand Up @@ -60,7 +61,7 @@ public class ReconnectMonoTests {
public void shouldExpireValueOnRacingDisposeAndNext() {
Hooks.onErrorDropped(t -> {});
Hooks.onNextDropped(System.out::println);
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final int index = i;
final CoreSubscriber<? super String>[] monoSubscribers = new CoreSubscriber[1];
Subscription mockSubscription = Mockito.mock(Subscription.class);
Expand Down Expand Up @@ -108,7 +109,7 @@ public void subscribe(CoreSubscriber<? super String> actual) {
@Test
public void shouldNotifyAllTheSubscribersUnderRacingBetweenSubscribeAndComplete() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);

Expand Down Expand Up @@ -151,7 +152,7 @@ public void shouldNotifyAllTheSubscribersUnderRacingBetweenSubscribeAndComplete(
@Test
public void shouldNotExpireNewlyResolvedValueIfSubscribeIsRacingWithInvalidate() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final int index = i;
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);
Expand Down Expand Up @@ -214,7 +215,7 @@ public void shouldNotExpireNewlyResolvedValueIfSubscribeIsRacingWithInvalidate()
@Test
public void shouldNotExpireNewlyResolvedValueIfSubscribeIsRacingWithInvalidates() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final int index = i;
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);
Expand Down Expand Up @@ -281,7 +282,7 @@ public void shouldNotExpireNewlyResolvedValueIfSubscribeIsRacingWithInvalidates(
@Test
public void shouldNotExpireNewlyResolvedValueIfBlockIsRacingWithInvalidate() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final int index = i;
final Mono<String> source =
Mono.fromSupplier(
Expand Down Expand Up @@ -347,7 +348,7 @@ public String get() {

@Test
public void shouldEstablishValueOnceInCaseOfRacingBetweenSubscribers() {
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold = TestPublisher.createCold();
cold.next("value" + i);

Expand Down Expand Up @@ -394,7 +395,7 @@ public void shouldEstablishValueOnceInCaseOfRacingBetweenSubscribers() {
@Test
public void shouldEstablishValueOnceInCaseOfRacingBetweenSubscribeAndBlock() {
Duration timeout = Duration.ofMillis(100);
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold = TestPublisher.createCold();
cold.next("value" + i);

Expand Down Expand Up @@ -441,7 +442,7 @@ public void shouldEstablishValueOnceInCaseOfRacingBetweenSubscribeAndBlock() {
@Test
public void shouldEstablishValueOnceInCaseOfRacingBetweenBlocks() {
Duration timeout = Duration.ofMillis(100);
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold = TestPublisher.createCold();
cold.next("value" + i);

Expand Down Expand Up @@ -486,7 +487,7 @@ public void shouldEstablishValueOnceInCaseOfRacingBetweenBlocks() {
@Test
public void shouldExpireValueOnRacingDisposeAndNoValueComplete() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);

Expand Down Expand Up @@ -524,7 +525,7 @@ public void shouldExpireValueOnRacingDisposeAndNoValueComplete() {
@Test
public void shouldExpireValueOnRacingDisposeAndComplete() {
Hooks.onErrorDropped(t -> {});
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);

Expand Down Expand Up @@ -564,7 +565,7 @@ public void shouldExpireValueOnRacingDisposeAndComplete() {
public void shouldExpireValueOnRacingDisposeAndError() {
Hooks.onErrorDropped(t -> {});
RuntimeException runtimeException = new RuntimeException("test");
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);

Expand Down Expand Up @@ -610,7 +611,7 @@ public void shouldExpireValueOnRacingDisposeAndError() {
public void shouldExpireValueOnRacingDisposeAndErrorWithNoBackoff() {
Hooks.onErrorDropped(t -> {});
RuntimeException runtimeException = new RuntimeException("test");
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold =
TestPublisher.createNoncompliant(TestPublisher.Violation.REQUEST_OVERFLOW);

Expand Down Expand Up @@ -886,19 +887,21 @@ public void shouldNotifyAllTheSubscribers() {

final ArrayList<MonoProcessor<String>> processors = new ArrayList<>(200);

for (int i = 0; i < 100; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final MonoProcessor<String> subA = MonoProcessor.create();
final MonoProcessor<String> subB = MonoProcessor.create();
processors.add(subA);
processors.add(subB);
RaceTestUtils.race(() -> reconnectMono.subscribe(subA), () -> reconnectMono.subscribe(subB));
}

Assertions.assertThat(reconnectMono.resolvingInner.subscribers).hasSize(204);
Assertions.assertThat(reconnectMono.resolvingInner.subscribers)
.hasSize(RaceTestConstants.REPEATS * 2 + 4);

sub1.dispose();

Assertions.assertThat(reconnectMono.resolvingInner.subscribers).hasSize(203);
Assertions.assertThat(reconnectMono.resolvingInner.subscribers)
.hasSize(RaceTestConstants.REPEATS * 2 + 3);

publisher.next("value");

Expand All @@ -917,7 +920,7 @@ public void shouldNotifyAllTheSubscribers() {

@Test
public void shouldExpireValueExactlyOnceOnRacingBetweenInvalidates() {
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold = TestPublisher.createCold();
cold.next("value");
final int timeout = 10;
Expand Down Expand Up @@ -959,7 +962,7 @@ public void shouldExpireValueExactlyOnceOnRacingBetweenInvalidates() {

@Test
public void shouldExpireValueExactlyOnceOnRacingBetweenInvalidateAndDispose() {
for (int i = 0; i < 10000; i++) {
for (int i = 0; i < RaceTestConstants.REPEATS; i++) {
final TestPublisher<String> cold = TestPublisher.createCold();
cold.next("value");
final int timeout = 10000;
Expand Down
Loading