bling

bling 查看完整档案

填写现居城市  |  填写毕业院校  |  填写所在公司/组织填写个人主网站
编辑
_ | |__ _ _ __ _ | '_ \| | | |/ _` | | |_) | |_| | (_| | |_.__/ \__,_|\__, | |___/ 该用户太懒什么也没留下

个人动态

bling 赞了文章 · 10月29日

深入理解 Go defer

在上一章节 《深入理解 Go panic and recover》 中,我们发现了 defer 与其关联性极大,还是觉得非常有必要深入一下。希望通过本章节大家可以对 defer 关键字有一个深刻的理解,那么我们开始吧。你先等等,请排好队,我们这儿采取后进先出 LIFO 的出站方式...

image

原文地址:深入理解 Go defer

特性

我们简单的过一下 defer 关键字的基础使用,让大家先有一个基础的认知

一、延迟调用

func main() {
    defer log.Println("EDDYCJY.")

    log.Println("end.")
}

输出结果:

$ go run main.go            
2019/05/19 21:15:02 end.
2019/05/19 21:15:02 EDDYCJY.

二、后进先出

func main() {
    for i := 0; i < 6; i++ {
        defer log.Println("EDDYCJY" + strconv.Itoa(i) + ".")
    }


    log.Println("end.")
}

输出结果:

$ go run main.go
2019/05/19 21:19:17 end.
2019/05/19 21:19:17 EDDYCJY5.
2019/05/19 21:19:17 EDDYCJY4.
2019/05/19 21:19:17 EDDYCJY3.
2019/05/19 21:19:17 EDDYCJY2.
2019/05/19 21:19:17 EDDYCJY1.
2019/05/19 21:19:17 EDDYCJY0.

三、运行时间点

func main() {
    func() {
         defer log.Println("defer.EDDYCJY.")
    }()

    log.Println("main.EDDYCJY.")
}

输出结果:

$ go run main.go 
2019/05/22 23:30:27 defer.EDDYCJY.
2019/05/22 23:30:27 main.EDDYCJY.

四、异常处理

func main() {
    defer func() {
        if e := recover(); e != nil {
            log.Println("EDDYCJY.")
        }
    }()

    panic("end.")
}

输出结果:

$ go run main.go 
2019/05/20 22:22:57 EDDYCJY.

源码剖析

$ go tool compile -S main.go 
"".main STEXT size=163 args=0x0 locals=0x40
    ...
    0x0059 00089 (main.go:6)    MOVQ    AX, 16(SP)
    0x005e 00094 (main.go:6)    MOVQ    $1, 24(SP)
    0x0067 00103 (main.go:6)    MOVQ    $1, 32(SP)
    0x0070 00112 (main.go:6)    CALL    runtime.deferproc(SB)
    0x0075 00117 (main.go:6)    TESTL    AX, AX
    0x0077 00119 (main.go:6)    JNE    137
    0x0079 00121 (main.go:7)    XCHGL    AX, AX
    0x007a 00122 (main.go:7)    CALL    runtime.deferreturn(SB)
    0x007f 00127 (main.go:7)    MOVQ    56(SP), BP
    0x0084 00132 (main.go:7)    ADDQ    $64, SP
    0x0088 00136 (main.go:7)    RET
    0x0089 00137 (main.go:6)    XCHGL    AX, AX
    0x008a 00138 (main.go:6)    CALL    runtime.deferreturn(SB)
    0x008f 00143 (main.go:6)    MOVQ    56(SP), BP
    0x0094 00148 (main.go:6)    ADDQ    $64, SP
    0x0098 00152 (main.go:6)    RET
    ...

首先我们需要找到它,找到它实际对应什么执行代码。通过汇编代码,可得知涉及如下方法:

  • runtime.deferproc
  • runtime.deferreturn

很显然是运行时的方法,是对的人。我们继续往下走看看都分别承担了什么行为

数据结构

在开始前我们需要先介绍一下 defer 的基础单元 _defer 结构体,如下:

type _defer struct {
    siz     int32
    started bool
    sp      uintptr // sp at time of defer
    pc      uintptr
    fn      *funcval
    _panic  *_panic // panic that is running defer
    link    *_defer
}

...
type funcval struct {
    fn uintptr
    // variable-size, fn-specific data here
}
  • siz:所有传入参数的总大小
  • started:该 defer 是否已经执行过
  • sp:函数栈指针寄存器,一般指向当前函数栈的栈顶
  • pc:程序计数器,有时称为指令指针(IP),线程利用它来跟踪下一个要执行的指令。在大多数处理器中,PC指向的是下一条指令,而不是当前指令
  • fn:指向传入的函数地址和参数
  • _panic:指向 _panic 链表
  • link:指向 _defer 链表

image

deferproc

func deferproc(siz int32, fn *funcval) {
    ...
    sp := getcallersp()
    argp := uintptr(unsafe.Pointer(&fn)) + unsafe.Sizeof(fn)
    callerpc := getcallerpc()

    d := newdefer(siz)
    ...
    d.fn = fn
    d.pc = callerpc
    d.sp = sp
    switch siz {
    case 0:
        // Do nothing.
    case sys.PtrSize:
        *(*uintptr)(deferArgs(d)) = *(*uintptr)(unsafe.Pointer(argp))
    default:
        memmove(deferArgs(d), unsafe.Pointer(argp), uintptr(siz))
    }

    return0()
}
  • 获取调用 defer 函数的函数栈指针、传入函数的参数具体地址以及PC (程序计数器),也就是下一个要执行的指令。这些相当于是预备参数,便于后续的流转控制
  • 创建一个新的 defer 最小单元 _defer,填入先前准备的参数
  • 调用 memmove 将传入的参数存储到新 _defer (当前使用)中去,便于后续的使用
  • 最后调用 return0 进行返回,这个函数非常重要。能够避免在 deferproc 中又因为返回 return,而诱发 deferreturn 方法的调用。其根本原因是一个停止 panic 的延迟方法会使 deferproc 返回 1,但在机制中如果 deferproc 返回不等于 0,将会总是检查返回值并跳转到函数的末尾。而 return0 返回的就是 0,因此可以防止重复调用

小结

这个函数中会为新的 _defer 设置一些基础属性,并将调用函数的参数集传入。最后通过特殊的返回方法结束函数调用。另外这一块与先前 《深入理解 Go panic and recover》 的处理逻辑有一定关联性,其实就是 gp.sched.ret 返回 0 还是 1 会分流至不同处理方式

newdefer

func newdefer(siz int32) *_defer {
    var d *_defer
    sc := deferclass(uintptr(siz))
    gp := getg()
    if sc < uintptr(len(p{}.deferpool)) {
        pp := gp.m.p.ptr()
        if len(pp.deferpool[sc]) == 0 && sched.deferpool[sc] != nil {
            ...
            lock(&sched.deferlock)
            d := sched.deferpool[sc]
            unlock(&sched.deferlock)
        }
        ...
    }
    if d == nil {
        systemstack(func() {
            total := roundupsize(totaldefersize(uintptr(siz)))
            d = (*_defer)(mallocgc(total, deferType, true))
        })
        ...
    }
    d.siz = siz
    d.link = gp._defer
    gp._defer = d
    return d
}
  • 从池中获取可以使用的 _defer,则复用作为新的基础单元
  • 若在池中没有获取到可用的,则调用 mallocgc 重新申请一个新的
  • 设置 defer 的基础属性,最后修改当前 Goroutine_defer 指向

通过这个方法我们可以注意到两点,如下:

  • deferGoroutine(g) 有直接关系,所以讨论 defer 时基本离不开 g 的关联
  • 新的 defer 总是会在现有的链表中的最前面,也就是 defer 的特性后进先出

小结

这个函数主要承担了获取新的 _defer 的作用,它有可能是从 deferpool 中获取的,也有可能是重新申请的

deferreturn

func deferreturn(arg0 uintptr) {
    gp := getg()
    d := gp._defer
    if d == nil {
        return
    }
    sp := getcallersp()
    if d.sp != sp {
        return
    }

    switch d.siz {
    case 0:
        // Do nothing.
    case sys.PtrSize:
        *(*uintptr)(unsafe.Pointer(&arg0)) = *(*uintptr)(deferArgs(d))
    default:
        memmove(unsafe.Pointer(&arg0), deferArgs(d), uintptr(d.siz))
    }
    fn := d.fn
    d.fn = nil
    gp._defer = d.link
    freedefer(d)
    jmpdefer(fn, uintptr(unsafe.Pointer(&arg0)))
}

如果在一个方法中调用过 defer 关键字,那么编译器将会在结尾处插入 deferreturn 方法的调用。而该方法中主要做了如下事项:

  • 清空当前节点 _defer 被调用的函数调用信息
  • 释放当前节点的 _defer 的存储信息并放回池中(便于复用)
  • 跳转到调用 defer 关键字的调用函数处

在这段代码中,跳转方法 jmpdefer 格外重要。因为它显式的控制了流转,代码如下:

// asm_amd64.s
TEXT runtime·jmpdefer(SB), NOSPLIT, $0-16
    MOVQ    fv+0(FP), DX    // fn
    MOVQ    argp+8(FP), BX    // caller sp
    LEAQ    -8(BX), SP    // caller sp after CALL
    MOVQ    -8(SP), BP    // restore BP as if deferreturn returned (harmless if framepointers not in use)
    SUBQ    $5, (SP)    // return to CALL again
    MOVQ    0(DX), BX
    JMP    BX    // but first run the deferred function

通过源码的分析,我们发现它做了两个很 “奇怪” 又很重要的事,如下:

  • MOVQ -8(SP), BP:-8(BX) 这个位置保存的是 deferreturn 执行完毕后的地址
  • SUBQ $5, (SP):SP 的地址减 5 ,其减掉的长度就恰好是 runtime.deferreturn 的长度

你可能会问,为什么是 5?好吧。翻了半天最后看了一下汇编代码...嗯,相减的确是 5 没毛病,如下:

    0x007a 00122 (main.go:7)    CALL    runtime.deferreturn(SB)
    0x007f 00127 (main.go:7)    MOVQ    56(SP), BP

我们整理一下思绪,照上述逻辑的话,那 deferreturn 就是一个 “递归” 了哦。每次都会重新回到 deferreturn 函数,那它在什么时候才会结束呢,如下:

func deferreturn(arg0 uintptr) {
    gp := getg()
    d := gp._defer
    if d == nil {
        return
    }
    ...
}

也就是会不断地进入 deferreturn 函数,判断链表中是否还存着 _defer。若已经不存在了,则返回,结束掉它。简单来讲,就是处理完全部 defer 才允许你真的离开它。果真如此吗?我们再看看上面的汇编代码,如下:

    。..
    0x0070 00112 (main.go:6)    CALL    runtime.deferproc(SB)
    0x0075 00117 (main.go:6)    TESTL    AX, AX
    0x0077 00119 (main.go:6)    JNE    137
    0x0079 00121 (main.go:7)    XCHGL    AX, AX
    0x007a 00122 (main.go:7)    CALL    runtime.deferreturn(SB)
    0x007f 00127 (main.go:7)    MOVQ    56(SP), BP
    0x0084 00132 (main.go:7)    ADDQ    $64, SP
    0x0088 00136 (main.go:7)    RET
    0x0089 00137 (main.go:6)    XCHGL    AX, AX
    0x008a 00138 (main.go:6)    CALL    runtime.deferreturn(SB)
    ...

的确如上述流程所分析一致,验证完毕

小结

这个函数主要承担了清空已使用的 defer 和跳转到调用 defer 关键字的函数处,非常重要

总结

我们有提到 defer 关键字涉及两个核心的函数,分别是 deferprocdeferreturn 函数。而 deferreturn 函数比较特殊,是当应用函数调用 defer 关键字时,编译器会在其结尾处插入 deferreturn 的调用,它们俩一般都是成对出现的

但是当一个 Goroutine 上存在着多次 defer 行为(也就是多个 _defer)时,编译器会进行利用一些小技巧, 重新回到 deferreturn 函数去消耗 _defer 链表,直到一个不剩才允许真正的结束

而新增的基础单元 _defer,有可能是被复用的,也有可能是全新申请的。它最后都会被追加到 _defer 链表的表头,从而设定了后进先出的调用特性

关联

参考

查看原文

赞 28 收藏 19 评论 4

bling 关注了用户 · 10月21日

CrazyCodes @crazycodes

