GoLang Concurrency

In Chapter 5 of our Golang Tutorial, we touched upon ‘Data Structures’. In this chapter, let’s explore ‘Concurrency’ in Golang. Here we go –

Large programs are made up of small programs. For example, a web server handles number of requests made from browser and returns the responses. Every request is like a small programs that are handled.

It is always best to run the small components at the same time (like web server handles the multiple requests at the same time). So, working on the multiple programs simultaneously is known as concurrency.

main.go
package main
import “fmt”
func main(){
show()
display()
}
func show(){
for i:=0;i<100;i++{
fmt.Println(“Show:”,i)
}
}
func display(){
for i:=0;i<100;i++{
fmt.Println(“Display:”,i)
}
}

If you observe in above program, the functions are executed sequentially i.e. one after another. Function Display( ) will wait till complete execution of Show( ). That will increase waiting time and could affect the performance of program.

For this, Go has great support and enhancement for concurrency using goroutines and channels.

Goroutines

Goroutine is like thread concept in java which is capable of running with the multiple independent functions. To make function as a goroutine, you just need to add go keyword before the function.

Every program contains one go routine with func main( ). Now in below programs, we have declared two more goroutines.

main.go
package main
import "fmt"
func main() {
go show()
go display()
}
func show() {
for i := 0; i < 100; i++ {
fmt.Println("show:", i)
}
}
func display() {
for i := 0; i < 100; i++ {
fmt.Println("display:", i)
}
}

In the above program. show() & display() will run independently as goroutine and give fast output but you can’t predict what output would come because they are working independently and it’ll be a mixed output.

Goroutines are lightweight and we can create thousands of them (any number). Goroutines has their own private stack and registers like thread and will execute from that stack only. If the main goroutine exits, the program will also exit.

WaitGroup

WaitGroup is the good concept in goroutines that will wait for other goroutines to finish their execution. Sometimes, for executing one activity, other activities need to complete.

As WaitGroup waits for no. of Goroutines to complete their execution, the main goroutine calls Add to set no. of goroutines to wait for. Then each goroutine runs and calls Done when finished their execution. At same time it will call Wait to block and wait until all goroutines finish their execution.

To use sync.WaitGroup :

• Create instance of sync.WaitGroup → var wg sync.WaitGroup
• Call Add(n) where n is no of goroutines to wait for → wg.Add(1)
• Execute defer Wg.Done( ) in each goroutine to indicate that goroutine is finished
• executing to the WaitGroup.
• Call wg.Wait( ) where we want to block

main.go
package main
import (
"fmt"
"sync"
)
var w sync.WaitGroup
func show(n int) {
for i := 0; i < n; i++ {
fmt.Println("fun show:", i)
}
defer w.Done()
}
func display(n int) {
for i := 0; i < n; i++ {
fmt.Println("fun display:", i)
}
defer w.Done()
}
func main() {
fmt.Println("Calling with sync WaitGroup")
w.Add(2)
go show(50)
go display(50)
w.Wait()
for i := 1; i <= 10; i++ {
fmt.Println("For", i)
}
}

In the above program, if we observe the show( ) and display( ) are goroutines which are added to WaitGroup that means main goroutine function have to wait till completion of this goroutines. When this function calls Defer.Done( ) it, indicates they are done with their execution and main goroutine also have to call Wait( ) on that waitgroup so that it will keep itself blocked.

Concurrency vs Parallelism

Concurrency is the composition of independently executing processes, while parallelism is the simultaneous execution of (possibly related) computations. Concurrency is dealing with lots of things at once. Parallelism is about doing things at once.

For parallelism , we just need to runtime.GOMAXPROCS(runtime.NumCPU()) in func init( ). init() is a function which is used to define some initialization program. Basically init () runs first before execution of main(). In init( ) we can provide some setup initialization logic.

Race Condition

A race condition occurs when two or more routines try to access the resource like variable or data structure and attempt to read or write the resources without regard to other routines. So it will create tremendous problems.

So Golang tooling introduced race detector. Race detector is code that is built in your program during the build process. Once your program starts, it will start to detect race condition. It is a really superb tool and does a great job.

For detecting whether there is a race condition in your program or not, run your program → open your cmd prompt → Go to your src folder of your project and run command → go run -race main.go

It will give the status of whether there is race in your program or not.

