Cytora Golang Academy: Part Three

The purpose of the Cytora Golang Academy blog post series is to share the material from our Goroutines & Channels talks that I have been working on at Cytora.

Everyone has thoroughly enjoyed preparing and listening to this series and I hope you will enjoy the series of magic show themed Cytora Golang Academy series too!

This is the third blog post in the series. You can find part one here and part two here if you missed them.

Master Illusionist: concurrency patterns

My third and final session in the Goroutines & Channels talk series rounds off the knowledge gained in the first two sessions with some useful concurrency patterns - that have applications outside of the world of our magic theme.

Unidirectional channels

The previous two sessions have explored the behaviour of channels, and understanding unidirectional channels will round off our knowledge.

• Bidirectional channels can be implicitly cast to unidirectional channels inside function scope, as seen in the example below.
• When using channels as function parameters, you can specify if a channel is meant to only send or receive values. <-chan is a receive-only channel, while chan<- is a send-only channel.
• The specificity of unidirectional channels increases the type-safety of the program. A compile error will be raised when attempting an unsupported operation.

Signalling to Goroutines that work is finished

The “Channel Card Shuffle” is a trick we saw in the second session, but we will now introduce the modification of passing cards back to the main Goroutine for it to reveal them. It makes sense for the main Goroutine, the show orchestrator, to perform the grand card reveal!

The card shuffler has a predetermined amount of cards that it writes to a send-only cardPicks channel every 2 seconds. Once it runs out of cards, the shuffler takes a down shuts down.

var tops = []string {"queen of hearts", "four of clubs", "jack of spades", "ace of diamonds"}
func channelShuffle(cardPicks chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("The magician starts shuffling the cards.")
    time.Sleep(1 * time.Second)

    for i := 0; i < len(tops); i++{
        time.Sleep(2 * time.Second)
        cardPicks <- tops[i] //send the topcard back
    }

    fmt.Println("The magician takes a bow!")
}

The main Goroutine starts a shuffler Goroutine, reveals only 2 cards from the shuffler and then attempts to instruct the shuffler to shut down by closing the cardPicks channel. This results in a panic, due to the behaviour of channels.

As we remember from session two, sending data on a closed channel panics to avoid blocking Goroutines. Closing the cards channel to signal we no longer wish to receive cards has backfired, as the shuffler continues to try to pass cards on the channel and then causes the Go runtime to panic. 💥

Signalling to Goroutines that work is finished is a common problem out in the wild/in production.

A solution is to use a signal channel. This is a channel whose sole purpose is to signal, not transfer information. It uses the empty struct{} to use as little memory as possible.

signal:= make(chan struct{})

The Goroutine that is sending to the channel should be the one to close the channel, but we can close an additional channel to tell the sending Goroutine to stop. This will avoid the panic we saw in the channel shuffle.

We rewrite the card shuffler to take two channels as parameters and listen to both of them using the select we saw introduced in the previous session. Once the shuffler receives a message on the signal channel, it will know to shut down and stop sending cards.

var topCards = []string {"queen of hearts", "four of clubs", "jack of spades", "ace of diamonds"}
func channelSignalShuffle(cardPicks chan<- string, signal <-chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("The magician starts shuffling the cards.")
    time.Sleep(1 * time.Second)

    i := 0
    for {
        select {
        case cardPicks <- topCards[i]:
            i++
            time.Sleep(2 * time.Second)
        case <- signal:
            close(cardPicks) // we won't be sending any more cards
            time.Sleep(2 * time.Second)
            fmt.Println("The magician takes a bow!")
            return
        }
    }
}

The main Goroutine reads the cards it is interested in and closes the signal channel once it is no longer interested in receiving more. The channel card shuffle now works like a charm! 🃏

Pipelines

Pipelines are a more interesting pattern. They are a series of stages/ Goroutines running the same function connected by channels.

In each stage, the Goroutines:

• Receive values from upstream via inbound channels
• Perform some function on that data, usually producing new values
• Send values downstream via outbound channels

Each stage has any number of inbound and outbound channels, except the first and last stages, which have only outbound or inbound channels, respectively.

The “Straitjacket Escape” performers consist of two Goroutines: an assistant and a magician. The assistant Goroutine puts the jacket on the magician and locks it in chains. The magician escapes from the jacket and the locks! 🔒 👏

The assistant receives a series of actions it must take and then writes them to a channel to communicate them with the magician. The magician receives a series of actions that the assistant has taken and then returns the opposite action to a channel.

