Skip to content

Commit b27c7f6

Browse files
robertroeserrobertroeser
authored
RSocket hanging on large channels (#438)
* fixed channel bug when you emit nothing, fixed channel bug when hanging with large number of items * renamed UnboudProcessor to UnboundedProcessor * adding timeouts to tests
1 parent 0582cbe commit b27c7f6

File tree

6 files changed

+521
-45
lines changed

6 files changed

+521
-45
lines changed

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

Lines changed: 37 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,19 @@
1616

1717
package io.rsocket;
1818

19-
import static io.rsocket.util.ExceptionUtil.noStacktrace;
20-
2119
import io.netty.buffer.Unpooled;
2220
import io.netty.util.collection.IntObjectHashMap;
2321
import io.rsocket.exceptions.ConnectionException;
2422
import io.rsocket.exceptions.Exceptions;
2523
import io.rsocket.internal.LimitableRequestPublisher;
24+
import io.rsocket.internal.UnboundedProcessor;
2625
import io.rsocket.util.PayloadImpl;
26+
import org.reactivestreams.Publisher;
27+
import org.reactivestreams.Subscriber;
28+
import reactor.core.Disposable;
29+
import reactor.core.publisher.*;
30+
31+
import javax.annotation.Nullable;
2732
import java.nio.channels.ClosedChannelException;
2833
import java.time.Duration;
2934
import java.util.Collection;
@@ -32,11 +37,8 @@
3237
import java.util.function.Consumer;
3338
import java.util.function.Function;
3439
import java.util.function.Supplier;
35-
import javax.annotation.Nullable;
36-
import org.reactivestreams.Publisher;
37-
import org.reactivestreams.Subscriber;
38-
import reactor.core.Disposable;
39-
import reactor.core.publisher.*;
40+
41+
import static io.rsocket.util.ExceptionUtil.noStacktrace;
4042

4143
/** Client Side of a RSocket socket. Sends {@link Frame}s to a {@link RSocketServer} */
4244
class RSocketClient implements RSocket {
@@ -52,7 +54,7 @@ class RSocketClient implements RSocket {
5254
private final IntObjectHashMap<Subscriber<Payload>> receivers;
5355
private final AtomicInteger missedAckCounter;
5456

55-
private final FluxProcessor<Frame, Frame> sendProcessor;
57+
private final UnboundedProcessor<Frame> sendProcessor;
5658

5759
private @Nullable Disposable keepAliveSendSub;
5860
private volatile long timeLastTickSentMs;
@@ -80,8 +82,7 @@ class RSocketClient implements RSocket {
8082
this.missedAckCounter = new AtomicInteger();
8183

8284
// DO NOT Change the order here. The Send processor must be subscribed to before receiving
83-
// connections
84-
this.sendProcessor = EmitterProcessor.<Frame>create().serialize();
85+
this.sendProcessor = new UnboundedProcessor<>();
8586

8687
if (!Duration.ZERO.equals(tickPeriod)) {
8788
long ackTimeoutMs = ackTimeout.toMillis();
@@ -98,8 +99,15 @@ class RSocketClient implements RSocket {
9899
})
99100
.subscribe();
100101
}
101-
102-
connection.onClose().doFinally(signalType -> cleanup()).doOnError(errorConsumer).subscribe();
102+
103+
connection
104+
.onClose()
105+
.doFinally(
106+
signalType -> {
107+
cleanup();
108+
})
109+
.doOnError(errorConsumer)
110+
.subscribe();
103111

104112
connection
105113
.send(sendProcessor)
@@ -205,7 +213,7 @@ public Flux<Payload> requestStream(Payload payload) {
205213

206214
@Override
207215
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
208-
return handleStreamResponse(Flux.from(payloads), FrameType.REQUEST_CHANNEL);
216+
return handleChannel(Flux.from(payloads), FrameType.REQUEST_CHANNEL);
209217
}
210218

211219
@Override
@@ -255,6 +263,7 @@ public Flux<Payload> handleRequestStream(final Payload payload) {
255263
} else if (contains(streamId) && !receiver.isTerminated()) {
256264
sendProcessor.onNext(Frame.RequestN.from(streamId, l));
257265
}
266+
sendProcessor.drain();
258267
})
259268
.doOnError(
260269
t -> {
@@ -268,7 +277,10 @@ public Flux<Payload> handleRequestStream(final Payload payload) {
268277
sendProcessor.onNext(Frame.Cancel.from(streamId));
269278
}
270279
})
271-
.doFinally(s -> removeReceiver(streamId));
280+
.doFinally(
281+
s -> {
282+
removeReceiver(streamId);
283+
});
272284
}));
273285
}
274286

@@ -291,11 +303,14 @@ private Mono<Payload> handleRequestResponse(final Payload payload) {
291303
return receiver
292304
.doOnError(t -> sendProcessor.onNext(Frame.Error.from(streamId, t)))
293305
.doOnCancel(() -> sendProcessor.onNext(Frame.Cancel.from(streamId)))
294-
.doFinally(s -> removeReceiver(streamId));
306+
.doFinally(
307+
s -> {
308+
removeReceiver(streamId);
309+
});
295310
}));
296311
}
297312

