PrismTech

Building Reactive Applications with DDS

Angelo Corsaro, PhD Chief Technology Officer OMG DDS SIG Co-Chair

angelo.corsaro@prismtech.com

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Reactive Systems were characteristic of “niche” applications domains such as Industrial, Military, Telco, Finance and Aerospace

• To address the challenges imposed contemporary Internet Applications an increasing movement of architects and developers is recognising the importance of the techniques and technologies used for building Reactive Systems

Reactive Systems Go Mainstream

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Defense and Aerospace

Integrated Modular Vetronics Training & Simulation Systems Naval Combat Systems

Air Traffic Control & Management Unmanned Air Vehicles Aerospace Applications

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Industrial Internet

Autonomous Vehicle Systems

Train Control / Management Systems Connected Medical Devices

Smart CitiesSmart Grids / Power Generation

Financial Services

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• The techniques and technologies traditionally used in a class of Reactive and Interactive Systems is gaining quite a bit of attention from the mainstream developers community

• Recently, the Reactive Manifesto was defined to summarise the key traits of “Reactive Applications”

Reactive ManifestoResponsive

Scalable

Event-Driven Resilient

http://www.reactivemanifesto.org/

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Event-Driven: modular, asynchronous

• Responsive: Timely react to stimuli

• Resilient: Tolerate failures functionally and ideally temporally

• Scalable: Gracefully scale up and down depending on load demands

Reactive ManifestoResponsive

Scalable

Event-Driven Resilient

http://www.reactivemanifesto.org/

Reactive Systems

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Reactive Systems1

• A Reactive System designates a permanently operating system that responds, reacts, to external stimuli at a speed determined by its environment, e.g. an industrial control system.

Reactive Systems

[1] Nicolas Halbwachs, Synchronous Programming of Reactive Systems

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Transformational Systems

• A Transformational System designates a system that terminates after transforming its input into its output, e.g. a compiler

Transformational Systems

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Interactive Systems

• An Interactive System designates a systems that permanently communicates with its environment at its own speed, e.g. a computer game

Interactive Systems

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Reactive

• react to an environment which “cannot wait”, e.g. reacting late has harsh consequences

• response time needs to be deterministic

• distributed concurrent systems

• data-centric

Reactive vs. Interactive SystemsInteractive

• react to an environment which “doesn’t like to wait”, e.g. reacting late induces QoS degradation

• response time needs to be acceptable

• distributed concurrent systems

• data-centric

The term reactive systems is increasingly used to denote interactive systems with stringent response time, performance and scalability constraints.

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Concurrency

• Aside from the concurrency between the system and its environment, it is natural to decompose these systems as an ensemble of concurrent components that cooperate to achieve the intended behaviour

Strict Temporal Requirements

• Reactive Systems have strict requirements with respect to the rate at which the need to process input as well as the response time for their outputs

Determinism

• The output of these systems are generally determined by the input and the occurrence time, e.g. no scheduling effects on the temporal properties of the output

Reliability

• Reactive Systems are often life/mission critical, as such reliability is of utmost importance

Reactive Systems Key Traits

Architectural Styles for Reactive Systems

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Stream Processing is a commonly adopted architectural style for building systems that operate over continuous (theoretically infinite) streams of data

• Stream Processing is often applied in:

- Reactive Systems

- Interactive Systems

- Signal Processing Systems

- Functional Stream Programming

- Data Flow Systems

Stream Processing

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Stream Processing Architectures are ideal for modelling systems that react to streams of data produced by the cyber-phisical world, such as data produced by sensors, the stock market, etc.

• Stream Processing Systems operate in real-time over streams and generate in turns other streams of data providing information on what is happening or suggesting actions to perform, such as by stock X, raise alarm Y, or detected spatial violation, etc.

Stream Processing

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Stream Processing Systems are typically modelled as collection of concurrent modules communicating via typed data channels

• Modules usually play one of the following roles:

- Sources: Injecting data into the System

- Filters/Actors: Performing some computation over sources

- Sinks: Consuming the data produced by the system

Stream Processing

Filter

Source

FilterFilter

Stream Sink

Stream Processing with DDS

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• A stream represents a potentially infinite sequence of data samples of a given type T

• As such, in DDS, a stream can be naturally represented with a Topic

• Example:

- Topic<PressureType>

- Topic<MarketData>

Streams in DDS

Filter

FilterFilter

Stream

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• A Topic defines a domain-wide information’s class

• A Topic is defined by means of a (name, type, qos) tuple, where

- name: identifies the topic within the domain

