1. 简介

最近在学习rust，恰好看到了skywalking的php扩展采用了rust编写。有用过Skywalking/CAT之类监控系统的同学应该知道，这类系统对我们开发工作帮助非常大，能够非常快的帮我们定位到问题的关键，比如说现在有一个api的请求响应非常慢，那我们就可以从系统提供的web ui中查询这个api请求的链路各个节点的耗时，从而精准的定位慢的关键。

但是这类系统搭建起来还是比较繁琐的，对于个人开发者或者一些小公司来说成本比较高，因此我在apache/skywalking-php的基础上对其进行精简和部分增强，去掉其上报到skywalking server的部分，将trace log写入到本地文件，在这个本地文件中会记录以下内容：

1. 调用CURL时，记录开始结束时间以及耗时，如果发生错误会将错误信息记录下来

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "CURL",
	"name": "https://error.blog.fanscore.cn/a/57/",
	"payload": {
		"http_code": "0",
		"query": "k1=v1&k2=k2&k3=v3",
		"curl_error": "Could not resolve host: error.blog.fanscore.cn"
	},
	"start_time": "10:19:03.596", // 时间格式%H:%M:%S%.3f
	"end_time": "10:19:03.602",
	"duration_in_micro": 5988 // 耗时
}

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "CURL",
	"name": "https://blog.fanscore.cn/a/57/",
	"payload": {
		"http_code": "200",
		"curl_error": "",
		"query": "k1=v1&k2=k2&k3=v3"
	},
	"start_time": "10:19:03.602",
	"end_time": "10:19:03.969",
	"duration_in_micro": 366647
}

2. 调用PDO函数时，记录开始结束时间以及耗时，如果发生错误会将错误信息记录下来

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "PDO",
	"name": "__construct",
	"payload": {
		"result": "unknown",
		"dsn": "mysql:host=127.0.0.1;dbname=blog;charset=utf8mb4"
	},
	"start_time": "10:19:03.969",
	"end_time": "10:19:03.980",
	"duration_in_micro": 11175
}
{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "PDO",
	"name": "query",
	"payload": {
		"statement": "select * from article",
		"result": "object(PDOStatement)"
	},
	"start_time": "10:19:03.980",
	"end_time": "10:19:03.985",
	"duration_in_micro": 5471
}
{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "PDO_STATEMENT",
	"name": "fetchAll",
	"payload": {
		"query_string": "select * from article",
		"result": "array(3)"
	},
	"start_time": "10:19:03.985",
	"end_time": "10:19:03.985",
	"duration_in_micro": 25
}

3. 捕获PHP代码中的错误

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "ERROR",
	"name": "E_WARNING: Undefined variable $undefined_value in /Users/orlion/workspace/nginx/www/ptrace/index.php on line 32",
	"payload": {},
	"start_time": "10:19:03.986",
	"end_time": "10:19:03.986",
	"duration_in_micro": 2
}

4. 捕获PHP代码中未捕获的异常

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "EXCEPTION",
	"name": "Exception: test exception in /Users/orlion/workspace/nginx/www/ptrace/index.php on line 34",
	"payload": {
		"trace": "#0 {main}"
	},
	"start_time": "10:19:03.986",
	"end_time": "10:19:03.986",
	"duration_in_micro": 1
}

5. 请求结束后会记录请求开始结束时间、状态码、GET/POST参数

{
	"trace_id": "b89143d7-0fda-43d5-a688-397aef0ee3ef",
	"kind": "URL",
	"name": "/index.php",
	"payload": {
		"$_GET": "{\"a\":\"1\",\"b\":\"2\",\"c\":\"3\"}",
		"$_POST": "[]",
		"method": "GET",
		"status_code": "200"
	},
	"start_time": "10:19:03.595",
	"end_time": "10:19:03.992",
	"duration_in_micro": 397178
}

2. 安装

1. Requirement

• 项目地址: https://github.com/Orlion/minitrace
• php 8.0+
• mac arm64

很遗憾，目前只提供mac arm64版本，后续会编译出linux版本，但因为依赖的phper-framework/phper的库不支持windows，因此短期内恐怕不能提供windows版本了。

1. 进入https://github.com/Orlion/minitrace/releases 下载编译好的扩展二进制文件到本地

2. 假设第一步将扩展下载到了/tmp/minitrace-v0.1.0-macos-arm64.dylib，编辑php.ini配置文件加入以下配置

   [minitrace]
   ;加载我们的扩展
   extension=/tmp/minitrace-v0.1.0-macos-arm64.dylib
   ;将trace数据输出到/tmp/minitrace.log
   minitrace.log_file = /tmp/minitrace.log
   

3. 重启fpm

3. 测试使用

编辑以下php文件

<?php

$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, 'https://error.blog.fanscore.cn/a/57/?k1=v1&k2=k2&k3=v3#aaa');
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($ch);

$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, 'https://blog.fanscore.cn/a/57/?k1=v1&k2=k2&k3=v3#aaa');
curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
$response = curl_exec($ch);

$host = '127.0.0.1';
$db   = 'blog';
$user = 'root';
$pass = '123456';
$charset = 'utf8mb4';
$dsn = "mysql:host=$host;dbname=$db;charset=$charset";
$options = [
    PDO::ATTR_ERRMODE            => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
    PDO::ATTR_EMULATE_PREPARES   => false,
];
$pdo = new PDO($dsn, $user, $pass, $options);
$stm = $pdo->query('select * from article');
$rows = $stm->fetchAll();
foreach($rows as $row) {
    print_r($row);
}


var_dump($undefined_value);

throw new Exception('test exception');
?>

然后在浏览器中请求该文件，打开/tmp/minitrace.log就能看到如下输出：

{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"CURL","name":"https://error.blog.fanscore.cn/a/57/","payload":{"http_code":"0","query":"k1=v1&k2=k2&k3=v3","curl_error":"Could not resolve host: error.blog.fanscore.cn"},"start_time":"10:19:03.596","end_time":"10:19:03.602","duration_in_micro":5988}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"CURL","name":"https://blog.fanscore.cn/a/57/","payload":{"http_code":"200","curl_error":"","query":"k1=v1&k2=k2&k3=v3"},"start_time":"10:19:03.602","end_time":"10:19:03.969","duration_in_micro":366647}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"PDO","name":"__construct","payload":{"result":"unknown","dsn":"mysql:host=127.0.0.1;dbname=blog;charset=utf8mb4"},"start_time":"10:19:03.969","end_time":"10:19:03.980","duration_in_micro":11175}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"PDO","name":"query","payload":{"statement":"select * from article","result":"object(PDOStatement)"},"start_time":"10:19:03.980","end_time":"10:19:03.985","duration_in_micro":5471}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"PDO_STATEMENT","name":"fetchAll","payload":{"query_string":"select * from article","result":"array(3)"},"start_time":"10:19:03.985","end_time":"10:19:03.985","duration_in_micro":25}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"ERROR","name":"E_WARNING: Undefined variable $undefined_value in /Users/orlion/workspace/nginx/www/ptrace/index.php on line 32","payload":{},"start_time":"10:19:03.986","end_time":"10:19:03.986","duration_in_micro":2}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"EXCEPTION","name":"Exception: test exception in /Users/orlion/workspace/nginx/www/ptrace/index.php on line 34","payload":{"trace":"#0 {main}"},"start_time":"10:19:03.986","end_time":"10:19:03.986","duration_in_micro":1}
{"trace_id":"b89143d7-0fda-43d5-a688-397aef0ee3ef","kind":"URL","name":"/index.php","payload":{"$_GET":"{\"a\":\"1\",\"b\":\"2\",\"c\":\"3\"}","$_POST":"[]","method":"GET","status_code":"200"},"start_time":"10:19:03.595","end_time":"10:19:03.992","duration_in_micro":397178}

前言

学习rust有一段时间了，也用rust写了两个小项目，过程中发现一些rust教程在所有权和引用这一章节的讲解还是不够丰富，有很多case没有讲到，对所有权和引用的理解不够深入，这就导致实际应用时经常卡在所有权和引用，后面查阅一些资料在社区请教一些大佬后才理解，因此将最近练习过程中遇到的一些所有权和引用方面的问题总结成本文，分享给大家，帮大家踩踩坑。

1. 所有权

let a = 1;
let b = a; // a拷贝给b
println!("{}", a); // 不会报错

a的值被拷贝给了b，a和b被存储在栈上，无需在堆上分配内存

let a = String::from("a");
let b = a;
println!("{}", a); // 会报错，上一行a的所有权转移给了b，a不能再使用了

新手在这里可能会产生疑问？当执行形如let b = a;这样的代码时，到底什么情况下发生拷贝，什么情况下转移所有权呢？问题的答案其实非常简单:

只要a实现了Copy trait，那么就会拷贝，如果没有实现则转移所有权

那么为什么不能拷贝呢？我们可以以String这个类型为例，String是一个复杂类型，由存储在栈上的堆指针、字符串长度、字符串容量组成。

我们假设这里也是拷贝，那么a和b都会持有这个堆指针，当变量离开作用域后，rust会自动清理堆内存，由于a和b都指向了同一位置，那么会释放两次，这就导致了bug。

