Crafting a concurrent queue in Golang
Tags: go, tutorial, programming, computerscience
Published: 2020-12-06 09:46:09

Queue is one of the most important data structure. Using queues, we can model many computational tasks easily and efficiently. Queues are used everywhere, in Operating systems, they are used to schedule processes, in routers, queues are used as buffers to store packets before processing them. Even cloud native applications composed of miroservices and workers use queues in between them for communication. Queues are extensively used for asynchronous processing.

### What is a Job queue?
If you have ever used node.js, you would be familiar with job queues already. The so called *Event Loop* in node is nothing but a job queue which schedules asynchronous operations to any on the free `uv-threadpool` workers. This type of pattern is common in many systems out there. So here is how queue is used for asynchronous processing, consider the scenario below:
   1. There will be a separate thread or a group of threads waiting for some kind of input.
   2. The main thread produces these inputs ( or tasks )


In this case, there has to be a medium which must act as a glue between input main thread and worker threads running in background. The medium should be designed in such a way that the input which is fed first should get a chance to be processed first, then the next. The input fed into this medium can be consumed by any of the free worker threads. Queue is the best data-structure for this scenario, because of FIFO (First-in, First-out) property, this property ensures ordering, if we make this data-structure somehow shareable between multiple threads we can achieve what we wanted to. 
![Job queue structure](https://dev-to-uploads.s3.amazonaws.com/i/qxj0blbkioigv2psbani.png)
This is how a job queue looks like, the main thread places input on to the queue and continues its operation without waiting for the output. Since the workers wait for input, any one of the worker pops the input from the queue and and processes it. Any of the thread which is free can pop the input. Since the thread which is doing the work will have no time to pop the next input from the queue, only the free thread will get a chance to pop from the queue and becomes busy. If no thread is free, the inputs will still be there in the queue following the right order.

#### What are the advantages?
Imagine a scenario where main thread did all the work, it would work much slower because the main thread cannot process next job until it finishes the current one. A single thread cannot run on multiple cores at once, that is the main reason it runs sequentially. But on a machine with multiple cores, this is not a good approach, because all other cores sit ideally without doing any work. Since we used thread pool here, the free threads in our pool runs on separate cores and we are exploiting parallelism, if we have a 8 core machine, we can now process 8 inputs at once in the same time we took to process a single input. The main thread can now concentrate on fetching new inputs rather than processing them, because putting an input on the queue doesn't take much time.

### Concurrent queue
Every advantage comes with a disadvantage as well. In the case we discussed above, we cannot use a normal queue, because it is not thread-safe. Since we have multiple threads operating on the same queue, synchronization problems will be obviously there, which we need to take care of. A concurrent-queue is a special queue designed to handle such issues. It allows multiple threads to operate on the same queue without any synchronization problems. In this blog we will be designing one such concurrent queue that can be used by multiple goroutines. The queue we design will have following properties:
   1. Can hold any type of data, i.e the queue is a generic implementation (`interface` type in golang)
   2. Expandable in size, it grows and shrinks as required. (i.e no pre-allocated memory)
   3. Thread-safe by nature, no synchronization issues.

#### Let's create a normal queue first
Synchronization issues has nothing to do with how data is stored, we only have to control how data is accessed. This allows us to implement a normal queue first and use it as a storage backend for the concurrent queue. 

Here is how we are going to design the queue:
   1. We will be using a doubly-linked list.
   2. The queue will have push and pop functions.
   3. We insert data at the head of the list and pop from the tail. This is how `push` and `pop` will be implemented.
   4. We make the queue to grow only until `maxSize`. So we keep track of current size every time an enqueue or dequeue occurs and take decisions based on the current size. Here is an implementation of the queue using doubly linked list.

```go
//Node storage of queue data
type Node struct {
	data interface{}
	prev *Node
	next *Node
}

//QueueBackend Backend storage of the queue, a double linked list
type QueueBackend struct {
	//Pointers to root and end
	head *Node
	tail *Node

	//keep track of current size
	size    uint32
	maxSize uint32
}

func (queue *QueueBackend) createNode(data interface{}) *Node {
	node := Node{}
	node.data = data
	node.next = nil
	node.prev = nil

	return &node
}

func (queue *QueueBackend) put(data interface{}) error {
	if queue.size >= queue.maxSize {
		err := errors.New("Queue full")
		return err
	}

	if queue.size == 0 {
		//new root node
		node := queue.createNode(data)
		queue.head = node
		queue.tail = node

		queue.size++

		return nil
	}

	//queue non-empty append to head
	currentHead := queue.head
	newHead := queue.createNode(data)
	newHead.next = currentHead
	currentHead.prev = newHead

	queue.head = currentHead
	queue.size++
	return nil
}

func (queue *QueueBackend) pop() (interface{}, error) {
	if queue.size == 0 {
		err := errors.New("Queue empty")
		return nil, err
	}

	currentEnd := queue.tail
	newEnd := currentEnd.prev

	if newEnd != nil {
		newEnd.next = nil
	}

	queue.size--
	if queue.size == 0 {
		queue.head = nil
		queue.tail = nil
	}

	return currentEnd.data, nil
}

func (queue *QueueBackend) isEmpty() bool {
	return queue.size == 0
}

func (queue *QueueBackend) isFull() bool {
	return queue.size >= queue.maxSize
}

```
We have created a `QueueBackend`, this is just a storage layer on top of which synchronization primitives are implemented. The code above is simple and anyone with basic data structures knowledge can understand it. Next we are going to implement synchronization primitives. 

### Adding concurrency support
We are going to create a `ConcurrentQueue` type which uses `QueueBackend` type internally. The `ConcurrentQueue` type is defined as shown below:

```go
//ConcurrentQueue concurrent queue
type ConcurrentQueue struct {
	//mutex lock
	lock *sync.Mutex

	//empty and full locks
	notEmpty *sync.Cond
	notFull  *sync.Cond

	//queue storage backend
	backend *QueueBackend
}
```
Here, the `backend` is a pointer variable which holds the address of the queue storage layer we created before. Now let us see what other members actually mean.
 1. `lock` : This is a member pointer of type `sync.Mutex`, a mutex is a synchronization primitive, mutex simply means a `lock`, lock can be either `locked` or `unclocked`. If the mutex is locked, other threads will wait until the lock is unlocked. This ensures only one thread can access the queue at a time. Here is how a thread will operate on the queue:
   a. The thread first locks the mutex, this way it gains exclusive control over the queue by avoiding other threads from accessing the queue at the same time.
   b. It then calls `euqueue` or `dequeue` over the queue and once that is done, it unlocks the mutex, allowing other waiting threads to get access to the queue.

This is more than enough to provide a basic synchronization support for the queue, but we actually need more. Once a thread gains control over the queue, it pops the data if present, if no data is present, it has to wait. One way to implement this wait operation is to use a while loop and simply keep checking the queue for data every time. This is not so cool because it runs in an infinite loop taking all the CPU cycles (the core usage goes to 100%). We need some mechanism that allows us to suspend the thread until data becomes available, we can do so by using `conditional locks`. We have defined two such locks, `notEmpty` and `notFull`. `notEmpty` means that, the queue is not empty and has some data, so anyone can pop. `notFull` means that, the queue is not full and the main thread can still push the data into it. If queue is not in `notEmpty` state, the consumer (worker threads) will wait. Similarly, if `notFull` condition is not met, the producer (main thread) will wait for queue to become empty. Now let us see how `euqueue` is implemented: 

```go
func (c *ConcurrentQueue) enqueue(data interface{}) error {
	c.lock.Lock()

	for c.backend.isFull() {
		//wait for empty
		c.notFull.Wait()
	}

	//insert
	err := c.backend.put(data)

	//signal notEmpty
	c.notEmpty.Signal()

	c.lock.Unlock()

	return err
}
``` 
The equeue function checks if anyone has already locked the queue, if not, it locks the queue to gain control, next it checks if the queue is full, if yes, it waits until it is not full. (in the line `c.notFull.Wait()`), After waiting, a new input is pushed into the queue, calling `push` over the queue backend. Now the queue is not empty, because there is a new input, `c.notEmpty.Signal()` is called, this notifies the threads waiting over `notEmpty` that a new entry has been inserted and the queue is no more empty, finally the lock is released, allowing other threads to operate on the queue.

Now, let us see how `dequeue` is implemented at the consumer's side:

```go

func (c *ConcurrentQueue) dequeue() (interface{}, error) {
	c.lock.Lock()

	for c.backend.isEmpty() {
		c.notEmpty.Wait()
	}

	data, err := c.backend.pop()

	//signal notFull
	c.notFull.Signal()

	c.lock.Unlock()

	return data, err
}
```
The consumer tries to access the queue for reading, if it is locked, it waits for the thread which is already accessing the queue to release the lock. Next, it checks if the queue is empty, if yes, it waits over `notEmpty` mutex. (in the line `c.notEmpty.Wait()`). Once the queue is not empty (remember `c.notEmpty.Signal()` in enqueue), it pops the entry and notifies the producer to push new input, since there is a space in the queue now. This is done by calling `c.notFull.Signal()`). 

