MoreRSS

site iconColobu | 鸟窝 晁岳攀修改

博客 编程 作者 开源
rpcx作者，出版《深入理解Go并发编程》等，中科大，先后在清华同方、Motorola、Comcast、新浪等公司工作。
RSS: https://colobu.com/atom.xml
请复制 RSS 到你的阅读器，或快速订阅到 :

Inoreader Feedly Follow Feedbin Local Reader

Colobu | 鸟窝 晁岳攀的 RSS 预览

趁着假期， 快速了解 Go io/fs 包

2025-01-30 00:44:01

Go 语言的 io/fs 包是 Go 1.16 版本引入的一个标准库包，它定义了文件系统的抽象接口。这个包提供了一种统一的方式来访问不同类型的文件系统，包括本地文件系统、内存文件系统、zip 文件等。

io/fs 包的主要作用

  • 抽象文件系统： io/fs 包定义了一组接口，用于描述文件系统的基本操作，如打开文件、读取目录等。通过这些接口，我们可以编写与具体文件系统无关的代码。
  • 统一访问方式： 无论底层文件系统是什么类型，只要实现了 io/fs 包定义的接口，就可以使用相同的代码进行访问。
  • 提高代码可测试性： 通过使用 io/fs 包，我们可以方便地mock文件系统，从而提高代码的可测试性。

io/fs 包的核心接口

  • fs.FS： 表示一个文件系统，定义了打开文件的方法 Open。
  • fs.File： 表示一个打开的文件，定义了读取、写入、关闭等方法。
  • fs.FileInfo： 表示文件的元信息，包括文件名、大小、修改时间等。
  • fs.DirEntry 接口表示一个目录项，它可以是文件或子目录。
  • fs.FileInfo 接口表示文件的元信息。
  • fs.FileMode 类型表示文件的权限和类型，它是一个位掩码。

还有一些基于fs.FS、fs.File等接口扩展的一些接口：

  • fs.GlobFS 接口扩展了 fs.FS 接口，增加了 Glob(pattern string) ([]string, error) 方法。该方法允许使用通配符模式匹配文件和目录。
  • fs.ReadDirFS 接口也扩展了 fs.FS 接口，增加了 ReadDir(name string) ([]fs.DirEntry, error) 方法。该方法用于读取指定目录下的所有文件和子目录。
  • fs.ReadDirFile 接口扩展了 fs.File 接口，增加了 ReadDir(n int) ([]fs.DirEntry, error) 方法。这个接口主要用于读取目录文件中的内容，返回一个 fs.DirEntry 列表。它通常用于实现了 fs.ReadDirFS 的文件系统。
  • fs.ReadFileFS 接口扩展了 fs.FS 接口，增加了 ReadFile(name string) ([]byte, error) 方法。这个接口允许直接读取指定文件的全部内容，返回字节切片。 它提供了一种更便捷的方式来读取文件内容，避免了先打开文件再读取的步骤。
  • fs.StatFS 接口也扩展了 fs.FS 接口，增加了 Stat(name string) (fs.FileInfo, error) 方法。该方法用于获取指定文件的元信息，返回一个 fs.FileInfo 对象。
  • fs.SubFS 接口也扩展了 fs.FS 接口，增加了 Sub(dir string) (fs.FS, error) 方法。该方法用于创建一个新的文件系统，它表示原始文件系统的一个子目录。这在需要限制访问文件系统的特定部分时非常有用。
  • fs.WalkDirFunc 类型定义了一个函数签名，用于 fs.WalkDir 函数的回调。

io/fs 包的应用场景

  • 访问不同类型的文件系统： 可以使用相同的代码访问本地文件系统、内存文件系统、zip 文件等。
  • 测试代码： 可以方便地mock文件系统，从而提高代码的可测试性。
  • 嵌入资源： 可以将静态资源嵌入到程序中，并使用 io/fs 包进行访问。

示例代码

示例代码一：fs.FS 接口

fs.FS 接口是 io/fs 包的核心，它表示一个文件系统。最常见的实现是 os.DirFS，它表示本地文件系统的一个目录。

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
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	// 创建一个表示当前目录的文件系统

	fsys := os.DirFS(".")
	// 打开一个文件

	f, err := fsys.Open("README.md")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()
	// 读取文件内容

	data := make([]byte, 100)
	n, err := f.Read(data)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(data[:n]))
}

这个例子展示了如何使用 os.DirFS 创建一个文件系统，然后使用 fsys.Open 方法打开一个文件并读取其内容。

示例代码二：fs.File 接口

fs.File 接口表示一个打开的文件，它提供了读取、写入、关闭等方法。

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
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	f, err := fsys.Open("README.md")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()
	// 获取文件信息

	info, err := f.Stat()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("File size:", info.Size())
}

这个例子展示了如何使用 f.Stat 方法获取文件的元信息。

示例代码三：fs.DirEntry 接口

fs.DirEntry 接口表示一个目录项，它可以是文件或子目录。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	entries, err := fs.ReadDir(fsys, ".")
	if err != nil {
		log.Fatal(err)
	}
	for _, entry := range entries {
		fmt.Println("Name:", entry.Name())
		fmt.Println("Is directory:", entry.IsDir())
	}
}

这个例子展示了如何使用 fs.ReadDir 函数读取目录中的所有条目，并使用 entry.Name 和 entry.IsDir 方法获取条目的名称和类型。

示例代码四：fs.GlobFS 接口

fs.GlobFS 接口扩展了 fs.FS 接口，增加了 Glob 方法，允许使用通配符模式匹配文件和目录。

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

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	if globFS, ok := fsys.(fs.GlobFS); ok {
		matches, err := globFS.Glob("*.go")
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println("Go files:", matches)
	}
}

这个例子展示了如何使用 fs.Glob 函数查找所有以 .go 结尾的文件。

示例代码五：fs.ReadDirFS 接口

fs.ReadDirFS 接口也扩展了 fs.FS 接口，增加了 ReadDir 方法，用于读取指定目录下的所有文件和子目录。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	if readDirFS, ok := fsys.(fs.ReadDirFS); ok {
		entries, err := readDirFS.ReadDir(".")
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println("Directory contents:")
		for _, entry := range entries {
			fmt.Println(entry.Name())
		}
	}
}

这个例子展示了如何使用 fs.ReadDir 函数读取目录中的所有条目。

示例代码六：fs.SubFS 接口

fs.SubFS 接口也扩展了 fs.FS 接口，增加了 Sub 方法，用于创建一个新的文件系统，它表示原始文件系统的一个子目录。

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
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	if subFS, ok := fsys.(fs.SubFS); ok {
		sub, err := subFS.Sub("subdir")
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println("Sub directory contents:")
		entries, err := fs.ReadDir(sub, ".")
		if err != nil {
			log.Fatal(err)
		}
		for _, entry := range entries {
			fmt.Println(entry.Name())
		}
	}
}

这个例子展示了如何使用 fs.Sub 函数创建一个表示子目录的文件系统，并读取其内容。

示例代码七：fs.WalkDirFunc 接口

fs.WalkDirFunc 类型定义了一个函数签名，用于 fs.WalkDir 函数的回调。

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
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

)

func main() {
	fsys := os.DirFS(".")
	err := fs.WalkDir(fsys, ".", func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		fmt.Println("Walking:", path)
		return nil

	})
	if err != nil {
		log.Fatal(err)
	}
}

这个例子展示了如何使用 fs.WalkDir 函数遍历目录，并使用 fs.WalkDirFunc 函数打印每个文件和目录的路径。

那些有趣的文件系统

内存文件系统

内存文件系统是一种虚拟文件系统，它将文件存储在内存中而不是磁盘上。内存文件系统通常用于临时存储数据，或者用于测试和调试目的。
这种文件系统速度非常快，但数据在程序退出后会丢失。Go 语言的 testing/fstest 包提供了一个 MapFS 包，可以方便地创建内存文件系统。

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
 

package main

import (
	"fmt"

	"io/fs"

	"log"

	"os"

	"testing/fstest"

)

