From 6bba66288c42aaf6f348c8cab12e77325a888036 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Tue, 6 Jun 2023 20:43:42 +0300 Subject: [PATCH 01/13] bumps lib versions Signed-off-by: Oleh Dokuka --- build.gradle | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/build.gradle b/build.gradle index 079a0e1d9..62ee687da 100644 --- a/build.gradle +++ b/build.gradle @@ -16,11 +16,11 @@ plugins { id 'com.github.sherter.google-java-format' version '0.9' apply false - id 'me.champeau.jmh' version '0.6.7' apply false - id 'io.spring.dependency-management' version '1.0.15.RELEASE' apply false + id 'me.champeau.jmh' version '0.7.1' apply false + id 'io.spring.dependency-management' version '1.1.0' apply false id 'io.morethan.jmhreport' version '0.9.0' apply false - id 'io.github.reyerizo.gradle.jcstress' version '0.8.13' apply false - id 'com.github.vlsi.gradle-extensions' version '1.76' apply false + id 'io.github.reyerizo.gradle.jcstress' version '0.8.15' apply false + id 'com.github.vlsi.gradle-extensions' version '1.89' apply false } boolean isCiServer = ["CI", "CONTINUOUS_INTEGRATION", "TRAVIS", "CIRCLECI", "bamboo_planKey", "GITHUB_ACTION"].with { @@ -33,21 +33,21 @@ subprojects { apply plugin: 'com.github.sherter.google-java-format' apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.31-SNAPSHOT' + ext['reactor-bom.version'] = '2020.0.32' ext['logback.version'] = '1.2.10' - ext['netty-bom.version'] = '4.1.90.Final' - ext['netty-boringssl.version'] = '2.0.59.Final' + ext['netty-bom.version'] = '4.1.93.Final' + ext['netty-boringssl.version'] = '2.0.61.Final' ext['hdrhistogram.version'] = '2.1.12' - ext['mockito.version'] = '4.4.0' + ext['mockito.version'] = '4.11.0' ext['slf4j.version'] = '1.7.36' - ext['jmh.version'] = '1.35' - ext['junit.version'] = '5.8.1' - ext['micrometer.version'] = '1.10.0' - ext['micrometer-tracing.version'] = '1.0.0' - ext['assertj.version'] = '3.22.0' + ext['jmh.version'] = '1.36' + ext['junit.version'] = '5.9.3' + ext['micrometer.version'] = '1.11.0' + ext['micrometer-tracing.version'] = '1.1.1' + ext['assertj.version'] = '3.24.2' ext['netflix.limits.version'] = '0.3.6' ext['bouncycastle-bcpkix.version'] = '1.70' - ext['awaitility.version'] = '4.1.1' + ext['awaitility.version'] = '4.2.0' group = "io.rsocket" From cb811cf6c77c59cc4b62d7caf356f203c92c22c5 Mon Sep 17 00:00:00 2001 From: OlegDokuka Date: Fri, 9 Jun 2023 20:57:59 +0300 Subject: [PATCH 02/13] increment version Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka Signed-off-by: OlegDokuka --- gradle.properties | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/gradle.properties b/gradle.properties index 7f8f4ca23..ce5421125 100644 --- a/gradle.properties +++ b/gradle.properties @@ -11,5 +11,5 @@ # See the License for the specific language governing permissions and # limitations under the License. # -version=1.1.4 -perfBaselineVersion=1.1.3 +version=1.1.5 +perfBaselineVersion=1.1.4 From f591f9d8313b79ea674ee7786000bb1bfd48e1d8 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Tue, 27 Feb 2024 13:08:04 +0200 Subject: [PATCH 03/13] update versions and fixes memleak in UnboundedProcessor (#1106) This PR updates dependencies and makes minor modifications to UnboundedProcessor due to repeating failures of UnboundedProcessorJCStreassTest, which started reproducing some unspotted issues. Motivation: UnboundedProcessor is a critical component in the RSocket-Java ecosystem and must work properly. After analysis of its internal state machine, it was spotted that sometimes: The request may not be delivered due to natural concurrency The terminal signal may not be delivered since it checks for demand which might be consumed already (due to natural concurrency) The final value could be delivered violating reactive-streams spec Modifications: This PR adds a minimal set of changes, preserving old implementation but eliminating the mentioned bugs --------- Signed-off-by: Oleh Dokuka --- build.gradle | 8 +- rsocket-core/build.gradle | 5 +- .../UnboundedProcessorStressTest.java | 52 ++++- .../java/io/rsocket/utils/FastLogger.java | 137 +++++++++++ .../rsocket/internal/UnboundedProcessor.java | 220 ++++++++++++++---- 5 files changed, 370 insertions(+), 52 deletions(-) create mode 100644 rsocket-core/src/jcstress/java/io/rsocket/utils/FastLogger.java diff --git a/build.gradle b/build.gradle index 62ee687da..ae8f05d06 100644 --- a/build.gradle +++ b/build.gradle @@ -33,10 +33,10 @@ subprojects { apply plugin: 'com.github.sherter.google-java-format' apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.32' - ext['logback.version'] = '1.2.10' - ext['netty-bom.version'] = '4.1.93.Final' - ext['netty-boringssl.version'] = '2.0.61.Final' + ext['reactor-bom.version'] = '2020.0.39' + ext['logback.version'] = '1.3.14' + ext['netty-bom.version'] = '4.1.106.Final' + ext['netty-boringssl.version'] = '2.0.62.Final' ext['hdrhistogram.version'] = '2.1.12' ext['mockito.version'] = '4.11.0' ext['slf4j.version'] = '1.7.36' diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 6f2056da0..da5b69b14 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -41,13 +41,14 @@ dependencies { testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' jcstressImplementation(project(":rsocket-test")) + jcstressImplementation 'org.slf4j:slf4j-api' jcstressImplementation "ch.qos.logback:logback-classic" jcstressImplementation 'io.projectreactor:reactor-test' } jcstress { - mode = 'quick' //quick, default, tough - jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.15" + mode = 'sanity' //sanity, quick, default, tough + jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.16" } jar { diff --git a/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java b/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java index 2f5e51f0e..a2d9fcf4d 100644 --- a/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java +++ b/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java @@ -3,6 +3,9 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.UnpooledByteBufAllocator; import io.rsocket.core.StressSubscriber; +import io.rsocket.utils.FastLogger; +import java.util.Arrays; +import java.util.ConcurrentModificationException; import org.openjdk.jcstress.annotations.Actor; import org.openjdk.jcstress.annotations.Arbiter; import org.openjdk.jcstress.annotations.Expect; @@ -14,6 +17,7 @@ import org.openjdk.jcstress.infra.results.L_Result; import reactor.core.Fuseable; import reactor.core.publisher.Hooks; +import reactor.util.Logger; public abstract class UnboundedProcessorStressTest { @@ -21,7 +25,9 @@ public abstract class UnboundedProcessorStressTest { Hooks.onErrorDropped(t -> {}); } - final UnboundedProcessor unboundedProcessor = new UnboundedProcessor(); + final Logger logger = new FastLogger(getClass().getName()); + + final UnboundedProcessor unboundedProcessor = new UnboundedProcessor(logger); @JCStressTest @Outcome( @@ -145,6 +151,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -270,6 +278,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -375,6 +385,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -476,6 +488,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -578,6 +592,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -701,6 +717,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -781,6 +799,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -837,9 +857,15 @@ public void arbiter(LLL_Result r) { + stressSubscriber.onErrorCalls * 2 + stressSubscriber.droppedErrors.size() * 3; + if (stressSubscriber.concurrentOnNext || stressSubscriber.concurrentOnComplete) { + throw new ConcurrentModificationException("boo"); + } + stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -892,6 +918,8 @@ public void arbiter(LLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1107,6 +1135,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1238,6 +1268,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1390,6 +1422,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1522,6 +1556,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1587,6 +1623,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1652,6 +1690,8 @@ public void arbiter(LLLL_Result r) { stressSubscriber.values.forEach(ByteBuf::release); r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + + checkOutcomes(this, r.toString(), logger); } } @@ -1678,6 +1718,16 @@ public void subscribe2() { @Arbiter public void arbiter(L_Result r) { r.r1 = stressSubscriber1.onErrorCalls + stressSubscriber2.onErrorCalls; + + checkOutcomes(this, r.toString(), logger); + } + } + + static void checkOutcomes(Object instance, String result, Logger logger) { + if (Arrays.stream(instance.getClass().getDeclaredAnnotationsByType(Outcome.class)) + .flatMap(o -> Arrays.stream(o.id())) + .noneMatch(s -> s.equalsIgnoreCase(result))) { + throw new RuntimeException(result + " " + logger); } } } diff --git a/rsocket-core/src/jcstress/java/io/rsocket/utils/FastLogger.java b/rsocket-core/src/jcstress/java/io/rsocket/utils/FastLogger.java new file mode 100644 index 000000000..c301d87cf --- /dev/null +++ b/rsocket-core/src/jcstress/java/io/rsocket/utils/FastLogger.java @@ -0,0 +1,137 @@ +package io.rsocket.utils; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import reactor.util.Logger; + +/** + * Implementation of {@link Logger} which is based on the {@link ThreadLocal} based queue which + * collects all the events on the per-thread basis.
Such logger is designed to have all events + * stored during the stress-test run and then sorted and printed out once all the Threads completed + * execution (inside the {@link org.openjdk.jcstress.annotations.Arbiter} annotated method.
+ * Note, this implementation only supports trace-level logs and ignores all others, it is intended + * to be used by {@link reactor.core.publisher.StateLogger}. + */ +public class FastLogger implements Logger { + + final Map> queues = new ConcurrentHashMap<>(); + + final ThreadLocal> logsQueueLocal = + ThreadLocal.withInitial( + () -> { + final ArrayList logs = new ArrayList<>(100); + queues.put(Thread.currentThread(), logs); + return logs; + }); + + private final String name; + + public FastLogger(String name) { + this.name = name; + } + + @Override + public String toString() { + return queues + .values() + .stream() + .flatMap(List::stream) + .sorted( + Comparator.comparingLong( + s -> { + Pattern pattern = Pattern.compile("\\[(.*?)]"); + Matcher matcher = pattern.matcher(s); + matcher.find(); + return Long.parseLong(matcher.group(1)); + })) + .collect(Collectors.joining("\n")); + } + + @Override + public String getName() { + return this.name; + } + + @Override + public boolean isTraceEnabled() { + return true; + } + + @Override + public void trace(String msg) { + logsQueueLocal.get().add(String.format("[%s] %s", System.nanoTime(), msg)); + } + + @Override + public void trace(String format, Object... arguments) { + trace(String.format(format, arguments)); + } + + @Override + public void trace(String msg, Throwable t) { + trace(String.format("%s, %s", msg, Arrays.toString(t.getStackTrace()))); + } + + @Override + public boolean isDebugEnabled() { + return false; + } + + @Override + public void debug(String msg) {} + + @Override + public void debug(String format, Object... arguments) {} + + @Override + public void debug(String msg, Throwable t) {} + + @Override + public boolean isInfoEnabled() { + return false; + } + + @Override + public void info(String msg) {} + + @Override + public void info(String format, Object... arguments) {} + + @Override + public void info(String msg, Throwable t) {} + + @Override + public boolean isWarnEnabled() { + return false; + } + + @Override + public void warn(String msg) {} + + @Override + public void warn(String format, Object... arguments) {} + + @Override + public void warn(String msg, Throwable t) {} + + @Override + public boolean isErrorEnabled() { + return false; + } + + @Override + public void error(String msg) {} + + @Override + public void error(String format, Object... arguments) {} + + @Override + public void error(String msg, Throwable t) {} +} diff --git a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java index 23ada95fe..c96a7aed2 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java @@ -32,6 +32,7 @@ import reactor.core.Scannable; import reactor.core.publisher.Flux; import reactor.core.publisher.Operators; +import reactor.util.Logger; import reactor.util.annotation.Nullable; import reactor.util.concurrent.Queues; import reactor.util.context.Context; @@ -51,6 +52,7 @@ public final class UnboundedProcessor extends Flux final Queue queue; final Queue priorityQueue; final Runnable onFinalizedHook; + @Nullable final Logger logger; boolean cancelled; boolean done; @@ -99,10 +101,19 @@ public UnboundedProcessor() { this(() -> {}); } + UnboundedProcessor(Logger logger) { + this(() -> {}, logger); + } + public UnboundedProcessor(Runnable onFinalizedHook) { + this(onFinalizedHook, null); + } + + UnboundedProcessor(Runnable onFinalizedHook, @Nullable Logger logger) { this.onFinalizedHook = onFinalizedHook; this.queue = new MpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE); this.priorityQueue = new MpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE); + this.logger = logger; } @Override @@ -153,7 +164,7 @@ public boolean tryEmitPrioritized(ByteBuf t) { } if (hasRequest(previousState)) { - drainRegular(previousState); + drainRegular((previousState | FLAG_HAS_VALUE) + 1); } } return true; @@ -189,7 +200,7 @@ public boolean tryEmitNormal(ByteBuf t) { } if (hasRequest(previousState)) { - drainRegular(previousState); + drainRegular((previousState | FLAG_HAS_VALUE) + 1); } } @@ -223,9 +234,7 @@ public boolean tryEmitFinal(ByteBuf t) { return true; } - if (hasRequest(previousState)) { - drainRegular(previousState); - } + drainRegular((previousState | FLAG_TERMINATED | FLAG_HAS_VALUE) + 1); } return true; @@ -279,9 +288,7 @@ public void onError(Throwable t) { return; } - if (hasRequest(previousState)) { - drainRegular(previousState); - } + drainRegular((previousState | FLAG_TERMINATED) + 1); } } @@ -318,18 +325,15 @@ public void onComplete() { return; } - if (hasRequest(previousState)) { - drainRegular(previousState); - } + drainRegular((previousState | FLAG_TERMINATED) + 1); } } - void drainRegular(long previousState) { + void drainRegular(long expectedState) { final CoreSubscriber a = this.actual; final Queue q = this.queue; final Queue pq = this.priorityQueue; - long expectedState = previousState + 1; for (; ; ) { long r = this.requested; @@ -351,7 +355,7 @@ void drainRegular(long previousState) { empty = t == null; } - if (checkTerminated(done, empty, a)) { + if (checkTerminated(done, empty, true, a)) { if (!empty) { release(t); } @@ -374,7 +378,7 @@ void drainRegular(long previousState) { done = this.done; empty = q.isEmpty() && pq.isEmpty(); - if (checkTerminated(done, empty, a)) { + if (checkTerminated(done, empty, false, a)) { return; } } @@ -401,7 +405,8 @@ void drainRegular(long previousState) { } } - boolean checkTerminated(boolean done, boolean empty, CoreSubscriber a) { + boolean checkTerminated( + boolean done, boolean empty, boolean hasDemand, CoreSubscriber a) { final long state = this.state; if (isCancelled(state)) { clearAndFinalize(this); @@ -415,8 +420,15 @@ boolean checkTerminated(boolean done, boolean empty, CoreSubscriber actual) { previousState = markSubscriberReady(this); + if (isSubscriberReady(previousState)) { + return; + } + if (this.outputFused) { if (isCancelled(previousState)) { return; @@ -523,7 +539,7 @@ public void subscribe(CoreSubscriber actual) { } if (hasRequest(previousState)) { - drainRegular(previousState); + drainRegular((previousState | FLAG_SUBSCRIBER_READY) + 1); } } @@ -549,7 +565,7 @@ public void request(long n) { } if (isSubscriberReady(previousState) && hasValue(previousState)) { - drainRegular(previousState); + drainRegular((previousState | FLAG_HAS_REQUEST) + 1); } } } @@ -727,7 +743,9 @@ static long markSubscribedOnce(UnboundedProcessor instance) { return state; } - if (STATE.compareAndSet(instance, state, state | FLAG_SUBSCRIBED_ONCE)) { + final long nextState = state | FLAG_SUBSCRIBED_ONCE; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mso", state, nextState); return state; } } @@ -743,7 +761,10 @@ static long markSubscriberReady(UnboundedProcessor instance) { for (; ; ) { long state = instance.state; - if (isFinalized(state) || isCancelled(state) || isDisposed(state)) { + if (isFinalized(state) + || isCancelled(state) + || isDisposed(state) + || isSubscriberReady(state)) { return state; } @@ -754,7 +775,9 @@ static long markSubscriberReady(UnboundedProcessor instance) { } } - if (STATE.compareAndSet(instance, state, nextState | FLAG_SUBSCRIBER_READY)) { + nextState = nextState | FLAG_SUBSCRIBER_READY; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " msr", state, nextState); return state; } } @@ -776,11 +799,13 @@ static long markRequestAdded(UnboundedProcessor instance) { } long nextState = state; - if (isSubscriberReady(state) && hasValue(state)) { + if (isWorkInProgress(state) || (isSubscriberReady(state) && hasValue(state))) { nextState = addWork(state); } - if (STATE.compareAndSet(instance, state, nextState | FLAG_HAS_REQUEST)) { + nextState = nextState | FLAG_HAS_REQUEST; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mra", state, nextState); return state; } } @@ -815,7 +840,9 @@ static long markValueAdded(UnboundedProcessor instance) { } } - if (STATE.compareAndSet(instance, state, nextState | FLAG_HAS_VALUE)) { + nextState = nextState | FLAG_HAS_VALUE; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mva", state, nextState); return state; } } @@ -840,12 +867,12 @@ static long markValueAddedAndTerminated(UnboundedProcessor instance) { if (isWorkInProgress(state)) { nextState = addWork(state); } else if (isSubscriberReady(state) && !instance.outputFused) { - if (hasRequest(state)) { - nextState = addWork(state); - } + nextState = addWork(state); } - if (STATE.compareAndSet(instance, state, nextState | FLAG_HAS_VALUE | FLAG_TERMINATED)) { + nextState = nextState | FLAG_HAS_VALUE | FLAG_TERMINATED; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, "mva&t", state, nextState); return state; } } @@ -867,16 +894,20 @@ static long markTerminatedOrFinalized(UnboundedProcessor instance) { } long nextState = state; - if (isSubscriberReady(state) && !instance.outputFused) { + if (isWorkInProgress(state)) { + nextState = addWork(state); + } else if (isSubscriberReady(state) && !instance.outputFused) { if (!hasValue(state)) { // fast path for no values and no work in progress nextState = FLAG_FINALIZED; - } else if (hasRequest(state)) { + } else { nextState = addWork(state); } } - if (STATE.compareAndSet(instance, state, nextState | FLAG_TERMINATED)) { + nextState = nextState | FLAG_TERMINATED; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mt|f", state, nextState); if (isFinalized(nextState)) { instance.onFinalizedHook.run(); } @@ -899,8 +930,9 @@ static long markCancelled(UnboundedProcessor instance) { return state; } - final long nextState = addWork(state); - if (STATE.compareAndSet(instance, state, nextState | FLAG_CANCELLED)) { + final long nextState = addWork(state) | FLAG_CANCELLED; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mc", state, nextState); return state; } } @@ -921,8 +953,9 @@ static long markDisposed(UnboundedProcessor instance) { return state; } - final long nextState = addWork(state); - if (STATE.compareAndSet(instance, state, nextState | FLAG_DISPOSED)) { + final long nextState = addWork(state) | FLAG_DISPOSED; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " md", state, nextState); return state; } } @@ -945,12 +978,10 @@ static long addWork(long state) { */ static long markWorkDone( UnboundedProcessor instance, long expectedState, boolean hasRequest, boolean hasValue) { - final long expectedMissed = expectedState & MAX_WIP_VALUE; for (; ; ) { final long state = instance.state; - final long missed = state & MAX_WIP_VALUE; - if (missed != expectedMissed) { + if (state != expectedState) { return state; } @@ -958,11 +989,12 @@ static long markWorkDone( return state; } - final long nextState = state - expectedMissed; - if (STATE.compareAndSet( - instance, - state, - nextState ^ (hasRequest ? 0 : FLAG_HAS_REQUEST) ^ (hasValue ? 0 : FLAG_HAS_VALUE))) { + final long nextState = + (state - (expectedState & MAX_WIP_VALUE)) + ^ (hasRequest ? 0 : FLAG_HAS_REQUEST) + ^ (hasValue ? 0 : FLAG_HAS_VALUE); + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " mwd", state, nextState); return nextState; } } @@ -991,8 +1023,9 @@ static void clearAndFinalize(UnboundedProcessor instance) { instance.clearUnsafely(); } - if (STATE.compareAndSet( - instance, state, (state & ~MAX_WIP_VALUE & ~FLAG_HAS_VALUE) | FLAG_FINALIZED)) { + long nextState = (state & ~MAX_WIP_VALUE & ~FLAG_HAS_VALUE) | FLAG_FINALIZED; + if (STATE.compareAndSet(instance, state, nextState)) { + log(instance, " c&f", state, nextState); instance.onFinalizedHook.run(); break; } @@ -1034,4 +1067,101 @@ static boolean isSubscriberReady(long state) { static boolean isSubscribedOnce(long state) { return (state & FLAG_SUBSCRIBED_ONCE) == FLAG_SUBSCRIBED_ONCE; } + + static void log( + UnboundedProcessor instance, String action, long initialState, long committedState) { + log(instance, action, initialState, committedState, false); + } + + static void log( + UnboundedProcessor instance, + String action, + long initialState, + long committedState, + boolean logStackTrace) { + Logger logger = instance.logger; + if (logger == null || !logger.isTraceEnabled()) { + return; + } + + if (logStackTrace) { + logger.trace( + String.format( + "[%s][%s][%s][%s-%s]", + instance, + action, + action, + Thread.currentThread().getId(), + formatState(initialState, 64), + formatState(committedState, 64)), + new RuntimeException()); + } else { + logger.trace( + String.format( + "[%s][%s][%s][%s-%s]", + instance, + action, + Thread.currentThread().getId(), + formatState(initialState, 64), + formatState(committedState, 64))); + } + } + + static void log( + UnboundedProcessor instance, String action, int initialState, int committedState) { + log(instance, action, initialState, committedState, false); + } + + static void log( + UnboundedProcessor instance, + String action, + int initialState, + int committedState, + boolean logStackTrace) { + Logger logger = instance.logger; + if (logger == null || !logger.isTraceEnabled()) { + return; + } + + if (logStackTrace) { + logger.trace( + String.format( + "[%s][%s][%s][%s-%s]", + instance, + action, + action, + Thread.currentThread().getId(), + formatState(initialState, 32), + formatState(committedState, 32)), + new RuntimeException()); + } else { + logger.trace( + String.format( + "[%s][%s][%s][%s-%s]", + instance, + action, + Thread.currentThread().getId(), + formatState(initialState, 32), + formatState(committedState, 32))); + } + } + + static String formatState(long state, int size) { + final String defaultFormat = Long.toBinaryString(state); + final StringBuilder formatted = new StringBuilder(); + final int toPrepend = size - defaultFormat.length(); + for (int i = 0; i < size; i++) { + if (i != 0 && i % 4 == 0) { + formatted.append("_"); + } + if (i < toPrepend) { + formatted.append("0"); + } else { + formatted.append(defaultFormat.charAt(i - toPrepend)); + } + } + + formatted.insert(0, "0b"); + return formatted.toString(); + } } From 7abe35ee592843de7cf11f4675615765c690d8a4 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 24 Jan 2025 21:31:24 +0000 Subject: [PATCH 04/13] Fix handling of rejected setup errors (#1117) Closes gh-1092 --- .../java/io/rsocket/core/RSocketServer.java | 12 +++++-- .../rsocket/resume/ServerRSocketSession.java | 3 ++ .../io/rsocket/core/RSocketServerTest.java | 36 ++++++++++++++++++- 3 files changed, 47 insertions(+), 4 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java index 0c68db6df..e969c39d2 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2025 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. @@ -463,8 +463,14 @@ private Mono acceptSetup( return interceptors .initSocketAcceptor(acceptor) .accept(setupPayload, wrappedRSocketRequester) - .doOnError( - err -> serverSetup.sendError(wrappedDuplexConnection, rejectedSetupError(err))) + .onErrorResume( + err -> + Mono.fromRunnable( + () -> + serverSetup.sendError( + wrappedDuplexConnection, rejectedSetupError(err))) + .then(wrappedDuplexConnection.onClose()) + .then(Mono.error(err))) .doOnNext( rSocketHandler -> { RSocket wrappedRSocketHandler = interceptors.initResponder(rSocketHandler); diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java b/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java index c4dc4d837..ad1b38375 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java @@ -287,6 +287,9 @@ public void setKeepAliveSupport(KeepAliveSupport keepAliveSupport) { @Override public void dispose() { + if (logger.isDebugEnabled()) { + logger.debug("Side[server]|Session[{}]. Disposing session", session); + } Operators.terminate(S, this); resumableConnection.dispose(); } diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java index 0b5ca38f7..a335ac1f3 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2025 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. @@ -23,11 +23,14 @@ import io.rsocket.Closeable; import io.rsocket.FrameAssert; import io.rsocket.RSocket; +import io.rsocket.exceptions.RejectedSetupException; import io.rsocket.frame.FrameType; import io.rsocket.frame.KeepAliveFrameCodec; import io.rsocket.frame.RequestResponseFrameCodec; +import io.rsocket.frame.SetupFrameCodec; import io.rsocket.test.util.TestDuplexConnection; import io.rsocket.test.util.TestServerTransport; +import io.rsocket.util.EmptyPayload; import java.time.Duration; import java.util.Random; import org.assertj.core.api.Assertions; @@ -164,4 +167,35 @@ public void unexpectedFramesBeforeSetup() { server.dispose(); transport.alloc().assertHasNoLeaks(); } + + @Test + public void ensuresErrorFrameDeliveredPriorConnectionDisposal() { + TestServerTransport transport = new TestServerTransport(); + Closeable server = + RSocketServer.create() + .acceptor( + (setup, sendingSocket) -> Mono.error(new RejectedSetupException("ACCESS_DENIED"))) + .bind(transport) + .block(); + + TestDuplexConnection connection = transport.connect(); + connection.addToReceivedBuffer( + SetupFrameCodec.encode( + ByteBufAllocator.DEFAULT, + false, + 0, + 1, + Unpooled.EMPTY_BUFFER, + "metadata_type", + "data_type", + EmptyPayload.INSTANCE)); + + StepVerifier.create(connection.onClose()).expectComplete().verify(Duration.ofSeconds(30)); + FrameAssert.assertThat(connection.pollFrame()) + .hasStreamIdZero() + .hasData("ACCESS_DENIED") + .hasNoLeaks(); + server.dispose(); + transport.alloc().assertHasNoLeaks(); + } } From b1dd3c54148daecc50695b6b2687e578fb641c0d Mon Sep 17 00:00:00 2001 From: sullis Date: Fri, 24 Jan 2025 13:31:58 -0800 Subject: [PATCH 05/13] support netty boringssl aarch_64 classifier (#1107) Signed-off-by: sullis --- rsocket-transport-netty/build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rsocket-transport-netty/build.gradle b/rsocket-transport-netty/build.gradle index 17756dbc6..39a5ceac5 100644 --- a/rsocket-transport-netty/build.gradle +++ b/rsocket-transport-netty/build.gradle @@ -22,7 +22,7 @@ plugins { } def os_suffix = "" -if (osdetector.classifier in ["linux-x86_64", "osx-x86_64", "windows-x86_64"]) { +if (osdetector.classifier in ["linux-x86_64", "linux-aarch_64", "osx-x86_64", "osx-aarch_64", "windows-x86_64"]) { os_suffix = "::" + osdetector.classifier } From 50c51c43c5963258af3fbc7562fb9bf2083782f2 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 24 Jan 2025 21:31:24 +0000 Subject: [PATCH 06/13] Fix handling of rejected setup errors (#1117) Closes gh-1092 --- .../java/io/rsocket/core/RSocketServer.java | 12 +++++-- .../rsocket/resume/ServerRSocketSession.java | 3 ++ .../io/rsocket/core/RSocketServerTest.java | 36 ++++++++++++++++++- 3 files changed, 47 insertions(+), 4 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java index 0c68db6df..e969c39d2 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2025 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. @@ -463,8 +463,14 @@ private Mono acceptSetup( return interceptors .initSocketAcceptor(acceptor) .accept(setupPayload, wrappedRSocketRequester) - .doOnError( - err -> serverSetup.sendError(wrappedDuplexConnection, rejectedSetupError(err))) + .onErrorResume( + err -> + Mono.fromRunnable( + () -> + serverSetup.sendError( + wrappedDuplexConnection, rejectedSetupError(err))) + .then(wrappedDuplexConnection.onClose()) + .then(Mono.error(err))) .doOnNext( rSocketHandler -> { RSocket wrappedRSocketHandler = interceptors.initResponder(rSocketHandler); diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java b/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java index c4dc4d837..ad1b38375 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java @@ -287,6 +287,9 @@ public void setKeepAliveSupport(KeepAliveSupport keepAliveSupport) { @Override public void dispose() { + if (logger.isDebugEnabled()) { + logger.debug("Side[server]|Session[{}]. Disposing session", session); + } Operators.terminate(S, this); resumableConnection.dispose(); } diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java index 0b5ca38f7..a335ac1f3 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2025 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. @@ -23,11 +23,14 @@ import io.rsocket.Closeable; import io.rsocket.FrameAssert; import io.rsocket.RSocket; +import io.rsocket.exceptions.RejectedSetupException; import io.rsocket.frame.FrameType; import io.rsocket.frame.KeepAliveFrameCodec; import io.rsocket.frame.RequestResponseFrameCodec; +import io.rsocket.frame.SetupFrameCodec; import io.rsocket.test.util.TestDuplexConnection; import io.rsocket.test.util.TestServerTransport; +import io.rsocket.util.EmptyPayload; import java.time.Duration; import java.util.Random; import org.assertj.core.api.Assertions; @@ -164,4 +167,35 @@ public void unexpectedFramesBeforeSetup() { server.dispose(); transport.alloc().assertHasNoLeaks(); } + + @Test + public void ensuresErrorFrameDeliveredPriorConnectionDisposal() { + TestServerTransport transport = new TestServerTransport(); + Closeable server = + RSocketServer.create() + .acceptor( + (setup, sendingSocket) -> Mono.error(new RejectedSetupException("ACCESS_DENIED"))) + .bind(transport) + .block(); + + TestDuplexConnection connection = transport.connect(); + connection.addToReceivedBuffer( + SetupFrameCodec.encode( + ByteBufAllocator.DEFAULT, + false, + 0, + 1, + Unpooled.EMPTY_BUFFER, + "metadata_type", + "data_type", + EmptyPayload.INSTANCE)); + + StepVerifier.create(connection.onClose()).expectComplete().verify(Duration.ofSeconds(30)); + FrameAssert.assertThat(connection.pollFrame()) + .hasStreamIdZero() + .hasData("ACCESS_DENIED") + .hasNoLeaks(); + server.dispose(); + transport.alloc().assertHasNoLeaks(); + } } From a5fbd96a815b1ef07333df61256d4756227a3642 Mon Sep 17 00:00:00 2001 From: sullis Date: Fri, 24 Jan 2025 13:31:58 -0800 Subject: [PATCH 07/13] support netty boringssl aarch_64 classifier (#1107) Signed-off-by: sullis --- rsocket-transport-netty/build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rsocket-transport-netty/build.gradle b/rsocket-transport-netty/build.gradle index 17756dbc6..39a5ceac5 100644 --- a/rsocket-transport-netty/build.gradle +++ b/rsocket-transport-netty/build.gradle @@ -22,7 +22,7 @@ plugins { } def os_suffix = "" -if (osdetector.classifier in ["linux-x86_64", "osx-x86_64", "windows-x86_64"]) { +if (osdetector.classifier in ["linux-x86_64", "linux-aarch_64", "osx-x86_64", "osx-aarch_64", "windows-x86_64"]) { os_suffix = "::" + osdetector.classifier } From 9bc30c44d601fa1e5d937901df979e68a0227fc1 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Tue, 28 Jan 2025 11:29:28 +0000 Subject: [PATCH 08/13] Update Logback version to 1.2.13 It must stay < 1.3 to work with slf4j 1.7. Signed-off-by: rstoyanchev --- build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index ae8f05d06..ccf03c4eb 100644 --- a/build.gradle +++ b/build.gradle @@ -34,7 +34,7 @@ subprojects { apply plugin: 'com.github.vlsi.gradle-extensions' ext['reactor-bom.version'] = '2020.0.39' - ext['logback.version'] = '1.3.14' + ext['logback.version'] = '1.2.13' ext['netty-bom.version'] = '4.1.106.Final' ext['netty-boringssl.version'] = '2.0.62.Final' ext['hdrhistogram.version'] = '2.1.12' From ccd67ba20d8a0c242901a180c8369a6315a6b626 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Fri, 31 Jan 2025 15:41:38 +0200 Subject: [PATCH 09/13] ensures connection is closed on keepalive timeout (#1118) * ensures connection is close on keepalive timeout Signed-off-by: Oleh Dokuka * fix format Signed-off-by: Oleh Dokuka * improve KeepaliveTest Signed-off-by: Oleh Dokuka * fix format and failing test Signed-off-by: Oleh Dokuka * adds reference to the original GH issue Signed-off-by: Oleh Dokuka * fixes google format Signed-off-by: Oleh Dokuka --------- Signed-off-by: Oleh Dokuka --- .../io/rsocket/core/RSocketRequester.java | 1 + .../io/rsocket/integration/KeepaliveTest.java | 190 ++++++++++++++++++ .../src/test/resources/logback-test.xml | 1 - 3 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java index 9e8d349bf..b8a9c00ff 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java @@ -312,6 +312,7 @@ private void tryTerminateOnKeepAlive(KeepAliveSupport.KeepAlive keepAlive) { () -> new ConnectionErrorException( String.format("No keep-alive acks for %d ms", keepAlive.getTimeout().toMillis()))); + getDuplexConnection().dispose(); } private void tryShutdown(Throwable e) { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java new file mode 100644 index 000000000..f05713215 --- /dev/null +++ b/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java @@ -0,0 +1,190 @@ +package io.rsocket.integration; + +import io.rsocket.Payload; +import io.rsocket.RSocket; +import io.rsocket.core.RSocketClient; +import io.rsocket.core.RSocketConnector; +import io.rsocket.core.RSocketServer; +import io.rsocket.frame.decoder.PayloadDecoder; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.server.CloseableChannel; +import io.rsocket.transport.netty.server.TcpServerTransport; +import io.rsocket.util.DefaultPayload; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.netty.tcp.TcpClient; +import reactor.netty.tcp.TcpServer; +import reactor.test.StepVerifier; +import reactor.util.retry.Retry; +import reactor.util.retry.RetryBackoffSpec; + +/** + * Test case that reproduces the following GitHub Issue + */ +public class KeepaliveTest { + + private static final Logger LOG = LoggerFactory.getLogger(KeepaliveTest.class); + private static final int PORT = 23200; + + private CloseableChannel server; + + @BeforeEach + void setUp() { + server = createServer().block(); + } + + @AfterEach + void tearDown() { + server.dispose(); + server.onClose().block(); + } + + @Test + void keepAliveTest() { + RSocketClient rsocketClient = createClient(); + + int expectedCount = 4; + AtomicBoolean sleepOnce = new AtomicBoolean(true); + StepVerifier.create( + Flux.range(0, expectedCount) + .delayElements(Duration.ofMillis(2000)) + .concatMap( + i -> + rsocketClient + .requestResponse(Mono.just(DefaultPayload.create(""))) + .doOnNext( + __ -> { + if (sleepOnce.getAndSet(false)) { + try { + LOG.info("Sleeping..."); + Thread.sleep(1_000); + LOG.info("Waking up."); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }) + .log("id " + i) + .onErrorComplete())) + .expectSubscription() + .expectNextCount(expectedCount) + .verifyComplete(); + } + + @Test + void keepAliveTestLazy() { + Mono rsocketMono = createClientLazy(); + + int expectedCount = 4; + AtomicBoolean sleepOnce = new AtomicBoolean(true); + StepVerifier.create( + Flux.range(0, expectedCount) + .delayElements(Duration.ofMillis(2000)) + .concatMap( + i -> + rsocketMono.flatMap( + rsocket -> + rsocket + .requestResponse(DefaultPayload.create("")) + .doOnNext( + __ -> { + if (sleepOnce.getAndSet(false)) { + try { + LOG.info("Sleeping..."); + Thread.sleep(1_000); + LOG.info("Waking up."); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }) + .log("id " + i) + .onErrorComplete()))) + .expectSubscription() + .expectNextCount(expectedCount) + .verifyComplete(); + } + + private static Mono createServer() { + LOG.info("Starting server at port {}", PORT); + + TcpServer tcpServer = TcpServer.create().host("localhost").port(PORT); + + return RSocketServer.create( + (setupPayload, rSocket) -> { + rSocket + .onClose() + .doFirst(() -> LOG.info("Connected on server side.")) + .doOnTerminate(() -> LOG.info("Connection closed on server side.")) + .subscribe(); + + return Mono.just(new MyServerRsocket()); + }) + .payloadDecoder(PayloadDecoder.ZERO_COPY) + .bind(TcpServerTransport.create(tcpServer)) + .doOnNext(closeableChannel -> LOG.info("RSocket server started.")); + } + + private static RSocketClient createClient() { + LOG.info("Connecting...."); + + Function reconnectSpec = + reason -> + Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10L)) + .doBeforeRetry(retrySignal -> LOG.info("Reconnecting. Reason: {}", reason)); + + Mono rsocketMono = + RSocketConnector.create() + .fragment(16384) + .reconnect(reconnectSpec.apply("connector-close")) + .keepAlive(Duration.ofMillis(100L), Duration.ofMillis(900L)) + .connect(TcpClientTransport.create(TcpClient.create().host("localhost").port(PORT))); + + RSocketClient client = RSocketClient.from(rsocketMono); + + client + .source() + .doOnNext(r -> LOG.info("Got RSocket")) + .flatMap(RSocket::onClose) + .doOnError(err -> LOG.error("Error during onClose.", err)) + .retryWhen(reconnectSpec.apply("client-close")) + .doFirst(() -> LOG.info("Connected on client side.")) + .doOnTerminate(() -> LOG.info("Connection closed on client side.")) + .repeat() + .subscribe(); + + return client; + } + + private static Mono createClientLazy() { + LOG.info("Connecting...."); + + Function reconnectSpec = + reason -> + Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10L)) + .doBeforeRetry(retrySignal -> LOG.info("Reconnecting. Reason: {}", reason)); + + return RSocketConnector.create() + .fragment(16384) + .reconnect(reconnectSpec.apply("connector-close")) + .keepAlive(Duration.ofMillis(100L), Duration.ofMillis(900L)) + .connect(TcpClientTransport.create(TcpClient.create().host("localhost").port(PORT))); + } + + public static class MyServerRsocket implements RSocket { + + @Override + public Mono requestResponse(Payload payload) { + return Mono.just("Pong").map(DefaultPayload::create); + } + } +} diff --git a/rsocket-transport-netty/src/test/resources/logback-test.xml b/rsocket-transport-netty/src/test/resources/logback-test.xml index b42db6df6..981d6d0b6 100644 --- a/rsocket-transport-netty/src/test/resources/logback-test.xml +++ b/rsocket-transport-netty/src/test/resources/logback-test.xml @@ -27,7 +27,6 @@ - From cff5cdbb16da6393efc04d8f0b80793e54f79026 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Fri, 31 Jan 2025 11:53:11 +0000 Subject: [PATCH 10/13] Log data in KEEPALIVE frame Fixes gh-1114 Signed-off-by: rstoyanchev --- .../main/java/io/rsocket/frame/FrameUtil.java | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java b/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java index 66d18c8a7..d581731a3 100644 --- a/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java +++ b/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java @@ -1,3 +1,18 @@ +/* + * Copyright 2015-2025 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 io.rsocket.frame; import io.netty.buffer.ByteBuf; @@ -99,8 +114,9 @@ private static ByteBuf getData(ByteBuf frame, FrameType frameType) { case REQUEST_CHANNEL: data = RequestChannelFrameCodec.data(frame); break; - // Payload and synthetic types + // Payload, KeepAlive and synthetic types case PAYLOAD: + case KEEPALIVE: case NEXT: case NEXT_COMPLETE: case COMPLETE: From 838e8fbfa2c45d72fa323c7e82e53490cc0d3d8b Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Fri, 31 Jan 2025 15:41:38 +0200 Subject: [PATCH 11/13] ensures connection is closed on keepalive timeout (#1118) * ensures connection is close on keepalive timeout Signed-off-by: Oleh Dokuka * fix format Signed-off-by: Oleh Dokuka * improve KeepaliveTest Signed-off-by: Oleh Dokuka * fix format and failing test Signed-off-by: Oleh Dokuka * adds reference to the original GH issue Signed-off-by: Oleh Dokuka * fixes google format Signed-off-by: Oleh Dokuka --------- Signed-off-by: Oleh Dokuka --- .../io/rsocket/core/RSocketRequester.java | 1 + .../io/rsocket/integration/KeepaliveTest.java | 190 ++++++++++++++++++ .../src/test/resources/logback-test.xml | 1 - 3 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java index 9e8d349bf..b8a9c00ff 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java @@ -312,6 +312,7 @@ private void tryTerminateOnKeepAlive(KeepAliveSupport.KeepAlive keepAlive) { () -> new ConnectionErrorException( String.format("No keep-alive acks for %d ms", keepAlive.getTimeout().toMillis()))); + getDuplexConnection().dispose(); } private void tryShutdown(Throwable e) { diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java new file mode 100644 index 000000000..f05713215 --- /dev/null +++ b/rsocket-transport-netty/src/test/java/io/rsocket/integration/KeepaliveTest.java @@ -0,0 +1,190 @@ +package io.rsocket.integration; + +import io.rsocket.Payload; +import io.rsocket.RSocket; +import io.rsocket.core.RSocketClient; +import io.rsocket.core.RSocketConnector; +import io.rsocket.core.RSocketServer; +import io.rsocket.frame.decoder.PayloadDecoder; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.server.CloseableChannel; +import io.rsocket.transport.netty.server.TcpServerTransport; +import io.rsocket.util.DefaultPayload; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.netty.tcp.TcpClient; +import reactor.netty.tcp.TcpServer; +import reactor.test.StepVerifier; +import reactor.util.retry.Retry; +import reactor.util.retry.RetryBackoffSpec; + +/** + * Test case that reproduces the following GitHub Issue + */ +public class KeepaliveTest { + + private static final Logger LOG = LoggerFactory.getLogger(KeepaliveTest.class); + private static final int PORT = 23200; + + private CloseableChannel server; + + @BeforeEach + void setUp() { + server = createServer().block(); + } + + @AfterEach + void tearDown() { + server.dispose(); + server.onClose().block(); + } + + @Test + void keepAliveTest() { + RSocketClient rsocketClient = createClient(); + + int expectedCount = 4; + AtomicBoolean sleepOnce = new AtomicBoolean(true); + StepVerifier.create( + Flux.range(0, expectedCount) + .delayElements(Duration.ofMillis(2000)) + .concatMap( + i -> + rsocketClient + .requestResponse(Mono.just(DefaultPayload.create(""))) + .doOnNext( + __ -> { + if (sleepOnce.getAndSet(false)) { + try { + LOG.info("Sleeping..."); + Thread.sleep(1_000); + LOG.info("Waking up."); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }) + .log("id " + i) + .onErrorComplete())) + .expectSubscription() + .expectNextCount(expectedCount) + .verifyComplete(); + } + + @Test + void keepAliveTestLazy() { + Mono rsocketMono = createClientLazy(); + + int expectedCount = 4; + AtomicBoolean sleepOnce = new AtomicBoolean(true); + StepVerifier.create( + Flux.range(0, expectedCount) + .delayElements(Duration.ofMillis(2000)) + .concatMap( + i -> + rsocketMono.flatMap( + rsocket -> + rsocket + .requestResponse(DefaultPayload.create("")) + .doOnNext( + __ -> { + if (sleepOnce.getAndSet(false)) { + try { + LOG.info("Sleeping..."); + Thread.sleep(1_000); + LOG.info("Waking up."); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }) + .log("id " + i) + .onErrorComplete()))) + .expectSubscription() + .expectNextCount(expectedCount) + .verifyComplete(); + } + + private static Mono createServer() { + LOG.info("Starting server at port {}", PORT); + + TcpServer tcpServer = TcpServer.create().host("localhost").port(PORT); + + return RSocketServer.create( + (setupPayload, rSocket) -> { + rSocket + .onClose() + .doFirst(() -> LOG.info("Connected on server side.")) + .doOnTerminate(() -> LOG.info("Connection closed on server side.")) + .subscribe(); + + return Mono.just(new MyServerRsocket()); + }) + .payloadDecoder(PayloadDecoder.ZERO_COPY) + .bind(TcpServerTransport.create(tcpServer)) + .doOnNext(closeableChannel -> LOG.info("RSocket server started.")); + } + + private static RSocketClient createClient() { + LOG.info("Connecting...."); + + Function reconnectSpec = + reason -> + Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10L)) + .doBeforeRetry(retrySignal -> LOG.info("Reconnecting. Reason: {}", reason)); + + Mono rsocketMono = + RSocketConnector.create() + .fragment(16384) + .reconnect(reconnectSpec.apply("connector-close")) + .keepAlive(Duration.ofMillis(100L), Duration.ofMillis(900L)) + .connect(TcpClientTransport.create(TcpClient.create().host("localhost").port(PORT))); + + RSocketClient client = RSocketClient.from(rsocketMono); + + client + .source() + .doOnNext(r -> LOG.info("Got RSocket")) + .flatMap(RSocket::onClose) + .doOnError(err -> LOG.error("Error during onClose.", err)) + .retryWhen(reconnectSpec.apply("client-close")) + .doFirst(() -> LOG.info("Connected on client side.")) + .doOnTerminate(() -> LOG.info("Connection closed on client side.")) + .repeat() + .subscribe(); + + return client; + } + + private static Mono createClientLazy() { + LOG.info("Connecting...."); + + Function reconnectSpec = + reason -> + Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(10L)) + .doBeforeRetry(retrySignal -> LOG.info("Reconnecting. Reason: {}", reason)); + + return RSocketConnector.create() + .fragment(16384) + .reconnect(reconnectSpec.apply("connector-close")) + .keepAlive(Duration.ofMillis(100L), Duration.ofMillis(900L)) + .connect(TcpClientTransport.create(TcpClient.create().host("localhost").port(PORT))); + } + + public static class MyServerRsocket implements RSocket { + + @Override + public Mono requestResponse(Payload payload) { + return Mono.just("Pong").map(DefaultPayload::create); + } + } +} diff --git a/rsocket-transport-netty/src/test/resources/logback-test.xml b/rsocket-transport-netty/src/test/resources/logback-test.xml index b42db6df6..981d6d0b6 100644 --- a/rsocket-transport-netty/src/test/resources/logback-test.xml +++ b/rsocket-transport-netty/src/test/resources/logback-test.xml @@ -27,7 +27,6 @@ - From d28d093aaef21bca01d8de12c2fb9be8492ed982 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Fri, 31 Jan 2025 11:53:11 +0000 Subject: [PATCH 12/13] Log data in KEEPALIVE frame Fixes gh-1114 Signed-off-by: rstoyanchev --- .../main/java/io/rsocket/frame/FrameUtil.java | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java b/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java index 66d18c8a7..d581731a3 100644 --- a/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java +++ b/rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java @@ -1,3 +1,18 @@ +/* + * Copyright 2015-2025 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 io.rsocket.frame; import io.netty.buffer.ByteBuf; @@ -99,8 +114,9 @@ private static ByteBuf getData(ByteBuf frame, FrameType frameType) { case REQUEST_CHANNEL: data = RequestChannelFrameCodec.data(frame); break; - // Payload and synthetic types + // Payload, KeepAlive and synthetic types case PAYLOAD: + case KEEPALIVE: case NEXT: case NEXT_COMPLETE: case COMPLETE: From 6e725d6d94f15de87ad7d67c8cc8e3e422e72acd Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Fri, 31 Jan 2025 13:58:46 +0000 Subject: [PATCH 13/13] Dependency upgrades Reactor 2020.0.39 -> 2020.0.47 Netty 4.1.106.Final -> 4.1.117.Final netty-tcnative-boringssl-static 2.0.62.Final -> 2.0.69.Final Micrometer 1.11.0 -> 1.11.12 Micrometer Tracing 1.1.1 -> 1.1.13 Signed-off-by: rstoyanchev --- build.gradle | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/build.gradle b/build.gradle index ccf03c4eb..b2122e4a5 100644 --- a/build.gradle +++ b/build.gradle @@ -33,17 +33,17 @@ subprojects { apply plugin: 'com.github.sherter.google-java-format' apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.39' + ext['reactor-bom.version'] = '2020.0.47' ext['logback.version'] = '1.2.13' - ext['netty-bom.version'] = '4.1.106.Final' - ext['netty-boringssl.version'] = '2.0.62.Final' + ext['netty-bom.version'] = '4.1.117.Final' + ext['netty-boringssl.version'] = '2.0.69.Final' ext['hdrhistogram.version'] = '2.1.12' ext['mockito.version'] = '4.11.0' ext['slf4j.version'] = '1.7.36' ext['jmh.version'] = '1.36' ext['junit.version'] = '5.9.3' - ext['micrometer.version'] = '1.11.0' - ext['micrometer-tracing.version'] = '1.1.1' + ext['micrometer.version'] = '1.11.12' + ext['micrometer-tracing.version'] = '1.1.13' ext['assertj.version'] = '3.24.2' ext['netflix.limits.version'] = '0.3.6' ext['bouncycastle-bcpkix.version'] = '1.70'