Concurrency Under Control: Channeling the Monitor Pattern in Go

Introduction

Many times when writing concurrent code, the need arises to protect a certain resource, or resources from concurrent access. A way of doing this is by using the Monitor pattern.

Implementing this pattern usually involves the use of the locks and condition variables. Implementing the pattern like this can be done in Go as well, however there is a much simpler way using channels. This offers several advantages:

1. Built-in synchronization: channels can handle all synchronization automatically, there is no need for explicit locks or waits, and operations block when needed , like when sending to a full buffer or receiving from an empty one.
2. Simplicity and readablity: using channels the implementation makes it concise, and very readable.
3. Safety: Channel operations are atomic. Also, there is no risk of forgetting to release locks, or notify waiting threads.
4. Go’s design philosophy: using channels allows for more idiomatic Go code, code which leverages language-specific features.
5. Error prevention: when using locks and condition variables, you will need to explicitly acquire a lock and also release, condition variables to signal whether a buffer is full or empty, and make sure signalling between threads is done correctly. This requires explicit and precise error handling.

Basically a monitor is a synchronization construct. What this comes down to is that only a single thread can execute a particular section of code at a particular time. This piece is called the critical section. A monitor also provides mechanisms for threads to wait for certain conditions to become true. In the example in this article it is the presence of items in the buffer.

Implementation

Start by importing these packages:

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

Line by line:

1. Declare the package.
2. The app will be writing to the console or terminal, so import the fmt packages.
3. Each go routine will wait for a random number of milliseconds, so import the math/rand package.
4. When running the go routines, the test app will use sync.WaitGroup as a synchronization mechanism to coordinate when all go routines have terminated. This allows the main() function to wait for multiple concurrent operations to finish before proceeding.
5. When producing or consuming items, the go routines wait a random number of milliseconds before producing or consuming the next item, that is why the time package is needed.

The Buffer struct

The Buffer type itself is quite simple:

type Buffer struct {
	items chan string
}

This consists of a channel of strings.

Creating this buffer is equally simple:

func CreateBuffer(capacity int) *Buffer {
	return &Buffer{
		items: make(chan string, capacity),
	}
}

All this function this is create a channel of a given capacity.

Adding items to this buffer goes like this:

func (b *Buffer) Produce(item string) {
	b.items <- item
	fmt.Printf("Producer: Added %s\n", item)
}

Line by line:

1. The function receives an item as its single parameter.
2. It tries to send this item to the channel, however if the channel is full, this will block execution. Once space has become available execution of the code will resume.
3. Once the item has been sent to the channel, a confirmation message is printed.

The Consume() works in a similar way:

func (b *Buffer) Consume() string {
	// Channel receive will block if buffer is empty
	item := <-b.items
	fmt.Printf("Consumer: Removed %s\n", item)
	return item
}

Line by line:

1. The first thing the method does is try to get an item from the channel. Execution blocks if the channel is empty. If items become available, execution resumes.
2. The item retrieved from the channel is assigned to item
3. Next a confirmation message is printed.
4. And the retrieved item is returned to the caller.

Creating and consuming items

To test this setup two functions are needed, one to produce items, and the next one to consume them. First the function to produce items:

func producerTask(buffer *Buffer, numItems int, wg *sync.WaitGroup) {
	defer wg.Done()

	for i := 0; i < numItems; i++ {
		item := fmt.Sprintf("item_%d", i)
		buffer.Produce(item)
		// Sleep for a random duration between 100ms and 125ms
		time.Sleep(time.Duration(100+rand.Intn(25)) * time.Millisecond)
	}
}

Line by line:

1. The function receives three parameter: the buffer which will receive the items, the number of items to be produced, and wait group which will be signalled once the function exits.
2. The first thing to do is make sure completion is signalled to the wait group, using the defer statement.
3. Next in a loop, produce items, and call the Produce() method on the buffer to put them to the buffer. Note that if the buffer is full execution will be suspended at this point.
4. Wait for a random number of milliseconds, before producing the next item.

