Pages

20 August 2017

Specs2 4.x

It has been now almost 3 years since the last major version of specs2. But why would a new major version of a testing library be even needed? Because the world doesn’t stop revolving!

In particular Scala can now be compiled to JavaScript and even natively. It is then only natural that specs2 users want to use specs2 on their Scala.js projects:

How hard can this be? First of all specs2 3.x relies on scalaz-stream which doesn’t have a Scala.js build, and more largely on scalaz (which has a Scala.js build).

This is where an issue becomes an opportunity! If I need to find an alternative to scalaz-stream why not remove scalaz altogether? Indeed it is very inconvenient for a test library to rely on third-party libraries which might conflict with users libraries. Because of this conflict I have to publish several versions of specs2 for different versions of scalaz (7.0.x, 7.1.x, 7.2.x and so on). This is pretty time-consuming, not only because of having to wait for more than 30 minutes for all the jars to be published but also because I have to adapt to some small differences between each scalaz version.

Removing scalaz and scalaz-stream is easier said than done:

  1. specs2 relies heavily on functional programming so I need the usual FP suspects: Functor, Applicative, Monad and so on

  2. specs2 interacts with the file system, writes out to the console, executes specifications concurrently so it needs a way to control effects

  3. specs2 is structured around the idea of a “stream of specification fragments” so it needs a streaming abstraction

All of this before I can even consider moving to a Scala.js build!

specs2 own FP / streaming infrastructure

Fortunately I have developed on the side many of the pieces needed above:

  1. eff is a library to handle effects, including concurrency and resources management

  2. producer is a simple streaming library designed to work with any Monad, including the Eff monad

  3. origami implements a notion of composable “folds” which can be applied to any stream, which is a pattern I (re)-discovered on earlier versions of specs2: reporting the results of an executed specification is just “folding” the specification, a bit like using foldLeft on a List

A copy of those libraries is now included in the specs2-common module.

The only thing missing was the FP abstractions. It turns out that it is not too hard to implement the whole menagerie:

  • Functor, Applicative, Monad, Traverse, Foldable, Monoid. Those typeclasses are very tied together
  • Tree and TreeLoc for manipulating trees

I am greatly indebted to the Scalaz and cats projects for this code, which I reduced to the only parts I needed for specs2. Those abstractions and datatypes are now available in the specs2-fp module (not intended for external use).

Execution model

At the heart of specs2 is a Fragment, which is a Description and an Execution yielding a Result. The Description holds an informal description of the specified system and the Execution runs some Scala code to verify that behaviour.

Using Scala.js imposes a big change is specs2. We can never await to the get the result of executing a Fragment. Also since scalaz.concurrent.Task is out of the equation, in specs2 4.x an Execution is implemented using a scala.concurrent.Future, like this:

case class Execution(
 run:       Option[Env => Future[() => Result]],
 executing: Option[Throwable Either Future[Result]])

This is a lot more complicated than just Future[Result]. Why is that?

run takes an action to execute () => Result, possibly concurrently Future[() => Result] and which might depend on the environment Env => Future[() => Result]. You might wonder why we don’t simply use Env => Future[Result]. This is to allow the user to add some behaviour around the result, for example with the AroundEach trait and the def around[R : AsResult](r: =>R): Result method.

We also want to “stream” results, which are executed concurrently and display them as soon as they are available. This is done by having executing holding either a Throwable if we could not even start the execution or the Result being currently computed.

Note that we can never call await in a Scala.js program so when we are executing a Fragment the best we can get is the equivalent of Future[Result] (in reality an Action[Result] which is a type possibly having more effects than just concurrency).

This triggers major changes in the implementation of specs2 and there are rippling effects up to the Runners used to run a specification. However most of the users of specs2 should not be concerned with these changes as the “user” DSL for specs2 isn’t modified. However if you are on the JVM and really need to await to get the result of a fragment execution you can do so by using one of the methods in the org.specs2.control.ExecuteActions object. For example result.runAction(executionEnv) will return an Error Either Result. Running the same method on Scala.js will throw an exception.

Scala.js modules

“specs2 runs on Scala.js” is ever going to be a partial truth. There are many limitations related to Scala.js and this means that, while specs2-core can be used on any Scala.js project, there are some specs2 features and modules which cannot be used with Scala.js. Here is a list (up to my current knowledge):

  1. the isolate argument to run each example in its own copy of the specification cannot be used (because this uses reflection to instantiate classes)

  2. no custom Selector/Executor/Printer/Notifier can be passed from the command line (for the same reason)

  3. the RandomSequentialExecution trait cannot be used because the implementation uses a TrieMap not available on Scala.js. This could be fixed in the future.

  4. running only examples which previously failed with was x on the command-line is not possible because this accesses file system

  5. all the modules where scala-xml is used don’t have a Scala.js version because there’s no scala-xml for Scala.js: specs2-form, specs2-html

  6. modules accessing the file system cannot be used: specs2-html, specs2-markdown

  7. the specs2-gwt module defines fragments based on the result of previous fragments, so it currently needs to await. This could be fixed potentially with a smart enough implementation in the future.

Breaking changes

A major version number is also the opportunity to fix some of the API flaws. I tried to limit the changes in the low-level API to the strict necessary but I also wanted to clean-up one important aspect of the high-level API.

When specs2 3.x came out I envisioned that there could be ways to create a specification from arguments passed on the command line, or from an ExecutionContext (to execute futures). With the addition of some traits like CommandLineArguments you could declare:

class MySpec extends Specification with CommandLineArguments {
  def is(args: CommandLine) = s2"""
    try this $ok
"""
}

or

class MySpec extends Specification { def is = s2"""
  try this $ex1
"""

  def ex1 = { implicit ec: ExecutionContext => Future(1) must be_==(1).await }
}

I later realized that it would be a lot more convenient to directly “inject” the command line arguments or the execution context in the specification like so:

class MySpec(args: CommandLine) extends Specification { def is = s2"""
  try this $ex1
"""

 def ex1 = { (1 + 1) must be_==(2).when(args.contains("universe-is-sane"))}
}
class MySpec(implicit ec: ExecutionContext) extends Specification { def is = s2"""
  try this $ex1
"""

  def ex1 = { Future(1) must be_==(1).await }
}

Now there were 2 ways to do the same thing, the second one being a lot easier than the first one. In specs2 4.x the traits supporting the first behavior org.specs2.specification.Environment, org.specs2.specification.CommandLineArguments, org.specs2.specification.ExecutionEnvironment have been removed. Same thing for the implicit conversions for functions like implicit ee: ExecutionEnv => Result to create examples.

This will demand some migration efforts for some users but the result will be more concise specifications.

Another word on injected execution environments. From specs2 3.8.9, 2 distinct execution environments are being used in specs2. One for the execution of the specification itself, the other one for the execution of the user examples (to map futures or await for results for example). This is to make sure that some incorrect user code doesn’t prevent the full specification from being executed.

With specs2 4.x we can go one step further and allow each specification to get its own dedicated execution environment and make sure that one specification execution doesn’t impact another one:

import org.specs2.specification.core.{Env, OwnExecutionEnv}

class MySpec extends(val env: Env) extends Specification with OwnExecutionEnv { def is = s2"""
    ...
"""
}

The injected env is used to pass command-line argument to a copy of the user execution env, implicitly available as a member of the OwnExecutionEnv trait. That trait also takes care of shutting down the execution environment once the specification has finished executing so as not to lose resources.

Feedback is essential

In conclusion, specs2 4.x should be a smaller gap from specs2 3.x than specs2 2.x to specs2 3.x. With a lot more to offer!

  1. no more dependencies for specs2-core

  2. a Scala.js version for specs2-core

  3. better control on execution environments

The release notes can be found here.

Please, please, please ask on gitter, on the mailing-list, on twitter if you have any issue. And a big thank you for reporting any mistake, big or small, that I may have made on this release!


21 September 2016

A neat trick from ICFP 2016

ICFP is an awesome conference where I can’t pretend I understand every presentation I’m attending but there is definitely lots of food for thought and some gems which makes it immensely worthwhile.

