You might have seen on my twitter that my current company MFG Labs has opensourced the library Akka-Stream Extensions. We have developed it with Alexandre Tamborrino and Damien Pignaud for our recent production projects based on Typesafe Akka-Stream.

In this article, I won’t explain all the reasons that motivated our choice of Akka-Stream at MFG Labs and the road towards our library Akka-Stream Extensions. Here, I’ll focus on one precise aspect of our choice: types… And I’ll tell you about a specific extension I’ve created for this project: ShapelessStream.

The code is there on github

Akka-Stream loves Types

As you may know, I’m a Type lover. Types are proofs and proofs are types. Proofs are what you want in your code to ensure in a robust & reliable way that it does what it pretends with the support of the compiler.

First of all, let’s remind that typesafety is a very interesting feature of Akka-Stream.

Akka-Stream most basic primitive Flow[A, B] represents a data-flow that accepts elements of type A and will return elements of type B. You can’t pass a C to it and you are sure that this flow won’t return any C for example.

At MFG Labs, we have inherited some Scala legacy code mostly based on Akka actors which provide a very good way to handle failures but which are not typesafe at all (till Akka Typed) and not composable. Developers using Akka tend to scatter the business logic in the code and it can become hard to maintain. It has appeared that in many cases where Akka was used to transform data in a flow, call external services, Akka-Stream would be a very good way to replace those actors:

• better type-safety,
• fluent & composable code with great builder DSL
• same Akka failure management
• buffer, parallel computing, backpressure out-of-the-box

Yes, it’s quite weird to say it but Akka-Stream helped us correct most problems that had been introduced using Akka (rightly or wrongly).

Multi-type flows

Ok, Akka-Stream promotes Types as first citizen in your data flows. That’s cool! But it appears that you often need to handle multiple types in the same input channel:

When you control completely the types in input, you can represent input types by a classic ADT:

1
2
3
4

sealed trait A
case class A1(...) extends A
case class A2(...) extends A
case class A3(...) extends A

… And manage it in Flow[A, B]:

1
2
3
4
5

Flow[A].map {
  case A1(...) => // some B
  case A2(...) => // some B
  case A3(...) => // some B
}

Nice but you need to wrap all input types in an ADT and this involves some boring code that can even be different for every custom flow.

Going further, in general, you don’t want to do that, you want to dispatch every element to a different flow according to its type:

… and merge all results of all flows in one single channel…

… and every flow has its own behavior in terms of parallelism, bufferization, data generation and back-pressure…

In Akka-Stream, if you wanted to build a flow corresponding to the previous schema, you would have to use:

• a FlexiRoute for the input dispatcher
• a FlexiMerge for the output merger.

Have a look at the doc and see that it requires quite a bunch of lines to write one of those. It’s really powerful but quite tedious to implement and not so typesafe after all. Moreover, you certainly would have to write one FlexiRoute and one FlexiMerge per use-case as the number of inputs types and return types depend on your context.

Miles Sabin to the rescue

In my latest project, this dispatcher/flows/merger pattern was required in multiple places and as I’m lazy, I wanted something more elegant & typesafe if possible to build this kind of flow graphs.

Thinking in terms of pure types and from an external point of view, we can see the previous dispatcher/flows/merger flow graph in pseudo-code as:

1
2
3
4

Flow[
  Input = A1 or A2 or A3, // in input it accepts A1 or A2 or A3
  Ouput = B1 or B2 or B3  // in output it generates B1 or B2 or B3
]

And to build the full flow graph, we need to provide a list of flows for all pairs of input/output types corresponding to our graph branches:

1

Flow[A1, B1] and Flow[A2, B2] and Flow[A3, B3]

In Shapeless, there are 2 very very very useful structures:

• Coproduct is a generalization of the well known Either. You have A or B in Either[A, B]. With Coproduct, you can have more than 2 alternatives A or B or C or D. So, for our previous external view of flow graph, using Coproduct, it could be written as:

1
2
3
4

Flow[
  A1 :+: A2 :+: A3 :+: CNil,
  B1 :+: B2 :+: B3 :+: CNil
]

• HList allows to build heterogenous List of elements keeping & tracking all types at compile time. For our previous list of flows, it would fit quite well as we want to match all input/output types of all flows. It would give:

1

Flow[A1, B1] :: Flow[A2, B2] :: Flow[A3, B3] :: HNil