因此rust这样解决问题：当a赋值给b后，rust认为a不再有效，因此a离开作用域之后不会二次释放，这就是把所有权从a转移到了b。a被赋值给b之后就失效了，因此不能再使用。

如果String实现了Copy trait，拷贝a给b时，把堆指针指向的数据也复制一遍，同时将新的堆指针给b，那么a和b就不会指向同一个位置，就不会二次释放，自然就不会发生二次释放的bug了。

以下类型实现了Copy trait * 所有整数类型，比如 u32 * 布尔类型，bool，它的值是 true 和 false * 所有浮点数类型，比如 f64 * 字符类型，char * 元组，当且仅当其包含的类型也都是 Copy 的时候。比如，(i32, i32) 是 Copy 的，但 (i32, String) 就不是 * 不可变引用 &T，注意: 可变引用 &mut T 是不可以 Copy的（如果Copy相当于两个指针指向一个位置，又会出现上面的二次释放的问题了）

1.1 结构体

结构体所有权问题比较复杂，这里单独拿出来分析。

先看一个简单的

struct User {
    age:
}

let user1 = User {
    age: 100,
};

let user2 = user1;
println!("{:}", user1); // 会报错，因为User没有实现Copy trait，所以user1的所有权转移给了user2
println!("{:}", user1.sign_in_count); // 会报错，user1已经无法使用了

这里要注意，虽然user1分配在栈上，但它没有实现Copy trait，仍然会发生所有权的转移

再看看一个复杂的

struct User {
    username: String,
    age: i128,
}

let user1 = User {
    username: String::from("user1"),
    age: 100,
};

let user2 = User {
    username: user1.username,
    age: user1.age
};

println!("{}", user1.age); // 不会报错，age发生了copy，而非所有权转移，可以继续使用
println!("{}", user1.username); // 会报错，username发生了所有权的转移
println!("{:}", user1); // 会报错

这里需要注意的是结构体内部的字段发生所有权转移后，会导致结构体本身也无法继续使用。但是其内部发生copy的值还是可以继续使用的，也就是user1.age还能继续使用不会报错的原因。

1.2 Option 所有权转移问题

我们先明确一个规则: 只要Option<T>中的T实现了Copy trait，那么Option<T>就实现了Copy trait

let a = Some(String::from("hello world!"));
let b = a.unwrap();
let c = a.unwrap(); // 这里会报错

我们分析下报错的原因，首先看unwrap的源码

pub const fn unwrap(self) -> T {
    match self {
        Some(val) => val,
        None => unwrap_failed(),
    }
}

从上面可以看到，调用unwrap时，因为Option<String>没有实现Copy trait，所以a发生了所有权转移，a的所有权转移到了unwrap里，所以第二次调用unwrap时就会报错。

解决办法就是调用as_ref/as_mut或者将Option<String>换成Option<&String>，rust中引用默认实现了Copy trait，所以Opiton<&String>不会发生所有权转移 看下as_ref的源码：

pub const fn as_ref(&self) -> Option<&T> {
    match *self {
        Some(ref x) => Some(x),
        None => None,
    }
}

2. 引用

2.1 可变引用

只能可变的引用一个可变变量

let a = 1;
let b = &mut a; // 会报错，无法可变引用一个不可变变量

同一时刻只能存在一个可变引用

let mut a = 1;
let b = &mut a;
*b = 2;
println!("{}", a); // 会报错，可以将a理解成1的一个引用，因为下一行println!("{}", b);所以b这个可变引用的生命周期还未结束，那么此时如果使用a，则违反了可变引用与不可变引用不能同时存在的规则
println!("{}", b);

2.2 解引用

结构体解引用

let user = String::from("user");
let user_ref = &user;
let _user_1 = *user_ref; // 报错

第三行会报错:

error[E0507]: cannot move out of `*user_ref` which is behind a shared reference
  --> src/main.rs:30:19
   |
30 |     let _user_1 = *user_ref;
   |                   ^^^^^^^^^ move occurs because `*user_ref` has type `String`, which does not implement the `Copy` trait

这个报错看到有解释说不能解引用获取到所有权(String没有实现Copy trait只能将user的所有权转移给_user_1)，但是这里将user的所有权转移给_user_1也并不会造成什么错误，所以我猜测是rust编译器限制了不能通过解引用间接转移所有权，只能直接转移。

这里还有个case：let _user_1 = &(*user_ref); 这种写法可以编译通过，猜测是编译器优化直接拷贝的引用，而不是先转移所有权再取引用。

3. 参考资料

• 有关解引用会获得所有权的理解
• 所有权 - Rust语言圣经(Rust Course)

1. 前言

几年前给公司前台业务一个QPS很高的接口做了一个优化，主要请求来源是当前在线用户，接口核心逻辑就是从codis中根据一个数字查询对应的用户id(小于1亿)，这两个数字的映射关系是不变的，可以理解为codis中有一个map[uint32]uint32的映射表，这个映射表只增不改。

因为接口对codis造成压力很大，因此决定在Go内存中将映射关系缓存下来，但由于这个映射表很大所以不能全部缓存中内存。因此结合业务逻辑决定引入了一个支持LRU淘汰策略的uint32 -> uint32的高性能缓存组件。

调研之后发现市面上Go的各种线程安全还支持LRU的缓存都是有锁的，性能可能受限，因此决定根据应用场景自己搞个特殊的缓存组件。

2. 实现原理

首先还是贴一下源码仓库地址: https://github.com/Orlion/intcache

2.1 结构体定义：

type IntCache struct {
	b          uint8
	buckets    [][8]uint64
	lruBuckets []uint32
}

func New(b uint8) *IntCache {
	cap := 1 << b
	return &IntCache{
		b:          b,
		buckets:    make([][8]uint64, cap),
		lruBuckets: make([]uint32, cap),
	}
}

如上图所示，一个IntCache有2^b个bucket和lruBucket，一个bucket有8个K-V对，一个K-V对使用uint64来存储，前32bit存储key，后32bit存储value。一个lruBucket有8个lru值，采用uint32存储，每4bit存储bucket对应的每个K-V对的lru值。

这里你可能会很奇怪为什么lru要单独存储，不要急，继续往下看，读流程时我会详细解释。

2.2 写流程

func (c *IntCache) Set(key uint32, value uint32) {
        if key == 0 && value == 0 {
		panic("key and value can't be 0")
	}
	bucketi := key & (1<<c.b - 1)
	for i := 0; i < 8; i++ {
		e := atomic.LoadUint64(&c.buckets[bucketi][i])
		if e == 0 {
			atomic.StoreUint64(&c.buckets[bucketi][i], uint64(key)<<32|uint64(value))
			c.updLru(bucketi, i)
			return
		}

		if uint32(e>>32) == key {
			e = uint64(key)<<32 | uint64(value)
			atomic.StoreUint64(&c.buckets[bucketi][i], e)
			c.updLru(bucketi, i)
			return
		}
	}

	// find the min lru
	lrus := atomic.LoadUint32(&c.lruBuckets[bucketi])
	var (
		minLru uint32
		mini   int
	)

	for i := 0; i < 8; i++ {
		lru := lrus | 0b1111<<uint32(i)
		if lru < minLru {
			minLru = lru
			mini = i
		}
	}

	atomic.StoreUint64(&c.buckets[bucketi][mini], uint64(key)<<32|uint64(value))
	c.updLru(bucketi, mini)
}

写入步骤如下：

1. key对容量取模，计算出key落到哪个桶里，然后遍历桶中8个槽(K-V对)
2. 如果遍历槽为0，说明这个槽还没有被占用，写入当前槽，并更新lru值为7，其他槽的lru值-1
3. 如果遍历槽的key等于当前的key，则更新这个槽的值，并更新lru值为7，其他槽的lru值-1
4. 如果遍历完后没有空槽也没有命中key，则找到lru值最小的，淘汰掉然后写入新key并更新lru值

2.3 读流程

func (c *IntCache) Get(key uint32) (value uint32, exists bool) {
	bucketi := key & (1<<c.b - 1)
	for i := 0; i < 8; i++ {
		e := atomic.LoadUint64(&c.buckets[bucketi][i])
                if e == 0 {
			break
		}

		if uint32(e>>32) == key {
			value = uint32(e)
			exists = true
			c.updLru(bucketi, i)
			break
		}
	}

	return
}

读取步骤如下：

1. key对容量取模，计算出key落到哪个桶里，然后遍历桶中8个槽(K-V对)
2. 如果遍历到的槽为0，说明后面的槽都是没有数据的，无需继续遍历
3. 如果遍历到的槽的key等于查询的key，则返回value，并更新lru值

3. 总结

3.1 为什么lru要单独存储

每个bucket占用8*8=64B，正好是一个x86 cpu cacheline的大小，刚好填满一个cacheline，这样遍历bucket上8个槽实际只需要读取第一个槽时访问一次内存，后续访问都会直接从cpu cache中读到（当然前提是没有写请求造成cacheline过期），这样可以充分利用cpu缓存。

