Conveyor: pipelines approach opposite to channels

Releases
1

To build a kafka/rabbitmq consuming pipeline with saturated stages in Go you normally need to spin a goroutine per stage and connect them with channels (whose buffer can’t be change in runtime). The logic of processing one message (or batch of messages) gets split between multiple functions not allowing you to use local variables and making it complicated to sync different branches of execution (in case of fan-outs).

Conveyor’s approach is orthogonal to that:

  • define a single function that processes an item (message or a batch),
  • add stage.MoveTo() gates between the stages in the function and
  • let the lib manage goroutines and ordering.
  • Supported pipeline topologies are deadlock-free: execution always moves forwards and earlier items have priority over later.
  • Keep the state of item processing in local variables in the function and use them at any stage you need.
  • Graceful queue size and concurrency limit changes, graceful shutdown and observability out of the box

Ready to answer questions. Don’t miss the great WebAssembly interactive demo

2

Interesting approach, here is some opinionated critique:

API-Ergonomics. I don’t like the “stage.moveTo()” syntax. It is not clear at all that something is locked and without an explicit “release” it is very unclear when it is released. Go provides a nice option with aquire() defer release(), which feels very Go-Native and provides stability with early returns or panics. I would prefer a readStage.aquire() and defer readStage.release()

I’m missing an easy option to parallelize the processing of individual stages - as it stands I would have to code this myself when using the library - whereas channels can easily have multiple concurrent readers, which execute multiple packets in parallel. In your implementation, if a single item is very slow in a single stage, that stage will be locked until this item is completely processed, implicitly blocking all prior stages from advancing. So a single slow network call would slow down the whole pipeline.

No automatic scaling/back pressure. A good pipeline system would provide some feedback via back pressure from later stages, so earlier stages can adjust their throughput dynamically to provide an even load.

1 Like
3

Thanks for the reply!

I don’t like the “stage.moveTo()” syntax

I went through several iterations and in the first version I used explicit Enter/Exit methods. defer doesn’t work well here since you need to exit a stage once you acquired a next stage (not when ItemProcessor returns). MoveTo does these 2 actions atomically not leaving a space for mistake but still allowing you keeping a previous stage (which is a rare case) via Retain. Enter/Exit semantics didn’t work well with fanout stages as well. Any not released stages get released (and not joined waves from tasks/Retain join) when function (ItemProcessor) returns.

The API design aims to avoid any possibilities of deadlocks, that’s why you need to claim tasks when moving to fanout stage. With Enter/Exit it becomes extremely difficult to understand and avoid mistakes.

With the current design you take existing 1-goroutine sequential fn and just put “MoveTo” calls as separators converting it to a fully-fledged pipeline (preserving the local state unlike chans approach)

I’m missing an easy option to parallelize the processing of individual stages

use stage.SetLimit() - it’s equivalent to what you describe. And it can be changed gracefully after conveyor.Run (see demo page)

No automatic scaling/back pressure.

Backpressure is provided in exact the same way as with channels. When an item can’t enter a next stage (since it’s occupied) it doesn’t unblock the current stage so next items can’t enter the current stage. This way back-pressure propagates down to the starting “Read” stage. On demo page it’s all visualised and clear to see

Thanks for your opinion! Hope you will find it useful for you. I spent long time thinking about all edge cases and use-cases for the library and use it in production for kafka messages processing. From architecture POV (when you want to keep business logic on service layer) it works better than channels

4

I personally would have preferred a solution with clear bounds like:

readStage.Do(func() {
...
...
...
})
filterStage.Do(func() {
...
...
...
})

This separated the code into clear boundaries between stages, giving you a clear visual guidance where the stages start and end. With stage.MoveTo() you could end up not calling it in some branches of complex if statements, or even accidentally call it twice in some branches of your workflow. And you can still have deadlocks, since one code-path could call a stage out of order, since stage.MoveTo is just a simple call which can happen anywhere in the code in any order (1,3,2…)

I find SetLimit() has some pitfalls, since each go routine will traverse the whole function sequentially. So If the read stage reads a batch of 100 messages (beacuse query batch-size of 100 is very efficient) then the next stage will not be able to easily work on these 100 messages in parallel, since a single read stage execution will always lead only to a single nexStage execution. I will have to manually create the logic to read 100 messages, cache them in a synchronized buffer and let the read stage only fetch one of these messages and go to the next stage, so SetLimit will actually work. This is something, which channels handle gracefully.

True back-pressure is more than just breaking when the queue is full. True back-pressure means dynamically throttling invocation speed and parallelization for optimal throughput. A full-fledged pipeline system will measure execution time and resource consumption of all stages and scale them accordingly for optimal throughput across all stages. So if the final stage starts to slow down, we don’t wait until all queues are full for breaking full stop, but send a message backwards to the other stages to slow down beforehand. On the other hand, if two stages are CPU bound, a full-fledged pipeline system will weigh their executions in a way to make sure a previous stage does not hinder throughput by consuming all CPU resources, while a later stage starves.

The difference is like drivers on a road, the minimal implementation leads to breaking on a traffic jam. But remarkable implementations would be like responsible driving and slowing traffic, before a traffic jam can occur.

1 Like
5

I considered Do semantics as well (it can be added easily) but rejected:

  • User can add some code between Do calls - it will execute out of any stage (breaks backpressure)
  • The main pro of “per-item” execution (opposite to chans) is that you can store the state of item processing in local variables inside fn and use them at any stage. Variables defined in one Do func will not be visible in another which forces you to pre-define variables at the top
    • One of my goals is to support easy migration for old 1-goroutine sequential code. In case of MoveTo you don’t need to extract stages in functions and extract variables, just put MoveTo calls at the boundaries
  • The init “read” stage is implicit and will not have Do anyway
  • Retain functionality will look less natural
  • We need to think how to react if Do is called inside another Do: it can be used to replace Retain (is a goroutine is used) but code like that diffucult to read and reason about

And you can still have deadlocks

Calling MoveTo for already entered stage or earlier stages panics

I find SetLimit() has some pitfalls

SetLimit is to allow multiple batches (items in conveyor terms), to be in a stage at the same time. It’s not for messages from your example.

In a simple example:

// implicit read stage
var batch []Message = read100Messages()

if err := processing.MoveTo(ctx); err != nil { ... }

// just spin errgroup with a goroutine for each of the message
// and wait

So conveyor is batch-centric and a batch moves through the stages normally till the last “commit” stage.

If you need gather/scatter and want to have “message processor” with limited concurrency (e.g. 150 messages) that will provide backpressure (which is absent in the example above), use FanOut stage with AddPool().SetLimit(150). Processing a message is a “Task” in it (similar to errgroup). Tasks of earlier batches/items have priority. If there are multiple pools, you claim tasks beforehand for all of them which eliminates the possibility of deadlocks with other items.

You can also use AddLane to create a sub-conveyor for messages (not batches)

True back-pressure is more than just breaking when the queue is full.

I agree that there are more sophisticated strategies than the basic one when back-pressure propagates through the stages one by one. Your example makes sense and I completely agree it may benefit in the cases you mention.

But in the lib I don’t know which strategy works better for your specific stages. Basic one behaves in the same way as channels would do. My goal is not to create a heavy framework but more like sync primitives.

There is observability support. Queue sizes and concurrency limits for stages are dynamic (can be gracefully adjusted in runtime). Given that features and understanding the nature of your specific stages you can implement a custom strategy.

Observability is pull-based in order not to lock on slow consumers. In the first version I didn’t come up with a clear and simple API for push-based “back-pressure controller/monitor”