- type: is the programming language type associated with the topic. Types are extensible and evolvable

- qos: is a collection of policies that express the non-functional properties of this topic, e.g. reliability, persistence, etc.

Topic

TopicType

Name

QoS

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

•As explained in the previous slide a topic defines a class/type of information

•Topics can be defined as Singleton or can have multiple Instances

•Topic Instances are identified by means of the topic key

•A Topic Key is identified by a tuple of attributes -- like in databases

•Remarks:

• A Singleton topic has a single domain-wide instance

• A “regular” Topic can have as many instances as the number of different key values, e.g., if the key is an 8-bit character then the topic can have 256 different instances

Topic and Instances

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Sources in DDS are mapped to DataWriters for a given Topic

• Sinks in DDS are mapped to DataReaders for a given Topic

Source and Sinks in DDS

Filter

Source

FilterFilter

Sink

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Filters implement bit and pieces of the application business logic

• Each Filter has one or more input stream and one or more output streams

• Obviously, a Filter consumes data from an input stream through a DDS data reader and pushed data into an output stream through a DDS data writer

Filters in DDS

Filter

FilterFilter

DDS Architectural Benefits

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Stream Processing Architectures in general and DDS-based architecture in particular promote loose coupling

• Anonymous Communication => No Spatial Coupling

• Non-Blocking and Asynchronous Interaction => No control or temporal coupling

In DDS the only thing that matters is the kind of information a module is interested in consuming or producing — who is producing the data or when the data has been produced is not relevant

Loose Coupling

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Location Transparency is an important architectural property as it makes it possible to completely decouple the logical decomposition of a system from its actual physical deployment

• DDS Automatic Discovery brings location transparency to the next level by enabling systems that are zero-conf, self-forming and self-healing

Location Transparency

Module

ModuleModule

Module

ModuleModule

Physical Deployment Unit

Logical Deployment Unit

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• DDS provides full control over temporal decoupling by means of its Durability Policy

• As such the data produced by a Source can be:

- Volatile: available for those that are around at the time at production

- Transient: available as far as the system/source is running

- Durable: available forever

Temporal DecouplingtSource

t

t

Sink

Sink

Volatile  Durability

tSource

t

t

Sink

Sink

Transient  Durability

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• DDS provides mechanism for detecting traditional faults as well as performance failures

• The Fault-Detection mechanism is controlled by means of the DDS Liveliness policy

• Performance Failures can be detected using the Deadline Policy which allows to receive notification when data is not received within the expected delays

Failure DetectionSource

tSink

Fault  Notification

TD

Source

tSink

Performance  Failure  Notification

P

t

P P

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• DDS provides a built-in fault-masking mechanism that allow to replicate Sources and transparently switch over when a failure occurs

• At any point in time the “active” source is the one with the highest strength. Where the strength is an integer parameter controller by the user

Fault-Masking

Source

tSink

Source t

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• DDS provides a built-in fault-masking mechanism that allow to replicate Sources and transparently switch over when a failure occurs

• At any point in time the “active” source is the one with the highest strength. Where the strength is an integer parameter controller by the user

Fault-Masking

Source

tSink

Source t

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Latency

- 15-20 usec Inter-Core Latency

- 45-75 usec over GBps Ethernet

• Throughput

- Up to 10+M msg/sec inter-core

- Up to 5M msg/sec inter node

Performance

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• The Data Distribution Service (DDS) was introduced as an OMG standard in 2006 to address Real-time and interoperable Data Sharing in Network Centric Systems

• Key requirements for the standard were high and predictable performance, and scalability to ultra-large-scale deployments

• DDS is recommended by key administration worldwide, e.g. DoD, MoD, EUROCAE, etc. and widely adopted across application domains, e.g., Automated Trading, Simulations, SCADA, Telemetry, etc.

Standard and Interoperable

Escape from Callback-Hell

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Reactive and Interactive Systems continuously interact with their environment

• Stimuli coming from the cyber-physical world are often represented as events to be handled by the Reactive/Interactive System

- e.g. think about the event from a GUI, or the data available event in DDS

• Events are often handled through some form of callbacks, such as listeners, functors, etc. This stile of event handling leads to inversion of control

Dealing with Events

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• The Inversion of Control induced by callback-based code in known to make applications brittle and harder to understand

• As an mental exercise, think about the call-back style code you’d have to write to in Java for drawing when the mouse is being dragged. Although the logic is simple, it is far cry from having a “declarative style”

• The problem of callback management, infamously known as Callback Hell, has become even more important due to the surge of Reactive Systems…

