Skip to content

Commit b87eed4

Browse files
OlegDokukaOlegDokuka
authored
improves ref-counting by eliminating redundant retain/release (rsocket#902)
improves ref counting and performance subsequently by eliminating redundant retain on the transport level hence we do not need to release this frame on the RSocketRequester/RSocketResponder level anymore Fixes all duplex connections to follow that strategy Signed-off-by: Oleh Dokuka <shadowgun@i.ua>
1 parent d614a28 commit b87eed4

12 files changed

Lines changed: 242 additions & 99 deletions

File tree

‎rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import static io.rsocket.keepalive.KeepAliveSupport.ClientKeepAliveSupport;
2020

2121
import io.netty.buffer.ByteBuf;
22-
import io.netty.util.ReferenceCountUtil;
2322
import io.rsocket.DuplexConnection;
2423
import io.rsocket.Payload;
2524
import io.rsocket.RSocket;
@@ -205,10 +204,14 @@ private void handleIncomingFrames(ByteBuf frame) {
205204
} else {
206205
handleFrame(streamId, type, frame);
207206
}
208-
frame.release();
209207
} catch (Throwable t) {
210-
ReferenceCountUtil.safeRelease(frame);
211-
throw reactor.core.Exceptions.propagate(t);
208+
super.getSendProcessor()
209+
.onNext(
210+
ErrorFrameCodec.encode(
211+
super.getAllocator(),
212+
0,
213+
new ConnectionErrorException("Unexpected error during frame handling", t)));
214+
this.tryTerminateOnConnectionError(t);
212215
}
213216
}
214217

‎rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,11 @@
1717
package io.rsocket.core;
1818

1919
import io.netty.buffer.ByteBuf;
20-
import io.netty.util.ReferenceCountUtil;
2120
import io.rsocket.DuplexConnection;
2221
import io.rsocket.Payload;
2322
import io.rsocket.RSocket;
2423
import io.rsocket.ResponderRSocket;
24+
import io.rsocket.exceptions.ConnectionErrorException;
2525
import io.rsocket.frame.ErrorFrameCodec;
2626
import io.rsocket.frame.FrameHeaderCodec;
2727
import io.rsocket.frame.FrameType;
@@ -39,7 +39,6 @@
3939
import org.slf4j.Logger;
4040
import org.slf4j.LoggerFactory;
4141
import reactor.core.Disposable;
42-
import reactor.core.Exceptions;
4342
import reactor.core.publisher.Flux;
4443
import reactor.core.publisher.Mono;
4544

@@ -314,10 +313,14 @@ private void handleFrame(ByteBuf frame) {
314313
"ServerRSocket: Unexpected frame type: " + frameType)));
315314
break;
316315
}
317-
ReferenceCountUtil.safeRelease(frame);
318316
} catch (Throwable t) {
319-
ReferenceCountUtil.safeRelease(frame);
320-
throw Exceptions.propagate(t);
317+
super.getSendProcessor()
318+
.onNext(
319+
ErrorFrameCodec.encode(
320+
super.getAllocator(),
321+
0,
322+
new ConnectionErrorException("Unexpected error during frame handling", t)));
323+
this.tryTerminateOnConnectionError(t);
321324
}
322325
}
323326

‎rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -387,29 +387,22 @@ private Mono<Void> acceptSetup(
387387
multiplexer,
388388
new InvalidSetupException(
389389
"Unsupported version: " + SetupFrameCodec.humanReadableVersion(setupFrame)))
390-
.doFinally(
391-
signalType -> {
392-
setupFrame.release();
393-
multiplexer.dispose();
394-
});
390+
.doFinally(signalType -> multiplexer.dispose());
395391
}
396392

397393
boolean leaseEnabled = leasesSupplier != null;
398394
if (SetupFrameCodec.honorLease(setupFrame) && !leaseEnabled) {
399395
return serverSetup
400396
.sendError(multiplexer, new InvalidSetupException("lease is not supported"))
401-
.doFinally(
402-
signalType -> {
403-
setupFrame.release();
404-
multiplexer.dispose();
405-
});
397+
.doFinally(signalType -> multiplexer.dispose());
406398
}
407399

408400
return serverSetup.acceptRSocketSetup(
409401
setupFrame,
410402
multiplexer,
411403
(keepAliveHandler, wrappedMultiplexer) -> {
412-
ConnectionSetupPayload setupPayload = new DefaultConnectionSetupPayload(setupFrame);
404+
ConnectionSetupPayload setupPayload =
405+
new DefaultConnectionSetupPayload(setupFrame.retain());
413406

414407
Leases<?> leases = leaseEnabled ? leasesSupplier.get() : null;
415408
RequesterLeaseHandler requesterLeaseHandler =

‎rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterSubscribersTest.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ void setUp() {
8282

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

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

94-
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), 1));
94+
if (requestType != FrameType.REQUEST_FNF && requestType != FrameType.METADATA_PUSH) {
95+
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), 1));
96+
}
9597