如果lru值与bucket存储在一起，那么系统中大量的读请求修改lru值就会造成cacheline过期的可能性就会变大，而如果分开存储，读请求不会造成cacheline过期。

你可能会问频繁的写入也会造成cacheline过期影响性能啊，但是我们这是一个典型的读多写少的系统，而且大量的bucket也降低了cacheline过期的几率。

3.2 缺陷

3.2.1 适用场景有限

由于我这个组件用在了在线用户访问的场景中，我将bucket数量设置为日活人数/8，hash冲突的几率还是比较小的，从监控看缓存命中率还是比较可观的。

但是由于不是严格的LRU，因此其他业务场景可能不适用。

3.2.2 value与lru值的更新不是原子的

因为要提高cpu cache命中率，因此value更新与lru更新是分离的，无法做到原子性，这也是很不严谨的，但是我们这个业务场景中不需要严谨的lru，所以可以忽略。

4. 基准测试

与fastcache对比的基准测试代码

func BenchmarkIntcache(b *testing.B) {
	rand.Seed(1)
	var B uint8 = 21
	m := New(B)
	for i := 0; i < b.N; i++ {
		intcacheBenchmarkFn(m)
	}
}

func BenchmarkFastcache(b *testing.B) {
	rand.Seed(1)
	m := fastcache.New(1 << 21 * (64 + 8))
	for i := 0; i < b.N; i++ {
		fastcacheBenchmarkFn(m)
	}
}

func intcacheBenchmarkFn(m *IntCache) {
	wg := &sync.WaitGroup{}
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			for j := 0; j < 300; j++ {
				key := rand.Uint32()
				m.Set(key, key)
				m.Get(key)
			}
			wg.Done()
		}()
	}
	wg.Wait()
}

func fastcacheBenchmarkFn(m *fastcache.Cache) {
	wg := &sync.WaitGroup{}
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			for j := 0; j < 300; j++ {
				key := rand.Uint32()
				b := make([]byte, 4) // uint64的大小为8字节
				binary.LittleEndian.PutUint32(b, key)
				m.Set(b, b)
				r := make([]byte, 4)
				m.Get(r, b)
			}
			wg.Done()
		}()
	}
	wg.Wait()
}

都是1000个协程并发读写300次，结果：

goos: darwin
goarch: arm64
pkg: github.com/Orlion/intcache
BenchmarkIntcache
BenchmarkIntcache-10                  14          78865301 ns/op
BenchmarkFastcache
BenchmarkFastcache-10                 10         113746746 ns/op
PASS
ok      github.com/Orlion/intcache      9.767s

可以看到我们这个实现要快一点。

1. 前言

书接上回 《统计一个数字二进制位1的个数》，现在我们已经知道如何快速计算出一个int64数字的二进制位1的个数，那么回到我们最初的需求，我们的目的是快速统计一个bitmap中二进制位1的个数，假设我们使用[]uint64来实现bitmap，那么如果要统计这个bitmap中二进制位1的个数，我们可以遍历每个元素，计算出每个uint64元素二进制位1的个数，最后加起来，代码大概如下：

type Bitmap []uint64

func (bitmap Bitmap) OnesCount() (count int) {
	for _, v := range bitmap {
		count += OnesCount64(v)
	}

	return
}

const m0 = 0x5555555555555555 // 01010101 ...
const m1 = 0x3333333333333333 // 00110011 ...
const m2 = 0x0f0f0f0f0f0f0f0f // 00001111 ...

// 计算出x中二进制位1的个数，该函数上篇文章有详细解释，看不懂可以再回去看下
func OnesCount64(x uint64) int {
	const m = 1<<64 - 1
	x = x>>1&(m0&m) + x&(m0&m)
	x = x>>2&(m1&m) + x&(m1&m)
	x = (x>>4 + x) & (m2 & m)
	x += x >> 8
	x += x >> 16
	x += x >> 32
	return int(x) & (1<<7 - 1)
}

这种实现方式在bitmap元素过多，切片长度过长的情况下，计算十分耗时。那么如何优化这段代码呢？

2. 优化

现代CPU一般都支持SIMD指令，通过SIMD指令可以并行执行多个计算，以加法运算为例，如果我们要计算{A0,A1,A2,A3}四个数与{B0,B1,B2,B3}的和，不使用SIMD指令的话，需要挨个计算A0+B0、A1+B1、A2+B2、A3+B3的和。使用SIMD指令的话，可以将{A0,A1,A2,A3}和{A0,A1,A2,A3}四个数加载到xmm(128bit)/ymm(256bit)/zmm(512bit)寄存器中，然后使用一条指令就可以同时计算对应的和。这样理论上可以获得N倍的性能提升。

我们可以采用SIMD指令将OnesCount64函数并行化，并行计算4个uint64数字的结果，代码实现如下：

在popcnt.go文件中定义SimdPopcntQuad函数

package popcnt

func SimdPopcntQuad(nums [4]uint64) [4]uint64

在popcnt.s文件中我们使用汇编实现SimdPopcntQuad函数

#include "textflag.h"

TEXT ·SimdPopcntQuad(SB),NOSPLIT,$0-64
    VMOVDQU nums+0(FP), Y0 // Y0 = x，将四个uint64数字加载到Y0寄存器
    MOVQ $0x5555555555555555, AX
    MOVQ AX, X9
    VPBROADCASTQ X9, Y5 // Y5 = m0 // 上面三行代码将4个m0加载到Y5寄存器
    MOVQ $0x3333333333333333, AX
    MOVQ AX, X9
    VPBROADCASTQ X9, Y6 // Y6 = m1 // 上面三行代码将4个m1加载到Y6寄存器
    MOVQ $0x0f0f0f0f0f0f0f0f, AX
    MOVQ AX, X9
    VPBROADCASTQ X9, Y7 // Y7 = m2 // 上面三行代码将4个m2加载到Y7寄存器
    MOVQ $0x7f, AX
    MOVQ AX, X9
    VPBROADCASTQ X9, Y8 // Y8 = m；上面三行代码将4个m3加载到Y8寄存器
    VPSRLQ $1, Y0, Y1 // Y1 = x>>1；Y0寄存器上四个uint64数字并行右移1位
    VPAND Y1, Y5, Y1 // Y1 = x>>1&m0；Y1寄存器上四个uint64数字并行与Y5寄存器上的四个m0并行与，结果存到Y1寄存器
    VPAND Y0, Y5, Y2 // Y2 = x&m0
    VPADDQ Y1, Y2, Y0 // x = x>>1&m0 + x&m0
    VPSRLQ $2, Y0, Y1 // Y1 = x>>2
    VPAND Y1, Y6, Y1 // Y1 = x>>2&m1
    VPAND Y0, Y6, Y2 // Y2 = x&m1
    VPADDQ Y1, Y2, Y0 // x = x>>2&m1 + x&m1
    VPSRLQ $4, Y0, Y1 // Y1 = x>>4
    VPAND Y1, Y7, Y1 // Y1 = x>>4&m2
    VPAND Y0, Y7, Y2 // Y2 = x&m2
    VPADDQ Y1, Y2, Y0 // x = x>>2&m2 + x&m2
    VPSRLQ $8, Y0, Y1 // Y1 = x >> 8
    VPADDQ Y1, Y0, Y0 // x += x >> 8
    VPSRLQ $16, Y0, Y1 // Y1 = x >> 16
    VPADDQ Y1, Y0, Y0 // x += x >> 16
    VPSRLQ $32, Y0, Y1 // Y1 = x >> 32
    VPADDQ Y1, Y0, Y0 // x += x >> 32
    VPAND Y0, Y8, Y0 // x & (1<<7-1)
    VMOVDQU Y0, ret+32(FP) // 将结果加载到内存中返回值的位置
    RET

Benchmark

理论上讲如此优化之后我们应该可以获得四倍的性能提升，所以我们写个基准测试验证下：

// 优化之后的并行计算测试
func BenchmarkSimdPopcntQuad(b *testing.B) {
        // 使用随机数防止编译阶段被编译器预先计算出来
	rand.Seed(time.Now().UnixNano())
	nums := [4]uint64{rand.Uint64(), rand.Uint64(), rand.Uint64(), rand.Uint64()}
	for i := 0; i < b.N; i++ {
		SimdPopcntQuad(nums)
	}
}

// 优化之前的顺序计算测试
func BenchmarkSerial(b *testing.B) {
        // 使用随机数防止编译阶段被编译器预先计算出来
	rand.Seed(time.Now().UnixNano())
	nums := [4]uint64{rand.Uint64(), rand.Uint64(), rand.Uint64(), rand.Uint64()}
	for i := 0; i < b.N; i++ {
		serialPopcntQuad(nums)
	}
}

func serialPopcntQuad(nums [4]uint64) [4]uint64 {
	return [4]uint64{uint64(bits.OnesCount64(nums[0])), uint64(bits.OnesCount64(nums[1])), uint64(bits.OnesCount64(nums[2])), uint64(bits.OnesCount64(nums[3]))}
}

运行后结果如下

