Go concurrency and sychronization - Part 2: Refactor iteration

As I pointed out in my previous post there are several misleading design issues associated with this solution. We had one channel to close two goroutines. This pattern of course can be made working by sending on the channel, but we could also utilize the two channels that we already use for synchronization.

So this is now an improved version with two seperate channels.

func main() {

    // we only use the two main channels
    printOdd := make(chan struct{})
    printEven := make(chan struct{})

    wg := sync.WaitGroup{}

    go func() {
        wg.Add(1)

        start := 0

        // since we only have a single channel to receive, we can now remove the select statement
        // and instead loop on the channel itself
        for range printEven {
            fmt.Println(start)
            start = start + 2
            time.Sleep(time.Second)
            printOdd <- struct{}{}
        }

        fmt.Println("finished odd printing")
        wg.Done()
    }()


    go func() {
        wg.Add(1)

        start := 1

        // same as above
        for range printOdd {
            fmt.Println(start)
            start = start + 2
            time.Sleep(time.Second)
            printEven <- struct{}{}
        }
        fmt.Println("finished odd printing")
        wg.Done()
    }()


    reader := bufio.NewReader(os.Stdin)
    fmt.Println("Press enter to cancel")
    fmt.Println("---------------------")
    // trigger the ping-pong
    printOdd <- struct{}{}

    reader.ReadString('\n')
    fmt.Println("finished")

    // so now we close each channel seperately
    close(printEven)
    close(printOdd)

    // maybe panics, because we are likely to send on a closed channel
    wg.Wait()

    // output: panic: send on closed channel
}

We now have a single channel to receive data from, so we may now replace the select statement from part one with a range loop.

The tricky one here is that we may send on a channel which is already closed. This leads to a panic and has to be catched. Since we do not have a chance to check if a channel is closed (without receiving) we just try to send some data in, and if it leads to a panic, we can just recover from it.

package main

import (
    "bufio"
    "fmt"
    "os"
    "sync"
    "time"
)

func main() {
    printOdd := make(chan struct{})
    printEven := make(chan struct{})

    wg := &sync.WaitGroup{}

    go func() {
        wg.Add(1)

        // move the Done() call into a deferred function
        // and now we are able to recover from a possible panic, as above
        defer func() {
            wg.Done()
            if err := recover(); err != nil {
                fmt.Println("send on closed channel occured, but recoverd")
            }
        }()

        start := 0

        for range printEven {
            fmt.Println(start)
            start = start + 2
            time.Sleep(time.Second)
            printOdd <- struct{}{}
        }
    }()

    go func() {
        wg.Add(1)

        // same as above
        defer func() {
            wg.Done()
            if err := recover(); err != nil {
                fmt.Println("send on closed channel occured, but recoverd")
            }
        }()

        start := 1

        for range printOdd {
            fmt.Println(start)
            start = start + 2
            time.Sleep(time.Second)
            printEven <- struct{}{}
        }
    }()

    reader := bufio.NewReader(os.Stdin)
    fmt.Println("Press enter to cancel")
    fmt.Println("---------------------")
    // trigger the ping-pong
    printEven <- struct{}{}

    reader.ReadString('\n')
    fmt.Println("finished")

    close(printEven)
    close(printOdd)
    wg.Wait()
}

So in the code snippet above, I redesigned the whole deferred function call. It's now more to do. The WaitGroup.Done() is called and of course the recover(). Since we can only call the recover in a defereed function, we have to make use of this approach. How can one tell when a panic() is triggerd? In any case, the current method returns instantly and therefore puts the deferred function onto the stack. In case of panic we now make sure recover from it.

Currently this solution looks very astonishing and is in fact pretty bullet-proof. But maybe you have noticed that there is the same code twice in my program. Of course there is some place for refactoring. This will be covered in the first section of the next part in this series. The other part is going to create a fully-fledged data pipeline out of it.

Stay curious!