You say you can’t do concurrency?

This is the 12th day of my participation in the November Gwen Challenge. Check out the event details: The last Gwen Challenge 2021

Author: lomtom

Personal website: Lomtom.top,

Personal public account: Bosiao Yuan

Your support is my biggest motivation.

The Go series:

1. Go (1) Basic introduction
2. Go (ii) structure
3. Go (3) Go configuration file
4. Go (4) Redis operation
5. Go (5) Go do not know how to use Gorm?
6. Go (6) come come come, teach you how to call remotely
7. You say you can’t do concurrency?

Do not communicate by sharing memory, but share memory by communicating.

Goroutine

The Go coroutine has a simple model: it is a function that runs concurrently with other Go coroutines in the same address space. It is lightweight and all that is consumed is the allocation of stack space. Stacks are also very small to start with, so they are cheap and only change as heap space is allocated (and freed) as needed.

Go coroutines can be multiplexed on multithreaded operating systems, so if one thread is blocked, say waiting for I/O, then another thread will run.

The design of the Go coroutine hides many of the complexities of thread creation and management.

Adding the go keyword before a function or method can call it in a new GO coroutine. When the call is complete, the Go coroutine also quietly exits. (The effect is somewhat like the ampersand in Unix shells, which makes commands run in the background.)

go myFunc()  // There is no need to wait for myFunc to run simultaneously
Copy the code

Anonymous functions are very convenient to call from a coroutine:

func TestGo(t *testing.T) {
	s := "How are you?"
	go func(a) {
		fmt.Println(s)
	}()
	ss := "The trail department is not good."
	fmt.Println(ss)
}
Copy the code

Result output:

Trail department is not good. How are youCopy the code

In Go, anonymous functions are closures: in effect, the lifetime of a reference variable within a function is now guaranteed to be the same as the active time of the function.

So, it is worth noting that if the main function completes but the methods following go do not, the program also stops. For example, in the method after Go, we let the function sleep for one second so that the main program will exit without printing s

func TestGo(t *testing.T) {
	s := "How are you?"
	go func(a) {
		time.Sleep(1 * time.Second)
		fmt.Println(s)
	}()
	ss := "The trail department is not good."
	fmt.Println(ss)
}
Copy the code

Output:

The trail department is not goodCopy the code

So how to avoid this situation? We’ll talk about that later

These functions are not very useful because they do not implement signal processing as it is done. So, we need channels.

Channel

Why channel? It doesn’t make sense to simply execute functions concurrently. Functions need to exchange data to make sense of executing functions concurrently.

A channel is a special type in the Go language.

A channel, like a conveyor belt or queue, always follows a First In First Out rule, ensuring the order In which data is sent and received. Each channel is a conduit of a concrete type, which requires the element type to be specified when declaring a channel.

Pipes, like maps, also need to make to allocate memory.

The resulting value acts as a reference to the underlying data structure. If an optional integer parameter is provided, it sets the buffer size for the channel.

The default value is zero, indicating unbuffered or synchronized channels.

1. Create (keyword chan)

c := make(chan int)            // Integer unbuffered channel
c := make(chan int.0)         // Integer unbuffered channel
c := make(chan *os.File, 100)  // Buffer channel for pointer to file
Copy the code

2. insert

c <- a
Copy the code

3. read

num := <- c
Copy the code

4. Declare read/write pipes

var c int
Copy the code

5. Declarations only write pipes

var c chan<- int
Copy the code

6. Declare a read-only pipe

var c <-chan int
Copy the code

Example:

func TestGo(t *testing.T) {
	// Create an unbuffered channel of type integer
	c := make(chan int)
	// Execute custom methods; At the end of the method, a signal is emitted on the channel
	go func(a) {
		// doSomething
		for i := 0; i < 5; i++ {
			// Send a signal
			c <- i
		}
		// Close the pipe
		close(c)
	}()
	// doSomething
	// Wait for the custom method to complete, then take the value from channel
	for i := range c{
		fmt.Println(i)
	}
}
Copy the code

The receiver blocks until it receives the data.

1. If the channel is unbuffered, the sender blocks until the receiver receives the value.
2. If the channel is buffered, the sender blocks only before the value is copied to the buffer;
3. If the buffer is full, the sender will wait until one of the receivers retrieves a value.

sync

sync.WaitGroup

Chan can be used in GO for communication, and go also provides waitGroups for synchronization.

The above example is a good example of the synchronization problem with not using WaitGroup

