Light intensity showers

Here F is a composed function: F = make_report . (api_call api_key)

The . operator lets you compose functions in Haskell. In general, if you have a function f and another function g, then (f . g) x is equivalent to doing (f (g x).

Why do we use mapM instead of map?

Because api_call returns a value of type IO String (this is a monad), and the make_report function takes an argument of type IO String and returns a value of type IO WeatherReport (another monad). The mapM function is like the map function, but it operates on lists of monads.

Actually mapM takes something like

and returns m

, that is, it takes a list of monads of T, and returns a monad of a list of T.

The definition of mapM is:

     mapM :: Monad m => (a -> m b) -> [a] -> m [b]

This explains why we do lreps <- preps.
The sortBy function requires a pure list; since preps is a monad that contains a list, what we do with this assignment is take the list out of the monad.

In general, if we have a monad m v, when we do x <- m v, we assign v to x.

What is a monad? It is the usual question from those who are starting to learn Haskell.

Remember when we did the following in the Toque y Fama game?

    acc <- getLine

What happens is that getLine is not a pure function, something we don’t like in Haskell. A pure function has no side effects, but getLine is a function that interacts with the outside world; it is what is called a destructive action (see revelations). Haskell protects us from the outside world through monads. In particular, instead of returning a String, what the getLine function does is return an IO String monad.

I like to think of monads as boxes. In Haskell, to talk to the outside world we use these boxes. The getLine function returns a box called IO whose content is a String.

Our program uses a lot of monads, since it has to interact with the OpenWeatherMap website, read environment variables from the operating system, write the weather report to the screen, measure the running time of a function, etc.

If you don’t understand much of this, don’t worry. Don’t try to grasp the concept of a monad right away; just write several programs in Haskell that use input/output functions, as in this exercise, and everything will fall into place with time. In any case, I am going to write a special post on Haskell later where I will explain this concept.

Erlang

Erlang implements the actor model for concurrency. What we will do is create actors that perform only one action. Traditionally an actor in Erlang sits in a loop receiving messages; in this case we wait for just one message and finish.

Let’s look at the synchronous version first

    buscar_reportes_con_map(Ciudades) -> 
        lists:map(
          fun(Ciudad) ->
               Url = crear_url_api(Ciudad),
                {Xml,Error} = llamar_api(Url),
              extraer_reporte(Xml, Ciudad, Error)  
            end, Ciudades).

It is a simple mapping of cities to reports, using a function declared inline. This function builds the URL, invokes the API and then extracts the report (which is the result).

the asynchronous version looks like this:

    buscar_reportes_con_map_async(Ciudades) -> 
        ReqIds = lists:map(
         fun(Ciudad) ->
               Url = crear_url_api(Ciudad),
                {Ciudad,llamar_api_async(Url)}
           end, Ciudades),
     recolectar(ReqIds).

The asynchronous version makes an asynchronous call to the API. In this case we map cities to the Ids of the actors.

Let’s see how the llamar_api_async() function works:

    llamar_api_async(Url) -> spawn(?MODULE, async_api_call, [Url]).

    async_api_call(Url) ->  % io:format("llamando asincronamenete url: ~s\n", [Url]), % descomentar para convencerse que es asincrono
       {Xml,Error} = llamar_api(Url),
      receive 
            {From, get_result} -> From ! {xml, Xml, Error}
    end. 

The llamar_api_async function invokes the spawn primitive, which creates an actor served by the async_api_call(Url) function.

This actor sits waiting for the get_result message. When it receives it, it sends back the result by doing From ! {xml, Xml, Error}.

To finish understanding this we need to look at the recolectar function:

    recolectar([Llam|Llams]) -> 
      {Ciudad,ReqId} = Llam,
      ReqId ! {self(), get_result},
        receive
            {xml, Xml, Error} -> 
                [extraer_reporte(Xml, Ciudad, Error) | recolectar(Llams)]
        end;
    recolectar([]) -> [].

The recolectar function takes a list of actor handles. This function sends a message to each actor:

     ReqId ! {self(), get_result}

The message has two parts, the identifier of the main process (self()) and the atom get_result.

Remember that each actor answers with the message {xml, Xml, Error}.In this drawing I tried to explain what happens. The left side initializes the actors (small green circles). The right side explains what happens afterwards. In this case it starts by sending the get_result message to each actor created. Each actor answers with the result of invoking the API, {xml, Xml, Error}.

Kotlin

With this language I decided to use a mechanism similar to the Scala solution.

The final code is basically this:

    // Sequential solution
    val reports = cities.map(::apiCall)
    printReports (reports)

The parallel solution ended up like this:

    // Parallel solution
    val reports = cities.par().map(::apiCall)
    printReports(reports.unpar().toList())
    reports.executorService.shutdown()

Kotlin doesn’t actually have the par() primitive for lists. To do it I used code published by Holger Brandl here.

This code extends collections with the par() primitive.

    /** A delegated tagging interface to allow for parallized extension functions */
    class ParCol<T>(val it: Iterable<T>, val executorService: ExecutorService) 
       : Iterable<T> by it

    /** Convert a stream into a parallel collection. */
    fun <T> Iterable<T>.par(numThreads: Int = Runtime.getRuntime().availableProcessors(),
                            executorService: ExecutorService = Executors.newFixedThreadPool(numThreads))
    : ParCol<T> {
        return ParCol(this, executorService);
    }

The code uses the ExecutorService class from Java 8. This is a service that lets you work with threads and concurrent code.

So when we do lista.par(), what we do is create an instance of the ParCol class that contains the list and adds an ExecutorService to it.

Another extension we add is the map() function:

    fun <T, R> ParCol<T>.map(transform: (T) -> R): ParCol<R> {
        val destination = ArrayList<R>(if (this is Collection<*>) this.size else 10)
        val futures = this.asIterable().map { 
            executorService.submit { 
                destination.add(transform(it)) 
            } 
        }
        futures.map { it.get() } // this will block until all are done
        return ParCol(destination, executorService)
    }

Here we create a list of futures, which are the ones that apply the transform() function to each element of the collection.

Then we wait for each of the calls to transform() to run.

Finally, the map() function creates a new instance of the ParCol class, this time with the result of computing transform() over all the elements of the initial collection.

To retrieve the result we use the unpar() function, which returns the list contained inside the ParCol instance.

    fun <T> ParCol<T>.unpar(): Iterable<T> {
        return this.it;
    }

This implementation gives us a hint of how to implement parmap in Java 8. Would anyone be up for solving this challenge using Java 8? If so, I invite you to add your proposal through a pull request to my repository: https://github.com/lnds/9d9l](//github.com/lnds/9d9l)

Let’s compare the results in Kotlin, to measure how efficient this solution is:

$ java -jar weather-1.0-SNAPSHOT-all.jar  Santiago Boston Londres Valdivia Antofagasta

Boston              max: 32,0  min: 12,2   actual:  17,6 niebla
Antofagasta         max: 17,0  min: 17,0   actual:  17,0 nubes rotas
Valdivia            max: 12,0  min: 12,0   actual:  12,0 llovizna
Santiago            max: 12,0  min: 12,0   actual:  12,0 chubasco de ligera intensidad
Londres             max:  8,5  min:  8,5   actual:   8,5 nubes dispersas
tiempo ocupado para generar el reporte:  0:00:1,372

In parallel:

$ java -jar weather-1.0-SNAPSHOT-all.jar -p Santiago Boston Londres Valdivia Antofagasta
Boston              max: 32,0  min: 12,2   actual:  17,6 niebla
Antofagasta         max: 17,0  min: 17,0   actual:  17,0 nubes rotas
Valdivia            max: 12,0  min: 12,0   actual:  12,0 llovizna
Santiago            max: 12,0  min: 12,0   actual:  12,0 chubasco de ligera intensidad
Londres             max:  8,5  min:  8,5   actual:   8,5 nubes dispersas
tiempo ocupado para generar el reporte:  0:00:0,516

The sequential version takes 1.372 seconds on my machine; the parallel version takes 0.516 seconds (less than half the time!).

With this we finish the implementation of this challenge in 9 languages. Just to wrap everything up I will publish another article with the details I didn’t cover, namely how to download from a URL (using HTTP) and parse an XML file.

Four solutions

In this article I have shown 4 additional solutions to this problem.

F#: at https://github.com/lnds/9d9l/tree/master/desafio2/fsharp

Haskell: at https://github.com/lnds/9d9l/tree/master/desafio2/haskell

Erlang: at https://github.com/lnds/9d9l/tree/master/desafio2/erlang

Kotlin: at https://github.com/lnds/9d9l/tree/master/desafio2/kotlin

Notes

(*) Title image taken from Wikipedia, by John Kerstholt, taken from here.