# go test -bench=. -v
=== RUN   TestSimdPopcntQuad
--- PASS: TestSimdPopcntQuad (0.00s)
goos: linux
goarch: amd64
pkg: github.com/Orlion/popcnt
cpu: Intel Core Processor (Broadwell, no TSX)
BenchmarkSimdPopcntQuad
BenchmarkSimdPopcntQuad-8        3693530               330.8 ns/op
BenchmarkSerial
BenchmarkSerial-8               539924296                2.232 ns/op
PASS
ok      github.com/Orlion/popcnt        2.993s

可以看到优化后的并行计算比原始的顺序计算慢了150倍😭，失败~

3. 分析

虽然优化失败了，但是我们还是要分析复盘下其中的原因，从中汲取一些经验，下面我们从两方面来分析下。

3.1 未优化函数为什么快?

首先我们可以看到未优化的函数serialPopcntQuad计算四个数字竟然只花了2ns，根据Numbers Everyone Should Know一文，访存的时间大概是100ns，这就有点离谱了，计算竟然不从内存加载我们的参数？

下面我们写段main函数，使用随机数来调用下serialPopcntQuad函数，然后反汇编看下汇编代码分析下。

func main() {
	rand.Seed(time.Now().UnixNano())
	nums := [4]uint64{rand.Uint64(), rand.Uint64(), rand.Uint64(), rand.Uint64()}
	results := serialPopcntQuad(nums)
	fmt.Println(results)
}

func serialPopcntQuad(nums [4]uint64) [4]uint64 {
	return [4]uint64{uint64(bits.OnesCount64(nums[0])), uint64(bits.OnesCount64(nums[1])), uint64(bits.OnesCount64(nums[2])), uint64(bits.OnesCount64(nums[3]))}
}

编译后反汇编：

从汇编代码中可以看到在调用bits.OnesCount64之前会判断cpu是否支持popcnt指令，如果支持则使用popcnt指令来计算而不是调用bits.OnesCount64来计算，恰好我机器支持popcnt指令，省略了bits.OnesCount64中的一堆计算，因此计算速度非常快。

3.2 优化后为什么慢?

正如3.1中所提到的，相较于cpu计算，访存的代价是非常高的，大概是100ns，而我们汇编代码中为了使用SIMD指令实现统计算法有大量的访存操作。

受限于本人对汇编掌握程度，上面的汇编代码质量应该是很差的，并不能证明SIMD性能差，可能有性能更高的实现，请各位大佬指点。

而且当前Go汇编在不指定编译参数的情况下只能采用旧函数调用约定，必须采用内存传参，所以导致最终基准测试的结果很差。

4. 收获

这一通瞎折腾虽然最终结果失败，但还是有很多收获的。首先真实的体会到了访存有多慢，所以日后在进行性能优化时就会注意这一点，尽量使代码能命中CPU缓存。

再一个就是之前并没有使用过SIMD指令，也没有接触过这种级别的优化，这次算是入门了。

后端选手，水平有限，各位计算机科学家见笑了。

5. 参考资料

1. 玩转SIMD指令编程

最近一个需求需要使用golang实现一个兼容redis的无压缩的bitmap，需要提供一个bitcoun函数来统计这个bitmap中二进制位1的个数，查了一圈并没有找到类似的第三方库，因此决定自己实现一个.（利用一切机会造轮子

1. 问题简化

问题本质实际就是给定一个数字，比如一个二进制数10101101，计算出这个数字中二进制位1的个数，对于10101101这个数字来说它有5个位为1，即：10101101

对于这个问题，最简单的办法就是挨位数，不过这个办法太笨了，没有逼格。

那么有没有银弹呢？答案是肯定的，而且还不止一种。

2. 查表法

对于一个8位的数字来说，它只有256个值，因此完全可以预先计算好每个值的二进制位1个个数写入到映射表中，使用时直接查询这张映射表即可。

伪代码如下所示：

var count1map = map[uint8]uint8 {
    0b0000_0000: 0,
    0b0000_0001: 1,
    ...
    0b1111_1111: 8,
}

func bitcount(x uint8) uint8 {
    return count1map[x]
}

3. 移位法

查表法虽然可以应对8位这样值数量有限的数字，但是对于uint64 or int64这样64位的数字来说，它的值数量是非常多的，我们无法在内存中维护这样巨大的映射表，因此不能使用查表法来解决

Golang在bits包中提供一个OnesCount64(x uint64) int的函数，可以计算一个64位数字中二进制为1的个数，其源码如下：

const m0 = 0x5555555555555555 // 01010101 ...
const m1 = 0x3333333333333333 // 00110011 ...
const m2 = 0x0f0f0f0f0f0f0f0f // 00001111 ...
const m3 = 0x00ff00ff00ff00ff // etc.
const m4 = 0x0000ffff0000ffff

func OnesCount64(x uint64) int {
	const m = 1<<64 - 1
	x = x>>1&(m0&m) + x&(m0&m)
	x = x>>2&(m1&m) + x&(m1&m)
	x = (x>>4 + x) & (m2 & m)
	x += x >> 8
	x += x >> 16
	x += x >> 32
	return int(x) & (1<<7 - 1)
}

初看起来是有点懵逼的，一顿位运算操作怎么就能把1的个数算出来了呢？

这段代码注释中标明其来源于Hacker’s Delight第5章

别着急，我们还是采用自底向上的思想来拆解下。

3.1 2位数字二进制位1的个数

我们先想一下如何计算2位的数字二进制位1的个数，答案是非常简单的：

func OnesCount2(x uint2) int {
    return (x & 0b01) + ((x >> 1) & 0b01)
}

x & 0b01就是求第0位是不是1，((x >> 1) & 0b01)就是求第1位是不是1，加起来就是x这个2位数字二进制位1的个数。

3.2 4位数字二进制位1的个数

对于一个4位数字，如1011，我们先按照3.1中的算法分别求出第3位与第2位即10 和 第1位与第0位即11的二进制位1的个数，然后再加起来就得出这个4位数字的二进制位1的个数了。

伪代码如下所示：

func OnesCount4(x uint4) int {
    x = x & 0b0101 + x >> 1 & 0b0101
    return x & 0b0011 + x >> 2 & 0b0011
}

计算过程如图：

3.3 8位数字二进制位1的个数

8位数字计算过程与4位计算过程本质是相同的，都是拆解组合，伪代码如下：

func OnesCount8(x uint8) int {
    x = x & 0b01010101 + x >> 1 & 0b01010101
    x = x & 0b00110011 + x >> 2 & 0b00110011
    return x & 0b00001111 + x >> 4 && 0b00001111
}

计算过程如下：

64位数字重复这个过程即可，回头看golang的代码应该就可以看懂了，这里就不再详细解释了。

另外这个算法过程还可以进一步优化，详细可以参考下：计算汉明权重的SWAR（SIMD within a Register）算法 感兴趣的可以研究一下，这里就不赘述了。

4. POPCNT指令

一些较新的CPU上支持POPCNT指令，可以通过硬件直接进行计算，Golang代码示例如下：

main.go文件

package main

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

func main() {
    rand.Seed(time.Now().Unix())
    for i := 0; i < 100; i++ {
        var num = rand.Uint64()
        if popcnt(num) != bits.OnesCount64(num) {

            panic(fmt.Sprintf("i: %d, popcnt(%b) = %d, bits.OnesCount64(%b) = %d", i, num, popcnt(num), num, bits.OnesCount64(num)))
        }
    }
    fmt.Println("ok")
}

func popcnt(x uint64) int

amd64.s 文件

#include "textflag.h"

TEXT main·popcnt(SB), NOSPLIT, $0-8
    MOVQ x+0(FP), AX // 将参数x移到AX寄存器
    BYTE $0xf3; BYTE $0x48; BYTE $0x0f; BYTE $0xb8; BYTE $0xc0  // 计算二进制X中1的个数，golang编译器不支持POPCNT指令，这行对应于POPCNT AX, AX
    MOVQ AX, ret+8(FP) // 将结果存入ret
    RET

1. 问题与背景

最近公司内部准备尝试使用下腾讯的TDSQL，因此组内同学写了一段很简单的查询TDSQL的go web程序，使用ab对其进行一个简单压测以获取TDSQL的性能表现，go代码如下：

package main

import (
    "crypto/md5"
    "database/sql"
    "fmt"
    "log"
    "math/rand"
    "net/http"
    "strconv"
    "time"

    "github.com/gin-gonic/gin"
    _ "github.com/go-sql-driver/mysql"
)