• Can we escape the Callback Hell?

The Callback Hell

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Reactive Programming is a paradigm based on a relaxed form of Synchronous Dataflow Programming

• The Reactive Programming is built around the notion of continuous time-varying values and propagation of change

• Reactive Programming facilitates the declarative development of non-blocking event driven applications — in essence you express what should be done and the runtime decides when to do it

• Reactive Programming was popularised in the context of Functional Programming Languages by the seminal done by Elliot and Hudak in 1997 as part of Fran — a framework for composing richly interactive multimedia animations

Reactive Programming

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Time for Common Operations

http://norvig.com/21-­‐days.html

execute typical instruction 1/1,000,000,000 sec = 1 nanosecfetch from L1 cache memory 0.5 nanosecbranch misprediction 5 nanosecfetch from L2 cache memory 7 nanosecMutex lock/unlock 25 nanosecfetch from main memory 100 nanosecsend 2K bytes over 1Gbps network 20,000 nanosecread 1MB sequentially from memory 250,000 nanosecfetch from new disk location (seek) 8,000,000 nanosecread 1MB sequentially from disk 20,000,000 nanosecsend packet US to Europe and back 150 milliseconds = 150,000,000 nanosec

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

!

There is an increasing number of Reactive Programming framework. The Rx framework introduced by Erik Meijer is becoming a reference

.NET Rx

• Arguably the first framework that brought Reactive Programming to the masses. Available for the .NET platform

