6 ms·

Go Concurrency Patterns: Pipelines and cancellation

247 points by enneff 13y ago

genwin 13y ago
I use channels to single-file data to a SQLite db. Data gets pushed into channels from multiple users / locations of the web app. A single goroutine (concurrently running function) reads from the channels one channel at a time and, for each channel, writes everything in the channel to the db in a single transaction. SQLite isn't the best at concurrency. Go's channels help resolve that issue nicely.
alexchamberlain 13y ago
I'm a C++ developer by day, and I haven't tried Go yet. This looks really intresting. There is a massive gap for some Dataflow Programming based languages; not sure Go is quite there yet, but we're moving in the right direction.
- AlexanderDhoore 13y ago
  The future of programming is somewhere in the field of dataflow and (functional) reactive programming. It brings easy state to the declarative programming world.
  - seanmcdirmid 13y ago
    Said often and probably false. Neither frp nor dataflow are very expressive. Probably something still imperative but more reactive is the future.
aray 13y ago
Great explanation in document form of these patterns. Also very succinct and easy to reference. For those that would rather read this in presentation form: http://talks.golang.org/2012/concurrency.slide#1 http://talks.golang.org/2012/concurrency.slide#1 And for those that would rather watch a video presentation: https://www.youtube.com/watch?v=f6kdp27TYZs https://www.youtube.com/watch?v=f6kdp27TYZs
- enneff 13y ago
  Both those links are in the original article, but they are not the same content.
songgao 13y ago
> In Go, we can do this by closing a channel, because a receive operation on a closed channel can always proceed immediately, yielding the element type's zero value. for n := range c { select { case out <- n: case <-done: } } This, is brilliant.
- jbert 13y ago
  I wasn't quite getting that code when I first read it. Now I can see it allows you do "discard" all the pending reads from the input channel (and avoid writing to a possibly-closed 'out' channel), but wouldn't it be better to break the loop in this case for an immediate exit? i.e. go for 'interrupt' semantics, rather than 'drain'?
  - Zariel 13y ago
    I think its so that the channel is empty, and will have no references so it will be garbage collected.
  - Sajmani 13y ago
    Yes, interrupt is better, and the article explains how to do that a few paragraphs later: for n := range in { select { case out <- n * n: case <-done: return } } If you allow pipeline stages to interrupt their receive loop instead of draining the inbound channel, then _all_ send operations need to be governed by done. Otherwise the interrupted receive loop may block the upstream sender.
  - songgao 13y ago
    The loop (particularly `range c`) keeps getting element out of `c` and assigning it to `n`. If it's "done", then `n` wouldn't be sent into `out`, but still gets read from the channel `c` constantly. In other words, it's the loop itself that makes sure everything sent into channel `c` is consumed. If `break` is used, for loop is terminated once it's "done", hence nothing would be reading from channel `c`. The previous station in the pipeline would then block at sending operation.
noname123 13y ago
Cool. Can the concurrency guru's out there tell me what is the difference between the "Go Concurrency Pattern" and the Java SingleThreadExecutor? (http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor%28%29 http://docs.oracle.com/javase/6/docs/api/java/util/concurren...) ExecutorService pipeline = ExecutorService.newSingleThreadExecutor(); FutureTask<String> step1 = new FutureTask<String>(new Callable<String>() { public String call() { return "step1"; }}); executor.execute(step1); // execute single-step pipeline step1.cancel(true); // interrupt the task
- enneff 13y ago
  There is no "Go Concurrency Pattern" (singular). The article demonstrates some ways to use Go's concurrency primitives. Obviously there is a massive syntactic difference between your Java example and the equivalent Go code, but the other major difference that jumps out at me is that cancellation is baked into the Java library, whereas in Go the programmer determines how each of their pipeline stages should be shut down.
  - crypto5 13y ago
    > Obviously there is a massive syntactic difference between your Java example and the equivalent Go code With Java 8 and lambdas, syntax will be much cleaner, i.e. you can write something like: executor.execute(() -> "step1");