func main() {
    r := gin.New()
    r.Use(gin.Logger())
    r.GET("/test", func(c *gin.Context) {
        c.JSON(200, gin.H{
            "message": "test",
        })
    })

    dbconnect, err := sql.Open("mysql", "user:passwd@tcp(10.43.0.43:3306)/dbname")
    if err != nil {
        panic(err)
    }
    dbconnect.SetMaxIdleConns(5)
    dbconnect.SetMaxOpenConns(10)
    dbconnect.SetConnMaxLifetime(time.Hour)
    dbconnect.SetConnMaxIdleTime(time.Hour)
    r.GET("tdsql_test", func(context *gin.Context) {
        muid := fmt.Sprintf("%x", md5.Sum([]byte(strconv.Itoa(rand.Intn(1000000000000)))))
        rows, err := dbconnect.Query(fmt.Sprintf("select muid from rtb_channel_0 where muid='%s'", muid))
        if err != nil {
            log.Fatal(err)
            context.JSON(http.StatusInternalServerError, gin.H{
                "error_code": -3,
            })
            return
        }

        rows.Close()

        context.JSON(http.StatusOK, gin.H{
            "error_code": 0,
            "error_msg":  muid,
            "data":       "result",
        })
    })

    r.Run(":9000")
}

ab压测命令如下：

ab -c 10 -n 500000 "http://127.0.0.1:9000/tdsql_test"

压测开始不久之后代码log.Fatal(err)就打印出了错误信息并退出了:

dial tcp 10.43.0.43:3306: connect: cannot assign requested address

这段错误信息是说无法连接到10.43.0.43:3306 ，原因是无法分配请求地址号，就是说本地端口号都被占用了。那么我们就开始进行排查，端口号究竟是被谁占满的？

2. 排查过程

2.1 通过netstat命令查看端口都被谁占用

netstat -nta | grep 10.43.0.43

有如下输出： 可以看到有大量处于TIME_WAIT状态的TCP连接，这些连接占用了大量的端口。那么这些TIMI_WAIT状态的TCP连接是从哪来的呢？

为了弄清楚这个问题，我们必须知道TIME_WAIT状态是怎么回事。

2.1.1 TIME_WAIT

上图是经典的TCP四次挥手断开连接的过程。可以看到在四次挥手的过程中，主动关闭连接的一端在收到对端发送的FIN包之后会进入TIME_WAIT状态，会等待2MSL之后才能真正关闭连接。

MSL: 最长报文段寿命（Maximum Segment Lifetime）,是一个工程值(经验值)，RFC标准是2分钟，不过有点太长了，一般是30秒，1分钟，2分钟。

为什么客户端TIME_WAIT状态等待2MSL呢？四次握手最后一步客户端向服务端响应ACK，后有两种情况：

1. 服务端没有收到ACK，这时服务端会超时重传FIN
2. 服务端收到了ACK，但是客户端不知道服务端有没有收到

无论1还是2，客户端都需要等待，要取这两种情况等待的最大值以应对最坏情况的发生，这个最坏的情况就是： 去向ACK消息的最大生存时间(MSL) + 来向FIN消息的最大生存时间(MSL) 。可以看到加起来正好是2MSL。等待2MSL，客户端就可以放心大胆的释放TCP连接了，此时可以使用该端口号连接任何服务器。

如果没有TIME_WAIT，新连接直接复用该连接占用的端口话，恰好回复的ACK包没有达到对端，导致对方重传FIN包，这时新连接就会被错误的关闭。

2.1.2 使用了连接池为什么还会出现大量的TIME_WAIT连接呢

首先大量的TIME_WAIT连接说明了我们的go程序建立了大量的连接然后又关闭了，但是理论上使用了连接池连接都应该得到复用，不会建立大量的连接才对。

这时我首先检查了是不是连接池的ConnMaxLifetime和ConnMaxIdleTime设置的太小导致连接被关闭。我回看了代码发现同事设置了一个小时的时长，那么就不可能是这个原因了。

然后我将怀疑的矛头指向了TDSQL，因为TDSQL是我们首次使用，之前使用Mysql时也没有遇到过这个问题，会不会是TDSQL发送/回复了某个特殊的包导致了客户端主动断开呢？

2.2 验证是否是TDSQL的问题

为了验证上述的猜想，使用tcpdump在服务器上抓了个包

tcpdump -i any host 10.43.0.43 -w output.pcap

然后将抓到的outout.pcap包down下来后丢到本机wireshark中进行分析，选择一个端口过滤下可以看到整个tcp连接的所有包。

可以看到TDSQL发送的都是正常的mysql协议包，并没有什么特殊的包，因此到这里基本可以确认不是TDSQL的锅。

那么排查的重点又回到了golang连接池，golang连接池为什么会主动断开连接？

2.3 golang为什么会主动断开连接？

由于golang整个sql包非常复杂，我们可以自底向上的思路来排查问题，首先我们找到mysql驱动包go-sql-driver/mysql中关闭连接的函数：

func (mc *mysqlConn) Close() (err error)

它位于connection.go文件中。

下面我们使用dlv 来启动上面的go程序

$ dlv debug main.go

进入到dlv控制台，然后在控制台中输入break mysql.(*mysqlConn).Close在这个函数上打个断点,继续输入c 命令让程序继续执行，然后使用ab命令进行这个web程序进行压测。

不出所料程序断在了mysql.(*mysqlConn).Close。然后我们使用bt命令来打印下调用栈：

可以清楚的看到整个核心调用链是：

rwos.Close -> driverConn.releaseConn -> DB.putConn -> driverConn.Close -> mysqlConn.Close

问题的关键是DB.putConn ,我们可以分析下源码：

func (db *DB) putConn(dc *driverConn, err error, resetSession bool) {
    ...
    added := db.putConnDBLocked(dc, nil)
    db.mu.Unlock()

    if !added {
        dc.Close()
        return
    }
}

func (db *DB) putConnDBLocked(dc *driverConn, err error) bool {
    if db.closed {
        return false
    }
    if db.maxOpen > 0 && db.numOpen > db.maxOpen {
        return false
    }
    if c := len(db.connRequests); c > 0 {
        var req chan connRequest
        var reqKey uint64
        for reqKey, req = range db.connRequests {
            break
        }
        delete(db.connRequests, reqKey) // Remove from pending requests.
        if err == nil {
            dc.inUse = true
        }
        req <- connRequest{
            conn: dc,
            err:  err,
        }
        return true
    } else if err == nil && !db.closed {
        if db.maxIdleConnsLocked() > len(db.freeConn) { // db.maxIdleConnsLocked()取自db.maxIdleCount
            db.freeConn = append(db.freeConn, dc)
            db.startCleanerLocked()
            return true
        }
        db.maxIdleClosed++
    }
    return false
}

从源码中我们可以得知是DB.putConnDBLocked返回了false导致了连接关闭。为了验证这一点可以继续使用dlv 在dc.Close()这一行打个断点，然后重新压测这个程序： 可以看到程序确实是走到了dc.Close()这一行，我们继续打印上下文的数据：

到这里我们就知道了是由于db.maxIdleCount == len(db.freeConn)导致了连接没有被复用。

db.maxIdleCount是我们代码中设置的dbconnect.SetMaxIdleConns(5)也就是5

那么问题的原因其实就很简单了，我们设置了最大闲置连接数为5，最大可建立连接数为10，那么进程中最多可出现10个连接，这10个连接中只有5个可以被丢回到连接池中复用，而另外5个连接由于超过了我们设置最大闲置连接数5所以不会被丢回到连接池中复用，因此使用完就close了。当并发高的情况下就会出现大量的连接打开与关闭。

3. 解决

最大闲置连接数设置成一个大于等于最大连接数的值即可，比如下面这样：

dbconnect.SetMaxIdleConns(10)
dbconnect.SetMaxOpenConns(10)

原文地址:https://blog.fanscore.cn/a/53/

1. 前言

本文是与世界分享我刚编的转发ntunnel_mysql.php的工具的后续，之前的实现有些拉胯，这次重构了下。需求背景是为了在本地macbook上通过开源的mysql可视化客户端(dbeaver、Sequel Ace等)访问我司测试环境的mysql，整个测试环境的如图所示:

那么就有以下几种方式:

• 客户端直连mysql #Pass# 测试环境mysql只提供了内网ip，只允许测试环境上的机器连接，因此不可行
• 通过ssh隧道连接 #Pass# 测试环境机器倒是可以ssh上去，但是只能通过堡垒机接入，且堡垒机不允许ssh隧道，因此不可行
• navicat http隧道连接 #Pass# 测试环境有机器提供了公网ip开放了http服务，因此技术上是可行的，但navicat非开源免费软件，我司禁止使用，因此不可行
• 测试环境选一台机器建立mysql代理转发请求 #Pass# 测试环境机器只开放了80端口，且已被nginx占用，因此不可行
• 内网穿透 这个想法很好，下次不要再想了

既然上面的方式都不行，那怎么办呢？因此我产生了一个大胆的想法

2. 一个大胆的想法

大概架构如下

首先，在本地pc上启动一个sidecar进程，该进程监听3306端口，实现mysql协议，将自己伪装为一个mysql server。本地pc上的mysql客户端连接到sidecar，发送请求数据包给sidecar，从sidecar读取响应包。

然后在测试环境某台机器上启动transport进程，该进程启动http服务，由nginx代理转发请求，相当于监听在80端口，然后连接到测试环境的mysql server。

sidecar会将来自客户端的请求包通过http请求转发给transport，transport将请求包转发到测试环境对应的mysql server，然后读取mysql的响应数据包，然后将响应数据包返回给sidecar，sidecar再将响应包返回给mysql客户端。