The gem I want to share here is a paper called “All sorts of permutations”.

This paper shows that you can use the filterM function with a predicate like a -> [True, False] and generate all the sublists of a list.

Similarly, you can evolve the sort function into sortM, taking a monadic comparison function: a -> a -> m Bool. If the comparison function is a -> a -> [True, False] the sortM function will generate the permutations of a list.

The paper goes on exploring if sortM generates all the permutations of a list or exactly the permutations of a list depending on the implementation of the sortM function (insertion sort, merge sort, quicksort,…) and tries to set various restrictions on the comparison function to get there.

At the end of the talk Philip Wadler asked the following question (paraphrased):

This is a neat trick to generate permutations but why should I teach it to my students, considering that there are more ‘classical’ implementations for getting the permutations of a list?"

I think that it is worth showing students because this sortM function gives us a very neat way of writing a QuickCheck generator for distinct elements!

It is indeed not uncommon to have to generate lists of distinct elements for testing applications. For example if you want to test a function scheduling different tasks and these tasks are supposed to be unique, but arriving in any order.

There is unfortunately no permutations :: [a] -> Gen [a] combinator in QuickCheck. So how can you produce random Lists with distinct elements?

One way is to take an existing list of distinct elements and shuffle it to get a random permutation of those elements. This is exactly what sortM gives us, we just need to use the right monad, which happens to be Gen instead of []!

Let’s take an example. Say you want to generate all the permutations of ["Homer", "Marge", "Bart", "Lisa"]

import Test.QuickCheck

-- An implementation of sortM
sortM :: Monad m => (a -> a -> m Bool) -> [a] -> m [a]
sortM _ [] = return []
sortM cmp (x : xs) = do
  ys <- sortM cmp xs
  insertM cmp x ys

insertM :: Monad m => (a -> a -> m Bool) -> a -> [a] -> m [a]
  insertM _ x [] = return [x]
  insertM cmp x yys@(y : ys) = do
    b <- cmp x y
    if b then return (x : yys)
    else fmap (y:) (insertM cmp x ys)

-- A generator for lists of distincts "Simpsons"
type Simpson = String

simpsons = ["Homer", "Marge", "Bart", "Lisa"]

distinctSimpsons :: Gen [Simpson]
distinctSimpsons = sortM (\_ _ -> choose(True, False)) simpsons

Let’s test it:

λ> sample distinctSimpsons
["Homer","Lisa","Marge","Bart"]
["Homer","Marge","Bart","Lisa"]
["Homer","Marge","Lisa","Bart"]
["Homer","Marge","Bart","Lisa"]
["Bart","Homer","Marge","Lisa"]
["Marge","Homer","Bart","Lisa"]
["Marge","Homer","Lisa","Bart"]
["Marge","Homer","Lisa","Bart"]
["Marge","Bart","Lisa","Homer"]
["Homer","Marge","Bart","Lisa"]
["Bart","Homer","Lisa","Marge"]

I think this answers Philip’s question, this is a nice application of the sortM “trick” which can be taught to students learning functional programming.

And what looks really promising is that this idea extends to the generation of other data structures. For example we can generate the partitions of a given set with groupByM:

partitionM :: Monad m => (a -> m Bool) -> [a] -> m ([a], [a])
partitionM _ [] = pure ([], [])
partitionM p (x : rest) = do
  b        <- p x
  (xs, ys) <- partitionM p rest
  if b then pure (x:xs, ys)
  else      pure (xs, x:ys)

groupByM :: Monad m => (a -> a -> m Bool) -> [a] -> m [[a]]
groupByM _ [] = pure []
groupByM cmp (x : rest) = do
  (xs, ys) <- partitionM (cmp x) rest
  ps       <- groupByM cmp ys
  pure $ (x:xs):ps

Then we can get all the set partitions of ["Homer","Lisa","Marge","Bart"]

λ> pl $ groupByM (\_ _ -> [True, False]) simpsons
[["Homer","Marge","Bart","Lisa"]]
[["Homer","Marge","Bart"],["Lisa"]]
[["Homer","Marge","Lisa"],["Bart"]]
[["Homer","Marge"],["Bart","Lisa"]]
[["Homer","Marge"],["Bart"],["Lisa"]]
[["Homer","Bart","Lisa"],["Marge"]]
[["Homer","Bart"],["Marge","Lisa"]]
[["Homer","Bart"],["Marge"],["Lisa"]]
[["Homer","Lisa"],["Marge","Bart"]]
[["Homer","Lisa"],["Marge"],["Bart"]]
[["Homer"],["Marge","Bart","Lisa"]]
[["Homer"],["Marge","Bart"],["Lisa"]]
[["Homer"],["Marge","Lisa"],["Bart"]]
[["Homer"],["Marge"],["Bart","Lisa"]]
[["Homer"],["Marge"],["Bart"],["Lisa"]]

or get some random partitions of that set:

λ> sample $ groupByM (\_ _ -> choose(True, False)) simpsons
[["Homer","Marge","Bart"],["Lisa"]]
[["Homer","Lisa"],["Marge","Bart"]]
[["Homer","Bart"],["Marge"],["Lisa"]]
[["Homer","Marge","Bart","Lisa"]]
[["Homer","Marge","Bart","Lisa"]]
[["Homer","Lisa"],["Marge"],["Bart"]]
[["Homer","Marge"],["Bart"],["Lisa"]]
[["Homer"],["Marge"],["Bart","Lisa"]]
[["Homer","Marge","Bart","Lisa"]]
[["Homer","Marge","Bart","Lisa"]]
[["Homer","Marge","Lisa"],["Bart"]]

Some partitions can be drawn twice but the generation should be uniform over the set of all possible partitions (it would be better to not take my word for it and prove it though :-)).

I suspect that this is more than a useful trick and that this deserves a more systematic treatment to understand all the data structures which we can generate by using the appropriate functions taking predicates or relations as parameters. But for now it is already useful as it is!


29 July 2015

Precise data types

Here is a new post about creating data types to encode different cases in applicationspp0pp, hosted on the Ambiata lab website: Precise data types.

24 December 2014

specs2-three

This post presents the next major version of specs2, specs2 3.0:

  • what are the motivations for a major version?
  • what are the main benefits and changes?
  • when will it be available?

The motivations

I started working on this new version a bit more than one year ago now. I had lots of different reasons for giving specs2 a good face-lift.

The Open Source reason

specs2 has largely been the effort of a single person. This has probably some advantages, like the possibility to maintain some kind of vision for the project, but also lots of drawbacks. Quality is one of them.

As a programmer I have all sorts of shortcomings. I have always been amazed by other people taking a look at my code and spotting obvious deficiencies either big or small (for example @jedws introduced named threads to me, much easier for debugging!). I want to maximize the possibility that other people will jump in to fix and extend the library as necessary (and be able to go on holidays for 3 weeks without a laptop :-)). Improving the code base can only help other people review my implementation.

The Design reason

In specs2 I’ve had this vision of a flow of “specification fragments” that would get created, executed, and then reported, possibly with different reporters for various output formats. This is not as easy at it seems:

  • I want fragments to be executed concurrently while being printed in sequence
  • the fragments should also be displayed as soon as executed
  • I want to be able to recreate a view of the sequence of fragments as a tree to be displayed in IDEs like Eclipse and Intellij

This is all done and working in specs2 < 3.0 but in a very clumsy way, subverting Scalaz Reducers to maintain state and try to compose reporters.

One of these reporters is a HTML reporter and I’ve always wanted to improve it. This was not something I was eager to change given the situation 1 year ago. Luckily scalaz-stream version 0.2 came out in December 2013 and allowed me to try out new ideas.

The Functional Programming reason

The major difference between specs and specs2 was the use of immutable data structures and the avoidance of exceptions for control flow. Yet there was still lots of side-effects!

I hadn’t fully grasped how to use the IO monad to structure my program. Fortunately I happen to work with the terrific @markhibberd and he showed me how to use a proper monad stack to track IO effects but also how to thread in configuration data and track errors.

The main benefits and changes