func main() {
	// 创建一个内存文件系统

	fsys := fstest.MapFS{
		"file1.txt": {Data: []byte("Hello, world!")},
		"dir1/file2.txt": {Data: []byte("This is file2.")},
	}
	// 打开一个文件

	f, err := fsys.Open("file1.txt")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()
	// 读取文件内容

	data := make([]byte, 100)
	n, err := f.Read(data)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(data[:n]))
    // 遍历文件系统

    err = fs.WalkDir(fsys, ".", func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return err
        }
        fmt.Println("Walking:", path)
        return nil

    })
    if err != nil {
        log.Fatal(err)
    }
}

也有一些第三方的库实现了内存文件系统，比如psanford/memfs,这是一个它的例子：

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
 

package main

import (
	"fmt"

	"io/fs"

	"github.com/psanford/memfs"

)

func main() {
	rootFS := memfs.New()
	err := rootFS.MkdirAll("dir1/dir2", 0777)
	if err != nil {
		panic(err)
	}
	err = rootFS.WriteFile("dir1/dir2/f1.txt", []byte("incinerating-unsubstantial"), 0755)
	if err != nil {
		panic(err)
	}
	err = fs.WalkDir(rootFS, ".", func(path string, d fs.DirEntry, err error) error {
		fmt.Println(path)
		return nil

	})
	if err != nil {
		panic(err)
	}
	content, err := fs.ReadFile(rootFS, "dir1/dir2/f1.txt")
	if err != nil {
		panic(err)
	}
	fmt.Printf("%s\n", content)
}

嵌入式文件系统

嵌入式文件系统将文件嵌入到程序中，这样可以方便地将静态资源打包到程序中。Go 语言标准库提供了一个 embed 包，可以方便地创建嵌入式文件系统。

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
 

package main

import (
	"embed"

	"fmt"

	"io/fs"

	"log"

)
//go:embed static

var staticFiles embed.FS

func main() {
	// 打开一个嵌入的文件

	f, err := staticFiles.Open("static/file1.txt")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()
	// 读取文件内容

	data := make([]byte, 100)
	n, err := f.Read(data)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(data[:n]))
    // 遍历嵌入式文件系统

    err = fs.WalkDir(staticFiles, "static", func(path string, d fs.DirEntry, err error) error {
        if err != nil {
            return err
        }
        fmt.Println("Walking:", path)
        return nil

    })
    if err != nil {
        log.Fatal(err)
    }
}

