Simple Implementation of the Monitor Object Pattern in Go for Easy Concurrency Control

Introduction

Sometimes in a multi-threaded program, there is a need to protect a resource from concurrent access, that is access by multiple threads. One way of doing this is by implementing a Monitor Object.

Implementing this pattern usually involves the use of the locks such as sync.Mutex and condition variables such as sync.Cond

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 in Go

A use case for this pattern would be a buffer, which is shared by a number of producers and consumers. Start by importing the following:

package main

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

Line by line:

1. This short program will need to output status information, so the fmt package is needed.
2. Each producer and each consumer will wait for a number of seconds before moving on, so the math/rand package is needed.
3. This example will be multithreaded, and it will be using things like Mutex and Cond so the sync package is needed.
4. Finally, because producers will need to sleep for a number of seconds, the time package is needed.

Time to define the Buffer struct:

type Buffer struct {
	capacity int
	buffer   []string
	mutex    sync.Mutex
	notEmpty *sync.Cond
	notFull  *sync.Cond
}

Line by line:

1. The buffer has a set capacity.
2. It will contain strings for this example. It would of course be possible to build a generic buffer.
3. The buffer field is the actual storage of the data.
4. Only one thread should be able to access the buffer at a single time, that is why the sync.Mutex is there.
5. The notEmpty condition variable allows consumers to wait when the buffer is empty, and get notified when new items are added.
6. Conversely the notFull allows producers to wait when the buffer is full and be notified when space becomes available.

Why are the two condition variables pointers and the mutex a value type?

1. First of all the sync.NewCond() method returns a pointer.
2. Condition variables maintain a complex internal state, including a list of waiting goroutines.
3. Copying such a variable would sever its connection the these goroutines.
4. In other words because of the relationship with the waiting goroutines, a condition variable needs a persistent identity, and therefore it is implemented as pointer here.
5. sync.Mutex is designed to be used as a value type. It can be embedded directly in structs without the use of pointers, and is even more memory-efficient when embedded directly.
6. Because of this the internal state of a mutex still works when initialized as a part of this struct.

Now we can create the Buffer:

func NewBuffer(capacity int) *Buffer {
	b := &Buffer{
		capacity: capacity,
		buffer:   make([]string, 0, capacity),
	}
	b.notEmpty = sync.NewCond(&b.mutex)
	b.notFull = sync.NewCond(&b.mutex)
	return b
}

Line by line:

1. Start by initializing the capacity and the slice, with zero length and the specified capacity.
2. Next initialize the two condition variables, which are associated with the same mutex. The sync.NewCond() method takes a Locker interface as its parameter, which is implemented by sync.Mutex and returns a condition variable which will use this lock for synchronization.
3. Finally return the created buffer.

The first thing we want to do is add items to this buffer:

func (b *Buffer) Produce(item string) {
	b.mutex.Lock()
	defer b.mutex.Unlock()

	for len(b.buffer) >= b.capacity {
		fmt.Printf("Producer: Buffer is full, waiting for space. Current buffer: %v\n", b.buffer)
		b.notFull.Wait()
	}

	b.buffer = append(b.buffer, item)
	fmt.Printf("Producer: Added %s. Current buffer: %v\n", item, b.buffer)
	b.notEmpty.Signal()
}

Line by line:

1. Start by acquiring an exclusive access to the buffer calling the Lock() method.
2. The following defer statement ensures that no matter how the method exits, the lock is released.
3. Next comes the critical section. If the buffer is full, a message is printed to the console. The call to Wait() call atomically releases the mutex . It then suspends the goroutines until another thread signals that space has become available. When that happens and the goroutine wakes up, the mutex is re-acquired and rechecks the condition to ensure there really is space available.
4. The method now knows there is space available, and the item is appended to the slice.
5. Finally notify any waiting consumer threads that data has become available.

The Consume() method works along the same lines:

func (b *Buffer) Consume() string {
	b.mutex.Lock()
	defer b.mutex.Unlock()

	
	for len(b.buffer) == 0 {
		fmt.Println("Consumer: Buffer is empty, waiting for items.")
		b.notEmpty.Wait()
	}

	item := b.buffer[0]
	b.buffer = b.buffer[1:]
	fmt.Printf("Consumer: Removed %s. Current buffer: %v\n", item, b.buffer)
	b.notFull.Signal()
	return item
}

Line by line:

1. Again, start by acquiring exclusive access to the buffer.
2. Make sure that however this method ends, the lock is released.
3. As long as the buffer is empty loop. The call to the Wait() method atomically releases the lock, and suspends the goroutine. When an item is available the goroutine wakes up, the lock is re-acquired, and the check for items is done again.
4. Now that there are items, the first item is taken, and removed (that is what the b.buffer=b.buffer[1:] does)
5. A message is sent to the screen.
6. Any producers are notified, so that they can add items to the buffer.
7. Finally, the retrieved item is returned.

To test this setup, write a 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)
		time.Sleep(time.Duration(100+rand.Intn(400)) * time.Millisecond)
	}
}

Line by line:

1. The function receives three parameters: the buffer to which the items will be written, the number of items to write, and a sync.WaitGroup which will be used to signal that the go-routine in which this function is executing has ended.
2. The first thing in the function is using the defer statement to make sure the sync.WaitGroup is notified when the function terminates.
3. The loop iterates numItems times, producing items, and adding them to the buffer using the Produce() method, after which it waits a random number of milliseconds.

And also a function to consume these items is needed:

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

	for i := 0; i < numItems; i++ {
		buffer.Consume()
		time.Sleep(time.Duration(200+rand.Intn(400)) * time.Millisecond)
	}
}

Line by line:

1. Like the producer function this function receives the buffer from which the items are received, the number of items to be received, and a sync.WaitGroup which signals this function has ended.
2. The first thing is to make sure the sync.WaitGroup is signalled when this function ends.
3. In the loop, Consume() is called, after which the go routine sleeps for a random number of milliseconds.

Testing time

Testing this setup is quite straightforward. Start by setting up the buffer, with a set capacity:

	bufferCapacity := 5
	sharedBuffer := NewBuffer(bufferCapacity)

Next define the number of producers and the number of consumers. By using some integer division the load is more or less evenly distributed:

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

To make sure the app waits for the termination for the different go-routines, there are two sync.WaitGroup variables, one for the producers and one for the consumers.

	var producerWg sync.WaitGroup
	var consumerWg sync.WaitGroup

Next, produces some items. In the loop execute the producerTask in a separate go routine. Note that for each go routine, the wait group is updated:

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

Next do the same for the consumers:

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

Now wait for all the go routines to end:

	producerWg.Wait()
	consumerWg.Wait()

And print out the final result:

	fmt.Println("\nAll tasks finished.")
	fmt.Printf("Final buffer state: %v\n", sharedBuffer.buffer)

The full main() function:

func main() {
	bufferCapacity := 5
	sharedBuffer := NewBuffer(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.")
	fmt.Printf("Final buffer state: %v\n", sharedBuffer.buffer)
}

Try running it, and you will see the buffer growing and shrinking.

Conclusion

Since Go was built for concurrent applications, writing this pattern was quite easy and it turned out to be quite straightforward. Especially using the sync.Cond struct makes signalling between go routines make life quite easy.

A possible enhancement would be to make the Monitor struct more generic.