Go Routines and Channels Done Right.

go_concurrency.md

Using Select to get output from multiple channels based on when they are ready:

package main

import (
	"fmt"
	"time"
)

func main() {
	channel1 := make(chan string)
	channel2 := make(chan string)

	go func() {
		for i := 0; i < 5; i++ {
			channel1 <- "I'll print every 100ms"
			time.Sleep(time.Millisecond * 100)
			
		}
	}()

	go func() {
		for i := 0; i < 5; i++ {
			channel2 <- "I'll print every 1s"
			time.Sleep(time.Second * 1)
			
		}
	}()

	for i := 0; i < 5; i++ {
		select {
		case message1 := <-channel1:
			fmt.Println(message1)
		case message2 := <-channel2:
			fmt.Println(message2)
		}
	}
}

Using default along with above example to do something when no channel has anything to send

package main

import (
	"fmt"
	"time"
)

func main() {
	channel1 := make(chan string)
	channel2 := make(chan string)

	go func() {
		for i := 0; i < 5; i++ {
			channel1 <- "I'll print every 100ms"
			time.Sleep(time.Millisecond * 100)
			
		}
	}()

	go func() {
		for i := 0; i < 5; i++ {
			channel2 <- "I'll print every 1s"
			time.Sleep(time.Second * 1)
			
		}
	}()

	for i := 0; i < 5; i++ {
		select {
		case message1 := <-channel1:
			fmt.Println(message1)
		case message2 := <-channel2:
			fmt.Println(message2)
		default:
			fmt.Println("No channel is ready")
		}
	}
}

Using Buffered Channel in Golang: No Deadlock happens in this example because it won't get blocked till capacity is hit

package main
import "fmt"

func main() {
	mychannel := make(chan int, 2)
	mychannel <- 10
	fmt.Println(<-mychannel)
}

WaitGroup

• A WaitGroup blocks a program and waits for a set of goroutines to finish before moving to the next steps of execution.

We can use WaitGroups through the following functions:

.Add(int): This function takes in an integer value which is essentially the number of goroutines which the waitgroup has to wait for. This function must be called before we execute a goroutine.
.Done(): This function is called within the goroutine to signal that the goroutine has successfully executed.
.Wait(): This function blocks the program until all the goroutines specified by Add() have invoked Done() from within.

package main
import ( 
  "fmt"
  "sync"
)

func WelcomeMessage(wg *sync.WaitGroup){
    fmt.Println("Welcome to Educative!")
    wg.Done()
}

func main() {
  var wg sync.WaitGroup

  wg.Add(2)

  go WelcomeMessage(&wg)
  go func(){  
    fmt.Println("Hello World!")
    wg.Done()
  }()

  wg.Wait()

}

Mutex

• To solve race conditions.
• A mutex, or a mutual exclusion prevents other processes from entering a critical section of data while a process occupies it.
• In general, the section of code which is accessing or updating the shared resources is called the critical section. Mutexes provide us with a locking mechanism which enables only one process to access the critical section at a time.

sync.mutex has two methods:

.Lock() : acquires/holds the lock
.Unlock(): releases the lock

package main
import ( "fmt"
        "sync")

func deposit(balance *int,amount int, myMutex *sync.Mutex, myWaitGroup *sync.WaitGroup){
    myMutex.Lock()
    *balance += amount //add amount to balance
    myMutex.Unlock()
    myWaitGroup.Done()
}

func withdraw(balance *int, amount int, myMutex *sync.Mutex, myWaitGroup *sync.WaitGroup){
    myMutex.Lock()
    *balance -= amount //subtract amount from balance
    myMutex.Unlock()
    myWaitGroup.Done()
}

func main() {
  
    balance := 100 
    var myWaitGroup sync.WaitGroup
    var myMutex sync.Mutex

    myWaitGroup.Add(2)
    go deposit(&balance,10, &myMutex, &myWaitGroup) //depositing 10
    withdraw(&balance, 50,&myMutex, &myWaitGroup) //withdrawing 50

    myWaitGroup.Wait()
    fmt.Println(balance) 


}

