GeekOut 2019: Don’t be Homer Simpson with your Reactor!

Don’t be Homer Simpson   with your Reactor! Sergei Egorov,
Pivotal @bsideup https://simpsonswiki.com/wiki/File:Springﬁeld_Nuclear_Power_Plant_6.png

About me • Staﬀ Engineer at Pivotal’s Spring R&D, working
on Project Reactor ⚛ • Berlin Spring User Group co-organizer • Testcontainers co-maintainer • Developer tools geek • … although spent most of my career building  highly scalable backend services @bsideup

My ﬁrst experience with  reactive programming was like…

@bsideup

Mono.fromRunnable()

Mono.fromRunnable() @bsideup

Mono.fromRunnable() My code @bsideup

Mono.fromRunnable() My code My code @bsideup

Mono.fromRunnable() My code My code My code Work! @bsideup

Mono.fromRunnable().subscribe() @bsideup

1 week later…

I agree with Heinrich Apfelmus that the essence of functional
reactive programming is to specify the dynamic behavior of a value completely at the time of declaration. “Reactive Sir” @bsideup

1 week of production later…

@bsideup

What could go wrong

Exceptions

// com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup

// com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); java.lang.IndexOutOfBoundsException: Source emitted
more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) @bsideup

more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? @bsideup

more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? Where is “com.example.demo” package? @bsideup

more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(FluxRange.java:129) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:107) at reactor.core.publisher.MonoSingle$SingleSubscriber.request(MonoSingle.java:94) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.trySchedule(MonoSubscribeOn.java:186) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.onSubscribe(MonoSubscribeOn.java:131) at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(MonoSingle.java:114) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:123) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Which source? Where is “com.example.demo” package? Why they closed Google Inbox? @bsideup

Assembly vs Subscription

Flux.range(0, 5) Assembly @bsideup

Flux.range(0, 5) .filter(it -> it > 3) Assembly @bsideup

Flux.range(0, 5) .filter(it -> it > 3) .map(it -> it.toString())
Assembly @bsideup

.single() Assembly @bsideup

.single() .subscribeOn(Schedulers.parallel()) Assembly @bsideup

.single() .subscribeOn(Schedulers.parallel()) .subscribe() Assembly @bsideup

FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 @bsideup

FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 *
= subscriber @bsideup

FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .onSubscribe(
) * = subscription @bsideup

FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Subscription current thread parallel-1 .request(max)
@bsideup

FluxRange FluxFilter FluxMap MonoSingle MonoSubscribeOn Execution current thread parallel-1 .onNext(0)
@bsideup

@bsideup