First of all, a happy maintainer! That goes without saying but my ability to fix bugs and add features will be improved a lot if I can better reason about the code :-).

Now for users…

For casual users there should be no changes! If you just use org.specs2.Specification or org.specs2.mutable.Specification with no other traits, you should not see any change (except in the User Guide, see below). For “advanced” users there are new benefits and API changes (in no particular order).

Refactored user guide

The existing User Guide has been divided into a lot more pages (around 60) and follows a pedagogical progression:

  • a Quick Start presenting a simple specification (and the mandatory link to the installation page)

  • some links from the Quick Start to the most common concepts: what is the structure of a Specification? Which matchers are available? How to run a Specification?

  • then, on each other page there is a presentation focusing on one topic plus additional links: "Now learn how to..." (what is the next thing you will probably need?) and "If you want to know more" (what is some more advanced topic that is related to this one?)

In addition to this refactoring there are some “tools” to help users find faster what they are looking for:

  • a search box

  • reference pages to summarize in one place some topics (matchers and run arguments for example)

  • a Troubleshooting page with the most common issues

You can have a first look at it here.

Generalized Reader pattern

One consequence of the “functional” re-engineering is that the environment is now available at different levels. By “environment”, I mean the org.specs.specification.core.Env class which gives you access to all the components necessary to execute a specification, among which:

  • the command line arguments
  • the lineLogger used to log results to the console (from Sbt)
  • the systemLogger used to log issues when instantiating the Specification for example
  • the execution environment, containing a reference to the thread pool used to execute the examples
  • the statsRepository to get and store execution statistics
  • the fileSystem which mediates all interactions with the file system (to read and write files)

I doubt that you will ever need all of this, but parts of the environment can be useful. For example, you can define the structure of your Specification based on command line arguments:

class MySpec extends Specification with CommandLineArguments { def is(args: CommandLine) = s2"""
  Do something here with a command line parameter ${args.valueOr("parameter1", "not found")}
"""
}

The CommandLineArguments uses your definition of the def is(args: CommandLine): Fragments method to build a more general method Env => Fragments which is the internal representation of a Specification (fragments that depend on the environment). This means that now you don’t have to skip examples based on condition (isDatabaseAvailable for example), you can simply remove them!

You can also use the environment, or part of it, to define examples:

class MySpec extends Specification { def is = s2"""

 Here are some examples using the environment.
 You can access
   the full environment                            $e1
   the command line arguments                      $e2
   the execution context to create a Scala future  $e3
   the executor service to create a Scalaz future  $e4

"""

  def e1 = { env: Env =>
    env.statisticsRepository.getStatistics(getClass.getName).runOption.flatten.foreach { stats =>
      println("the previous results for this specification are "+stats)
    }
    ok
  }

  def e2 = { args: CommandLine =>
    if (args.boolOr("doit", false)) success
    else skipped
  }

  def e3 = { implicit executionContext: ExecutionContext =>
    scala.concurrent.Future(1); ok
  }

  def e4 = { implicit executorService: ExecutorService =>
    scalaz.concurrent.Future(1); ok
  }
}

Better reporting framework

This paragraph is mostly relevant to people who want to extend specs2 with additional outputs. The reporting framework has been refactored around 4 concepts:

A Runner (for example the SbtRunner)

  • instantiates the specification class to execute
  • creates the execution environment (arguments, thread pool)
  • instantiates a Reporter
  • instantiates Printers and starts the execution

A Reporter

  • reads the previous execution statistics if necessary
  • selects the fragments to execute
  • executes the specification fragments
  • calls the printers for printing out the results
  • saves the execution statistics

A Printer

  • prepares the environment for printing
  • uses a Fold to print or to gather execution data. For example the TextPrinter prints results to the console as soon as they are available and the HtmlPrinter

A Fold

  • has a Sink[Task, (T, S)] (see scalaz-stream for the definition of a Sink) to perform side-effects (like writing to a file)
  • has a fold: (T, S) => S method to accumulate some state (to compute statistics for example, or create an index)
  • has an init: S element to initialize the state
  • has a last(s: S): Task[Unit] method to perform one last side-effect with the final state once all the fragments have been executed

It is unlikely that you will create a new Runner (except if you build an Eclipse plugin for example) but you can create custom reporters and printers by passing the reporter <classname> and printer <classname> options as arguments. Note also that Folds are composable so if you need 2 outputs you can create a Printer that will compose 2 folds into 1.

The Html printer

Pandoc

The Html printer has been reworked to use Pandoc as a templating system and Markdown engine. I decided this move to Pandoc for several reasons:

  • Pandoc is one of the libraries that is officially endorsing the CommonMark format
  • I’ve had less corner cases with rendering mixed html/markdown with Pandoc than previously
  • Pandoc opens the possibility to render other markup languages than CommonMark, LaTeX for example

However this comes with a huge drawback, you need to have Pandoc installed as a command line tool on your machine. If Pandoc is not installed, specs2 will use a default template renderer, but won’t render CommonMark.

Templates

I’ve extracted a specs2.html template (and a corresponding specs2.css stylesheet) and it is possible for you to substitute another template (with the html.template option) if you want your html files to be displayed differently. This template is using the Pandoc template system so it is pretty primitive but should still cover most cases.

Better API

The specs2 API has been split into a lot more traits to support various objectives:

  • support the new execution model with scalaz-stream

  • make it possible to separate the DSL methods from the core ones (see Lightweight spec)

  • offer a better Fragment API

Let’s start with the heart of specs2, the Fragment.

FragmentFactory methods

Advanced specs2 users need to tweak the creation of fragments. For example, when using a “template specification”:

abstract class DatabaseSpec extends Specification {
  override def map(fs: => Fragments): Fragments =
    step(startDb) ^ fs ^ step(closeDb)
}

In the DatabaseSpec you are using different methods to work with Fragments. The ^ method to append them, the step method to create a Step fragment. Those 2 methods are part of the Fragment API. Here is a list of the main changes, compared to specs2 < 3.0

  • first of all there is only one Fragment type (instead of Text, Step, Example,…). This type contains a Description and an Execution. By combining different types of Descriptions and Executions it is possible to recreate all the previous specs2 < 3.0 types

  • however you don’t need to create a Fragment by yourself, what you do is invoke the FragmentFactory methods: example, step, text,… This now unifies the notation between immutable and mutable specifications because in specs2 < 3.0 you would write step in a mutable specification and Step in an immutable one (Step is now deprecated)

  • there is no ExampleFactory trait anymore since it has been subsumed by methods on the FragmentFactory trait (so this will break code for people who were intercepting Example creation to inject additional behaviour)

Finally those “core” objects have been moved under the org.specs2.specification.core package, in order to restructure the org.specs2.specification package into

  • core: Fragment Description, SpecificationStructure…

  • dsl: all the syntactic sugar FragmentsDsl, ExampleDsl, ActionDsl…

  • process: the “processing” classes Selector, Executor, StatisticsRepository…

  • create: traits to create the specification FragmentFactory, AutoExamples, S2StringContext (for s2 string interpolation)…

FragmentsDsl methods

When you want to assemble Fragments together you will need the FragmentsDsl trait to do so (mixed-in the Specification trait, you don’t have to add it).

The result of appending 2 Fragments is a Fragments object. The Fragments class has changed in specs2 3.0. It doesn’t hold a reference to the specification title and the specification arguments anymore, this is now the role of the SpecStructure. So in summary:

  • a Specification is a function Env => SpecStructure

  • a SpecStructure contains: a SpecHeader, some Arguments and Fragments

  • Fragments is a sequence of Fragments (actually a scalaz-stream Process[Task, Fragment])

The FragmentsDsl api allows to combine almost everything into Fragments with the ^ operator:

  • a String to a Seq[Fragment]
  • 2 Fragments
  • 1 Fragments and a String

One advantage of this fine-grained decomposition of the fragments API is that there is now a Spec lightweight trait.

Lightweight Spec trait

