goroutine调度
0.1、索引
https://blog.waterflow.link/articles/
创新互联公司主要从事成都网站制作、成都做网站、外贸营销网站建设、网页设计、企业做网站、公司建网站等业务。立足成都服务曲水,十年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:13518219792
1、进程
一个进程包含可以由任何进程分配的公共资源。这些资源包括但不限于内存地址空间、文件句柄、设备和线程。
一个进程会包含下面一些属性:
- Process ID:进程ID
- Process State:进程状态
- Process Priority:进程优先级
- Program Counter:程序计数器
- General purpose register:通用寄存器
- List of open files:打开的文件列表
- List of open devices:打开的设备列表
- Protection information:保护信息
- List of the child process:子进程列表
- Pending alarms:待定警告
- Signals and signal handlers:信号和信号处理程序
- Accounting information:记账信息
2、线程
线程是轻量级的进程,一个线程将在进程内的所有线程之间共享进程的资源,如代码、数据、全局变量、文件和内存地址空间。但是栈和寄存器不会共享,每个线程都有自己的栈和寄存器
线程的优点:
- 提高系统的吞吐量
- 提高响应能力
- 由于属性更少,上下文切换更快
- 多核CPU的有效利用
- 资源共享(代码、数据、地址空间、文件、全局变量)
3、用户级线程
用户级线程也称为绿色线程,如:C 中的coroutine、Go 中的 goroutine 和 Ruby 中的 Fiber
该进程维护一个内存地址空间,处理文件,以及正在运行的应用程序的设备和线程。操作系统调度程序决定哪些线程将在任何给定的 CPU 上接收时间
因此,与耗时和资源密集型的进程创建相比,在一个进程中创建多个用户线程(goroutine)效率更高。
4、goroutine
在Go中用户级线程被称作Goroutine,在创建goroutine时需要做到:
- 易于创建
- 轻量级
- 并发执行
- 可扩展
- 无限堆栈(最大堆栈大小在 64 位上为 1 GB,在 32 位上为 250 MB。)
- 处理阻塞调用
- 高效 (work stealing)
其中阻塞调用可能是下面一些原因:
- 在channel中收发数据
- 网络IO调用
- 阻塞的系统调用
- 计时器
- 互斥操作(Mutex)
为什么go需要调度goroutine?
Go 使用称为 goroutine 的用户级线程,它比内核级线程更轻且更便宜。 例如,创建一个初始 goroutine 将占用 2KB 的堆栈大小,而内核级线程将占用 8KB 的堆栈大小。 还有,goroutine 比内核线程有更快的创建、销毁和上下文切换,所以 go 调度器 需要退出来调度 goroutine。OS 不能调度用户级线程,OS 只知道内核级线程。 Go 调度器 将 goroutine 多路复用到内核级线程,这些线程将在不同的 CPU 内核上运行
什么时候会调度goroutine?
如果有任何操作应该或将会影响 goroutine 的执行,比如 goroutine 的启动、等待执行和阻塞调用等……
go调度 如何将 goroutine 多路复用到内核线程中?
1、1:1调度(1个线程对应一个goroutine)
- 并行执行(每个线程可以在不同的内核上运行)
- 可以工作但是代价太高
- 内存至少〜32k(用户堆栈和内核堆栈的内存)
- 性能问题(系统调用)
- 没有无限堆栈
2、N:1调度(在单个内核线程上多路复用所有 goroutine)
- 没有并行性(即使有更多 CPU 内核可用,也只能使用单个 CPU 内核)
我们看下下面的例子,只为go分配了1个processer去处理2个goroutine:
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
func main() {
// 分配 1 个逻辑处理器供调度程序使用
runtime.GOMAXPROCS(1)
var wg sync.WaitGroup
wg.Add(2)
fmt.Println("Starting Goroutines")
// 开一个go协程打印字母
go func() {
defer wg.Done()
time.Sleep(time.Second)
// 打印3次字母
for count := 0; count < 3; count++ {
for ch := 'a'; ch < 'a'+26; ch++ {
fmt.Printf("%c ", ch)
}
fmt.Println()
}
}()
// 开一个go协程打印数字
go func() {
defer wg.Done()
// 打印3次数字
for count := 0; count < 3; count++ {
for n := 1; n <= 26; n++ {
fmt.Printf("%d ", n)
}
fmt.Println()
}
}()
// 等待返回
fmt.Println("Waiting To Finish")
wg.Wait()
fmt.Println("\nTerminating Program")
}
看下结果:
go run main.go
Starting Goroutines
Waiting To Finish
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
a b c d e f g h i j k l m n o p q r s t u v w x y z
a b c d e f g h i j k l m n o p q r s t u v w x y z
a b c d e f g h i j k l m n o p q r s t u v w x y z
Terminating Program
可以看到这俩个goroutine是串行执行的,要么先完成第一个goroutine,要么先完成第二个goroutine,并不是并发执行的。
那如何去实现并发执行呢?
我们同样设置runtime.GOMAXPROCS为1,但是在goroutine中我们在不同的时机加入阻塞goroutine的时间函数time.Sleep,我们看下会有什么不同的结果。
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
func main() {
// 分配 1 个逻辑处理器供调度程序使用
runtime.GOMAXPROCS(1)
var wg sync.WaitGroup
wg.Add(2)
fmt.Println("Starting Goroutines")
// 开一个go协程打印字母
go func() {
defer wg.Done()
time.Sleep(time.Second)
// 打印3次字母
for count := 0; count < 3; count++ {
for ch := 'a'; ch < 'a'+26; ch++ {
if count == 0 {
time.Sleep(10 * time.Millisecond)
}
if count == 1 {
time.Sleep(30 * time.Millisecond)
}
if count == 2 {
time.Sleep(50 * time.Millisecond)
}
fmt.Printf("%c ", ch)
}
fmt.Println()
}
}()
// 开一个go协程打印数字
go func() {
defer wg.Done()
// 打印3次数字
for count := 0; count < 3; count++ {
for n := 1; n <= 26; n++ {
if count == 0 {
time.Sleep(20 * time.Millisecond)
}
if count == 1 {
time.Sleep(40 * time.Millisecond)
}
if count == 2 {
time.Sleep(60 * time.Millisecond)
}
fmt.Printf("%d ", n)
}
fmt.Println()
}
}()
// 等待返回
fmt.Println("Waiting To Finish")
wg.Wait()
fmt.Println("\nTerminating Program")
}
看下结果:
go run main.go
Starting Goroutines
Waiting To Finish
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26
1 2 3 4 5 6 7 8 9 10 11 a 12 b c d e 13 f g h i 14 j k l m 15 n o p 16 q r s t 17 u v w x 18 y z
19 a b 20 c 21 d 22 e f 23 g 24 h 25 i j 26
k l 1 m n 2 o p 3 q r 4 s t 5 u v 6 w x 7 y z
8 a 9 b 10 c 11 d 12 e f 13 g 14 h 15 i 16 j 17 k l 18 m 19 n 20 o 21 p 22 q r 23 s 24 t 25 u 26
v w x y z
Terminating Program
通过上面的结果我们可以看到,当goroutine1阻塞时,go调度器会调度goroutine2执行。
我们可以得出:
- 即使我们将 runtime.GOMAXPROCS(1) 设置为 1,程序也在并发运行
- Running 状态的 Goroutine 数量最大为 1,Block Goroutine 可以多于一个,其他所有 Goroutine 都处于 Runnable 状态
3、线程池
- 在需要时创建一个线程,这意味着如果有 goroutine 要运行但所有其他线程都忙,则创建一个线程
- 一旦线程完成其执行而不是销毁重用它
- 这可以更快的创建goroutine,因为我们可以重用线程
- 但是还有更多的内存消耗,性能问题,并且没有无限堆栈。
4、M: N 线程共享运行队列调度(GMP)
- M代表系统线程的数量
- N代表goroutine的数量
- goroutine 的创建成本很低,我们可以完全控制 goroutine 的整个生命周期,因为它是在用户空间中创建的
- 创建一个操作系统线程很昂贵,我们无法控制它,但是使用多个线程我们可以实现并行
- 在这个模型中,多个 goroutine 被多路复用到内核线程中
我们上面提到过导致goroutine阻塞调用可能是下面一些原因:
- 在channel中收发数据
- 网络IO调用
- 阻塞的系统调用
- 计时器
- 互斥操作(Mutex)
下面看一些goroutine阻塞的例子:
package main
import (
"time"
"fmt"
"sync"
"os"
"net/http"
"io/ioutil"
)
// 全局变量
var worker int
func writeToFile(wg *sync.WaitGroup,){
defer wg.Done()
file, _ := os.OpenFile("file.txt", os.O_RDWR|os.O_CREATE, 0755) // 系统调用阻塞
resp, _ := http.Get("https://blog.waterflow.link/articles/") // 网络IO阻塞
body, _ := ioutil.ReadAll(resp.Body) // 系统调用阻塞
file.WriteString(string(body))
}
func workerCount(wg *sync.WaitGroup, m *sync.Mutex, ch chan string) {
// Lock() 给共享资源上锁
// 独占访问状态,
// 增加worker的值,
// Unlock() 释放锁
m.Lock() // Mutex阻塞
worker = worker + 1
ch <- fmt.Sprintf("Worker %d is ready",worker)
m.Unlock()
// 返回, 通知WaitGroup完成
wg.Done()
}
func printWorker(wg *sync.WaitGroup, done chan bool, ch chan string){
for i:=0;i<100;i++{
fmt.Println(<-ch) // Channel阻塞
}
wg.Done()
done <-true
}
func main() {
ch :=make(chan string)
done :=make(chan bool)
var mu sync.Mutex
var wg sync.WaitGroup
for i:=1;i<=100;i++{
wg.Add(1)
go workerCount(&wg,&mu,ch)
}
wg.Add(2)
go writeToFile(&wg)
go printWorker(&wg,done,ch)
wg.Wait()
<-done // Channel阻塞
<-time.After(1*time.Second) // Timer阻塞
close(ch)
close(done)
}
下面我们看看go调度器在上面这些例子中是如何工作的:
- 如果一个 goroutine 在通道上被阻塞,则通道有等待队列,所有阻塞的 goroutine 都列在等待队列中,并且很容易跟踪。 在阻塞调用之后,它们将被放入 schedular 的全局运行队列中,OS Thread 将再次按照 FIFO 的顺序选择 goroutine。
- M1,M2,M3尝试从全局G队列中获取G
- M1获取锁并拿到G1,然后释放锁
- M3获取锁拿到G2,然后释放锁
- M2获取锁拿到G3,然后释放锁
- G1在ch1的channel中阻塞,然后添加到ch1的等待队列。导致M1空闲
- M1不能闲着,从全局队列获取锁拿到G4,然后释放锁
- G3阻塞在ch2的channel中,然后被放到ch2的等待队列。导致M2空闲
- M2获取锁拿到G5,然后释放锁
- 此时G3在ch2结束阻塞,被放到全局队列尾部等待执行
- G1在ch1结束阻塞,被放到全局队列尾部等待执行
- G4,G5,G2执行完成
- M1,M2,M3重复步骤1-4
-
互斥锁、定时器和网络 IO 使用相同的机制
-
如果一个 goroutine 在系统调用中被阻塞,那么情况就不同了,因为我们不知道内核空间发生了什么。 通道是在用户空间中创建的,因此我们可以完全控制它们,但在系统调用的情况下,我们没法控制它们。
-
阻塞系统调用不仅会阻塞 goroutine 还会阻塞内核线程。
-
假设一个 goroutine 被安排在一个内核线程上的系统调用,当一个内核线程完成执行时,它将唤醒另一个内核线程(线程重用),该线程将拾取另一个 goroutine 并开始执行它。 这是一个理想的场景,但在实际情况下,我们不知道系统调用将花费多少时间,因此我们不能依赖内核线程来唤醒另一个线程,我们需要一些代码级逻辑来决定何时 在系统调用的情况下唤醒另一个线程。 这个逻辑在 golang 中实现为 runtime·entersyscall()和 runtime·exitsyscall()。 这意味着内核线程的数量可以超过核心的数量。
-
当对内核进行系统调用时,它有两个关键点,一个是进入时机,另一个是退出时机。
- M1,M2试着从全局队列拿G
- M1获取锁并拿到G1,然后释放锁
- M2获取锁并拿到G2,然后释放锁
- M2阻塞在系统调用,没有可用的内核线程,所以go调度器创建一个新的线程M3
- M3获取锁并拿到G3,然后释放锁
- 此时M2结束阻塞状态,重新把G2放到全局队列(G2由阻塞变为可执行状态)。M2虽然是空闲状态,但是go调度器不会销毁它,而是自旋发现新的可执行的goroutine。
- G1,G3执行结束
- M1,M3重复步骤1-3
操作系统可以支持多少内核线程?
在 Linux 内核中,此参数在文件 /proc/sys/kernel/threads-max 中定义,该文件用于特定内核。
sh:~$ cat /proc/sys/kernel/threads-max
这里输出表示内核最多可以执行个线程
每个 Go 程序可以支持多少个 goroutine?
调度中没有内置对 goroutine 数量的限制。
每个 GO程序 可以支持多少个内核线程?
默认情况下,运行时将每个程序限制为最多 10,000 个线程。可以通过调用 runtime/debug 包中的 SetMaxThreads 函数来更改此值。
总结:
- 内核线程数可以多于内核数
- 轻量级 goroutine
- 处理 IO 和系统调用
- goroutine并行执行
- 不可扩展(所有内核级线程都尝试使用互斥锁访问全局运行队列。因此,由于竞争,这不容易扩展)
5、M:N 线程分布式运行队列调度器
为了解决每个线程同时尝试访问互斥锁的可扩展问题,维护每个线程的本地运行队列
- 每个线程状态(本地运行队列)
- 仍然有一个全局运行队列
- M1,M2,M3,M4扫描本地可运行队列
- M1,M2,M3,M4从各自的本地队列取出G4,G6,G1,G3
从上面的动图可以看到:
- 从本地队列拿G是不需要加锁的
- 可运行 goroutine 的全局队列需要锁
结论:
- 轻量级 goroutine
- 处理 IO 和 SystemCalls
- goroutine 并行执行
- 可扩展
- 高效
如果线程数大于内核数,那么会有什么问题呢?
在分布式运行队列调度中,我们知道每个线程都有自己的本地运行队列,其中包含有关接下来将执行哪个 goroutine 的信息。 同样由于系统调用,线程数会增加,并且大多数时候它们的本地运行队列是空的。 因此,如果线程数大于核心数,则每个线程必须扫描所有线程本地运行队列,并且大部分时间它们是空的,所以如果线程过多,这个过程是耗时的并且解决方案 效率不高,因此我们需要将线程扫描限制为使用 M:P:N 线程模型求解的常数。
6、M:P: N 线程
- P 代表处理器,它是运行 go 代码所需的资源。 处理器结构详细信息 https://github.com/golang/go/blob/63e129ba1c458db23f0752d106ed088a2cf/src/runtime/runtime2.go#L601
- M 代表工作线程或机器。 机器线程结构详细信息 https://github.com/golang/go/blob/63e129ba1c458db23f0752d106ed088a2cf/src/runtime/runtime2.go#L519
- G 代表 goroutine。 Goroutine 结构细节 https://github.com/golang/go/blob/63e129ba1c458db23f0752d106ed088a2cf/src/runtime/runtime2.go#L407
- 通常,P的数量与逻辑处理器的数量相同
- 逻辑处理器与物理处理器不同(比如我的mac逻辑处理器是8,无力处理器是4)
- 在启动main goroutine之前创建P
如何检查逻辑处理器的数量?
package main
import (
"fmt"
"runtime"
)
func main() {
fmt.Println(runtime.NumCPU())
}
分布式 M:P:N 调度例子
- M1,M2各自扫描P1,P2的队列
- M1,M2从各自的P1,P2中取出G3,G1执行
在系统调用期间执行P的切换
- M1,M2各自扫描P1,P2的队列
- M1,M2从各自的P1,P2中取出G3,G1执行
- G1即将进入系统调用,所以在这之前G1会唤醒另一个线程M3,并将P2切换到M3
- M3扫描P2并取出G2运行
- 一旦G1变为非阻塞,它将被推送到全局队列等待运行
在work-stealing期间,只需要扫描固定数量的队列,因为逻辑处理器的数量是有限的。
如何选择下一个要运行的 goroutine ?
Go 调度器 将按以下顺序检查以选择下一个要执行的 goroutine
-
本地运行队列
-
全局运行队列
- M1,M2,M3各自扫描本地队列P1,P2,P3
- M1,M2,M3各自从P1,P2,P3取出G3,G1,G5
- G5完成,M3扫描本地队列P3发现空,然后扫描全局队列
- M3将从全局队列获取一定数量的G(G6,G7),保存到本地队列P3
- 现在M3从本地队列P3取出G6执行
-
Network poller
- M1,M2,M3各自扫描本地队列P1,P2,P3
- M1,M2,M3各自从P1,P2,P3取出G3,G1,G6
- G6执行完成,M3扫描P3发现是空的,然后扫描全局队列
- 但是全局队列也是空的,然后就检查网络轮询中已就绪的G
- 网络轮询中有一个已就绪的G2,所以M3取出G2并执行
-
Work Stealing
- M1,M2,M3各自扫描本地队列P1,P2,P3
- M1,M2,M3各自从P1,P2,P3取出G3,G1,G6
- G6执行完成,M3扫描P3发现是空的,然后扫描全局队列
- 但是全局队列也是空的,然后就检查网络轮询中已就绪的G
- 但是网络轮询中没有已就绪的G,所以M3随机的从其他P中窃取一半的G到P3
- 如果随机选中的P中没有要执行的G,就会重试4次,从其他P获取
总结:
- 轻量级 goroutine
- 处理 IO 和系统调用
- goroutine 的并行执行
- 可扩展
- 高效/工作窃取
Go 调度的局限性
- FIFO 对局部性原则不利
- 没有 goroutine 优先级的概念(不像 Linux 内核)
- 没有强抢占 -> 没有强公平或延迟保证
- 它没有意识到系统拓扑 -> 没有真实的位置。有一个旧的 NUMA 感知调度程序提案。此外,建议使用 LIFO 队列,这样 CPU 内核缓存中更有可能有数据。
翻译自:
https://mukeshpilaniya.github.io/posts/Go-Schedular/
分享题目:goroutine调度
分享网址:http://scyanting.com/article/dsoicoj.html