Skip to content

improves ref-counting by eliminating redundant retain/release #902

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
Aug 3, 2020
Merged

improves ref-counting by eliminating redundant retain/release #902

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
11 changes: 7 additions & 4 deletions rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
import static io.rsocket.keepalive.KeepAliveSupport.ClientKeepAliveSupport;

import io.netty.buffer.ByteBuf;
import io.netty.util.ReferenceCountUtil;
import io.rsocket.DuplexConnection;
import io.rsocket.Payload;
import io.rsocket.RSocket;
Expand Down Expand Up @@ -205,10 +204,14 @@ private void handleIncomingFrames(ByteBuf frame) {
} else {
handleFrame(streamId, type, frame);
}
frame.release();
} catch (Throwable t) {
ReferenceCountUtil.safeRelease(frame);
throw reactor.core.Exceptions.propagate(t);
super.getSendProcessor()
.onNext(
ErrorFrameCodec.encode(
super.getAllocator(),
0,
new ConnectionErrorException("Unexpected error during frame handling", t)));
this.tryTerminateOnConnectionError(t);
}
}

Expand Down
13 changes: 8 additions & 5 deletions rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,11 @@
package io.rsocket.core;

import io.netty.buffer.ByteBuf;
import io.netty.util.ReferenceCountUtil;
import io.rsocket.DuplexConnection;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.ResponderRSocket;
import io.rsocket.exceptions.ConnectionErrorException;
import io.rsocket.frame.ErrorFrameCodec;
import io.rsocket.frame.FrameHeaderCodec;
import io.rsocket.frame.FrameType;
Expand All @@ -39,7 +39,6 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

Expand Down Expand Up @@ -314,10 +313,14 @@ private void handleFrame(ByteBuf frame) {
"ServerRSocket: Unexpected frame type: " + frameType)));
break;
}
ReferenceCountUtil.safeRelease(frame);
} catch (Throwable t) {
ReferenceCountUtil.safeRelease(frame);
throw Exceptions.propagate(t);
super.getSendProcessor()
.onNext(
ErrorFrameCodec.encode(
super.getAllocator(),
0,
new ConnectionErrorException("Unexpected error during frame handling", t)));
this.tryTerminateOnConnectionError(t);
}
}

Expand Down
15 changes: 4 additions & 11 deletions rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -387,29 +387,22 @@ private Mono<Void> acceptSetup(
multiplexer,
new InvalidSetupException(
"Unsupported version: " + SetupFrameCodec.humanReadableVersion(setupFrame)))
.doFinally(
signalType -> {
setupFrame.release();
multiplexer.dispose();
});
.doFinally(signalType -> multiplexer.dispose());
}

boolean leaseEnabled = leasesSupplier != null;
if (SetupFrameCodec.honorLease(setupFrame) && !leaseEnabled) {
return serverSetup
.sendError(multiplexer, new InvalidSetupException("lease is not supported"))
.doFinally(
signalType -> {
setupFrame.release();
multiplexer.dispose();
});
.doFinally(signalType -> multiplexer.dispose());
}

return serverSetup.acceptRSocketSetup(
setupFrame,
multiplexer,
(keepAliveHandler, wrappedMultiplexer) -> {
ConnectionSetupPayload setupPayload = new DefaultConnectionSetupPayload(setupFrame);
ConnectionSetupPayload setupPayload =
new DefaultConnectionSetupPayload(setupFrame.retain());

Leases<?> leases = leaseEnabled ? leasesSupplier.get() : null;
RequesterLeaseHandler requesterLeaseHandler =
Expand Down
10 changes: 7 additions & 3 deletions rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterSubscribersTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ void setUp() {

@ParameterizedTest
@MethodSource("allInteractions")
void singleSubscriber(Function<RSocket, Publisher<?>> interaction) {
void singleSubscriber(Function<RSocket, Publisher<?>> interaction, FrameType requestType) {
Flux<?> response = Flux.from(interaction.apply(rSocketRequester));

AssertSubscriber assertSubscriberA = AssertSubscriber.create();
Expand All @@ -91,7 +91,9 @@ void singleSubscriber(Function<RSocket, Publisher<?>> interaction) {
response.subscribe(assertSubscriberA);
response.subscribe(assertSubscriberB);

connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), 1));
if (requestType != FrameType.REQUEST_FNF && requestType != FrameType.METADATA_PUSH) {
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), 1));
}

assertSubscriberA.assertTerminated();
assertSubscriberB.assertTerminated();
Expand All @@ -111,7 +113,9 @@ void singleSubscriberInCaseOfRacing(
RaceTestUtils.race(
() -> response.subscribe(assertSubscriberA), () -> response.subscribe(assertSubscriberB));

connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), i));
if (requestType != FrameType.REQUEST_FNF && requestType != FrameType.METADATA_PUSH) {
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), i));
}

