Flight of the Flux

A look at Reactor's execution model

Transcript

1. Flight of the Flux a look at Reactor's execution model
2. Flight of the Flux a look at Reactor's execution model @simonbasle
3. What we'll learn
5. Nothing happens until you subscribe, Assembly time vs Execution time
6. Nothing happens until you subscribe, Assembly time vs Execution time Cold or Hot?
7. Nothing happens until you subscribe, Assembly time vs Execution time Cold or Hot? Concurrent agnostic
8. Cold or Hot? Concurrent agnostic Schedulers, PublishOn vs SubscribeOn Nothing happens until you subscribe, Assembly time vs Execution time
9. Cold or Hot? Concurrent agnostic Schedulers, PublishOn vs SubscribeOn Work Stealing Nothing happens until you subscribe, Assembly time vs Execution time
10. Cold or Hot? Concurrent agnostic Schedulers, PublishOn vs SubscribeOn Work Stealing Operator Fusion Nothing happens until you subscribe, Assembly time vs Execution time
11. Cold or Hot? Concurrent agnostic Schedulers, PublishOn vs SubscribeOn Work Stealing Operator Fusion Nothing happens until you subscribe, Assembly time vs Execution time
13. Publisher Subscriber
15. ...
16. ...
17. ...
18. Subscription
19. Nothing Happens until you subscribe Assembly time
20. Nothing happen until you subscribe
21. Assembly time
22. this.myFlux = Flux.just("foo") .map(String::length); MyTest#init()
23. Execution time
24. Execution time Subscription
25. this.myFlux .subscribe(); MyTest#test()
26. what would I see in a stack trace?
27. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx)
28. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx) ACCIO INTERESTING PARTS!
29. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx) ACCIO INTERESTING PARTS!
30. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx)
31. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx)
32. java.lang.ArithmeticException: / by zero at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:115) at reactor.core.publisher.FluxJust$WeakScalarSubscription.request(FluxJust.java:99) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:156) at reactor.core.publisher.BlockingSingleSubscriber.onSubscribe(BlockingSingleSubscriber.java:49) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90) at reactor.core.publisher.FluxJust.subscribe(FluxJust.java:70) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63) at reactor.core.publisher.Flux.subscribe(Flux.java:6873) at mycode.test(MyTest.java:xx)
33. checkpoint()
34. Hot or Cold ?
35. Cold publisher
43. always true... until it's false " Nothing happen until you subscribe "
44. Hot publisher
47. Concurrent Agnostic
48. do I have to accept a particular execution model?
49. No, we provide tools to fine tune
50. But, frameworks & libraries built on top might be opinionated
51. what about time operators?
52. ...they do need to change threads
53. Schedulers & how to switch threads
54. Scheduler abstraction
55. , part part
56. publishOn
57. Flux.op1.op2.publishOn.op3.op4 .subscribe
58. 1 2 Flux.op1.op2.publishOn.op3.op4 .subscribe
59. Thread calling subscribe() is where data flows initially 1
60. After publishOn data flows in Thread 2 2
61. subscribeOn
62. Flux.op1.op2 .subscribeOn .op3.op4 .subscribe
63. 2 Flux.op1.op2 .subscribeOn .op3.op4 .subscribe 1
64. subscribeOn() changes where sequence subscription happens... 1
65. ... which is also where data flows initially 2
66. and the two together?
67. Flux .op1 .subscribeOn .op2 .op4 .subscribe( v -> doSomething(v)) .publishOn
68. Flux .op1 .subscribeOn .op2 .op4 .subscribe( v -> doSomething(v)) .publishOn
69. Flux .op1 .subscribeOn .op2 .op4 .subscribe( v -> doSomething(v)) .publishOn 1 2 3 4
70. Work Stealing
71. so where do (most) operators run, again?
72. publishOn
73. what is work stealing?
74. 1 operator that combines data coming from 2 threads
75. shared queue, drain loop
76. offer
77. offer work
78. offer offer work
79. offer offer work steal exit
80. offer offer work steal exit
81. offer offer work steal exit work
82. offer offer work steal exit work exit
83. Operator Fusion
84. macro fusion
85. a.then(b).then(c).then(d)
86. a.then(b,c,d)
87. micro fusion
88. a.map(->).map(->).filter(->)
89. a.map(->).map(->).filter(->)
90. a.map(->).map(->).filter(->) bounded queues (for backpressure)
91. a.map(->).map(->).filter(->)
92. a.map(->).map(->).filter(->) single transverse bounded queue
94. Questions?
95. Thanks!
96. • geese flight: CC0 (via pxhere) • tools: CC0 via (pxhere) • harry potter: © Warner Bros (fair use) • badly drawn ron: CC0 (via pixabay) • fire and ice cubes: CC0 (via pxhere) • blank canvas: CC0 (via publicdomainpictures) • rail switch: CC0 (via pxhere) • pickpocket: CC-By by Wellcome Library, London (via wikimedia commons) • mask icon: CC-BySa by Angelus (via wikimedia commons) • metal fusion: CC-BySa by P Sakthy (via wikimedia commons) • The End sand: CC0 by ElektroPlan (via pixabay) Image Credits