reactive cloud-native networking...built by leaders in microservices and cloud computing • open...
TRANSCRIPT
• Arsalan Farooq CEO• Founder/CEO of Convirture
(acquired by Accelerite)• Founded Oracle’s ASLM
division and grew it to a multi-national organization
• Robert Roeser CINO• Netflix Edge Platform team• RSocket core contributor• Principal Architect at Nike
on realtime analytics system serving 30M+ users
Speakers
Built by leaders in microservices and cloud computing
• Open Source Layer 5/6 communication protocol
• Reactive streams semantics
• Application-level flow control
• Binary Encoding
• Asynchronous Message-Passing
What’s needed for high performance
Microservice Networking?
• Temporal Decoupling
• Spacial Decoupling
• Binary
• Application-Flow control
�5
High Performance Microservices
Spacial Decoupling
A message’s sender should not directly call a destination
The execution time processing a call should not affect the caller
Loose Coupling
Temporal Decoupling
RSocket is loosely coupled using message-passing.
What’s the difference between Message-Passing and Event-Driven?
�9
Message-Passing vs Event-Driven
Message-Passing Event-Driven
A message is a payload sent to a specific destination An event is signal emitted by a component reaching a specific stage
Message-driven focuses on addressable recipients Event drive focuses on listen for events
Errors passed as messages - can be easy sent to caller
Error handling in Event driven very complex - think dead letter queues
Bi-directional communication is easy Bi-directional communication is very hard in Event-driven
�10
Event-Driven
Emits Events Listens for Events
Queue
�11
Event-Driven
Emits Events Listens for Events
Queue
�12
Event-Driven
Emits Events Listens for Events
Queue
�13
Event-Driven
Emits Events Listens for Events
Queue
�14
Event-Driven
Emits Events Listens for Events
Queue
�15
Event-Driven
Emits Events Listens for Events
Queue
response???
�16
Event-Driven
Emits Events Listens for Events
Queue
exception???
�17
Event-Driven
Emits Events Listens for Events
Queue
???backpressure
�18
Message-Passing
Dest 1
Dest 2
Dest 3
Dest 4
Dest 5
Dest 6
�19
Message-Passing
Dest 1
Dest 2
Dest 3
Dest 4
Dest 5
Dest 6
�20
Message-Passing
Dest 1
Dest 2
Dest 3
Dest 4
Dest 5
Dest 6
�21
Message-Passing
Dest 1
Dest 2
Dest 3
Dest 4
Dest 5
Dest 6
�22
RSocket - Message-Passing
Dest 1 Dest 2
�23
RSocket - Message-Passing
Dest 1 Dest 2requestChannel
�24
RSocket - Message-Passing
Dest 1 Dest 2
request(3)
requestChannel
�25
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
�26
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload
�27
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload Payload
�28
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload Payloadrequest(2)
�29
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload Payloadrequest(2)
complete
�30
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload Payloadrequest(2)
�31
RSocket - Message-Passing
Dest 1 Dest 2
Payload
request(3)
requestChannel
Payload Payload
exception
request(2)
Spacial Decoupling
RSocket messages are binary frames over dedicated streams RSocket’s APIs are non-blocking
RSocket - Loose Coupling
Temporal Decoupling
What type of Messages does RSocket use?
Binary Payloads
Efficiently encoded binary payloads
Resumable lens over bytes to access messages
RSocket’s Messages
Flyweight Pattern
How does RSocket Process Messages?
It’s easy to get processing wrong.
Key Insightpublic class Counter1 { private int count;
private int count() { //count = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) { count++; } return count; }
public static void main(String... args) { final Counter1 counter1 = new Counter1();
final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); } histogram .getIntervalHistogram() .outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Key Insightpublic class Counter1 { private int count;
private int count() { //count = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) { count++; } return count; }
public static void main(String... args) { final Counter1 counter1 = new Counter1();
final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); } histogram .getIntervalHistogram() .outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Single
Count Counter Main
Key Insightpublic class Counter1 { private int count;
private int count() { //count = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) { count++; } return count; }
public static void main(String... args) { final Counter1 counter1 = new Counter1();
final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); } histogram .getIntervalHistogram() .outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Takes about 1 microsecond.
Single
Count Counter Main
Key Insightpublic class Counter2 { private static AtomicIntegerFieldUpdater<Counter2> COUNT = AtomicIntegerFieldUpdater.newUpdater(Counter2.class, "count"); private volatile int count;
private int count() throws Exception { int target = Integer.MAX_VALUE; count = 0; ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) { executor.execute(() -> { while (COUNT.incrementAndGet(Counter2.this) < target) Thread.yield(); }); }
while (COUNT.get(Counter2.this) < target) Thread.yield(); return count; }
public static void main(String... args) throws Exception { final Counter2 counter = new Counter2(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Key Insightpublic class Counter2 { private static AtomicIntegerFieldUpdater<Counter2> COUNT = AtomicIntegerFieldUpdater.newUpdater(Counter2.class, "count"); private volatile int count;
private int count() throws Exception { int target = Integer.MAX_VALUE; count = 0; ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) { executor.execute(() -> { while (COUNT.incrementAndGet(Counter2.this) < target) Thread.yield(); }); }
while (COUNT.get(Counter2.this) < target) Thread.yield(); return count; }
public static void main(String... args) throws Exception { final Counter2 counter = new Counter2(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Multi-threaded
Count
Counter
MainCounter
Counter
Key Insightpublic class Counter2 { private static AtomicIntegerFieldUpdater<Counter2> COUNT = AtomicIntegerFieldUpdater.newUpdater(Counter2.class, "count"); private volatile int count;
private int count() throws Exception { int target = Integer.MAX_VALUE; count = 0; ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) { executor.execute(() -> { while (COUNT.incrementAndGet(Counter2.this) < target) Thread.yield(); }); }
while (COUNT.get(Counter2.this) < target) Thread.yield(); return count; }
public static void main(String... args) throws Exception { final Counter2 counter = new Counter2(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Didn’t wait for it to finish.
Multi-threaded
Count
Counter
MainCounter
Counter
Key Insightpublic class Counter3 { private static AtomicIntegerFieldUpdater<Counter3> WIP = AtomicIntegerFieldUpdater.newUpdater(Counter3.class, "wip");
private volatile int wip; private volatile int count;
private int count() throws Exception { count = 0; int parties = Runtime.getRuntime().availableProcessors(); for (int i = 0; i < parties; i++) { Thread thread = new Thread( () -> { while (count < Integer.MAX_VALUE) { // Use work in progress flag to let only one thread count at a time if (WIP.compareAndSet(Counter3.this, 0, 1)) { // Update the volatile variable count = doCount(); // No more work - set WIP back to zero WIP.set(Counter3.this, 0); } else Thread.yield(); } }); thread.start(); } while (count < Integer.MAX_VALUE) Thread.yield(); return count; }
private int doCount() { int c = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) c++; return c; }
public static void main(String... args) throws Exception { final Counter3 counter1 = new Counter3(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Key Insightpublic class Counter3 { private static AtomicIntegerFieldUpdater<Counter3> WIP = AtomicIntegerFieldUpdater.newUpdater(Counter3.class, "wip");
private volatile int wip; private volatile int count;
private int count() throws Exception { count = 0; int parties = Runtime.getRuntime().availableProcessors(); for (int i = 0; i < parties; i++) { Thread thread = new Thread( () -> { while (count < Integer.MAX_VALUE) { // Use work in progress flag to let only one thread count at a time if (WIP.compareAndSet(Counter3.this, 0, 1)) { // Update the volatile variable count = doCount(); // No more work - set WIP back to zero WIP.set(Counter3.this, 0); } else Thread.yield(); } }); thread.start(); } while (count < Integer.MAX_VALUE) Thread.yield(); return count; }
private int doCount() { int c = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) c++; return c; }
public static void main(String... args) throws Exception { final Counter3 counter1 = new Counter3(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Drain Loop
Count Counter
Main
CounterCount
… …
Key Insight
Less than a millisecond.
public class Counter3 { private static AtomicIntegerFieldUpdater<Counter3> WIP = AtomicIntegerFieldUpdater.newUpdater(Counter3.class, "wip");
private volatile int wip; private volatile int count;
private int count() throws Exception { count = 0; int parties = Runtime.getRuntime().availableProcessors(); for (int i = 0; i < parties; i++) { Thread thread = new Thread( () -> { while (count < Integer.MAX_VALUE) { // Use work in progress flag to let only one thread count at a time if (WIP.compareAndSet(Counter3.this, 0, 1)) { // Update the volatile variable count = doCount(); // No more work - set WIP back to zero WIP.set(Counter3.this, 0); } else Thread.yield(); } }); thread.start(); } while (count < Integer.MAX_VALUE) Thread.yield(); return count; }
private int doCount() { int c = 0; for (int i = 0; i < Integer.MAX_VALUE; i++) c++; return c; }
public static void main(String... args) throws Exception { final Counter3 counter1 = new Counter3(); final Recorder histogram = new Recorder(3600000000000L, 3);
for (int i = 0; i < 100; i++) { long start = System.nanoTime(); counter1.count(); long stop = System.nanoTime(); histogram.recordValue(stop - start); }
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Drain Loop
Count Counter
Main
CounterCount
… …
Key Insightpublic class Counter4 {
private volatile int count;
private Mono<Integer> count() { return Mono.fromSupplier( () -> { int c = count; for (int i = 0; i < Integer.MAX_VALUE; i++) { c++; } count = c; return c; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Key Insightpublic class Counter4 {
private volatile int count;
private Mono<Integer> count() { return Mono.fromSupplier( () -> { int c = count; for (int i = 0; i < Integer.MAX_VALUE; i++) { c++; } count = c; return c; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Drain Loop
Count Counter
Main
CounterCount
… …
Key Insight
Less than a millisecond.
public class Counter4 {
private volatile int count;
private Mono<Integer> count() { return Mono.fromSupplier( () -> { int c = count; for (int i = 0; i < Integer.MAX_VALUE; i++) { c++; } count = c; return c; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Drain Loop
Count Counter
Main
CounterCount
… …
• Reactive Streams abstracts async message passing
• No difference in message passing within and across a binary boundary to the caller
• RSocket formally defines a protocol to make Reactive Streams work across a binary boundary
• Can swap RSocket in place of Reactive Streams calls
�49
Reactive Streams / RSocket Connection
Reactive Stream / RSocket Connectionpublic class Counter4 {
private volatile int count;
private Mono<Integer> count() { return Mono.fromSupplier( () -> { int c = count; for (int i = 0; i < Integer.MAX_VALUE; i++) { c++; } count = c; return c; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); } }
Reactive Stream / RSocket Connectionpublic class Counter4 { private RSocket rSocket; public Counter4() { this.rSocket = RSocketFactory.connect().transport(TcpClientTransport.create(9090)).start().block(); } @Override public Mono<Integer> count() { return rSocket .requestResponse(ByteBufPayload.create(new byte[0])) .map( payload -> { int i = payload.data().readInt(); payload.release(); return i; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); }
Reactive Stream / RSocket Connectionpublic class Counter4 { private RSocket rSocket; public Counter4() { this.rSocket = RSocketFactory.connect().transport(TcpClientTransport.create(9090)).start().block(); } @Override public Mono<Integer> count() { return rSocket .requestResponse(ByteBufPayload.create(new byte[0])) .map( payload -> { int i = payload.data().readInt(); payload.release(); return i; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); }
Reactive Stream / RSocket Connectionpublic class Counter4 { private RSocket rSocket; public Counter4() { this.rSocket = RSocketFactory.connect().transport(TcpClientTransport.create(9090)).start().block(); } @Override public Mono<Integer> count() { return rSocket .requestResponse(ByteBufPayload.create(new byte[0])) .map( payload -> { int i = payload.data().readInt(); payload.release(); return i; }); }
public static void main(String... args) throws Exception { Counter4 counter = new Counter4(); final Recorder histogram = new Recorder(3600000000000L, 3); Flux.range(1, 100) .concatMap( ignore -> { long start = System.nanoTime(); return counter .count() .doFinally(s -> histogram.recordValue(System.nanoTime() - start)); }) .then() .block();
histogram.getIntervalHistogram().outputPercentileDistribution(System.out, 5, 1000.0, false); }
RSocket
Count Counter
Main
CounterCount
… …
• Temporal Decoupling
• Spacial Decoupling
• Binary
• Application-Flow control
�54
High Performance Microservices
• Temporal Decoupling - non-blocking api
• Spacial Decoupling - binary frames sent over dedicated stream
• Binary - binary payloads access with flyweights
• Application-Flow control - flow control information passed via messages
�55
RSocket Performance
Real-world
performance?
• Acme Air is a a realistic application benchmark that runs as part of Istio's nightly release pipeline
• Tests run on a standard 4-node GCP Kubernetes cluster• n1-standard-4 (4 vCPUs, 15
GB)
• https://ibmcloud-perf.istio.io
Source: IBM Acme Air Key Performance Indicator
• Acme Air is a a realistic application benchmark that runs as part of Istio's nightly release pipeline
• Tests run on a standard 4-node GCP Kubernetes cluster• n1-standard-4 (4 vCPUs, 15
GB)
• https://ibmcloud-perf.istio.io
Enterprise
Source: IBM Acme Air Key Performance Indicator
Source: Evaluating Critical Performance Needs for Microservices and Cloud-Native
• Acme Air is a a realistic application benchmark that runs as part of Istio's nightly release pipeline
• Tests run on a standard 4-node GCP Kubernetes cluster• n1-standard-4 (4 vCPUs, 15
GB)
• https://ibmcloud-perf.istio.io
Enterprise
Source: IBM Acme Air Key Performance Indicator
Source: Evaluating Critical Performance Needs for Microservices and Cloud-Native