func assistant(actions []string) <-chan int {
    out := make(chan int)
    go func() {
        for i, item := range actions {
            fmt.Printf("The assistant %s!\n", item)
            time.Sleep(2 * time.Second)
            out <- i
        }
        close(out)
    }()
    return out
}
func magician(items <-chan int) <-chan string {
    magicianActions := []string{"unlocks jacket", "unties jacket", "takes off jacket"}
    out := make(chan string)
    go func() {
        for i := range items {
            time.Sleep(2 * time.Second)
            out <- magicianActions[i]
        }
        close(out)
    }()
    return out
}

The “Straitjacket Escape” trick uses pipelines for passing the index of the assistant’s actions to the magician, which then knows what the opposite action should be.

The main Goroutine sends the actions it must take and lets the performers do their trick. It then listens to the magician’s opposite actions and prints them out to reveal the escape, as the final step of this short pipeline.

Buffered channels & worker pools

Next up, we examine the implementation of worker pools using buffered channels. This is an important pattern in production and is a worthwhile example to understand and master as it allows the preallocation and reuse of concurrency resources.

A worker pool needs the following core functionalities:

• Creation of a pool of Goroutines which listen on an input buffered channel waiting for jobs to be assigned
• Addition of jobs to the input buffered channel
• Writing results to an output buffered channel after job completion
• Read and print results from the output buffered channel

The “Infinite Hat” trick starts up three concealer Goroutines which repeatedly take items and place them into an infinitely large hat. The contents of impossibly large hat are then revealed by the main Goroutine! 🎩

The concealer worker Goroutine receives items on a channel and sends them off to the hat channel to hide them in the hat.

func concealerWorker(id int, items <- chan string, hat chan<- string, wg *sync.WaitGroup) {
    defer wg.Done()
    for item := range items {
        time.Sleep(1 * time.Second)
        fmt.Printf("Concealer %d hides %s in the hat!\n", id, item)
        time.Sleep(1 * time.Second)
        hat <- item
    }
}

The concealer worker pool starts up a predetermined number of workers and passes them the channels they need to do their work. It also uses a sync.WaitGroup to make sure that it waits for all the concealers to finish their work before shutting down.

func createConcealerPool(items <- chan string, hat chan<- string, noOfWorkers int) {
    var wg sync.WaitGroup // know when all workers have completed
    for i := 0; i < noOfWorkers; i++ {
        wg.Add(1)
        go concealerWorker(i, items, hat, &wg)
    }
    wg.Wait()
    fmt.Println("Concealers have finished their work!")
    close(hat) // all concealers have finished no more to conceal/reveal
}

The fetchItems Goroutine writes the items it wishes placed in the hat by the concealers to the items channel. Once it has finished its item, it closes down its channel to signal no more items will be incoming.

func fetchItems(itemsCh chan <- string) {
    items := []string{"bowling ball", "flowers", "rabbit", "melon", "dove", "boot"}
    for _, item := range items {
        time.Sleep(1 * time.Second)
        itemsCh <- item // populate the items channel
    }
    fmt.Println("Finished fetching items!")
    close(itemsCh) // all done with the items
}

The main Goroutine creates all the channels required and starts up the worker pool and the item fetcher. The concealer workers hide items in the hat until the channel is closed. The main Goroutine then reveals the items from the hat channel to show off the impossibly large infinite hat.

An interesting thing to note about this trick is that there is no dedicated Hat struct. The Hat is simply represented by a channel which holds the items that concealers are placing in it. 🎩

Real world examples

In the first session, I discussed the use of Goroutines in production. Let’s revisit the list and imagine some solutions using the knowledge from all three sessions.

• Background processing of large files: could be implemented by splitting the file into batches and using pipelines for processing
• Handling user requests in web servers: spinning off a Goroutine per request then communicating via channels; we can use signal channels or contexts for cancellation/work stopping
• Pushing tracking events/logs in the background: could be implemented using buffered channels and worker pools to limit throughput and reuse workers

The concurrency patterns we’ve learned this session have a wide variety of applications… and not just in the field of magic! 🎩

Final curtain drop/Conclusions

This brings us to our conclusions and final curtain drop. The magic tricks in this session demonstrate the implementations of some common concurrency patterns.

The full list of magic tricks presented in the session is:

• Card Shuffle trick: closing a signal channel to tell Goroutines to finish their work
• Straitjacket Escape trick: using the pipeline pattern to pass actions to different workers and produce different results
• Infinite Hat trick: using the worker pool pattern to have workers pick up data as soon as it’s ready

Your humble Cytora magician takes a final bow, takes her hat off to you and thanks you for staying for the show! 🎩

Talk Materials

My slide deck was designed to be well documented and self standing.

You can find it here. It fully elaborates on the short synopsis I have written in this blogpost.