https://github.com/CrazyCodes... 我的博客
_
| |__ __ _
| '_ \| | | |/ _` |
| |_) | |_| | (_| |
|_.__/ \__,_|\__, |

         |___/   感谢生命可以让我成为一名程序员

                         CrazyCodes To Author

关注 4671

bling 赞了文章 · 10月21日

PHP8.x 你必须知道的这些新特性

前言

Hello 大家好,我是CrazyCodes,距离上次发文已经过去4个月的时间,今年是悲惨的一年,也是奋发的一年,我会发布一些更好更实用的文章与大家分享,谢谢大家一直以来的支持。

本篇是我参加《2020 PHP开发者峰会》 Nikita分享内了解到的一些知识与大家分享

Nikita 是PHP8的核心开发者。
PHP8的版本会在今年11月26日与各位开发者见面,敬请期待

JIT

值得被提起的则是JIT新的特性,它会将PHP代码转换为传统的机器码,而并非通过zend虚拟机来运行,这样大大的增加了运行速度,但并不向下兼容,这意味着你不能通过像PHP5升级到PHP7那样获得该特性。

JIT可以通过php.ini去设置,例如这样

opcache.jit=on // on 代表打开,则off代表关闭

注解

PHP8版本彻底把注解扶正,当然在这之前像 Symfony,hyperf通过php-parser加入注解的使用方法,但这毕竟不属于PHP8内核真正的部分,在PHP8的版本中,但依旧需要反射

new ReflecationProperty(User::class,"id");

去获取到注解部分,看来注解在PHP的历史长河中还是需要继续不断完善的。

类中的成员变量

小的知识点

在PHP8之前,我们一般会这样定义一个类,首先要设置成员变量,然后在构造或者某一个方法为它赋值。

class User{
    public $username;
    public $phone;
    public $sex;
    
    public function __contruct(
        $username,$phone,$sex
    ){
        $this->username = $username;
        $this->phone = $phone;
        $this->sex = $sex;
    }
}

但在PHP8上,我们可以这样做

class User{
    public function __contruct(
        public string $username = "zhangsan",
        public string $phone = "110110";
        public string $sex = "男"
    ){}
}

命名参数

当我们创建一个函数时,例如

function roule($name,$controller,$model){
    // ... code
}

在调用这个函数时,我们需要顺序输入参数

roule("user/login","UserController","login");

但在PHP8中,我们可以这样做

roule(name:"user/login",controller:"UserController",model:"login");

因为可以需要输入参数名来区分传入的字段,那么在一些函数中,类比中间某项这段需要默认值,那我们就可以跳过这个字段

function roule($name,$controller="UserController",$model){
    // ... code
}
roule(name:"user/login",model:"login");

当然也可以以传统方式与其相结合

roule("user/login",model:"login");

联合类型

在PHP7中,我们在强制函数返回类型时是这样做的

function create() : bool

那么在PHP8中你可以使用多种预测类型

function create() : bool|string

当然在传参时也可以这样做

function create(bool|string $userId)

并且也可以设置类型NULL和TRUE,FALSE了。

总结

以上是PHP8主要的一些特性,所有表达和案例都是在Nikita的基础上描述的,并没有直接照搬,当然Nikita的演讲并不仅仅只有这些,为了保持对峰会主办方的尊重,还请各位移步至
https://www.itdks.com/Home/Ac...
观看Nikita的完整演讲。

致谢

感谢你看到这里,希望本篇文章对你的技术生涯多一分动力,谢谢!

查看原文

赞 17 收藏 7 评论 6

bling 赞了文章 · 10月14日

Linux IO模式及 select、poll、epoll详解

注:本文是对众多博客的学习和总结,可能存在理解错误。请带着怀疑的眼光,同时如果有错误希望能指出。

同步IO和异步IO,阻塞IO和非阻塞IO分别是什么,到底有什么区别?不同的人在不同的上下文下给出的答案是不同的。所以先限定一下本文的上下文。

本文讨论的背景是Linux环境下的network IO。

一 概念说明

在进行解释之前,首先要说明几个概念:
- 用户空间和内核空间
- 进程切换
- 进程的阻塞
- 文件描述符
- 缓存 I/O

用户空间与内核空间

现在操作系统都是采用虚拟存储器,那么对32位操作系统而言,它的寻址空间(虚拟存储空间)为4G(2的32次方)。操作系统的核心是内核,独立于普通的应用程序,可以访问受保护的内存空间,也有访问底层硬件设备的所有权限。为了保证用户进程不能直接操作内核(kernel),保证内核的安全,操心系统将虚拟空间划分为两部分,一部分为内核空间,一部分为用户空间。针对linux操作系统而言,将最高的1G字节(从虚拟地址0xC0000000到0xFFFFFFFF),供内核使用,称为内核空间,而将较低的3G字节(从虚拟地址0x00000000到0xBFFFFFFF),供各个进程使用,称为用户空间。

进程切换

为了控制进程的执行,内核必须有能力挂起正在CPU上运行的进程,并恢复以前挂起的某个进程的执行。这种行为被称为进程切换。因此可以说,任何进程都是在操作系统内核的支持下运行的,是与内核紧密相关的。

从一个进程的运行转到另一个进程上运行,这个过程中经过下面这些变化:
1. 保存处理机上下文,包括程序计数器和其他寄存器。
2. 更新PCB信息。
3. 把进程的PCB移入相应的队列,如就绪、在某事件阻塞等队列。
4. 选择另一个进程执行,并更新其PCB。
5. 更新内存管理的数据结构。
6. 恢复处理机上下文。

注:总而言之就是很耗资源,具体的可以参考这篇文章:进程切换

进程的阻塞

正在执行的进程,由于期待的某些事件未发生,如请求系统资源失败、等待某种操作的完成、新数据尚未到达或无新工作做等,则由系统自动执行阻塞原语(Block),使自己由运行状态变为阻塞状态。可见,进程的阻塞是进程自身的一种主动行为,也因此只有处于运行态的进程(获得CPU),才可能将其转为阻塞状态。当进程进入阻塞状态,是不占用CPU资源的

文件描述符fd

文件描述符(File descriptor)是计算机科学中的一个术语,是一个用于表述指向文件的引用的抽象化概念。

文件描述符在形式上是一个非负整数。实际上,它是一个索引值,指向内核为每一个进程所维护的该进程打开文件的记录表。当程序打开一个现有文件或者创建一个新文件时,内核向进程返回一个文件描述符。在程序设计中,一些涉及底层的程序编写往往会围绕着文件描述符展开。但是文件描述符这一概念往往只适用于UNIX、Linux这样的操作系统。

缓存 I/O

缓存 I/O 又被称作标准 I/O,大多数文件系统的默认 I/O 操作都是缓存 I/O。在 Linux 的缓存 I/O 机制中,操作系统会将 I/O 的数据缓存在文件系统的页缓存( page cache )中,也就是说,数据会先被拷贝到操作系统内核的缓冲区中,然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。

缓存 I/O 的缺点:
数据在传输过程中需要在应用程序地址空间和内核进行多次数据拷贝操作,这些数据拷贝操作所带来的 CPU 以及内存开销是非常大的。

二 IO模式

刚才说了,对于一次IO访问(以read举例),数据会先被拷贝到操作系统内核的缓冲区中,然后才会从操作系统内核的缓冲区拷贝到应用程序的地址空间。所以说,当一个read操作发生时,它会经历两个阶段:
1. 等待数据准备 (Waiting for the data to be ready)
2. 将数据从内核拷贝到进程中 (Copying the data from the kernel to the process)

正式因为这两个阶段,linux系统产生了下面五种网络模式的方案。
- 阻塞 I/O(blocking IO)
- 非阻塞 I/O(nonblocking IO)
- I/O 多路复用( IO multiplexing)
- 信号驱动 I/O( signal driven IO)
- 异步 I/O(asynchronous IO)

注:由于signal driven IO在实际中并不常用,所以我这只提及剩下的四种IO Model。

阻塞 I/O(blocking IO)

在linux中,默认情况下所有的socket都是blocking,一个典型的读操作流程大概是这样:
clipboard.png

当用户进程调用了recvfrom这个系统调用,kernel就开始了IO的第一个阶段:准备数据(对于网络IO来说,很多时候数据在一开始还没有到达。比如,还没有收到一个完整的UDP包。这个时候kernel就要等待足够的数据到来)。这个过程需要等待,也就是说数据被拷贝到操作系统内核的缓冲区中是需要一个过程的。而在用户进程这边,整个进程会被阻塞(当然,是进程自己选择的阻塞)。当kernel一直等到数据准备好了,它就会将数据从kernel中拷贝到用户内存,然后kernel返回结果,用户进程才解除block的状态,重新运行起来。

所以,blocking IO的特点就是在IO执行的两个阶段都被block了。

非阻塞 I/O(nonblocking IO)

linux下,可以通过设置socket使其变为non-blocking。当对一个non-blocking socket执行读操作时,流程是这个样子:
clipboard.png

当用户进程发出read操作时,如果kernel中的数据还没有准备好,那么它并不会block用户进程,而是立刻返回一个error。从用户进程角度讲 ,它发起一个read操作后,并不需要等待,而是马上就得到了一个结果。用户进程判断结果是一个error时,它就知道数据还没有准备好,于是它可以再次发送read操作。一旦kernel中的数据准备好了,并且又再次收到了用户进程的system call,那么它马上就将数据拷贝到了用户内存,然后返回。

所以,nonblocking IO的特点是用户进程需要不断的主动询问kernel数据好了没有。

I/O 多路复用( IO multiplexing)

IO multiplexing就是我们说的select,poll,epoll,有些地方也称这种IO方式为event driven IO。select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。它的基本原理就是select,poll,epoll这个function会不断的轮询所负责的所有socket,当某个socket有数据到达了,就通知用户进程。

clipboard.png

当用户进程调用了select,那么整个进程会被block,而同时,kernel会“监视”所有select负责的socket,当任何一个socket中的数据准备好了,select就会返回。这个时候用户进程再调用read操作,将数据从kernel拷贝到用户进程。

所以,I/O 多路复用的特点是通过一种机制一个进程能同时等待多个文件描述符,而这些文件描述符(套接字描述符)其中的任意一个进入读就绪状态,select()函数就可以返回。

这个图和blocking IO的图其实并没有太大的不同,事实上,还更差一些。因为这里需要使用两个system call (select 和 recvfrom),而blocking IO只调用了一个system call (recvfrom)。但是,用select的优势在于它可以同时处理多个connection。

所以,如果处理的连接数不是很高的话,使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好,可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快,而是在于能处理更多的连接。)

在IO multiplexing Model中,实际中,对于每一个socket,一般都设置成为non-blocking,但是,如上图所示,整个用户的process其实是一直被block的。只不过process是被select这个函数block,而不是被socket IO给block。

异步 I/O(asynchronous IO)

inux下的asynchronous IO其实用得很少。先看一下它的流程:
clipboard.png

用户进程发起read操作之后,立刻就可以开始去做其它的事。而另一方面,从kernel的角度,当它受到一个asynchronous read之后,首先它会立刻返回,所以不会对用户进程产生任何block。然后,kernel会等待数据准备完成,然后将数据拷贝到用户内存,当这一切都完成之后,kernel会给用户进程发送一个signal,告诉它read操作完成了。

总结

blocking和non-blocking的区别

调用blocking IO会一直block住对应的进程直到操作完成,而non-blocking IO在kernel还准备数据的情况下会立刻返回。

synchronous IO和asynchronous IO的区别

在说明synchronous IO和asynchronous IO的区别之前,需要先给出两者的定义。POSIX的定义是这样子的:
- A synchronous I/O operation causes the requesting process to be blocked until that I/O operation completes;
- An asynchronous I/O operation does not cause the requesting process to be blocked;

两者的区别就在于synchronous IO做”IO operation”的时候会将process阻塞。按照这个定义,之前所述的blocking IO,non-blocking IO,IO multiplexing都属于synchronous IO。

有人会说,non-blocking IO并没有被block啊。这里有个非常“狡猾”的地方,定义中所指的”IO operation”是指真实的IO操作,就是例子中的recvfrom这个system call。non-blocking IO在执行recvfrom这个system call的时候,如果kernel的数据没有准备好,这时候不会block进程。但是,当kernel中数据准备好的时候,recvfrom会将数据从kernel拷贝到用户内存中,这个时候进程是被block了,在这段时间内,进程是被block的。

而asynchronous IO则不一样,当进程发起IO 操作之后,就直接返回再也不理睬了,直到kernel发送一个信号,告诉进程说IO完成。在这整个过程中,进程完全没有被block。

各个IO Model的比较如图所示:
clipboard.png

通过上面的图片,可以发现non-blocking IO和asynchronous IO的区别还是很明显的。在non-blocking IO中,虽然进程大部分时间都不会被block,但是它仍然要求进程去主动的check,并且当数据准备完成以后,也需要进程主动的再次调用recvfrom来将数据拷贝到用户内存。而asynchronous IO则完全不同。它就像是用户进程将整个IO操作交给了他人(kernel)完成,然后他人做完后发信号通知。在此期间,用户进程不需要去检查IO操作的状态,也不需要主动的去拷贝数据。

三 I/O 多路复用之select、poll、epoll详解

select,poll,epoll都是IO多路复用的机制。I/O多路复用就是通过一种机制,一个进程可以监视多个描述符,一旦某个描述符就绪(一般是读就绪或者写就绪),能够通知程序进行相应的读写操作。但select,poll,epoll本质上都是同步I/O,因为他们都需要在读写事件就绪后自己负责进行读写,也就是说这个读写过程是阻塞的,而异步I/O则无需自己负责进行读写,异步I/O的实现会负责把数据从内核拷贝到用户空间。(这里啰嗦下)

select