遵循上述的基本原理，我将其实现出来: https://github.com/Orlion/hersql。但是在描述hersql的实现细节之前我们有必要了解下mysql协议

3. mysql协议

mysql客户端与服务端交互过程主要分为两个阶段：握手阶段与命令阶段。交互流程如下：

在最新版本中，握手过程比上面要复杂，会多几次交互

3.1 握手阶段

在握手阶段，3次握手建立tcp连接后服务端会首先发送一个握手初始化包，包含了 * 协议版本号：指示所使用的协议版本。 * 服务器版本：指示MySQL服务器版本的字符串。 * 连接ID：在当前连接中唯一标识客户端的整数。 * 随机数据：包含一个随机字符串，用于后续的身份验证。 * 服务器支持的特性标志：指示服务器支持的客户端功能的位掩码。 * 字符集：指示服务器使用的默认字符集。 * 默认的身份验证插件名（低版本没有该数据）

随后客户端会发送一个登录认证包，包含了：

• 协议版本号：指示所使用的协议版本。
• 用户名：用于身份验证的用户名。
• 加密密码：客户端使用服务端返回的随机数对密码进行加密
• 数据库名称：连接后要使用的数据库名称。
• 客户端标志：客户端支持的功能的位掩码。
• 最大数据包大小：客户端希望接收的最大数据包大小。
• 字符集：客户端希望使用的字符集。
• 插件名称：客户端希望使用的身份验证插件的名称。

服务端收到客户端发来的登录认证包验证通过后会发送一个OK包，告知客户端连接成功，可以转入命令交互阶段

在mysql 8.0默认的身份验证插件为caching_sha2_password，低版本为mysql_native_password，两者的验证交互流程有所不同个，caching_sha2_password在缓存未命中的情况下还会多几次交互。另外如果服务端与客户端的验证插件不同的话，也是会多几次交互。

3.2 命令阶段

在命令阶段，客户端会发送命令请求包到服务端。数据包的第一个字节标识了当前请求的类型，常见的命令有：

• COM_QUERY命令，执行SQL查询语句。
• COM_INIT_DB命令，连接到指定的数据库。
• COM_QUIT命令，关闭MySQL连接。
• COM_FIELD_LIST命令，列出指定表的字段列表。
• COM_PING命令，向MySQL服务器发送PING请求。
• COM_STMT_系列预处理语句命令

请求响应的模式是客户端会发一个请求包，服务端会回复n(n>=0)个响应包

最后客户端断开连接时会主动发送一个COM_QUIT命令包通知服务端断开连接

4. hersql数据流转过程

在了解mysql协议之后我们就可以来看下hersql的数据流转过程了。

transport连接mysql server时必须要知道目标数据库的地址与端口号(mysql client连接的是sidecar)，所以hersql要求mysql client需要在数据库名中携带目标数据库的地址与端口号。

transport发给mysql server的登录请求包中需要包含用mysql server发来的随机数加密之后的密码，但是mysql client给到sidecar的登录请求包中的密码是用sidecar给的随机数加密的，因此无法直接拿来使用，所以hersql要求mysql client需要在数据库名中携带密码原文，transport会用mysql server给的随机数进行加密， 这也是hersql的局限。

5. hersql使用

上面介绍了一堆原理性的东西，那么如何使用呢？

5.1 在一台能够请求目标mysql server的机器上部署hersql transport

首先你需要下载下来hersql的源码：https://github.com/Orlion/hersql，还需要安装下golang，这些都完成后你就可以启动hersql transport了。但是先别着急，我先解释下transport的配置文件tranport.example.yaml:

server:
  # transport http服务监听的地址
  addr: :8080

log:
  # 标准输出的日志的日志级别
  stdout_level: debug
  # 文件日志的日志级别
  level: error
  # 文件日志的文件地址
  filename: ./storage/transport.log
  # 日志文件的最大大小（以MB为单位）, 默认为 100MB。日志文件超过此大小会创建个新文件继续写入
  maxsize: 100
  # maxage 是根据文件名中编码的时间戳保留旧日志文件的最大天数。 
  maxage: 168
  # maxbackups 是要保留的旧日志文件的最大数量。默认是保留所有旧日志文件。
  maxbackups: 3
  # 是否应使用 gzip 压缩旋转的日志文件。默认是不执行压缩。
  compress: false

你可以根据你的需求修改配置，然后就可以启动transport了

$ go run cmd/transport/main.go -conf=transport.example.yaml

一般情况下都是会先编译为可执行文件，由systemd之类的工具托管transport进程，保证transport存活。这里简单期间直接用go run起来

5.2 在你本地机器部署启动hersql sidecar

同样的，你需要下载下来hersql的源码：https://github.com/Orlion/hersql，提前安装好golang。修改下sidecar的配置文件sidecar.example.yaml:

server:
  # sidecar 监听的地址，之后mysql client会连接这个地址
  addr: 127.0.0.1:3306
  # transport http server的地址
  transport_addr: http://x.x.x.x:xxxx
log:
  # 与transport配置相同

就可以启动sidecar了

$ go run cmd/sidecar/main.go -conf=sidecar.example.yaml

同样的，一般情况下也都是会先编译为可执行文件，mac上是launchctl之类的工具托管sidecar进程，保证sidecar存活。这里简单期间直接用go run起来

5.3 客户端连接

上面的步骤都执行完成后，就可以打开mysql客户端使用了。数据库地址和端口号需要填写sidecar配置文件中的addr地址，sidercar不会校验用户名和密码，因此用户名密码可以随意填写

重点来了: 数据库名必须要填写，且必须要按照以下格式填写

[username[:password]@][protocol[(address)]]/dbname[?param1=value1&...&paramN=valueN]

举个例子：

root:123456@tcp(10.10.123.123:3306)/BlogDB

如图所示：

5.4 举个例子

目标mysql服务器

• 地址：10.10.123.123:3306
• 数据库：BlogDB
• 用户名：root
• 密码：123456

可以直连目标mysql服务器的机器

• 地址：10.10.123.100
• 开放端口：8080

那么transport可以配置为

server:
  addr: :8080

sidecar可以配置为

server:
  addr: 127.0.0.1:3306
  transport_addr: http://10.10.123.100:8080

客户端连接配置

• 服务器地址：127.0.0.1
• 端口: 3306
• 数据库名root:123456@tcp(10.10.123.123:3306)/BlogDB

5.5 局限

hersql目前只支持mysql_native_password的认证方式，mysql8默认的认证方式是caching_sha2_password，所以如果要通过hersql连接mysql8需要注意登录用户的认证方式是否是mysql_native_password，如果是caching_sha2_password那暂时是无法使用的。

6. 参考资料

• MySQL协议分析
• MySQL: Client/Server Protocol

如果hersql对你有帮助欢迎点个star

背景

最近接到一个需求，开发中使用了redis georadius命令取附近给定距离内的点。完工后对服务进行压测后发现georadius的性能比预期要差，因此我分析了georadius的源码，并对原始的实现方案进行了优化，总结成了本文。

我们生产环境使用的redis版本为4.0.13，因此本文redis源码皆为4.0.13版本的源码

redis geo原理

往redis中添加坐标的命令是GEOADD key longitude latitude member [longitude latitude member ...]，实际上redis会将经纬度转成一个52bit的整数作为zset的score，然后添加到zset中，所以实际上redis geo底层就是个zset，你甚至可以直接使用zset的命令来操作一个geo类型的key。

那么经纬度是如何转成52bit整数的呢？业内广泛使用的方法是首先对经纬度分别按照二分法编码，然后将各自的编码交叉组合成最后的编码。我们以116.505021, 39.950898这个坐标为例看下如何编码：

• 第一次二分操作，把经度分为两个区间：[-180,0)和[0,180]，116.505021落在右区间，因此用1表示第一次编码后的值
• 第二次二分操作，把[0,180]分为两个区间[0,90)和[90,180]，116.505021落在右区间，因此用1表示第二次编码后的值
• 第三次二分操作，把[90,180]分为两个区间[90,135)和[135,180]，116.505021落在左区间，因此用0表示第二次编码后的值
• 按照这种方法依次处理，做完5次后，得到经度值的5位编码值：11010

• 按照同样的方法对纬度值进行编码，得到纬度值的5位编码值：10111

然后将经度编码11010和纬度编码值10111交叉得到最终geohash值1110011101

通常会使用base32将编码值转成字符串表示的hash值，与本文无关这里不多做介绍

根据如上的算法通常可以直观的写出如下的代码：

// 该代码来源于https://github.com/HDT3213/godis/blob/master/lib/geohash/geohash.go
func encode0(latitude, longitude float64, bitSize uint) ([]byte, [2][2]float64) {
	box := [2][2]float64{
		{-180, 180}, // lng
		{-90, 90},   // lat
	}
	pos := [2]float64{longitude, latitude}
	hash := &bytes.Buffer{}
	bit := 0
	var precision uint = 0
	code := uint8(0)
	for precision < bitSize {
		for direction, val := range pos {
			mid := (box[direction][0] + box[direction][1]) / 2
			if val < mid {
				box[direction][1] = mid
			} else {
				box[direction][0] = mid
				code |= bits[bit]
			}
			bit++
			if bit == 8 {
				hash.WriteByte(code)
				bit = 0
				code = 0
			}
			precision++
			if precision == bitSize {
				break
			}
		}
	}
	if code > 0 {
		hash.WriteByte(code)
	}
	return hash.Bytes(), box
}