main.go
package main
import (
"fmt"
"sync"
)
var w sync.WaitGroup
var cnt int
func increment(s string) {
for i := 1; i <= 10; i++ {
x := cnt
x++
cnt = x
fmt.Println(s, i, "Counter:", cnt)
}
defer w.Done()
}
func main() {
fmt.Println("Starting")
w.Add(2)
go increment("show:") /*Both goroutines accessing increment( ) */
go increment("display:")
w.Wait()
fmt.Println("Final Counter:", cnt)
}

Note: I’ve run this program through command prompt.

Mutex

Mutex stands for Mutual Exclusion. Mutex is used for achieving synchronization in Golang and for accessing data safely for multiple goroutines.
Package sync provides this synchronization primitive but higher-level synchronization is always better with channels.

Go provides mutual exclusion with this method

type Mutex
→ func (m *Mutex) Lock ( )
→ func (m *Mutex) Unlock( )

main.go
package main
import (
"fmt"
"sync"
"time"
)
var w sync.WaitGroup
var cnt int
var mutex sync.Mutex //create mutex type variable
func increment(s string) {
for i := 1; i <= 10; i++ {
time.Sleep(3 * time.Millisecond)
mutex.Lock() //Locking while incrementing
x := cnt
x++
cnt = x
fmt.Println(s, i, "Counter:", cnt)
mutex.Unlock() //Unlocking
}
defer w.Done()
}
func main() {
fmt.Println("Starting")
w.Add(2)
go increment("show:")
go increment("display:")
w.Wait()
fmt.Println("Final Counter:", cnt)
}

Now observe the output with a race detector as there is no any race condition and counter values for display and show are not mixing.

Atomicity

Don’t communicate by sharing memory; share memory by communicating is the main proverb of Golang projects ideas.

Atomicity is like mutex for managing state of a user. From Go 1.4 , there is another library offered by Go for achieving thread safety in sync/atomic and has been providing low-level primitives.

It provides thread safe and lock free way.

main.go
package main
import (
"fmt"
"math/rand"
"sync"
"sync/atomic"
"time"
)
var wg sync.WaitGroup
var counter int64
func main() {
wg.Add(2)
go increment("show:")
go increment("display:")
wg.Wait()
fmt.Println("Final Counter:", counter)
}
func increment(s string) {
for i := 0; i < 20; i++ {
time.Sleep(time.Duration(rand.Intn(3)) * time.Millisecond)
atomic.AddInt64(&counter, 1)
fmt.Println(s, i, "Counter:", atomic.LoadInt64(&counter))
/* access without race */
}
defer wg.Done()
}

Here, counter value is showing atomic and not affected by these two goroutines.

Channel

Channels are like pipes that connect concurrent goroutines and passes the data through it. It is a way of sending and receiving value from one goroutine to another goroutine in FIFO manner.

While working with thread-based programming, the shared variables need to protect as they might behave differently and gives wrong results. Also in threading, we need to place locks, avoid deadlocks and serialization of data.

Channels provide higher-level synchronization as they do not share the data. They allow only one channel to access the data even if we are passing.

In order to create a channel, we need to use the make( ) which we have already seen while creating maps and slices. A channel is just created for passing specific type.

Example –

ch:=make(chan type,buffer_size)

ch:=make(chan int,2) where chan is variable and this will pass only integer goroutines and 1 specifying that our channel has 1 value to pass. This is known as the buffered channel.

main.go
package main
import (
"fmt"
"time"
)
func main() {
c := make(chan int)
go func() {
for i := 0; i < 10; i++ {
fmt.Println("Receiving to Channel", i)
c <- i
}
}()

go func() {
for {
fmt.Println()
fmt.Println("Sending from Channel", <-c)
}
close(c) //close channel
}()

time.Sleep(time.Second)
}

In this program, we have created integer channel that can pass only integer values from one goroutine to another goroutine and the functions are anonymous goroutines.

Here, when channel receives c<-i values it stop still something takes values off from channel. After taking out these values again channel will proceed further and like this way, it will pass the values.

By default, sends and receives block until the other side is ready. This provides guaranteed synchronization without locks and conditions.

Channel_name <- value //sends value to channel

value:=<-channel_name //receives from channel
//Data flows in arrow direction

Channel Internal Structure contains three queues –

• Receiving goroutine queue→ This queue is also linked list without size limit. The receiving channel information is stored in this queue along with goroutine.
• Sending goroutine queue→ This queue is also linked list without size limit. The sending channel information is stored in this queue along with goroutine.
• The value of buffer queue→ This is a circular queue acting as a buffer. We need to specify its capacity while creating a channel. If the values in the buffer reach its capacity then the channel is called full. If values are there then the channel is called empty.

Channel operations are:

• len(ch) → Current no. of values in the buffer
• cap(ch) → buffer capacity of channel
• close(ch) → close the channel when no longer it is needed.Closing nil or already closed channel will give panic error
• Send value to channel by using c<-value. Depending on status of channel, sending g operation may succeed to send a value to the channel, block the sending goroutine, or make the sending goroutine panic.
• Receive (and take out) a value, from the channel by using the form value,ok=<-ch where the second value ok is optional, it reports whether the first received value,v, was sent before the channel was closed. Depending on the status of the channel, a receiving operation may succeed to receive a value from the channel or block the receiving goroutine. A receiving operation will never make the receiving goroutine panic.

For Range On Channel:

For range syntax used from channels. The loop iteratively receives value from channel until it gets closed and no more values stored in a buffer.

In map, slice, array it needs two iteration variable but for channel most, one iteration variable is needed.

for a=range channel { is equivalent to for {
//use v a,ok=<-channel
} if !ok{
break
}
//use v
}

Select – Case operations are also there to perform on a channel.

Channel Rules

Buffered Channels

Channels can be buffer. We need to just provide the length as 2nd argument to make for initializing with size.

ch:=make(chan int,100)

A sender needs to close the channel to indicate that channel is no more going to receive the values.

N-to-1: Many functions writing to the same channel.

main.go
package main
import (
"fmt"
"sync"
)
var w sync.WaitGroup
func main() {
c := make(chan int)
w.Add(2)
go func() {
for i := 0; i < 10; i++ {
c <- i
}
w.Done()
}()

go func() {
for i := 0; i < 10; i++ {
c <- i
}
w.Done()
}()
go func() {
w.Wait()
close(c)
fmt.Println("Channel closed")
}()
for n := range c {
fmt.Println(n)
}
}

Here in this program, two goroutines are writing to the same channel and ‘w’ Waitgroup is shared.

Semaphore: Semaphore is variable that can change depending on programmer defined condition.This variable is then uses a condition to control access to some system resources. (like sending messages by holding flags in certain condition).

main.go
package main
import (
"fmt"
)
func main() {

c := make(chan int)
ok := make(chan bool)
go func() {
for i := 1; i <= 10; i++ {
c <- i
}
ok <- true
}()
go func() {
for i := 11; i <= 20; i++ {
c <- i
}
ok <- true
}()
go func() {
<-ok //wait till ‘true’ comes
<-ok
close(c) /* if you put this code outside of goroutine
output will display blank screen
}()
for n := range c { //receiving from channel
fmt.Println(n)
}
}

In the above example, we have created channel one is int and another one is bool. So like instead of doing waitgroup w.Done() we are putting true on channel c.

1-to-N: One channel writing to many functions

main.go
package main
import (
"fmt"
)
func main() {
nums := 10
c := make(chan int)
ok := make(chan bool)

go func() {
for i := 0; i < 10; i++ {
c <- i
}
close(c)
}()
for i := 0; i < nums; i++ { //writing to 10 functions at a times
go func() {
for n := range c {
fmt.Println(n)
}
ok <- true
}()
}
for i := 0; i < nums; i++ {
<-ok
}
}

Here channel is passing data to 10 channels.

Pass return channels: we can pass channel to functions and also can return the channel.

main.go
package main
import "fmt"
func main() {
c := send()
cSum := receive(c)
for n := range cSum {
fmt.Println(n)
}
}
func send() chan int { //channel as return type
out := make(chan int)
go func() {
for i := 0; i < 10; i++ {
out <- i
}
close(out)
}()
return out
}
func receive(c chan int) chan int { //channel as arg and return type
out := make(chan int)
go func() {
var sum int
for n := range c {
sum += n
}
out <- sum
close(out)
}()
return out
}

Channel Direction

While using channels as function parameters, you can specify the channel meant for use whether to only send or receive of for both. If the channel is not represented by any direction that means the channel is bidirectional i.e. it can send as well as receive values.

package main
import "fmt"
func main() {
c := increment()
cSum := pull(c)
for n := range cSum {
fmt.Println(n)
}
}
func increment() <-chan int { //receive only channel
out := make(chan int)
go func() {
for i := 0; i < 10; i++ {
out <- i
}
close(out)
}()
return out
}
func pull(c <-chan int) <-chan int {
out := make(chan int)
go func() {
var sum int
for n := range c {
sum += n
}
out <- sum
close(out)
}()
return out
}

Well, this was all about ‘Concurrency’. In our next chapter, we will be focusing on ‘Error Handling’ in Golang. Make sure you check it out.