Compilation times can be a problem with Scala and specs2 makes it worse by providing lots of implicit methods in a standard Specification to provide various DSLs. In specs2 3.0 there is a Spec trait which contains a reduced number of implicits to:

  • create a s2 string for an “Acceptance Specification”
  • create should and in blocks in a “Unit Specification”
  • to create expectations with must
  • to add arguments to the specification (like sequential)

If you use that trait and you find yourself missing an implicit you will have to either:

  • use the Specification class instead

  • search specs2 for the trait or object providing the missing implicit. There is no magic recipe for this but the MustMatchers trait and the Spec2StringContext trait should bring most of the missing implicits in scope

It is possible that this trait will be adjusted to find the exact balance between expressivity and compile times but I hope it will remain pretty stable.

Durations

When specs2 started, the package scala.concurrent.duration didn’t exist. This is why there was a Duration type in specs2 < 3.0 and a TimeConversions trait. Of course this introduced annoying collisions with the implicits coming from scala.concurrent.duration when that one came around.

There is no reason to go on using specs2 Durations anymore now so you can use the standard Scala durations everywhere specs2 expects a Duration.

Contexts

Context management has been slowly evolving in specs2. In specs2 3.0 we end up with the following traits:

  • BeforeAll do something before all the examples (you had to use a Step in specs2 < 3.0)
  • BeforeEach do something before each example (was BeforeExample in specs2 < 3.0)
  • AfterEach do something after each example (was AfterExample in specs2 < 3.0)
  • BeforeAfterEach do something before/after each example (was BeforeAfterExample in specs2 < 3.0)
  • ForEach[T] provide an element of type T (a “fixture”) to some each example (was FixtureExample[T] in specs2 < 3.0)
  • AfterAll do something after all the examples examples (you had to use a Step in specs2 < 3.0)
  • BeforeAfterAll do something before/after all the examples examples (you had to use a Step in specs2 < 3.0)

There are some other cool things you can do. For example set a time-out for all examples based on a command line parameter:

trait ExamplesTimeout extends EachContext with MustMatchers with TerminationMatchers {

  def context: Env => Context = { env: Env =>
    val timeout = env.arguments.commandLine.intOr("timeout", 1000 * 60).millis
    upTo(timeout)(env.executorService)
  }

  def upTo(to: Duration)(implicit es: ExecutorService) = new Around {
    def around[T : AsResult](t: =>T) = {
      lazy val result = t

      val termination =
        result must terminate(retries = 10,
                              sleep = (to.toMillis / 10).millis).orSkip((ko: String) => "TIMEOUT: "+to)

      if (!termination.toResult.isSkipped) AsResult(result)
      else termination.toResult
    }
  }
}

The ExamplesTimeout trait extends EachContext which is a generalization of the xxxEach traits. With the EachContext trait you get access to the environment to define the behaviour used to “decorate” each example. So, in that case, we use a timeout command line parameter to create an Around context that will timeout each example if necessary. You can also note that this Around context uses the executorService passed by the environment so you don’t have to worry about resources management for your Specification.

Included specifications

As I was reworking the implementation of specs2 I also looked for ways to simplify its internal model. In specs2 < 3.0 you can nest a specification inside another one. This adds some significant complexity because a nested specification have its own arguments, its own title. For example during the execution of the inner specification we need to be careful enough to override the outer specification arguments with the inner ones.

I decided to let go of this functionality in favor of a view of specifications as “referencing” each other, with 2 types of references:

  • “link” reference
  • “see” reference

The idea is to model dependency relationships with “link” and weaker relationships with “see” (when you just want to mention that some information is present in another specification).

Then there are 2 modes of execution:

  • the default one
  • the “all” mode

By default when a specification is executed, the Runner will try to display the status of “linked” specifications but not “see” specifications. If you use the all argument then we collect all the “linked” specifications transitively and run them respecting dependencies (if s1 has a link to s2, then s2 is executed first).

This is particularly important for HTML reporting when the structure of “link” references is used to produce a table of contents and “see” references are merely used to display HTML links.

Online specifications

I find this exciting even if I don’t know if I will ever use this feature! (it has been requested in the past though).

In specs2 < 3.0 there is a clear distinction between the “creation” time and the “execution” time for specification. Once you have defined your examples you can not add new ones based on your execution results. But wait! this is more or less the property of a Monad! “Produce an action based on the value returned by another action”. Since specs2 3.0 is using scalaz-stream Process under the covers which is a Monad, this means that it is now possible to do the following:

class WikipediaBddSpec extends Specification with Online { def is = s2"""

  All the pages mentioning the term BDD must contain a reference to specs2 $e1

"""

    def e1 = {
      val pages = Wikipedia.getPages("BDD")

      // if the page is about specs2, add more examples to check the links
      (pages must contain((_:Page) must mention("specs2"))) continueWith
       pagesSpec(pages)
    }

    /** create one example per linked page */
    def pagesSpec(pages: Seq[Page]): Fragments = {
      val specs2Links = pages.flatMap(_.getLinks).filter(_.contains("specs2"))

      s2"""
       The specs2 links must be active
       ${Fragments.foreach(specs2Links)(active)}
      """
    }

    def active(link: HtmlLink) =
      s2"""
      The page at ${link.url} must be active ${ link must beActive }"""
  }

The specification above is “dynamic” in the sense that it creates more examples based on the tested data. All Wikipedia pages for BDD must mention “specs2” and for each linked page (which we can’t know in advance) we create a new example specifying that the link must be active.

ScalaCheck

The ScalaCheck trait has been reworked and extended to provide the following features:

  • you can specify Arbitrary[T], Gen[T], Shrink[T], T => Pretty instances at the property level (for any or all of the arguments)
  • you can easily collect argument values by appending .collectXXX to the property (XXX depends on the argument you want to collect. collect1 for the first, collectAll for all)
  • you can override default parameters from the command line. For example pass scalacheck.mintestsok 10000
  • you can set individual before, after actions to be executed before and after the property executes to do some setup/teardown

Also, specs2 was previously doing some message reformatting on top of ScalaCheck but now the ScalaCheck original messages have been preserved to keep the consistency between the 2 libraries.

Note: the ScalaCheck trait stays in the org.specs2 package but all the traits it depends on now live in the org.specs2.scalacheck package.

Bits and pieces

This section is about various small things which have changed with specs2 3.0:

Implicit context

There is no more implicit context when you use the .await method to match futures. This means that you have to either import the scala.concurrent.ExecutionContext.global context or to use a function ExecutionContext => Result to define your examples:

s2"""
An example using an ExecutionContext $e1
"""

  def e1 = { implicit ec: ExecutionContext =>
    // use the context here
    ok
  }
Foreach methods

It is possible now to create several examples or results with a foreach method which will not return Unit:

// create several examples
Fragment.foreach(1 to 10)(i => "example "+i ! ok)

// create several examples with breaks in between
Fragments.foreach(1 to 10)(i => ("example "+i ! ok) ^ br)

// create several results for a sequence of numbers
Result.foreach(1 to 10)(i => i must_== i)
Removed syntax
  • (action: Any).before to create a “before” context is removed (same thing for after)
  • function.forAll to create a Prop from a function
Dependencies
  • specs2 3.0 uses scalacheck 1.12.1
  • you need to use a recent version of sbt, like 0.13.7
  • you need to upgrade to scalaz-specs2 0.4.0-SNAPSHOT for compatibility

Can I use it?

specs2 3.0 is now available as specs2-core-3.0-M2 on Sonatype. I am making it available for early testing and feedback. Please use the mailing-list or the github issues to ask questions and tell me if there is anything going wrong with this new version. I will incorporate your comments in this blog post, serving as a migration guide.

Special thanks

  • to Clinton Freeman who started the re-design of the specs2 home page more than one year ago and sparked this whole refactoring
  • to Pavel Chlupacek and Frank Thomas for patiently answering many of my questions about scalaz-stream
  • to Paul Chiusano for starting scalaz-stream in the first place!
  • to Mark Hibberd for his guidance with functional programming

06 March 2014

Streaming with previous and next

The Scalaz streams library is very attractive but it might feel unfamiliar because this is not your standard collection library.

This short post shows how to produce a stream of elements from another stream so that we get a triplet with: the previous element, the current element, the next element.