int select (int n, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

select 函数监视的文件描述符分3类,分别是writefds、readfds、和exceptfds。调用后select函数会阻塞,直到有描述副就绪(有数据 可读、可写、或者有except),或者超时(timeout指定等待时间,如果立即返回设为null即可),函数返回。当select函数返回后,可以 通过遍历fdset,来找到就绪的描述符。

select目前几乎在所有的平台上支持,其良好跨平台支持也是它的一个优点。select的一 个缺点在于单个进程能够监视的文件描述符的数量存在最大限制,在Linux上一般为1024,可以通过修改宏定义甚至重新编译内核的方式提升这一限制,但 是这样也会造成效率的降低。

poll

int poll (struct pollfd *fds, unsigned int nfds, int timeout);

不同与select使用三个位图来表示三个fdset的方式,poll使用一个 pollfd的指针实现。

struct pollfd {
    int fd; /* file descriptor */
    short events; /* requested events to watch */
    short revents; /* returned events witnessed */
};

pollfd结构包含了要监视的event和发生的event,不再使用select“参数-值”传递的方式。同时,pollfd并没有最大数量限制(但是数量过大后性能也是会下降)。 和select函数一样,poll返回后,需要轮询pollfd来获取就绪的描述符。

从上面看,select和poll都需要在返回后,通过遍历文件描述符来获取已经就绪的socket。事实上,同时连接的大量客户端在一时刻可能只有很少的处于就绪状态,因此随着监视的描述符数量的增长,其效率也会线性下降。

epoll

epoll是在2.6内核中提出的,是之前的select和poll的增强版本。相对于select和poll来说,epoll更加灵活,没有描述符限制。epoll使用一个文件描述符管理多个描述符,将用户关系的文件描述符的事件存放到内核的一个事件表中,这样在用户空间和内核空间的copy只需一次。

一 epoll操作过程

epoll操作过程需要三个接口,分别如下:

int epoll_create(int size);//创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);

1. int epoll_create(int size);
创建一个epoll的句柄,size用来告诉内核这个监听的数目一共有多大,这个参数不同于select()中的第一个参数,给出最大监听的fd+1的值,参数size并不是限制了epoll所能监听的描述符最大个数,只是对内核初始分配内部数据结构的一个建议
当创建好epoll句柄后,它就会占用一个fd值,在linux下如果查看/proc/进程id/fd/,是能够看到这个fd的,所以在使用完epoll后,必须调用close()关闭,否则可能导致fd被耗尽。

2. int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
函数是对指定描述符fd执行op操作。
- epfd:是epoll_create()的返回值。
- op:表示op操作,用三个宏来表示:添加EPOLL_CTL_ADD,删除EPOLL_CTL_DEL,修改EPOLL_CTL_MOD。分别添加、删除和修改对fd的监听事件。
- fd:是需要监听的fd(文件描述符)
- epoll_event:是告诉内核需要监听什么事,struct epoll_event结构如下:

struct epoll_event {
  __uint32_t events;  /* Epoll events */
  epoll_data_t data;  /* User data variable */
};

//events可以是以下几个宏的集合:
EPOLLIN :表示对应的文件描述符可以读(包括对端SOCKET正常关闭);
EPOLLOUT:表示对应的文件描述符可以写;
EPOLLPRI:表示对应的文件描述符有紧急的数据可读(这里应该表示有带外数据到来);
EPOLLERR:表示对应的文件描述符发生错误;
EPOLLHUP:表示对应的文件描述符被挂断;
EPOLLET: 将EPOLL设为边缘触发(Edge Triggered)模式,这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT:只监听一次事件,当监听完这次事件之后,如果还需要继续监听这个socket的话,需要再次把这个socket加入到EPOLL队列里

3. int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
等待epfd上的io事件,最多返回maxevents个事件。
参数events用来从内核得到事件的集合,maxevents告之内核这个events有多大,这个maxevents的值不能大于创建epoll_create()时的size,参数timeout是超时时间(毫秒,0会立即返回,-1将不确定,也有说法说是永久阻塞)。该函数返回需要处理的事件数目,如返回0表示已超时。

二 工作模式

 epoll对文件描述符的操作有两种模式:LT(level trigger)ET(edge trigger)。LT模式是默认模式,LT模式与ET模式的区别如下:
  LT模式:当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序可以不立即处理该事件。下次调用epoll_wait时,会再次响应应用程序并通知此事件。
  ET模式:当epoll_wait检测到描述符事件发生并将此事件通知应用程序,应用程序必须立即处理该事件。如果不处理,下次调用epoll_wait时,不会再次响应应用程序并通知此事件。

1. LT模式

LT(level triggered)是缺省的工作方式,并且同时支持block和no-block socket.在这种做法中,内核告诉你一个文件描述符是否就绪了,然后你可以对这个就绪的fd进行IO操作。如果你不作任何操作,内核还是会继续通知你的。

2. ET模式

ET(edge-triggered)是高速工作方式,只支持no-block socket。在这种模式下,当描述符从未就绪变为就绪时,内核通过epoll告诉你。然后它会假设你知道文件描述符已经就绪,并且不会再为那个文件描述符发送更多的就绪通知,直到你做了某些操作导致那个文件描述符不再为就绪状态了(比如,你在发送,接收或者接收请求,或者发送接收的数据少于一定量时导致了一个EWOULDBLOCK 错误)。但是请注意,如果一直不对这个fd作IO操作(从而导致它再次变成未就绪),内核不会发送更多的通知(only once)

ET模式在很大程度上减少了epoll事件被重复触发的次数,因此效率要比LT模式高。epoll工作在ET模式的时候,必须使用非阻塞套接口,以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。

3. 总结

假如有这样一个例子:
1. 我们已经把一个用来从管道中读取数据的文件句柄(RFD)添加到epoll描述符
2. 这个时候从管道的另一端被写入了2KB的数据
3. 调用epoll_wait(2),并且它会返回RFD,说明它已经准备好读取操作
4. 然后我们读取了1KB的数据
5. 调用epoll_wait(2)......

LT模式:
如果是LT模式,那么在第5步调用epoll_wait(2)之后,仍然能受到通知。

ET模式:
如果我们在第1步将RFD添加到epoll描述符的时候使用了EPOLLET标志,那么在第5步调用epoll_wait(2)之后将有可能会挂起,因为剩余的数据还存在于文件的输入缓冲区内,而且数据发出端还在等待一个针对已经发出数据的反馈信息。只有在监视的文件句柄上发生了某个事件的时候 ET 工作模式才会汇报事件。因此在第5步的时候,调用者可能会放弃等待仍在存在于文件输入缓冲区内的剩余数据。

当使用epoll的ET模型来工作时,当产生了一个EPOLLIN事件后,
读数据的时候需要考虑的是当recv()返回的大小如果等于请求的大小,那么很有可能是缓冲区还有数据未读完,也意味着该次事件还没有处理完,所以还需要再次读取:

while(rs){
  buflen = recv(activeevents[i].data.fd, buf, sizeof(buf), 0);
  if(buflen < 0){
    // 由于是非阻塞的模式,所以当errno为EAGAIN时,表示当前缓冲区已无数据可读
    // 在这里就当作是该次事件已处理处.
    if(errno == EAGAIN){
        break;
    }
    else{
        return;
    }
  }
  else if(buflen == 0){
     // 这里表示对端的socket已正常关闭.
  }

 if(buflen == sizeof(buf){
      rs = 1;   // 需要再次读取
 }
 else{
      rs = 0;
 }
}

Linux中的EAGAIN含义

Linux环境下开发经常会碰到很多错误(设置errno),其中EAGAIN是其中比较常见的一个错误(比如用在非阻塞操作中)。
从字面上来看,是提示再试一次。这个错误经常出现在当应用程序进行一些非阻塞(non-blocking)操作(对文件或socket)的时候。

例如,以 O_NONBLOCK的标志打开文件/socket/FIFO,如果你连续做read操作而没有数据可读。此时程序不会阻塞起来等待数据准备就绪返回,read函数会返回一个错误EAGAIN,提示你的应用程序现在没有数据可读请稍后再试。
又例如,当一个系统调用(比如fork)因为没有足够的资源(比如虚拟内存)而执行失败,返回EAGAIN提示其再调用一次(也许下次就能成功)。

三 代码演示

下面是一段不完整的代码且格式不对,意在表述上面的过程,去掉了一些模板代码。

#define IPADDRESS   "127.0.0.1"
#define PORT        8787
#define MAXSIZE     1024
#define LISTENQ     5
#define FDSIZE      1000
#define EPOLLEVENTS 100

listenfd = socket_bind(IPADDRESS,PORT);

struct epoll_event events[EPOLLEVENTS];

//创建一个描述符
epollfd = epoll_create(FDSIZE);

//添加监听描述符事件
add_event(epollfd,listenfd,EPOLLIN);

//循环等待
for ( ; ; ){
    //该函数返回已经准备好的描述符事件数目
    ret = epoll_wait(epollfd,events,EPOLLEVENTS,-1);
    //处理接收到的连接
    handle_events(epollfd,events,ret,listenfd,buf);
}

//事件处理函数
static void handle_events(int epollfd,struct epoll_event *events,int num,int listenfd,char *buf)
{
     int i;
     int fd;
     //进行遍历;这里只要遍历已经准备好的io事件。num并不是当初epoll_create时的FDSIZE。
     for (i = 0;i < num;i++)
     {
         fd = events[i].data.fd;
        //根据描述符的类型和事件类型进行处理
         if ((fd == listenfd) &&(events[i].events & EPOLLIN))
            handle_accpet(epollfd,listenfd);
         else if (events[i].events & EPOLLIN)
            do_read(epollfd,fd,buf);
         else if (events[i].events & EPOLLOUT)
            do_write(epollfd,fd,buf);
     }
}

//添加事件
static void add_event(int epollfd,int fd,int state){
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);
}

//处理接收到的连接
static void handle_accpet(int epollfd,int listenfd){
     int clifd;     
     struct sockaddr_in cliaddr;     
     socklen_t  cliaddrlen;     
     clifd = accept(listenfd,(struct sockaddr*)&cliaddr,&cliaddrlen);     
     if (clifd == -1)         
     perror("accpet error:");     
     else {         
         printf("accept a new client: %s:%d\n",inet_ntoa(cliaddr.sin_addr),cliaddr.sin_port);                       //添加一个客户描述符和事件         
         add_event(epollfd,clifd,EPOLLIN);     
     } 
}

//读处理
static void do_read(int epollfd,int fd,char *buf){
    int nread;
    nread = read(fd,buf,MAXSIZE);
    if (nread == -1)     {         
        perror("read error:");         
        close(fd); //记住close fd        
        delete_event(epollfd,fd,EPOLLIN); //删除监听 
    }
    else if (nread == 0)     {         
        fprintf(stderr,"client close.\n");
        close(fd); //记住close fd       
        delete_event(epollfd,fd,EPOLLIN); //删除监听 
    }     
    else {         
        printf("read message is : %s",buf);        
        //修改描述符对应的事件,由读改为写         
        modify_event(epollfd,fd,EPOLLOUT);     
    } 
}

//写处理
static void do_write(int epollfd,int fd,char *buf) {     
    int nwrite;     
    nwrite = write(fd,buf,strlen(buf));     
    if (nwrite == -1){         
        perror("write error:");        
        close(fd);   //记住close fd       
        delete_event(epollfd,fd,EPOLLOUT);  //删除监听    
    }else{
        modify_event(epollfd,fd,EPOLLIN); 
    }    
    memset(buf,0,MAXSIZE); 
}

//删除事件
static void delete_event(int epollfd,int fd,int state) {
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,&ev);
}

//修改事件
static void modify_event(int epollfd,int fd,int state){     
    struct epoll_event ev;
    ev.events = state;
    ev.data.fd = fd;
    epoll_ctl(epollfd,EPOLL_CTL_MOD,fd,&ev);
}

//注:另外一端我就省了

四 epoll总结

在 select/poll中,进程只有在调用一定的方法后,内核才对所有监视的文件描述符进行扫描,而epoll事先通过epoll_ctl()来注册一 个文件描述符,一旦基于某个文件描述符就绪时,内核会采用类似callback的回调机制,迅速激活这个文件描述符,当进程调用epoll_wait() 时便得到通知。(此处去掉了遍历文件描述符,而是通过监听回调的的机制。这正是epoll的魅力所在。)

epoll的优点主要是一下几个方面:
1. 监视的描述符数量不受限制,它所支持的FD上限是最大可以打开文件的数目,这个数字一般远大于2048,举个例子,在1GB内存的机器上大约是10万左 右,具体数目可以cat /proc/sys/fs/file-max察看,一般来说这个数目和系统内存关系很大。select的最大缺点就是进程打开的fd是有数量限制的。这对 于连接数量比较大的服务器来说根本不能满足。虽然也可以选择多进程的解决方案( Apache就是这样实现的),不过虽然linux上面创建进程的代价比较小,但仍旧是不可忽视的,加上进程间数据同步远比不上线程间同步的高效,所以也不是一种完美的方案。

  1. IO的效率不会随着监视fd的数量的增长而下降。epoll不同于select和poll轮询的方式,而是通过每个fd定义的回调函数来实现的。只有就绪的fd才会执行回调函数。

如果没有大量的idle -connection或者dead-connection,epoll的效率并不会比select/poll高很多,但是当遇到大量的idle- connection,就会发现epoll的效率大大高于select/poll。

参考

用户空间与内核空间,进程上下文与中断上下文[总结]
进程切换
维基百科-文件描述符
Linux 中直接 I/O 机制的介绍
IO - 同步,异步,阻塞,非阻塞 (亡羊补牢篇)
Linux中select poll和epoll的区别
IO多路复用之select总结
IO多路复用之poll总结
IO多路复用之epoll总结

查看原文

赞 577 收藏 1041 评论 63

bling 赞了文章 · 10月14日

Golang - 调度剖析【第一部分】

简介

首先,Golang 调度器的设计和实现让我们的 Go 程序在多线程执行时效率更高,性能更好。这要归功于 Go 调度器与操作系统(OS)调度器的协同合作。不过在本篇文章中,多线程 Go 程序在设计和实现上是否与调度器的工作原理完全契合不是重点。重要的是对系统调度器和 Go 调度器,它们是如何正确地设计多线程程序,有一个全面且深入的理解。

本章多数内容将侧重于讨论调度器的高级机制和语义。我将展示一些细节,让你可以通过图像来理解它们是如何工作的,可以让你在写代码时做出更好的决策。因为原理和语义是必备的基础知识中的关键。

系统调度

操作系统调度器是一个复杂的程序。它们要考虑到运行时的硬件设计和设置,其中包括但不限于多处理器核心、CPU 缓存和 NUMA,只有考虑全面,调度器才能做到尽可能地高效。值得高兴的是,你不需要深入研究这些问题,就可以大致上了解操作系统调度器是如何工作的。

你的代码会被翻译成一系列机器指令,然后依次执行。为了实现这一点,操作系统使用线程(Thread)的概念。线程负责顺序执行分配给它的指令。一直执行到没有指令为止。这就是我将线程称为“执行流”的原因。

你运行的每个程序都会创建一个进程,每个进程都有一个初始线程。而后线程可以创建更多的线程。每个线程互相独立地运行着,调度是在线程级别而不是在进程级别做出的。线程可以并发运行(每个线程在单个内核上轮流运行),也可以并行运行(每个线程在不同的内核上同时运行)。线程还维护自己的状态,以便安全、本地和独立地执行它们的指令。

如果有线程可以执行,操作系统调度器就会调度它到空闲的 CPU 核心上去执行,保证 CPU 不闲着。它还必须模拟一个假象,即所有可以执行的线程都在同时地执行着。在这个过程中,调度器还会根据优先级不同选择线程执行的先后顺序,高优先级的先执行,低优先级的后执行。当然,低优先级的线程也不会被饿着。调度器还需要通过快速而明智的决策尽可能减少调度延迟。

为了实现这一目标,算法在其中做了很多工作,且幸运的是,这个领域已经积累了几十年经验。为了我们能更好地理解这一切,接下来我们来看几个重要的概念。