func TestGo(t *testing.T) {
	s := "Hello world."
	go func(a) {
		time.Sleep(1 * time.Second)
		fmt.Println(s)
	}()
	ss := "The trail department is not good."
	fmt.Println(ss)
}
Copy the code

So here’s another example, I’m going to print it ten times in my main function, ten times in myFunc, and in theory, it’s going to print it.

Sleep(time.second *1) in myFunc to simulate the situation where myFunc is not finished but the main program is finished

func TestGo1(t *testing.T) {
	go myFunc()
	for i := 0; i < 10; i++ {
		fmt.Println("Main () test, this is no." + strconv.Itoa(i) + "Time")
	}
	wg.Wait()
}

func myFunc(a) {
	for i := 0; i < 10; i++ {
		fmt.Println("Test () test, this is number one." + strconv.Itoa(i) + "Time")
		time.Sleep(time.Second*1)}}Copy the code

The output is as follows:

Main () test, this is the 0th main() test, this is the first main() test, this is the second main() test, this is the third main() test, this is the fourth main() test, this is the fifth main() test, this is the sixth main() test, This is the seventh main() test, this is the eighth main() test, this is the ninth testtest(alpha) test, this is zeroCopy the code

Sleep(time.second *1) in the main function loop

func TestGo1(t *testing.T) {
	go myFunc()
	for i := 0; i < 10; i++ {
		fmt.Println("Main () test, this is no." + strconv.Itoa(i) + "Time")
		time.Sleep(time.Second*1)
	}
	wg.Wait()
}

func myFunc(a) {
	for i := 0; i < 10; i++ {
		fmt.Println("Test () test, this is number one." + strconv.Itoa(i) + "Time")
		time.Sleep(time.Second*1)}}Copy the code

The output is as follows:

Main () test, this is the 0th testtest() test, this is the 0th main() test, this is the first testtest() test, this is the first main() test, this is the second testtest() Test, this is the second timetest() test, this is the third main() test, this is the fourth main() testtest() Test, this is the fourth testtest() test, this is the fifth main() test, this is the sixth main() testtest() Test, this is the sixth timetest() test, this is the seventh main() test, this is the eighth main() testtest() Test, this is the eighth timetest() test, this is the ninth main() testCopy the code

This worked as we expected, but officially we don’t know how fast or how long the code will execute, so a second that works in theory is a drag in practice.

Then you can use WaitGroup to control.

1. As soon as I start a coroutine, I Add(1), so I start a coroutine
2. Done() is required to delete the coroutine from the wait group
3. Only after all coroutines have been Done() will subsequent code Wait() continue.
4. The number of coroutines of Add must correspond to the number of coroutines of Done; otherwise, deadlock errors are reported

func TestGo1(t *testing.T) {
	var wg sync.WaitGroup
	go myFunc(&wg)
	for i := 0; i < 10; i++ {
		fmt.Println("Main () test, this is no." + strconv.Itoa(i) + "Time")
		//time.Sleep(time.Second*1)
	}
	wg.Wait()
}

func myFunc(wg *sync.WaitGroup) {
	wg.Add(1)
	for i := 0; i < 10; i++ {
		fmt.Println("Test () test, this is number one." + strconv.Itoa(i) + "Time")
		time.Sleep(time.Second*1)
	}
	wg.Done()
}
Copy the code

Output:

Main () test, this is the 0th main() test, this is the first main() test, this is the second main() test, this is the third main() test, this is the fourth main() test, this is the fifth main() testtest() test, this is the 0th main() test, this is the 6th main() test, this is the 7th main() test, this is the 8th main() test, this is the 9th testtest() Test, this is the first testtest() Test, this is the second timetest() Test, this is the third timetest() Test, this is the fourth testtest() Test, this is the fifth timetest() Test, this is the sixth timetest() Test, this is the seventh timetest() Test, this is the eighth timetest() Test, this is the ninth timeCopy the code

sync.Once

In many programming scenarios we need to ensure that certain operations are performed only once in high concurrency scenarios, such as loading configuration files only once, closing channels only once, etc.

The sync package in the Go language provides a solution for a once-only scenario: sync.once.

Sync.once provides only one Do method:

func (o *Once) Do(f func(a)) {
	if atomic.LoadUint32(&o.done) == 0 {
		o.doSlow(f)
	}
}
Copy the code

In use, we just pass in the method that needs to be executed.

var db *gorm.DB

var loadDbConf sync.Once

// GetDb gets the connection
func GetDb(a) *gorm.DB {
	loadDbConf.Do(DbInit)
	return db
}

