Channels-driven concurrency with Go

go-channels-1-generator.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (1) Generator: function that returns the channel

func boring(msg string) <-chan string { // Return receive-only channel of strings
    c := make(chan string)
    go func() { // We launch goroutine from inside the function
        for i := 0; ; i++ {
            c <- fmt.Sprintf("%s %d", msg, i)
            time.Sleep(time.Duration(rand.Intn(1e3)) * time.Milliseconds)
        }
    }()
    return c
}

c := boring("boring!") // Function returning a channel.
for i := 0; i < 5; i++ {
    fmt.Printf("You say: %q\n", <-c)
}
fmt.Println("You're boring: I'm leaving.")

// More instances
func main() {
    joe := boring("Joe")
    ann := boring("Ann")
    
    for i := 0; i < 10; i++ {
        fmt.Println(<-joe)
        fmt.Println(<-ann)
    }
    fmt.Println("You're boring: I'm leaving.")
}

go-channels-2-multiplexing.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (2) Fan-in

func fanIn(input1, input2 <-chan string) <-chan string {
    c := make(chan)
    go func() { for { c <- <-input1 } }()
    go func() { for { c <- <-input2 } }()
    return c
}

c := fanIn(boring("Joe"), boring("Ann"))
for i := 0; i < 10; i++ {
    fmt.Println(<-c)
}
fmt.Println("You're boring: I'm leaving.")

go-channels-3-seq.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (3) Restoring sequencing

type Message struct {
    str string
    wait chan bool
}

// Each speaker must wait for a go-ahead
for i :=0; i < 5; i++ {
    msg1 := <-c; fmt.Println(msg1.str)
    msg2 := <-c; fmt.Println(msg2.str)
    msg1.wait <- true
    msg2.wait <- true
}

waitForIt := make(chan bool) // Shared between all messages.

func boring(msg string) <-chan string { // Return receive-only channel of strings
    c := make(chan string)
    go func() { // We launch goroutine from inside the function
        for i := 0; ; i++ {
            c <- Message( fmt.Sprintf("%s: %d", mgs, i), waitForIt )
            time.Sleep(time.Duration(rand.Intn(2e3)) * time.Milliseconds)
            <-waitForIt        
        }
    }()
    return c
}

go-channels-4-select.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (4) Select control structure

// Fan-in using select 
func fanIn(input1, input2 <-chan string) <-chan string {
    c := make(chan)
    go func() { 
        for {
            select {
                case s := <-input1; c <-s 
                case s := <-input2; c <-s 
            }    
        } 
    }()
    return c
}

// Timeout using select
func main() {
    c := boring("Joe")
    for {
        select {
            case s := <-c:
                fmt.Println(s)
            case <-time.After(1 * time.Second):
                fmt.Println("You're too slow!")
                return
        }
    }
}

// Timeout for whole conversation using select
func main() {
    c := boring("Joe")
    timeout := time.After(1 * time.Second)
    for {
        select {
            case s := <-c:
                fmt.Println(s)
            case <-timeout:
                fmt.Println("You talk too much!")
                return
        }
    }
}

go-channels-5-quit.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (5) Quit channel

// Simple boolean flag
func main() {
    quit := make(chan bool)
    c := boring("Joe") 
    for i := rand.Intn(10); i >= 0; i-- { fmt.Println(<-c) }
    quit <- true
}


func boring(msg string, quit chan bool) <-chan string { 
    c := make(chan string)
    go func() { 
        for {
            select {
                case c <- fmt.Sprintf("%s %d", msg, i):
                    // do nothing
                case <-quit:
                    return
            }
        }
    }()
    return c
}

// "Cleanup" function on quit

func main() {
    quit := make(chan string)
    c := boring("Joe") 
    for i := rand.Intn(10); i >= 0; i-- { fmt.Println(<-c) }
    quit <- "Bye!"
    fmt.Printf("Joe says: %q\n", <-quit)
}


func boring(msg string, quit chan bool) <-chan string { 
    c := make(chan string)
    go func() { 
        for {
            select {
                case c <- fmt.Sprintf("%s %d", msg, i):
                    // do nothing
                case <-quit:
                    // cleanup()
                    quit <- "See you!"
                    return
            }
        }
    }()
    return c
}

go-channels-6-chain.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (6) Daisy-chain

func f(left, right chan int) {
    left <- 1 + <- right
} 

func main() {
    const n = 100000
    leftmost := make(chan int)
    right := leftmost
    left := leftmost
    for i := 0; i < n; i++ {
        right := make(chan int)
        go f(left, right)
        left = right
    }
    go func(c chan int) { c <- 1 }(right)
    fmt.Println(<-leftmost)
}

go-channels-7-search.go

// Channels-driven concurrency with Go
// Code examples from Rob Pike's talk on Google I/O 2012:
// http://www.youtube.com/watch?v=f6kdp27TYZs&feature=youtu.be
//
// Concurrency is the key to designing high performance network services. 
// Go's concurrency primitives (goroutines and channels) provide a simple and efficient means 
// of expressing concurrent execution. In this talk we see how tricky concurrency 
// problems can be solved gracefully with simple Go code.

// (7) Fake Google search

var (
    Web = fakeSearch("web")
    Image = fakeSearch("image")
    Video = fakeSearch("video")
)

type Search func(query string) Result

func fakeSearch(kind string) Search {
    return func(query string) Result {
        time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        return Result(fmt.Sprintf("%s result for %q\n", kind, query))
    }
}

// Replication in order to avoid timeouts
func First(query string, replicas ...Search) Result {
    c := make(chan Result)
    searchReplica := func(i, int) { c <- replicas[i](query) }
    for i := range replicas {
        go searchReplica(i)
    }
    return <-c
}

// Main search function
func Google(query sting) (results []Result) {
    c := make(chan Result)

    go func() { c <- First(query, Web1, Web2) }()
    go func() { c <- First(query, Image1, Image2) }()
    go func() { c <- First(query, Video1, Video2) }()

    timeout := time.After(80 * time.Millisecond)

    for i := 0; i < 3; i++ {
        select {
            case result := <-c:
                results = append(results, result)
            case <-timeout:
                fmt.Println("timed out.")
                return
        }
    }
}