From Pipe to Flow

Earlier we talked about the beauty of the pipe: a complex transactional job can be broken down into small components (or processing steps) and chained together with a pipe. Here’s a concrete example: as a video playback app with ads, one of TubiTV’s core features is that whenever an ad opportunity arises, it walks through the qualifying ad vendors, sends them ad requests, filters the responses for suitable ads, and finally plays them to the user. This is the basic function of every video playback application, and the requirement can be further abstracted into:

• Fetch a list of URLs
• Send network requests
• Parse the data returned by each response
• Use rule engines to filter out the results we don’t want
• Normalize the filtered results into a format we can handle
• Aggregate
• Decide the final result

The corresponding pseudocode structure looks like this:

getAdUrls(params)
  .pipe(urlRequester)
  .pipe(responseParser)
  .pipe(ruleEngines)
  .pipe(normalizer)
  .pipe(aggregator)
  .on('data', reducer)

This style of processing is simple, easy to understand, highly extensible, and offers excellent concurrency. As sample code it is beautiful. In real-world usage, however, the problem appears: how do we make this flow parallelize better (or, parallelize the way we want it to)?

Note that the goal here is parallel, not merely concurrent. Suppose we have 5,000 qualifying URLs and we want them spread in parallel across all available cores — whether that’s a 16-core single machine or a cluster with 800 cores in total, we want processing to be as fast as possible. For a problem of that scale, the solution comes quickly to mind: do map/reduce on a Spark cluster. But what if the problem isn’t that big? Suppose a single request only has a few dozen URLs to process and the work needn’t cross physical machines — Spark is overkill. What then?

This doesn’t stump most engineers with some experience. Using a message queue (such as rabbitMQ or SQS) and decoupling producer from consumer, we can turn the pipeline above into a distributed pipeline. But every component you add to a system brings extra problems. What used to be one piece of software is now three (or four), and all three need operations:

• Managing the producer: a producer starting from getAdUrls() pushes the fetched URLs into the message queue
• Managing the message queue system itself (though with SQS there’s nothing to operate)
• Managing the distributed consumers (workers): they read URLs from the queue, process them through the same pipeline, and push the results into another message queue
• Managing the reducer (which can be the same entity as the producer): one or more aggregators / reducers pull from the message queue of results and reduce them into the final result

In some application scenarios this is a great approach — fan-out / fan-in over messages is light and nimble. But it does add complexity, and cross-process enqueue / dequeue greatly increases latency: the OS hiccups and who knows when the next dequeue will happen…

Can we do something at the language (framework) level?

Yes. We know that in Scala and Clojure there’s pmap() for parallel list processing. With pmap, the pseudocode above can be expressed as:

  (reduce reducer
    []
    (pmap
      (comp normalizer ruleEngines responseParser urlRequester)
      (getAdUrls params)))

Of course, putting several processing stages under a single thread isn’t necessarily a good thing: some steps are fast and need only a little parallelism, others are slow and need a lot. So in some situations this is better:

  (def req (partial pmap urlRequester))
  (def parser (partial pmap responseParser))
  (def engine (partial pmap ruleEngines))
  (def norm (partial pmap normalizer))
  (reduce reducer
    []
    ((comp norm engine parser req)
      (getAdUrls params)))

Alas, pmap() isn’t necessarily an obedient servant: you can’t give it orders about which part should pull harder and which part should ease off. Maybe most of the time you don’t need that flexibility — but using it or not is one thing; having flexibility is always better than not having it. So you might wish you could do this:

getAdUrls(params)
  .flow(urlRequester, 1)
  .flow(responseParser, 2)
  .flow(ruleEngines, 8)
  .flow(normalizer, 4)
  .flow(aggregator, 1)
  .on('data', reducer)

Borrowing an analogy from hardware design:

• getAdUrls is the input to the whole piece of hardware: rx
• reducer is the output of the whole piece of hardware: tx
• All the intermediate steps are processing engines; the order in which data flows is: urlRequester -> responseParser -> ruleEngines -> normalizer -> aggregator
• The ratio among engines is 1:2:8:4:1 — you can think of one urlRequester as corresponding to eight ruleEngines

This ratio doesn’t have to be stable (that would involve reserving resources up front); it can be best effort. Allowing programmers to set the ratio of runtime compute resources without writing detailed imperative code buys two things:

• Code stays simple and less error-prone
• Parallel performance is better, and back pressure is handled better
• The ratio can be adjusted dynamically according to the use case and the target environment

Of course, this is still just an idea; no language or library today seems able to accomplish it. Erlang (Elixir) / Akka’s actor-model-based concurrency model comes closest to realizing this idea — after all, resources are broken down to a very fine granularity, and each engine above can have its own process pool (poolboy). Yet to realize those few lines of pseudocode still takes an enormous amount of work and all kinds of boilerplate. Elixir recently introduced Experimental.GenStage, which comes very close to this idea. I believe that as time goes on and CPU parallelism grows stronger, more and more programming languages will make their mark on parallel computing (look — over the past few years, improvements in language-level concurrency have gone marrow-deep).