Detailed graphic and text explanation of coroutines in GoLang

Coroutine is a lightweight thread implementation in Go language, managed by the Go runtime.

Add the go keyword before a function call, and the call will be executed concurrently in a new goroutine. When the called function returns, this goroutine also ends automatically. It should be noted that if this function has a return value, the return value will be discarded.

Look at the following example first:

func Add(x, y int) {
    z := x + y
    fmt.Println(z)
}

func main() {
    for i:=0; i<10; i++ {
        go Add(i, i)
    }
}

When you execute the above code, you will find that nothing is printed on the screen and the program exits.

For the above example, the main() function starts 10 goroutines and then returns. At this time, the program exits, and the started goroutine that executes Add() has no time to execute. We want the main() function to wait for all goroutines to exit before returning, but how do we know that all goroutines have exited? This leads to the problem of communication between multiple goroutines.

In engineering, there are two most common concurrent communication models: shared memory and messages.

Look at the following example. 10 goroutines share the variable counter. After each goroutine is executed, the counter value is increased by 1. Because 10 goroutines are executed concurrently, we also introduce a lock, as well. It's the lock variable in the code. In the main() function, use a for loop to continuously check the counter value. When its value reaches 10, it means that all goroutines have been executed. At this time, main() returns and the program exits.

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

var counter int = 0

func Count(lock *sync.Mutex) {
    lock.Lock()
    counter++
    fmt.Println("counter =", counter)
    lock.Unlock()
}


func main() {

    lock := &sync.Mutex{}

    for i:=0; i<10; i++ {
        go Count(lock)
    }

    for {
        lock.Lock()

        c := counter

        lock.Unlock()

        runtime.Gosched()　　　　// 出让时间片

        if c >= 10 {
            break
        }
    }
}

The above example uses a lock variable (a type of shared memory) to synchronize the coroutine. In fact, the Go language mainly uses the message mechanism (channel) as the communication model.

channel

The message mechanism considers that each concurrent unit is a self-contained, independent individual, and has its own variables, but in These variables are not shared between different concurrent units. Each concurrent unit has only one input and output, which is messages.

Channel is the communication method between goroutines provided by Go language at the language level. We can use channels to pass messages between multiple goroutines. Channel is an intra-process communication method, so the process of passing objects through channels is consistent with the parameter passing behavior when calling functions. For example, pointers can also be passed.
Channel is type-related. A channel can only pass one type of value. This type needs to be specified when declaring the channel.

The declaration form of channel is:

var chanName chan ElementType

For example, declare a channel passing int type:

var ch chan int

Use the built-in function make() to define a channel :

ch := make(chan int)

In the usage of channel, the most common ones include writing and reading:

// 将一个数据value写入至channel，这会导致阻塞，直到有其他goroutine从这个channel中读取数据
ch <- value

// 从channel中读取数据，如果channel之前没有写入数据，也会导致阻塞，直到channel中被写入数据为止
value := <-ch

By default, the receiving and sending of channel are blocked unless the other end is ready good.

We can also create a buffered channel:

c := make(chan int, 1024)

// 从带缓冲的channel中读数据
for i:=range c {
　　...
}

At this time, create a channel of type int with a size of 1024. Even if there is no reader, the writer can always go to Writing to the channel will not block until the buffer is filled.

You can close channels that are no longer used:

close(ch)

The channel should be closed at the producer's location. If it is closed at the consumer's location, it will easily cause panic;

In a A receive operation (<-ch) on a closed channel always returns immediately, and the return value is a zero value of the corresponding type.

Now use channels to rewrite the above example:

func Count(ch chan int) {
    ch <- 1
    fmt.Println("Counting")
}

func main() {

    chs := make([] chan int, 10)

    for i:=0; i<10; i++ {
        chs[i] = make(chan int)
        go Count(chs[i])
    }

    for _, ch := range(chs) {
        <-ch
    }
}

In this example, an array containing 10 channels is defined, and each channel in the array is assigned to 10 Different goroutines. After each goroutine completes, a data is written to the goroutine. This operation blocks until the channel is read.