执行指令

程序计数器(PC),有时称为指令指针(IP),线程利用它来跟踪下一个要执行的指令。在大多数处理器中,PC指向的是下一条指令,而不是当前指令。
图片描述
如果你之前看过 Go 程序的堆栈跟踪,那么你可能已经注意到了每行末尾的这些十六进制数字。如下:

goroutine 1 [running]:
   main.example(0xc000042748, 0x2, 0x4, 0x106abae, 0x5, 0xa)
       stack_trace/example1/example1.go:13 +0x39                 <- LOOK HERE
   main.main()
       stack_trace/example1/example1.go:8 +0x72                  <- LOOK HERE

这些数字表示 PC 值与相应函数顶部的偏移量。+0x39PC 偏移量表示在程序没中断的情况下,线程即将执行的下一条指令。如果控制权回到主函数中,则主函数中的下一条指令是0+x72PC 偏移量。更重要的是,指针前面的指令是当前正在执行的指令。

下面是对应的代码
https://github.com/ardanlabs/gotraining/blob/master/topics/go/profiling/stack_trace/example1/example1.go

07 func main() {
08     example(make([]string, 2, 4), "hello", 10)
09 }

12 func example(slice []string, str string, i int) {
13    panic("Want stack trace")
14 }

十六进制数+0x39表示示例函数内的一条指令的 PC 偏移量,该指令位于函数的起始指令后面第57条(10进制)。接下来,我们用 objdump 来看一下汇编指令。找到第57条指令,注意,runtime.gopanic那一行。