That is it, we have implemented basic `enqueue` and `dequeue` methods with synchronization support. We also implement a miscellaneous method called `getSize` which simply returns the size, even this is done with lock because it might return the wrong value if another thread push/pops the data at the same time.

```go
func (c *ConcurrentQueue) getSize() uint32 {
	c.lock.Lock()
	size := c.backend.size
	c.lock.Unlock()

	return size
}
```

To wrap this entire thing, we create a method called `NewConcurrentQueue` which simply creates an instance of `ConcurrentQueue` type and initializes it with proper values. Here we also initialize all the necessary mutexes. Look at it's code below:
```go
//NewConcurrentQueue Creates a new queue
func NewConcurrentQueue(maxSize uint32) *ConcurrentQueue {
	queue := ConcurrentQueue{}

	//init mutexes
	queue.lock = &sync.Mutex{}
	queue.notFull = sync.NewCond(queue.lock)
	queue.notEmpty = sync.NewCond(queue.lock)

	//init backend
	queue.backend = &QueueBackend{}
	queue.backend.size = 0
	queue.backend.head = nil
	queue.backend.tail = nil

	queue.backend.maxSize = maxSize
	return &queue
}
```
I was able to create a simple HTTP work pool using this queue implementation as a job queue. You can checkout the complete source code here : [httppool](https://github.com/Narasimha1997/httppool)

Thanks for reading this!