// DbInit Initializes the database connection pool
func DbInit(a) {
	newLogger := logger.New(
		log.New(os.Stdout, "\r\n", log.LstdFlags), // io writer
		logger.Config{
			SlowThreshold:             time.Second, // Slow SQL threshold
			LogLevel:                  logger.Info, // Log level
			IgnoreRecordNotFoundError: true.// Ignore ErrRecordNotFound error for logger
			Colorful:                  true.// Disable color
		},
	)
	conn, err1 := gorm.Open(mysql.Open(mySQLUri()), &gorm.Config{
		Logger: newLogger,
	})
	iferr1 ! =nil {
		log.Printf("mysql connect get failed.%v", err1)
		return
	}
	db = conn
	log.Printf("mysql init success")}Copy the code

type Once struct {
	done uint32
	m    Mutex
}
Copy the code

Sync.once actually contains a mutex that keeps the Boolean and the data safe, and a Boolean that registers whether the initialization is complete.

This is designed to ensure that initialization is concurrency safe and that initialization is not performed more than once.

sync.Map

The map built into the Go language is not concurrency safe.

var m = make(map[string]int)

func get(key string) int {
	return m[key]
}

func set(key string, value int) {
	m[key] = value
}

func TestGo5(t *testing.T) {
	wg := sync.WaitGroup{}
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(n int) {
			key := strconv.Itoa(n)
			set(key, n)
			fmt.Printf("k=:%v,v:=%v\n", key, get(key))
			wg.Done()
		}(i)
	}
	wg.Wait()
}
Copy the code

The above code may be fine when opening a small number of goroutines, but fatal Error: Concurrent map writes occurs when executing the above code when there are too many.

k=:0,v:=0
k=:2,v:=2
k=:4,v:=4
k=:5,v:=5
k=:6,v:=6
k=:7,v:=7
k=:1,v:=1
k=:8,v:=8
k=:3,v:=3
k=:11,v:=11
k=:10,v:=10
k=:12,v:=12
fatal error: concurrent map writes
k=:13,v:=13
k=:14,v:=14

goroutine 38 [running]:
runtime.throw(0xe6715c, 0x15)
	E:/program/go/src/runtime/panic.go:1117 +0x79 fp=0xc000337ec8 sp=0xc000337e98 pc=0x7ac6f9
runtime.mapassign_faststr(0xd9b800, 0xc00003c2a0, 0xe8046a, 0x2, 0x0)
Copy the code

In situations like this, a map lock is required to ensure concurrency security. The SYNC package for the Go language provides a concurrency security version of Map – sync.map – out of the box.

Out of the box means it can be used directly without using the make function initialization as the built-in map. Sync. Map also has built-in operation methods such as Store, Load, LoadOrStore, Delete, and Range.

var m1 = sync.Map{}

func TestGo6(t *testing.T) {
	wg := sync.WaitGroup{}
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(n int) {
			k := strconv.Itoa(n)
			m1.Store(k,n)
			v, _ := m1.Load(k)
			fmt.Printf("k=:%v,v:=%v\n", k, v)
			wg.Done()
		}(i)
	}
	wg.Wait()
}
Copy the code

parallelization

Sum of 1 minus n

Another application of these designs is parallel computing on multiple CPU cores.

If the computation process can be divided into blocks that can be executed independently, it can send a signal to the channel at the end of each calculation, thus enabling parallel processing.

In the following example, I need to calculate the sum from 1 to n, which is usually a simple loop.

func TestGo(t *testing.T){
	n := 100000
	var s int
	for i := 0; i < n; i++{ s += i } log.Println(s) }Copy the code

However, for the large amount of data, it is obviously not suitable for our code, so we can use parallel computing to achieve.

func TestGo3(t *testing.T){
	t1 := time.Now()
	n := 100
	num := 10
	c := make(chan int,num)
	for i := 0; i < num; i++ {go func(a) {
			start := n / num * i
			end := n / num * (i + 1)
			var s int
			forj := start; j < end; j++ { s += j } c <- s }() }var s int
	for i := 0; i < num; i++ { s += <- c } t2 := time.Since(t1) log.Println(t2) log.Println(s) }Copy the code

You can also borrow WaitGroup

