Skip a part of the stream when it's stalled

user_4664091 0
Viewed 64

I have a situation where messages are passed through a component that could possibly delay them.

Under stress I'd like to skip this component, so that no more than X messages could be delayed simultaneously. Overflowing messages will skip this stage and move to the next stage of the steam.

Messages are stalled within this stage till their future is done, or up to one minute, whichever comes first.

I can probably implement a custom GraphStage similarly to this buffer example, or use divertTo with some counter to cause messages to skip the stalled component, but it feels like there might be an easier approach in akka streams

akka-stream
1 Answers

I've been playing around your use case and came up with a solution based on an Akka Actor representing a counter and an asynchronous map stage:

The idea being that up to 3 elements can be processed at any given time and that based on a counter whose max capacity is 2 we'd only allow at most 2 of these elements to simultaneously be processed by the slow component.

This way, there's always one thread of processing reserved to upstream elements that would be branched out from the slow component and directly reach downstream.

Let's first define a basic Counter with a maximum capacity as an Akka Actor:

class Counter(max: Int) extends Actor {
  private var count: Int = 0

  override def receive: Receive = {
    case TryAndLock if count < max =>
      count += 1
      sender ! true
    case TryAndLock =>
      sender ! false
    case Release =>
      count -= 1
  }
}

sealed trait CounterAction
case object TryAndLock extends CounterAction
case object Release extends CounterAction

val counter = system.actorOf(Props(new Counter(max = 2)))

It holds a mutable count variable that can be incremented via a TryAndLock request, but only if the count hasn't yet reached the maximum capacity, and can be decremented via a Release request.

We're using an Actor so that concurrent lock and release operations from within the following mapAsync stage are correctly handled without race conditions.

Then, it's just a matter of using a mapAsyncUnordered stage with a parallelism just 1 unit above the counter's max capacity.

Any element passing through the asynchronous stage will query the Counter to try and lock a resource. If a resource's been locked, then the element will be thrown into the slow component. If not, it will skip it. Elements are passed into the slow component until we reach the counter's max capacity, at which point any new element will be skipped until an element exits the slow component and releases a resource from the counter.

We can't simply use a mapAsync because elements would keep their upstream's order when existing the stage, making skipped elements wait for elements processed by the slow component to finish before being produced downstream. Thus the necessity to use the mapAsyncUnordered instead.

Let's define an example with at most 2 elements processed at the same time by the slow component and an asynchronous map whose parallelism is 3:

Source(0 to 15)
  .throttle(1, 50.milliseconds)
  .mapAsyncUnordered(parallelism = 3) { i =>
    (counter ? TryAndLock).map {
      case locked: Boolean if locked =>
        val result = slowTask(i)
        counter ! Release
        result
      case _ =>
        skipTask(i)
    }
  }
  .runForeach(println)

with for instance these two functions that'll simulate the slow component (slowTask) and what to do when skipping the slow component (skipTask):

def slowTask(value: Int): String = {
  val start = Instant.now()
  Thread.sleep(250)
  s"$value - processed - $start - ${Instant.now()}"
}
def skipTask(value: Int): String =
  s"$value - skipped - ${Instant.now()}"

which results in something like:

2 - skipped - 2020-06-03T19:07:19.410Z
3 - skipped - 2020-06-03T19:07:19.468Z
4 - skipped - 2020-06-03T19:07:19.518Z
5 - skipped - 2020-06-03T19:07:19.569Z
1 - processed - 2020-06-03T19:07:19.356Z - 2020-06-03T19:07:19.611Z
0 - processed - 2020-06-03T19:07:19.356Z - 2020-06-03T19:07:19.611Z
8 - skipped - 2020-06-03T19:07:19.719Z
9 - skipped - 2020-06-03T19:07:19.769Z
10 - skipped - 2020-06-03T19:07:19.819Z
6 - processed - 2020-06-03T19:07:19.618Z - 2020-06-03T19:07:19.869Z
12 - skipped - 2020-06-03T19:07:19.919Z
7 - processed - 2020-06-03T19:07:19.669Z - 2020-06-03T19:07:19.921Z
14 - skipped - 2020-06-03T19:07:20.019Z
15 - skipped - 2020-06-03T19:07:20.070Z
11 - processed - 2020-06-03T19:07:19.869Z - 2020-06-03T19:07:20.122Z
13 - processed - 2020-06-03T19:07:19.968Z - 2020-06-03T19:07:20.219Z

where the first part is the index of the upstream element, the second part is the transformation the element's been applied on (either processed when entering the slow component or skipped) and the last part is a timestamp so that we visualise when things are happening.

The 2 first elements entering the stage (0 and 1) are processed by the slow component and a bunch of following elements (2, 3, 4 and 5) are skipping the slow stage until these 2 first elements complete and additional elements can enter the slow stage. And so on.

Gilfoyle

user_92971440
Related