Go Concurrency

Goroutines

A Tour of Go: Goroutines

package main

import (
    "fmt"
    "time"
)

func say(s string) {
    for i := 0; i < 3; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Println(s)
    }
}

Serial execution

Execute

func main() {
    say("hello")
    say("world")
}

Concurrent Execution

Execute

func main() {
    go say("hello")
    say("world")
}

go says("hello"):

• is a non-blocking call: it returns "immediately",
  and doesn't wait for the completion of the function.

• its gets executed:

  • in the background,

  • in its own lightweight process

  • with its own execution thread

• concurrently with say("hello")

Experiment

Run the same code several times.

What's going on?

Pitfall

• Concurrency non-determinism

• Programs are harder to understand

• Beware the Heisenbugs!

A Quick and Dirty Fix

Make sure that all messages are printed:

func main() {
    go say("hello")
    go say("world")
    time.Sleep(3 * time.Second)
}

A Proper Fix

WaitGroup

import "sync"

func main() {
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        say("hello")
        wg.Done()
	}()
    say("world")
    wg.Wait()
}

Data races

package main

import "time"

var counter = 0

func add(i int) {
    counter += i
}


func main() {
    for {
        println(counter)
        //  Hammer the counter!
        for i := 0; i < 1000; i++ {
            go add(1)
            go add(-1)
        }
        //  Let the dust settle
        time.Sleep(time.Second) 
    }
}

Ooops

Shared Variables

• In C/C++, you would use a lock (mutex) to ensure that at most one process can access the counter variable at any given moment.

This also works in Go, but it's not idiomatic. Instead:

Don't communicate by sharing memory;
share memory by communicating.

The core communication device is a channel.

Channels

Create

Channels are typed and (optionally) buffered:

message := make(chan string, 1)

numbers := make(chan int, 10)

ready := make(chan bool) // same as make(chan bool, 0)

Read & Write

message <- "Hello world!"
s := <-message
fmt.Println(s)

t := <-message //  blocked

message <- "Hello"
message <- "world! //  blocked

numbers <- 0
numbers <- 1
numbers <- 2
fmt.Println(<-numbers)
fmt.Println(<-numbers)
fmt.Println(<-numbers)
fmt.Println(<-numbers) //  blocked

Unbuffered channels seem useless at first sight

ready <- true // 

status := <-ready // 

... but they actually shine

• in a concurrent setting

• as a synchronisation mecanism!

We're not ready yet

var ready = make(chan bool)

func Greetings() {
    for i:=0; i<10; i++ {
        fmt.Println("Hello world!")
    }
    ready <- true
} 

func main() {
    fmt.Println("begin")
    go Greetings()
    <-ready // Wait for the end!
    fmt.Println("end")
}

Parallel Computations

var c = make(chan int, 2)

func sum(s []int) {
    sum := 0
    for _, v := range s {
        sum += v
    }
    c <- sum
}

func main() {
    s := []int{7, 2, 8, -9, 4, 0}
    go sum(s[:len(s)/2])
    go sum(s[len(s)/2:])
    x, y := <-c, <-c
    println(x, y, x+y)
}

Throttling Process

func throttled(input chan string) chan string {
    output := make(chan string)
    go func() {
        for {
            output <- <-input
            time.Sleep(3 * time.Second)
        }
    }()
    return output
}

func main() {
    input := make(chan string)
    output := throttled(input)
    for i:=0; i<10; i++ {
        input <- "Hello world!"
        fmt.Println(<-output)
    }
}

No more data races

Channels: safe to use concurrently

var counter = 0
var ch = make(chan int, 100)

func add(i int) {
    ch <- i
}

func counterHandler() {
    for {
        counter += <-ch
    }
}

func main() {
    go counterHandler() //  New!
    go display() // 
    for {
        time.Sleep(time.Second) // 
        //  Hammer the counter!
        for i := 0; i < 1000; i++ {
            go add(1)
            go add(-1)
        }
    }
}

Timers

Sequential implementation of two concurrent processes:

for {
    for i:=0; i < 10; i++ {
        // Executed at ~10Hz
        fmt.Println("FAST")
        time.Sleep(time.Second / 10) 
    }
    // Executed at ~1Hz
    fmt.Println("SLOW") 
}

Consider instead

func Printer(m string, d time.Duration) {
    for {
        fmt.Print(m)
        time.Sleep(d)
    }
}

func main() {
    go Printer("FAST", time.Second / 10)
    Printer("SLOW", time.Second)
}

Even better

func Printer(m string, d time.Duration) {
    for {
        wait := time.After(d)
        fmt.Println(m)
        <-wait
    }
}

func main() {
    go Printer("FAST", time.Second / 10)
    Printer("SLOW", time.Second)
}

or use the time module Ticker & Timer API!

func HappyNewYear(year int) chan struct{} {
    trigger := make(chan struct{})
    newYear := time.Date(
        year, time.January, 
        0, 0, 0, 0, 0, time.UTC)

    go func() {
        for {
            if time.Now().After(newYear) {
                fmt.Println(" Happy New Year!")
                trigger <- struct{}{}
                return
            }
            time.Sleep(time.Second)
        }
    }()

    return trigger
}

func main() {
    <-HappyNewYear(2024)
}