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

Jump to bottom

Avoid allocating instances of AtomicReference in Uni #660

Closed
wants to merge 2 commits into from
Closed

Avoid allocating instances of AtomicReference in Uni #660

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4704146
Avoid allocating excessive amounts of AtomicReference instances
Sanne Aug 24, 2021
9be8567
Improve abstraction over internal state of UniOperatorProcessor
Sanne Aug 24, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 3 additions & 3 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniFailOnTimeout.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ public void onSubscribe(UniSubscription subscription) {
timeoutFuture = executor.schedule(this::doTimeout, timeout.toMillis(), TimeUnit.MILLISECONDS);
} catch (RejectedExecutionException e) {
// Executor out of service.
upstream.set(CANCELLED);
getAndSetUpstreamSubscription(CANCELLED);
subscription.cancel();
downstream.onSubscribe(EmptyUniSubscription.DONE);
downstream.onFailure(e);
Expand All @@ -61,7 +61,7 @@ public void onSubscribe(UniSubscription subscription) {

@Override
public void onItem(I item) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
UniSubscription sub = getAndSetUpstreamSubscription(CANCELLED);
if (sub != CANCELLED) {
if (timeoutFuture != null) {
timeoutFuture.cancel(false);
Expand All @@ -72,7 +72,7 @@ public void onItem(I item) {

@Override
public void onFailure(Throwable failure) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
UniSubscription sub = getAndSetUpstreamSubscription(CANCELLED);
if (sub != CANCELLED) {
if (timeoutFuture != null) {
timeoutFuture.cancel(false);
Expand Down
2 changes: 1 addition & 1 deletion implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnCancellation.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ public void onFailure(Throwable failure) {
@Override
public void cancel() {
if (state.compareAndSet(State.INIT, State.CANCELLED)) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
UniSubscription sub = getAndSetUpstreamSubscription(CANCELLED);
callback.run();
if (sub != null) {
sub.cancel();
Expand Down
2 changes: 1 addition & 1 deletion implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnCancellationCall.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ public void onFailure(Throwable failure) {
@Override
public void cancel() {
if (state.compareAndSet(State.INIT, State.CANCELLED)) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
UniSubscription sub = getAndSetUpstreamSubscription(CANCELLED);
execute().subscribe().with(
ignoredItem -> {
if (sub != null) {
Expand Down
2 changes: 1 addition & 1 deletion implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnFailureFlatMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public UniOnFailureFlatMapProcessor(UniSubscriber<? super I> downstream) {

@Override
public void onSubscribe(UniSubscription subscription) {
if (upstream.get() == null) {
if (getCurrentUpstreamSubscription() == null) {
super.onSubscribe(subscription);
} else if (innerSubscription == null) {
this.innerSubscription = subscription;
Expand Down
2 changes: 1 addition & 1 deletion implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnItemOrFailureFlatMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public UniOnItemOrFailureFlatMapProcessor(UniSubscriber<? super O> downstream) {

@Override
public void onSubscribe(UniSubscription subscription) {
if (upstream.get() == null) {
if (getCurrentUpstreamSubscription() == null) {
super.onSubscribe(subscription);
} else if (innerSubscription == null) {
this.innerSubscription = subscription;
Expand Down
2 changes: 1 addition & 1 deletion implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnItemTransformToUni.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ public UniOnItemTransformToUniProcessor(UniSubscriber<? super O> downstream) {

@Override
public void onSubscribe(UniSubscription subscription) {
if (upstream.get() == null) {
if (getCurrentUpstreamSubscription() == null) {
super.onSubscribe(subscription);
} else if (innerSubscription == null) {
this.innerSubscription = subscription;
Expand Down
31 changes: 24 additions & 7 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOperatorProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
import static io.smallrye.mutiny.helpers.EmptyUniSubscription.CANCELLED;
import static io.smallrye.mutiny.helpers.EmptyUniSubscription.DONE;

import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import io.smallrye.mutiny.helpers.ParameterValidation;
import io.smallrye.mutiny.infrastructure.Infrastructure;
Expand All @@ -13,15 +13,19 @@
public abstract class UniOperatorProcessor<I, O> implements UniSubscriber<I>, UniSubscription {

protected final UniSubscriber<? super O> downstream;
protected final AtomicReference<UniSubscription> upstream = new AtomicReference<>();

private static final AtomicReferenceFieldUpdater<UniOperatorProcessor, UniSubscription> updater = AtomicReferenceFieldUpdater
.newUpdater(UniOperatorProcessor.class, UniSubscription.class, "upstream");

private volatile UniSubscription upstream;

public UniOperatorProcessor(UniSubscriber<? super O> downstream) {
this.downstream = ParameterValidation.nonNull(downstream, "downstream");
}

@Override
public void onSubscribe(UniSubscription subscription) {
if (upstream.compareAndSet(null, subscription)) {
if (updater.compareAndSet(this, null, subscription)) {
jponge marked this conversation as resolved.
Show resolved Hide resolved
downstream.onSubscribe(this);
} else {
subscription.cancel();
Expand All @@ -31,15 +35,15 @@ public void onSubscribe(UniSubscription subscription) {
@Override
@SuppressWarnings("unchecked")
public void onItem(I item) {
UniSubscription subscription = upstream.getAndSet(CANCELLED);
UniSubscription subscription = updater.getAndSet(this, CANCELLED);
if (subscription != CANCELLED) {
downstream.onItem((O) item);
}
}

@Override
public void onFailure(Throwable failure) {
UniSubscription subscription = upstream.getAndSet(CANCELLED);
UniSubscription subscription = updater.getAndSet(this, CANCELLED);
if (subscription != CANCELLED) {
downstream.onFailure(failure);
} else {
Expand All @@ -49,13 +53,26 @@ public void onFailure(Throwable failure) {

@Override
public void cancel() {
UniSubscription subscription = upstream.getAndSet(CANCELLED);
UniSubscription subscription = updater.getAndSet(this, CANCELLED);
if (subscription != null && subscription != CANCELLED && subscription != DONE) {
subscription.cancel();
}
}

public boolean isCancelled() {
return upstream.get() == CANCELLED;
return upstream == CANCELLED;
}

protected final UniSubscription getCurrentUpstreamSubscription() {
return upstream;
}

protected final UniSubscription getAndSetUpstreamSubscription(UniSubscription newValue) {
return updater.getAndSet(this, newValue);
}

protected final boolean compareAndSetUpstreamSubscription(UniSubscription expect, UniSubscription update) {
return updater.compareAndSet(this, expect, update);
}

}
4 changes: 2 additions & 2 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniRetryAtMost.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ public UniRetryAtMostProcessor(UniSubscriber<? super T> downstream) {
@Override
public void onSubscribe(UniSubscription subscription) {
int count = counter.incrementAndGet();
if (upstream.compareAndSet(null, subscription)) {
if (compareAndSetUpstreamSubscription(null, subscription)) {
if (count == 1) {
downstream.onSubscribe(this);
}
Expand All @@ -61,7 +61,7 @@ public void onFailure(Throwable failure) {
downstream.onFailure(failure);
return;
}
UniSubscription previousSubscription = upstream.getAndSet(null);
UniSubscription previousSubscription = getAndSetUpstreamSubscription(null);
if (previousSubscription != null) {
previousSubscription.cancel();
}
Expand Down