Now the function to consume the items:

func consumerTask(buffer *Buffer, numItems int, wg *sync.WaitGroup) {
	defer wg.Done()

	for i := 0; i < numItems; i++ {
		buffer.Consume()
		// Sleep for a random duration between 200ms and 1600ms
		time.Sleep(time.Duration(200+rand.Intn(1400)) * time.Millisecond)
	}
}

Line by line:

1. The function receives three parameters: the buffer from which the items are consumed, the number of items to consume and a wait group which is used for signalling the function has ended.
2. The first thing to do is make sure the wait group is signalled when the function ends using the defer statement.
3. Next in a loop, consume the numItems items. Note that the call to Consume() will suspend execution if the buffer is empty and no item can be retrieved.
4. After an item has been retrieved, the function waits for a random number of milliseconds before consuming the next item.

Testing time

Now with this all out of the way, time to test!

First set up the buffer itself, with a capacity of five:

	bufferCapacity := 5
	sharedBuffer := CreateBuffer(bufferCapacity)

Next set the number of producers and consumers. Also set the number of consumers, producers and the number of items each producers produces. Using a very rough formula, calculate the number items each consumer must consume, so spread the load evenly.

	numProducers := 2
	numConsumers := 3
	itemsPerProducer := 10
	itemsToConsume := numProducers * itemsPerProducer / numConsumers // Distribute items roughly

Now declare to sync.WaitGroup instances to track the completion of the producer and the consumer go routines. Having two separate wait groups allows the program to wait for both groups independently.

Next starting producing and consuming items. This follows the same pattern both for producers and consumers:

1. Increment the appropiate wait group counter.
2. Launch task in a separate go routine.
3. In each task, like in producerTask() a call to Done() is made to decrement the wait group counter.

	for i := 0; i < numProducers; i++ {
		producerWg.Add(1)
		go producerTask(sharedBuffer, itemsPerProducer, &producerWg)
	}

	// Start consumer goroutines
	for i := 0; i < numConsumers; i++ {
		consumerWg.Add(1)
		go consumerTask(sharedBuffer, itemsToConsume, &consumerWg)
	}

Next wait for all producers and consumers to finish:

	producerWg.Wait()
	consumerWg.Wait()

Finally print a message to the screen, to show all tasks have finished, and show how many items have been left in the buffer.

	fmt.Println("\nAll tasks finished.")

	// Count any remaining items in the buffer
	remaining := len(sharedBuffer.items)
	if remaining > 0 {
		fmt.Printf("Final buffer state: %d items remaining\n", remaining)
	} else {
		fmt.Println("Final buffer state: empty")
	}

The full main() function looks like this:

func main() {
	bufferCapacity := 5
	sharedBuffer := CreateBuffer(bufferCapacity)

	numProducers := 2
	numConsumers := 3
	itemsPerProducer := 10
	itemsToConsume := numProducers * itemsPerProducer / numConsumers 

	var producerWg sync.WaitGroup
	var consumerWg sync.WaitGroup


	for i := 0; i < numProducers; i++ {
		producerWg.Add(1)
		go producerTask(sharedBuffer, itemsPerProducer, &producerWg)
	}

	for i := 0; i < numConsumers; i++ {
		consumerWg.Add(1)
		go consumerTask(sharedBuffer, itemsToConsume, &consumerWg)
	}


	producerWg.Wait()
	consumerWg.Wait()

	fmt.Println("\nAll tasks finished.")


	remaining := len(sharedBuffer.items)
	if remaining > 0 {
		fmt.Printf("Final buffer state: %d items remaining\n", remaining)
	} else {
		fmt.Println("Final buffer state: empty")
	}
}

Conclusion

Even though the monitor could be implemented using locks and condition variables in Go, using channels makes the code shorter and more readable in this example.