java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(Mono at reactor.core.publisher.FluxRange$RangeSubscription.fastPath(Flux at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxR at reactor.core.publisher.MonoSingle$SingleSubscriber.request(Mono at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. at reactor.core.publisher.MonoSingle$SingleSubscriber.onSubscribe(M at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68) at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58) at reactor.core.publisher.Mono.subscribe(Mono.java:3711) at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber. @bsideup

Demo http://eskipaper.com/homer-jay-simpson-cartoon.html

How ReactorDebugAgent works? @bsideup

// com/example/demo/Example.java Flux.range(0, 5) .single() .subscribeOn(Schedulers.parallel()) .subscribe(); @bsideup

// com/example/demo/Example.java Flux.range(0, 5) .checkpoint("Example.java:16") .single() .checkpoint("Example.java:17") .subscribeOn(Schedulers.parallel()) .checkpoint(“Example.java:18") .subscribe();
@bsideup

Demo outcomes • Use .checkpoint(“something”) to “mark” reactive “milestones” •
Read about Hooks.onOperatorDebug()… • … but use reactor-tools’ ReactorDebugAgent (works in prod too) • https://spring.io/blog/2019/03/06/ﬂight-of-the-ﬂux-1-assembly-vs- subscription - great article from Simon Basle about the internals @bsideup

Blocking calls! https://blog.tfd.co.uk/2010/10/15/jackrabbit-performance/ @bsideup

How Reactor schedules tasks @bsideup

Default thread pools Schedulers.parallel() - N threads, where N matches
the CPUs count. Schedulers.single() - 1 thread handling all submitted tasks Schedulers.elastic() - dynamic, thread caching pool @bsideup

Mono.delay(ofSeconds(1)) @bsideup

Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) @bsideup

Mono.delay(ofSeconds(1)) Mono.delay(ofSeconds(1), Schedulers.parallel()) Every non-instant operation   runs on the
parallel scheduler  by default! @bsideup

parallel-1 parallel-2 parallel-3 parallel-4 Non-blocking execution @bsideup

parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution @bsideup

parallel-1 parallel-2 parallel-3 parallel-4 Blocking execution ⚠ no more tasks
scheduled   until this task returns @bsideup

How to ﬁx?

Project Loom

@bsideup

Use non-blocking APIs, or…

parallel-1 parallel-2 parallel-3 parallel-4 Custom thread pool! custom-1 custom-2 @bsideup

Demo http://eskipaper.com/homer-jay-simpson-cartoon.html

How BlockHound works? @bsideup

public class Thread implements Runnable { public static native void
sleep(long millis); } @bsideup

public class Thread implements Runnable { private static native void
$$BlockHound$$_sleep(long millis); public static void sleep(long millis) { $$BlockHound$$_sleep(millis); } } @bsideup

public class Thread implements Runnable { private static native void
$$BlockHound$$_sleep(long millis); public static void sleep(long millis) { reactor.BlockHoundRuntime.checkBlocking( "java.lang.Thread", "sleep", /*method modifiers*/ ); $$BlockHound$$_sleep(millis); } } @bsideup

Blocking check static void checkBlocking(String className, String methodName, int modifiers)
{ if (Boolean.FALSE == IS_ALLOWED.get()) { // Report } } static final ThreadLocal<Boolean> IS_ALLOWED = ThreadLocal.withInitial(() -> { if (threadPredicate.test(Thread.currentThread())) { return false; } else { // Optimization: use Three-state (true, false, null) // where `null` is `not non-blocking` return null; } }); @bsideup

Allowed/disallowed @bsideup

Allowed/disallowed class ClassLoader { // ... public Class<?> loadClass(String name)
{ return loadClass(name, false); } } @bsideup

{ return loadClass(name, false); } } @bsideup Blocking!

{ Boolean previous = BlockHoundRuntime.IS_ALLOWED.get(); BlockHoundRuntime.IS_ALLOWED.set( previous != null ? true : null ); try { // Original call return loadClass(name, false); } finally { BlockHoundRuntime.IS_ALLOWED.set(previous); } } } @bsideup

Demo outcomes • Blocking calls are bad, m’kay? • They
may sneak into your production system • Use https://github.com/reactor/BlockHound to detect them • Supports multiple frameworks (Reactor, RxJava, etc) • … and maybe even Kotlin:   https://github.com/Kotlin/kotlinx.coroutines/issues/1031 • Use a dedicated pool for the necessary blocking calls, or schedule them on the Schedulers.elastic() built-in pool if they happen rarely @bsideup

“I use async APIs and I am safe!”

Yeah… sure.

Are you sure?

KafkaProducer#send(ProducerRecord,Callback) @bsideup

KafkaProducer#send(ProducerRecord,Callback) “Asynchronously send a record to a topic and invoke
the provided callback when the send has been acknowledged.” - Javadoc @bsideup

the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) @bsideup

the provided callback when the send has been acknowledged.” - Javadoc java.lang.Error: Blocking call! java.lang.Object#wait at reactor.BlockHound$Builder.lambda$new$0(BlockHound.java:154) at reactor.BlockHound$Builder.lambda$install$8(BlockHound.java:254) at reactor.BlockHoundRuntime.checkBlocking(BlockHoundRuntime.java:43) at java.lang.Object.wait(Object.java) at org.apache.kafka.clients.Metadata.awaitUpdate(Metadata.java:181) at org.apache.kafka.clients.producer.KafkaProducer.waitOnMetadata(KafkaProducer.java:938) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:823) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) https://issues.apache.org/jira/browse/KAFKA-3539 @bsideup

long remainingWaitMs = maxWaitMs; long elapsed; // Issue metadata requests
until we have metadata for the topic or maxWaitTimeMs is exceeded. // In case we already have cached metadata for the topic, but the requested partition is greater // than expected, issue an update request only once. This is necessary in case the metadata // is stale and the number of partitions for this topic has increased in the meantime. do { log.trace("Requesting metadata update for topic {}.", topic); metadata.add(topic); int version = metadata.requestUpdate(); sender.wakeup(); try { metadata.awaitUpdate(version, remainingWaitMs); } catch (TimeoutException ex) { // Rethrow with original maxWaitMs to prevent logging exception with remainingWaitMs throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); } cluster = metadata.fetch(); elapsed = time.milliseconds() - begin; if (elapsed >= maxWaitMs) throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms."); if (cluster.unauthorizedTopics().contains(topic)) throw new TopicAuthorizationException(topic); remainingWaitMs = maxWaitMs - elapsed; partitionsCount = cluster.partitionCountForTopic(topic); } while (partitionsCount == null); waitOnMetadata(record.topic(), record.partition(), maxBlockTimeMs); Default is “60 seconds” @bsideup

@bsideup

How to “instrument” Reactor

Schedulers.onScheduleHook(fn) @bsideup

Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("myHook", runnable -> { println("Before every scheduled runnable");
return () -> { println("Before execution"); runnable.run(); println("After execution"); }; });

Schedulers.onScheduleHook(fn) @bsideup Schedulers.onScheduleHook("mdc", runnable -> { String userId = MDC.get("userId");
return () -> { MDC.put("userId", userId); try { runnable.run(); } finally { MDC.remove("userId"); } }; });

Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .subscribe();

Hooks.onLastOperator(fn) @bsideup Flux.range(0, 5) .map(it -> it.toString()) .single() .subscribeOn(Schedulers.parallel()) .transform(fn)
.subscribe();

Your Reactor should be like:

https://9gag.com/gag/aqKWPwY

Takeaways • io.projectreactor:reactor-tools:3.3.0.RELEASE • Flux#checkpoint() / Flux#log() • Hooks class
• https://github.com/reactor/BlockHound • Async APIs - trust, but verify @bsideup

@bsideup bsideup

GeekOut 2019: Don’t be Homer Simpson with your ...

GeekOut 2019: Don’t be Homer Simpson with your Reactor!

Transcript

More Decks by Sergei Egorov

Other Decks in Programming

Featured