Skip to content
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

Handle exceptions from Subscribers #2553

Merged
merged 7 commits into from
Mar 10, 2020
Merged

Handle exceptions from Subscribers #2553

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d0857ce
Handle exceptions from Subscribers
minwoox Mar 4, 2020
e131ab1
Address comments by @trustin
minwoox Mar 6, 2020
f3a4641
Address comments by @anuraaga
minwoox Mar 6, 2020
1f2b076
Address comments by @trustin
minwoox Mar 9, 2020
7b3e5dd
Add to notice.txt
minwoox Mar 9, 2020
3e2d586
Address comments by @ikhoon and @trustin
minwoox Mar 10, 2020
8893ee6
Address comments by @ikhoon
minwoox Mar 10, 2020
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
4 changes: 0 additions & 4 deletions core/src/main/java/com/linecorp/armeria/client/endpoint/healthcheck/HttpHealthChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,10 +174,6 @@ public void onSubscribe(Subscription subscription) {

@Override
public void onNext(HttpObject obj) {
// There's no chance that an exception is raised from the underlying logic, so we don't do
// try catch not to propagate the exception to the publisher.
// https://github.com/reactive-streams/reactive-streams-jvm#2.13

if (closeable.isClosing()) {
subscription.cancel();
return;
Expand Down
4 changes: 0 additions & 4 deletions core/src/main/java/com/linecorp/armeria/common/HttpMessageAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,10 +103,6 @@ public final void onComplete() {

@Override
public final void onNext(HttpObject o) {
// There's no chance that an exception is raised from the underlying logic, so we don't do try catch not
// to propagate the exception to the publisher.
// https://github.com/reactive-streams/reactive-streams-jvm#2.13

if (o instanceof HttpHeaders) {
onHeaders((HttpHeaders) o);
} else {
Expand Down
36 changes: 23 additions & 13 deletions core/src/main/java/com/linecorp/armeria/common/stream/AbstractStreamMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.MoreObjects;
import com.spotify.futures.CompletableFutures;
Expand All @@ -40,6 +42,8 @@

abstract class AbstractStreamMessage<T> implements StreamMessage<T> {

private static final Logger logger = LoggerFactory.getLogger(AbstractStreamMessage.class);

static final CloseEvent SUCCESSFUL_CLOSE = new CloseEvent(null);
static final CloseEvent CANCELLED_CLOSE = new CloseEvent(CancelledSubscriptionException.INSTANCE);
static final CloseEvent ABORTED_CLOSE = new CloseEvent(AbortedStreamException.INSTANCE);
Expand Down Expand Up @@ -145,16 +149,23 @@ static void failLateSubscriber(SubscriptionImpl subscription, Subscriber<?> late
final Throwable cause = abortedOrLate(oldSubscriber);

if (subscription.needsDirectInvocation()) {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(cause);
lateSubscriber(lateSubscriber, cause);
} else {
subscription.executor().execute(() -> {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(cause);
lateSubscriber(lateSubscriber, cause);
});
}
}

private static void lateSubscriber(Subscriber<?> lateSubscriber, Throwable cause) {
minwoox marked this conversation as resolved.
Show resolved Hide resolved
try {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(cause);
} catch (Exception e) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Global comment: Should we catch a Throwable?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reactive-streams/reactive-streams-jvm#107 (comment)
I think they are guiding not to catch the fatal ones. 🤔

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there are Error that aren't fatal, so better to catch Throwable and we may need a propagateIfFatal like in zipkin

// Taken from RxJava throwIfFatal, which was taken from scala
  public static void propagateIfFatal(Throwable t) {
    if (t instanceof VirtualMachineError) {
      throw (VirtualMachineError) t;
    } else if (t instanceof ThreadDeath) {
      throw (ThreadDeath) t;
    } else if (t instanceof LinkageError) {
      throw (LinkageError) t;
    }
  }

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, let me add this method to Exceptions. Thanks!

Copy link
Contributor

@ikhoon ikhoon Mar 6, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we don't have to follow the Reactive Streams TCK, just FYI, which also regards InterruptedException as fatal.
https://github.com/reactive-streams/reactive-streams-jvm/blob/master/tck/src/main/java/org/reactivestreams/tck/flow/support/NonFatal.java#L35
It is copied from Scala code.
https://github.com/scala/scala/blob/v2.13.1/src/library/scala/util/control/NonFatal.scala#L41

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah yes, I saw it from ReactiveX/RxJava#748 (comment)
and I think they intentionally omit that. But I didn't think about it deeply. 😆

logger.warn("Subscriber should not throw an exception. subscriber: {}", lateSubscriber, e);
}
}

T prepareObjectForNotification(SubscriptionImpl subscription, T o) {
ReferenceCountUtil.touch(o);
onRemoval(o);
Expand Down Expand Up @@ -231,7 +242,8 @@ boolean cancelRequested() {
@Override
public void request(long n) {
if (n <= 0) {
invokeOnError(new IllegalArgumentException(
// Just abort the publisher so subscriber().onError(e) is called and resources are cleaned up.
publisher.abort(new IllegalArgumentException(
"n: " + n + " (expected: > 0, see Reactive Streams specification rule 3.9)"));
return;
}
Expand All @@ -245,14 +257,6 @@ public void cancel() {
publisher.cancel();
}

private void invokeOnError(Throwable cause) {
if (needsDirectInvocation()) {
subscriber.onError(cause);
} else {
executor.execute(() -> subscriber.onError(cause));
}
}

// We directly run callbacks for event loops if we're already on the loop, which applies to the vast
// majority of cases.
boolean needsDirectInvocation() {
Expand Down Expand Up @@ -291,6 +295,9 @@ void notifySubscriber(SubscriptionImpl subscription, CompletableFuture<?> comple
if (cause == null) {
try {
subscriber.onComplete();
} catch (Exception e) {
logger.warn("Subscriber.onComplete() should not raise an exception. subscriber: {}",
trustin marked this conversation as resolved.
Show resolved Hide resolved
subscriber, e);
} finally {
completionFuture.complete(null);
}
Expand All @@ -299,6 +306,9 @@ void notifySubscriber(SubscriptionImpl subscription, CompletableFuture<?> comple
if (subscription.notifyCancellation || !(cause instanceof CancelledSubscriptionException)) {
subscriber.onError(cause);
}
} catch (Exception e) {
logger.warn("Subscriber.onError() should not raise an exception. subscriber: {}",
trustin marked this conversation as resolved.
Show resolved Hide resolved
subscriber, e);
} finally {
completionFuture.completeExceptionally(cause);
}
Expand Down
37 changes: 31 additions & 6 deletions core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@

import org.jctools.queues.MpscChunkedArrayQueue;
import org.reactivestreams.Subscriber;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.linecorp.armeria.common.util.UnstableApi;

Expand Down Expand Up @@ -66,6 +68,8 @@
@UnstableApi
public class DefaultStreamMessage<T> extends AbstractStreamMessageAndWriter<T> {

private static final Logger logger = LoggerFactory.getLogger(DefaultStreamMessage.class);

@SuppressWarnings("rawtypes")
private static final AtomicReferenceFieldUpdater<DefaultStreamMessage, SubscriptionImpl>
subscriptionUpdater = AtomicReferenceFieldUpdater.newUpdater(
Expand Down Expand Up @@ -118,18 +122,30 @@ SubscriptionImpl subscribe(SubscriptionImpl subscription) {

final Subscriber<Object> subscriber = subscription.subscriber();
if (subscription.needsDirectInvocation()) {
invokedOnSubscribe = true;
subscriber.onSubscribe(subscription);
subscribe0(subscription, subscriber);
} else {
subscription.executor().execute(() -> {
invokedOnSubscribe = true;
subscriber.onSubscribe(subscription);
subscribe0(subscription, subscriber);
});
}

return subscription;
}

private void subscribe0(SubscriptionImpl subscription, Subscriber<Object> subscriber) {
try {
invokedOnSubscribe = true;
subscriber.onSubscribe(subscription);
} catch (Exception e) {
if (setState(State.OPEN, State.CLEANUP) || setState(State.CLOSED, State.CLEANUP)) {
notifySubscriberOfCloseEvent(subscription, newCloseEvent(e));
} else {
logger.warn("Subscriber.onSubscribe() should not raise an exception. subscriber: {}",
subscriber, e);
}
}
}

@Override
public void abort() {
abort0(AbortedStreamException.get());
Expand Down Expand Up @@ -328,7 +344,7 @@ private void notifySubscriber0() {

for (;;) {
if (state == State.CLEANUP) {
cleanup();
cleanupObjects();
return;
}

Expand Down Expand Up @@ -375,6 +391,15 @@ private boolean notifySubscriberWithElements(SubscriptionImpl subscription) {
try {
o = prepareObjectForNotification(subscription, o);
subscriber.onNext(o);
} catch (Exception e) {
if (setState(State.OPEN, State.CLEANUP) || setState(State.CLOSED, State.CLEANUP)) {
notifySubscriberOfCloseEvent(subscription, newCloseEvent(e));
} else {
logger.warn("Subscriber.onNext({}) should not raise an exception. subscriber: {}",
o, subscriber, e);
}

return false;
} finally {
inOnNext = false;
}
Expand Down Expand Up @@ -435,7 +460,7 @@ private boolean setState(State oldState, State newState) {
return stateUpdater.compareAndSet(this, oldState, newState);
}

private void cleanup() {
private void cleanupObjects() {
Throwable cause = null;
for (;;) {
final Object e = queue.poll();
Expand Down
46 changes: 38 additions & 8 deletions core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessageDuplicator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects.ToStringHelper;
Expand All @@ -64,6 +66,8 @@
@UnstableApi
public class DefaultStreamMessageDuplicator<T> implements StreamMessageDuplicator<T> {

private static final Logger logger = LoggerFactory.getLogger(DefaultStreamMessageDuplicator.class);

@SuppressWarnings("rawtypes")
private static final AtomicIntegerFieldUpdater<DefaultStreamMessageDuplicator> unsubscribedUpdater =
AtomicIntegerFieldUpdater.newUpdater(DefaultStreamMessageDuplicator.class, "unsubscribed");
Expand Down Expand Up @@ -286,9 +290,13 @@ private void doSubscribe(DownstreamSubscription<T> subscription) {

private static void failLateProcessorSubscriber(DownstreamSubscription<?> subscription) {
final Subscriber<?> lateSubscriber = subscription.subscriber();
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(
new IllegalStateException("duplicator is closed or no more downstream can be added."));
try {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(
new IllegalStateException("duplicator is closed or no more downstream can be added."));
} catch (Exception e) {
logger.warn("Subscriber should not throw an exception. subscriber: {}", lateSubscriber, e);
}
}

void unsubscribe(DownstreamSubscription<T> subscription, @Nullable Throwable cause) {
Expand All @@ -311,6 +319,9 @@ private void doUnsubscribe(DownstreamSubscription<T> subscription, @Nullable Thr
if (cause == null) {
try {
subscriber.onComplete();
} catch (Exception e) {
logger.warn("Subscriber.onComplete() should not raise an exception. subscriber: {}",
subscriber, e);
} finally {
completionFuture.complete(null);
doCleanupIfLastSubscription();
Expand All @@ -322,6 +333,9 @@ private void doUnsubscribe(DownstreamSubscription<T> subscription, @Nullable Thr
if (subscription.notifyCancellation || !(cause instanceof CancelledSubscriptionException)) {
subscriber.onError(cause);
}
} catch (Exception e) {
logger.warn("Subscriber.onError() should not raise an exception. subscriber: {}",
subscriber, e);
} finally {
completionFuture.completeExceptionally(cause);
doCleanupIfLastSubscription();
Expand Down Expand Up @@ -513,8 +527,12 @@ private static void failLateSubscriber(EventExecutor executor,
final Throwable cause = abortedOrLate(oldSubscriber);

executor.execute(() -> {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(cause);
try {
lateSubscriber.onSubscribe(NoopSubscription.INSTANCE);
lateSubscriber.onError(cause);
} catch (Exception e) {
logger.warn("Subscriber should not throw an exception. subscriber: {}", lateSubscriber, e);
}
});
}

Expand Down Expand Up @@ -644,17 +662,26 @@ void clearSubscriber() {
subscriber = NeverInvokedSubscriber.get();
}

// only called from processor.processorExecutor
// Called from processor.processorExecutor
void invokeOnSubscribe() {
if (invokedOnSubscribe) {
return;
}
invokedOnSubscribe = true;

if (executor.inEventLoop()) {
subscriber.onSubscribe(this);
invokeOnSubscribe0();
} else {
executor.execute(() -> subscriber.onSubscribe(this));
executor.execute(this::invokeOnSubscribe0);
}
}

// Called from the executor of the subscriber.
void invokeOnSubscribe0() {
try {
subscriber.onSubscribe(this);
} catch (Exception e) {
processor.unsubscribe(this, e);
}
}

Expand Down Expand Up @@ -800,6 +827,9 @@ private boolean doSignalSingle(SignalQueue signals) {
inOnNext = true;
try {
subscriber.onNext(obj);
} catch (Exception e) {
processor.unsubscribe(this, e);
return false;
} finally {
inOnNext = false;
}
Expand Down
Loading