func TestGo3(t *testing.T){
	t1 := time.Now()
	n := 100
	num := 10
	c := make(chan int,num)
	var wg sync.WaitGroup
	for i := 0; i < num; i++ { wg.Add(1)
		go func(a) {
			start := n / num * i
			end := n / num * (i + 1)
			var s int
			forj := start; j < end; j++ { s += j } c <- s wg.Done() }() } wg.Wait()close(c)
	var s int
	for item := range c{
		s += item
	}
	t2 := time.Since(t1)
	log.Println(t2)
	log.Println(s)
}
Copy the code

We start separate processing blocks in the loop, one processing per CPU.

They might finish and end up out of order, but that doesn’t matter; We simply receive after all the Go coroutines have started and count the completion signals in the channel.

Instead of setting the num constant value directly, we can ask the Runtime for a reasonable value.

The runtime.NumCPU function returns the number of cores on the hardware CPU, used like this:

var num = runtime.NumCPU()
Copy the code

Another function to know is runtime.GOMAXPROCS, which returns the user set number of available cpus. By default, runtime.NumCPU is used, but can be command-line environment variables, or this function can be called with positive integer arguments. If we pass 0, it will return the value, and if we respect the user’s allocation of resources,

You should write:

var numCPU = runtime.GOMAXPROCS(0)
Copy the code

Be careful not to confuse concurrency with parallelism: Concurrency is a way of building a program with components that can be executed independently, whereas parallelism is the idea of running calculations in parallel across multiple cpus for efficiency.

Although the concurrency nature of Go makes it easier for some problems to be constructed in parallel, Go is still a concurrent rather than parallel language, and the Go model is not suitable for all parallel problems.

The problem

However, if you run past the previous code, you will see that the later calculation is actually inaccurate.

The value from 1 to 100 should be 4950, and the output is rarely the same, much less 4950.

The problem is in Go’s for loop, which is reused on each iteration, so the I variable is shared across all Go coroutines, which is not what we want.

We need to make sure that I is unique for every Go coroutine.

There are several ways to do this:

1. One: write one more layer of calls

Split the following anonymous methods into a generic method and pass in I as an argument.

func TestGo3(t *testing.T){
	t1 := time.Now()
	n := 100
	num := 10
	c := make(chan int,num)
	var wg sync.WaitGroup
	for i := 0; i < num; i++ { wg.Add(1)
		go myFunc1(n,num,i,c,&wg)
	}
	wg.Wait()
	close(c)
	var s int
	for item := range c{
		s += item
	}
	t2 := time.Since(t1)
	log.Println(t2)
	log.Println(s)
}

func myFunc1(n,num,i int,c chan int,wg *sync.WaitGroup) {
	start := n / num * i
	end := n / num * (i + 1)
	var s int
	forj := start; j < end; j++ { s += j } c <- s wg.Done() }Copy the code

2. The second: closure for passed parameters (with I as the parameter)

func TestGo3(t *testing.T){
	t1 := time.Now()
	n := 100
	num := 10
	c := make(chan int,num)
	var wg sync.WaitGroup
	for i := 0; i < num; i++ { wg.Add(1)
		// pass in I as an argument
		go func(i int) {
			start := n / num * i
			end := n / num * (i + 1)
			var s int
			forj := start; j < end; j++ { s += j } c <- s wg.Done() }(i) } wg.Wait()close(c)
	var s int
	for item := range c{
		s += item
	}
	t2 := time.Since(t1)
	log.Println(t2)
	log.Println(s)
}
Copy the code

3. Third: restatement

func TestGo3(t *testing.T){
	t1 := time.Now()
	n := 100
	num := 10
	c := make(chan int,num)
	var wg sync.WaitGroup
	for i := 0; i < num; i++ { wg.Add(1)
		// restate
		i := i
		go func(a) {
			start := n / num * i
			end := n / num * (i + 1)
			var s int
			forj := start; j < end; j++ { s += j } c <- s wg.Done() }() } wg.Wait()close(c)
	var s int
	for item := range c{
		s += item
	}
	t2 := time.Since(t1)
	log.Println(t2)
	log.Println(s)
}
Copy the code

I := I seems a little strange to write, but it is legal and common to do so in Go.

You get a new version of the variable with the same name to deliberately mask the loop variable locally, making it unique to each Go coroutine.

The performance comparison

Traditional methods and parallel computing:

I don’t care about the out of bounds case, because he still computes every time.

The value of n	traditional	parallel
10000000	3.7146 ms	1.058 ms
100000000	39.9925 ms	7.1645 ms
1000000000	344.0599 ms	49.9023 ms
10000000000	3.4797346 s	501.5713 ms
100000000000	34.5406926 s	4.650136 s

The results are pretty obvious.