• RWMutex stands for Reader/Writer mutual exclusion and is essentially the same as Mutex, but it gives the lock to more than one reading process or just a writing process.
• RWMutex provides us with more control over memory.

Generator Pattern

• pattern of concurrency, which is about generators that return channels as returning argument.
• Generators return the next value in a sequence each time they are called. This means that each value is available as an output before the generator computes the next value. Hence, this pattern is used to introduce parallelism in our program.

package main

import "fmt"

func updatePosition(name string) <-chan string { 
	positionChannel := make(chan string)

	go func() {
		for i := 0; ; i++ {
			positionChannel <- fmt.Sprintf("%s %d", name , i)
		}
	}()

	return positionChannel
}

func main() {
	positionChannel1 := updatePosition("Legolas :") 
	positionChannel2 := updatePosition("Gandalf :")
	

	for i := 0; i < 5; i++ {
		fmt.Println(<-positionChannel1)
		fmt.Println(<-positionChannel2)
	}

	fmt.Println("Done with getting updates on positions.")
}

Output:

Legolas : 1
Gandalf : 1
Legolas : 2
Gandalf : 2
Legolas : 3
Gandalf : 3
Legolas : 4
Gandalf : 4
Done with getting updates on positions.

Fan-In, Fan-Out

• Fan-In, Fan-Out techniques which are used to multiplex and demultiplex data in Go.

• These two operations are blocking each other

    fmt.Println(<-positionChannel1)
    fmt.Println(<-positionChannel2)

• They are taking turns not only in printing value on to the console but also in proceeding to the next computation.

package main

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

func updatePosition(name string) <-chan string { 
	positionChannel := make(chan string)

	go func() {
		for i := 0; ; i++ {
			positionChannel <- fmt.Sprintf("%s %d", name , i)
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
		}
	}()

	return positionChannel
}

func fanIn(mychannel1, mychannel2 <-chan string) <-chan string {
	mychannel := make(chan string) 

	go func() { 
		for {
			mychannel <- <-mychannel1 
		}
	}()

	go func() { 
		for {
			mychannel <- <-mychannel2
		}
	}()

	return mychannel
}


func main() {
	positionsChannel := fanIn(updatePosition("Legolas :"), updatePosition("Gandalf :"))
	

	for i := 0; i < 10; i++ {
		fmt.Println(<-positionsChannel)
	}

	fmt.Println("Done with getting updates on positions.")
}

• Output is random now.
• Here we take in two channels as input and specify one channel as the return argument, i.e., mychannel which is the Fan-In channel.
• The only difference comes from the fact that mychannel1 and mychannel2 are communicating with mychannel now instead of directly communicating with the main routine as before.

Sequencing

• A pattern used for sequencing in a program by sending channel over a channel.
• If we don’t want to block the code and introduce sequence to our program instead of randomness

package main

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


type CookInfo struct {
	foodCooked  string
	waitForPartner chan bool 
}



func cookFood(name string) <-chan CookInfo { 
	
	cookChannel := make(chan CookInfo)
	wait := make(chan bool)
	go func() {
		for i := 0; ; i++ {
			cookChannel<- CookInfo{fmt.Sprintf("%s %s", name,"Done") , wait}
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)

			<-wait
		}
	}()

	return cookChannel
}

func fanIn(mychannel1, mychannel2 <-chan CookInfo) <-chan CookInfo {
	mychannel := make(chan CookInfo)

	go func() { 
		for {
			mychannel <- <-mychannel1 
		}
	}()

	go func() { 
		for {
			mychannel <- <-mychannel2
		}
	}()

	return mychannel
}


func main() {
	gameChannel := fanIn(cookFood("Player 1 : "), cookFood("Player 2 :"))
	

	for round := 0; round < 3; round++ {
		food1 := <-gameChannel
		fmt.Println(food1.foodCooked)

		food2 := <-gameChannel
		fmt.Println(food2.foodCooked)

		food1.waitForPartner <- true
		food2.waitForPartner <- true

		fmt.Printf("Done with round %d\n", round+1)
	}

	fmt.Println("Done with the competition")
}