可以看到基本就是上述算法的实际描述，但是redis源码中却是另外一种算法：

int geohashEncode(const GeoHashRange *long_range, const GeoHashRange *lat_range,
                  double longitude, double latitude, uint8_t step,
                  GeoHashBits *hash) {
    // 参数检查此处代码省略
    ...
    
    double lat_offset =
        (latitude - lat_range->min) / (lat_range->max - lat_range->min);
    double long_offset =
        (longitude - long_range->min) / (long_range->max - long_range->min);

    lat_offset *= (1 << step);
    long_offset *= (1 << step);
    // lat_offset与long_offset交叉
    hash->bits = interleave64(lat_offset, long_offset);
    return 1;
}

那么该如何理解redis的这种算法呢？我们假设经度用3位来编码 可以看到编码值从左到右实际就是从000到111依次加1递进的，给定的经度值在这条线的位置(偏移量)就是其编码值。假设给定经度值为50，那么它在这条线的偏移量就是(50 - -180) / (180 - -180) * 8 = 5即101

georadius原理

georadius命令格式为GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]，以给定的经纬度为中心， 返回键包含的位置元素当中， 与中心的距离不超过给定最大距离的所有位置元素。

首先需要明确一点的是并非两个坐标点编码相近其距离越近，以上图为例，虽然A所在区块的编码与C所在区块编码较之B更相近，但实际B点距离A点更近。为了避免这种问题redis中会先计算出给定点东南西北以及东北、东南、西北、西南八个区块以及自己身所在的区块即九宫格区域内所有坐标点，然后计算与当前点的距离，再进一步筛选出符合距离条件的点。

假设要查附近100km的点，那么要保证矩形的边长要大于100km，才能保证能获取到所有符合条件的点，地球半径约6372.797km，第一次分割后可以得到四个东西长6372.797*π，南北长3186.319*π，继续切割：

分割到第七次时南北长49.787*π，如果再切分长度为24.894*π，长度小于100km，因此停止分割，所以如果要查附近100km的点，我们需要的精度为7

redis中根据给定的距离估算出需要的精度的代码如下

const double MERCATOR_MAX = 20037726.37;

uint8_t geohashEstimateStepsByRadius(double range_meters, double lat) {
    if (range_meters == 0) return 26;
    int step = 1;
    while (range_meters < MERCATOR_MAX) {
        range_meters *= 2;
        step++;
    }
    step -= 2;
    // 高纬度地区地球半径小因此适当降低精度
    if (lat > 66 || lat < -66) {
        step--;
        if (lat > 80 || lat < -80) step--;
    }

    if (step < 1) step = 1;
    if (step > 26) step = 26;
    return step;
}

调用encode0函数就能计算出给定点在step = geohashEstimateStepsByRadius()精度级别所在矩形区域的geohash值。接下来计算该矩形区域附近的八个区域。

...
// 调用encode0函数计算geohash
geohashEncode(&long_range,&lat_range,longitude,latitude,steps,&hash);
// 计算出附近八个区域
geohashNeighbors(&hash,&neighbors);
...

一个区域的东侧区域只要将经度的编码值+1即可，反之西侧区域只要将经度编码值-1即可，北侧区域只要将纬度的编码值+1即可，南侧区域只要将纬度的编码值-1即可。对应redis源码如下:

void geohashNeighbors(const GeoHashBits *hash, GeoHashNeighbors *neighbors) {
    neighbors->east = *hash;
    neighbors->west = *hash;
    neighbors->north = *hash;
    neighbors->south = *hash;
    neighbors->south_east = *hash;
    neighbors->south_west = *hash;
    neighbors->north_east = *hash;
    neighbors->north_west = *hash;
    // 纬度加1就是东侧区域
    geohash_move_x(&neighbors->east, 1);
    geohash_move_y(&neighbors->east, 0);
    // 纬度减1就是西侧区域
    geohash_move_x(&neighbors->west, -1);
    geohash_move_y(&neighbors->west, 0);
    // 精度减1就是南侧区域
    geohash_move_x(&neighbors->south, 0);
    geohash_move_y(&neighbors->south, -1);

    geohash_move_x(&neighbors->north, 0);
    geohash_move_y(&neighbors->north, 1);

    geohash_move_x(&neighbors->north_west, -1);
    geohash_move_y(&neighbors->north_west, 1);

    geohash_move_x(&neighbors->north_east, 1);
    geohash_move_y(&neighbors->north_east, 1);

    geohash_move_x(&neighbors->south_east, 1);
    geohash_move_y(&neighbors->south_east, -1);

    geohash_move_x(&neighbors->south_west, -1);
    geohash_move_y(&neighbors->south_west, -1);
}

如上图所示，当给定点在中心区域的东北侧时，西北、西、西南、南、东南五个方向的区域中的所有点距离给定点肯定超过了给定距离，所以可以过滤掉，redis代码如下所示：

if (steps >= 2) {
    if (area.latitude.min < min_lat) {
        GZERO(neighbors.south); // 南侧区域置零，过滤南侧区域
        GZERO(neighbors.south_west);
        GZERO(neighbors.south_east);
    }
    if (area.latitude.max > max_lat) {
        GZERO(neighbors.north);
        GZERO(neighbors.north_east);
        GZERO(neighbors.north_west);
    }
    if (area.longitude.min < min_lon) {
        GZERO(neighbors.west);
        GZERO(neighbors.south_west);
        GZERO(neighbors.north_west);
    }
    if (area.longitude.max > max_lon) {
        GZERO(neighbors.east);
        GZERO(neighbors.south_east);
        GZERO(neighbors.north_east);
    }
}

计算出区块后下一步就需要将九宫格区域中的所有坐标点拿出来，依次计算与给定点的距离，然后过滤出符合给定距离的点

// 遍历九宫格内所有点，依次计算与给定点的距离，然后过滤出符合给定距离的点添加到ga中
int membersOfAllNeighbors(robj *zobj, GeoHashRadius n, double lon, double lat, double radius, geoArray *ga) {
    GeoHashBits neighbors[9];
    unsigned int i, count = 0, last_processed = 0;
    int debugmsg = 1;

    neighbors[0] = n.hash;
    neighbors[1] = n.neighbors.north;
    neighbors[2] = n.neighbors.south;
    neighbors[3] = n.neighbors.east;
    neighbors[4] = n.neighbors.west;
    neighbors[5] = n.neighbors.north_east;
    neighbors[6] = n.neighbors.north_west;
    neighbors[7] = n.neighbors.south_east;
    neighbors[8] = n.neighbors.south_west;

    // 遍历九宫格
    for (i = 0; i < sizeof(neighbors) / sizeof(*neighbors); i++) {
        ...
        // 当给定距离过大时，区块可能会重复
        if (last_processed &&
            neighbors[i].bits == neighbors[last_processed].bits &&
            neighbors[i].step == neighbors[last_processed].step)
        {
            continue;
        }
        // 取出宫格内所有点，依次计算距离，符合条件后添加到ga中
        count += membersOfGeoHashBox(zobj, neighbors[i], ga, lon, lat, radius);
        last_processed = i;
    }
    return count;
}

int membersOfGeoHashBox(robj *zobj, GeoHashBits hash, geoArray *ga, double lon, double lat, double radius) {
    GeoHashFix52Bits min, max;
    // 根据区块的geohash值计算出对应的zset的score的上下限[min,max]
    scoresOfGeoHashBox(hash,&min,&max);
    // 取出底层的zset中的[min,max]范围内的元素，依次计算距离，符合条件后添加到ga中
    return geoGetPointsInRange(zobj, min, max, lon, lat, radius, ga);
}

georadius优化

从上一节中可以看到，给定距离范围越大，则九宫格区域越大，九宫格区域内的点就越多，而每个点都需要计算与中间点的距离，距离计算又涉及到大量的三角函数计算，所以这部分计算是十分消耗CPU的。又因为redis工作线程是单线程的，因此无法充分利用多核，无法通过增加redis server的CPU核数来提升性能，只能添加从库。

距离计算算法及优化可以看下美团的这篇文章: https://tech.meituan.com/2014/09/05/lucene-distance.html

对于这个问题，我们可以将九宫格以及距离计算部分提升到我们的应用程序即redis客户端来进行，步骤如下： * 在客户端计算出九宫格区域，然后转为zset score的范围 * 使用zrangebyscore命令从redis取出score范围内的所有点 * 遍历所有点依次计算与给定点的距离，筛选出符合距离条件的点

陌陌好像也是使用了这种方案：https://mp.weixin.qq.com/s/DL2P49y4R1AE2MIdkxkZtQ

由于我们使用golang进行开发，因此我将redis中的georadius部分代码转为了golang代码，并整理成一个库开源在了github：https://github.com/Orlion/go-georadius