After all goroutines are started, data is read from 10 channels in sequence. This operation is also blocked before the corresponding channel writes data. In this way, the channel is used to implement a lock-like function, and ensures that main() returns only after all goroutines are completed.

In addition, when we pass a channel variable to a function, we can limit the operations on this channel in the function by specifying it as a one-way channel variable.

Declaration of one-way channel variables:

var ch1 chan int  　　　　// 普通channel
var ch2 chan <- int 　　 // 只用于写int数据
var ch3 <-chan int 　　 // 只用于读int数据

You can convert a channel into one-way through type conversion:

ch4 := make(chan int)
ch5 := <-chan int(ch4)   // 单向读
ch6 := chan<- int(ch4)  //单向写

The role of one-way channel is somewhat similar to c The const keyword in is used to follow the "principle of least privilege" in the code.

For example, using a one-way read channel in a function:

func Parse(ch <-chan int) {
    for value := range ch {
        fmt.Println("Parsing value", value) 
    }
}

As a native type, channel itself can also be passed through the channel, such as the following streaming processing structure:

type PipeData struct {
    value int
    handler func(int) int
    next chan int
}

func handle(queue chan *PipeData) {
    for data := range queue {
        data.next <- data.handler(data.value)
    }
}

select

In UNIX, the select() function is used to monitor a group of descriptors. This mechanism is often used to implement high-concurrency sockets. Server program. The Go language directly supports the select keyword at the language level, which is used to deal with asynchronous IO issues. The general structure is as follows:

select {
    case <- chan1:
    // 如果chan1成功读到数据
    
    case chan2 <- 1:
    // 如果成功向chan2写入数据

    default:
    // 默认分支
}

select is blocking by default, and will only occur when there is sending or receiving in the monitored channel. When running, when multiple channels are ready, select randomly selects one for execution.

Go语言没有对channel提供直接的超时处理机制，但我们可以利用select来间接实现，例如：

timeout := make(chan bool, 1)

go func() {
    time.Sleep(1e9)
    timeout <- true
}()

switch {
    case <- ch:
    // 从ch中读取到数据

    case <- timeout:
    // 没有从ch中读取到数据，但从timeout中读取到了数据
}

这样使用select就可以避免永久等待的问题，因为程序会在timeout中获取到一个数据后继续执行，而无论对ch的读取是否还处于等待状态。

并发

早期版本的Go编译器并不能很智能的发现和利用多核的优势，即使在我们的代码中创建了多个goroutine，但实际上所有这些goroutine都允许在同一个CPU上，在一个goroutine得到时间片执行的时候其它goroutine都会处于等待状态。

实现下面的代码可以显式指定编译器将goroutine调度到多个CPU上运行。

import "runtime"...
runtime.GOMAXPROCS(4)

PS：runtime包中有几个处理goroutine的函数，

调度

Go调度的几个概念：

M：内核线程；

G：go routine，并发的最小逻辑单元，由程序员创建；

P：处理器，执行G的上下文环境，每个P会维护一个本地的go routine队列；

 除了每个P拥有一个本地的go routine队列外，还存在一个全局的go routine队列。

具体调度原理：

1、P的数量在初始化由GOMAXPROCS决定；

2、我们要做的就是添加G；

3、G的数量超出了M的处理能力，且还有空余P的话，runtime就会自动创建新的M；

4、M拿到P后才能干活，取G的顺序：本地队列>全局队列>其他P的队列，如果所有队列都没有可用的G，M会归还P并进入休眠；

一个G如果发生阻塞等事件会进行阻塞，如下图：

G发生上下文切换条件：

系统调用；

读写channel；

gosched主动放弃，会将G扔进全局队列；

如上图，一个G发生阻塞时，M0让出P，由M1接管其任务队列；当M0执行的阻塞调用返回后，再将G0扔到全局队列，自己则进入睡眠（没有P了无法干活）；

The above is the detailed content of Detailed graphic and text explanation of coroutines in GoLang. For more information, please follow other related articles on the PHP Chinese website!