Upgrade to Pro — share decks privately, control downloads, hide ads and more …

DevoxxFR 2018: Spéléo Reactor

DevoxxFR 2018: Spéléo Reactor

Simon Baslé

April 18, 2018
Tweet

More Decks by Simon Baslé

Other Decks in Programming

Transcript

  1. • introduction / reactive streams • map • la fusion

    • concatMap • la “drain loop”
  2. Publisher Subscriber pousse des éléments produit consomme feedback “0 à

    N onNext suivi d’au plus 1 (onComplete ou onError)”
  3. grâce à elle, le Subscriber peut notifier de sa capacité

    de traitement via Subscription#request(long)
  4. Modèle dit “Asynchronous Push-Pull” Le Publisher envoie au max le

    nombre demandé, en asynchrone (au fil de l’eau)
  5. les onNext, ainsi que les événements terminaux, sont en série

    ie. pas d’appel parallèle à onNext pendant un onNext
  6. request & cancel peuvent se produire en parallèle (de onNext

    et d’eux même) req(2) req(7) req(1)
  7. Autres Concepts Importants souscription vers le haut, données vers le

    bas rien ne se passe avant `subscribe()` enchaîner les opérateurs
  8. (dans la plupart des cas) la source des données ne

    commence à émettre que lors de la souscription
  9. final class FluxMap<T, R> extends FluxOperator<T, R> { final Function<?

    super T, ? extends R> mapper; FluxMap(Flux<? extends T> source, Function<? super T, ? extends R> mapper) { super(source); this.mapper = Objects.requireNonNull(mapper, "mapper"); } @Override @SuppressWarnings("unchecked") public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new MapSubscriber<>(actual, mapper)); } }
  10. final class FluxMap<T, R> extends FluxOperator<T, R> { final Function<?

    super T, ? extends R> mapper; FluxMap(Flux<? extends T> source, Function<? super T, ? extends R> mapper) { super(source); this.mapper = Objects.requireNonNull(mapper, "mapper"); } @Override @SuppressWarnings("unchecked") public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new MapSubscriber<>(actual, mapper)); } }
  11. static final class MapSubscriber<T, R> implements InnerOperator<T, R> { final

    CoreSubscriber<? super R> actual; final Function<? super T, ? extends R> mapper; boolean done; Subscription s; MapSubscriber(CoreSubscriber<? super R> actual, Function<? super T, ? extends R> mapper) { this.actual = actual; this.mapper = mapper; } ...
  12. static final class MapSubscriber<T, R> implements InnerOperator<T, R> { final

    CoreSubscriber<? super R> actual; final Function<? super T, ? extends R> mapper; boolean done; Subscription s; MapSubscriber(CoreSubscriber<? super R> actual, Function<? super T, ? extends R> mapper) { this.actual = actual; this.mapper = mapper; }
  13. final CoreSubscriber<? super R> actual; Subscription s; ... @Override public

    void onSubscribe(Subscription s) { if (Operators.validate(this.s, s)) { this.s = s; actual.onSubscribe(this); } }
  14. final CoreSubscriber<? super R> actual; boolean done; ... @Override public

    void onComplete() { if (done) { return; } done = true; actual.onComplete(); }
  15. final CoreSubscriber<? super R> actual; boolean done; ... @Override public

    void onError(Throwable t) { if (done) { return; } done = true; actual.onError(t); }
  16. static final class MapSubscriber<T, R> implements InnerOperator<T, R> { ...

    Subscription s; ... @Override public void request(long n) { s.request(n); } @Override public void cancel() { s.cancel(); } }
  17. @Override public void onNext(T t) { if (done) { return;

    } R v; try { v = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null value."); } catch (Throwable e) { onError(e); return; } actual.onNext(v); }
  18. @Override public void onNext(T t) { if (done) { return;

    } R v; try { v = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null value."); } catch (Throwable e) { onError(e); return; } actual.onNext(v); }
  19. La Fusion map peut être fusionné comment la fusion est-elle

    implémentée ? quel impact sur la chaîne d’opérateurs ?
  20. final class FluxMapFuseable<T, R> extends FluxOperator<T, R> implements Fuseable {

    final Function<? super T, ? extends R> mapper; FluxMapFuseable(Flux<? extends T> source, Function<? super T, ? extends R> mapper) { super(source); this.mapper = Objects.requireNonNull(mapper, "mapper"); } @Override public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new MapFuseableSubscriber<>(actual, mapper)); }
  21. final class FluxMapFuseable<T, R> extends FluxOperator<T, R> implements Fuseable {

    final Function<? super T, ? extends R> mapper; FluxMapFuseable(Flux<? extends T> source, Function<? super T, ? extends R> mapper) { super(source); this.mapper = Objects.requireNonNull(mapper, "mapper"); } @Override public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new MapFuseableSubscriber<>(actual, mapper)); }
  22. static final class MapFuseableSubscriber<T, R> implements InnerOperator<T, R>, QueueSubscription<R> {

    final CoreSubscriber<? super R> actual; final Function<? super T, ? extends R> mapper; boolean done; QueueSubscription<T> s; int sourceMode; MapFuseableSubscriber(CoreSubscriber<? super R> actual, Function<? super T, ? extends R> mapper) { this.actual = actual; this.mapper = mapper; }
  23. static final class MapFuseableSubscriber<T, R> implements InnerOperator<T, R>, QueueSubscription<R> {

    final CoreSubscriber<? super R> actual; final Function<? super T, ? extends R> mapper; boolean done; QueueSubscription<T> s; int sourceMode; MapFuseableSubscriber(CoreSubscriber<? super R> actual, Function<? super T, ? extends R> mapper) { this.actual = actual; this.mapper = mapper; }
  24. @Override public int requestFusion(int requestedMode) { int m; if ((requestedMode

    & Fuseable.THREAD_BARRIER) != 0) { return Fuseable.NONE; } else { m = s.requestFusion(requestedMode); } sourceMode = m; return m; }
  25. @Override public int requestFusion(int requestedMode) { int m; if ((requestedMode

    & Fuseable.THREAD_BARRIER) != 0) { return Fuseable.NONE; } else { m = s.requestFusion(requestedMode); } sourceMode = m; return m; }
  26. @Override public void onNext(T t) { if (sourceMode == ASYNC)

    { actual.onNext(null); } else { if (done) { return; } R v; try { v = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null value."); } catch (Throwable e) { onError(e); return;
  27. @Override public void onNext(T t) { if (sourceMode == ASYNC)

    { actual.onNext(null); } else { if (done) { return; } R v; try { v = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null value."); } catch (Throwable e) { onError(e); return;
  28. @Override @Nullable public R poll() { T v = s.poll();

    if (v != null) { //can throw exceptions return Objects.requireNonNull(mapper.apply(v)); } return null; }
  29. @Override @Nullable public R poll() { T v = s.poll();

    if (v != null) { //can throw exceptions return Objects.requireNonNull(mapper.apply(v)); } return null; }
  30. @Override public boolean isEmpty() { return s.isEmpty(); } @Override public

    void clear() { s.clear(); } @Override public int size() { return s.size(); }
  31. final class FluxConcatMap<T, R> extends FluxOperator<T, R> { final Function<?

    super T, ? extends Publisher<? extends R>> mapper; final Supplier<? extends Queue<T>> queueSupplier; final int prefetch; FluxConcatMap(Flux<? extends T> source, Function<? super T, ? extends Publisher<? extends R>> mapper, Supplier<? extends Queue<T>> queueSupplier, int prefetch) { super(source); if (prefetch <= 0) throw new IllegalArgumentException("prefetch <= 0"); this.mapper = Objects.requireNonNull(mapper, "mapper"); this.queueSupplier = Objects.requireNonNull(queueSupplier); this.prefetch = prefetch; } @Override public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new ConcatMapImmediate<>(s, mapper, queueSupplier, prefetch)); }
  32. final class FluxConcatMap<T, R> extends FluxOperator<T, R> { final Function<?

    super T, ? extends Publisher<? extends R>> mapper; final Supplier<? extends Queue<T>> queueSupplier; final int prefetch; FluxConcatMap(Flux<? extends T> source, Function<? super T, ? extends Publisher<? extends R>> mapper, Supplier<? extends Queue<T>> queueSupplier, int prefetch) { super(source); if (prefetch <= 0) throw new IllegalArgumentException("prefetch <= 0"); this.mapper = Objects.requireNonNull(mapper, "mapper"); this.queueSupplier = Objects.requireNonNull(queueSupplier); this.prefetch = prefetch; } @Override public void subscribe(CoreSubscriber<? super R> actual) { source.subscribe(new ConcatMapImmediate<>(s, mapper, queueSupplier, prefetch)); }
  33. static final class ConcatMapImmediate<T, R> { final CoreSubscriber<? super R>

    actual; final Function<? super T, ? extends Publisher<? extends R>> mapper; final Supplier<? extends Queue<T>> queueSupplier; final int prefetch; final int limit; Subscription s; int consumed; volatile Queue<T> queue; volatile boolean done; volatile boolean cancelled; volatile boolean active; volatile Throwable error; volatile int wip; volatile int guard; int sourceMode;
  34. static final class ConcatMapImmediate<T, R> { final CoreSubscriber<? super R>

    actual; final Function<? super T, ? extends Publisher<? extends R>> mapper; final Supplier<? extends Queue<T>> queueSupplier; final int prefetch; final int limit; Subscription s; int consumed; volatile Queue<T> queue; volatile boolean done; volatile boolean cancelled; volatile boolean active; volatile Throwable error; volatile int wip; volatile int guard; int sourceMode;
  35. static final AtomicReferenceFieldUpdater<ConcatMapImmediate, Throwable> ERROR = AtomicReferenceFieldUpdater.newUpdater(ConcatMapImmediate.class, Throwable.class, "error"); static

    final AtomicIntegerFieldUpdater<ConcatMapImmediate> WIP = AtomicIntegerFieldUpdater.newUpdater(ConcatMapImmediate.class,"wip"); static final AtomicIntegerFieldUpdater<ConcatMapImmediate> GUARD = AtomicIntegerFieldUpdater.newUpdater(ConcatMapImmediate.class, "guard"); final ConcatMapInner<R> inner;
  36. @Override public void onSubscribe(Subscription s) { if (Operators.validate(this.s, s)) {

    this.s = s; if (s instanceof Fuseable.QueueSubscription) { if (m == Fuseable.SYNC) { ... } else if (m == Fuseable.ASYNC) { ... } else { queue = queueSupplier.get(); } } else { queue = queueSupplier.get(); } actual.onSubscribe(this); s.request(Operators.unboundedOrPrefetch(prefetch)); } }
  37. ... if (s instanceof Fuseable.QueueSubscription) { Fuseable.QueueSubscription<T> f = (Fuseable.QueueSubscription<T>)

    s; int m = f.requestFusion(Fuseable.ANY); if (m == Fuseable.SYNC) { sourceMode = Fuseable.SYNC; queue = f; done = true; actual.onSubscribe(this); drain(); return; } else if (m == Fuseable.ASYNC) { sourceMode = Fuseable.ASYNC; queue = f; } else { queue = queueSupplier.get(); } } ...
  38. ... if (s instanceof Fuseable.QueueSubscription) { Fuseable.QueueSubscription<T> f = (Fuseable.QueueSubscription<T>)

    s; int m = f.requestFusion(Fuseable.ANY); if (m == Fuseable.SYNC) { sourceMode = Fuseable.SYNC; queue = f; done = true; actual.onSubscribe(this); drain(); return; } else if (m == Fuseable.ASYNC) { sourceMode = Fuseable.ASYNC; queue = f; } else { queue = queueSupplier.get(); } } ... drain( )? on va y revenir...
  39. @Override public void onComplete() { done = true; drain(); }

    @Override public void request(long n) { inner.request(n); } @Override public void cancel() { if (!cancelled) { cancelled = true; inner.cancel(); s.cancel(); } }
  40. @Override public void onError(Throwable t) { if (Exceptions.addThrowable(ERROR, this, t))

    { inner.cancel(); if (GUARD.getAndIncrement(this) == 0) { t = Exceptions.terminate(ERROR, this); if (t != TERMINATED) { actual.onError(t); } } } else { Operators.onErrorDropped(t, actual.currentContext()); } }
  41. @Override public void onError(Throwable t) { if (Exceptions.addThrowable(ERROR, this, t))

    { inner.cancel(); if (GUARD.getAndIncrement(this) == 0) { t = Exceptions.terminate(ERROR, this); if (t != TERMINATED) { actual.onError(t); } } } else { Operators.onErrorDropped(t, actual.currentContext()); } }
  42. @Override public void onNext(T t) { if (sourceMode == Fuseable.ASYNC)

    { drain(); } else if (!queue.offer(t)) { onError(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL); } else { drain(); } }
  43. @Override public void onNext(T t) { if (sourceMode == Fuseable.ASYNC)

    { drain(); } else if (!queue.offer(t)) { onError(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL); } else { drain(); } } on va y revenir je vous dis...
  44. pour l’instant, précisons juste que dans le drain( ) on

    va attacher le inner au Publisher créé par la fonction
  45. @Override public void onNext(T t) { if (sourceMode == Fuseable.ASYNC)

    { drain(); } else if (!queue.offer(t)) { onError(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL); } else { drain(); } }
  46. @Override public void innerNext(R value) { if (guard == 0

    && GUARD.compareAndSet(this, 0, 1)) { actual.onNext(value); if (GUARD.compareAndSet(this, 1, 0)) { return; } Throwable e = Exceptions.terminate(ERROR, this); if (e != TERMINATED) { actual.onError(e); } } }
  47. @Override public void innerError(Throwable e) { if (Exceptions.addThrowable(ERROR, this, e))

    { s.cancel(); if (GUARD.getAndIncrement(this) == 0) { e = Exceptions.terminate(ERROR, this); if (e != TERMINATED) { actual.onError(e); } } } else { Operators.onErrorDropped(e, actual.currentContext()); } }
  48. static final class ConcatMapInner<R> extends Operators.MultiSubscriptionSubscriber<R, R> { final FluxConcatMapImmediate<?,

    R> parent; long produced; ConcatMapInner(FluxConcatMapSupport<?, R> parent) { super(Operators.emptySubscriber()); this.parent = parent; } @Override public void onNext(R t) { produced++; parent.innerNext(t); } ...
  49. static final class ConcatMapInner<R> extends Operators.MultiSubscriptionSubscriber<R, R> { final FluxConcatMapImmediate<?,

    R> parent; long produced; ConcatMapInner(FluxConcatMapSupport<?, R> parent) { super(Operators.emptySubscriber()); this.parent = parent; } @Override public void onNext(R t) { produced++; parent.innerNext(t); } ...
  50. static final class ConcatMapInner<R> extends Operators.MultiSubscriptionSubscriber<R, R> { ... @Override

    public void onError(Throwable t) { long p = produced; if (p != 0L) { produced = 0L; produced(p); } parent.innerError(t); } ... }
  51. static final class ConcatMapInner<R> extends Operators.MultiSubscriptionSubscriber<R, R> { ... @Override

    public void onComplete() { long p = produced; if (p != 0L) { produced = 0L; produced(p); } parent.innerComplete(); } } }
  52. void drain() { if (WIP.getAndIncrement(this) == 0) { for (;

    ; ) { if (cancelled) return; if (!active) { ... } if (WIP.decrementAndGet(this) == 0) { break; } } } } }
  53. void drain() { if (WIP.getAndIncrement(this) == 0) { for (;

    ; ) { if (cancelled) return; if (!active) { ... } if (WIP.decrementAndGet(this) == 0) { break; } } } } }
  54. if (cancelled) return; if (!active) { boolean d = done;

    T v; try { v = queue.poll(); } catch (Throwable e) { actual.onError(e); return; } boolean empty = v == null; if (d && empty) { actual.onComplete(); return; } ...
  55. ... if (!empty) { Publisher<? extends R> p; try {

    p = Objects.requireNonNull(mapper.apply(v)); } catch (Throwable e) { actual.onError(e); return; } if (sourceMode != Fuseable.SYNC) { ... } active = true; p.subscribe(inner); } } if (WIP.decrementAndGet(this) == 0) { break; }
  56. ... if (sourceMode != Fuseable.SYNC) { int c = consumed

    + 1; if (c == limit) { consumed = 0; s.request(c); } else { consumed = c; } } active = true; p.subscribe(inner); } } if (WIP.decrementAndGet(this) == 0) { break; }
  57. void drain() { if (WIP.getAndIncrement(this) == 0) { for (;

    ; ) { if (cancelled) return; if (!active) { ... } if (WIP.decrementAndGet(this) == 0) { break; } } } } }
  58. thread 1 thread 2 thread 3 sans work stealing onNext(3)

    empile 3 request( 1 ) émet 1 2, 3
  59. request( 2 ) thread 1 thread 2 thread 3 sans

    work stealing émet 2, 3 onNext(3) empile 3 request( 1 ) émet 1
  60. thread 1 thread 2 thread 3 sans work stealing onNext(3)

    empile 3 bloqué draine bloqué draine
  61. thread 1 thread 2 thread 3 avec work stealing onNext(3)

    empile 3 request( 1 ) émet 1 2, 3
  62. request( 2 ) thread 1 thread 2 thread 3 avec

    work stealing onNext(3) empile 3 request( 1 ) émet 1 émet 2, 3
  63. request( 2 ) thread 1 thread 2 thread 3 avec

    work stealing onNext(3) empile 3 request( 1 ) émet 1 émet 2, 3 met juste la requête à jour et quitte la boucle WIP cette drainLoop a “gagné”