原本的写法是:

client.GeoRadius(key, longitude, latitude, &redis.GeoRadiusQuery{
	Radius:    1000,
	Unit:      "m", // 距离单位
	Count:     1,          // 返回1条
	WithCoord: true,       // 将位置元素的经纬度一并返回
	WithDist:  true,       // 一并返回距离
})

改造后：

ga := make([]redis.Z, 0)
ranges := geo.NeighborRanges(longitude, latitude, 1000)
for _, v := range ranges {
    zs, _ := client.ZRangeByScoreWithScores(key, redis.ZRangeBy{
		Min: strconv.Itoa(int(v[0])),
		Max: strconv.Itoa(int(v[1])),
	}).Result()
	for _, z := range zs {
	    dist := geox.GetDistanceByScore(longitude, latitude, uint64(z.Score))
		if dist < 1000 {
		    ga = append(ga, z)
		}
	}
}

压测结果对比

43w坐标点，取附近50000m（九宫格内有14774点，符合条件的点约6000个）

50km优化前

Concurrency Level:      5
Time taken for tests:   89.770 seconds
Complete requests:      5000
Failed requests:        0
Write errors:           0
Total transferred:      720000 bytes
HTML transferred:       0 bytes
Requests per second:    55.70 [#/sec] (mean)
Time per request:       89.770 [ms] (mean)
Time per request:       17.954 [ms] (mean, across all concurrent requests)
Transfer rate:          7.83 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.0      0       0
Processing:    23   90  10.7     90     159
Waiting:       23   89  10.7     89     159
Total:         23   90  10.7     90     159

Percentage of the requests served within a certain time (ms)
  50%     90
  66%     93
  75%     96
  80%     97
  90%    102
  95%    107
  98%    111
  99%    116
 100%    159 (longest request)

50km优化后

Concurrency Level:      5
Time taken for tests:   75.447 seconds
Complete requests:      5000
Failed requests:        0
Write errors:           0
Total transferred:      720000 bytes
HTML transferred:       0 bytes
Requests per second:    66.27 [#/sec] (mean)
Time per request:       75.447 [ms] (mean)
Time per request:       15.089 [ms] (mean, across all concurrent requests)
Transfer rate:          9.32 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.0      0       0
Processing:    21   75  14.2     75     159
Waiting:       21   75  14.1     75     159
Total:         21   75  14.2     75     159

Percentage of the requests served within a certain time (ms)
  50%     75
  66%     80
  75%     84
  80%     86
  90%     92
  95%     98
  98%    104
  99%    111
 100%    159 (longest request)

可以看到性能并没有巨大的提升，我们减小距离范围到5km（符合条件的点有130个）再看下压测结果

5km优化前

Concurrency Level:      5
Time taken for tests:   14.006 seconds
Complete requests:      5000
Failed requests:        0
Write errors:           0
Total transferred:      720000 bytes
HTML transferred:       0 bytes
Requests per second:    356.99 [#/sec] (mean)
Time per request:       14.006 [ms] (mean)
Time per request:       2.801 [ms] (mean, across all concurrent requests)
Transfer rate:          50.20 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.0      0       0
Processing:     2   14   5.5     12      33
Waiting:        2   14   5.5     12      33
Total:          2   14   5.5     12      34

Percentage of the requests served within a certain time (ms)
  50%     12
  66%     16
  75%     19
  80%     20
  90%     22
  95%     23
  98%     27
  99%     28
 100%     34 (longest request)

5km优化后

Concurrency Level:      5
Time taken for tests:   16.661 seconds
Complete requests:      5000
Failed requests:        0
Write errors:           0
Total transferred:      720000 bytes
HTML transferred:       0 bytes
Requests per second:    300.11 [#/sec] (mean)
Time per request:       16.661 [ms] (mean)
Time per request:       3.332 [ms] (mean, across all concurrent requests)
Transfer rate:          42.20 [Kbytes/sec] received

Connection Times (ms)
              min  mean[+/-sd] median   max
Connect:        0    0   0.0      0       0
Processing:     3   17   5.8     16      66
Waiting:        3   16   5.8     16      66
Total:          3   17   5.8     16      66

Percentage of the requests served within a certain time (ms)
  50%     16
  66%     20
  75%     21
  80%     22
  90%     24
  95%     26
  98%     28
  99%     30
 100%     66 (longest request)

可以看到当优化后性能更差了

猜测造成这个结果的原因应该是附近5km九宫格内的点比较少，所以优化后实际没减少多少距离计算，但多了n(n<=9)倍的请求数，多了额外的命令解析与响应内容的消耗，因此这种优化方案仅仅适用于附近点特别多的情况。

参考资料

• 美团-地理空间距离计算优化
• Golang 实现 Redis(9): 使用GeoHash 搜索附近的人
• 日请求量过亿，谈陌陌的Feed服务优化之路
• GeoHash核心原理解析

Connection reset by peer

网络编程中某一端可能会产生Connection reset by peer的报错，这是因为收到了对端发送的RST包。RST包是在tcp异常关闭时发出的，产生的情形很多。 我在写cat-agent 也遇到了这个报错，经排查发现是客户端发送数据后没有读取服务端的响应而直接关闭了连接（之所以不读取就关闭是因为客户端为提高发送效率不care服务端返回，后来为避免大量的RST包选择服务端不回包来解决）

EPOLL_CLOEXEC

epoll_create1() flags中有个EPOLL_CLOEXEC，类似的还有open()函数的O_CLOEXEC，它们的用处在于：当父进程fork出子进程之后，子进程会继承父进程的文件描述符，当子进程执行exec系统调用之后保存文件描述符的变量就不存在了，这些继承过来的文件描述符将无法关闭。 xxx_CLOEXEC选项就可以标识当子进程执行exec系统调用之后就内核自动关闭继承过来的文件描述符。

syscall.Forklock

在go标准库和其他一些网络库创建socket通常会有如下的代码: syscall.ForkLock.RLock() s, err = socketFunc(family, sotype, proto) if err == nil { syscall.CloseOnExec(s) } syscall.ForkLock.RUnlock() syscall.CloseOnExec(s)与上一节的xxx_CLOEXEC选项作用的相同的。这里ForkLock的作用是保证fork操作时socket的创建与设置CloseOnExec原子性。

持续更新中

背景

先说背景吧，我司目前是通过phpMyAdmin来操作mysql，phpMyAdmin每次都要输入账号密码，而且我们的数据库实例又非常多，每次登录还要先搜索一番。虽然搞了一些chrome插件来简化这个过程，但是个人还是不太满意，所以考虑使用桌面客户端来代替。 然而测试环境的mysql是无法直连的，虽然有些桌面客户端（比如navicat）提供了ssh隧道或者http代理来连接，但是我们开发机只开放了http server的端口，没有开放ssh（只能通过跳板机连接）。所以一些桌面客户端直连和通过ssh隧道来连接是不可能了，http代理方式倒是可以连接，但是貌似只有navicat提供了这个功能，而navicat是收费的。得，所有常规道路都不通了。 只能靠骚操作了

方案

1、内网穿透！把内网ssh服务暴露到公网去！

如上图 1. 首先开发机上agent进程同时连接代理服务器proxy server与本地ssh，做双向转发 2. 然后代理服务器proxy server监听来自我的mac上的桌面客户端的ssh隧道连接，做该连接和proxy server <=> agent这个tcp连接的双向转发 3. 这样mac上的桌面客户端就能建立到内网开发机的ssh隧道了 很好！我们现在就能通过ssh隧道连接到测试环境mysql了！ 不过，过不了多久我们可能会遇到下面的情况↓↓↓

2、用http隧道代理tcp连接进而在80端口上提供ssh服务

先简单提下http隧道的原理吧： 1. 首先客户端会发放CONNECT请求给代理服务器，告诉代理服务器它要连接的目标ip+port 2. 代理服务器建立与目标ip+port的tcp连接然后返回给客户端HTTP 200 Connection Established报文 3. 这时候就建立起了连接，后续客户端到代理服务器的所有tcp报文都会被代理服务器转发到目标ip+port的tcp连接上 > 详情可以参考这篇文章：HTTP 隧道代理原理和实现 这个方案存在以下两个问题： 第一、80端口被nginx监听，而nginx不支持http tunnel，虽然有些插件可以实现，但是考虑到开发机许多同学都在使用，添加插件还是有风险的。 第二、与方案1同样的，偷开放ssh服务这件事就不能干🙈

3、做一个mysql server，解析客户端请求转发到http代理

先看图： 在我的mac上起一个服务：hersql，“伪装”成mysql server，接收来自桌面客户端的连接，按照mysql协议从报文中解析出请求的db与sql，然后请求开发机上的navicat提供的代理工具：ntunnel_mysql.php，ntunnel_mysql.php连接到db并执行sql，将结果返回给hersql，hersql将查询结果按照mysql协议包装成报文返回给桌面客户端。 完美！

hersql

在方案3中我实现了hersql并开源到了github: github.com/Orlion/hersql，供大家借鉴