0%

《Golang》Context原理

Context应用场景

  1. 我有一个任务,需要分配给很多g共同完成,而且我可以随时取消这个任务,当我取消时,正在工作的所有g都会停下来
  2. 我有一个紧急任务,需要分配给很多g共同完成,而且我可以随时取消这个任务,当我取消时,正在工作的所有g都会停下来,而且我规定下面的g超过多久时间没有完成自己的任务就不要做了
  3. 跟TLS线程本地存储类似的功能

派遣任务的g通过Context.Cancel()取消任务,接受任务的g通过Context.Done()接收取消或者超时的消息

如果派遣任务的g没有Cancel就退出了(进程没有退出,进程退出后操作系统会回收所有内存,不存在泄漏),而接受任务的g任务还没做完,那么就出现了g泄漏以及内存泄漏(g如果做的是一个持续性的任务,那么这个g就变成僵尸g,造成g泄漏,而这个g持有的内存也不会释放给mheap,自然也不会释放给操作系统,造成内存泄漏)

TLS线程本地存储由内核实现的,存放在线程的结构体TIB中,通过系统调用arch_prctl可以设置TIB,同时设置了TLS

TLS是线程的私有存储空间,实现线程中内部共享而外部无法访问的功能

其实通过参数的传递也可以实现TLS的功能,如下:

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
27
28
29
30
31
32
33
34
package main

import (
"fmt"
"time"
)

func test(tls map[string]string) {
fmt.Println(tls["a"])
tls["b"] = "2"
}

func test2(tls map[string]string) {
fmt.Println(tls["aa"])
tls["b"] = "2"
}

func main() {
go func() {
tls := make(map[string]string, 0)
tls["a"] = "1"
test(tls)
test2(tls)
}()

go func() {
tls := make(map[string]string, 0)
tls["aa"] = "11"
test(tls)
test2(tls)
}()

time.Sleep(3 * time.Second)
}

Golang中对于Goroutine并没有GLS这样一个实现,但是可以通过valueCtx实现差不多的效果(其实就是参数传递),严格讲并没有实现TLS的功能(线程之间互相无法访问各自的TLS)

源码浅析

源码中实现了多种Context,下面分别介绍

emptyCtx

啥也没有,不能被取消、不能传递值、没有超时机制

通过Background()或者TODO()方法获得这种Context

Background()TODO()都是单例

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
type emptyCtx int

func (*emptyCtx) Deadline() (deadline time.Time, ok bool) {
return
}

func (*emptyCtx) Done() <-chan struct{} {
return nil
}

func (*emptyCtx) Err() error {
return nil
}

func (*emptyCtx) Value(key interface{}) interface{} {
return nil
}

func (e *emptyCtx) String() string {
switch e {
case background:
return "context.Background"
case todo:
return "context.TODO"
}
return "unknown empty Context"
}

var (
background = new(emptyCtx)
todo = new(emptyCtx)
)

func Background() Context {
return background
}

func TODO() Context {
return todo
}

cancelCtx

是一种可以取消的Context,通过WithCancel()方法创建

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
type CancelFunc func()

// WithCancel returns a copy of parent with a new Done channel. The returned
// context's Done channel is closed when the returned cancel function is called
// or when the parent context's Done channel is closed, whichever happens first.
//
// Canceling this context releases resources associated with it, so code should
// call cancel as soon as the operations running in this Context complete.
func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
c := newCancelCtx(parent)
propagateCancel(parent, &c)
return &c, func() { c.cancel(true, Canceled) }
}

// newCancelCtx returns an initialized cancelCtx.
func newCancelCtx(parent Context) cancelCtx {
return cancelCtx{Context: parent}
}

// A cancelCtx can be canceled. When canceled, it also cancels any children
// that implement canceler.
type cancelCtx struct {
Context

mu sync.Mutex // protects following fields
done chan struct{} // created lazily, closed by first cancel call
children map[canceler]struct{} // set to nil by the first cancel call
err error // set to non-nil by the first cancel call
}

func (c *cancelCtx) Value(key interface{}) interface{} {
if key == &cancelCtxKey {
return c
}
return c.Context.Value(key) // 从父ctx中获取值
}

func (c *cancelCtx) Done() <-chan struct{} {
c.mu.Lock()
if c.done == nil { // done没有的话就初始化
c.done = make(chan struct{})
}
d := c.done
c.mu.Unlock()
return d
}

func (c *cancelCtx) Err() error {
c.mu.Lock()
err := c.err
c.mu.Unlock()
return err
}