298-
private Flux<Payload> handleStreamResponse(Flux<Payload> request, FrameType requestType) {
313+
private Flux<Payload> handleChannel(Flux<Payload> request, FrameType requestType) {
299314
return started.thenMany(
300315
Flux.defer(
301316
new Supplier<Flux<Payload>>() {
@@ -328,6 +343,7 @@ public Flux<Payload> get() {
328343
}
329344

330345
if (_firstRequest) {
346+
AtomicBoolean firstPayload = new AtomicBoolean(true);
331347
Flux<Frame> requestFrames =
332348
request
333349
.transform(
@@ -345,19 +361,10 @@ public Flux<Payload> get() {
345361
})
346362
.map(
347363
new Function<Payload, Frame>() {
348-
boolean firstPayload = true;
349364

350365
@Override
351366
public Frame apply(Payload payload) {
352-
boolean _firstPayload = false;
353-
synchronized (this) {
354-
if (firstPayload) {
355-
firstPayload = false;
356-
_firstPayload = true;
357-
}
358-
}
359-
360-
if (_firstPayload) {
367+
if (firstPayload.compareAndSet(true, false)) {
361368
return Frame.Request.from(
362369
streamId, requestType, payload, l);
363370
} else {
@@ -372,6 +379,9 @@ public Frame apply(Payload payload) {
372379
sendOneFrame(
373380
Frame.PayloadFrame.from(
374381
streamId, FrameType.COMPLETE));
382+
if (firstPayload.get()) {
383+
receiver.onComplete();
384+
}
375385
}
376386
});
377387

@@ -522,6 +532,7 @@ private void handleFrame(int streamId, FrameType type, Frame frame) {
522532
if (sender != null) {
523533
int n = Frame.RequestN.requestN(frame);
524534
sender.increaseRequestLimit(n);
535+
sendProcessor.drain();
525536
}
526537
break;
527538
}

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

Lines changed: 19 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,29 @@
1616

1717
package io.rsocket;
1818

19-
import static io.rsocket.Frame.Request.initialRequestN;
20-
import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
21-
import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
22-
2319
import io.netty.buffer.ByteBuf;
2420
import io.netty.buffer.Unpooled;
2521
import io.netty.util.collection.IntObjectHashMap;
2622
import io.rsocket.exceptions.ApplicationException;
2723
import io.rsocket.internal.LimitableRequestPublisher;
24+
import io.rsocket.internal.UnboundedProcessor;
2825
import io.rsocket.util.PayloadImpl;
29-
import java.util.Collection;
30-
import java.util.function.Consumer;
31-
import javax.annotation.Nullable;
3226
import org.reactivestreams.Publisher;
3327
import org.reactivestreams.Subscriber;
3428
import org.reactivestreams.Subscription;
3529
import reactor.core.Disposable;
36-
import reactor.core.publisher.*;
30+
import reactor.core.publisher.Flux;
31+
import reactor.core.publisher.Mono;
32+
import reactor.core.publisher.SignalType;
33+
import reactor.core.publisher.UnicastProcessor;
34+
35+
import javax.annotation.Nullable;
36+
import java.util.Collection;
37+
import java.util.function.Consumer;
38+
39+
import static io.rsocket.Frame.Request.initialRequestN;
40+
import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
41+
import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
3742

3843
/** Server side RSocket. Receives {@link Frame}s from a {@link RSocketClient} */
3944
class RSocketServer implements RSocket {
@@ -45,7 +50,7 @@ class RSocketServer implements RSocket {
4550
private final IntObjectHashMap<Subscription> sendingSubscriptions;
4651
private final IntObjectHashMap<UnicastProcessor<Payload>> channelProcessors;
4752

48-
private final FluxProcessor<Frame, Frame> sendProcessor;
53+
private final UnboundedProcessor<Frame> sendProcessor;
4954
private Disposable receiveDisposable;
5055

5156
RSocketServer(
@@ -58,7 +63,7 @@ class RSocketServer implements RSocket {
5863

5964
// DO NOT Change the order here. The Send processor must be subscribed to before receiving
6065
// connections
61-
this.sendProcessor = EmitterProcessor.<Frame>create().serialize();
66+
this.sendProcessor = new UnboundedProcessor<>();
6267

6368
connection
6469
.send(sendProcessor)
@@ -302,7 +307,10 @@ private Mono<Void> handleRequestResponse(int streamId, Mono<Payload> response) {
302307
.doOnError(errorConsumer)
303308
.onErrorResume(t -> Mono.just(Frame.Error.from(streamId, t)))
304309
.doOnNext(sendProcessor::onNext)
305-
.doFinally(signalType -> removeSubscription(streamId))
310+
.doFinally(
311+
signalType -> {
312+
removeSubscription(streamId);
313+
})
306314
.then();
307315
}
308316

0 commit comments

Comments
 (0)