assertSubscriberA.assertTerminated();
assertSubscriberB.assertTerminated();
Expand Down
40 changes: 20 additions & 20 deletions rsocket-core/src/test/java/io/rsocket/internal/ClientServerInputMultiplexerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,52 +68,52 @@ public void clientSplits() {
.doOnNext(f -> setupFrames.incrementAndGet())
.subscribe();

source.addToReceivedBuffer(errorFrame(1));
source.addToReceivedBuffer(errorFrame(1).retain());
assertEquals(1, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(1));
source.addToReceivedBuffer(errorFrame(1).retain());
assertEquals(2, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(leaseFrame());
source.addToReceivedBuffer(leaseFrame().retain());
assertEquals(3, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(keepAliveFrame());
source.addToReceivedBuffer(keepAliveFrame().retain());
assertEquals(4, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(2));
source.addToReceivedBuffer(errorFrame(2).retain());
assertEquals(4, clientFrames.get());
assertEquals(1, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(0));
source.addToReceivedBuffer(errorFrame(0).retain());
assertEquals(5, clientFrames.get());
assertEquals(1, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(metadataPushFrame());
source.addToReceivedBuffer(metadataPushFrame().retain());
assertEquals(5, clientFrames.get());
assertEquals(2, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(setupFrame());
source.addToReceivedBuffer(setupFrame().retain());
assertEquals(5, clientFrames.get());
assertEquals(2, serverFrames.get());
assertEquals(1, setupFrames.get());

source.addToReceivedBuffer(resumeFrame());
source.addToReceivedBuffer(resumeFrame().retain());
assertEquals(5, clientFrames.get());
assertEquals(2, serverFrames.get());
assertEquals(2, setupFrames.get());

source.addToReceivedBuffer(resumeOkFrame());
source.addToReceivedBuffer(resumeOkFrame().retain());
assertEquals(5, clientFrames.get());
assertEquals(2, serverFrames.get());
assertEquals(3, setupFrames.get());
Expand Down Expand Up @@ -141,52 +141,52 @@ public void serverSplits() {
.doOnNext(f -> setupFrames.incrementAndGet())
.subscribe();

source.addToReceivedBuffer(errorFrame(1));
source.addToReceivedBuffer(errorFrame(1).retain());
assertEquals(1, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(1));
source.addToReceivedBuffer(errorFrame(1).retain());
assertEquals(2, clientFrames.get());
assertEquals(0, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(leaseFrame());
source.addToReceivedBuffer(leaseFrame().retain());
assertEquals(2, clientFrames.get());
assertEquals(1, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(keepAliveFrame());
source.addToReceivedBuffer(keepAliveFrame().retain());
assertEquals(2, clientFrames.get());
assertEquals(2, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(2));
source.addToReceivedBuffer(errorFrame(2).retain());
assertEquals(2, clientFrames.get());
assertEquals(3, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(errorFrame(0));
source.addToReceivedBuffer(errorFrame(0).retain());
assertEquals(2, clientFrames.get());
assertEquals(4, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(metadataPushFrame());
source.addToReceivedBuffer(metadataPushFrame().retain());
assertEquals(3, clientFrames.get());
assertEquals(4, serverFrames.get());
assertEquals(0, setupFrames.get());

source.addToReceivedBuffer(setupFrame());
source.addToReceivedBuffer(setupFrame().retain());
assertEquals(3, clientFrames.get());
assertEquals(4, serverFrames.get());
assertEquals(1, setupFrames.get());

source.addToReceivedBuffer(resumeFrame());
source.addToReceivedBuffer(resumeFrame().retain());
assertEquals(3, clientFrames.get());
assertEquals(4, serverFrames.get());
assertEquals(2, setupFrames.get());

source.addToReceivedBuffer(resumeOkFrame());
source.addToReceivedBuffer(resumeOkFrame().retain());
assertEquals(3, clientFrames.get());
assertEquals(4, serverFrames.get());
assertEquals(3, setupFrames.get());
Expand Down
32 changes: 31 additions & 1 deletion rsocket-core/src/test/java/io/rsocket/test/util/LocalDuplexConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,13 @@
import io.netty.buffer.ByteBufAllocator;
import io.rsocket.DuplexConnection;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Operators;

public class LocalDuplexConnection implements DuplexConnection {
private final ByteBufAllocator allocator;
Expand Down Expand Up @@ -55,7 +58,34 @@ public Mono<Void> send(Publisher<ByteBuf> frame) {

@Override
public Flux<ByteBuf> receive() {
return receive.doOnNext(f -> System.out.println(name + " - " + f.toString()));
return receive
.doOnNext(f -> System.out.println(name + " - " + f.toString()))
.transform(
Operators.<ByteBuf, ByteBuf>lift(
(__, actual) ->
new CoreSubscriber<ByteBuf>() {

@Override
public void onSubscribe(Subscription s) {
actual.onSubscribe(s);
}

@Override
public void onNext(ByteBuf byteBuf) {
actual.onNext(byteBuf);
byteBuf.release();
}

@Override
public void onError(Throwable t) {
actual.onError(t);
}

@Override
public void onComplete() {
actual.onComplete();
}
}));
}

@Override
Expand Down
29 changes: 28 additions & 1 deletion rsocket-core/src/test/java/io/rsocket/test/util/TestDuplexConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,16 @@
import java.util.concurrent.LinkedBlockingQueue;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Operators;

/**
* An implementation of {@link DuplexConnection} that provides functionality to modify the behavior
Expand Down Expand Up @@ -83,7 +86,31 @@ public Mono<Void> send(Publisher<ByteBuf> frames) {

@Override
public Flux<ByteBuf> receive() {
return received;
return received.transform(
Operators.<ByteBuf, ByteBuf>lift(
(__, actual) ->
new CoreSubscriber<ByteBuf>() {
@Override
public void onSubscribe(Subscription s) {
actual.onSubscribe(s);
}

@Override
public void onNext(ByteBuf byteBuf) {
actual.onNext(byteBuf);
byteBuf.release();
}

@Override
public void onError(Throwable t) {
actual.onError(t);
}

@Override
public void onComplete() {
actual.onComplete();
}
}));
}

@Override
Expand Down
Loading