MergeSort in Golang

package main
import "fmt"

func Merge(left, right [] int) [] int{
  merged := make([] int, 0, len(left) + len(right))
  for len(left) > 0 || len(right) > 0{
    if len(left) == 0 {
      return append(merged,right...)
    }else if len(right) == 0 {
      return append(merged,left...)
    }else if left[0] < right[0] {
      merged = append(merged, left[0])
      left = left[1:]
    }else{
      merged = append(merged, right [0])
      right = right[1:]
    }
  }
  return merged
}

func MergeSort(data [] int) [] int {
  if len(data) <= 1 {
    return data
  }
  done := make(chan bool)
  mid := len(data)/2
  var left [] int
  
  go func(){
    left = MergeSort(data[:mid])
    done <- true
  }()
  right := MergeSort(data[mid:])
  <-done
  return Merge(left,right)

}

func main(){
  data := [] int{9,4,3,6,1,2,10,5,7,8}
  fmt.Printf("%v\n%v\n", data, MergeSort(data))

}

Concurrency is about dealing with lots of things at once.

Parallelism is about doing lots of things at once

Mutex lock once acquired must be released to be able to acquire another inside

package main
import (
  "fmt"
  "sync"
  "time")

func main() {
   myMutex := sync.Mutex{}
   myMutex.Lock()
   go func() {
    myMutex.Lock()
    fmt.Println("I am in the goroutine")
    myMutex.Unlock()
  }()
  fmt.Println("I am in main routine")
  myMutex.Unlock()
  time.Sleep(time.Second*1)
  
}

I am in main routine
I am in the goroutine

So first we acquire the lock on line 9 and then we cannot hold another lock on line 11. The lock is released on line 16 after printing I am in main routine on to the console. Only then, the goroutine is able to acquire the lock on line 11 and print I am in the goroutine before releasing that lock on line 13 and exiting the program.

Runtime Package for setting parallelism or checking the number of cpus being used

package main

import (
    "runtime"
    "fmt"
)

func main() {
    fmt.Printf("GOMAXPROCS is %d\n", runtime.GOMAXPROCS(3))
    fmt.Printf("GOMAXPROCS is %d\n", runtime.GOMAXPROCS(0))
    fmt.Printf("NumCPU is %d\n", runtime.NumCPU())
}

Generates an array of random numbers and prints all the values after doubling them

• Fan-Out

package main

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

func main() {
	var myNumbers [10]int
	for i := 0; i < 10; i++{ 
		rand.Seed(time.Now().UnixNano())
		myNumbers[i]=rand.Intn(50)
	}
	
	mychannelOut := channelGenerator(myNumbers)

	mychannel1 := double(mychannelOut)
	mychannel2 := double(mychannelOut)

	mychannelIn := fanIn(mychannel1, mychannel2)


	for i := 0; i < len(myNumbers); i++ {
		fmt.Println(<-mychannelIn)
	}
}

func channelGenerator(numbers [10]int) <-chan string {
	channel := make(chan string)
	go func() {
		for _, i := range numbers {
			channel  <-  strconv.Itoa(i)
		}
		close(channel)
	}()
	return channel 
}

func double(inputchannel <-chan string) <-chan string {
	channel := make(chan string)
	go func() {
		for i := range inputchannel {
			num, err := strconv.Atoi(i)
			 if err != nil {
      			
  			 }
			 channel <- fmt.Sprintf("%d * 2 = %d", num,num*2)
		}
		close(channel)
	}()
	return channel
}


func fanIn(inputchannel1, inputchannel2 <-chan string) <-chan string {
	channel := make(chan string)
	go func() {
		for {
			select {
			case message1 := <-inputchannel1:  
				channel <- message1
			case message2 := <-inputchannel2:  
				channel <- message2
			}
		}
	}()
	return channel
}

• The Fan-In, Fan-Out techniques can be pretty useful when we have to divide work among jobs and then combine the results from those jobs.

Range and Close

