Project: Concurrent Pipeline

A concurrent pipeline is a chain of stages that process items in parallel. A stage is one step that takes items in, changes them, and passes them on.

One stage transforms items in a queue

The stage takes each number from the input queue. It doubles the number and puts the result in the output queue.

import scala.concurrent.{Future, Await}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
def runStage(input: LinkedBlockingQueue[Int], output: LinkedBlockingQueue[Int], f: Int => Option[Int]): Future[Unit] =
  Future {
    var item = input.take()
    while item != -1 do
      f(item) match
        case Some(v) => output.put(v)
        case None => ()
      item = input.take()
    output.put(-1)
  }
def drain(queue: LinkedBlockingQueue[Int]): List[Int] =
  val out = ListBuffer.empty[Int]
  var done = false
  while !done do
    val v = queue.take()
    if v == -1 then done = true else out += v
  out.toList
@main def main(): Unit =
  val input = new LinkedBlockingQueue[Int]()
  val output = new LinkedBlockingQueue[Int]()
  val stage = runStage(input, output, (v: Int) => Some(v * 2))
  List(1, 2, 3).foreach(v => input.put(v))
  input.put(-1)
  Await.result(stage, Duration.Inf)
  println(drain(output))

Two stages run at the same time

The first stage doubles each number. The second stage keeps only the numbers above four.

import scala.concurrent.{Future, Await}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
def runStage(input: LinkedBlockingQueue[Int], output: LinkedBlockingQueue[Int], f: Int => Option[Int]): Future[Unit] =
  Future {
    var item = input.take()
    while item != -1 do
      f(item) match
        case Some(v) => output.put(v)
        case None => ()
      item = input.take()
    output.put(-1)
  }
def drain(queue: LinkedBlockingQueue[Int]): List[Int] =
  val out = ListBuffer.empty[Int]
  var done = false
  while !done do
    val v = queue.take()
    if v == -1 then done = true else out += v
  out.toList
@main def main(): Unit =
  val first = new LinkedBlockingQueue[Int]()
  val middle = new LinkedBlockingQueue[Int]()
  val last = new LinkedBlockingQueue[Int]()
  val firstStage = runStage(first, middle, (v: Int) => Some(v * 2))
  val secondStage = runStage(middle, last, (v: Int) => if v > 4 then Some(v) else None)
  List(1, 2, 3, 4, 5).foreach(v => first.put(v))
  first.put(-1)
  Await.result(firstStage, Duration.Inf)
  Await.result(secondStage, Duration.Inf)
  println(drain(last))

A runner wires any list of stages

The runner builds one queue per gap between stages. It starts every stage, feeds the items, and drains the last queue.

import scala.concurrent.{Future, Await}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
def runStage(input: LinkedBlockingQueue[Int], output: LinkedBlockingQueue[Int], f: Int => Option[Int]): Future[Unit] =
  Future {
    var item = input.take()
    while item != -1 do
      f(item) match
        case Some(v) => output.put(v)
        case None => ()
      item = input.take()
    output.put(-1)
  }
def drain(queue: LinkedBlockingQueue[Int]): List[Int] =
  val out = ListBuffer.empty[Int]
  var done = false
  while !done do
    val v = queue.take()
    if v == -1 then done = true else out += v
  out.toList
def runPipeline(items: List[Int], stages: List[Int => Option[Int]]): List[Int] =
  val queues = List.fill(stages.size + 1)(new LinkedBlockingQueue[Int]())
  val futures = stages.zipWithIndex.map(pair => runStage(queues(pair._2), queues(pair._2 + 1), pair._1))
  items.foreach(v => queues.head.put(v))
  queues.head.put(-1)
  futures.foreach(f => Await.result(f, Duration.Inf))
  drain(queues.last)
@main def main(): Unit =
  val result = runPipeline(List(1, 2, 3, 4), List((v: Int) => Some(v * 2), (v: Int) => if v >= 6 then Some(v) else None))
  println(result)

Example

import scala.concurrent.{Future, Await}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
def runStage(input: LinkedBlockingQueue[Int], output: LinkedBlockingQueue[Int], f: Int => Option[Int]): Future[Unit] =
  Future {
    var item = input.take()
    while item != -1 do
      f(item) match
        case Some(v) => output.put(v)
        case None => ()
      item = input.take()
    output.put(-1)
  }
def drain(queue: LinkedBlockingQueue[Int]): List[Int] =
  val out = ListBuffer.empty[Int]
  var done = false
  while !done do
    val v = queue.take()
    if v == -1 then done = true else out += v
  out.toList
def runPipeline(items: List[Int], stages: List[Int => Option[Int]]): List[Int] =
  val queues = List.fill(stages.size + 1)(new LinkedBlockingQueue[Int]())
  val futures = stages.zipWithIndex.map(pair => runStage(queues(pair._2), queues(pair._2 + 1), pair._1))
  items.foreach(v => queues.head.put(v))
  queues.head.put(-1)
  futures.foreach(f => Await.result(f, Duration.Inf))
  drain(queues.last)
@main def main(): Unit =
  val result = runPipeline(List(1, 2, 3, 4, 5, 6), List((v: Int) => Some(v * 2), (v: Int) => if v > 5 then Some(v) else None, (v: Int) => Some(v + 1)))
  println(result)
  println(result.size)

Key points

Queue: a queue is a waiting line that holds items until a stage is ready for them.
Stage: a stage is one step that transforms each item it receives.
Future: a future is a handle for work that runs in the background.
Sentinel: a sentinel is a special value that marks the end of the items.
Drain: to drain a queue is to take items from it until the sentinel arrives.
Order: the final list keeps the order the items left the last stage.

In this example, the runner wires three stages together. The first stage doubles each number. The second stage keeps only numbers above five. The third stage adds one to each survivor. The program prints the final list in order and then its size.

Note

Await.result waits for a future to finish before the program prints its result.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Scastie

import scala.concurrent.{Future, Await}
import scala.concurrent.duration.Duration
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.LinkedBlockingQueue
import scala.collection.mutable.ListBuffer
def runStage(input: LinkedBlockingQueue[Int], output: LinkedBlockingQueue[Int], f: Int => Option[Int]): Future[Unit] =
  Future {
    var item = input.take()
    while item != -1 do
      f(item) match
        case Some(v) => output.put(v)
        case None => ()
      item = input.take()
    output.put(-1)
  }
def drain(queue: LinkedBlockingQueue[Int]): List[Int] =
  val out = ListBuffer.empty[Int]
  var done = false
  while !done do
    val v = queue.take()
    if v == -1 then done = true else out += v
  out.toList
def runPipeline(items: List[Int], stages: List[Int => Option[Int]]): List[Int] =
  val queues = List.fill(stages.size + 1)(new LinkedBlockingQueue[Int]())
  val futures = stages.zipWithIndex.map(pair => runStage(queues(pair._2), queues(pair._2 + 1), pair._1))
  items.foreach(v => queues.head.put(v))
  queues.head.put(-1)
  futures.foreach(f => Await.result(f, Duration.Inf))
  drain(queues.last)
@main def main(): Unit =
  val result = runPipeline(List(1, 2, 3, 4, 5, 6), List((v: Int) => Some(v * 2), (v: Int) => if v > 5 then Some(v) else None, (v: Int) => Some(v + 1)))
  println(result)
  println(result.size)

Exercise

What does the sentinel value do in this pipeline?

It doubles the last item It tells each stage that no more items are coming It sorts the final list It starts a new queue