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

Use atomic field updaters in selected Uni places #669

Merged
merged 4 commits into from Aug 31, 2021
Merged

Use atomic field updaters in selected Uni places #669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2c20281
Avoid allocating excessive amounts of AtomicReference instances
Sanne Aug 24, 2021
f77575b
Improve abstraction over internal state of UniOperatorProcessor
Sanne Aug 24, 2021
4e3f5fa
Use further atomic field updaters in selected places (Uni)
jponge Aug 30, 2021
e06ce7d
Use CaS methods in UniOperatorProcessor
jponge Aug 31, 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
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
26 changes: 16 additions & 10 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnCancellation.java
View file
Open in desktop
Expand Up @@ -2,7 +2,7 @@

import static io.smallrye.mutiny.helpers.EmptyUniSubscription.CANCELLED;

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

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.operators.AbstractUni;
Expand All @@ -11,6 +11,7 @@
import io.smallrye.mutiny.subscription.UniSubscription;

public class UniOnCancellation<T> extends UniOperator<T, T> {

private final Runnable callback;

public UniOnCancellation(Uni<T> upstream, Runnable callback) {
Expand All @@ -20,7 +21,7 @@ public UniOnCancellation(Uni<T> upstream, Runnable callback) {

@Override
public void subscribe(UniSubscriber<? super T> subscriber) {
AbstractUni.subscribe(upstream(), new UniOnCancellationProcessor(subscriber));
AbstractUni.subscribe(upstream(), new UniOnCancellationProcessor<T>(callback, subscriber));
}

private enum State {
Expand All @@ -29,32 +30,37 @@ private enum State {
CANCELLED
}

private class UniOnCancellationProcessor extends UniOperatorProcessor<T, T> {
private static class UniOnCancellationProcessor<T> extends UniOperatorProcessor<T, T> {

private final Runnable callback;

public UniOnCancellationProcessor(UniSubscriber<? super T> downstream) {
private volatile State state = State.INIT;
private static final AtomicReferenceFieldUpdater<UniOnCancellationProcessor, State> stateUpdater = AtomicReferenceFieldUpdater
.newUpdater(UniOnCancellationProcessor.class, State.class, "state");

public UniOnCancellationProcessor(Runnable callback, UniSubscriber<? super T> downstream) {
super(downstream);
this.callback = callback;
}

private final AtomicReference<State> state = new AtomicReference<>(State.INIT);

@Override
public void onItem(T item) {
if (state.compareAndSet(State.INIT, State.DONE)) {
if (stateUpdater.compareAndSet(this, State.INIT, State.DONE)) {
downstream.onItem(item);
}
}

@Override
public void onFailure(Throwable failure) {
if (state.compareAndSet(State.INIT, State.DONE)) {
if (stateUpdater.compareAndSet(this, State.INIT, State.DONE)) {
downstream.onFailure(failure);
}
}

@Override
public void cancel() {
if (state.compareAndSet(State.INIT, State.CANCELLED)) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
if (stateUpdater.compareAndSet(this, State.INIT, State.CANCELLED)) {
UniSubscription sub = getAndSetUpstreamSubscription(CANCELLED);
callback.run();
if (sub != null) {
sub.cancel();
Expand Down
25 changes: 15 additions & 10 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniOnCancellationCall.java
View file
Open in desktop
Expand Up @@ -3,7 +3,7 @@
import static io.smallrye.mutiny.helpers.EmptyUniSubscription.CANCELLED;
import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;

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

import io.smallrye.mutiny.Uni;
Expand All @@ -24,7 +24,7 @@ public UniOnCancellationCall(Uni<? extends I> upstream, Supplier<Uni<?>> supplie

@Override
public void subscribe(UniSubscriber<? super I> subscriber) {
AbstractUni.subscribe(upstream(), new UniOnCancellationCallProcessor(subscriber));
AbstractUni.subscribe(upstream(), new UniOnCancellationCallProcessor<I>(supplier, subscriber));
}

private enum State {
Expand All @@ -33,32 +33,37 @@ private enum State {
CANCELLED
}

private class UniOnCancellationCallProcessor extends UniOperatorProcessor<I, I> {
private static class UniOnCancellationCallProcessor<I> extends UniOperatorProcessor<I, I> {

public UniOnCancellationCallProcessor(UniSubscriber<? super I> downstream) {
private final Supplier<Uni<?>> supplier;

private volatile State state = State.INIT;
private static final AtomicReferenceFieldUpdater<UniOnCancellationCallProcessor, State> stateUpdater = AtomicReferenceFieldUpdater
.newUpdater(UniOnCancellationCallProcessor.class, State.class, "state");

public UniOnCancellationCallProcessor(Supplier<Uni<?>> supplier, UniSubscriber<? super I> downstream) {
super(downstream);
this.supplier = supplier;
}

private final AtomicReference<State> state = new AtomicReference<>(State.INIT);

@Override
public void onItem(I item) {
if (state.compareAndSet(State.INIT, State.DONE)) {
if (stateUpdater.compareAndSet(this, State.INIT, State.DONE)) {
downstream.onItem(item);
}
}

@Override
public void onFailure(Throwable failure) {
if (state.compareAndSet(State.INIT, State.DONE)) {
if (stateUpdater.compareAndSet(this, State.INIT, State.DONE)) {
downstream.onFailure(failure);
}
}

@Override
public void cancel() {
if (state.compareAndSet(State.INIT, State.CANCELLED)) {
UniSubscription sub = upstream.getAndSet(CANCELLED);
if (stateUpdater.compareAndSet(this, State.INIT, State.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
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
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
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
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 (compareAndSetUpstreamSubscription(null, subscription)) {
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 = getAndSetUpstreamSubscription(CANCELLED);
if (subscription != CANCELLED) {
downstream.onItem((O) item);
}
}

@Override
public void onFailure(Throwable failure) {
UniSubscription subscription = upstream.getAndSet(CANCELLED);
UniSubscription subscription = getAndSetUpstreamSubscription(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 = getAndSetUpstreamSubscription(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);
}

}
27 changes: 16 additions & 11 deletions implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniRetryAtMost.java
View file
Open in desktop
Expand Up @@ -3,7 +3,7 @@
import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;
import static io.smallrye.mutiny.helpers.ParameterValidation.positive;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.function.Predicate;

import io.smallrye.mutiny.Uni;
Expand All @@ -25,21 +25,26 @@ public UniRetryAtMost(Uni<T> upstream, Predicate<? super Throwable> predicate, l

@Override
public void subscribe(UniSubscriber<? super T> subscriber) {
AbstractUni.subscribe(upstream(), new UniRetryAtMostProcessor(subscriber));
AbstractUni.subscribe(upstream(), new UniRetryAtMostProcessor<T>(this, subscriber));
}

private class UniRetryAtMostProcessor extends UniOperatorProcessor<T, T> {
private static class UniRetryAtMostProcessor<T> extends UniOperatorProcessor<T, T> {

private final AtomicInteger counter = new AtomicInteger(0);
private final UniRetryAtMost<T> uniRetryAtMost;

public UniRetryAtMostProcessor(UniSubscriber<? super T> downstream) {
private volatile int counter = 0;
private static final AtomicIntegerFieldUpdater<UniRetryAtMostProcessor> counterUpdater = AtomicIntegerFieldUpdater
.newUpdater(UniRetryAtMostProcessor.class, "counter");

public UniRetryAtMostProcessor(UniRetryAtMost<T> uniRetryAtMost, UniSubscriber<? super T> downstream) {
super(downstream);
this.uniRetryAtMost = uniRetryAtMost;
}

@Override
public void onSubscribe(UniSubscription subscription) {
int count = counter.incrementAndGet();
if (upstream.compareAndSet(null, subscription)) {
int count = counterUpdater.incrementAndGet(this);
if (compareAndSetUpstreamSubscription(null, subscription)) {
if (count == 1) {
downstream.onSubscribe(this);
}
Expand All @@ -57,21 +62,21 @@ public void onFailure(Throwable failure) {
if (!testPredicate(failure)) {
return;
}
if (counter.get() > maxAttempts) {
if (counter > uniRetryAtMost.maxAttempts) {
downstream.onFailure(failure);
return;
}
UniSubscription previousSubscription = upstream.getAndSet(null);
UniSubscription previousSubscription = getAndSetUpstreamSubscription(null);
if (previousSubscription != null) {
previousSubscription.cancel();
}
AbstractUni.subscribe(upstream(), this);
AbstractUni.subscribe(uniRetryAtMost.upstream(), this);
}

private boolean testPredicate(Throwable failure) {
boolean passes;
try {
passes = predicate.test(failure);
passes = uniRetryAtMost.predicate.test(failure);
} catch (Throwable e) {
downstream.onFailure(e);
return false;
Expand Down