So, from an external point of view, the process of building our dispatcher/flows/merger flow graph looks like a Function taking aHlist of flowsin input and returning the builtFlow of Coproducts`:

1
2

Flow[A1, B1] :: Flow[A2, B2] :: Flow[A3, B3] :: HNil =>
  Flow[A1 :+: A2 :+: A3 :+: CNil, B1 :+: B2 :+: B3 :+: CNil]

Let’s write it in terms of Shapeless Scala code:

1
2
3
4
5
6
7
8
9

/**
 * Builds at compile-time a fully typed-controlled flow that transforms a HList of Flows to a Flow of the Coproduct of inputs to Coproduct of outputs.
 *
 * @param a Hlist of flows Flow[A1, B1] :: FLow[A2, B2] :: ... :: Flow[An, Bn] :: HNil
 * @return the flow of the Coproduct of inputs and the Coproduct of outputs Flow[A1 :+: A2 :+: ... :+: An :+: CNil, B1 :+: B2 :+: ... +: Bn :+: CNil, Unit]
 */
def coproductFlow[HL <: HList, CIn <: Coproduct, COut <: Coproduct](
  flows: HL
): Flow[CIn, COut, Unit]

Fantastic !!!

Now the question is how can we build at compile-time this Flow[CIn, COut, Unit] from an HList of Flows and be sure that the compiler checks all links are correctly typed and all types are managed by the provided flows?

Akka-Stream Graph Mutable builders

An important concept in Akka-Stream is the separation of concerns between:

• constructing/describing a data-flow
• materializing with live resources (like actor system)
• running the data-flow by plugging live sources/sinks on it (like web, file, hdfs, queues etc…).

For the curious, you find the same idea in scalaz-stream but in a FP-purer way as scalaz-stream directly relies on Free concepts that formalize this idea quite directly.

Akka-Stream has taken a more custom way to respond to these requirements. To build complex data flows, it provides a very nice DSL described here. This DSL is based on the idea of a mutable structure used while building your graph until you decide to fix it definitely into an immutable structure.

An example from the doc:

1
2
3
4
5
6
7
8
9
10
11
12
13

val g = FlowGraph.closed() { implicit builder: FlowGraph.Builder[Unit] =>
  import FlowGraph.Implicits._
  val in = Source(1 to 10)
  val out = Sink.ignore

  val bcast = builder.add(Broadcast[Int](2))
  val merge = builder.add(Merge[Int](2))

  val f1, f2, f3, f4 = Flow[Int].map(_ + 10)

  in ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> out
              bcast ~> f4 ~> merge
}

builder is the mutable structure used to build the flow graph using the DSL inside the {...} block.

The value g is the immutable structure resulting from the builder that will later be materialized and run using live resources.

Please remark that once built, g value can reused and materialized/run several times, it is just the description of your flow graph.

This idea of mutable builders is really interesting in general: mutability in the small can help a lot to make your building block efficient and easy to write/read without endangering immutability in the large.

Hacking mutable builders with Shapeless

My intuition was to hack these mutable Akka-Stream builders using Shapeless type-dependent mechanics to build a Flow of Coproducts from an HList of Flows…

Let’s show the real signature of coproductFlow:

1
2
3
4
5
6
7
8
9
10
11
12

def coproductFlow[HL <: HList, CIn <: Coproduct, COut <: Coproduct, CInOutlets <: HList, COutInlets <: HList](
  flows: HL
)(
  implicit
    flowTypes: FlowTypes.Aux[HL, CIn, COut],
    obuild: OutletBuilder.Aux[CIn, CInOutlets],
    ibuild: InletBuilder.Aux[COut, COut, COutInlets],
    otrav: ToTraversable.Aux[CInOutlets, List, Outlet[_]],
    itrav: ToTraversable.Aux[COutInlets, List, Inlet[COut]],
    selOutletValue: SelectOutletValue.Aux[CIn, CInOutlets],
    flowBuilder: FlowBuilderC.Aux[CIn, COut, CInOutlets, HL, COutInlets]
): Flow[CIn, COut, Unit]

Frightening!!!!!!!

No, don’t be, it’s just the transcription in types of the requirements to build the full flow.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

def coproductFlow[HL <: HList, CIn <: Coproduct, COut <: Coproduct, CInOutlets <: HList, COutInlets <: HList](
  flows: HL
)(
  implicit
    // 1 - Introspects HList of Flows to extract
    //      * all input types in CIn Coproduct
    //      * all output types in COut Coproduct
    flowTypes: FlowTypes.Aux[HL, CIn, COut],

    // 2 - Builds all Akka-Stream outlets branching the FlexiRoute
    //     outlets to the right Flow inlet in the HList
    obuild: OutletBuilder.Aux[CIn, CInOutlets],

    // 3 - Builds all Akka-Stream inlets branching the outlets of each Flow
    //     in the HList to the right inlet of FlexiMerge
    ibuild: InletBuilder.Aux[COut, COut, COutInlets],

    // 4 - Technical structures to be able to traverse
    //     and select all those previously built thingies
    otrav: ToTraversable.Aux[CInOutlets, List, Outlet[_]],
    itrav: ToTraversable.Aux[COutInlets, List, Inlet[COut]],
    selOutletValue: SelectOutletValue.Aux[CIn, CInOutlets],

    // 5 - The AkkaStream mutable Builder that:
    //      * builds the input FlexiRoute with the right output types
    //      * plugs all FlexiRoute outlets to the right flow inlet
    //      * builds the output FlexiMerge with the right input types
    //      * plugs all flow outlets to the right inlet of FlexiMerge
    flowBuilder: FlowBuilderC.Aux[CIn, COut, CInOutlets, HL, COutInlets]
): Flow[CIn, COut, Unit]

The Scala code might seem a bit ugly to a few of you. That’s not false but keep in mind what we have done: mixing shapeless-style recursive implicit typeclass inference with the versatility of Akka-Stream mutable builders… And we were able to build our complex flow graph, to check all types and to plug all together at compile-time…

Sample

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

// 1 - Create a type alias for your coproduct
type C = Int :+: String :+: Boolean :+: CNil

// The sink to consume all output data
val sink = Sink.fold[Seq[C], C](Seq())(_ :+ _)

// 2 - a sample source wrapping incoming data in the Coproduct
val f = FlowGraph.closed(sink) { implicit builder => sink =>
  import FlowGraph.Implicits._
  val s = Source(() => Seq(
    Coproduct[C](1),
    Coproduct[C]("foo"),
    Coproduct[C](2),
    Coproduct[C](false),
    Coproduct[C]("bar"),
    Coproduct[C](3),
    Coproduct[C](true)
  ).toIterator)

// 3 - our typed flows
  val flowInt = Flow[Int].map{i => println("i:"+i); i}
  val flowString = Flow[String].map{s => println("s:"+s); s}
  val flowBool = Flow[Boolean].map{s => println("s:"+s); s}

// >>>>>> THE IMPORTANT THING
// 4 - build the coproductFlow in a 1-liner
  val fr = builder.add(ShapelessStream.coproductFlow(flowInt :: flowString :: flowBool :: HNil))
// <<<<<< THE IMPORTANT THING

// 5 - plug everything together using akkastream DSL
  s ~> fr.inlet
       fr.outlet ~> sink
}

// 6 - run it
f.run().futureValue.toSet should equal (Set(
  Coproduct[C](1),
  Coproduct[C]("foo"),
  Coproduct[C](2),
  Coproduct[C](false),
  Coproduct[C]("bar"),
  Coproduct[C](3),
  Coproduct[C](true)
))

FYI, Shapeless Coproduct provides a lot of useful operations on Coproducts such as unifying all types or merging Coproducts together.

Some compile errors now ?

Imagine you forget to manage one type of the Coproduct in the HList of flows:

1
2
3
4
5
6

...

// 4 - build the coproductFlow in a 1-liner
  val fr = builder.add(ShapelessStream.coproductFlow(flowInt :: flowString :: HNil))

..

If you compile, it will produce this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

ShapelessExtensionsSpec.scala:97: overloaded method value ~> with alternatives:
[error]   (to: akka.stream.SinkShape[C])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]   (to: akka.stream.Graph[akka.stream.SinkShape[C], _])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]   [Out](flow: akka.stream.FlowShape[C,Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](junction: akka.stream.UniformFanOutShape[C,Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](junction: akka.stream.UniformFanInShape[C,Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](via: akka.stream.Graph[akka.stream.FlowShape[C,Out],Any])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   (to: akka.stream.Inlet[C])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit
[error]  cannot be applied to (akka.stream.Inlet[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.CIn]]])
[error]       s ~> fr.inlet
[error]         ^
[error] /Users/pvo/workspaces/mfg/akka-stream-extensions/extensions/shapeless/src/test/scala/ShapelessExtensionsSpec.scala:98: overloaded method value ~> with alternatives:
[error]   (to: akka.stream.SinkShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]]])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]   (to: akka.stream.Graph[akka.stream.SinkShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]]], _])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]   [Out](flow: akka.stream.FlowShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]],Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](junction: akka.stream.UniformFanOutShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]],Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](junction: akka.stream.UniformFanInShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]],Out])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   [Out](via: akka.stream.Graph[akka.stream.FlowShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]],Out],Any])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])akka.stream.scaladsl.FlowGraph.Implicits.PortOps[Out,Unit] <and>
[error]   (to: akka.stream.Inlet[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]]])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit
[error]  cannot be applied to (sink.Shape)
[error]            fr.outlet ~> sink
[error]                      ^

OUCHHHH, this is a mix of the worst error of Akka-Stream and the kind of errors you get with Shapeless :)

Don’t panic, breathe deep and just tell yourself that in this case, it just means that your types do not fit well

In general, the first line and the last lines are the important ones.

For input:

1
2
3
4
5

ShapelessExtensionsSpec.scala:97: overloaded method value ~> with alternatives:
[error]   (to: akka.stream.SinkShape[C])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]  cannot be applied to (akka.stream.Inlet[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.CIn]]])
[error]       s ~> fr.inlet
[error]         ^

It just means you try to plug a C == Int :+: String :+: Bool :+: CNil to a Int :+: String :+: CNil and the compiler is angry against you!!

For output:

1
2
3
4
5

ShapelessExtensionsSpec.scala:98: overloaded method value ~> with alternatives:
[error]   (to: akka.stream.SinkShape[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]]])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit <and>
[error]   (to: akka.stream.Inlet[shapeless.:+:[Int,shapeless.:+:[String,com.mfglabs.stream.extensions.shapeless.FlowTypes.last.COut]]])(implicit b: akka.stream.scaladsl.FlowGraph.Builder[_])Unit
[error]  cannot be applied to (sink.Shape)
[error]            fr.outlet ~> sink

It just means you try to plug a Int :+: String :+: CNil to a C == Int :+: String :+: Bool :+: CNil and the compiler is 2X-angry against you!!!

Conclusion

Mixing the power of Shapeless compile-time type dependent structures and Akka-Stream mutable builders, we are able to build at compile-time a complex dispatcher/flows/merger flow graph that checks all types and all flows correspond to each other and plugs all together in a 1-liner…

This code is the first iteration on this principle but it appeared to be so efficient and I trusted the mechanism so much (nothing happens at runtime but just at compile-time) that I’ve put that in production two weeks ago. It runs like a charm.

Finally, there are a few specificities/limitations to know:

• Wrapping input data into the Coproduct is still the boring part with some pattern matching potentially. But this is like Json/Xml validation, you need to validate only the data you expect. Yet I expect to reduce the work soon by providing a Scala macro that will generate this part for you as it’s just mechanical…

• Wrapping everything in Coproduct could have some impact on performance if what you expect is pure performance but in my use-cases IO are so much more impacting that this is not a problem…

• coproductFlow is built with a custom FlexiRoute with DemandFromAll condition & FlexiMerge using ReadAny condition. This implies :

  • the order is NOT guaranteed due to the nature of used FlexiRoute & FlexiMerge and potentially to the flows you provide in your HList (each branch flow has its own parallelism/buffer/backpressure behavior and is not necessarily a 1-to-1 flow).

  • the slowest branch will slow down all other branches (as with a broadcast). To manage these issues, you can add buffers in your branch flows to allow other branches to go on pulling input data

The future?

• A macro generating the Coproduct wrapping flow

• Some other flows based on Shapeless

Have more backpressured and typed fun…

Draft FreeR code is on Github

Introduction

I’ve recently pushed some Free code & doc to the cool project cats and I had a few more ideas in my head on optimizing Free and never took time to make them concrete. I’ve just found this time during my holidays…

Free Monad is often used to represent embedded DSL in functional programming languages like Haskell or Scala. One generally represents his grammar with a simple Functor ADT representing the available operations. Then, from within your programming language, Free Monad provides the facilities to:

• build in a monadic & stack-safe way a static program composed of a sequence of operations,
• compile this program,
• run it.

To know more about the way to use Free and some more specific theory, please refer to recent draft doc I’ve pushed on cats

The well-known classic representation in Scala is the following:

1
2
3

sealed abstract class Free[F[_], A]
case class Pure[F[_], A](a: A) extends Free[F, A]
case class Suspend[F[_], A](a: F[Free[F, A]]) extends Free[F, A]

Please note that F[_] should be a Functor to take advantage of Free construction.

Building a program can then just be a classic sequence of monadic bind/flatMap on Free[S[_], _]:

1
2
3
4
5

for {
  a <- doA(...)
  b <- doB(a, ...)
  c <- doC(c, ...)
} yield (c)

This actually constructs a recursive structure looking like:

1

Suspend(F(Suspend(F(Suspend(F(....(Pure(a))))))))

It can be seen as a left-associated sequence of operations and as any left-associated structure, appending an element to it has a quadratic complexity. So the more you flatMap, the longer (in n²) it will take to drill down the structure.

So if you try such code:

1
2
3
4
5
6

def gen[I](i: I): Trampoline[I] = Trampoline.suspend(Trampoline.done(i))

//(a flatMap (b flatMap (c flatMap (...))))
def lftBind[S[_]](n: Int)(gen: Int => S[Int])(implicit M: Monad[S]) = {
  (1 to n).foldLeft(gen(0)){ case (acc, i) => acc flatMap { a => gen(i) } }
}

You will see that it has a quadratic curve in terms of execution time when you increase n.

First weakness of classic Free is its left-associativity that induces a quadratic complexity when flatMapping

Solving left-associated quadratic complexity

To solve it, the immediate idea is to make Free right-associative instead of left associative (This idea was proposed by Kiselyov & al. in a paper and called Continuation-Passing-Style or also Codensity construction)

This is already done in current scalaz/cats.Free by adding a new element to the Free ADT:

1
2
3
4
5
6

/** Call a subroutine and continue with the given function. */
sealed abstract class Gosub[S[_], B] extends Free[S, B] {
  type C
  val a: () => Free[S, C]
  val f: C => Free[S, B]
}

If you test the same previous code, it has a linear behavior when n increases.

Quadratic observability

In this great paper Reflection Without Remorse, Atze van der Ploeg & Oleg Kiselyov show that classic Free are subject to another tricky quadratic behavior when, within your sequence of operations, one need to observe the current internal state of Free.

Observing the state requires to drill down the recursive Free structure explicitly and go up and again down then up and again and again. As explained in the paper, this case is quite tricky because it’s very hard to see that it will happen. The deeper is the structure, the longer it takes to observe current state. It’s important to note that right-association doesn’t help in this case and that complexity is once again in O(n²).

The second weakness of Free is its quadratic complexity when observing internal state.

To solve it, in Reflection Without Remorse, they propose a very interesting approach by changing the representation of Free and take advantage of its sequential nature.

A Free becomes the association of 2 elements:

• a FreeView representing current internal state of the Free
• a sequence of bind/flatMap functions stored in an efficient data structure that can prepent/append in O(1).

For the data structure, they propose to use a type-aligned dequeue to keep track of all types.

I have tried to implement this structure using a typed-aligned FingerTree in Scala. The code is here. The result is pretty interesting but not much efficient: it has a linear behavior for left-association & observability but…

• for lower values of n, FingerTree costs far too much to build
• memory cost is so high that it triggers the JVM GC far too soon and limitates a lot what you can do
• allocated memory locality isn’t good and requires huge amounts of memory jumps.
• type-alignment makes code quite ugly (yes Scala type inference isn’t as powerful as Haskell in this case and laziness of Haskell helps a lot for FingerTree)

As a conclusion, the idea is really nice on paper but in practice, we need to find something that costs less than this type-aligned dequeue (even if my FingerTree code is really raw, too strict and not optimized at all).

FreeR hybrid structure

I wanted to improve Free behavior and decided to create a new version of it called FreeR thinking in terms of efficient Scala…

I really liked the idea of representing a Free as a pure sequence of operations with a view of current internal state.

To gain in efficiency, I decided to choose another efficient append/prepend data structure, optimized and very well known: Vector providing:

• append/prepend in O(1) (in average),
• random access in constant time,
• quite good locality.

Then, I’ve decided to relax a lot type alignment and manipulate Any values internally and cast/reify to the right types when required.

BTW, I plagiarized some code written by Alois Cochard for his new IO model in Scalaz/8.0 branch… Alois is a great dev & had made concrete the idea I had in my head so why rewrite them differently? uh ;)

I also decided to reify the 2 kinds of operations:

• Bind for flatMap/bind calls
• Map for map calls

So a Free becomes:

1
2
3
4
5
6
7
8

// to mitigate the shock of Any in the code
// and provide helpers for casting/reifying
type Val = Any

case class FreeR[S[_], A](
  head: FreeView[S, Val],
  ops: Ops = Vector.empty
) extends FreeR[S, A]

with FreeView as:

1
2
3
4
5
6
7
8
9
10

// The view of Free internal state can be of 2 types:
sealed abstract class FreeView[S[_], A]

object FreeView {
  // Pure value
  case class Pure[S[_], A](a: A) extends FreeView[S, A]

  // Impure computation determined by the Functor S
  case class Impure[S[_], A](a: S[FreeR[S, A]]) extends FreeView[S, A]
}

and Ops are:

1
2
3
4
5

sealed trait Op
object Op {
  case class Map(f: Val => Val) extends Op
  case class Bind[S[_]](f: Val => FreeR[S, Val]) extends Op
}

FYI This code is less than 300 lines so nothing really horrible except a few ugly casts ;)

Left Association

The code used for testing can be found here

FreeR behavior is linear even for millions of flatMap (until the GC triggers naturally) whereas classic Free has clearly quadratic curve.

Observability

The code used for testing can be found here

FreeR behavior is quite linear even for millions of flatMap (until the GC triggers naturally) whereas classic Free has clearly quadratic curve.

Right association complexity

I finally tried to check the behavior of my new FreeR when using flatMap in a right associated way like:

1
2
3
4

// (... flatMap (_ => c flatMap (_ => b flatMap (_ => a))))
def rgtBind[S[_]](n: Int)(gen: Int => S[Int])(implicit M: Monad[S]) = {
  (1 to n).foldLeft(gen(n)){ case (acc, i) => gen(n-i) flatMap { _ => acc } }
}

This is not so frequent code but anyway, Free should be efficient for left & right associated code.

Using FreeR as described previously, I’ve discovered that it wasn’t efficient in right association when increasing n because it allocates recursively a lot of Vector with one element and it becomes slower and slower apparently (I’m not even sure of the real cause of it).

I’ve refined my representation by distinguishing 3 kinds of Free in my ADT:

1
2
3
4
5
6
7
8

// No op
case class FreeR0[S[_], A](head: FreeView[S, Val]) extends FreeR[S, A]

// One single op (typically the right association case)
case class FreeR1[S[_], A](head: FreeView[S, Val], op: Op) extends FreeR[S, A]

// Multiple ops
case class FreeRs[S[_], A](head: FreeView[S, Val], ops: Ops = Vector.empty) extends FreeR[S, A]

With this optimization, here is the performance in right association:

It is quite comparable to classic Free for n under 1 million but it becomes quite bad when n becomes big. Yet, it remains for more efficient than previous representation with just Vector.

I need to work more on this issue (apparently GC is triggered too early) to see if more optimizations for right association can be found…

Cherry on the cake: map-fusion optimization

Imagine doing a lot of map operations on a Free like:

1
2
3

def mapalot(n: Int): Trampoline[Long] = {
  (1 to n).foldLeft(Trampoline.done(0L)){ case (acc, i) => acc map { a => a + 1 } }
}

If you think just a bit, you will clearly see that:

1

a.map(f).map(g) == a.map(g compose f)

This is called map-fusion and as you may have deduced already, my decision to reify explicitly Bind and Map operations was made in this purpose.

If I can know there are several Map operations in a row, I can fusion them in one single Map by just calling mapFusion on a Free to optimize it:

1
2
3
4
5
6
7

val free = FreeRTools.mapalot(x)

// optimized free
val freeOpt = free.mapFusion

// run the trampoline
freeOpt.run

Here is the difference in performance between FreeR and FreeR.mapFusion:

As you can see, mapFusion can be very interesting in some cases.

Conclusion

Finally, I have created a new representation of Free using:

• type-relaxed version of Reflection w/o Remorse
• sequence of operations managed by Scala Vector
• reification of Bind & Map operations
• differenciation of None/Single/Multiple operations cases
• Map Fusion optimization

It allows to have a Free with:

• Linear behavior in Left-Assocation, Observability,
• Stack-safety is sill ensured,
• Right-Association should be optimized because it still has a too high cost for bigger n (yet it is far more acceptable than other alternatives),
• Map Fusion can provide an interesting optimization when using multiple consecutive Map operations,
• For small n, the cost is a bit higher than basic Free but quite low and acceptable.

It is really interesting as it makes Free more and more usable in real-life problems without having to rewrite the code bootstrapped with Free in a more optimized way. I personally find it quite promising!

Please remark that this code has been written for the great project cats that will soon be a viable & efficient alternative for functional structures in Scala.

The full code is there.

Don’t hesitate to test, find bugs, contribute, give remarks, ideas…

Have fun in FreeR world…

SCALEDN, EDN Scala API

Scaledn is a Scala EDN parser (runtime & compile-time)/serializer/validator based on :

• Parboiled2,
• Shapeless,
• Generic Validation
• Scala Macros

It works only in Scala 2.11.x

The code & sample apps can be found on Github

Why EDN?…

Because Json is not so good & quite limitating

EDN is described as an extensible data notation specified (not really standardized) there. Clojure & Datalog used in Datomic are supersets of EDN.

EDN allows much more things than Json while keeping the same simplicity.

Here are the main points making EDN great to represent & exchange Data.

EDN manages number types far better than Json

For Json, all numbers (floating or integer, exponential or not) are all considered in the same way so numbers can only be mapped to the biggest number format: BigDecimal. It is really bad in terms of semantics and performance.

In EDN, numbers can be :

• 64bits integer aka Long in Scala: 12345
• 64bits floating point numbers & exponentials aka Double in Scala: 123.45e-9
• Natural Integers aka BigInt in Scala: 1234567891234N
• Exact Floating Number aka BigDecimal in Scala: 123.4578972345M

EDN knows much more about collections

Collections in Json are just:

• lists of heterogenous json values
• maps of key strings and json values.

In EDN, you can have:

• heterogenous lists

1

(1 true "toto)

• heterogenous vectors/arrays

1

[1 true "toto]

• heterogenous sets

1

#{1 true "toto}

• heterogenous maps with heterogenous keys & values

1

{1 "toto", "foo" 2}

EDN accepts characters & unicode

Json doesn’t know about characters outside strings.

EDN can manage chars:

1
2
3
4
5
6
7
8
9
10
11
12

// simple char
\c

// special chars
\newline
\return
\space
\tag
\\

// unicode
\u0308

EDN accepts comments & discarded values

There are special syntaxes:

• comments are lines starting with ;

1

; this is a comment

• values starting with #_ are parsed but discarded

1

"toto" 3 #_discarded 1.234

EDN knows about symbols & keywords

These are notions that don’t exist in Json.

Symbols can reference anything external or internal that you want to identify. A Symbol can have a namespace such as:

1

foo.foo2/bar

Keywords are unique identifiers or enumerated values that can be reused in your data structure. A Keyword is just a symbol preceded by a : such as

1

:foo.foo2/bar

EDN is extensible using tags

EDN is an extensible format using tags starting with # such as:

1

#foo/bar value

When parsing EDN format, the parser should provide tag handlers that can be applied when a tag is discovered. In this way, you can extend default format with your own formats.

EDN specifies 2 tag handlers by default:

• #inst "1985-04-12T23:20:50.52Z" for RFC-3339 instants
• #uuid "f81d4fae-7dec-11d0-a765-00a0c91e6bf6" for UUID

EDN has no root node & can be streamed

Json is defined to have a root map node: { key : value } or [ ... ].

Json can’t accept single values outside of this. So Json isn’t really meant to be streamed as you need to find closing tags to finish parsing a value.

EDN doesn’t require this and can consist in multiple heterogenous values:

1

1 123.45 "toto" true nil (1 2 3)

As a consequence, EDN can be used to stream your data structures.

Conclusion: EDN should be preferred to Json

All of these points make EDN a far better & stricter & more evolutive notation to represent data structures than Json. It can be used in the same way as Json but could make a far better RPC string format than Json.

I still wonder why Json has become the de-facto standard except for the reason that the not so serious Javascript language parses it natively and because people were so sick about XML that they would have accepted anything changing their daily life.

But JS could also parse EDN without any problem and all more robust & typed backend languages would earn a lot from using EDN instead of JSON for their interfaces.

EDN could be used in REST API & also for streaming API. That’s exactly why, I wanted to provide a complete Scala API for EDN to test this idea a bit further.

Scaledn insight

Runtime Parsing

Scaledn can be used to parse the EDN string or arrays of chars received by your API.

All types described in EDN format are isomorphic to Scala types so I’ve decided to skip the complete AST wrapping those types and directly parse to Scala types.

• "foobar" is parsed to String
• 123 is parsed to Long
• (1 2 3) is parsed to List[Long]
• (1 "toto" 3) is parsed to List[Any]
• {"toto" 1 "tata" 2} is parsed to Map[String, Long]
• {1 "toto" 2 "tata"} is parsed to Map[Long, String]
• {1 "toto" true 3} is parsed to Map[Any, Any]
• etc…

The parser (based on Parboiled2) provides 2 main functions:

1
2
3
4
5
6
7
8

import scaledn._
import parser._

// parses only the first EDN value discovered in the String input
def parseEDN(in: ParserInput): Try[EDN] = ...

// parses all EDN values discovered in the String input
def parseEDNs(in: ParserInput): Try[Seq[EDN]] = ...

If you look in common package, you’ll see that EDN is just an alias for Any ;)

Here is how you can use it:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import scaledn._
import parser._

// Single Value
parseEDN("""{1 "foo", "bar" 1.234M, :foo/bar [1,2,3]} #_foo/bar :bar/foo""") match {
  case Success(t) => \/-(t)
  case Failure(f : org.parboiled2.ParseError) => -\/(parser.formatError(f))
}

// Multiple Value
parseEDNs("""{1 "foo", "bar" 1.234M, :foo/bar [1,2,3]} :bar/foo""").success.value should be (
  Vector(
    Map(
      1L -> "foo",
      "bar" -> BigDecimal("1.234"),
      EDNKeyword(EDNSymbol("foo/bar", Some("foo"))) -> Vector(1, 2, 3)
    ),
    EDNKeyword(EDNSymbol("bar/foo", Some("bar")))
  )
))

Some people will think Any is a bit too large and I agree but it’s quite practical to use. Moreover, using validation explained a bit later, you can parse your EDN and then map it to a stronger typed scala structure and then Any disappears.

Compile-time parsing with Macros

When you use static EDN structures in your Scala code, you can write them in their string format and scaledn can parse them at compile-time using Scala macros and thus prevent a lot of errors you can encounter in dynamic languages.

The macro mechanism is based on quasiquotes & whitebox macro contexts which allow to infer types of your parsed EDN structures at compile-time. For example:

1
2
3
4
5

> val s:Long = EDN("\"toto\"")

[error]  found   : String("toto")
[error]  required: Long
[error]     val e: Long = EDN("\"toto\"")

Whooohooo magic :)

Classic Scala types

Here is how you can use it:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import scaledn._
import macros._

// All types are just for info and can be omitted below, the macro infers them quite well
val e: String = EDN("\"toto\"")

val bt: Boolean = EDN("true")

val bf: Boolean = EDN("false")

val l: Long = EDN("123")

val d: Double = EDN("123.456")

val bi: BigInt = EDN("123M")

val bd: BigDecimal = EDN("123.456N")

val s: EDNSymbol = EDN("foo/bar")

val kw: EDNKeyword = EDN(":foo/bar")

// Homogenous collection inferred as Vecto[String]
val vector: Vector[String] = EDN("""["tata" "toto" "tutu"]""")

// multiple heterogenous values inferred as Seq[Any]
val s = EDNs("""(1 2 3) "toto" [true false] :foo/bar""")
// note the small s at the end of EDN to inform the macro there are several values

Shapeless heterogenous collections

EDN allows to manipulate heterogenous collections. In Scala, when one thinks heterogenous collection, one thinks Shapeless. Scaledn macros can parse & map your EDN stringified structures to Scala strongly typed structures.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import scaledn._
import macros._

import shapeless.{HNil, ::}
import shapeless.record._
import shapeless.syntax.singleton._

// Heterogenous list
val s = EDNH("""(1 "toto" true)""")
s should equal (1L :: "toto" :: true :: HNil)

// Heterogenous Map/Record
val s3 = EDNH("""{1 "toto" true 1.234 "foo" (1 2 3)}""")
s3 should equal (
  1L ->> "toto" ::
  true ->> 1.234 ::
  "foo" ->> List(1L, 2L, 3L) ::
  HNil
)

please note the H in EDNH for heterogenous

I must say using these macros, it might be even simpler to write Shapeless hlists or records than using scala API ;)

Macro API

Scaledn provides different macros depending on the depth of introspection you require in your collection with respect to heterogeneity.

Have a look directly at Macro API

Mixing macro with Scala string interpolation

Following ideas implemented by Daniel James in Datomisca, scaledn proposes to use String interpolation mixed with parsing macro such as:

1
2
3
4
5
6
7
8
9
10
11
12

import scaledn._
import macros._

import shapeless.{HNil, ::}

val l = 123L
val s = List("foo", "bar")

val r: Long = EDN(s"$l")

val r1: Seq[Any] = EDN(s"($l $s)")
val r2: Long :: List[String] :: HNil = EDNH(s"($l $s)")

Nothing to add, macros are cool sometimes :)

Runtime validation of EDN to Scala

When writing REST or external API, the received data can never be trusted before being validated. So, you generally try to validate what is received and map it to a strong-typed structures. For example:

1
2
3
4
5
6
7
8
9
10
11

// parse the received string input
parseEDN("""{ 1 "toto" 2 "tata" 3 "tutu" }""")
// then validate it to a Scala type
  .map(validate[Map[Long, String]])
  .success.value should be (
    play.api.data.mapping.Success(Map(
      1L -> "toto",
      2L -> "tata",
      3L -> "tutu"
    ))
  )

The validation API is the following:

1
2
3
4

import scaledn._
import validate._

def validate[T](edn: EDN)(implicit r: RuleLike[EDN, T]): Validation[EDN, T] = r.validate(edn)

Scaledn validation is based on Generic Validation API developed by my MFGLabs’s colleague & friend Julien Tournay. This API was developed for Play Framework & Typesafe last year to generalize Json validation API to all data formats. But it will never be integrated in Play as Typesafe considers it to be too pure Scala & pure FP-oriented. Yet, we use this API in production at MFGLabs and maintain/extend it ourselves.

As explained before, Scaledn parser parses EDN values directly to Scala types as they are bijective so validation is often just a runtime cast and not very interesting in general.

What’s much more interesting is to validate to Shapeless HList, Records and even more interesting to CaseClasses & Tuples based on Shapeless fantastic auto-generated Generic macros.

Let’s take a few examples to show the power of this feature:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import scaledn._
import validate._

import play.api.data.mapping._
import shapeless.{HNil, ::}

case class CP(cp: Int)
case class Address(street: String, cp: CP)
case class Person(name: String, age: Int, addr: Address)
// Remark that NO implicits must be declared on our case classes

// HLISTS
parseEDN("""(1 "toto" true nil)""").map(
  validate[Long :: String :: Boolean :: EDNNil.type :: HNil]
).success.value should be (
  Success(1L :: "toto" :: true :: EDNNil :: HNil)
)

// TUPLES
parseEDN("""("toto" 34 {"street" "chboing", "cp" {"cp" 75009}})""").map(
  validate[Tuple3[String, Int, Address]]
).success.value should be (
  Success(("toto", 34, Address("chboing", CP(75009))))
)

// CASECLASSES
parseEDN("""("toto" 34 ("chboing" (75009)))""").map(
  validate[Person]
).success.value should be (
  Success(Person("toto", 34, Address("chboing", CP(75009))))
)

parseEDN("""{"name" "toto", "age" 34, "addr" {"street" "chboing", "cp" {"cp" 75009}}}""").map(
  validate[Person]
).success.value should be (
  Success(Person("toto", 34, Address("chboing", CP(75009))))
)

I think here you can see the power of this validation feature without writing any boilerplate…

Serializing Scala to EDN

Using Generic Validation API, you can also write scala structures to any other data format.

Scaledn provides serialization from scala structures to EDN Strings. For example:

1
2
3
4
5

import scaledn._
import write._

toEDNString("toto") should equal ("\"toto\"")
toEDNString(List(1, 2, 3)) should equal ("""(1 2 3)""")

The write API is the following:

1
2
3
4

import scaledn._
import write._

def toEDNString[I](i: I)(implicit w: WriteLike[I, String]): String = w.writes(i)

Once again, what’s more interesting is using shapeless & caseclasses & tuples.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import scaledn._
import write._

import shapeless.{HNil, ::}

// HLIST
toEDNString(1 :: true :: List(1L, 2L, 3L) :: HNil) should equal ("""(1 true (1 2 3))""")

// TUPLE
toEDNString((23, true)) should equal ("""(23 true)""")

// CASE CLASS
case class Address(street: String, cp: Int)
case class Person(name: String, age: Int, addr: Address)
// Remark that NO implicits must be declared on our case classes

toEDNString(Person("toto", 34, Address("chboing", 75009))) should equal (
  """{"name" "toto", "age" 34, "addr" {"street" "chboing", "cp" 75009}}"""
)

TODO

This project is a first draft so it requires a bit more work.

Here are a few points to work on:

• patch remaining glitches/bugs
• write more tests for all cases
• study streamed parser asap
• write sample apps

Don’t hesitate to test, find bugs, contribute, give remarks, ideas…

Have fun in EDN world…

Not an article, just some reflections on this idea…

You know what is a functor?

• 2 categories C & D, simplifying a category as:

  • a set of objects with some arrows/morphisms/functions between objects: f: x -> y
  • those morphisms are associative h . (g . f) = (h . g) .f where . is the composition (g . f)(x) = g(f(x)))
  • for each object there is an identity morphism id(x) = x -> x

• a functor F associates :

  • each object x of C with an object of F(x) of D.
  • each morphism f: x -> y of C with an element F(f): F(x) -> F(y) of D such that:
    • F(id(x)) = id(F(x))
    • F(g . f) = F(g) . F(f)

A Functor is a mapping (an homomorphism) between categories that conserves the structure of the category (the morphisms, the relation between objects) whatever the kind of objects those categories contain.

In scalaz, here is the definition of a Functor:

1
2
3
4
5
6
7

trait Functor[F[_]] {
  ////

  /** Lift `f` into `F` and apply to `F[A]`. */
  def map[A, B](fa: F[A])(f: A => B): F[B]
  ...
}

You can see the famous map function that you can find in many structures in Scala : List[_], Option[_], Map[_, _], Future[_], etc…

Why? Because all these structures are Functors between the categories of Scala types…

Math is everywhere in programming & programming is Math so don’t try to avoid it ;)

So you can write a Functor[List] or Functor[Option] as those structures are monoids.

Now let’s consider HList the heterogenous List provided by Miles Sabin’s fantastic Shapeless. HList looks like a nice Functor.

1
2
3
4

(1 :: true :: HNil) map (
   (i:Int)     => i.toString ,
   (b:Boolean) => b.toString)
) => ("1" :: "true" :: HNil)

Ok, it’s a bit more complex as this functor requires not one function but several for each element type constituting the HList, a kind of polymorphic function. Hopefully, Shapeless provides exactly a structure to represent this: Poly

What about writing a functor for HList?

Scalaz Functor isn’t very helpful (ok I just copy the HMonoid text & tweak it ;)):

To be able to write a Functor of HList, we need something else based on multiple different types…

I spent a few hours having fun on this idea with Shapeless and tried to implement a Functor for heterogenous structures like HList, Sized and even not heterogenous structures.

Here are the working samples.

Here is the code based on pseudo-dependent types as shapeless.

The signature of the HFunctor as a map function as expected:

1
2
3
4
5
6
7
8
9

trait HFunctor[HA, F <: Poly] {
    type Real

    trait HMapper[P <: Poly, In] extends DepFn1[In] { type Out }

    val hmapper: HMapper[F, HA]

    def map(ha: HA)(f: F): hmapper.Out = hmapper(ha)
  }

This is just a sandbox to open discussion on this idea so I won’t explain more and let the curious ones think about it…

Have F(un)!

Not an article, just some reflections on this idea…

You know what is a monoid?

• a binary operation taking 2 elements and giving another element e x e -> e (aka a SemiGroup)
• a Identity id element id . e = e . id = e (also called zero element)

(and some associativity)

In scalaz, here is the definition:

1
2
3
4
5
6
7

trait Monoid[F] extends Semigroup[F] { self =>
  ////
  /** The identity element for `append`. */
  def zero: F
  def append(f1: F, f2: => F): F
  ...
}

You can see the zero & the SemiGroup.append operations, right?

So you can write a Monoid[Int] or Monoid[List[A]] as those structures are monoids.

Now let’s consider HList the heterogenous List provided by Miles Sabin’s fantastic Shapeless. HList looks like a nice monoid.

1

(1 :: "toto" :: HNil) ++ (true :: HNil) => (1 :: "toto" :: true :: HNil)

What about writing a monoid for HList?

Scalaz monoid isn’t very helpful because our monoid operations would look like:

1
2

def zero: HNil.type
  def append(f1: H1 <: HList, f2: => H2 <: HList): H3 <: HList

So, to be able to write a Monoid of HList, we need something else based on multiple different types…

I spent a few hours having fun on this idea with Shapeless and tried to implement a Monoid for heterogenous structures like HList, Nat, Sized and even not heterogenous structures.

Here are the working samples.

Here is the code based on pseudo-dependent types as shapeless.

The signature of the HMonoid shows the zero and the Semigroup as expected:

1

trait HMonoid[A, B] extends HZero with HSemiGroup[A, B]

This is just a sandbox to open discussion on this idea so I won’t explain more and let the curious ones think about it…

Have Monoids of Fun!

The code & sample apps can be found on Github

Forget the buzz title, this project is still very draft but it’s time to expel it out of my R&D sandbox as imperfect as it might be… Before I lose my sanity while wandering in Scala macro hygiene ;)

What?

Daemonad is a nasty Scala macro that aims at:

• marking where you manipulate monads or stacks of monads
• compile-checking monadic behavior & implicit monad instances
• allowing to snoop monad values deep into (some) monad stacks in the same way as ScalaAsync i.e. in a pseudo-imperative way.

This project is NOT yet stable, NOT very robust so use it at your own risks but we can discuss about it…

Here is what you can write right now.

1
2
3
4
5
6
7
8
9

Await.result(
  monadic[Future, Option] {
    val a = Future ( Some(9) )
    val b = Some(7)
    val c = 10
    if(snoop2(a) < 10) snoop1(b) + 10
    else c
  }, duration.Duration("1 second")
) should equal (Some(17))

Motivations

1 - Experiment writing a very ugly Scala macro

I wanted to write a huge & complex Scala macro that would move pieces of code, create more code, types etc…

I wanted to know the difficulties that it implies.

I felt reckless, courageous!

Nothing could stop me!!!!

Result: I was quite insane and I will certainly write a post-mortem article about it to show the horrible difficulties I’ve encountered. My advice: let people like hit their head against the wall and wait for improved hygienic macros that should come progressively before writing big macros ;)

2 - Investigate ScalaAsync generalization to all monads + (some) monad stacks

I had investigated ScalaAsync code and thought it would be possible to generalize it to all kinds of monads and go further by managing monad stacks.

Result : Simple monads are easy to manage (as seen also in scala-workflow which I discovered very recently) and some monad stacks can be managed with Scalaz monad transformers.

But don’t think you can use all kinds of monad transformers: the limits of Scala compiler with type-lambdas in macros and my very own limits blocked me from going as far as I expected.

So for now, it can manage Future/Option/List stacks & also Either \/ using type aliases.

3 - Explicitly Mark monadic blocks

There are 2 ways of seeing monads:

You don’t need or you don’t want to know what is a monad…

… And yet you use it everyday/everywhere.

This is what most of us (and it’s so shameful) do using those cool map/flatMap functions provided by Scala libraries that allow to access the values inside Future, List, Option in a protected way etc… That’s enough for you need in your everyday life, right?

You want to know or you know what is a monad …

… and you want to use them on purpose.

This is what hippy developers do in advanced Scala using Scalaz or even crazier in pure FP languages like Haskell.

Guess what I prefer?

Here is the kind of code I’d like to write :

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// I write my datastructure without any map/flatMap function
case class Toto[A](a: A)

// Hey I proved Toto was a monad (yes believe me)

// Let's bring this concept into my scope
implicit object TotoMonad extends Monad[Toto] {
  def bind[A, B](fa: Toto[A])(f: A => Toto[B]): Toto[B] = {
    f(fa.a)
  }

  def point[A](a: => A): Toto[A] = Toto(a)
}

...
// I create my toto
val toto = Toto("this is toto")

...

// Suddenly I decide that I must use this monadic behavior of toto
monadic[Toto] {
  val a = <snoop_value_inside_monad>(toto) // outside the monadic block, you shall not do that
  do_something_with_value(a)
  // The compiler takes care that my structure is used in a pure monadic way
  // and returns a monad Toto of the right type
}
...

Ok I speak about pure functional programming and then about snooping the value out of the monad. This might seem a bit useless or even stupid compared to using directly Monad facilities. I agree and I still wonder about the sanity of this project but I’m stubborn and I try to finish what I start ;)

Back to code Sample

1
2
3
4
5
6
7
8
9

Await.result(
  monadic[Future, Option] {
    val a = Future ( Some(9) )
    val b = Some(7)
    val c = 10
    if(snoop2(a) < 10) snoop1(b) + 10
    else c
  }, duration.Duration("1 second")
) should equal (Some(17))

What does it do ?

• monadic marks the monadic block
• monadic[Future, Option] declares that you manipulate a stack Future[Option] (and no other)
• snoopX means that you want to snoop the monad value at X-th level (1, 2, 3, 4 and no more for now)
• the macro checks for implicit instances of monads (here List, Option, Future) and monad transformers (here OptionT & ListT) for this stack
• the macro translates this code into crazy embedded Monad.bind/point/lift/run…
• snoop2 is used in first position: if you have used snoop1, the macro would have rejected your monadic block. It’s logical, when you use flatMap, you always start with the deeper stack of monad and I chose not to change the order of your code as I find this macro is already far too intrusive :)

I’m sure you don’t want to see the code you would have to write for this, this is quite long and boring.

Let just say that this code is generated by a Scala macro for you.

The current generated code isn’t optimized at all and quite redundant but this is for next iterations.

What is working ?

• stacks with List/Option/Either up to depth 4
• custom Monads
• a few preliminary checkings that prevent you from writing stupid code but not so much
• if/then/else & case/match in some cases

What isn’t working ?

• monadic block assigned to a val not explicitly typed.
• many things or edge-cases with mixed monad depth and if/match.
• can’t use advanced monad transformers like StateT or WriterT in monadic block because Scala compiler doesn’t allow what I expected with type lambdas. This needs to be studied further.

A very stupid example to finish with 4-depth stack

1
2
3
4
5
6
7
8
9
10
11
12

it should """snoop4 on stupid stack""" in {
    type S[T] = ({ type l[T] = \/[String, T] })#l[T]
    Await.result(
      monadic[Future, S, List, Option] {
        val a: Future[S[List[Option[Int]]]] = Future(\/-(List(Some(5), Some(10))))
        val b: S[List[Option[Int]]] = \/-(List(Some(1), Some(2)))
        val c: List[Option[Int]] = List(Some(3), Some(4))
        val d: Option[Int] = Some(2)
        (snoop4(a) + snoop3(b) * 2 - snoop2(c)) / snoop1(d)
      }, duration.Duration("1 second")
    ) should equal (\/-(List(Some(2), Some(1), Some(3), Some(2), Some(4), Some(4), Some(5), Some(5))))
  }

Note that:

• you have to use a type alias to Scalaz \/ to one parametric type.
• you have to help a bit the compiler about type alias S or it will infer \/[A, B] which is not what we want for the monad. This might seem tedious but I’ll study if I can go around this.
• look at the result: you have a 2 elements list and at the end, you have a 8 elements list: WHAT???? No it’s normal, this is the result flatmap between first and second and third list. 2*2*2 = 8… nothing strange but it can be surprising at first glance ;)

TODO

• refactor all code because it’s ugly, not robust and redundant!!!
• rely on MonadTrans[F[_], _] instead of hardcoding monad transformers as now.
• accept custom MonadTrans provided in the user code.
• steal some inspiration from scala-workflow because I find this code cool.

Special Thanks

• Eugene Burmako (@xeno_by) for helping me each time I was lost in macros
• Jason Zaugg (@retronym) for Scala Async and splicer
• Daniel James (@dwhjames) for the snoop name
• Thibaut Duplessis (@ornicar) for the monad stack idea

Have a look at the code on Github.

Have snoop22(macrofun)!

Synopsis

The code & sample apps can be found on Github

Zpark-Zstream I article was a PoC trying to use Scalaz-Stream instead of DStream with Spark-Streaming. I had deliberately decided not to deal with fault-tolerance & stream graph persistence to keep simple but without it, it was quite useless for real application…

[Part 3/3] Fancy Spark Machine Learning with NIO client/server & DStream…

Let’s remind that I’m not an expert in ML but more a student. So if I tell or do stupid ML things, be indulgent ;)

Here is what I propose:

• Train a collaborative filtering rating model for a recommendation system (as explained in Spark doc there) using a first NIO server and a client as presented in part 2.

• When model is trained, create a second server that will accept client connections to receive data.

• Stream/merge all received data into one single stream, dstreamize it and perform streamed predictions using previous model.

Train collaborative filtering model

Training client

As explained in Spark doc about collaborative filtering, we first need some data to train the model. I want to send those data using a NIO client.

Here is a function doing this:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

//////////////////////////////////////////////////
// TRAINING CLIENT
def trainingClient(addr: InetSocketAddress): Process[Task, Bytes] = {

  // naturally you could provide much more data
  val basicTrainingData = Seq(
    // user ID, item ID, Rating
    "1,1,5.0",
    "1,2,1.0",
    "1,3,5.0",
    "1,4,1.0",
    "2,4,1.0",
    "2,2,5.0"
  )
  val trainingProcess =
    Process.emitAll(basicTrainingData map (s => Bytes.of((s+"\n").getBytes)))

  // sendAndCheckSize is a special client sending all data emitted 
  // by the process and verifying the server received all data 
  // by acknowledging all data size
  val client = NioClient.sendAndCheckSize(addr, trainingProcess)

  client
}

Training server

Now we need the training NIO server waiting for the training client to connect and piping the received data to the model.

Here is a useful function to help creating a server as described in previous article part:

1
2
3
4
5
6
7
8
9
10
11
12
13

def server(addr: InetSocketAddress): (Process[Task, Bytes], Signal[Boolean]) = {

  val stop = async.signal[Boolean]
  stop.set(false).run

  // this is a server that is controlled by a stop signal
  // and that acknowledges all received data by their size
  val server =
    ( stop.discrete wye NioServer.ackSize(addr) )(wye.interrupt)

  // returns a stream of received data & a signal to stop server
  (server, stop)
}

We can create the training server with it:

1
2
3
4

val trainingAddr = NioUtils.localAddress(11100)
//////////////////////////////////////////////////
// TRAINING SERVER
val (trainingServer, trainingStop) = server(trainingAddr)

trainingServer is a Process[Task, Bytes] streaming the training data received from training client. We are going to train the rating model with them.

Training model

To train a model, we can use the following API:

1
2
3
4
5
6
7
8

// the rating with user ID, product ID & rating
case class Rating(val user: Int, val product: Int, val rating: Double)

// A RDD of ratings
val ratings: RDD[Rating] = ...

// train the model with it
val model: MatrixFactorizationModel = ALS.train(ratings, 1, 20, 0.01)

Building RDD[Rating] from server stream

Imagine that we have a continuous flow of training data that can be very long.

We want to train the model with just a slice of this flow. To do this, we can:

• dstreamize the server output stream
• run the dstream for some time
• retrieve the RDDs received during this time
• union all of those RDDs
• push them to the model

Here is the whole code with previous client:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

val trainingAddr = NioUtils.localAddress(11100)

//////////////////////////////////////////////////
// TRAINING SERVER
val (trainingServer, trainingStop) = server(trainingAddr)

//////////////////////////////////////////////////
// TRAINING CLIENT
val tclient = trainingClient(trainingAddr)

//////////////////////////////////////////////////
// DStreamize server received data
val (trainingServerSink, trainingDstream) = dstreamize(
  trainingServer
      // converts bytes to String (doesn't care about encoding, it shall be UTF8)
      .map  ( bytes => new String(bytes.toArray) )
      // rechunk received strings based on a separator \n 
      // to keep only triplets: "USER_ID,PROD_ID,RATING"
      .pipe (NioUtils.rechunk { s:String => (s.split("\n").toVector, s.last == '\n') } )
  , ssc
)

//////////////////////////////////////////////////
// Prepare dstream output 
// (here we print to know what has been received)
trainingDstream.print()

//////////////////////////////////////////////////
// RUN

// Note the time before
val before = new Time(System.currentTimeMillis)

// Start SSC
ssc.start()

// Launch server
trainingServerSink.run.runAsync( _ => () )

// Sleeps a bit to let server listen
Thread.sleep(300)

// Launches client and awaits until it ends
tclient.run.run

// Stop server
trainingStop.set(true).run

// Note the time after
val after = new Time(System.currentTimeMillis)

// retrieves all dstreamized RDD during this period
val rdds = trainingDstream.slice(
  before.floor(Milliseconds(1000)), after.floor(Milliseconds(1000))
)

// unions them (this operation can be expensive)
val union: RDD[String] = new UnionRDD(ssc.sparkContext, rdds)

// converts "USER_ID,PROD_ID,RATING" triplets into Ratings
val ratings = union map { e =>
  e.split(',') match {
    case Array(user, item, rate) => Rating(user.toInt, item.toInt, rate.toDouble)
  }
}

// finally train the model with it
val model = ALS.train(ratings, 1, 20, 0.01)

// Predict
println("Prediction(1,3)=" + model.predict(1, 3))

//////////////////////////////////////////////////
// Stop SSC
ssc.stop()

Run it

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

-------------------------------------------
Time: 1395079621000 ms
-------------------------------------------
1,1,5.0
1,2,1.0
1,3,5.0
1,4,1.0
2,4,1.0
2,2,5.0

-------------------------------------------
Time: 1395079622000 ms
-------------------------------------------

-------------------------------------------
Time: 1395079623000 ms
-------------------------------------------

-------------------------------------------
Time: 1395079624000 ms
-------------------------------------------

-------------------------------------------
Time: 1395079625000 ms
-------------------------------------------

Prediction(1,3)=4.94897842056338

Fantastic, we have trained our model in a very fancy way, haven’t we?

Personally, I find it interesting that we can take advantage of both APIs…

Predict Ratings

Now that we have a trained model, we can create a new server to receive data from clients for rating prediction.

Prediction client

Firstly, let’s generate some random data to send for prediction.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

//////////////////////////////////////////////////
// PREDICTION CLIENT
def predictionClient(addr: InetSocketAddress): Process[Task, Bytes] = {

  // PREDICTION DATA
  def rndData =
    // userID
    (Math.abs(scala.util.Random.nextInt) % 4 + 1).toString +
    // productID
    (Math.abs(scala.util.Random.nextInt) % 4 + 1).toString +
    "\n"

  val rndDataProcess = Process.eval(Task.delay{ rndData }).repeat

  // a 1000 elements process emitting every 10ms
  val predictDataProcess =
    (Process.awakeEvery(10 milliseconds) zipWith rndDataProcess){ (_, s) => Bytes.of(s.getBytes) }
      .take(1000)

  val client = NioClient.sendAndCheckSize(addr, predictDataProcess)

  client
}

Prediction server

1
2
3
4

val predictAddr = NioUtils.localAddress(11101)
//////////////////////////////////////////////////
// PREDICTION SERVER
val (predictServer, predictStop) = server(predictAddr)

Prediction Stream

predictServer is the stream of data to predict. Let’s stream it to the model by dstreamizing it and transforming all built RDDs by passing them through model

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

//////////////////////////////////////////////////
// DStreamize server
val (predictServerSink, predictDstream) = dstreamize(
  predictServer
      // converts bytes to String (doesn't care about encoding, it shall be UTF8)
      .map  ( bytes => new String(bytes.toArray) )
      // rechunk received strings based on a separator \n
      .pipe (NioUtils.rechunk { s:String => (s.split("\n").toVector, s.last == '\n') } )
  , ssc
)

//////////////////////////////////////////////////
// pipe dstreamed RDD to prediction model
// and print result
predictDstream map { _.split(',') match {
  // converts to integers required by the model (USER_ID, PRODUCT_ID)
  case Array(user, item) => (user.toInt, item.toInt)
}} transform { rdd =>
  // prediction happens here
  model.predict(rdd)
} print()

Running all in same StreamingContext

I’ve discovered a problem here because the recommendation model is built in a StreamingContext and uses RDDs built in it. So you must use the same StreamingContext for prediction. So I must build my training dstreamized client/server & prediction dstreamized client/server in the same context and thus I must schedule both things before starting this context.

Yet the prediction model is built from training data received after starting the context so it’s not known before… So it’s very painful and I decided to be nasty and consider the model as a variable that will be set later. For this, I used a horrible SyncVar to set the prediction model when it’s ready… Sorry about that but I need to study more about this issue to see if I can find better solutions because I’m not satisfied about it at all…

So here is the whole training/predicting painful code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

//////////////////////////////////////////////////
// TRAINING

val trainingAddr = NioUtils.localAddress(11100)

// TRAINING SERVER
val (trainingServer, trainingStop) = server(trainingAddr)

// TRAINING CLIENT
val tclient = trainingClient(trainingAddr)

// DStreamize server
val (trainingServerSink, trainingDstream) = dstreamize(
  trainingServer
      // converts bytes to String (doesn't care about encoding, it shall be UTF8)
      .map  ( bytes => new String(bytes.toArray) )
      // rechunk received strings based on a separator \n
      .pipe (NioUtils.rechunk { s:String => (s.split("\n").toVector, s.last == '\n') } )
  , ssc
)

// THE HORRIBLE SYNCVAR CLUDGE (could have used atomic but not better IMHO)
var model = new SyncVar[org.apache.spark.mllib.recommendation.MatrixFactorizationModel]
// THE HORRIBLE SYNCVAR CLUDGE (could have used atomic but not better IMHO)


//////////////////////////////////////////////////
// PREDICTING
val predictAddr = NioUtils.localAddress(11101)

// PREDICTION SERVER
val (predictServer, predictStop) = server(predictAddr)

// PREDICTION CLIENT
val pClient = predictionClient(predictAddr)

// DStreamize server
val (predictServerSink, predictDstream) = dstreamize(
  predictServer
      // converts bytes to String (doesn't care about encoding, it shall be UTF8)
      .map  ( bytes => new String(bytes.toArray) )
      // rechunk received strings based on a separator \n
      .pipe ( NioUtils.rechunk { s:String => (s.split("\n").toVector, s.last == '\n') } )
  , ssc
)

// Piping received data to the model
predictDstream.map {
  _.split(',') match {
    case Array(user, item) => (user.toInt, item.toInt)
  }
}.transform { rdd =>
  // USE THE HORRIBLE SYNCVAR
  model.get.predict(rdd)
}.print()

//////////////////////////////////////////////////
// RUN ALL
val before = new Time(System.currentTimeMillis)

// Start SSC
ssc.start()

// Launch training server
trainingServerSink.run.runAsync( _ => () )

// Sleeps a bit to let server listen
Thread.sleep(300)

// Launch training client
tclient.run.run

// Await SSC termination a bit
ssc.awaitTermination(1000)
// Stop training server
trainingStop.set(true).run
val after = new Time(System.currentTimeMillis)

val rdds = trainingDstream.slice(before.floor(Milliseconds(1000)), after.floor(Milliseconds(1000)))
val union: RDD[String] = new UnionRDD(ssc.sparkContext, rdds)

val ratings = union map {
  _.split(',') match {
    case Array(user, item, rate) => Rating(user.toInt, item.toInt, rate.toDouble)
  }
}

// SET THE HORRIBLE SYNCVAR
model.set(ALS.train(ratings, 1, 20, 0.01))

println("**** Model Trained -> Prediction(1,3)=" + model.get.predict(1, 3))

// Launch prediction server
predictServerSink.run.runAsync( _ => () )

// Sleeps a bit to let server listen
Thread.sleep(300)

// Launch prediction client
pClient.run.run

// Await SSC termination a bit
ssc.awaitTermination(1000)
// Stop server
predictStop.set(true).run