• range function which lets us iterate over elements in different data structures. Using this function, we can range over the items we receive on a channel until it is closed.

package main
import "fmt"

type Money struct{
  amount int
  year int
}

func sendMoney(parent chan Money){

  for i:=0; i<=18; i++ {
    parent <- Money{5000*i,i}  
  }
  close(parent)
}

func main() {
  money := make(chan Money)

  go sendMoney(money)

  for kidMoney:= range money {
    fmt.Printf("Money received by kid in year %d : %d\n", kidMoney.year, kidMoney.amount) 
  }
}

For-Select Loop

• we can combine the for and select statements in order to communicate over channels in Go.

for { //either range over a channel or loop infinitely
  select {
    //handle channel operations
  }
}

• the for-loop below enables us to execute the select statement infinitely. This is pretty handy in situations where we have to keep taking inputs from different channels.

package main
import "fmt"
func getNews(newsChannel chan string){
  NewsArray := [] string {"Roger Federer wins the Wimbledon","Space Exploration has reached new heights","Wandering cat prevents playground accident"}
  for _, news := range NewsArray{
    newsChannel <- news 
  }

  newsChannel <- "Done"
  close(newsChannel)

}

func main() {
  myNewsChannel := make(chan string)

  go getNews(myNewsChannel)
  
  for {
    select{
      case news := <-myNewsChannel:
        fmt.Println(news)
        if news == "Done"{
          return
        }
      default:
    }
    
  }
}

Quit Channel

• A pattern related to quitting from a select statement.

• Normal Approach using return in loop

package main
import ( "fmt"
         "math/rand" 
         "time")

func Race(channel, quit chan string, i int) {
  
  channel <- fmt.Sprintf("Car %d started!", i)  
    for{
      rand.Seed(time.Now().UnixNano())
      time.Sleep(time.Duration(rand.Intn(500)+500) * time.Millisecond)
      quit <- fmt.Sprintf("Car %d reached the finishing line!", i)
    }

}

func main() {
  
  channel := make(chan string)
  quit := make(chan string)
  
  for i:=0; i < 3; i++{
    go Race(channel,quit,i)
  }

  for{
    select{
      case raceUpdates := <-channel:
        fmt.Println(raceUpdates)
      case winnerAnnoucement := <-quit:
        fmt.Println(winnerAnnoucement)  
       return
      
    }
  }
}

• using quit channel
• wg is used to halt the execution once the program has run

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

var wg sync.WaitGroup

func Race(channel, quit chan string, i int) {
  
  channel <- fmt.Sprintf("Car %d started!", i)  
    for{
      rand.Seed(time.Now().UnixNano())
      time.Sleep(time.Duration(rand.Intn(500)+500) * time.Millisecond)
      quit <- fmt.Sprintf("Car %d reached the finishing line!", i)
      fmt.Println(<-quit)
      wg.Done()
    }
}

func main() {

  channel := make(chan string)
  quit := make(chan string)
  wg.Add(1)
  for i:=0; i < 3; i++{
    go Race(channel,quit,i)
  }

  for{
    select{
      case raceUpdates := <-channel:
        fmt.Println(raceUpdates)
      case winnerAnnoucement := <-quit:
        fmt.Println(winnerAnnoucement)
        quit <- "You win!"
        wg.Wait()  
        return
      
    }
  }
}

Timeout using Select Statement

• A pattern which uses the time.After function in a select statement.
• time.after returns a channel that blocks the code for the specified duration. After that duration, the channel delivers the current time but only once.

package main
import ( "fmt"
         "time"
         "math/rand")


func main() {
  dynamite := make(chan string)

  go func(){
    rand.Seed(time.Now().UnixNano())
    time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
    dynamite <- "Dynamite Diffused!"
  }()
    
  for{
    select
    { case s := <-dynamite:
        fmt.Println(s)
        return
      case <-time.After(time.Duration(rand.Intn(500)) * time.Millisecond):
        fmt.Println("Dynamite Explodes!")  
        return      
    }
  }
}

Google Talks about Advanced Concurrency in Go

• YT Video
• Slides