$ go tool objdump -S -s "main.example" ./example1
TEXT main.example(SB) stack_trace/example1/example1.go
func example(slice []string, str string, i int) {
  0x104dfa0        65488b0c2530000000    MOVQ GS:0x30, CX
  0x104dfa9        483b6110              CMPQ 0x10(CX), SP
  0x104dfad        762c                  JBE 0x104dfdb
  0x104dfaf        4883ec18              SUBQ $0x18, SP
  0x104dfb3        48896c2410            MOVQ BP, 0x10(SP)
  0x104dfb8        488d6c2410            LEAQ 0x10(SP), BP
    panic("Want stack trace")
  0x104dfbd        488d059ca20000        LEAQ runtime.types+41504(SB), AX
  0x104dfc4        48890424              MOVQ AX, 0(SP)
  0x104dfc8        488d05a1870200        LEAQ main.statictmp_0(SB), AX
  0x104dfcf        4889442408            MOVQ AX, 0x8(SP)
  0x104dfd4        e8c735fdff            CALL runtime.gopanic(SB)
  0x104dfd9        0f0b                  UD2              <--- 这里是 PC(+0x39)

记住: PC 是下一个指令,而不是当前指令。上面是基于 amd64 的汇编指令的一个很好的例子,该 Go 程序的线程负责顺序执行。

线程状态

另一个重要的概念是线程状态,它描述了调度器在线程中的角色。
线程可以处于三种状态之一: 等待中(Waiting)待执行(Runnable)执行中(Executing)

等待中(Waiting):这意味着线程停止并等待某件事情以继续。这可能是因为等待硬件(磁盘、网络)、操作系统(系统调用)或同步调用(原子、互斥)等原因。这些类型的延迟是性能下降的根本原因。

待执行(Runnable):这意味着线程需要内核上的时间,以便执行它指定的机器指令。如果有很多线程都需要时间,那么线程需要等待更长的时间才能获得执行。此外,由于更多的线程在竞争,每个线程获得的单个执行时间都会缩短。这种类型的调度延迟也可能导致性能下降。

执行中(Executing):这意味着线程已经被放置在一个核心上,并且正在执行它的机器指令。与应用程序相关的工作正在完成。这是每个人都想要的。

工作类型

线程可以做两种类型的工作。第一个称为 CPU-Bound,第二个称为 IO-Bound

CPU-Bound:这种工作类型永远也不会让线程处在等待状态,因为这是一项不断进行计算的工作。比如计算 π 的第 n 位,就是一个 CPU-Bound 线程。

IO-Bound:这是导致线程进入等待状态的工作类型。比如通过网络请求对资源的访问或对操作系统进行系统调用。

上下文切换

诸如 Linux、Mac、 Windows 是一个具有抢占式调度器的操作系统。这意味着一些重要的事情。首先,这意味着调度程序在什么时候选择运行哪些线程是不可预测的。线程优先级和事件混在一起(比如在网络上接收数据)使得无法确定调度程序将选择做什么以及什么时候做。

其次,这意味着你永远不能基于一些你曾经历过但不能保证每次都发生的行为来编写代码。如果应用程序中需要确定性,则必须控制线程的同步和协调管理。

在核心上交换线程的物理行为称为上下文切换。当调度器将一个正在执行的线程从内核中取出并将其更改状态为一个可运行的线程时,就会发生上下文切换。

上下文切换的代价是高昂的,因为在核心上交换线程会花费很多时间。上下文切换的延迟取决于不同的因素,大概在在 50 到 100 纳秒之间。考虑到硬件应该能够合理地(平均)在每个核心上每纳秒执行 12 条指令,那么一次上下文切换可能会花费 600 到 1200 条指令的延迟时间。实际上,上下文切换占用了大量程序执行指令的时间。

如果你在执行一个 IO-Bound 程序,那么上下文切换将是一个优势。一旦一个线程更改到等待状态,另一个处于可运行状态的线程就会取而代之。这使得 CPU 总是在工作。这是调度器最重要的之一,最好不要让 CPU 闲下来。

而如果你在执行一个 CPU-Bound 程序,那么上下文切换将成为性能瓶颈的噩梦。由于线程总是有工作要做,所以上下文切换阻碍了工作的进展。这种情况与 IO-Bound 类型的工作形成了鲜明对比。

少即是多

在早期处理器只有一个核心的时代,调度相对简单。因为只有一个核心,所以物理上在任何时候都只有一个线程可以执行。其思想是定义一个调度程序周期,并尝试在这段时间内执行所有可运行线程。算法很简单:用调度周期除以需要执行的线程数。

例如,如果你将调度器周期定义为 10ms(毫秒),并且你有 2 个线程,那么每个线程将分别获得 5ms。如果你有 5 个线程,每个线程得到 2ms。但是,如果有 1000 个线程,会发生什么情况呢?给每个线程一个时间片 10μs (微秒)?错了,这么干是愚蠢的,因为你会花费大量的时间在上下文切换上,而真正的工作却做不成。

你需要限制时间片的长度。在最后一个场景中,如果最小时间片是 2ms,并且有 1000 个线程,那么调度器周期需要增加到 2s(秒)。如果有 10000 个线程,那么调度器周期就是 20s。在这个简单的例子中,如果每个线程使用它的全时间片,那么所有线程运行一次需要花费 20s。

要知道,这是一个非常简单的场景。在真正进行调度决策时,调度程序需要考虑和处理比这更多的事情。你可以控制应用程序中使用的线程数量。当有更多的线程要考虑,并且发生 IO-Bound 工作时,就会出现一些混乱和不确定的行为。任务需要更长的时间来调度和执行。

这就是为什么游戏规则是“少即是多”。处于可运行状态的线程越少,意味着调度开销越少,每个线程执行的时间越长。完成的工作会越多。如此,效率就越高。

寻找一个平衡

你需要在 CPU 核心数和为应用程序获得最佳吞吐量所需的线程数之间找到平衡。当涉及到管理这种平衡时,线程池是一个很好的解决方案。将在第二部分中为你解析,Go 并不是这样做的。

CPU 缓存

从主存访问数据有很高的延迟成本(大约 100 到 300 个时钟周期),因此处理器核心使用本地高速缓存来将数据保存在需要的硬件线程附近。从缓存访问数据的成本要低得多(大约 3 到 40 个时钟周期),这取决于所访问的缓存。如今,提高性能的一个方面是关于如何有效地将数据放入处理器以减少这些数据访问延迟。编写多线程应用程序也需要考虑 CPU 缓存的机制。

图片描述

数据通过cache lines在处理器和主存储器之间交换。cache line是在主存和高速缓存系统之间交换的 64 字节内存块。每个内核都有自己所需的cache line的副本,这意味着硬件使用值语义。这就是为什么多线程应用程序中内存的变化会造成性能噩梦。

当并行运行的多个线程正在访问相同的数据值,甚至是相邻的数据值时,它们将访问同一cache line上的数据。在任何核心上运行的任何线程都将获得同一cache line的副本。

图片描述

如果某个核心上的一个线程对其cache line的副本进行了更改,那么同一cache line的所有其他副本都必须标记为dirty的。当线程尝试对dirty cache line进行读写访问时,需要向主存访问(大约 100 到 300 个时钟周期)来获得cache line的新副本。

也许在一个 2 核处理器上这不是什么大问题,但是如果一个 32 核处理器在同一cache line上同时运行 32 个线程来访问和改变数据,那会发生什么?如果一个系统有两个物理处理器,每个处理器有16个核心,那又该怎么办呢?这将变得更糟,因为处理器到处理器的通信延迟更大。应用程序将会在主存中周转,性能将会大幅下降。

这被称为缓存一致性问题,还引入了错误共享等问题。在编写可能会改变共享状态的多线程应用程序时,必须考虑缓存系统。

调度决策场景

假设我要求你基于我给你的信息编写操作系统调度器。考虑一下这个你必须考虑的情况。记住,这是调度程序在做出调度决策时必须考虑的许多有趣的事情之一。

启动应用程序,创建主线程并在核心1上执行。当线程开始执行其指令时,由于需要数据,正在检索cache line。现在,线程决定为一些并发处理创建一个新线程。下面是问题:

  1. 进行上下文切换,切出核心1的主线程,切入新线程?这样做有助于提高性能,因为这个新线程需要的相同部分的数据很可能已经被缓存。但主线程没有得到它的全部时间片。
  2. 新线程等待核心1在主线程完成之前变为可用?线程没有运行,但一旦启动,获取数据的延迟将被消除。
  3. 线程等待下一个可用的核心?这意味着所选核心的cache line将被刷新、检索和复制,从而导致延迟。然而,线程将启动得更快,主线程可以完成它的时间片。

有意思吗?这些是系统调度器在做出调度决策时需要考虑的有趣问题。幸运的是,不是我做的。我能告诉你的就是,如果有一个空闲核心,它将被使用。你希望线程在可以运行时运行。

结论

本文的第一部分深入介绍了在编写多线程应用程序时需要考虑的关于线程和系统调度器的问题。这些是 Go 调度器也要考虑的事情。在下一篇文章中,我将解析 Go 调度器的语义以及它们如何与这些信息相关联,并通过一些示例程序来展示。

查看原文

赞 60 收藏 43 评论 5

bling 赞了文章 · 9月25日

谈谈前后端分离及认证选择

前几年,web开发领域中「前后端分离」比较火,现如今已逐渐成为事实标准。但是究竟什么是前后端分离?又为什么要前后端分离呢?

什么是前后端分离?为什么要前后端分离?

前后端分离,说的更多的是一种架构上的概念。在传统的web架构中,比如经典的MVC,会分数据层、逻辑层、视图层。这个视图层即我们所说的前端了,映射到代码层面,就是html、js、css等代码文件。数据层和逻辑层更多的是后端部分,例如我们的 .java.go.py等文件。这些文件会在一个工程中,并不会单独的开发、测试、部署。

在前后端分离的架构中,前端和后端是分开的,分别在不同的工程中。前端有专门的前端开发人员来进行开发、测试,后端则有后端开发人员来进行开发、测试,他们之间通过API来交互。

前后端分离有这么几个好处:

1/ 解耦了前后端的工作人员 让前端和后端分别交给更擅长的人来做,细化了工种,可以更加的专精。前端人员来关心用户体验、UI设计、交互渲染;后端人员更关注业务逻辑、性能保障、安全等方面。在项目进度方面,前后端可以并行开发,而互不影响,加快了整体的项目进度。

2/ 解耦了前后端的代码 后端只需提供API服务,不再与静态文件交互。后端可以使用更复杂的分布式、微服务架构,提供更好的性能和稳定性保障。同时前端除了PC端之外,移动端也可以使用相同的一套后端服务。

看到这里,前后端分离被广泛应用也可以理解了。

大家需要注意,并不是所有的项目都需要前后端分离,像是大型的项目,开发人员很多,人员分工明确,这种团队配置下,使用前后端分离可增加工作效率提高系统质量。但是团队人员少,分工不那么明确的情况下,再采用前后端分离的架构,只会增加开发成本和系统复杂度。前后端分离是一个好的架构思路,但是需要看具体的业务和人员情况,切勿盲目的跟从。

前后端分离常用的认证方式

前后端分离中前后端的交互是通过API进行的,那么其中的认证是少不了的。前后端分离中常用的认证方式有下面几种:

  • Session-Cookie
  • Token 验证
  • OAuth(开放授权)

Session-Cookie 方式

Session-Cookie 方式是我们开发web应用时最常用的认证方式。它的认证过程一般是这样的:

  • 1/ 用户浏览器向服务器发起认证请求,将用户名和密码发送给服务器。
  • 2/ 服务器认证用户名和密码,若通过则创建一个session对话,并将用户信息保存到session中。session的信息可以是保存到服务器文件、共享外部存储、数据库等存储中,等下次请求时查询验证使用。
  • 3/ 服务器会将该session的唯一标识ID,返回给用户浏览器,并保存在cookie中。
  • 4/ 用户请求其他页面时,浏览器会自动将用户的cookie携带上,并发起接口请求,服务端收到请求后,从cookie解析出sessionID, 根据这个sessionID 查询登录后并保存好的session,若有则说明用户已登录,放行。

该方式是MVC架构中最常用的认证方案,在前后端分离中也是可以用的。几乎所有的Web框架都默认集成了Session-Cookie的认证方式,而且对Session-Cookie方式的安全性和稳定性方面都有很成熟的处理方案。

当前端代码使用后端web框架当做web容器驱动时,Session-Cookie 方案可作为首选的认证方案。

Token 方式

Token 方式是不同系统交互、前后端架构常用的认证方式。Token 方式的认证流程如下:

  • 1/ 用户使用用户名和密码登录,将用户名和密码发送给服务器。
  • 2/ 服务器验证用户名和密码,若正确,则签发token,返回给用户。
  • 3/ 用户收到token后,将其存储起来,web服务一般为localStrage 或cookie。
  • 4/ 用户请求其他资源页面时,会携带token,一般放到header 或参数中,发送给服务端。
  • 5/ 服务器收到后,验证token,判断用户的正确性。

JWT(JSON Web Token)是最常用的一种Token认证方式,已成为Token认证的标准事实。JWT 方式将Token 分段,使其可以保持少量数据,还增加了签名验证,确保了token的安全性。JWT 网上介绍的资料很多,这里不再赘述。不了解的,可参考下边这些资料:

OAuth 方式

OAuth(Open Authorization)是一个开放标准,允许用户授权第三方网站访问他们存储在服务端的用户信息。我们常见的的QQ、微信等第三方登录便是Auth认证方式。OAuth协议有1.0和2.0两个版本。相比较1.0版,2.0版整个授权验证流程更简单更安全,也是目前最主要的用户身份验证和授权方式。

OAuth更像是一种授权机制。数据的所有者告诉系统,同意授权第三方应用进入系统,获取这些数据。系统从而产生一个短期的进入令牌(token),用来代替密码,供第三方应用使用。

在单纯的前后端分离系统中,OAuth并不是常用的方式,它更多的应用在不同系统之间的授权交互。

对比思考

刨去不常用的OAuth,这里对比两种前两种常用的认证方式 JWT Auth 和 Session-Cookie Auth ,到底谁才是前后端分离认证的最佳实践呢。从下面几个方向分析比对。

可扩展性

Session-Cookie 是有状态的服务,在服务端保存了session的信息。当服务端扩容的时候,需要考虑到session的共享问题,这个问题已有成熟的解决放方案,可使用session复制、共享、持久化等方式解决,大多数的分布式Web框架已经集成了处理方案。JWT 验证方式是无状态的服务,服务端可随意扩缩容。

Session-Cookie 方式基于Cookie,也就是必须是浏览器或支持Cookie的浏览器封装的框架,纯移动端无法使用。JWT 不同,不依赖Cookie, 只要在本地可存储即可。

安全性

Web开发中常见的两个安全问题 XSS(跨站点脚本攻击) 和 CRSF (跨站点请求伪造)。前者利用注入脚本到用户认证网站上,执行恶意脚本代码。后者则利用浏览器访问后端自动携带cookie的机制,来跨站伪造请求。XSS 只要我们对注入端,进行过滤、转义就能解决,CRSF 是我们重点关注的。

在Session-Cookie认证方式中,因为把SessionID保存在了Cookie中,很容易引起CRSF攻击。在大多数的WEB框架中有集成解决方案,如Django 的csrftoken 、Beego的xsrfToken 等。在使用Session-Cookie方案时建议开启web框架的csrf功能。

JWT 认证,可以把Token存放在Cookie或localstorage。建议存在localstorage,这样就彻底避免了 CRSF 攻击。

另外JWT有几个安全性的问题,需要注意:

  • 1/ JWT是明文编码 JWT 的编码是明文Base64的一个编码,是可以反编译的。在使用JWT传输信息的时候,不要放置重要敏感信息,最好使用https。
  • 2/ JWT 泄露问题 解决JWT的泄露问题是一个平衡的问题。有三种处理方式由轻到重,看你业务重要性酌情选择:

    • 将JWT 的过期时间设置的很短,即使泄露也无关紧要。
    • 在服务端设计JWT的黑名单机制,将泄露的Token 加黑名单即可。
    • 保存签发的JWT,当JWT泄露时,直接设置失效。

性能

Session-Cookie方案,因为后端服务存储了Session信息,在认证的时候需要查询,当有大量认证的时候是非常耗费资源的。JWT 可以把信息放到token中,只需要验证解码,使用签名验证token即可,相对来说效率会有提升。

从上面三个方面,我们分析了Session-Cookie和JWT 方式各自的优缺点,和面对问题的一些应对方案。相信大家会有自己的心里选择。

抛开业务场景谈技术都是耍流氓。不同的业务场景,不同的架构设计,适用的认证方式也是不同的。这里按我自己的经验总结了下,什么情况下该使用那种认证方式,大家可参考。

适用Session-Cookie认证方案的情况:

  • 项目只有web端的情况;
  • 项目人员配置少,且前后端开发都会参与;
  • 项目前后端分离不彻底,前端使用后端web框架作为服务容器启动;

使用 JWT 认证方案的情况:

  • 项目人员配置充足,分工明确;
  • 项目除web端外还有移动端;
  • 临时的授权需求;
  • 纯后端系统之间的交互。

本文围绕前后端分离这个话题总结分享了前后端分离时的认证方案。这些仅仅是通用的一般方案,在具体的业务场景中,还有很多不典型的扩展的验证方案也是极好的,欢迎大家留言讨论自己心中的最佳认证方案。

参考及扩展阅读

查看原文

赞 44 收藏 29 评论 2

bling 赞了文章 · 9月15日

总结了才知道,原来channel有这么多用法!

这篇文章总结了channel的10种常用操作,以一个更高的视角看待channel,会给大家带来对channel更全面的认识。

在介绍10种操作前,先简要介绍下channel的使用场景、基本操作和注意事项。

channel的使用场景

把channel用在数据流动的地方

  1. 消息传递、消息过滤
  2. 信号广播
  3. 事件订阅与广播
  4. 请求、响应转发
  5. 任务分发
  6. 结果汇总
  7. 并发控制
  8. 同步与异步
  9. ...

channel的基本操作和注意事项

channel存在3种状态

  1. nil,未初始化的状态,只进行了声明,或者手动赋值为nil
  2. active,正常的channel,可读或者可写
  3. closed,已关闭,千万不要误认为关闭channel后,channel的值是nil

channel可进行3种操作

  1. 关闭

把这3种操作和3种channel状态可以组合出9种情况

操作nil的channel正常channel已关闭channel
<- ch阻塞成功或阻塞读到零值
ch <-阻塞成功或阻塞panic
close(ch)panic成功panic

对于nil通道的情况,也并非完全遵循上表,有1个特殊场景:当nil的通道在select的某个case中时,这个case会阻塞,但不会造成死锁。

参考代码请看:https://dave.cheney.net/2014/...

下面介绍使用channel的10种常用操作。

1. 使用for range读channel

  • 场景:当需要不断从channel读取数据时
  • 原理:使用for-range读取channel,这样既安全又便利,当channel关闭时,for循环会自动退出,无需主动监测channel是否关闭,可以防止读取已经关闭的channel,造成读到数据为通道所存储的数据类型的零值。
  • 用法:
for x := range ch{
    fmt.Println(x)
}

2. 使用_,ok判断channel是否关闭

  • 场景:读channel,但不确定channel是否关闭时
  • 原理:读已关闭的channel会得到零值,如果不确定channel,需要使用ok进行检测。ok的结果和含义:

    • true:读到数据,并且通道没有关闭。
    • false:通道关闭,无数据读到。
  • 用法:
if v, ok := <- ch; ok {
    fmt.Println(v)
}

3. 使用select处理多个channel

  • 场景:需要对多个通道进行同时处理,但只处理最先发生的channel时
  • 原理:select可以同时监控多个通道的情况,只处理未阻塞的case。当通道为nil时,对应的case永远为阻塞,无论读写。特殊关注:普通情况下,对nil的通道写操作是要panic的
  • 用法:
// 分配job时,如果收到关闭的通知则退出,不分配job
func (h *Handler) handle(job *Job) {
    select {
    case h.jobCh<-job:
        return 
    case <-h.stopCh:
        return
    }
}

4. 使用channel的声明控制读写权限

  • 场景:协程对某个通道只读或只写时
  • 目的:A. 使代码更易读、更易维护,B. 防止只读协程对通道进行写数据,但通道已关闭,造成panic。
  • 用法:

    • 如果协程对某个channel只有写操作,则这个channel声明为只写。
    • 如果协程对某个channel只有读操作,则这个channe声明为只读。
// 只有generator进行对outCh进行写操作,返回声明
// <-chan int,可以防止其他协程乱用此通道,造成隐藏bug
func generator(int n) <-chan int {
    outCh := make(chan int)
    go func(){
        for i:=0;i<n;i++{
            outCh<-i
        }
    }()
    return outCh
}

// consumer只读inCh的数据,声明为<-chan int
// 可以防止它向inCh写数据
func consumer(inCh <-chan int) {
    for x := range inCh {
        fmt.Println(x)
    }
}

5. 使用缓冲channel增强并发

  • 场景:并发
  • 原理:有缓冲通道可供多个协程同时处理,在一定程度可提高并发性。
  • 用法:
// 无缓冲
ch1 := make(chan int)
ch2 := make(chan int, 0)
// 有缓冲
ch3 := make(chan int, 1)
func test() {
    inCh := generator(100)
    outCh := make(chan int, 10)

    // 使用5个`do`协程同时处理输入数据
    var wg sync.WaitGroup
    wg.Add(5)
    for i := 0; i < 5; i++ {
        go do(inCh, outCh, &wg)
    }

    go func() {
        wg.Wait()
        close(outCh)
    }()

    for r := range outCh {
        fmt.Println(r)
    }
}

func generator(n int) <-chan int {
    outCh := make(chan int)
    go func() {
        for i := 0; i < n; i++ {
            outCh <- i
        }
        close(outCh)
    }()
    return outCh
}

func do(inCh <-chan int, outCh chan<- int, wg *sync.WaitGroup) {
    for v := range inCh {
        outCh <- v * v
    }

    wg.Done()
}

6. 为操作加上超时

  • 场景:需要超时控制的操作
  • 原理:使用selecttime.After,看操作和定时器哪个先返回,处理先完成的,就达到了超时控制的效果
  • 用法:
func doWithTimeOut(timeout time.Duration) (int, error) {
    select {
    case ret := <-do():
        return ret, nil
    case <-time.After(timeout):
        return 0, errors.New("timeout")
    }
}

func do() <-chan int {
    outCh := make(chan int)
    go func() {
        // do work
    }()
    return outCh
}

7. 使用time实现channel无阻塞读写

  • 场景:并不希望在channel的读写上浪费时间
  • 原理:是为操作加上超时的扩展,这里的操作是channel的读或写
  • 用法:
func unBlockRead(ch chan int) (x int, err error) {
    select {
    case x = <-ch:
        return x, nil
    case <-time.After(time.Microsecond):
        return 0, errors.New("read time out")
    }
}

func unBlockWrite(ch chan int, x int) (err error) {
    select {
    case ch <- x:
        return nil
    case <-time.After(time.Microsecond):
        return errors.New("read time out")
    }
}

注:time.After等待可以替换为default,则是channel阻塞时,立即返回的效果

8. 使用close(ch)关闭所有下游协程

  • 场景:退出时,显示通知所有协程退出
  • 原理:所有读ch的协程都会收到close(ch)的信号
  • 用法:
func (h *Handler) Stop() {
    close(h.stopCh)

    // 可以使用WaitGroup等待所有协程退出
}

// 收到停止后,不再处理请求
func (h *Handler) loop() error {
    for {
        select {
        case req := <-h.reqCh:
            go handle(req)
        case <-h.stopCh:
            return
        }
    }
}

9. 使用chan struct{}作为信号channel

  • 场景:使用channel传递信号,而不是传递数据时
  • 原理:没数据需要传递时,传递空struct
  • 用法:
// 上例中的Handler.stopCh就是一个例子,stopCh并不需要传递任何数据
// 只是要给所有协程发送退出的信号
type Handler struct {
    stopCh chan struct{}
    reqCh chan *Request
}

10. 使用channel传递结构体的指针而非结构体

  • 场景:使用channel传递结构体数据时
  • 原理:channel本质上传递的是数据的拷贝,拷贝的数据越小传输效率越高,传递结构体指针,比传递结构体更高效
  • 用法:
reqCh chan *Request

// 好过
reqCh chan Request

11. 使用channel传递channel

  • 场景:使用场景有点多,通常是用来获取结果。
  • 原理:channel可以用来传递变量,channel自身也是变量,可以传递自己。
  • 用法:下面示例展示了有序展示请求的结果,另一个示例可以见另外文章的版本3
package main

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

func main() {
    reqs := []int{1, 2, 3, 4, 5, 6, 7, 8, 9}

    // 存放结果的channel的channel
    outs := make(chan chan int, len(reqs))
    var wg sync.WaitGroup
    wg.Add(len(reqs))
    for _, x := range reqs {
        o := handle(&wg, x)
        outs <- o
    }

    go func() {
        wg.Wait()
        close(outs)
    }()

    // 读取结果,结果有序
    for o := range outs {
        fmt.Println(<-o)
    }
}

// handle 处理请求,耗时随机模拟
func handle(wg *sync.WaitGroup, a int) chan int {
    out := make(chan int)
    go func() {
        time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
        out <- a
        wg.Done()
    }()
    return out
}

你有哪些channel的奇淫巧技,说来看看?

  1. 如果这篇文章对你有帮助,请点个赞/喜欢,感谢
  2. 本文作者:大彬
  3. 如果喜欢本文,随意转载,但请保留此原文链接:http://lessisbetter.site/2019/01/20/golang-channel-all-usage/

查看原文

赞 55 收藏 36 评论 16

bling 赞了文章 · 9月15日

Go:指针能优化性能吗?【译】

趁着元旦休假+春节,尝试把2018年期间让我受益的一些文章、问答,翻译一下。
欢迎指正、讨论,希望对你也有所帮助。
原文链接:Go: Are pointers a performance optimization?

以下,开始正文

过去几周时间,我回答了许多关于使用指针优化性能的问题。似乎很多人在这方面都感到困惑。这也可以理解,指针确实是个复杂的话题。 希望这篇文章对你有所帮助。

简而言之:不是使用指针就一定代表着性能优化。

如果要彻底解释这篇文章涉及的所有细节,那篇幅可能会长到没人愿意看。所以,我精简了一下,试图用中等篇幅也能涵盖想说明的高级概念。

阅读时需要说明一点:本文讨论的是微优化,性能优化都是极其细微的。在进行微优化之前,需先进行基准测试,否则很可能看不到明显的效果。代码易读性才是第一要义。

什么是指针?

指针底层代表的就是内存地址,指针解引用可以访问到内存存储的具体数据值。

应用指针之后是如何起到性能优化作用的?

函数调用时,变量传递实际上是将变量重新复制了一份,传给函数。多数情况下,指针都要比变量本身占用更小的空间。

通常,指针大小和系统的架构体系保持一致。32位系统就是32bit,64位系统,即是64bit大小。像bool、float等标量类型,占用的空间都小于等于指针;而多字段的符合类型,指针占的空间更少。

所有,我们的想法就基于复制指针比复制原值更高效。一定程度上,这样想没问题。但是性能问题涉及广泛,除了复制成本之外,还有很多因素要考虑。

指针是否会对性能产生负面影响?

会。主要出于两方面的考量:

  1. 解引用虽然耗能很小,但积少成多,不得不虑。
  2. 通过指针共享的数据,是放在堆上的。堆数据的清理是GC负责的,这也会产生开销。随着堆上数据增多,GC的工作量变大,对项目的性能影响也不容忽视。

堆与栈

堆和栈是两个让人头疼的概念,但是我们不得不直面它们。我在这尽量用简短的篇幅讨论完。如果没能快速理解也没关系,我曾经也没能很快理解。

栈:函数局部空间

每当函数被调用,都会分配一块栈空间来存储函数局部变量。函数占用的栈空间大小在编译时已经确定。函数返回时,这块空间就给下一个函数调用使用了,也无需立刻清理。虽然这个分配和使用过程要耗费资源,但相对来讲,消耗很小。

堆:共享数据空间

如上所述,函数返回后,局部变量会被销毁(译者注:空间被复用或者彻底回收,局部变量不再存在)。如果返回是非指针变量,返回值会被复制给调用者,存在于调用者的栈空间中。

但是,如果返回的是指针(译者注:也就是函数局部空间的地址),指针指向的数据就要保存在栈空间之外,这样才能保证函数返回后,数据仍然可以访问。这就是堆的用处。

与堆相关的性能问题有这些:

  1. 堆空间需要从runtime申请,虽然开销很小,也不能不考虑;
  2. 如果运行时没有足够空间,就需要系统调用了,这是额外的开销;
  3. 一旦数据占用了堆空间,就要一直占用到没有指针再指向它。这时候,需要GC来清理。GC会找到堆中所有没有被饮用的值,标记为空闲(译者注:请参考Go垃圾回收的三色标记)。垃圾越多,GC耗时越大,系统性能越差。

那为什么还要用指针?

  1. 指针能修改传参,提供了一种共享数据的方式。
  2. 指针能区分零值,确定你的变量是否被赋值了。

总结

指针可以节省复制的开销,但同时要考虑解引用和垃圾回收带来的影响。在我看来,性能分析结果显示复制是瓶颈之前,不应该考虑把指针作为优化方案。计算机在复制方面的速度可是极快的。

希望这篇文章可以让你认识到指针是可以派上用场的,但也不要因为要用而用。

默认还是使用值而不是地址吧。除非语法满足不了你了。

查看原文

赞 9 收藏 4 评论 2

bling 赞了文章 · 9月15日

golang协程池设计

Why Pool

go自从出生就身带“高并发”的标签,其并发编程就是由groutine实现的,因其消耗资源低,性能高效,开发成本低的特性而被广泛应用到各种场景,例如服务端开发中使用的HTTP服务,在golang net/http包中,每一个被监听到的tcp链接都是由一个groutine去完成处理其上下文的,由此使得其拥有极其优秀的并发量吞吐量

for {
        // 监听tcp
        rw, e := l.Accept()
        if e != nil {
            .......
        }
        tempDelay = 0
        c := srv.newConn(rw)
        c.setState(c.rwc, StateNew) // before Serve can return
        // 启动协程处理上下文
        go c.serve(ctx)
}

虽然创建一个groutine占用的内存极小(大约2KB左右,线程通常2M左右),但是在实际生产环境无限制的开启协程显然是不科学的,比如上图的逻辑,如果来几千万个请求就会开启几千万个groutine,当没有更多内存可用时,go的调度器就会阻塞groutine最终导致内存溢出乃至严重的崩溃,所以本文将通过实现一个简单的协程池,以及剖析几个开源的协程池源码来探讨一下对groutine的并发控制以及多路复用的设计和实现。

一个简单的协程池

过年前做过一波小需求,是将主播管理系统中信息不完整的主播找出来然后再到其相对应的直播平台爬取完整信息并补全,当时考虑到每一个主播的数据都要访问一次直播平台所以就用应对每一个主播开启一个groutine去抓取数据,虽然这个业务量还远远远远达不到能造成groutine性能瓶颈的地步,但是心里总是不舒服,于是放假回来后将其优化成从协程池中控制groutine数量再开启爬虫进行数据抓取。思路其实非常简单,用一个channel当做任务队列,初始化groutine池时确定好并发量,然后以设置好的并发量开启groutine同时读取channel中的任务并执行, 模型如下图
图片描述

实现

type SimplePool struct {
    wg   sync.WaitGroup
    work chan func() //任务队列
}

func NewSimplePoll(workers int) *SimplePool {
    p := &SimplePool{
        wg:   sync.WaitGroup{},
        work: make(chan func()),
    }
    p.wg.Add(workers)
    //根据指定的并发量去读取管道并执行
    for i := 0; i < workers; i++ {
        go func() {
            defer func() {
                // 捕获异常 防止waitGroup阻塞
                if err := recover(); err != nil {
                    fmt.Println(err)
                    p.wg.Done()
                }
            }()
            // 从workChannel中取出任务执行
            for fn := range p.work {
                fn()
            }
            p.wg.Done()
        }()
    }
    return p
}
// 添加任务
func (p *SimplePool) Add(fn func()) {
    p.work <- fn
}

// 执行
func (p *SimplePool) Run() {
    close(p.work)
    p.wg.Wait()
}

测试

测试设定为在并发数量为20的协程池中并发抓取一百个人的信息, 因为代码包含较多业务逻辑所以sleep 1秒模拟爬虫过程,理论上执行时间为5秒

func TestSimplePool(t *testing.T) {
    p := NewSimplePoll(20)
    for i := 0; i < 100; i++ {
        p.Add(parseTask(i))
    }
    p.Run()
}

func parseTask(i int) func() {
    return func() {
        // 模拟抓取数据的过程
        time.Sleep(time.Second * 1)
        fmt.Println("finish parse ", i)
    }
}

图片描述

这样一来最简单的一个groutine池就完成了

go-playground/pool

上面的groutine池虽然简单,但是对于每一个并发任务的状态,pool的状态缺少控制,所以又去看了一下go-playground/pool的源码实现,先从每一个需要执行的任务入手,该库中对并发单元做了如下的结构体,可以看到除工作单元的值,错误,执行函数等,还用了三个分别表示,取消,取消中,写 的三个并发安全的原子操作值来标识其运行状态。

// 需要加入pool 中执行的任务
type WorkFunc func(wu WorkUnit) (interface{}, error)

// 工作单元
type workUnit struct {
    value      interface{}    // 任务结果 
    err        error          // 任务的报错
    done       chan struct{}  // 通知任务完成
    fn         WorkFunc    
    cancelled  atomic.Value   // 任务是否被取消
    cancelling atomic.Value   // 是否正在取消任务
    writing    atomic.Value   // 任务是否正在执行
}

接下来看Pool的结构

type limitedPool struct {
    workers uint            // 并发量 
    work    chan *workUnit  // 任务channel
    cancel  chan struct{}   // 用于通知结束的channel
    closed  bool            // 是否关闭
    m       sync.RWMutex    // 读写锁,主要用来保证 closed值的并发安全
}

初始化groutine池, 以及启动设定好数量的groutine

// 初始化pool,设定并发量
func NewLimited(workers uint) Pool {
    if workers == 0 {
        panic("invalid workers '0'")
    }
    p := &limitedPool{
        workers: workers,
    }
    p.initialize()
    return p
}

func (p *limitedPool) initialize() {
    p.work = make(chan *workUnit, p.workers*2)
    p.cancel = make(chan struct{})
    p.closed = false
    for i := 0; i < int(p.workers); i++ {
        // 初始化并发单元
        p.newWorker(p.work, p.cancel)
    }
}

// passing work and cancel channels to newWorker() to avoid any potential race condition
// betweeen p.work read & write
func (p *limitedPool) newWorker(work chan *workUnit, cancel chan struct{}) {
    go func(p *limitedPool) {

        var wu *workUnit

        defer func(p *limitedPool) {
            // 捕获异常,结束掉异常的工作单元,并将其再次作为新的任务启动
            if err := recover(); err != nil {

                trace := make([]byte, 1<<16)
                n := runtime.Stack(trace, true)

                s := fmt.Sprintf(errRecovery, err, string(trace[:int(math.Min(float64(n), float64(7000)))]))

                iwu := wu
                iwu.err = &ErrRecovery{s: s}
                close(iwu.done)

                // need to fire up new worker to replace this one as this one is exiting
                p.newWorker(p.work, p.cancel)
            }
        }(p)

        var value interface{}
        var err error

        for {
            select {
            // workChannel中读取任务
            case wu = <-work:

                // 防止channel 被关闭后读取到零值
                if wu == nil {
                    continue
                }

                // 先判断任务是否被取消
                if wu.cancelled.Load() == nil {
                    // 执行任务
                    value, err = wu.fn(wu)
                    wu.writing.Store(struct{}{})
                    
                    // 任务执行完在写入结果时需要再次检查工作单元是否被取消,防止产生竞争条件
                    if wu.cancelled.Load() == nil && wu.cancelling.Load() == nil {
                        wu.value, wu.err = value, err
                        close(wu.done)
                    }
                }
            // pool是否被停止
            case <-cancel:
                return
            }
        }

    }(p)
}

往POOL中添加任务,并检查pool是否关闭

func (p *limitedPool) Queue(fn WorkFunc) WorkUnit {
    w := &workUnit{
        done: make(chan struct{}),
        fn:   fn,
    }

    go func() {
        p.m.RLock()
        if p.closed {
            w.err = &ErrPoolClosed{s: errClosed}
            if w.cancelled.Load() == nil {
                close(w.done)
            }
            p.m.RUnlock()
            return
        }
        // 将工作单元写入workChannel, pool启动后将由上面newWorker函数中读取执行
        p.work <- w
        p.m.RUnlock()
    }()

    return w
}

在go-playground/pool包中, limitedPool的批量并发执行还需要借助batch.go来完成

// batch contains all information for a batch run of WorkUnits
type batch struct {
    pool    Pool          // 上面的limitedPool实现了Pool interface
    m       sync.Mutex    // 互斥锁,用来判断closed
    units   []WorkUnit    // 工作单元的slice, 这个主要用在不设并发限制的场景,这里忽略
    results chan WorkUnit // 结果集,执行完后的workUnit会更新其value,error,可以从结果集channel中读取
    done    chan struct{} // 通知batch是否完成
    closed  bool
    wg      *sync.WaitGroup
}
//  go-playground/pool 中有设置并发量和不设并发量的批量任务,都实现Pool interface,初始化batch批量任务时会将之前创建好的Pool传入newBatch
func newBatch(p Pool) Batch {
    return &batch{
        pool:    p,
        units:   make([]WorkUnit, 0, 4), // capacity it to 4 so it doesn't grow and allocate too many times.
        results: make(chan WorkUnit),
        done:    make(chan struct{}),
        wg:      new(sync.WaitGroup),
    }
}

// 往批量任务中添加workFunc任务
func (b *batch) Queue(fn WorkFunc) {

    b.m.Lock()
    if b.closed {
        b.m.Unlock()
        return
    }
    //往上述的limitPool中添加workFunc
    wu := b.pool.Queue(fn)

    b.units = append(b.units, wu) // keeping a reference for cancellation purposes
    b.wg.Add(1)
    b.m.Unlock()
    
    // 执行完后将workUnit写入结果集channel
    go func(b *batch, wu WorkUnit) {
        wu.Wait()
        b.results <- wu
        b.wg.Done()
    }(b, wu)
}

// 通知批量任务不再接受新的workFunc, 如果添加完workFunc不执行改方法的话将导致取结果集时done channel一直阻塞
func (b *batch) QueueComplete() {
    b.m.Lock()
    b.closed = true
    close(b.done)
    b.m.Unlock()
}

// 获取批量任务结果集
func (b *batch) Results() <-chan WorkUnit {
    go func(b *batch) {
        <-b.done
        b.m.Lock()
        b.wg.Wait()
        b.m.Unlock()
        close(b.results)
    }(b)
    return b.results
}

测试

func SendMail(int int) pool.WorkFunc {
    fn := func(wu pool.WorkUnit) (interface{}, error) {
        // sleep 1s 模拟发邮件过程
        time.Sleep(time.Second * 1)
        // 模拟异常任务需要取消
        if int == 17 {
            wu.Cancel()
        }
        if wu.IsCancelled() {
            return false, nil
        }
        fmt.Println("send to", int)
        return true, nil
    }
    return fn
}

func TestBatchWork(t *testing.T) {
    // 初始化groutine数量为20的pool
    p := pool.NewLimited(20)
    defer p.Close()
    batch := p.Batch()
    // 设置一个批量任务的过期超时时间
    t := time.After(10 * time.Second)
    go func() {
        for i := 0; i < 100; i++ {
            batch.Queue(SendMail(i))
        }
        batch.QueueComplete()
    }()
    // 因为 batch.Results 中要close results channel 所以不能将其放在LOOP中执行
    r := batch.Results()
LOOP:
    for {
        select {
        case <-t:
        // 登台超时通知
            fmt.Println("recived timeout")
            break LOOP
     
        case email, ok := <-r:
        // 读取结果集
            if ok {
                if err := email.Error(); err != nil {
                    fmt.Println("err", err.Error())
                }
                fmt.Println(email.Value())
            } else {
                fmt.Println("finish")
                break LOOP
            }
        }
    }
}

    

图片描述
图片描述
接近理论值5s, 通知模拟被取消的work也正常取消

go-playground/pool在比起之前简单的协程池的基础上, 对pool, worker的状态有了很好的管理。但是,但是问题来了,在第一个实现的简单groutine池和go-playground/pool中,都是先启动预定好的groutine来完成任务执行,在并发量远小于任务量的情况下确实能够做到groutine的复用,如果任务量不多则会导致任务分配到每个groutine不均匀,甚至可能出现启动的groutine根本不会执行任务从而导致浪费,而且对于协程池也没有动态的扩容和缩小。所以我又去看了一下ants的设计和实现。

ants

ants是一个受fasthttp启发的高性能协程池, fasthttp号称是比go原生的net/http快10倍,其快速高性能的原因之一就是采用了各种池化技术(这个日后再开新坑去读源码), ants相比之前两种协程池,其模型更像是之前接触到的数据库连接池,需要从空余的worker中取出一个来执行任务, 当无可用空余worker的时候再去创建,而当pool的容量达到上线之后,剩余的任务阻塞等待当前进行中的worker执行完毕将worker放回pool, 直至pool中有空闲worker。 ants在内存的管理上做得很好,除了定期清除过期worker(一定时间内没有分配到任务的worker),ants还实现了一种适用于大批量相同任务的pool, 这种pool与一个需要大批量重复执行的函数锁绑定,避免了调用方不停的创建,更加节省内存。

先看一下ants的pool 结构体 (pool.go)

type Pool struct {
    // 协程池的容量 (groutine数量的上限)
    capacity int32
    // 正在执行中的groutine
    running int32
    // 过期清理间隔时间
    expiryDuration time.Duration
    // 当前可用空闲的groutine
    workers []*Worker
    // 表示pool是否关闭
    release int32
    // lock for synchronous operation.
    lock sync.Mutex
    // 用于控制pool等待获取可用的groutine
    cond *sync.Cond
    // 确保pool只被关闭一次
    once sync.Once
    // worker临时对象池,在复用worker时减少新对象的创建并加速worker从pool中的获取速度
    workerCache sync.Pool
    // pool引发panic时的执行函数
    PanicHandler func(interface{})
}

接下来看pool的工作单元 worker (worker.go)

type Worker struct {
    // worker 所属的poo;
    pool *Pool
    // 任务队列
    task chan func()
    // 回收时间,即该worker的最后一次结束运行的时间
    recycleTime time.Time
}

执行worker的代码 (worker.go)

func (w *Worker) run() {
    // pool中正在执行的worker数+1
    w.pool.incRunning()
    go func() {
        defer func() {
            if p := recover(); p != nil {
                //若worker因各种问题引发panic, 
                //pool中正在执行的worker数 -1,         
                //如果设置了Pool中的PanicHandler,此时会被调用
                w.pool.decRunning()
                if w.pool.PanicHandler != nil {
                    w.pool.PanicHandler(p)
                } else {
                    log.Printf("worker exits from a panic: %v", p)
                }
            }
        }()
        
        // worker 执行任务队列
        for f := range w.task {
            //任务队列中的函数全部被执行完后,
            //pool中正在执行的worker数 -1, 
            //将worker 放回对象池
            if f == nil {
                w.pool.decRunning()
                w.pool.workerCache.Put(w)
                return
            }
            f()
            //worker 执行完任务后放回Pool 
            //使得其余正在阻塞的任务可以获取worker
            w.pool.revertWorker(w)
        }
    }()
}

了解了工作单元worker如何执行任务以及与pool交互后,回到pool中查看其实现, pool的核心就是取出可用worker提供给任务执行 (pool.go)

// 向pool提交任务
func (p *Pool) Submit(task func()) error {
    if 1 == atomic.LoadInt32(&p.release) {
        return ErrPoolClosed
    }
    // 获取pool中的可用worker并向其任务队列中写入任务
    p.retrieveWorker().task <- task
    return nil
}


// **核心代码** 获取可用worker
func (p *Pool) retrieveWorker() *Worker {
    var w *Worker

    p.lock.Lock()
    idleWorkers := p.workers
    n := len(idleWorkers) - 1
  // 当前pool中有可用worker, 取出(队尾)worker并执行
    if n >= 0 {
        w = idleWorkers[n]
        idleWorkers[n] = nil
        p.workers = idleWorkers[:n]
        p.lock.Unlock()
    } else if p.Running() < p.Cap() {
        p.lock.Unlock()
        // 当前pool中无空闲worker,且pool数量未达到上线
        // pool会先从临时对象池中寻找是否有已完成任务的worker,
        // 若临时对象池中不存在,则重新创建一个worker并将其启动
        if cacheWorker := p.workerCache.Get(); cacheWorker != nil {
            w = cacheWorker.(*Worker)
        } else {
            w = &Worker{
                pool: p,
                task: make(chan func(), workerChanCap),
            }
        }
        w.run()
    } else {
        // pool中没有空余worker且达到并发上限
        // 任务会阻塞等待当前运行的worker完成任务释放会pool
        for {
            p.cond.Wait() // 等待通知, 暂时阻塞
            l := len(p.workers) - 1
            if l < 0 {
                continue
            }
            // 当有可用worker释放回pool之后, 取出
            w = p.workers[l]
            p.workers[l] = nil
            p.workers = p.workers[:l]
            break
        }
        p.lock.Unlock()
    }
    return w
}

// 释放worker回pool
func (p *Pool) revertWorker(worker *Worker) {
    worker.recycleTime = time.Now()
    p.lock.Lock()
    p.workers = append(p.workers, worker)
    // 通知pool中已经获取锁的groutine, 有一个worker已完成任务
    p.cond.Signal()
    p.lock.Unlock()
}

在批量并发任务的执行过程中, 如果有超过5纳秒(ants中默认worker过期时间为5ns)的worker未被分配新的任务,则将其作为过期worker清理掉,从而保证pool中可用的worker都能发挥出最大的作用以及将任务分配得更均匀
(pool.go)

// 该函数会在pool初始化后在协程中启动
func (p *Pool) periodicallyPurge() {
    // 创建一个5ns定时的心跳
    heartbeat := time.NewTicker(p.expiryDuration)
    defer heartbeat.Stop()

    for range heartbeat.C {
        currentTime := time.Now()
        p.lock.Lock()
        idleWorkers := p.workers
        if len(idleWorkers) == 0 && p.Running() == 0 && atomic.LoadInt32(&p.release) == 1 {
            p.lock.Unlock()
            return
        }
        n := -1
        for i, w := range idleWorkers {
            // 因为pool 的worker队列是先进后出的,所以正序遍历可用worker时前面的往往里当前时间越久
            if currentTime.Sub(w.recycleTime) <= p.expiryDuration {
                break
            }    
            // 如果worker最后一次运行时间距现在超过5纳秒,视为过期,worker收到nil, 执行上述worker.go中 if n == nil 的操作
            n = i
            w.task <- nil
            idleWorkers[i] = nil
        }
        if n > -1 {
            // 全部过期
            if n >= len(idleWorkers)-1 {
                p.workers = idleWorkers[:0]
            } else {
            // 部分过期
                p.workers = idleWorkers[n+1:]
            }
        }
        p.lock.Unlock()
    }
}

测试

func TestAnts(t *testing.T) {
    wg := sync.WaitGroup{}
    pool, _ := ants.NewPool(20)
    defer pool.Release()
    for i := 0; i < 100; i++ {
        wg.Add(1)
        pool.Submit(sendMail(i, &wg))
    }
    wg.Wait()
}

func sendMail(i int, wg *sync.WaitGroup) func() {
    return func() {
        time.Sleep(time.Second * 1)
        fmt.Println("send mail to ", i)
        wg.Done()
    }
}

图片描述
这里虽只简单的测试批量并发任务的场景, 如果大家有兴趣可以去看看ants的压力测试, ants的吞吐量能够比原生groutine高出N倍,内存节省10到20倍, 可谓是协程池中的神器。

借用ants作者的原话来说:
然而又有多少场景是单台机器需要扛100w甚至1000w同步任务的?基本没有啊!结果就是造出了屠龙刀,可是世界上没有龙啊!也是无情…

Over

一口气从简单到复杂总结了三个协程池的实现,受益匪浅, 感谢各开源库的作者, 虽然世界上没有龙,但是屠龙技是必须练的,因为它就像存款,不一定要全部都用了,但是一定不能没有!

查看原文

赞 46 收藏 35 评论 6

bling 赞了文章 · 9月15日

Go 程序是如何编译成目标机器码的

今天我们一起来研究 Go 1.11 的编译器,以及它将 Go 程序代码编译成可执行文件的过程。以便了解我们日常使用的工具是如何工作的。
本文还会带你了解 Go 程序为什么这么快,以及编译器在这中间起到了什么作用。

首先,编译器的三个阶段:

  1. 逐行扫描源代码,将之转换为一系列的 token,交给 parser 解析。
  2. parser,它将一系列 token 转换为 AST(抽象语法树),用于下一步生成代码。
  3. 最后一步,代码生成,会利用上一步生成的 AST 并根据目标机器平台的不同,生成目标机器码。

注意:下面使用的代码包(go/scannergo/parsergo/tokengo/ast)主要是让我们可以方便地对 Go 代码进行解析和生成,做出更有趣的事情。但是 Go 本身的编译器并不是用这些代码包实现的。

扫描代码,进行词法分析

任何编译器的第一步都是将源代码文本分解成 token,由扫描程序(也称为词法分析器)完成。token 可以是关键字,字符串,变量名,函数名等等。每一个有效的词都由 token 表示。
在 Go 中,我们写在代码上的 "package""main""func" 这些都是 token

token 由代码中的位置,类型和原始文本组成。我们可以使用 go/scanner 和 go/token 包在 Go 程序中自己执行扫描程序。这意味着我们可以像编译器那样扫描检视自己的代码。
下面,我们将通过一个打印 Hello World 的示例来展示 token

package main

import (
    "fmt"
    "go/scanner"
    "go/token"
)

func main() {
    src := []byte(`
package main

import "fmt"

func main() {
    fmt.Println("Hello, world!")
}
`)

    var s scanner.Scanner
    fset := token.NewFileSet()
    file := fset.AddFile("", fset.Base(), len(src))
    s.Init(file, src, nil, 0)

    for {
        pos, tok, lit := s.Scan()
        fmt.Printf("%-6s%-8s%q\n", fset.Position(pos), tok, lit)

        if tok == token.EOF {
            break
        }
    }
}

首先通过源代码字符串创建 token 集合并初始化 scan.Scanner,它将逐行扫描我们的源代码。
接下来循环调用 Scan() 并打印每个 token 的位置,类型和文本字符串,直到遇到文件结束(EOF)标记。

输出:

2:1   package "package"
2:9   IDENT   "main"
2:13  ;       "\n"
4:1   import  "import"
4:8   STRING  "\"fmt\""
4:13  ;       "\n"
6:1   func    "func"
6:6   IDENT   "main"
6:10  (       ""
6:11  )       ""
6:13  {       ""
7:2   IDENT   "fmt"
7:5   .       ""
7:6   IDENT   "Println"
7:13  (       ""
7:14  STRING  "\"Hello, world!\""
7:29  )       ""
7:30  ;       "\n"
8:1   }       ""
8:2   ;       "\n"
8:3   EOF     ""

以第一行为例分析这个输出,第一列 2:1 表示扫描到了源代码第二行第一个字符,第二列 package 表示 tokenpackage,第三列 "package" 表示源代码文本。
我们可以看到在 Scanner 执行过程中将 \n 换行符标记成了 ; 分号,像在 C 语言中是用分号表示一行结束的。这就解释了为什么 Go 不需要分号:它们是在词法分析阶段由 Scanner 智能地解释的。

语法分析

源代码扫描完成后,扫描结果将被传递给语法分析器。语法分析是编译的一个阶段,它将 token 转换为 抽象语法树(AST)
AST 是源代码的结构化表示。在 AST 中,我们将能够看到程序结构,比如函数和常量声明。

我们使用 go/parsergo/ast 来打印完整的 AST

package main

import (
    "go/ast"
    "go/parser"
    "go/token"
    "log"
)

func main() {
    src := []byte(`
package main

import "fmt"

func main() {
    fmt.Println("Hello, world!")
}
`)

    fset := token.NewFileSet()

    file, err := parser.ParseFile(fset, "", src, 0)
    if err != nil {
        log.Fatal(err)
    }

    ast.Print(fset, file)
}

输出:

     0  *ast.File {
     1  .  Package: 2:1
     2  .  Name: *ast.Ident {
     3  .  .  NamePos: 2:9
     4  .  .  Name: "main"
     5  .  }
     6  .  Decls: []ast.Decl (len = 2) {
     7  .  .  0: *ast.GenDecl {
     8  .  .  .  TokPos: 4:1
     9  .  .  .  Tok: import
    10  .  .  .  Lparen: -
    11  .  .  .  Specs: []ast.Spec (len = 1) {
    12  .  .  .  .  0: *ast.ImportSpec {
    13  .  .  .  .  .  Path: *ast.BasicLit {
    14  .  .  .  .  .  .  ValuePos: 4:8
    15  .  .  .  .  .  .  Kind: STRING
    16  .  .  .  .  .  .  Value: "\"fmt\""
    17  .  .  .  .  .  }
    18  .  .  .  .  .  EndPos: -
    19  .  .  .  .  }
    20  .  .  .  }
    21  .  .  .  Rparen: -
    22  .  .  }
    23  .  .  1: *ast.FuncDecl {
    24  .  .  .  Name: *ast.Ident {
    25  .  .  .  .  NamePos: 6:6
    26  .  .  .  .  Name: "main"
    27  .  .  .  .  Obj: *ast.Object {
    28  .  .  .  .  .  Kind: func
    29  .  .  .  .  .  Name: "main"
    30  .  .  .  .  .  Decl: *(obj @ 23)
    31  .  .  .  .  }
    32  .  .  .  }
    33  .  .  .  Type: *ast.FuncType {
    34  .  .  .  .  Func: 6:1
    35  .  .  .  .  Params: *ast.FieldList {
    36  .  .  .  .  .  Opening: 6:10
    37  .  .  .  .  .  Closing: 6:11
    38  .  .  .  .  }
    39  .  .  .  }
    40  .  .  .  Body: *ast.BlockStmt {
    41  .  .  .  .  Lbrace: 6:13
    42  .  .  .  .  List: []ast.Stmt (len = 1) {
    43  .  .  .  .  .  0: *ast.ExprStmt {
    44  .  .  .  .  .  .  X: *ast.CallExpr {
    45  .  .  .  .  .  .  .  Fun: *ast.SelectorExpr {
    46  .  .  .  .  .  .  .  .  X: *ast.Ident {
    47  .  .  .  .  .  .  .  .  .  NamePos: 7:2
    48  .  .  .  .  .  .  .  .  .  Name: "fmt"
    49  .  .  .  .  .  .  .  .  }
    50  .  .  .  .  .  .  .  .  Sel: *ast.Ident {
    51  .  .  .  .  .  .  .  .  .  NamePos: 7:6
    52  .  .  .  .  .  .  .  .  .  Name: "Println"
    53  .  .  .  .  .  .  .  .  }
    54  .  .  .  .  .  .  .  }
    55  .  .  .  .  .  .  .  Lparen: 7:13
    56  .  .  .  .  .  .  .  Args: []ast.Expr (len = 1) {
    57  .  .  .  .  .  .  .  .  0: *ast.BasicLit {
    58  .  .  .  .  .  .  .  .  .  ValuePos: 7:14
    59  .  .  .  .  .  .  .  .  .  Kind: STRING
    60  .  .  .  .  .  .  .  .  .  Value: "\"Hello, world!\""
    61  .  .  .  .  .  .  .  .  }
    62  .  .  .  .  .  .  .  }
    63  .  .  .  .  .  .  .  Ellipsis: -
    64  .  .  .  .  .  .  .  Rparen: 7:29
    65  .  .  .  .  .  .  }
    66  .  .  .  .  .  }
    67  .  .  .  .  }
    68  .  .  .  .  Rbrace: 8:1
    69  .  .  .  }
    70  .  .  }
    71  .  }
    72  .  Scope: *ast.Scope {
    73  .  .  Objects: map[string]*ast.Object (len = 1) {
    74  .  .  .  "main": *(obj @ 27)
    75  .  .  }
    76  .  }
    77  .  Imports: []*ast.ImportSpec (len = 1) {
    78  .  .  0: *(obj @ 12)
    79  .  }
    80  .  Unresolved: []*ast.Ident (len = 1) {
    81  .  .  0: *(obj @ 46)
    82  .  }
    83  }

分析这个输出,在 Decls 字段中,包含了代码中所有的声明,例如导入、常量、变量和函数。在本例中,我们只有两个:导入fmt包主函数
为了进一步理解它,我们可以看看下面这个图,它是上述数据的表示,但只包含类型,红色代表与节点对应的代码:

图片描述

main函数由三个部分组成:NameTypeBodyName 是值为 main 的标识符。由 Type 字段指定的声明将包含参数列表和返回类型(如果我们指定了的话)。正文由一系列语句组成,里面包含了程序的所有行,在本例中只有一行fmt.Println("Hello, world!")

我们的一条 fmt.Println 语句由 AST 中很多部分组成。
该语句是一个 ExprStmt表达式语句(expression statement),例如,它可以像这里一样是一个函数调用,它可以是字面量,可以是一个二元运算(例如加法和减法),当然也可以是一元运算(例如自增++,自减--,否定!等)等等。
同时,在函数调用的参数中可以使用任何表达式。

然后,ExprStmt 又包含一个 CallExpr,它是我们实际的函数调用。里面又包括几个部分,其中最重要的部分是 FunArgs
Fun 包含对函数调用的引用,在这种情况下,它是一个 SelectorExpr,因为我们从 fmt 包中选择 Println 标识符。
但是至此,在 AST 中,编译器还不知道 fmt 是一个包,它也可能是 AST 中的一个变量。

Args 包含一个表达式列表,它是函数的参数。这里,我们将一个文本字符串传递给函数,因而它由一个类型为 STRINGBasicLit 表示。

显然,AST 包含了许多信息,我们不仅可以分析出以上结论,还可以进一步检查 AST 并查找文件中的所有函数调用。下面,我们将使用 go/ast 包中的 Inspect 函数来递归地遍历树,并分析所有节点的信息。

package main

import (
    "fmt"
    "go/ast"
    "go/parser"
    "go/printer"
    "go/token"
    "os"
)

func main() {
    src := []byte(`
package main

import "fmt"

func main() {
    fmt.Println("Hello, world!")
}
`)

    fset := token.NewFileSet()

    file, err := parser.ParseFile(fset, "", src, 0)
    if err != nil {
        fmt.Println(err)
    }

    ast.Inspect(file, func(n ast.Node) bool {
        call, ok := n.(*ast.CallExpr)
        if !ok {
            return true
        }

        printer.Fprint(os.Stdout, fset, call.Fun)
        
        return false
    })
}

输出:

fmt.Println

上面代码的作用是查找所有节点以及它们是否为 *ast.CallExpr 类型,上面也说过这种类型是函数调用。如果是,则使用 go/printer 包打印 Fun 中存在的函数的名称。

构建出 AST 后,将使用 GOPATH 或者在 Go 1.11 及更高版本中的 modules 解析所有导入。然后,执行类型检查,并做一些让程序运行更快的初级优化。

代码生成

在解析导入并做了类型检查之后,我们可以确认程序是合法的 Go 代码,然后就走到将 AST 转换为(伪)目标机器码的过程。

此过程的第一步是将 AST 转换为程序的低级表示,特别是转换为 静态单赋值(SSA)表单。这个中间表示不是最终的机器代码,但它确实代表了最终的机器代码。 SSA 具有一组属性,会使应用优化变得更容易,其中最重要的是在使用变量之前总是定义变量,并且每个变量只分配一次。

在生成 SSA 的初始版本之后,将执行一些优化。这些优化适用于某些代码,可以使处理器执行起来更简单且更快速。例如,可以做 死码消除。还有比如可以删除某些 nil 检查,因为编译器可以证明这些检查永远不会出错。

现在通过最简单的例子来说明 SSA 和一些优化过程:

package main

import "fmt"

func main() {
    fmt.Println(2)
}

如你所见,此程序只有一个函数和一个导入。它会在运行时打印 2。但是,此例足以让我们了解SSA。

为了显示生成的 SSA,我们需要将 GOSSAFUNC 环境变量设置为我们想要跟踪的函数,在本例中为main 函数。我们还需要将 -S 标识传递给编译器,这样它就会打印代码并创建一个HTML文件。我们还将编译Linux 64位的文件,以确保机器代码与您在这里看到的相同。
在终端执行下面的命令:

GOSSAFUNC=main GOOS=linux GOARCH=amd64 go build -gcflags -S main.go

会在终端打印出所有的 SSA,同时也会生成一个交互式的 ssa.html 文件,我们用浏览器打开它。

图片描述

当你打开 ssa.html 时,将显示很多阶段,其中大部分都已折叠。start 阶段是从 AST 生成的SSAlower 阶段将非机器特定的 SSA 转换为机器特定的 SSA,最后的 genssa 就是生成的机器代码。

start 阶段的代码如下:

b1:
    v1  = InitMem <mem>
    v2  = SP <uintptr>
    v3  = SB <uintptr>
    v4  = ConstInterface <interface {}>
    v5  = ArrayMake1 <[1]interface {}> v4
    v6  = VarDef <mem> {.autotmp_0} v1
    v7  = LocalAddr <*[1]interface {}> {.autotmp_0} v2 v6
    v8  = Store <mem> {[1]interface {}} v7 v5 v6
    v9  = LocalAddr <*[1]interface {}> {.autotmp_0} v2 v8
    v10 = Addr <*uint8> {type.int} v3
    v11 = Addr <*int> {"".statictmp_0} v3
    v12 = IMake <interface {}> v10 v11
    v13 = NilCheck <void> v9 v8
    v14 = Const64 <int> [0]
    v15 = Const64 <int> [1]
    v16 = PtrIndex <*interface {}> v9 v14
    v17 = Store <mem> {interface {}} v16 v12 v8
    v18 = NilCheck <void> v9 v17
    v19 = IsSliceInBounds <bool> v14 v15
    v24 = OffPtr <*[]interface {}> [0] v2
    v28 = OffPtr <*int> [24] v2
If v19 → b2 b3 (likely) (line 6)

b2: ← b1
    v22 = Sub64 <int> v15 v14
    v23 = SliceMake <[]interface {}> v9 v22 v22
    v25 = Copy <mem> v17
    v26 = Store <mem> {[]interface {}} v24 v23 v25
    v27 = StaticCall <mem> {fmt.Println} [48] v26
    v29 = VarKill <mem> {.autotmp_0} v27
Ret v29 (line 7)

b3: ← b1
    v20 = Copy <mem> v17
    v21 = StaticCall <mem> {runtime.panicslice} v20
Exit v21 (line 6)

这个简单的程序就已经产生了相当多的 SSA(总共35行)。然而,很多都是引用,可以消除很多(最终的SSA版本有28行,最终的机器代码版本有18行)。

每个 v 都是一个新变量,可以点击来查看它被使用的位置。b 是块,这里有三块:b1,b2,b3。b1 始终会执行,b2b3 是条件块,满足条件才执行。
我们来看 b1 结尾处的 If v19 → b2 b3 (likely)。单击该行中的 v19 可以查看它定义的位置。可以看到它定义为 IsSliceInBounds <bool> v14 v15,通过 Go 编译器源代码,我们知道 IsSliceInBounds 的作用是检查 0 <= arg0 <= arg1。然后单击 v14v15 看看在哪定义的,我们会看到 v14 = Const64 <int> [0],Const64 是一个常量 64 位整数。 v15 定义一样,放在 args1 的位置。所以,实际执行的是 0 <= 0 <= 1,这显然是正确的。

编译器也能够证明这一点,当我们查看 opt 阶段(“机器无关优化”)时,我们可以看到它已经重写了 v19ConstBool <bool> [true]。结果就是,在 opt deadcode 阶段,b3 条件块被删除了,因为永远也不会执行到 b3

下面来看一下 Go 编译器在把 SSA 转换为 机器特定的SSA 之后所做的另一个更简单的优化,基于amd64体系结构的机器代码。下面,我们将比较 lowerlowered deadcode
lower

b1:
    BlockInvalid (6)
b2:
    v2 (?) = SP <uintptr>
    v3 (?) = SB <uintptr>
    v10 (?) = LEAQ <*uint8> {type.int} v3
    v11 (?) = LEAQ <*int> {"".statictmp_0} v3
    v15 (?) = MOVQconst <int> [1]
    v20 (?) = MOVQconst <uintptr> [0]
    v25 (?) = MOVQconst <*uint8> [0]
    v1 (?) = InitMem <mem>
    v6 (6) = VarDef <mem> {.autotmp_0} v1
    v7 (6) = LEAQ <*[1]interface {}> {.autotmp_0} v2
    v9 (6) = LEAQ <*[1]interface {}> {.autotmp_0} v2
    v16 (+6) = LEAQ <*interface {}> {.autotmp_0} v2
    v18 (6) = LEAQ <**uint8> {.autotmp_0} [8] v2
    v21 (6) = LEAQ <**uint8> {.autotmp_0} [8] v2
    v30 (6) = LEAQ <*int> [16] v2
    v19 (6) = LEAQ <*int> [8] v2
    v23 (6) = MOVOconst <int128> [0]
    v8 (6) = MOVOstore <mem> {.autotmp_0} v2 v23 v6
    v22 (6) = MOVQstore <mem> {.autotmp_0} v2 v10 v8
    v17 (6) = MOVQstore <mem> {.autotmp_0} [8] v2 v11 v22
    v14 (6) = MOVQstore <mem> v2 v9 v17
    v28 (6) = MOVQstoreconst <mem> [val=1,off=8] v2 v14
    v26 (6) = MOVQstoreconst <mem> [val=1,off=16] v2 v28
    v27 (6) = CALLstatic <mem> {fmt.Println} [48] v26
    v29 (5) = VarKill <mem> {.autotmp_0} v27
Ret v29 (+7)

在HTML中,某些行是灰色的,这意味着它们将在下一个阶段中被删除或修改。
例如,v15 (?) = MOVQconst <int> [1] 显示为灰色。点击 v15,我们看到它在其他地方都没有使用,而 MOVQconst 基本上与我们之前看到的 Const64 相同,只针对amd64的特定机器。我们把 v15 设置为1。但是,v15 在其他地方都没有使用,所以它是无用的(死的)代码并且可以消除。

Go 编译器应用了很多这类优化。因此,虽然 AST 生成的初始 SSA 可能不是最快的实现,但编译器将SSA优化为更快的版本。 HTML 文件中的每个阶段都有可能发生优化。

如果你有兴趣了解 Go 编译器中有关 SSA 的更多信息,请查看 Go 编译器的 SSA 源代码
这里定义了所有的操作以及优化。

结论

Go 是一种非常高效且高性能的语言,由其编译器及其优化支撑。要了解有关 Go 编译器的更多信息,源代码的 README 是不错的选择。

查看原文

赞 36 收藏 19 评论 4

认证与成就

  • 获得 0 次点赞
  • 获得 1 枚徽章 获得 0 枚金徽章, 获得 0 枚银徽章, 获得 1 枚铜徽章

擅长技能
编辑

(゚∀゚ )
暂时没有

开源项目 & 著作
编辑

(゚∀゚ )
暂时没有

注册于 2018-08-28
个人主页被 67 人浏览