// cancel closes c.done, cancels each of c's children, and, if
// removeFromParent is true, removes c from its parent's children.
func (c *cancelCtx) cancel(removeFromParent bool, err error) {
if err == nil {
panic("context: internal error: missing cancel error")
}
c.mu.Lock() // cancel函数可能会被多个g调用,比如多个g做任务a,只要一个完成了就取消任务b。所以这里加锁
if c.err != nil { // 如果ctx已经出错(这个错是已经取消的错)了,也就是如果ctx已经取消了,则返回
c.mu.Unlock()
return // already canceled
}
c.err = err // 命令的发布者在cancel时会传递一个"已经取消"的err
if c.done == nil { // done没有的话(没有任何持有此命令的g使用Done函数接收取消指令),就给一个公用的closedchan,表示chan已经关闭,即已经发送取消指令
c.done = closedchan
} else {
close(c.done) // done存在的话,关闭chan,这样所有使用Done接收取消指令的g都可以收到消息
}
for child := range c.children { // 同时要取消所有子ctx
// NOTE: acquiring the child's lock while holding parent's lock.
child.cancel(false, err) // 保留父子关系
}
c.children = nil // 上面保留关系是为了性能考虑,这里移除所有孩子ctx,提高性能
c.mu.Unlock()

if removeFromParent {
removeChild(c.Context, c)
}
}

timerCtx

timerCtx是一种既可以取消又具有超时机制的Context,只是在cancelCtx基础上加了一个截止时间。由WithDeadline()或者WithTimeout()创建

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) {
if cur, ok := parent.Deadline(); ok && cur.Before(d) { // 如果父的截止时间比新的子ctx的还早,则不管截止时间,直接返回一个cancelCtx
// The current deadline is already sooner than the new one.
return WithCancel(parent)
}
c := &timerCtx{
cancelCtx: newCancelCtx(parent),
deadline: d,
}
propagateCancel(parent, c)
dur := time.Until(d)
if dur <= 0 { // 截止时间已经过了,则直接取消任务
c.cancel(true, DeadlineExceeded) // deadline has already passed
return c, func() { c.cancel(false, Canceled) }
}
c.mu.Lock()
defer c.mu.Unlock()
if c.err == nil {
c.timer = time.AfterFunc(dur, func() { // 启动一个定时器,到了截止时间就取消任务
c.cancel(true, DeadlineExceeded)
})
}
return c, func() { c.cancel(true, Canceled) }
}

type timerCtx struct {
cancelCtx
timer *time.Timer // Under cancelCtx.mu.

deadline time.Time
}

func (c *timerCtx) Deadline() (deadline time.Time, ok bool) {
return c.deadline, true
}

func (c *timerCtx) cancel(removeFromParent bool, err error) {
c.cancelCtx.cancel(false, err)
if removeFromParent {
// Remove this timerCtx from its parent cancelCtx's children.
removeChild(c.cancelCtx.Context, c)
}
c.mu.Lock()
if c.timer != nil {
c.timer.Stop() // 停掉定时器
c.timer = nil
}
c.mu.Unlock()
}

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
return WithDeadline(parent, time.Now().Add(timeout))
}

valueCtx

valueCtx是一个可以传值的Context

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
func WithValue(parent Context, key, val interface{}) Context {
if key == nil {
panic("nil key")
}
if !reflectlite.TypeOf(key).Comparable() {
panic("key is not comparable")
}
return &valueCtx{parent, key, val}
}

type valueCtx struct {
Context
key, val interface{}
}

func (c *valueCtx) Value(key interface{}) interface{} {
if c.key == key {
return c.val
}
return c.Context.Value(key)
}

上面说了,valueCtx可以差不多实现TLS的功能

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
package main

import (
"context"
"fmt"
"time"
)

func main() {
c := context.WithValue(context.Background(), "haha", "hehe")
fmt.Println(c.Value("haha"))
go func(c context.Context) {
c1 := context.WithValue(c, "dede", "111")
fmt.Println(c1.Value("haha")) // 子context可以访问父context的数据,所以严格讲并没有实现TLS的功能
}(c)

time.Sleep(3 * time.Second)
fmt.Println(c.Value("dede")) // 访问不到其他g的数据
}

总结

  1. 如果父cancel,所有子都cancel,且所有子都会被移除,且自己也被其父移除。子cancel不会影响父
  2. 各种Context可以潜逃组合使用,比如一个全能的Context可以这样创建:context.WithTimeout(context.WithValue(context.Background(), “haha”, “xixi”), 10 * time.Second)
  3. Context是并发安全的。通过Mutex锁实现



微信关注我,及时接收最新技术文章