9698
assertSubscriberA.assertTerminated();
9799
assertSubscriberB.assertTerminated();
@@ -111,7 +113,9 @@ void singleSubscriberInCaseOfRacing(
111113
RaceTestUtils.race(
112114
() -> response.subscribe(assertSubscriberA), () -> response.subscribe(assertSubscriberB));
113115

114-
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), i));
116+
if (requestType != FrameType.REQUEST_FNF && requestType != FrameType.METADATA_PUSH) {
117+
connection.addToReceivedBuffer(PayloadFrameCodec.encodeComplete(connection.alloc(), i));
118+
}
115119

116120
assertSubscriberA.assertTerminated();
117121
assertSubscriberB.assertTerminated();

‎rsocket-core/src/test/java/io/rsocket/internal/ClientServerInputMultiplexerTest.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -68,52 +68,52 @@ public void clientSplits() {
6868
.doOnNext(f -> setupFrames.incrementAndGet())
6969
.subscribe();
7070

71-
source.addToReceivedBuffer(errorFrame(1));
71+
source.addToReceivedBuffer(errorFrame(1).retain());
7272
assertEquals(1, clientFrames.get());
7373
assertEquals(0, serverFrames.get());
7474
assertEquals(0, setupFrames.get());
7575

76-
source.addToReceivedBuffer(errorFrame(1));
76+
source.addToReceivedBuffer(errorFrame(1).retain());
7777
assertEquals(2, clientFrames.get());
7878
assertEquals(0, serverFrames.get());
7979
assertEquals(0, setupFrames.get());
8080

81-
source.addToReceivedBuffer(leaseFrame());
81+
source.addToReceivedBuffer(leaseFrame().retain());
8282
assertEquals(3, clientFrames.get());
8383
assertEquals(0, serverFrames.get());
8484
assertEquals(0, setupFrames.get());
8585

86-
source.addToReceivedBuffer(keepAliveFrame());
86+
source.addToReceivedBuffer(keepAliveFrame().retain());
8787
assertEquals(4, clientFrames.get());
8888
assertEquals(0, serverFrames.get());
8989
assertEquals(0, setupFrames.get());
9090

91-
source.addToReceivedBuffer(errorFrame(2));
91+
source.addToReceivedBuffer(errorFrame(2).retain());
9292
assertEquals(4, clientFrames.get());
9393
assertEquals(1, serverFrames.get());
9494
assertEquals(0, setupFrames.get());
9595

96-
source.addToReceivedBuffer(errorFrame(0));
96+
source.addToReceivedBuffer(errorFrame(0).retain());
9797
assertEquals(5, clientFrames.get());
9898
assertEquals(1, serverFrames.get());
9999
assertEquals(0, setupFrames.get());
100100

101-
source.addToReceivedBuffer(metadataPushFrame());
101+
source.addToReceivedBuffer(metadataPushFrame().retain());
102102
assertEquals(5, clientFrames.get());
103103
assertEquals(2, serverFrames.get());
104104
assertEquals(0, setupFrames.get());
105105

106-
source.addToReceivedBuffer(setupFrame());
106+
source.addToReceivedBuffer(setupFrame().retain());
107107
assertEquals(5, clientFrames.get());
108108
assertEquals(2, serverFrames.get());
109109
assertEquals(1, setupFrames.get());
110110

111-
source.addToReceivedBuffer(resumeFrame());
111+
source.addToReceivedBuffer(resumeFrame().retain());
112112
assertEquals(5, clientFrames.get());
113113
assertEquals(2, serverFrames.get());
114114
assertEquals(2, setupFrames.get());
115115

116-
source.addToReceivedBuffer(resumeOkFrame());
116+
source.addToReceivedBuffer(resumeOkFrame().retain());
117117
assertEquals(5, clientFrames.get());
118118
assertEquals(2, serverFrames.get());
119119
assertEquals(3, setupFrames.get());
@@ -141,52 +141,52 @@ public void serverSplits() {
141141
.doOnNext(f -> setupFrames.incrementAndGet())
142142
.subscribe();
143143

144-
source.addToReceivedBuffer(errorFrame(1));
144+
source.addToReceivedBuffer(errorFrame(1).retain());
145145
assertEquals(1, clientFrames.get());
146146
assertEquals(0, serverFrames.get());
147147
assertEquals(0, setupFrames.get());
148148

149-
source.addToReceivedBuffer(errorFrame(1));
149+
source.addToReceivedBuffer(errorFrame(1).retain());
150150
assertEquals(2, clientFrames.get());
151151
assertEquals(0, serverFrames.get());
152152
assertEquals(0, setupFrames.get());
153153

154-
source.addToReceivedBuffer(leaseFrame());
154+
source.addToReceivedBuffer(leaseFrame().retain());
155155
assertEquals(2, clientFrames.get());
156156
assertEquals(1, serverFrames.get());
157157
assertEquals(0, setupFrames.get());
158158

159-
source.addToReceivedBuffer(keepAliveFrame());
159+
source.addToReceivedBuffer(keepAliveFrame().retain());
160160
assertEquals(2, clientFrames.get());
161161
assertEquals(2, serverFrames.get());
162162
assertEquals(0, setupFrames.get());
163163

164-
source.addToReceivedBuffer(errorFrame(2));
164+
source.addToReceivedBuffer(errorFrame(2).retain());
165165
assertEquals(2, clientFrames.get());
166166
assertEquals(3, serverFrames.get());
167167
assertEquals(0, setupFrames.get());
168168

169-
source.addToReceivedBuffer(errorFrame(0));
169+
source.addToReceivedBuffer(errorFrame(0).retain());
170170
assertEquals(2, clientFrames.get());
171171
assertEquals(4, serverFrames.get());
172172
assertEquals(0, setupFrames.get());
173173

174-
source.addToReceivedBuffer(metadataPushFrame());
174+
source.addToReceivedBuffer(metadataPushFrame().retain());
175175
assertEquals(3, clientFrames.get());
176176
assertEquals(4, serverFrames.get());
177177
assertEquals(0, setupFrames.get());
178178

179-
source.addToReceivedBuffer(setupFrame());
179+
source.addToReceivedBuffer(setupFrame().retain());
180180
assertEquals(3, clientFrames.get());
181181
assertEquals(4, serverFrames.get());
182182
assertEquals(1, setupFrames.get());
183183

184-
source.addToReceivedBuffer(resumeFrame());
184+
source.addToReceivedBuffer(resumeFrame().retain());
185185
assertEquals(3, clientFrames.get());
186186
assertEquals(4, serverFrames.get());
187187
assertEquals(2, setupFrames.get());
188188

189-
source.addToReceivedBuffer(resumeOkFrame());
189+
source.addToReceivedBuffer(resumeOkFrame().retain());
190190
assertEquals(3, clientFrames.get());
191191
assertEquals(4, serverFrames.get());
192192
assertEquals(3, setupFrames.get());

‎rsocket-core/src/test/java/io/rsocket/test/util/LocalDuplexConnection.java‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,13 @@
2020
import io.netty.buffer.ByteBufAllocator;
2121
import io.rsocket.DuplexConnection;
2222
import org.reactivestreams.Publisher;
23+
import org.reactivestreams.Subscription;
24+
import reactor.core.CoreSubscriber;
2325
import reactor.core.publisher.DirectProcessor;
2426
import reactor.core.publisher.Flux;
2527
import reactor.core.publisher.Mono;
2628
import reactor.core.publisher.MonoProcessor;
29+
import reactor.core.publisher.Operators;
2730

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

5659
@Override
5760
public Flux<ByteBuf> receive() {
58-
return receive.doOnNext(f -> System.out.println(name + " - " + f.toString()));
61+
return receive
62+
.doOnNext(f -> System.out.println(name + " - " + f.toString()))
63+
.transform(
64+
Operators.<ByteBuf, ByteBuf>lift(
65+
(__, actual) ->
66+
new CoreSubscriber<ByteBuf>() {
67+
68+
@Override
69+
public void onSubscribe(Subscription s) {
70+
actual.onSubscribe(s);
71+
}
72+
73+
@Override
74+
public void onNext(ByteBuf byteBuf) {
75+
actual.onNext(byteBuf);
76+
byteBuf.release();
77+
}
78+
79+
@Override
80+
public void onError(Throwable t) {
81+
actual.onError(t);
82+
}
83+
84+
@Override
85+
public void onComplete() {
86+
actual.onComplete();
87+
}
88+
}));
5989
}
6090

6191
@Override

‎rsocket-core/src/test/java/io/rsocket/test/util/TestDuplexConnection.java‎

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,16 @@
2424
import java.util.concurrent.LinkedBlockingQueue;
2525
import org.reactivestreams.Publisher;
2626
import org.reactivestreams.Subscriber;
27+
import org.reactivestreams.Subscription;
2728
import org.slf4j.Logger;
2829
import org.slf4j.LoggerFactory;
30+
import reactor.core.CoreSubscriber;
2931
import reactor.core.publisher.DirectProcessor;
3032
import reactor.core.publisher.Flux;
3133
import reactor.core.publisher.FluxSink;
3234
import reactor.core.publisher.Mono;
3335
import reactor.core.publisher.MonoProcessor;
36+
import reactor.core.publisher.Operators;
3437

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

8487
@Override
8588
public Flux<ByteBuf> receive() {
86-
return received;
89+
return received.transform(
90+
Operators.<ByteBuf, ByteBuf>lift(
91+
(__, actual) ->
92+
new CoreSubscriber<ByteBuf>() {
93+
@Override
94+
public void onSubscribe(Subscription s) {
95+
actual.onSubscribe(s);
96+
}
97+
98+
@Override
99+
public void onNext(ByteBuf byteBuf) {
100+
actual.onNext(byteBuf);
101+
byteBuf.release();
102+
}
103+
104+
@Override
105+
public void onError(Throwable t) {
106+
actual.onError(t);
107+
}
108+
109+
@Override
110+
public void onComplete() {
111+
actual.onComplete();
112+
}
113+
}));
87114
}
88115

89116
@Override

0 commit comments

Comments
 (0)