RxJava (http://bit.ly/rxjava)

• An Open Source implementation of Rx for the JVM contributed by NetFlix, it supports, Java, Scala, Kotlin, Clojure, Groovy and JRuby

Reactive Programming Libraries

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• RxJava is a Java implementation of .NET Reactive Extensions — a library for composing asynchronous and event-based programs by using observable sequences.

• RxJava allow you to compose streams of data declaratively while abstracting away low level concerns such as threading, synchronization, concurrent data structures, and non-blocking I/O

• The two main concepts at the foundation of RxJava are Observables and Observers

RxJava Overview

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• RxJava Observables allow the composition of flows and sequences of asynchronous data

• You can think of Observables as some kinds of asynchronous collections that push data to you

• In essence an observable can be created from any synchronous or asynchronous stream of data.

- you can create an observables that represents the data coming into a DDS Data Reader, the events from a Button, or even time

- you can also create an observable that represent an actual container

RxJava Observables

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Observable Examples

t1 2 3

val  ticks:  Observable[Long]  =  Observable.interval(1  seconds)

n4 50

t1 2 3

val  evens:  Observable[Long]  =  ticks.filter(n  =>  n  %  2  ==  0)

n4 50 x x x

tval  bufs:  Observable[Seq[Long]]  =  evens.buffer(2,1)

n0 4 6 82

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• The Observer receives notification concerning new values, errors or the completion of the stream — e.g. no more data

RxJava Observers

trait  Observer[-­‐T]  {    def  onNext(value:  T):  Unit    def  onError(error:  Throwable):  Unit    def  onCompleted():  Unit  }

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Beside from the standard RxJava/RxScala Observable, each framework that wants to integrate with RxJava/RxScala has to provide its own factory methods to create observables

• For DDS you have to use the DdsObservable defined as follows:

DDS Observables

object  DdsObservable  {  

   def  fromDataReaderData[T](dr:  DataReader[T]):  Observable[T]  

   def  fromDataReaderEvents[T](dr:  DataReader[T]):  Observable[ReaderEvent[T]]  

   //  more  methods  which  we’ll  ignore  for  the  time  being  

}

Coding Lab

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• We will use the omnipresent Shapes Application to illustrate the benefits of using Reactive Programming with DDS

• Three Topics

- Circle, Square, Triangle

• One Type:

Shapes Application

struct ShapeType { string color; long x; long y; long shapesize; }; #pragma keylist ShapeType color

Spotted shapes represent subscriptions

Pierced shapes represent publications

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• To appreciate the joys of Reactive lets look first at how to implement a Shapes Logger using traditional “callback” programming

Logging Shapes

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Step 1: Implement Listenerclass  ShapesListener  extends  DataReaderListener[ShapeType]  {  !    def  onRequestedDeadlineMissed(p1:  RequestedDeadlineMissedEvent[ShapeType]):  Unit  !    def  onRequestedIncompatibleQos(p1:  RequestedIncompatibleQosEvent[ShapeType]):  Unit  !    def  onSampleRejected(p1:  SampleRejectedEvent[ShapeType]):  Unit  !    def  onLivelinessChanged(p1:  LivelinessChangedEvent[ShapeType]):  Unit  !    def  onDataAvailable(p1:  DataAvailableEvent[ShapeType]):  Unit  !    def  onSubscriptionMatched(p1:  SubscriptionMatchedEvent[ShapeType]):  Unit  !    def  onSampleLost(p1:  SampleLostEvent[ShapeType]):  Unit  }

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• To keep things simple and more compact we’ll use the Scala binding

• From the listener code you’d retrieve the data reader (dr) and do something like this:

Step 2: Read and Log Data

def  onDataAvailable(p1:  DataAvailableEvent[ShapeType]):  Unit  =  {        dr.read.foreach(s  =>  println("("  +  s.color  +  ",  "  +  s.x  +  ",  "  +  s.y))    }  !

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• To be fair, if you use Moliere, the Scala API for OpenSplice you could simply write:

Molière Makes it Better

dr  listen  {      case  DataAvailable(_)    =>  {          dr  read()  foreach  (s  =>  println("("  +  s.color  +  ",  "  +  s.x  +  ",  "  +  s.y))      }  }

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Reactive Logger

val  shapes  =  DdsObservable.fromDataReaderData(dr)  shapes.subscribe(s  =>            println("("  +  s.color  +  ",  "  +  s.x  +  ",  "  +  s.y)  )

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

To show a bit more the expressive power of Reactive programming, let’s see how we can solve the following problems

“Shapes Problems”

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• We want to show a triangle that positioned midway between every single pair of circle and square of the same colour

Problem 1: Shape in the Middle

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Shapes Marble Diagrams

t

t

• Conceptually, we need to select on stream, look at it one element at the time and match it with the corresponding item in the other stream

• Then, we can produce the average

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Shapes Marble Diagrams

t

t

t

x x

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Shapes Marble Diagrams

t

t

t

x x x x

P.S.: The dropping may seem a bit counter-intuitive but it is in part due to how the DDS streams work

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Midpoint Triangle        val  circles  =  DdsObservable.fromDataReaderData  {              DataReader[ShapeType](Topic[ShapeType](circle))          }          val  squares  =  DdsObservable.fromDataReaderData  {              DataReader[ShapeType](Topic[ShapeType](square))          }          val  ttopic  =  Topic[ShapeType](triangle)          val  tdw  =  DataWriter[ShapeType](ttopic)  !!        //  Compute  the  average  between  circle  and  square  of  matching  color  with  flatMap          val  triangles  =  circles.flatMap  {              c  =>  squares.dropWhile(_.color  !=  c.color).take(1).map  {                  s  =>  new  ShapeType(s.color,  (s.x  +  c.x)/2,  (s.y  +  c.y)/2,  (s.shapesize  +  c.shapesize)/4)              }          }  !        triangles.subscribe(tdw.write(_))  

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

Using for-comprehension            //  Compute  the  average  between  circle  and  square  of  matching  colour  with  flatMap          val  triangles  =  circles.flatMap  {              c  =>  squares.dropWhile(_.color  !=  c.color).take(1).map  {                  s  =>  new  ShapeType(s.color,  (s.x  +  c.x)/2,  (s.y  +  c.y)/2,  (s.shapesize  +  c.shapesize)/4)              }          }  

           //  Compute  the  average  between  circle  and  square  of  matching  colour  with  for  comprehension          val  triangles  =  for  {                c  <-­‐  circles;                s  <-­‐  squares.dropWhile(_.color  !=  c.color).take(1)          }  yield  new  ShapeType(s.color,  (s.x  +  c.x)/2,  (s.y  +  c.y)/2,  (s.shapesize  +  c.shapesize)/4)  

Summing Up

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• Many of the requirements of traditional Reactive Systems are applicable to many mainstream application domains

• Stream Processing Architecture and Reactive Programming can greatly help in the design and implementation of Reactive Systems that are efficient, responsive, resilient, scalable and highly available

• As proven through years of deployments DDS provides the ideal infrastructure technology for Reactive Systems

• The combination of DDS with Reactive Programming makes for a very powerful platform!

Concluding Remarks

Cop

yrig

ht P

rism

Tech

, 201

4

PrismTech

• OpenSplice DDS [http://prismtech.com]

• RxJava [http://bit.ly/rxjava] (OpenSouce Rx Implementation from NetFlix)

• Moliere [http://bit.ly/nuvo-moliere] (This includes Rx Binding for DDS)

References