Handling Golang Concurrency using Channels : Part 3
The only place I learnt about golang was contributing to open Source projects like Kubernetes and Kubernetes Sigs.
I was only involved with making changes and enhancements of features but never handled issues of concurrency for large scale systems.
Now as part of Distributed systems Course at CMU I have been working on projects that deals with concurrency to a very large extent and if I had to go back and learn about concurrency from the basics I would start with how channels operate.
Well we are part 3 of Golang Concurrency and in this part we are going to run a simple practical example which demonstrates how to use golang concurrency.
The project idea
Perform add and subtract operations on a list of length n concurrently without deadlocks.
We are going to use a list of length 4 and we are going to run two functions add and subtract on a particular index which increments or decrements a random value in that particular index.
Lastly we shall print the result of the list.
First pass without channels
This is the initial code and each part is explained below.
Add: Add a value to an index in the array data
Subtract: Decrement a value to an index in the array data
Read: Read content of the array at that index
Two goroutines are started which adds and subtracts random values at random indexes in the array and lastly we read the list.
The code is written below.
package main
import (
"fmt"
"math/rand"
"time"
)
func generateRandom(a, b int) int {
return rand.Intn(b-a+1) + a
}
func Read(data []int,index int) int {
return data[index]
}
func Add(data []int, index int, value int) int {
data[index] += value
return data[index]
}
func Subtract(data []int, index int, value int) int {
data[index] -= value
return data[index]
}
func main() {
var data = []int{1,2,3,4}
for i:=0;i<=5;i++ {
go func() {
index := generateRandom(0,3)
value := generateRandom(100,400)
Add(data, index,value)
}()
go func() {
index := generateRandom(0,3)
value := generateRandom(100,400)
Subtract(data, index, value)
}()
}
time.Sleep(1*time.Second)
fmt.Println("#####################")
for index, _ := range data {
value := Read(data, index)
fmt.Printf("The value for index %d received is this %d", index, value)
}
time.Sleep(1*time.Second)
}Lets run the code with race flag and see the output.
go run -race main.go
We run into a data race error on doing this so clearly we are trying to share a space of memory or critical section by two operations.
Lets use Channels
So now lets start our channels synchronization implementation.
We will first create two structs just to make our lives easier
type operation struct {
index int
value int
}The operation type will contain the index we wish to change and the value we wish to increment or decrement.
type Action struct {
data []int
readChan chan int
addChan chan operation
subChan chan operation
}A action struct which contains our main array and three channels which are used for reading , adding or subtracting data.
The add and subtract channels are of type operation since we have to work with both index and value.
Now we will run a goroutine which will synchronize our operations.
func (a Action)HandleRequest() {
for {
select {
case index:= <- a.readChan:
value := a.Read(index)
fmt.Printf("The data READ for index %v is %v\n", index, value)
case w := <- a.addChan:
value := a.Add( w.index, w.value)
fmt.Printf("The data ADD for index %v is %v and result %d\n", w.index, w.value, value)
case w := <- a.subChan:
value := a.Subtract( w.index, w.value)
fmt.Printf("The data SUBTRACT for index %v is %v and result %d\n", w.index, w.value, value)
}
}
}The HandleRequest will take in incoming requests and pass them onto the relevant channels.
Furthermore we will convert the add and subtract operations to be methods.
func (a Action)Read(index int) int {
return a.data[index]
}
func (a Action)Add( index int, value int) int {
a.data[index] += value
return a.data[index]
}
func (a Action)Subtract( index int, value int) int {
a.data[index] -= value
return a.data[index]
}Finally we run the HandleRequest goroutine.
go a.HandleRequest()So our entire code looks like this.
package main
import (
"fmt"
"math/rand"
"time"
)
type operation struct {
index int
value int
}
type Action struct {
data []int
readChan chan int
addChan chan operation
subChan chan operation
}
func generateRandom(a, b int) int {
return rand.Intn(b-a+1) + a
}
func (a Action)Read(index int) int {
return a.data[index]
}
func (a Action)Add( index int, value int) int {
a.data[index] += value
return a.data[index]
}
func (a Action)Subtract( index int, value int) int {
a.data[index] -= value
return a.data[index]
}
func main() {
var data = []int{1,2,3,4}
a := Action{
data: data,
readChan : make(chan int),
addChan : make(chan operation),
subChan: make(chan operation),
}
go a.HandleRequest()
for i:=0;i<=5;i++ {
go func() {
index := generateRandom(0,3)
value := generateRandom(100,400)
w := operation {
index: index,
value : value,
}
a.addChan <- w
return
}()
go func() {
index := generateRandom(0,3)
value := generateRandom(100,400)
w := operation {
index: index,
value : value,
}
a.subChan <- w
return
}()
}
time.Sleep(1*time.Second)
fmt.Println("#####################")
for index, _ := range a.data {
a.readChan <- index
}
// allow both the goroutines to complete or finishes too fast
time.Sleep(1*time.Second)
}
func (a Action)HandleRequest() {
for {
select {
case index:= <- a.readChan:
value := a.Read(index)
fmt.Printf("The data READ for index %v is %v\n", index, value)
case w := <- a.addChan:
value := a.Add( w.index, w.value)
fmt.Printf("The data ADD for index %v is %v and result %d\n", w.index, w.value, value)
case w := <- a.subChan:
value := a.Subtract( w.index, w.value)
fmt.Printf("The data SUBTRACT for index %v is %v and result %d\n", w.index, w.value, value)
}
}
}Running the same code with the race flag now produces this output.
So what exactly happens.
When the add goroutine is called the value and index is packed into an operation body and pushed to the add channel.
This is what happens for the subtract case too.
However it is up to the HandleRequest go routine to make the decision which channel operation it will execute first , the main idea being that both Add and Subtract does not execute together.
Part 1 can be found here: https://diptochakrabarty.medium.com/handling-golang-concurrency-using-channels-and-select-part-1-3e760ef4d8ae
Part 2 can be found here: https://medium.com/@diptochakrabarty/handling-concurrency-in-golang-part-2-80c3b4d4a8db
If you like it please consider clapping and subscribing.