- songgao 13y ago
  I guess an obvious difference is that Go has these stuff built into runtime and syntax, which makes those light-weight threads / tasks a first class thing. Furthermore, is there a way in "Java SingleThreadExecutor" and its related components that allow automatic scheduling tasks on different threads? For example, you are on a quad-core and you want to have 4 threads running together, with millions of tasks mapped to 4 threads and dynamically balanced.
  - saryant 13y ago
obilgic 13y ago
The beauty of Go is, I have been writing these pipeline functions without even being explicitly taught the pattern.
- RobinUS2 13y ago
  Likewise, it simply makes sense if you have a proper understanding of channels and go-routines.
Jabbles 13y ago
Note the use of receive-only channels: func merge(cs ...<-chan int) <-chan int { Inside the function you cannot send to a member of cs - it will give you a compile-error. Outside the function you cannot send to its return value. An example to play with: http://play.golang.org/p/6AJLpZkLBe http://play.golang.org/p/6AJLpZkLBe
- AYBABTME 13y ago
  I've cleanly used that in a lil' game I'm working on. Players send their move to the game on a channel they expose through Move(). They receive regular updates of the game state on a channel they expose through Update(). type Player interface { Name() string Move() <-chan Move Update() chan<- State } > https://github.com/aybabtme/bomberman/blob/master/player/player.go#L30-L34 https://github.com/aybabtme/bomberman/blob/master/player/pla... The game reads the player moves, if moves are available[1]: for pState, player := range g.Players { if pState.Alive { select { case m := <-player.Move(): movePlayer(g, board, pState, m) default: } } } > https://github.com/aybabtme/bomberman/blob/master/bomberman.go#L250-L251 https://github.com/aybabtme/bomberman/blob/master/bomberman.... The game sends each turn's update to players that are ready to listen for it: for pState, player := range game.Players { pState.Board = board.Clone() pState.Turn = game.Turn() select { case player.Update() <- *pState: default: } } > https://github.com/aybabtme/bomberman/blob/master/bomberman.go#L262-L269 https://github.com/aybabtme/bomberman/blob/master/bomberman.... This leans to surprisingly clean implementations: * A websocket player [2] * A keyboard player [3] * A random AI player [4] Note that the rest of the game (`bomberman.go`) is a hairy ball that needs refactoring. [1]: this, to prevent an unresponsive player from hanging the game, or an AI player from unfairly computing longer than it's opponent [2]: https://github.com/aybabtme/bomberweb/blob/master/player/player.go https://github.com/aybabtme/bomberweb/blob/master/player/pla... [3]: https://github.com/aybabtme/bomberman/blob/master/player/input/input.go https://github.com/aybabtme/bomberman/blob/master/player/inp... [4]: https://github.com/aybabtme/bomberman/blob/master/player/ai/random.go https://github.com/aybabtme/bomberman/blob/master/player/ai/...
codygman 13y ago
Interesting, I'll have to compare pipelines in Go to pipelines in Haskell.
- signa11 13y ago
  the 'squinting at power series' paper by mcllroy is very cool demonstration of this technique using new-squeak
  - vanderZwan 13y ago
    To help the lazy: http://swtch.com/~rsc/thread/squint.pdf http://swtch.com/~rsc/thread/squint.pdf (it's also linked at the bottom of the article)
    - jerf 13y ago
      That appears to compare Go channels to "simple" lazy Haskell. A more apt Haskel comparison would be to something like "pipes" or "conduit", which will be quite different.
- signa11 13y ago
  > Interesting, I'll have to compare pipelines in Go to pipelines in Haskell check out the "power series, power serious" paper by doug-macllroy. he does that...
  - codygman 13y ago
    As jerf said, this is a little different since it's comparing to lazy io rather than pipes or conduit.
tete 13y ago
Node.js has another beautiful way of doing this. It's called streams there. https://github.com/substack/stream-handbook https://github.com/substack/stream-handbook http://nodejs.org/api/stream.html http://nodejs.org/api/stream.html

247 points by enneff 13y ago

I use channels to single-file data to a SQLite db. Data gets pushed into channels from multiple users / locations of the web app. A single goroutine (concurrently running function) reads from the channels one channel at a time and, for each channel, writes everything in the channel to the db in a single transaction. SQLite isn't the best at concurrency. Go's channels help resolve that issue nicely.

alexchamberlain 13y ago

I'm a C++ developer by day, and I haven't tried Go yet. This looks really intresting. There is a massive gap for some Dataflow Programming based languages; not sure Go is quite there yet, but we're moving in the right direction.

AlexanderDhoore 13y ago
The future of programming is somewhere in the field of dataflow and (functional) reactive programming. It brings easy state to the declarative programming world.
- seanmcdirmid 13y ago
  Said often and probably false. Neither frp nor dataflow are very expressive. Probably something still imperative but more reactive is the future.

aray 13y ago

Great explanation in document form of these patterns. Also very succinct and easy to reference. For those that would rather read this in presentation form: http://talks.golang.org/2012/concurrency.slide#1 http://talks.golang.org/2012/concurrency.slide#1 And for those that would rather watch a video presentation: https://www.youtube.com/watch?v=f6kdp27TYZs https://www.youtube.com/watch?v=f6kdp27TYZs

enneff 13y ago
Both those links are in the original article, but they are not the same content.

songgao 13y ago

> In Go, we can do this by closing a channel, because a receive operation on a closed channel can always proceed immediately, yielding the element type's zero value. for n := range c { select { case out <- n: case <-done: } } This, is brilliant.

jbert 13y ago
I wasn't quite getting that code when I first read it. Now I can see it allows you do "discard" all the pending reads from the input channel (and avoid writing to a possibly-closed 'out' channel), but wouldn't it be better to break the loop in this case for an immediate exit? i.e. go for 'interrupt' semantics, rather than 'drain'?
- Zariel 13y ago
  I think its so that the channel is empty, and will have no references so it will be garbage collected.
- Sajmani 13y ago
  Yes, interrupt is better, and the article explains how to do that a few paragraphs later: for n := range in { select { case out <- n * n: case <-done: return } } If you allow pipeline stages to interrupt their receive loop instead of draining the inbound channel, then _all_ send operations need to be governed by done. Otherwise the interrupted receive loop may block the upstream sender.
- songgao 13y ago
  The loop (particularly `range c`) keeps getting element out of `c` and assigning it to `n`. If it's "done", then `n` wouldn't be sent into `out`, but still gets read from the channel `c` constantly. In other words, it's the loop itself that makes sure everything sent into channel `c` is consumed. If `break` is used, for loop is terminated once it's "done", hence nothing would be reading from channel `c`. The previous station in the pipeline would then block at sending operation.

noname123 13y ago

Cool. Can the concurrency guru's out there tell me what is the difference between the "Go Concurrency Pattern" and the Java SingleThreadExecutor? (http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/Executors.html#newSingleThreadExecutor%28%29 http://docs.oracle.com/javase/6/docs/api/java/util/concurren...) ExecutorService pipeline = ExecutorService.newSingleThreadExecutor(); FutureTask<String> step1 = new FutureTask<String>(new Callable<String>() { public String call() { return "step1"; }}); executor.execute(step1); // execute single-step pipeline step1.cancel(true); // interrupt the task

enneff 13y ago
There is no "Go Concurrency Pattern" (singular). The article demonstrates some ways to use Go's concurrency primitives. Obviously there is a massive syntactic difference between your Java example and the equivalent Go code, but the other major difference that jumps out at me is that cancellation is baked into the Java library, whereas in Go the programmer determines how each of their pipeline stages should be shut down.
- crypto5 13y ago
  > Obviously there is a massive syntactic difference between your Java example and the equivalent Go code With Java 8 and lambdas, syntax will be much cleaner, i.e. you can write something like: executor.execute(() -> "step1");
songgao 13y ago
I guess an obvious difference is that Go has these stuff built into runtime and syntax, which makes those light-weight threads / tasks a first class thing. Furthermore, is there a way in "Java SingleThreadExecutor" and its related components that allow automatic scheduling tasks on different threads? For example, you are on a quad-core and you want to have 4 threads running together, with millions of tasks mapped to 4 threads and dynamically balanced.
- saryant 13y ago

obilgic 13y ago

The beauty of Go is, I have been writing these pipeline functions without even being explicitly taught the pattern.

RobinUS2 13y ago
Likewise, it simply makes sense if you have a proper understanding of channels and go-routines.

Jabbles 13y ago

Note the use of receive-only channels: func merge(cs ...<-chan int) <-chan int { Inside the function you cannot send to a member of cs - it will give you a compile-error. Outside the function you cannot send to its return value. An example to play with: http://play.golang.org/p/6AJLpZkLBe http://play.golang.org/p/6AJLpZkLBe

AYBABTME 13y ago
I've cleanly used that in a lil' game I'm working on. Players send their move to the game on a channel they expose through Move(). They receive regular updates of the game state on a channel they expose through Update(). type Player interface { Name() string Move() <-chan Move Update() chan<- State } > https://github.com/aybabtme/bomberman/blob/master/player/player.go#L30-L34 https://github.com/aybabtme/bomberman/blob/master/player/pla... The game reads the player moves, if moves are available[1]: for pState, player := range g.Players { if pState.Alive { select { case m := <-player.Move(): movePlayer(g, board, pState, m) default: } } } > https://github.com/aybabtme/bomberman/blob/master/bomberman.go#L250-L251 https://github.com/aybabtme/bomberman/blob/master/bomberman.... The game sends each turn's update to players that are ready to listen for it: for pState, player := range game.Players { pState.Board = board.Clone() pState.Turn = game.Turn() select { case player.Update() <- *pState: default: } } > https://github.com/aybabtme/bomberman/blob/master/bomberman.go#L262-L269 https://github.com/aybabtme/bomberman/blob/master/bomberman.... This leans to surprisingly clean implementations: * A websocket player [2] * A keyboard player [3] * A random AI player [4] Note that the rest of the game (`bomberman.go`) is a hairy ball that needs refactoring. [1]: this, to prevent an unresponsive player from hanging the game, or an AI player from unfairly computing longer than it's opponent [2]: https://github.com/aybabtme/bomberweb/blob/master/player/player.go https://github.com/aybabtme/bomberweb/blob/master/player/pla... [3]: https://github.com/aybabtme/bomberman/blob/master/player/input/input.go https://github.com/aybabtme/bomberman/blob/master/player/inp... [4]: https://github.com/aybabtme/bomberman/blob/master/player/ai/random.go https://github.com/aybabtme/bomberman/blob/master/player/ai/...

codygman 13y ago

Interesting, I'll have to compare pipelines in Go to pipelines in Haskell.

signa11 13y ago
the 'squinting at power series' paper by mcllroy is very cool demonstration of this technique using new-squeak
- vanderZwan 13y ago
  To help the lazy: http://swtch.com/~rsc/thread/squint.pdf http://swtch.com/~rsc/thread/squint.pdf (it's also linked at the bottom of the article)
  - jerf 13y ago
    That appears to compare Go channels to "simple" lazy Haskell. A more apt Haskel comparison would be to something like "pipes" or "conduit", which will be quite different.
signa11 13y ago
> Interesting, I'll have to compare pipelines in Go to pipelines in Haskell check out the "power series, power serious" paper by doug-macllroy. he does that...
- codygman 13y ago
  As jerf said, this is a little different since it's comparing to lazy io rather than pipes or conduit.

tete 13y ago

Node.js has another beautiful way of doing this. It's called streams there. https://github.com/substack/stream-handbook https://github.com/substack/stream-handbook http://nodejs.org/api/stream.html http://nodejs.org/api/stream.html