这个例子展示了如何使用embed.FS类型创建一个嵌入式文件系统，并使用staticFiles.Open` 方法打开一个嵌入的文件。

云存储文件系统

有一些第三方库提供了将 S3 存储桶挂载为本地文件系统的功能，这样我们就可以像访问本地文件一样访问 S3 文件。例如，go-cloud 库就提供了对多种云存储服务的统一访问接口，包括 S3。

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
 

package main

import (
	"context"

	"fmt"

	"io/fs"

	"log"

	"gocloud.dev/blob"

	_ "gocloud.dev/blob/gcs" // 引入 GCS 驱动，如果使用其他云存储服务，请引入相应的驱动

)

func main() {
	// 设置 S3 存储桶 URL

	bucketURL := "gs://my-bucket"

	// 创建一个 blob.Bucket

	bucket, err := blob.OpenBucket(context.Background(), bucketURL)
	if err != nil {
		log.Fatal(err)
	}
	defer bucket.Close()
	// 创建一个 fs.FS

	fsys := blob.NewFS(bucket)
	// 现在可以使用 fsys 进行文件操作

	err = fs.WalkDir(fsys, ".", func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		fmt.Println("Walking:", path)
		return nil

	})
	if err != nil {
		log.Fatal(err)
	}
}

这个例子展示了如何使用 gocloud.dev/blob 包将 google GCS 存储桶挂载为本地文件系统，并使用 fs.WalkDir 函数遍历存储桶中的文件。

扫描全国的公网IP需要多久？

2025-01-27 00:38:47

自从加入百度负责物理网络的监控业务之后，我大部分的都是编写各种各样额度底层的网络程序。业余时间我也是编写一些有趣的网络程序，不仅仅是兴趣，也是为未来的某个业务探索一下技术方案。

而且这次，我想知道，就在我这一个10年前的小mini机器4核机器上，在家庭网络中扫描全国(中国大陆)的所有的公网IP地址需要多少时间。

利用它，我可以知道和全国各省市的运营商、云服务商的联通情况。有没有运营商的出口故障以及IP已没有被运营商或者有关部门劫持。

TL;DR: 一共扫描了3亿个地址(343142912)，当前ping的通的IP 592万个(5923768)，耗时1小时(1h2m57.973755197s)。

这次我重构了以前的一个扫描公网IP的程序。先前的程序使用gopacket收发包，也使用gopacket组装包。但是gopacket很讨厌的的一个地方是它依赖libpcap库，没有办法在禁用CGO的情况下。

事实上利用Go的扩展包icmp和ipv4,我们完全可以不使用gopacket实现这个功能，本文介绍具体的实现。

程序的全部代码在：https://github.com/smallnest/fishfinder

程序的主要架构

程序使用ICMP协议进行探测。

首先它启动一个goroutine解析全国的IP地址。IP地址文件每一行都是一个网段，它对每一个网段解析成一组IP地址，把这组IP地址扔进input channel。

一个发送goroutine从input通道中接收IP地址，然后组装成ICMP echo包发送给每一个IP地址，它只负责发送，发送完所有的地址就返回。

一个接收goroutine处理接收到的ICMP reply 回包，并将结果写入到output channel中。

主程序不断的从output中接收已经有回包的IP并打印到日志中，直到所有的IP都处理完就退出。

这里涉及到并发编程的问题，几个goroutine怎么协调：

  • IP解析和任务分发goroutine和发送goroutine通过input通讯。分发goroutine处理完所有的IP后，就会关闭input通知发送goroutine。
  • 发送goroutine得知input关闭，就知道已经处理完所有的IP,发送完最后的IP后把output关闭。
  • 接收goroutine往output发送接收到回包的IP, 如果output关闭，再往output发送就会panic,程序中捕获了panic。不过还没到这一步主程序应该就退出了。
  • 主程序从output读取IP, 一旦output关闭，主程序就打印统计信息后推出。

如果你对Go并发编程有疑问，可以阅读极客时间上的《Go并发编程实战课》专栏，或者图书《深入理解Go并发编程》。
如果你是Rust程序员，不就我会推出《Go并发编程实战课》姊妹专栏，专门介绍Rust并发编程。
如果你对网络编程感兴趣，今年我还想推出《深入理解网络编程》的专栏或者图书，如果你感兴趣，欢迎和我探讨。

主程序的代码如下：

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
 

package main

import (
	"flag"

	"time"

	"github.com/kataras/golog"

)

var (
	protocol = flag.String("p", "icmp", "The protocol to use (icmp, tcp or udp)")
)
// 嵌入ip.sh

func main() {
	flag.Parse()
	input := make(chan []string, 1024)
	output := make(chan string, 1024)
	scanner := NewICMPScanner(input, output)
	var total int

	var alive int

	golog.Infof("start scanning")
	start := time.Now()
	// 将待探测的IP发送给send goroutine

	go func() {
		lines := readIPList()
		for _, line := range lines {
			ips := cidr2IPList(line)
			input <- ips
			total += len(ips)
		}
		close(input)
	}()
	// 启动 send goroutine

	scanner.Scan()
	// 接收 send goroutine 发送的结果, 直到发送之后5秒结束

	for ip := range output {
		golog.Infof("%s is alive", ip)
		alive++
	}
	golog.Infof("total: %d, alive: %d, time: %v", total, alive, time.Since(start))
}

接下来介绍三个三个主要goroutine的逻辑。

公网IP获取以及任务分发

首先你需要到互联网管理中心下载中国大陆所有的注册的IP网段，这是从亚太互联网络信息中心下载的公网IP信息，实际上可以探测全球的IP,这里以中国大陆的公网IP为例。

通过下面的代码转换成网段信息：

1
2
3
 
#!/bin/bash
wget -c -O- http://ftp.apnic.net/stats/apnic/delegated-apnic-latest | awk -F '|' '/CN/&&/ipv4/ {print $4 "/" 32-log($5)/log(2)}' | cat > ipv4.txt

ipv4.txt文件中是一行行的网段：

1
2
3
4
5
6
7
8
 

1.0.1.0/24


1.0.2.0/23


1.0.8.0/21


1.0.32.0/19


1.1.0.0/24


1.1.2.0/23


1.1.4.0/22

...

数据量不大，我们全读取进来(如果太多的话我们就流式读取了)。
解析每一行的网段，转换成IP地址列表，然后发送给input通道。
等处理完就把inpout通道关闭。

1
2
3
4
5
6
7
8
9
 

go func() {
	lines := readIPList()
	for _, line := range lines {
		ips := cidr2IPList(line)
		input <- ips
		total += len(ips)
	}
	close(input)
}()

发送逻辑

我使用了ICMPScanner结构体来管理发送和接收的逻辑。看名字你也可以猜测到我们将来还可以使用TCP/UDP等协议进行探测。

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
 

type ICMPScanner struct {
	src net.IP
	input  chan []string

	output chan string

}
// 调大缓存区
// sysctl net.core.rmem_max
// sysctl net.core.wmem_max

func NewICMPScanner(input chan []string, output chan string) *ICMPScanner {
	localIP := getLocalIP()
	s := &ICMPScanner{
		input:  input,
		output: output,
		src:    net.ParseIP(localIP),
	}
	return s
}

func (s *ICMPScanner) Scan() {
	go s.recv()
	go s.send(s.input)
}

send方法负责发送ICMP包，recv方法负责接收ICMP包。

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
 
// send sends a single ICMP echo request packet for each ip in the input channel.

func (s *ICMPScanner) send(input chan []string) error {
	defer func() {
		time.Sleep(5 * time.Second)
		close(s.output)
		golog.Infof("send goroutine exit")
	}()
	id := os.Getpid() & 0xffff
	// 创建 ICMP 连接

	conn, err := icmp.ListenPacket("ip4:icmp", s.src.String())
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	// 不负责接收数据

	filter := createEmptyFilter()
	if assembled, err := bpf.Assemble(filter); err == nil {
		conn.IPv4PacketConn().SetBPF(assembled)
	}
    ... // 先忽略，后面再介绍

	return nil

}

send方法中，我们首先创建一个ICMP连接，我通过icmp包创建了一个连接，然后设置了一个BPF过滤器，过滤掉我们不关心的包。
这是一个技巧，这个连接我们不关心接收到的包，只关心发送的包，所以我们设置了一个空的过滤器。

这个设计本来是为了将来的性能扩展做准备，可以创建多个连接用来更快的发送。不过目前我们只使用一个连接，所以这个连接其实可以和接收goroutine共享，目前的设计还是发送和接收使用各自的连接。

接下来就是发送的逻辑了，也就是上面省略的部分：

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
 
seq := uint16(0)

for ips := range input {
	for _, ip := range ips {
		dst, err := net.ResolveIPAddr("ip", ip)
		if err != nil {
			golog.Fatalf("failed to resolve IP address: %v", err)
		}
		// 构造 ICMP 报文

		msg := &icmp.Message{
			Type: ipv4.ICMPTypeEcho,
			Code: 0,
			Body: &icmp.Echo{
				ID:   id,
				Seq:  int(seq),
				Data: []byte("Hello, are you there!"),
			},
		}
		msgBytes, err := msg.Marshal(nil)
		if err != nil {
			golog.Errorf("failed to marshal ICMP message: %v", err)
		}
		// 发送 ICMP 报文

		_, err = conn.WriteTo(msgBytes, dst)
		if err != nil {
			golog.Errorf("failed to send ICMP message: %v", err)
		}
		seq++
	}
}

发送循环从input通道中读取IP地址，然后构造ICMP echo报文，发送到目标地址。

  • 从 input channel 读取 IP 列表
  • 对每个 IP 执行以下操作：
    1. 解析 IP 地址
    2. 构造 ICMP echo 请求报文
    3. 序列化报文
    4. 发送到目标地址

icmp报文中的ID我们设置为进程的PID，在接收的时候可以用来判断是否是我们发送的回包。

接收逻辑

接收逻辑比较简单，我们只需要接收ICMP回包，然后解析出IP地址，然后发送到output通道。

首先我们创建一个ICMP连接，然后循环接收ICMP回包，解析出IP地址，然后发送到output通道。

我们只需处理ICMPTypeEchoReply类型的回包，然后判断ID是否是我们发送的ID，如果是就把对端的IP发送到output通道。

我们通过ID判断回包针对我们的场景就足够了，不用再判断seq甚至payload信息。

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
 

func (s *ICMPScanner) recv() error {
	defer recover()
	id := os.Getpid() & 0xffff
	// 创建 ICMP 连接

	conn, err := icmp.ListenPacket("ip4:icmp", "0.0.0.0")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	// 接收 ICMP 报文

	reply := make([]byte, 1500)
	for {
		n, peer, err := conn.ReadFrom(reply)
		if err != nil {
			log.Fatal(err)
		}
		// 解析 ICMP 报文

		msg, err := icmp.ParseMessage(protocolICMP, reply[:n])
		if err != nil {
			golog.Errorf("failed to parse ICMP message: %v", err)
			continue

		}
		// 打印结果

		switch msg.Type {
		case ipv4.ICMPTypeEchoReply:
			echoReply, ok := msg.Body.(*icmp.Echo)
			if !ok {
				continue

			}
			if echoReply.ID == id {
				s.output <- peer.String()
			}
		}
	}
}

可以看到，200行代码基本就可以我们扫描全国公网IP的程序了。你也可以尝试扫描一下全球的IP地址，看看需要多少时间。

对了，下面是我运行这个程序的输出:

1
2
3
4
5
 
...
[INFO] 2025/01/26 22:01 223.255.236.221 is alive
[INFO] 2025/01/26 22:01 223.255.252.9 is alive
[INFO] 2025/01/26 22:01 send goroutine exit
[INFO] 2025/01/26 22:01 total: 343142912, alive: 5923768, time: 1h2m57.973755197s

Go中秘而不宣的数据结构: 四叉堆，不是普通的二叉堆

2024-11-18 22:47:50

Go语言中Timer以及相关的Ticker、time.After、time.AfterFunc 等定时器最终是以四叉堆的数据形式存放的。

全局的 timer 堆也经历过三个阶段的重要升级。

  • Go 1.9 版本之前，所有的计时器由全局唯一的四叉堆维护，goroutine间竞争激烈。
  • Go 1.10 - 1.13，全局使用 64 个四叉堆维护全部的计时器，通过分片减少了竞争的压力，但是本质上还是没有解决 1.9 版本之前的问题
  • Go 1.14 版本之后，每个 P 单独维护一个四叉堆，避免了goroutine的竞争。 (后面我们再介绍 per-P 的数据结构)

常见的堆(heap)常常以二叉堆的形式实现。可是为什么Go timer使用四叉堆呢？

以最小堆为例，下图展示了二叉堆和四叉堆的区别：

  • 二叉堆：每个节点最多有2个子节点；四叉堆：每个节点最多有4个子节点
  • 在相同节点数下，四叉堆的高度更低，约为二叉堆的一半（log₄n vs log₂n）
  • 对于最小堆来说， 父节点的值小于等于子节点的值。

父节点和子节点的索引计算也略有不同。二叉堆的父子索引如下：

1
2
3
 
parent = (i - 1) // 2

left_child = 2 * i + 1

right_child = 2 * i + 2

四叉堆的父子索引如下：

1
2
3
 
parent = (i - 1) // 4

first_child = 4 * i + 1

last_child = 4 * i + 4

他们的操作时间复杂度:

因为四叉树的高度相对更低，所以四叉堆适合数据量特别大，需要减少树的高度的场景， Go的timer很久以前(11年前)就使用四叉树来实现Timer的保存，当然Go开发者也是根据测试结果选择了四叉树，最早的这个提交可以查看: ## code review 13094043: time: make timers heap 4-ary (Closed)

在Go的运行时中，四叉堆的实现在 src/runtime/time.go 文件中，可以查看源码实现。timers数据结构代表Timer的集合，每个P都有一个timers实例，用于维护当前P的所有Timer。

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
 
// A timers is a per-P set of timers.

type timers struct {
    // 互斥锁保护timers; 虽然timers是每个P的，但是调度器可以访问另一个P的timers，所以我们必须锁定。

	mu mutex
    // heap是一组计时器，按heap[i].when排序。这就是一个四叉堆，虽然没有明确的说明。

    // 必须持有锁才能访问这个堆。

	heap []timerWhen
    // len是heap的长度的原子副本。

	len atomic.Uint32
    // zombies是堆中标记为删除的计时器的数量。

	zombies atomic.Int32
	raceCtx uintptr

    // minWhenHeap是最小的heap[i].when值(= heap[0].when)。

    // wakeTime方法使用minWhenHeap和minWhenModified来确定下一个唤醒时间。

    // 如果minWhenHeap = 0，表示堆中没有计时器。

	minWhenHeap atomic.Int64
    // minWhenModified是具有timerModified位设置的计时器的最小heap[i].when的下界。

    // 如果minWhenModified = 0，表示堆中没有timerModified计时器。

	minWhenModified atomic.Int64
}

type timerWhen struct {
	timer *timer
	when  int64

}

func (ts *timers) lock() {
	lock(&ts.mu)
}

func (ts *timers) unlock() {
	ts.len.Store(uint32(len(ts.heap)))
	unlock(&ts.mu)
}

同时Timer结构体还引用了Timers, 这叫你中有我，我中有你，这样的设计是为了方便Timer的管理，Timer的创建、删除、执行都是通过Timers来实现的。

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

type timer struct {
	mu mutex
	astate atomic.Uint8 
	state  uint8       
	isChan bool       
	blocked uint32

	when   int64

	period int64

	f      func(arg any, seq uintptr, delay int64)
	arg    any
	seq    uintptr

	ts *timers // 注意这里

	sendLock mutex
	isSending atomic.Int32
}

我们来看看对这个堆操作的一些方法。

timerHeapN定义了堆是四叉堆，也就是每个节点最多有4个子节点。

1
 

const timerHeapN = 4

堆常用的辅助方法就是siftUp和siftDown，分别用于上浮和下沉操作。

下面是上浮的方法，我把一些跟踪检查的代码去掉了。整体看代码还是比较简单的，就是不停的上浮，直到找到合适的位置。

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
 
// siftUp将位置i的计时器在堆中合适的位置，通过将其向堆的顶部移动。

func (ts *timers) siftUp(i int) {
	heap := ts.heap
	if i >= len(heap) {
		badTimer()
	}
    // 注意下面两行我们保存了当前i的计时器和它的when值

	tw := heap[i] 
	when := tw.when
	if when <= 0 {
		badTimer()
	}
	for i > 0 {
		p := int(uint(i-1) / timerHeapN) // 父节点 (i-1)/4

		if when >= heap[p].when { // 如果父节点的when <= 当前节点的when，那么就不需要再上浮了

			break

		}
		heap[i] = heap[p] // 父节点下沉到当前的i

		i = p // i指向父节点, 继续循环上浮检查

	}
    // 如果发生了上浮，那么最后将tw放到上浮到的合适位置

	if heap[i].timer != tw.timer {
		heap[i] = tw
	}
}

类似的，下面是下沉的方法：

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
 
// siftDown将位置i的计时器放在堆中的正确位置，通过将其向堆的底部移动。

func (ts *timers) siftDown(i int) {
	heap := ts.heap
	n := len(heap)
	if i >= n {
		badTimer()
	}
    // 如果已经是叶子节点，不用下沉了

	if i*timerHeapN+1 >= n {
		return

	}
    // 保存当前i的计时器和when值

	tw := heap[i]
	when := tw.when
	if when <= 0 {
		badTimer()
	}
    // 从左子节点开始，找到最小的when值，然后将当前节点下沉到这个位置

	for {
		leftChild := i*timerHeapN + 1 // 左子节点

		if leftChild >= n {
			break

		}
		w := when
		c := -1

		for j, tw := range heap[leftChild:min(leftChild+timerHeapN, n)] { // 从左子节点开始遍历子节点,找到小于当前w的最小的子节点

			if tw.when < w {
				w = tw.when
				c = leftChild + j
			}
		}
		if c < 0 { // 如果没有找到比当前节点更小的子节点，那么就不用下沉了

			break

		}
        // 将当前节点下沉到最小的子节点

		heap[i] = heap[c]
		i = c
	}
    // 如果发生了下沉，那么最后将tw放到下沉到的合适位置

	if heap[i].timer != tw.timer {
		heap[i] = tw
	}
}

比上浮略微复杂，因为需要在兄弟节点中找到最小的节点，然后将当前节点下沉到这个位置。

对于一个任意的slice,我们可以把它初始化为一个四叉堆，方法如下：

1
2
3
4
5
6
7
8
9
10
 

func (ts *timers) initHeap() {
	if len(ts.heap) <= 1 {
		return

	}
    
    // 从最后一个非叶子节点开始，依次下沉

	for i := int(uint(len(ts.heap)-1-1) / timerHeapN); i >= 0; i-- {
		ts.siftDown(i)
	}
}

当然timers还有一些辅助timer处理的一些方法，很多和四叉堆没有关系了，我就不一一介绍了，我主要介绍几个和四叉堆相关的方法。

这里吐槽一下，这个time.go文件中代码组织很乱，timer和timers的方法都穿插在一起。理论应该是timer方法和timers方法分开，这样更清晰。或者把timers抽取到一个单独的文件中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
 

func (ts *timers) deleteMin() {
    // 得到堆顶元素

	t := ts.heap[0].timer
	if t.ts != ts {
		throw("wrong timers")
	}
	t.ts = nil // 将timer的ts置为nil，自此和ts一别两宽，再无瓜葛

    // 将最后一个元素设置为堆顶

	last := len(ts.heap) - 1

	if last > 0 {
		ts.heap[0] = ts.heap[last]
	}
	ts.heap[last] = timerWhen{} // 将最后一个元素置为空

	ts.heap = ts.heap[:last] // 缩减slice,剔除最后的空元素

	if last > 0 { // 将堆顶元素下沉

		ts.siftDown(0)
	}
	ts.updateMinWhenHeap()
	if last == 0 {
		ts.minWhenModified.Store(0)
	}
}

增加一个timer到堆中：

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

func (ts *timers) addHeap(t *timer) {
	if netpollInited.Load() == 0 {
		netpollGenericInit()
	}
	if t.ts != nil {
		throw("ts set in timer")
	}
    // 设置timer的ts为当前的timers,从此执子之手，笑傲江湖

	t.ts = ts
    // 添加到最后

	ts.heap = append(ts.heap, timerWhen{t, t.when})
	ts.siftUp(len(ts.heap) - 1) // 上浮它到合适的位置

	if t == ts.heap[0].timer {
		ts.updateMinWhenHeap()
	}
}

n叉堆

d-ary 堆或 d-heap 是一种优先队列数据结构，是二进制堆的泛化，其中节点有d个子节点而不是 2 个子节点。因此，二进制堆是2堆，而三元堆是3堆。根据 Tarjan 和 Jensen 等人的说法，d-ary堆是由 Donald B. Johnson 1975 年发明的。

此数据结构允许比二进制堆更快地执行降低优先级操作(因为深度更浅了)，但代价是删除最小操作速度较慢。这种权衡导致算法的运行时间更长，其中降低优先级操作比删除最小操作更常见。此外，d-ary堆比二进制堆具有更好的内存缓存行为，尽管理论上最坏情况下的运行时间更长，但它们在实践中运行得更快。与二进制堆一样，d-ary堆是一种就地数据结构，除了在堆中存储项目数组所需的存储空间外，它不使用任何额外的存储空间。

在Go生态圈已经有相应的库实现这个数据结构，比如ahrav/go-d-ary-heap,所以如果你有类似场景的需求，或者想对比测试，你可以使用这个库。

导入库:

1
 

import "github.com/ahrav/go-d-ary-heap"

下面的例子是创建三叉最小堆和四叉最大堆的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 

package main

import (
    "fmt"

    "github.com/ahrav/go-d-ary-heap"

)

func main() {
    // Create a min-heap for integers with a branching factor of 3.

    minHeap := heap.NewHeap[int](3, func(a, b int) bool { return a < b })
    // Create a max-heap for integers with a branching factor of 4.

    maxHeap := heap.NewHeap[int](4, func(a, b int) bool { return a > b })
}

往堆中增加元素：

1
2
3
4
5
6
7
 
minHeap.Push(10)
minHeap.Push(5)
minHeap.Push(15)
maxHeap.Push(10)
maxHeap.Push(5)
maxHeap.Push(15)

从堆中移除最值:

1
2
 
fmt.Println(minHeap.Pop()) // Outputs: 5

fmt.Println(maxHeap.Pop()) // Outputs: 15

返回但是不移除最值:

1
2
 
fmt.Println(minHeap.Peek()) // Assuming more elements were added, outputs the smallest

fmt.Println(maxHeap.Peek()) // Assuming more elements were added, outputs the largest

HeapMap, 一个混合功能的数据结构Go语言实现

2024-11-17 17:17:13

今天在准备《秘而不宣》系列下一篇文章时，思绪飘散了，突然想到使用 Heap 的功能再加 HashTable (Map) 的功能，可以构造一种新的数据结构，然后把我聚合程序中的数据聚合数据结构替换掉，总之思绪翩翩。然后在网上搜了一下，这种数据结构其实早就有了，名字叫 HeapMap。

HeapMap (也叫做 PriorityMap) 是一种结合了堆和哈希映射的数据结构，常用于需要按键排序并进行高效查找的场景。它可以在优先级队列的基础上，使用哈希映射来提供快速访问和更新。HeapMap 在实现过程中利用堆的有序性和哈希表的快速查找能力，以支持按键排序和常数时间查找。

Go 语言支付 Rob Pike 在他的 Rob Pike's 5 Rules of Programming 第 5 条就指出：

  • Data dominates. If you've chosen the right data structures and organized things well, the algorithms will almost always be self-evident. Data structures, not algorithms, are central to programming.
    数据为王。如果你选择了合适的数据结构并进行了良好的组织，算法通常会变得显而易见。编程的核心在于数据结构，而非算法。

所以，如果在合适的场景下，针对它的特点，使用 HeapMap 会取得事半功倍的效果。

HeapMap 的主要特点

  1. 堆的特点：HeapMap 内部通过堆来维护键的顺序，可以快速获取最小或最大键。堆提供了插入和删除堆顶元素的 O(log n) 时间复杂度。
  2. 哈希映射的特点：HeapMap 同时使用哈希映射以支持快速查找。哈希映射的查找、插入、删除等操作在理想情况下时间复杂度为 O(1)。
  3. 用途：HeapMap 适合需要频繁按键排序和快速查找的场景，比如带有优先级的缓存、调度系统、任务优先队列等。

HeapMap 的基本结构

  • 堆（Heap）：用来维持按键的顺序，堆可以是最小堆或最大堆，根据具体需求决定。
  • 哈希映射（Map）：用来存储每个键值对，并支持通过键快速查找元素。

你使用一个 container/heap + map 很容易实现一个 HeapMap, 其实我们没必要自己去写一个重复的轮子了，网上其他语言比如 Rust、Java 都有现成的实现，Go 语言中也有一个很好的实现：nemars/heapmap

HeapMap 的实现

nemars/heapmap 这个库是去年增加到 github 中的，我是第一个 star 它的人。我们看看它是怎么实现的。

结构体定义

1
2
3
4
5
6
7
8
9
10
11
 

type Entry[K comparable, V, P any] struct {
	Key      K
	Value    V
	Priority P
	index    int

}

type heapmap[K comparable, V, P any] struct {
	h pq[K, V, P]
	m map[K]*Entry[K, V, P]
}

Entry 代表这个数据结构中的一个节点 (元素、条目) , 它包含 key、value 值，还有优先级，index 记录它在堆的实现数组中的索引。

heapmap 代表 HeapMap 的实现，它包含两个字段，第一个字段其实就是 Heap 的实现，为了方便实现泛型，它就自己实现了一个堆。第二个字段就是一个 map 对象了。

典型的方法

数据结构定义清楚了，那就就可以实现它的方法了。它实现了一些便利的方法，我们值关注几个实现就好了。

Len 方法
1
2
3
 

func (hm *heapmap[K, V, P]) Len() int {
	return len(hm.m)
}

读取h字段或者m字段的长度都可以。

Peek 方法

返回root元素。
最小堆就是返回最小的元素，最大堆就是返回最大的元素。

1
2
3
4
5
6
 

func (hm *heapmap[K, V, P]) Peek() (Entry[K, V, P], bool) {
	if hm.Empty() {
		return Entry[K, V, P]{}, false

	}
	return *hm.h.entries[0], true

}
Pop 方法

弹出root元素。

1
2
3
4
5
6
7
8
 

func (hm *heapmap[K, V, P]) Pop() (Entry[K, V, P], bool) {
	if hm.Empty() {
		return Entry[K, V, P]{}, false

	}
	e := *heap.Pop(&hm.h).(*Entry[K, V, P])
	delete(hm.m, e.Key)
	return e, true

}

注意涉及到元素的删除操作，要同时删除 map 中的元素。

Push 方法 (Set 方法)

其实作者没有实现 Push 方法，而是使用Set 方法来实现的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 

func (hm *heapmap[K, V, P]) Set(key K, value V, priority P) {
	if e, ok := hm.m[key]; ok {
		e.Value = value
		e.Priority = priority
		heap.Fix(&hm.h, e.index)
		return

	}
	e := &Entry[K, V, P]{
		Key:      key,
		Value:    value,
		Priority: priority,
	}
	heap.Push(&hm.h, e)
	hm.m[key] = e
}

Set方法有两个功能。如果元素的Key已经存在，那么就是更新元素，并且根据优先级进行调整。
如果元素的Key不存在，那么就是插入元素。

Get 方法

Get 方法就是获取任意的元素。

1
2
3
4
5
6
7
 

func (hm *heapmap[K, V, P]) Get(key K) (Entry[K, V, P], bool) {
	if e, ok := hm.m[key]; ok {
		return *e, true

	}
	return Entry[K, V, P]{}, false

}

有一点你需要注意的是，这个数据结构不是线程安全的，如果你需要线程安全的话，你可以使用 sync.Mutex/sync.RWMutex 来保护它。

Go中秘而不宣的数据结构 CacheLinePad：精细化优化

2024-11-17 16:19:01

在现代多核处理器中，高效的缓存机制极大地提升了程序性能，而“伪共享”问题却常常导致缓存机制的低效。

1. 背景

cacheline 本文中有时又叫做 缓存行

在现代多核处理器中，三级缓存通常分为三级：L1、L2 和 L3，每一级缓存的大小、速度和共享方式都不同：

  • L1 缓存：这是速度最快的缓存，通常每个 CPU 核心都有独立的 L1 缓存。L1 缓存分为两个部分：一个用于存储指令（L1I），另一个用于存储数据（L1D）。L1 缓存的容量一般较小（通常 32KB - 64KB），但是读取速度极快，以极低的延迟为 CPU 核心提供服务。

  • L2 缓存：L2 缓存通常比 L1 缓存大一些，容量一般在 256KB - 1MB 左右，每个 CPU 核心通常也会有独立的 L2 缓存。虽然 L2 缓存的访问速度比 L1 缓存稍慢，但它仍然显著快于主存。

  • L3 缓存：这是三级缓存中容量最大的，通常在 8MB - 64MB 或更大。L3 缓存往往由所有 CPU 核心共享，并且主要用于减少核心之间的数据传输延迟。L3 缓存的读取速度比 L1、L2 缓存慢，但相对主存依然较快。对于多核处理器，L3 缓存是多核心之间协作的重要纽带。

CPU缓存将数据划分成若干个 cacheline，使得 CPU 访问特定数据时，能以 cacheline 为单位加载或存储数据。cacheline 的大小通常是固定的，x86 架构中常见的 cacheline 大小是 64 字节，而 Apple M 系列等一些 ARM 架构处理器上可能达到 128 字节。

在 CPU 执行程序时，若数据在某级缓存中命中，整个 cacheline 会从该缓存加载到寄存器中；若数据不在 L1 缓存中，则会依次查找 L2、L3 缓存，并最终在主存中查找并加载到缓存。由于 cacheline 是缓存操作的基本单位，每次数据传输都是以 cacheline 为最小粒度的。

比如在 mac mini m2 机器是，我们可以查看此 CPU 的缓存行大小为 128 字节：

Linux 下可以查看另外一台机器的各级别缓存行大小为 64 字节：

1.1 伪共享 (False Sharing)

伪共享 是指多个线程访问同一个 cache line 中的不同变量时，导致频繁的缓存失效（cache invalidation），从而大大降低程序性能。伪共享通常在多线程编程中发生，因为在多个线程中，如果两个或多个线程操作的变量在同一个 cache line 中，但它们并没有真正的共享关系，每个线程对其变量的写操作会导致其他线程的缓存失效。这样，CPU 核心会不断地将数据写回并重新加载，产生了不必要的资源浪费。

设有两个线程，各自操作两个独立的变量 x 和 y：

1
2
3
4
 

type Data struct {
    x int64 // 线程A更新的变量

    y int64 // 线程B更新的变量

}

如果变量 x 和 y 位于同一个 cache line 中，那么线程 A 更新 x 后，线程 B 也会因为缓存失效而重新加载 y，尽管 B 实际上并未使用 x 的值。这种情况下，虽然两个变量并没有直接共享，但每次写操作都会导致另一方的缓存失效，从而形成了伪共享。

1.2 如何避免伪共享？

伪共享会对性能产生严重影响，但可以通过以下几种方法来优化：

  1. 变量对齐（Padding）：将每个变量扩展至一个完整的 cacheline，以防止多个线程访问同一个 cacheline。例如，可以在变量之间添加填充数据来分隔不同的 cacheline (假定 CPU 缓存行是 64 字节)：
1
2
3
4
5
 

type Data struct {
    x int64           // 线程A更新的变量

    _ [7]int64        // 填充7个int64以对齐至64字节的cache line大小

    y int64           // 线程B更新的变量

}
  1. 将变量分散到不同的结构体中：对于经常被多个线程更新的变量，可以考虑将它们分散到不同的结构体，避免同一结构体被多个线程同时频繁更新。
  2. 使用原子变量：在某些情况下，可以使用原子变量进行更新。虽然这不会彻底消除伪共享，但可以减少缓存一致性带来的开销。
  3. 绑定 CPU 核心（CPU Affinity）：可以将线程绑定到指定的 CPU 核心上，从而减少多个线程同时访问同一块缓存的数据的几率。

1.3 单线程的缓存行污染问题

虽然单线程不会出现伪共享的问题，但是单线程程序仍然有一些缓存优化的空间：

  • 避免缓存行污染：在单线程程序中，如果频繁访问的变量分布在不同的 cache line 上，会导致缓存频繁更替，增加缓存开销。优化时可以将频繁使用的数据集中在同一个 cache line 内，减少 CPU 从内存加载数据的频率。
  • 数据布局优化：对于单线程程序，也可以通过调整数据的内存布局，让程序更好地利用缓存。将经常一起访问的数据放在连续的内存中，以提高缓存命中率。
    比如下面一个测试，
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
 

package main

import (
	"testing"

)
// NonAlignedStruct 未对齐的结构体，补充后占24个字节

type NonAlignedStruct struct {
	a byte // 1字节,补齐7字节

	b int64 // 8字节

	c byte // 1字节,补齐7字节

}
// AlignedStruct 已对齐的结构体，补充后占16个字节

type AlignedStruct struct {
	b int64 // 8字节

	a byte // 1字节

	c byte // 1字节

	_ [6]byte // 填充6个字节，总共16个字节

}

const arraySize = 1024 * 1024


var (
	nonAlignedArray [arraySize]NonAlignedStruct
	alignedArray    [arraySize]AlignedStruct
	result          int64

)
// 初始化数组

func init() {
	for i := 0; i < arraySize; i++ {
		nonAlignedArray[i] = NonAlignedStruct{
			a: byte(i),
			b: int64(i),
			c: byte(i),
		}
		alignedArray[i] = AlignedStruct{
			a: byte(i),
			b: int64(i),
			c: byte(i),
		}
	}
}
// BenchmarkNonAligned 测试未对齐结构体的性能

func BenchmarkNonAligned(b *testing.B) {
	var sum int64

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		for j := 0; j < arraySize; j++ {
			sum += nonAlignedArray[j].b // 读取未对齐结构体的字段

		}
	}
	result = sum // 防止编译器优化

}
// BenchmarkAligned 测试已对齐结构体的性能

func BenchmarkAligned(b *testing.B) {
	var sum int64

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		for j := 0; j < arraySize; j++ {
			sum += alignedArray[j].b // 读取已对齐结构体的字段

		}
	}
	result = sum // 防止编译器优化

}

可以看到读取对齐的结构体性能要远远好于未对齐的结构体。


很多高性能的库都会采用 CacheLine 优化的数据结构，比如 Java 生态圈知名的 LMAX Disruptor。 Go 标准库中也有类似的优化，让我们一起来看看它的实现和应用场景。

2. Go 运行时中的 CacheLine

2.1 运行时中的 CacheLinePad

我们支持，Go 语言支持不同的 CPU 架构，不同的 CPU 架构的缓存行的大小也可能不同，Go 语言是如何统一的呢？
方法很简单，就是针对不同的 CPU 架构，定义不同大小的缓存行。

首先定义统一的结构和变量:

1
2
3
4
5
6
 
// CacheLinePad 用来填充结构体，避免伪共享

type CacheLinePad struct{ _ [CacheLinePadSize]byte }
// CacheLineSize 是 CPU 的缓存行大小，不同的 CPU 架构可能不同.
// 目前 Go 运行时没有检测真实的缓存行大小，所以代码实现使用每个 GOARCH 的常量 CacheLinePadSize 作为近似值。

var CacheLineSize uintptr = CacheLinePadSize

然后针对不同的 CPU 架构定义不同的缓存行大小。
比如arm64的CPU, 文件go/src/internal/cpu/cpu_arm64.go中定义了缓存行大小为128字节：

go/src/internal/cpu/cpu_arm64.go
1
2
3
4
 
// CacheLinePadSize is used to prevent false sharing of cache lines.
// We choose 128 because Apple Silicon, a.k.a. M1, has 128-byte cache line size.
// It doesn't cost much and is much more future-proof.

const CacheLinePadSize = 128

比如64bit的龙芯， 缓存行大小是64字节，文件go/src/internal/cpu/cpu_loong64.go中定义了缓存行大小为64字节：

go/src/internal/cpu/cpu_loong64.go
1
2
3
 
// CacheLinePadSize is used to prevent false sharing of cache lines.
// We choose 64 because Loongson 3A5000 the L1 Dcache is 4-way 256-line 64-byte-per-line.

const CacheLinePadSize = 64

又比如x86和amd64的CPU， 缓存行大小是64字节，文件go/src/internal/cpu/cpu_x86.go中定义了缓存行大小为64字节：

go/src/internal/cpu/cpu_x86.go
1
2
3
4
5
 
//go:build 386 || amd64

package cpu

const CacheLinePadSize = 64

所以Go运行时是根据它支持的不同的 CPU 架构，定义不同的缓存行大小，以此来避免伪共享问题。

但是这个数据结构是定义在Go运行时internal库中，不对外暴露，那么我们怎么用的？

2.2 golang.org/x/sys/cpu

没关系，Go的扩展库golang.org/x/sys/cpu中提供了CacheLinePad的定义，我们可以直接使用。

1
 

type CacheLinePad struct{ _ [cacheLineSize]byte }

它的实现和Go运行时中的一样，只是把CacheLinePad暴露出来了，所以我们可以在自己的项目中直接使用。

2.3 Go运行时中的应用场景

在这个系列的上一篇文章中，我们介绍了treap, treap使用在semTable中，semTable是Go运行时中的一个数据结构，用来管理semaphore的等待队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 

type semaRoot struct {
	lock  mutex
	treap *sudog        // root of balanced tree of unique waiters.

	nwait atomic.Uint32 // Number of waiters. Read w/o the lock.

}

var semtable semTable
// Prime to not correlate with any user patterns.

const semTabSize = 251


type semTable [semTabSize]struct {
	root semaRoot
	pad  [cpu.CacheLinePadSize - unsafe.Sizeof(semaRoot{})]byte

}

等并发读取semTable时，由于semTable中的root是一个semaRoot结构体，semaRoot中有mutex，treap等字段，这些字段可能会被不同的CPU核心同时访问，导致伪共享问题。
为了解决伪共享问题，它增加了一个Pad字段，补齐字段的大小到CacheLineSize，这样就可以避免伪共享问题。当然这里可以确定semaRoot的大小不会超过一个CacheLineSize。

mheap 结构体中展示了另外一种场景，将部分字段使用CacheLinePad隔开， 避免arenas字段和上面的字段之间的伪共享问题。

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
 

type mheap struct {
	_ sys.NotInHeap
	lock mutex
	pages pageAlloc // page allocation data structure

	sweepgen uint32 // sweep generation, see comment in mspan; written during STW

	allspans []*mspan // all spans out there

	pagesInUse         atomic.Uintptr // pages of spans in stats mSpanInUse

	pagesSwept         atomic.Uint64  // pages swept this cycle

	pagesSweptBasis    atomic.Uint64  // pagesSwept to use as the origin of the sweep ratio

	sweepHeapLiveBasis uint64         // value of gcController.heapLive to use as the origin of sweep ratio; written with lock, read without

	sweepPagesPerByte  float64        // proportional sweep ratio; written with lock, read without

	reclaimIndex atomic.Uint64
	reclaimCredit atomic.Uintptr
	_ cpu.CacheLinePad // prevents false-sharing between arenas and preceding variables

	
	arenas [1 << arenaL1Bits]*[1 << arenaL2Bits]*heapArena
    ...
}

go/src/runtime/stack.go中stackpool结构体中也使用了CacheLinePad，展示了另外一种用法：

1
2
3
4
 

var stackpool [_NumStackOrders]struct {
	item stackpoolItem
	_    [(cpu.CacheLinePadSize - unsafe.Sizeof(stackpoolItem{})%cpu.CacheLinePadSize) % cpu.CacheLinePadSize]byte

}

因为item的大小不确定，可能小于一个CacheLineSize，也可能大于一个CacheLineSize，所以这里对CacheLinePad求余，只需补充一个小于CacheLineSize的字节即可。

一般软件开发中，我们不需要关心这些细节，但是当我们需要优化性能时，了解这些底层的实现，可以帮助我们更好的理解和优化程序。

Go中秘而不宣的数据结构 Treap：随机化的二叉搜索树

2024-11-17 16:19:00

treap 是一棵二叉树，它同时维护二叉搜索树 （BST） 和堆的属性, 所以由此得名 (tree + heap   ⇒  treap)。

从形式上讲，treap （tree + heap） 是一棵二叉树，其节点包含两个值，一个 key 和一个 priority，这样 key 保持 BST 属性，priority 是一个保持 heap 属性的随机值（至于是最大堆还是最小堆并不重要）。相对于其他的平衡二叉搜索树，treap的特点是实现简单，且能基本实现随机平衡的结构。属于弱平衡树。

treap 由 Raimund Siedel 和 Cecilia Aragon 于 1989 年提出。

treap 通常也被称为“笛卡尔树”，因为它很容易嵌入到笛卡尔平面中：

具体来说，treap 是一种在二叉树中存储键值对 (X,Y) 的数据结构，其特点是：按 X 值满足二叉搜索树的性质，同时按 Y 值满足二叉堆的性质。如果树中某个节点包含值 (X₀,Y₀)，那么：

  • 左子树中所有节点的X值都满足 X ≤ X₀ (BST 属性)
  • 右子树中所有节点的X值都满足 X₀ ≤ X (BST 属性)
  • 左右子树中所有节点的Y值都满足 Y ≤ Y₀ （堆属性。这里以最大堆为例）

在这种实现中，  X是键（同时也是存储在 Treap 中的值），并且  Y称为优先级。如果没有优先级，则 treap 将是一个常规的二叉搜索树。

优先级（前提是每个节点的优先级都不相同）的特殊之处在于：它们可以确定性地决定树的最终结构（不会受到插入数据顺序的影响）。这一点是可以通过相关定理来证明的。
这里有个巧妙的设计：如果我们随机分配这些优先级值，就能在平均情况下得到一棵比较平衡的树（避免树退化成链表）。这样就能保证主要操作（如查找、插入、删除等）的时间复杂度保持在 O(log N) 水平。
正是因为这种随机分配优先级的特点，这种数据结构也被称为"随机二叉搜索树"。

Treap维护堆性质的方法用到了旋转，且只需要进行两种旋转操作，因此编程复杂度较红黑树、AVL树要小一些。

红黑树的操作：
插入
以最大堆为例
给节点随机分配一个优先级，先和二叉搜索树的插入一样，先把要插入的点插入到一个叶子上，然后跟维护堆一样进行以下操作：

  1. 如果当前节点的优先级比父节点大就进行2. 或3. 的操作
  2. 如果当前节点是父节点的左子叶就右旋
  3. 如果当前节点是父节点的右子叶就左旋。

删除

因为 treap满足堆性质，所以只需要把要删除的节点旋转到叶节点上，然后直接删除就可以了。具体的方法就是每次找到优先级最大的子叶，向与其相反的方向旋转，直到那个节点被旋转到了叶节点，然后直接删除。

查找

和一般的二叉搜索树一样，但是由于 treap的随机化结构，Treap中查找的期望复杂度是 O(logn)

以上是 treap 数据结构的背景知识，如果你想了解更多而关于 treap 的知识，你可以参考

Go 运行时的 treap 和用途

在 Go 运行时 sema.go#semaRoot 中，定义了一个数据结构 semaRoot:

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
 

type semaRoot struct {
	lock  mutex
	treap *sudog        // 不重复的等待者(goroutine)的平衡树(treap)的根节点

	nwait atomic.Uint32 // 等待者(goroutine)的数量

}

type sudog struct {
	g *g
	next *sudog
	prev *sudog
	elem unsafe.Pointer // data element (may point to stack)

	acquiretime int64

	releasetime int64

	ticket      uint32

	isSelect bool

	success bool

	waiters uint16

	parent   *sudog // semaRoot binary tree

	waitlink *sudog // g.waiting list or semaRoot

	waittail *sudog // semaRoot

	c        *hchan // channel

}

这是Go语言互斥锁(Mutex)底层实现中的关键数据结构，用于管理等待获取互斥锁的goroutine队列。我们已经知道，在获取 sync.Mutex 时，如果锁已经被其它 goroutine 获取，那么当前请求锁的 goroutine 会被 block 住，就会被放入到这样一个数据结构中 (所以你也知道这个数据结构中的 goroutine 都是唯一的，不重复)。

semaRoot 保存了一个平衡树，树中的 sudog 节点都有不同的地址 (s.elem) ,每个 sudog 可能通过 s.waitlink 指向一个链表，该链表包含等待相同地址的其他 sudog。对具有相同地址的 sudog 内部链表的操作时间复杂度都是O(1).。扫描顶层semaRoot列表的时间复杂度是 O(log n),其中 n 是具有被阻塞goroutine的不同地址的数量（这些地址会散列到给定的semaRoot）。

semaRoot 的 treap *sudog 其实就是一个 treap, 我们来看看它的实现。

增加一个元素（入队）

增加一个等待的goroutine(sudog)到 semaRoot 的 treap 中，如果 lifo 为 true，则将 s 替换到 t 的位置，否则将 s 添加到 t 的等待列表的末尾。

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
 

func (root *semaRoot) queue(addr *uint32, s *sudog, lifo bool) {
	// 设置这个要加入的节点

	s.g = getg()
	s.elem = unsafe.Pointer(addr)
	s.next = nil

	s.prev = nil

	s.waiters = 0

	var last *sudog
	pt := &root.treap
	// 从根节点开始

	for t := *pt; t != nil; t = *pt { // ①

		// 如果地址已经在列表中,则加入到这个地址的链表中

		if t.elem == unsafe.Pointer(addr) {
			// 如果地址已经在列表中，并且指定了先入后出flag,这是一个替换操作

			if lifo {
				// 替换操作

				*pt = s
				s.ticket = t.ticket
				... // 把t的各种信息复制给s

			} else {
				// 增加到到等待列表的末尾

				if t.waittail == nil {
					t.waitlink = s
				} else {
					t.waittail.waitlink = s
				}
				t.waittail = s
				s.waitlink = nil

				if t.waiters+1 != 0 {
					t.waiters++
				}
			}
			return

		}
		last = t
		// 二叉搜索树查找

		if uintptr(unsafe.Pointer(addr)) < uintptr(t.elem) { // ②

			pt = &t.prev
		} else {
			pt = &t.next
		}
	}
	// 为新节点设置ticket.这个ticket是一个随机值，作为随机堆的优先级，用于保持treap的平衡。

	s.ticket = cheaprand() | 1 // ③

	s.parent = last
	*pt = s
	// 根据优先级(ticket)旋转以保持treap的平衡

	for s.parent != nil && s.parent.ticket > s.ticket { // ④

		if s.parent.prev == s {
			root.rotateRight(s.parent) // ⑤

		} else {
			if s.parent.next != s {
				panic("semaRoot queue")
			}
			root.rotateLeft(s.parent) // ⑥

		}
	}
}

① 是遍历 treap 的过程，当然它是通过搜索二叉树的方式实现。 addr就是我们一开始讲的treap的key，也就是 s.elem。
首先检查 addr 已经在 treap 中，如果存在，那么就把 s 加入到 addr 对应的 sudog 链表中，或者替换掉 addr 对应的 sudog。

这个addr, 如果对于sync.Mutex来说，就是 Mutex.sema字段的地址。

1
2
3
4
 

type Mutex struct {
	state int32

	sema  uint32

}

所以对于阻塞在同一个sync.Mutex上的goroutine，他们的addr是相同的，所以他们会被加入到同一个sudog链表中。
如果是不同的sync.Mutex锁，他们的addr是不同的，那么他们会被加入到这个treap不同的节点。

进而，你可以知道，这个rootSema是维护多个sync.Mutex的等待队列的，可以快速找到不同的sync.Mutex的等待队列,也可以维护同一个sync.Mutex的等待队列。
这给了我们启发，如果你有类似的需求，可以参考这个实现。

③就是设置这个节点的优先级，它是一个随机值，用于保持treap的平衡。这里有个技巧就是总是把优先级最低位设置为1，这样保证优先级不为0.因为优先级经常和0做比较，我们将最低位设置为1，就表明优先级已经设置。

④ 就是将这个新加入的节点旋转到合适的位置，以保持treap的平衡。这里的旋转操作就是上面提到的左旋和右旋。稍后看。

移除一个元素（出队）

对应的，还有出对的操作。这个操作就是从treap中移除一个节点，这个节点就是一个等待的goroutine(sudog)。

dequeue 搜索并找到在semaRoot中第一个因addr而阻塞的goroutine。
比如需要唤醒一个goroutine, 让它继续执行(比如直接将锁交给它，或者唤醒它去争抢锁)。

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
 

func (root *semaRoot) dequeue(addr *uint32) (found *sudog, now, tailtime int64) {
	ps := &root.treap
	s := *ps
	for ; s != nil; s = *ps { // ①， 二叉搜索树查找

		if s.elem == unsafe.Pointer(addr) { // ②

			goto Found
		}
		if uintptr(unsafe.Pointer(addr)) < uintptr(s.elem) {
			ps = &s.prev
		} else {
			ps = &s.next
		}
	}
	return nil, 0, 0

Found: // ③

	...
	if t := s.waitlink; t != nil { // ④

		*ps = t
		...
	} else { // ⑤

		// 旋转s到叶节点，以便删除

		for s.next != nil || s.prev != nil {
			if s.next == nil || s.prev != nil && s.prev.ticket < s.next.ticket {
				root.rotateRight(s)
			} else {
				root.rotateLeft(s)
			}
		}
		// ⑤ 删除s

		if s.parent != nil {
			if s.parent.prev == s {
				s.parent.prev = nil

			} else {
				s.parent.next = nil

			}
		} else {
			root.treap = nil

		}
		tailtime = s.acquiretime
	}
	... // 清理s的不需要的信息

	return s, now, tailtime
}

① 是遍历 treap 的过程，当然它是通过搜索二叉树的方式实现。 addr就是我们一开始讲的treap的key，也就是 s.elem。如果找到了，就跳到 Found 标签。如果没有找到，就返回 nil。

④是检查这个地址上是不是有多个等待的goroutine，如果有，就把这个节点替换成链表中的下一个节点。把这个节点从treap中移除并返回。
如果就一个goroutine，那么把这个移除掉后，需要旋转treap，直到这个节点被旋转到叶节点，然后删除这个节点。

这里的旋转操作就是上面提到的左旋和右旋。

左旋 rotateLeft

rotateLeft 函数将以 x 为根的子树左旋，使其变为 y 为根的子树。
左旋之前的结构为 (x a (y b c))，旋转后变为 (y (x a b) c)。

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
 

func (root *semaRoot) rotateLeft(x *sudog) {
	// p -> (x a (y b c))

	p := x.parent
	y := x.next
	b := y.prev
	y.prev = x // ①

	x.parent = y // ②

	x.next = b // ③

	if b != nil {
		b.parent = x // ④

	}
	y.parent = p // ⑤

	if p == nil {
		root.treap = y // ⑥

	} else if p.prev == x { // ⑦

		p.prev = y
	} else {
		if p.next != x {
			throw("semaRoot rotateLeft")
		}
		p.next = y
	}
}

具体步骤：

  • 将 y 设为 x 的父节点(②)，x 设为 y 的左子节点(①)。
  • 将 b 设为 x 的右子节点(③)，并更新其父节点为 x(④)。
  • 更新 y 的父节点为 p(⑤)，即 x 的原父节点。如果 p 为 nil，则 y 成为新的树根(⑥)。
  • 根据 y 是 p 的左子节点还是右子节点，更新对应的指针(⑦)。


左旋为

右旋 rotateRight

rotateRight 旋转以节点 y 为根的树。
将 (y (x a b) c) 变为 (x a (y b c))。

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
 

func (root *semaRoot) rotateRight(y *sudog) {
	// p -> (y (x a b) c)

	p := y.parent
	x := y.prev
	b := x.next
	x.next = y // ①

	y.parent = x // ②

	y.prev = b // ③

	if b != nil {
		b.parent = y // ④

	}
	x.parent = p // ⑤

	if p == nil {
		root.treap = x // ⑥

	} else if p.prev == y { // ⑦

		p.prev = x
	} else {
		if p.next != y {
			throw("semaRoot rotateRight")
		}
		p.next = x
	}
}

具体步骤：

  • 将 y 设为 x 的右子节点(①), x 设为 y 的父节点(②)
  • 将 b 设为 y 的左子节点(③)，并更新其父节点为 y(④)
  • 更新 x 的父节点为 p(⑤)，即 y 的原父节点。如果 p 为 nil，则 x 成为新的树根(⑥)
  • 根据 x 是 p 的左子节点还是右子节点，更新对应的指针(⑦)


右旋为

理解了左旋和右旋，你就理解了出队代码中这一段为什么把当前节点旋转到叶结点中了：

1
2
3
4
5
6
7
8
 
// 旋转s到叶节点，以便删除

for s.next != nil || s.prev != nil {
	if s.next == nil || s.prev != nil && s.prev.ticket < s.next.ticket {
		root.rotateRight(s)
	} else {
		root.rotateLeft(s)
	}
}

整体上看，treap这个数据结构确实简单可维护。左旋和右旋的代码量很少，结合图看起来也容易理解。 出入队的代码也很简单，只是简单的二叉搜索树的操作，加上旋转操作。

这是我介绍的Go秘而不宣的数据结构第三篇，希望你喜欢。你还希望看到Go运行时和标准库中的哪些数据结构呢，欢迎留言。

我会不定期的从关注者列表并点赞文章的同学中选出一位，送出版商和出版社的老师赠送的书，欢迎参与。