With Scala collections

With regular Scala collections, this is not too hard. We first create a list of all the previous elements. We create them as options because there will not be a previous element for the first element of the list. Then we create a list of next elements (also a list of options) and we zip everything with the input list:

def withPreviousAndNext[T] = (list: List[T]) => {
  val previousElements = None +: list.map(Some(_)).dropRight(1)
  val nextElements     = list.drop(1).map(Some(_)) :+ None

  // plus some flattening of the triplet
  (previousElements zip list zip nextElements) map { case ((a, b), c) => (a, b, c) }
}

withPreviousAndNext(List(1, 2, 3))

> List((None,1,Some(2)), (Some(1),2,Some(3)), (Some(2),3,None))

And streams

The code above can be translated pretty straightforwardly to scalaz processes:

def withPreviousAndNext[F[_], T] = (p: Process[F, T]) => {
  val previousElements = emit(None) fby p.map(Some(_))
  val nextElements     = p.drop(1).map(Some(_)) fby emit(None)

  (previousElements zip p zip nextElements).map { case ((a, b), c) => (a, b, c) }
}

val p1 = emitAll((1 to 3).toSeq).toSource
withPreviousAndNext(p1).runLog.run

> Vector((None,1,Some(2)), (Some(1),2,Some(3)), (Some(2),3,None))

However what we generally want with streams is combinators which you can pipe onto a given Process. We want to write

def withPreviousAndNext[T]: Process1[T, T] = ???

val p1 = emitAll((1 to 3).toSeq).toSource
// produces the stream of (previous, current, next)
p1 |> withPreviousAndNext

How can we write this?

As a combinator

The trick is to use recursion to keep state and this is actually how many of the process1 combinators in the library are written. Let's see how this works on a simpler example. What happens if we just want a stream where elements are zipped with their previous value? Here is what we can write:

def withPrevious[T]: Process1[T, (Option[T], T)] = {

  def go(previous: Option[T]): Process1[T, (Option[T], T)] =
    await1[T].flatMap { current =>
      emit((previous, current)) fby go(Some(current))
    }

  go(None)
}

val p1 = emitAll((1 to 3).toSeq).toSource
(p1 |> withPrevious).runLog.run

> Vector((None,1), (Some(1),2), (Some(2),3))

Inside the withPrevious method we recursively call go with the state we need to track. In this case we want to keep track of each previous element (and the first call is with None because there is no previous element for the first element of the stream). Then go awaits a new element. Each time there is a new element, we emit it, then call recursively go which is again going to wait for the next element, knowing that the new previous element is now current.

We can do something similar, but a bit more complex for withNext:

def withNext[T]: Process1[T, (T, Option[T])] = {
  def go(current: Option[T]): Process1[T, (T, Option[T])] =
    await1[T].flatMap { next =>
      current match {
        // accumulate the first element
        case None    => go(Some(next))
        // if we have a current element, emit it with the next
        // but when there's no more next, emit it with None
        case Some(c) => (emit((c, Some(next))) fby go(Some(next))).orElse(emit((c, None)))
      }
    }

  go(None)
}

val p1 = emitAll((1 to 3).toSeq).toSource
(p1 |> withNext).runLog.run

> Vector((1,Some(2)), (2,Some(3)), (2,None))

Here, we start by accumulating the first element of the stream, and then, when we get to the next, we emit both of them. And we make a recursive call remembering what is now the current element. But the process we return in flatMap has an orElse clause. It says "by the way, if you don't have anymore elements (no more next), just emit current and None".

Now with both withPrevious and withNext we can create a withPreviousAndNext process:

def withPreviousAndNext[T]: Process1[T, (Option[T], T, Option[T])] = {
  def go(previous: Option[T], current: Option[T]): Process1[T, (Option[T], T, Option[T])] =
    await1[T].flatMap { next =>
      current.map { c =>
        emit((previous, c, Some(next))) fby go(Some(c), Some(next))
      }.getOrElse(
          go(previous, Some(next))
        ).orElse(emit((current, next, None)))
    }
  go(None, None)
}

val p1 = emitAll((1 to 3).toSeq).toSource
(p1 |> withPreviousAndNext).runLog.run

> Vector((None,1,Some(2)), (Some(1),2,Some(3)), (Some(2),3,None))

The code is pretty similar but this time we keep track of both the "previous" element and the "current" one.

emit(last paragraph)

I hope this will help beginners like me to get started with scalaz-stream and I'd be happy if scalaz-stream experts out there leave comments if there's anything which can be improved (is there an effective way to combine withPrevious and withNext to get withPreviousAndNext?

I finally need to add that, in order to get proper performance/side-effect control for the withNext and withPreviousAndNext processes you need to use the lazy branch of scalaz-stream. It contains a fix for orElse which prevents it to be evaluated more than necessary.

10 December 2013

The revenge of the chunks

This series of posts feels like a whole saga for something which should have a quick an easy way to demonstrate the obvious superiority of functional programming over a simple loop. In the first post. Then the second post was about defining proper scalaz-stream combinators to do the same thing, and particularly how to "chunk" the processing in order to get good performances.

However as I was writing unit tests for my requirements I realized that the problem was harder than I thought. In particular, the files I'm processing can have several sections made of HEADERs and TRAILERs. When you create chunks of lines to process this results in a number of combinations that need to be analysed. A chunk can:

  • start with a HEADER but not finish with a TRAILER which is in another chunk
  • contain lines only
  • contains lines + a TRAILER + a new HEADER
  • and so on...

For each of these cases it is necessary to use the current state and the contents of the lines to determine if the file is malformed or not. This is a lot less easy that previously.

All the combinations

This is what I came up with:

  def process(path: String, targetName: String, chunkSize: Int = 10000): String \/ File = {

    val targetPath = path.replace(".DAT", "")+targetName

    val read = 
      linesRChunk(path, chunkSize) |> 
      validateLines.map(lines => lines.mkString("\n"))

    val task = 
      ((read |> process1.intersperse("\n") |> 
      process1.utf8Encode) to io.fileChunkW(targetPath)).run

    task.attemptRun.leftMap(_.getMessage).map(_ => new File(targetPath))
  }

  /**
   * validate that the lines have the right sequence of HEADER/column names/lines/TRAILER
   * and the right number of lines
   */
  def validateLines: Process1[Vector[String], Vector[String]] = {

    // feed lines into the lines parser with a given state
    // when it's done, follow by parsing with a new state
    def parse(lines: Vector[String], state: LineState, newState: LineState) =
      emit(lines) |> linesParser(state) fby linesParser(newState)

    // parse chunks of lines
    def linesParser(state: LineState): Process1[Vector[String], Vector[String]] = {
      receive1[Vector[String], Vector[String]] { case lines =>
        lines match {
          case first +: rest if isHeader(first) =>
            if (state.openedSection) fail("A trailer is missing")
            else
              parse(lines.drop(2),
                    state.open,
                    LineState(lines.count(isHeader) > lines.count(isTrailer), 
                              lines.drop(2).size))

          case first +: rest if isTrailer(first) =>
            val expected = "\\d+".r.findFirstIn(first).map(_.toInt).getOrElse(0)

            if (!state.openedSection)             
              fail("A header is missing")
            else if (state.lineCount != expected) 
              fail(s"expected $expected lines, got ${state.lineCount}")
            else {
              val dropped = lines.drop(1)
              parse(dropped,
                    state.restart,
                    LineState(dropped.count(isHeader) > dropped.count(isTrailer), 
                              dropped.size))
            }

          case first +: rest =>
            if (!state.openedSection) fail("A header is missing")
            else {
              val (first, rest) = lines.span(line => !isTrailer(line))
              emit(first) fby
              parse(rest, state.addLines(first.size), state.addLines(lines.size))
            }

          case Vector() => halt
        }
      }
    }
    // initialise the parsing expecting a HEADER
    linesParser(LineState())
  }


  private def fail(message: String) = Halt(new Exception(message))
  private def isHeader(line: String) = line.startsWith("HEADER|")
  private def isTrailer(line: String) = line.startsWith("TRAILER|")

The bulk of the code is the validateLines process which verifies the file structure:

  • if the first line of this chunk is a HEADER the next line needs to be skipped, we know we opened a new section, and we feed the rest to the lines parser again. However we fail the process if we were not expecting a HEADER there

  • if the first line of this chunk is a TRAILER we do something similar but we also check the expected number of lines

  • otherwise we try to emit as many lines as possible until the next HEADER or TRAILER and we recurse

This is a bit complex because we need to analyse the first element of the chunk, then emit the rest and calculate the new state we will have when this whole chunk is emitted. On the other hand the processor is easy to test because I don't have to read or write files to check it. This would be a bit more difficult to do with the loop version.

But unfortunately not all the tests are green. One is still not passing. What if there is no ending TRAILER in the file? How can I raise an exception? There's no process to run, because there are no more lines to process! My test is pending for now, and I'll post the solution once I have it (maybe there's a smarter way to rewrite all of this?).

Is it worth it?

This was definitely worth it for me in terms of learning the scalaz-stream library. However in terms of pure programmer "productivity", for this kind of requirements, it feels like an overkill. The imperative solution is very easy to come up with and there is no problems with performances. This should change once streaming parsing is available (see the roadmap). Probably this use case will just be expressed as a one-liner. In the light of this post I'm just curious how the implementation will deal with chunking.

09 December 2013

`runState` 0 - combinators 1

In my previous blog post I was trying to implement a runState method with scalaz-stream to process a file and try to validate its internal structure. That was however not a good solution because:

  • it doesn't use combinators but a special purpose runState method
  • it stackoverflows on large files!

It turns out that there is a much better way of dealing with this use case.

Combinators

First of all it is possible to propagate some state with scalaz-stream without having to write a special runState method. The following uses only combinators to do the job:

def process(path: String, targetName: String): String \/ File = {

  val HEADER  = "HEADER(.*)".r
  val TRAILER = "TRAILER\\|(\\d+)".r

  val lineOrTrailer: Process1[String, String]  = {
    def go(lines: Int): Process1[String, String] =
      receive1[String, String] {
        case TRAILER(count) => 
          if (count.toInt == lines) halt 
          else Halt(new Exception(s"Expected $count lines, but got $lines"))
        case HEADER(h)      => 
          Halt(new Exception(s"Didn't expected a HEADER here: $h"))
        case s              => 
          emit(s) fby go(lines + 1)
      }
    go(0)
  }

  val linesStructure =
    discardRegex("HEADER.*") fby
    discardLine              fby
    lineOrTrailer

  val read       = io.linesR(path) |> linesStructure
  val targetPath = path.replace(".DAT", "")+targetName
  val task       = 
    ((read |> process1.intersperse("\n") |> 
     process1.utf8Encode) to io.fileChunkW(targetPath)).run

  task.attemptRun.leftMap(_.getMessage).map(_ => new File(targetPath))
}

val discardLine = receive1[String, String] { _ => halt }

/** discard a line if it matches the expected pattern */
def discardRegex(pattern: String): Process1[String,String] = {
  val compiled = Pattern.compile(pattern)
  receive1[String, String] { line =>
    if (compiled.matcher(line).matches) halt
    else Halt(new Exception(s"Failed to parse $line, does not match regex: $pattern"))
  }
}

With the code above, processing a file amounts to:

  • reading the lines
  • analysing them with linesStructure which propagates the current state, the number of lines already processed, with a recursive method (go) calling itself
  • writing the lines to a new file

The linesStructure method almost looks like a parser combinators expression with parsers sequenced with the fby ("followed by") method.

That looks pretty good but... it performs horribly! With the good-old "loop school", it took 8 seconds to process a 700M file:

def processLoop(path: String, targetName: String): String \/ File = {

  val targetPath = path.replace(".DAT", "")+targetName
  val writer = new FileWriter(targetPath)
  val source = scala.io.Source.fromFile(new File(path))
  var count = 0
  var skipNextLine = false
  try {
    source.getLines().foreach { line =>
      if (line.startsWith("HEADER")) skipNextLine = true
      else if (skipNextLine) { skipNextLine = false }
      else if (line.startsWith("TRAILER")) {
        val expected = line.drop(8).headOption.map(_.toInt).getOrElse(0)
        if (expected != count) throw new Exception(s"expected $expected, got $count")
      }
      else {
        count = count + 1
        writer.write(line)
      }
    }
  } catch {
    case t: Throwable => t.getMessage.left
  } finally {
    source.close
    writer.close
  }
  new File(targetPath).right
}

With the nice, "no-variables, no loop", method it took almost,... 8 minutes!

Chunky streaming

It is fortunately possible to retrieve correct performances by "chunking" the lines before processing them. To do this, we need a new combinator, very close to the io.linesR combinator in scalaz-stream:

// read a file, returning one "chunk" of lines at the time
def linesRChunk(filename: String, chunkSize: Int = 10000): Process[Task, Vector[String]] =
 io.resource(Task.delay(scala.io.Source.fromFile(filename)))(src => Task.delay(src.close)) { src =>
    lazy val lines = src.getLines.sliding(chunkSize, chunkSize) // A stateful iterator
    Task.delay {
      if (lines.hasNext) lines.next.toVector
      else throw End
    }
 }

Now we can process each chunk with:

def process(path: String, targetName: String, bufferSize: Int = 1): String \/ File = {

  val HEADER  = "HEADER(.*)".r
  val TRAILER = "TRAILER\\|(\\d+)".r

  def linesParser(state: LineState): Process1[Vector[String], Vector[String]] = {

    def onHeader(rest: Vector[String]) =
      (emit(rest) |> linesParser(ExpectLineOrTrailer(0))) fby
      linesParser(ExpectLineOrTrailer(rest.size))

    def onLines(ls: Vector[String], actual: Int) =
      emit(ls) fby linesParser(ExpectLineOrTrailer(actual + ls.size))

    def onTrailer(ls: Vector[String], count: Int, actual: Int) =
      if ((actual + ls.size) == count) emit(ls)
      else fail(new Exception(s"expected $count lines, got $actual"))

    receive1[Vector[String], Vector[String]] { case lines =>
      (lines, state) match {
        case (Vector(),                  _)                      => 
          halt
        case (HEADER(_) +: cols +: rest, ExpectHeader)           => 
          onHeader(rest)
        case (_,                         ExpectHeader)           => 
          fail(new Exception("expected a header"))
        case (ls :+ TRAILER(count),      ExpectLineOrTrailer(n)) =>
          onTrailer(ls, count.toInt, n)
        case (ls,                        ExpectLineOrTrailer(n)) => 
          onLines(ls, n)
      }
    }
  }

  val targetPath = path.replace(".DAT", "")+targetName

  val read = linesRChunk(path, bufferSize) |> 
             linesParser(ExpectHeader).map(lines => lines.mkString("\n"))
  val task = 
    ((read |> process1.intersperse("\n") |> 
    process1.utf8Encode) to io.fileChunkW(targetPath)).run

  task.attemptRun.leftMap(_.getMessage).map(_ => new File(targetPath))
}

The linesParser method uses receive1 to analyse:

  • the current state: are we expecting a HEADER, or some lines followed by a TRAILER?
  • the current chunk of lines

When we expect a HEADER and we have one, we skip the row containing the column names (see onHeader), we emit the rest of the lines to the linesParser (this is the recursive call) and we change the state to ExpectLineOrTrailer. If we get some lines with no TRAILER, we emit those lines and make a recursive call to linesParser with an incremented count to signal how many lines we've emitted so far (in the onLines method). Finally, if we get some lines and a TRAILER we check that the expected number of lines is equal to the actual one before emitting the lines and stopping the processing (no more recursive call in onTrailer).

For reference, here are the state objects used to track the current processing state:

sealed trait LineState

case object ExpectHeader                            extends LineState
case class  ExpectLineOrTrailer(lineCount: Int = 0) extends LineState

This new way of processing lines gets us:

  • a readable state machine with clear transitions, which was my first objective
  • adequate performances; it takes around 10 seconds to process a 700M file which is slightly more than the processLoop version but acceptable

One other explored avenue

It took me a loooooooooooong time to get there. I think I hit this issue when trying to use the built-in chunk combinator. When using chunk, my parser was being fed the same lines several times. For a chunk of 10 lines, I first had the first line, then the first 2, then the first 3,... Even with a modified version of chunk the performances were still very bad. This is why I wrote my own linesRChunk.

Now I got something working I hope that this will boost other's development time and show that it is possible to avoid loops + variables in that case!

05 December 2013

`runState` for a scalaz-stream Process

I was preparing to post this on the scalaz mailing-list but I thought that a short blog post could serve as a reference for other people as well. The following assumes that you have a good knowledge of Scalaz (at least of what's covered in my "Essence of the Iterator Pattern" post and some familiarity with the scalaz-stream library.

My use case

What I want to do is very common, just process a bunch of files! More precisely I want to (this is slightly simplified):

  1. read some pipe delimited files

  2. validate that the files have the proper internal structure:
    one(HEADER marker)
    one(column names)
    many(lines of pipe delimited values)
    one(TRAILER marker with total number of lines since the header)

  3. output only the lines which are not markers to another file

Scalaz stream

The excellent chapter 15 of Functional Programming in Scala highlights some of the potential problems with processing files:

  • you need to make sure you are closing resources properly even in the face of exceptions
  • you want to be able to easily compose small processing functions together instead of having a gigantic loop and a bunch of variables
  • you want to control the amount of data that is in memory at any moment in time

Based on the ideas of the book, Paul Chiusano created scalaz-stream, a library providing lots of combinators for doing this kind of input/output streaming operations (and more!).

A state machine for the job

My starting point for addressing our requirements is to devise a State object representing the both the expected file structure and the fact that some lines need to be filtered out. First of all I need to model the kind of lines I'm expecting when reading the file:

sealed trait LineState

case object ExpectHeader                            extends LineState
case object ExpectHeaderColumns                     extends LineState
case class  ExpectLineOrTrailer(lineCount: Int = 0) extends LineState

As you can see ExpectLineOrTrailer contains a counter to keep track of the number of lines seen so far.

Then I need a method (referred as the State function below) to update this state when reading a new line:

def lineState(line: String): State[Throwable \/ LineState, Option[String]] =
  State { state: Throwable \/ LineState =>
    def t(message: String) = new Exception(message).left

    (state, line) match {
      case (\/-(ExpectHeader), HeaderLine(_))           =>
        (ExpectHeaderColumns.right, None)
      case (\/-(ExpectHeaderColumns), _)                =>
        (ExpectLineOrTrailer(0).right, None)
      case (\/-(ExpectHeader), _)                       =>
        (t("expecting a header"), None)
      case (\/-(ExpectLineOrTrailer(n)), HeaderLine(_)) =>
        (t("expecting a line or a trailer"), None)
      case (\/-(ExpectLineOrTrailer(n)), TrailerLine(e)) =>
        if (n == e) (ExpectHeader.right, None)
        else        (t(s"wrong number of lines, expecting $e, got $n"), None)
      case (\/-(ExpectLineOrTrailer(n)), _)             =>
        (ExpectLineOrTrailer(n + 1).right, Some(line))
      case (-\/(e), _)                                  =>
        (state, None)
  }
}

The S type parameter (in the State[S, A] type) used to keep track of the "state" is Throwable \/ LineState. I'm using the "Left" part of the disjunction to represent processing errors. The error type itself is a Throwable. Originally I was using any type E but we'll see further down why I had to use exceptions. The value type A I extract from State[S, A] is going to be Option[String] in order to output None when I encounter a marker line.

This is all pretty good, functional and testable. But how can I use this state machine with a scalaz-stream Process?

runState

After much head scratching and a little help from the mailing-list (thanks Pavel!) I realized that I had to write a new driver for a Process. Something which would understand what to do with a State. Here is what I came up with:

def runState[F[_], O, S, E <: Throwable, A](p: Process[F, O])
                                           (f: O => State[E \/ S, Option[A]], initial: S)
                                           (implicit m: Monad[F], c: Catchable[F]) = {

  def go(cur: Process[F, O], init: S): F[Process[F, A]] = {
    cur match {
      case Halt(End) => m.point(Halt(End))
      case Halt(e)   => m.point(Halt(e))

      case Emit(h: Seq[O], t: Process[F, O]) => {
        println("emitting lines here!")
        val state = h.toList.traverseS(f)
        val (newState, result) = state.run(init.right)
        newState.fold (
          l => m.point(fail(l)),
          r => go(t, r).map(emitAll(result.toSeq.flatten) ++ _)
        )
      }

      case Await(req, recv, fb: Process[F, O], cl: Process[F, O]) =>
        m.bind (c.attempt(req.asInstanceOf[F[Any]])) { _.fold(
        { case End => go(fb, init)
          case e   => go(cl.causedBy(e), init) },
        o => go(recv.asInstanceOf[Any => Process[F ,O]](o), init)) }
    }
  }
  go(p, initial)
}

This deserves some comments :-)

The idea is to recursively analyse what kind of Process we're currently dealing with:

  1. if this is a Halt(End) we've terminated processing with no errors. We then return an empty Seq() in the context of F (hence the m.point operation). F is the monad that provides us input values so we can think of all the computations happening here as happening inside F (probably a scalaz.concurrent.Task when reading file lines)

  2. if this is a Halt(error) we use the Catchable instance for F to instruct the input process what to do in the case of an error (probably close the file, clean up resources,...)

  3. if this is an Emit(values, rest) we traverseS the list of values in memory with our State function and we use the initial value to get: 1. the state at the end of the traversal, 2. all the values returned by our State at each step of its execution. Note that the traversal will happen on all the values in memory, there won't be any short-circuiting if the State indicates an error. Also, this is important, the traverseS method is not trampolined. This means that we will get StackOverflow exceptions if the "chunks" that we are processing are too big. On the other hand we will avoid trampolining on each line so we should get good performances. If there was an error we stop all processing and return the error otherwise we emit all the values collected by the State appended to a recursive call to go

  4. if this is an Await Process we attempt to read input values, with c.attempt, and use the recv function to process them. We can do that "inside the F monad" by using the bind (or flatMap) method. The resulting Process is sent to go in order to be processed with the State function

Note what we do in case 2. when the newState returns an exception.left. We create a Process.fail process with the exception. This is why I used a Throwable to represent errors in the State function.

Now let's see how to use this new "driver".

Let's use it

First of all, we create a test file:

import scalaz.stream._
import Process._

val lines = """|HEADER|file
               |header1|header2
               |val11|val12
               |val21|val22
               |val21|val22
               |TRAILER|3""".stripMargin

// save 100 times the lines above in a file
(fill(100)(lines).intersperse("\n").pipe(process1.utf8Encode)
  .to(io.fileChunkW("target/file.dat")).run.run

Then we read the file but we buffer 50 lines at the time to control our memory usage:

val lines = io.linesR("target/file.dat").buffer(50)

We're now ready to run the state function:

// this task processes the lines with our State function
// the initial State is `ExpectHeader` because this is what we expect the first line to be
val stateTask: Task[Process[Task, String]] = runState(lines)(lineState, ExpectHeader)

// this one outputs the lines to a result file
// separating each line with a new line and encoding it in UTF-8
val outputTask: Task[Unit] = stateTask.flatMap(_.intersperse("\n").pipe(process1.utf8Encode)
                                      .to(io.fileChunkW("target/result.dat")).run)

// if the processing throws an Exception it will be retrieved here
val result: Throwable \/ Unit = task.attemptRun

When we finally run the Task, the result is either ().right if we were able to read, process, and write back to disc or exception.left if there was any error in the meantime, including when checking if the file has a valid structure.

The really cool thing about all of this is that we can now precisely control the amount of memory consumed during our processing by using the buffer method. In the example above we buffer 50 lines at the time then we process them in memory using traverseS. This is why I left a println statement in the runState method. I wanted to see "with my own eyes" how buffering was working. We could probably load more lines but the trade-off will then be that the stack that is consumed by traverseS will grow and that we might face StackOverflow exceptions.

I haven't done yet any benchmark but I can imagine lots of different ways to optimise the whole thing for our use case.

try { blog } finally { closing remarks }

I'm only scratching the surface of the scalaz-stream library and there is still a big possibility that I completely misunderstood something obvious!

First, it is important to say that you might not need to implement the runState method if you don't have complex validation requirements. There are 2 methods, chunkBy and chunkBy2, which allow to create "chunks" of lines based on a given line (for chunk) or pair of lines (for chunk2) naturally serving as "block" delimiters in the read file (for example a pair of "HEADER" followed by a "TRAILER" in my file).

Second, it is not yet obvious to me if I should use ++ or fby when I'm emitting state-processed lines + "the rest" (in step 2 when doing: emitAll(result.toSeq.flatten) ++ _). The difference has to do with error/termination management (the fallback process of Await) and I'm still unclear on how/when to use this.

Finally I would say that the scalaz-stream library is intriguing in terms of types. A process is Process[F[_], O] where O is the type of the output and the type of the input is... nowhere? Actually it is in the Await[F[_], A, O] constructor as a forall type. That's not all. In Await you have the type of request, F[A], a function to process elements of type A: recv: A => Process[F, O] but no way to extract or map the value A from the request to pass it to the recv method! The only way to do that is to provide an additional constraint to the "driver method" by saying, for example, that there is an implicit Monad[F] somewhere. This is the first time that I see a design where we build structures and then we give them properties when we want to use them. Very unusual.

I hope this can help other people exploring the library and, who knows, some of this might end up being part of it. Let's see what Paul and others think...

27 July 2013

Endorsing the move on to Java 6

This is a short public announcement to say that, as the maintainer of an open-source project, I support the move to Java 6 this year. I encourage other OSS projects, especially in the Scala ecosystem, to support this move as well.

20 June 2013

A Zipper and Comonad example

There are some software concepts which you hear about and after some time you roughly understand what they are. But you still wonder: "where can I use this?". "Zippers" and "Comonads" are like that. This post will show an example of:

  • using a Zipper for a list
  • using the Comonad cojoin operation for the Zipper
  • using the new specs2 contain matchers to specify collection properties

The context for this example is simply to specify the behaviour of the following function:

def partition[A](seq: Seq[A])(relation: (A, A) => Boolean): Seq[NonEmptyList[A]]

Intuitively we want to partition the sequence seq into groups so that all the elements in a group have "something in common" with at least one other element. Here is a concrete example

val near = (n1: Int, n2: Int) => math.abs(n1 - n2) <= 1
partition(Seq(1, 2, 3, 7, 8, 9))(near)

> List(NonEmptyList(1, 2, 3), NonEmptyList(7, 8, 9))

Properties

If we want to encode this behaviour with ScalaCheck properties we need to check at least 3 things:

  1. for each element in a group, there exists at least another related element in the group
  2. for each element in a group, there doesn't exist a related element in any other group
  3. 2 elements which are not related must end up in different groups

How do we translate this to some nice Scala code?

Contain matchers

"for each element in a group, there exists another related element in the same group"

prop { (list: List[Int], relation: (Int, Int) => Boolean) =>
  val groups = partition(list)(relation)
  groups must contain(relatedElements(relation)).forall
}

The property above uses a random list, a random relation, and does the partitioning into groups. We want to check that all groups satisfy the property relatedElements(relation). This is done by:

  • using the contain matcher
  • passing it the relatedElements(relation) function to check a given group
  • do this check forall groups

The relatedElements(relation) function we pass has type NEL[Int] => MatchResult[NEL[Int]] (type NEL[A] = NonEmptyList[A]) and is testing each group. What does it do? It checks that each element of a group has at least one element that is related to it.

def relatedElements(relation: (Int, Int) => Boolean) = (group: NonEmptyList[Int]) => {
  group.toZipper.cojoin.toStream must contain { zipper: Zipper[Int] =>
    (zipper.lefts ++ zipper.rights) must contain(relation.curried(zipper.focus)).forall
  }
}

This function is probably a bit mysterious so we need to dissect it.

Zipper

In the relatedElements function we need to check each element of a group in relation to the other elements. This means that we need to traverse the sequence, while keeping the context of where we are in the traversal. This is exactly what a Zipper is good at!

A List Zipper is a structure which keeps the focus on one element of the list and can return the elements on the left or the elements on the right. So in the code above we transform the group into a Zipper with the toZipper method. Note that this works because the group is a NonEmptyList. This wouldn't work with a regular List because a Zipper cannot be empty, it needs something to focus on:

// a zipper for [1, 2, 3, 4, 5, 6, 7, 8, 9]
//     lefts      focus    rights
// [  [1, 2, 3]     4      [5, 6, 7, 8, 9]  ]

Now that we have a Zipper that is focusing on the one element of the group. But we don't want to test only one element, we want to test all of them, so we need to get all the possible zippers over the original group!

Cojoin

It turns out that there is a method doing exactly this for Zippers, it is called cojoin. I won't go here into the full explanation of what a Comonad is, but the important points are:

  • Zipper has a Comonad instance
  • Comonad has a cojoin method with this signature cojoin[A](zipper: Zipper[A]): Zipper[Zipper[A]]

Thanks to cojoin we can create a Zipper of all the Zippers, turn it into a Stream[Zipper[Int]] and do the checks that really matters to us

def relatedElements(relation: (Int, Int) => Boolean) = (group: NonEmptyList[Int]) => {
  group.toZipper.cojoin.toStream must contain { zipper: Zipper[Int] =>
    val otherElements = zipper.lefts ++ zipper.rights
    otherElements must contain(relation.curried(zipper.focus))
  }
}

We get the focus of the Zipper, an element, and we check it is related to at least one other element in that group. This is easy because the Zipper gives us all the other elements on the left and on the right.

Cobind

If you know a little bit about Monads and Comonads you know that there is a dualism between join in Monads and cojoin in Comonads. But there is also one between bind and cobind. Is it possible to use cobind then to implement the relatedElements function? Yes it is, and the result is slightly different (arguably less understandable):

def relatedElements(relation: (Int, Int) => Boolean) = (group: NonEmptyList[Int]) => {
  group.toZipper.cobind { zipper: Zipper[Int] =>
    val otherElements = zipper.lefts ++ zipper.rights
    otherElements must contain(relation.curried(zipper.focus))
  }.toStream must contain((_:MatchResult[_]).isSuccess).forall
}

In this case we cobind each zipper with a function that will check if there are related elements in the groups. This will gives us back a Zipper of results and we need to make sure that it full of success values.

Second property

"for each element in a group, there doesn't exist a related element in another group"

prop { (list: List[Int], relation: (Int, Int) => Boolean) =>
  val groups = partition(list)(relation)
  groups match {
    case Nil          => list must beEmpty
    case head :: tail => nel(head, tail).toZipper.cojoin.toStream must not contain(relatedElementsAcrossGroups(relation))
  }
}

This property applies the same technique but now across groups of elements by creating a Zipper[NonEmptyList[Int]] instead of a Zipper[Int] as before:

def relatedElementsAcrossGroups(relation: (Int, Int) => Boolean) = (groups: Zipper[NonEmptyList[Int]]) =>
  groups.focus.list must contain { e1: Int =>
    val otherGroups = (groups.lefts ++ groups.rights).map(_.list).flatten
    otherGroups must contain(relation.curried(e1))
  }

Note that the ability to "nest" the new specs2 contain matchers is very useful in this situation.

Last property

Finally the last property is much easier because it doesn't require any context to be tested. For this property we just make sure that no element is ever related to another one and check that they end up partitioned into distinct groups.

"2 elements which are not related must end up in different groups"

prop { (list: List[Int]) =>
  val neverRelated = (n1: Int, n2: Int) => false
  val groups = partition(list)(neverRelated)
  groups must have size(list.size)
}

Conclusion

Building an intuition for those crazy concepts is really what counts. For me it was "traversal with a context". Then I was finally able to spot it in my own code.