程序设计 - Felix021 - So far so good

Jul 18

踩坑记： Go 服务灵异 panic

felix021 @ 2020-7-18 17:31 [IT » 程序设计] 评论(3) , 引用(0) , 阅读(4705) | Via 本站原创

这个坑比较新鲜，周一刚填完，还冒着冷气。

- 1 -

在字节跳动，我们线上服务的所有 log 都通过统一的日志库采集到流式日志服务、落地 ES 集群，配上字节云超(sang)级(xin)强(bing)大(kuang)的监控能力，每一条 panic log 都可以触发一个打给值班同学的电话。

所以我们常常不选电话，只选飞书 ↓↓↓

但毕竟是 panic，大部分 case 都会迅速被就地正法，除了少数排查费劲、又不对线上产生太大影响的，比如这一个：

Error: invalid memory address or nil pointer dereference
Traceback:
goroutine 68532877 [running]:
...
src/encoding/json/encode.go:880 +0x59
encoding/json.stringEncoder(0xcb9fead550, ...)
...
src/encoding/json/encode.go:298 +0xa5
encoding/json.Marshal(0x1ecb9a0, ...)
...
/path/to/util.SendData(0xca813cd300)

注：为了方便阅读，略有简化。

你看，它可以被 recover 兜住（不会把服务搞挂），而且出现频率很低（每天几次甚至没有），考虑到在每天数百亿请求中的占比，解决它的 ROI 实在太低，所以就耽搁了一段时间 ~~且不用担心背 P0 的锅~~。

- 2 -

其实之前 S 同学和我都关注过这个 panic ，从上面的 Error log 可以看到，错误发生在调用 json.Marshal 的时候，调用方的代码大概长这样：

func SendData(...) {
data := map[string]interface{} {
"code": ctx.ErrorCode,
"message": ctx.Message,
"step": ctx.StepName,
}
msg, err := json.Marshal(data)
...
}

注：实际map有更多key/value，这里略作简化。

看这代码，第一反应是：这**也能 panic ？

找到对应的 json 库源码（encode.go第880行，对应下面第5行）：

func (e *encodeState) string(s string, escapeHTML bool) {
e.WriteByte('"')
start := 0
for i := 0; i < len(s); {
if b := s[i]; b < utf8.RuneSelf {
...

—— 也只是从string里逐个读取字符，看着并没什么猫饼。

由于 panic 发生在官方 json 库里，不适合修改并部署到全量机器；引入第三方 json 库又涉及很多依赖问题，所以当时没再跟进。

直到最近 panic 频率逐渐升高， H 和 L 同学实在看不下去了。

- 3 -

L 同学的思路是，既然这个 panic 能被 recover 兜住，那为什么不看看 panic 时这个 map 里装了什么呢？

于是代码就变成了这样：

defer func() {
if p := recover(); p != nil {
log.Warnf("Error: %v, data: %v", p, data)
}
}()
data := map[string]...

然后 panic 顺利转移到了 log.Warnf 这一行[doge]

- 4 -

不管怎么说成功地转移了问题，只要把 log.Warnf 这一行注释掉……

作为一个追求极致的 ByteDancer，L 同学抵制住了诱惑并尝试了新的思路，既然从 panic log 看到是跪在了一个 string 上，那至少先看看是哪一个string：

data := make(map[string]interface{})
defer func() {
if p := recover(); p != nil {
for k, v := range data {
log.Warnf("CatchMe: k=%v", k)
log.Warnf("CatchMe: v=%v", v)
}
}
}()
...

改起来倒是很简单，赶在这个 ~~需要上班的~~ 周日下午发了车，晚上就捉到了一个case。

通过线上 log，我们发现错误出现在 "step" 这个 key 上（log里有输出key、但没输出value），value 本应是 ctx.StepName 这个 string。

可是 string 这种看起来人畜无害的 immutable 的 type 为什么会导致 panic 呢？

- 5 -

通过走读代码得知，在遇到异常的时候，我们会往 ctx.StepName 写入这个异常点的名称，就像这样：

const STEP_XX = "XX"

func XX(...) {
if err := process(); err != nil {
ctx.StepName = STEP_XX
}
}

一边读一边写，有那么点并发的味道了。

考虑到我们为了降低媒体感知的超时率，将整个广告的召回流程包装成一个带时间限制的任务：

finished := make(chan struct{})
timer := time.NewTimer(duration)
go recall(finished)
select {
case <-finished:
sendResponse()
case <- timer.C:
sendTimeoutResponse()
}

因此在一个请求流程中，确实可能会出现并发读写 ctx.StepName 这个 string object 的情况。

但如何实锤是这儿挖的坑呢？

- 6 -

在线上服务中直接验证这一点不太容易，但是 H 同学做了一个简单的 POC，大概像这样：

const (
FIRST = "WHAT THE"
SECOND = "F*CK"
)

func main() {
var s string
go func() {
i := 1
for {
i = 1 - i
if i == 0 {
s = FIRST
} else {
s = SECOND
}
time.Sleep(10)
}
}()

for {
fmt.Println(s)
time.Sleep(10)
}
}

代码一跑起来就有点味道了：

$ go run poc.go
WHAT THE
F*CK
...
WHAT
WHAT
WHAT
F*CKGOGC
...

虽然没看到 panic，但是确实看到了点奇怪的东西（严正声明：不是故意要吐槽GO的GC）。

再用 go 的 race detector 瞅瞅：

$ go run -race poc.go >/dev/null
==================
WARNING: DATA RACE
Write at 0x00c00011c1e0 by goroutine 7:
main.main.func1()
poc.go:19 +0x66（赋值那行）

Previous read at 0x00c00011c1e0 by main goroutine:
main.main()
poc.go:28 +0x9d（println那行）

这下可算是实锤了。

- 7 -

那么为什么 string 的并发读写会出现这种现象呢？

这就得从 string 底层的数据结构说起了。在 go 的 reflect 包里有一个 type StringHeader ，对应的就是 string 在 go runtime的表示：

type StringHeader struct {
Data uintptr
Len int
}

可以看到， string 由一个指针（指向字符串实际内容）和一个长度组成。

比如说我们可以这么玩弄 StringHeader：

s := "hello"
p := *(*reflect.StringHeader)(unsafe.Pointer(&s))
fmt.Println(p.Len)

对于这样一个 struct ，golang 无法保证原子性地完成赋值，因此可能会出现goroutine 1 刚修改完指针（Data）、还没来得及修改长度（Len），goroutine 2 就读取了这个string 的情况。

因此我们看到了 "WHAT" 这个输出 —— 这就是将 s 从 "F*CK" 改成 "WHAT THE" 时，Data 改了、Len 还没来得及改的情况（仍然等于4）。

至于 "F*CKGOGC" 则正好相反，而且显然是出现了越界，只不过越界访问的地址仍然在进程可访问的地址空间里。

- 8 -

既然问题定位到了，解决起来就很简单了。

最直接的方法是使用 sync.Mutex：

func (ctx *Context) SetStep(step string) {
ctx.Mutex.Lock()
defer ctx.Mutex.Unlock()
ctx.StepName = Step
}

但 Mutex 性能不够好（lock does not scale with the number of the processors），对于这种读写冲突概率很小的场景，性能更好的方案是将 ctx.StepName 类型改成 atomic.Value，然后

ctx.StepName.Store(step)

注：也可以改成 *string 然后使用 atomic.StorePointer

实际上，Golang 不保证任何单独的操作是原子性的，除非使用 atomic 包里提供的原语或加锁。

- 9 -

大结局：周一下午 H 同学提交了修复代码并完成发布，这个 panic 就再没出现了。

总结一下：
* string 没有看起来那么人畜无害
* 并发的坑可以找 -race 帮帮忙
* 记得使用 mutex 或 atomic

最后留下一个小问题供思考：

这说了半天并没有完全复现 panic，不过文中已经给了足够多的工具，你能想到怎么办吗？

Jun 25

踩坑记#2：Go服务锁死

felix021 @ 2020-6-25 22:48 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(2128) | Via 本站原创

接着上一篇-内存暴涨坑再挖个坟，讲讲去年踩的另一个坑。

---

前方低能

那是去年7月的一天，被透过落地玻璃的宇宙中心五道口的夕阳照着的正在工位搬砖的我，突然听到一阵骚乱，转头一看，收到夺命连环call的D同学反馈，流量严重异常。

点开报警群，一串异常赫然在目：

点击查看更多...(阅读全文)

Jun 21

golang: bufio.Scanner 的坑

felix021 @ 2020-6-21 01:03 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(2949) | Via 本站原创

之前从网上找的一段代码，按行读取文件:

inFile, err := os.Open("xxx.log")
if err != nil {
fmt.Fprintf(os.Stderr, "open failed: %v\n", err)
return
}
defer inFile.Close()

scanner := bufio.NewScanner(inFile)
for scanner.Scan() {
line := scanner.Bytes()
//do sth. with line
}

看起来没问题，用起来也没问题，直到踩了个坑：针对某个特定的文件，读取到某一行以后就不再继续了。

既然总能复现，那就好解决，我的一个常用方法是：制造一个总能复现的case，并不断缩小case的规模。

例如这个case，把那一行单独拿出来，通过二分找到出问题的位置。

原以为是该行有特殊字符导致触发了什么奇怪的逻辑，但经过不断尝试，发现临界点是该行长度 = 65536 的时候，正好会触发错误。

这么整的数字（2^16, 64KB）必然是代码里的特殊逻辑了，翻了一下 bufio 的源码，果然有一个

const (
//...(一堆注释)...
MaxScanTokenSize = 64 * 1024
)

搜索这个常量在代码里的引用：

func NewScanner(r io.Reader) *Scanner {
return &Scanner{
r: r,
split: ScanLines,
maxTokenSize: MaxScanTokenSize,
}
}

...

func (s *Scanner) Scan() bool {
....
if len(s.buf) >= s.maxTokenSize || len(s.buf) > maxInt/2 {
s.setErr(ErrTooLong)
return false
}
...
}

在 for 循环后加上一句：

if scanner.Err() != nil {
fmt.Fprintf(os.Stderr, "scan err: %v\n", scanner.Err())
}

实锤：

引用

scan err: bufio.Scanner: token too long

那怎么解决呢？

MaxScanTokenSize 上面的注释是这么写的：

引用

// MaxScanTokenSize is the maximum size used to buffer a token
// unless the user provides an explicit buffer with Scanner.Buffer.
// The actual maximum token size may be smaller as the buffer
// may need to include, for instance, a newline.

于是最终版的解决方案是这样：

...
scanner := bufio.NewScanner(inFile)
buf := make([]byte, 0, bufio.MaxScanTokenSize * 10) //根据自己的需要调整这个倍数
scanner.Buffer(buf, cap(buf))

for scanner.Scan() {
line := scanner.Bytes()
//do sth. with line
}

if scanner.Err() != nil {
fmt.Fprintf(os.Stderr, "scan err: %v\n", scanner.Err())
}

真是丑陋的api啊。

Sep 21

使用pprof和go-torch排查golang的性能问题

felix021 @ 2019-9-21 15:19 [IT » 程序设计] 评论(1) , 引用(0) , 阅读(6362) | Via 本站原创

最近线上服务压力很大，api的p99有点扛不住。

广告业务对延时的要求普遍比较严格，有些adx设置的超时时间低至100ms，因此亟需找出性能热点。

根据对目前系统情况的估计（和metrics埋点数据），大致估计问题出在广告的正排环节。

使用 pprof 也证明了这一块确实是热点：

引用

$ go tool pprof http://$IP:$PORT/debug/pprof/profile
...
(pprof) top 10
Showing nodes accounting for 25.50s, 32.63% of 78.14s total
Dropped 1533 nodes (cum <= 0.39s)
Showing top 10 nodes out of 284
flat flat% sum% cum cum%
4.56s 5.84% 5.84% 4.87s 6.23% syscall.Syscall
4.03s 5.16% 10.99% 4.03s 5.16% runtime.aeshashbody
3.50s 4.48% 15.47% 6.01s 7.69% git.xxx.org/xxx/target.NewFilter
2.78s 3.56% 19.03% 3.73s 4.77% runtime.mapaccess2_fast64
2.63s 3.37% 22.40% 4.52s 5.78% runtime.mapiternext
2.08s 2.66% 25.06% 2.16s 2.76% runtime.heapBitsForObject
1.65s 2.11% 27.17% 1.93s 2.47% runtime.mapaccess1_fast64
1.57s 2.01% 29.18% 2.96s 3.79% runtime.mapaccess2
1.43s 1.83% 31.01% 2.06s 2.64% runtime.runqgrab
1.27s 1.63% 32.63% 1.27s 1.63% runtime.epollwait
(pprof) png
Generating report in profile001.png (使用生成的线框图查看耗时）

其中第三行 NewFilter 就是正排过滤函数。因为一些历史原因，系统里不是所有定向条件都使用了倒排，正排实现起来毕竟简单、容易理解，而一旦开了这个口子，就会有越来越多正排加进来，推测是这个原因导致了性能的逐渐衰退。

经过讨论，D同学花了一周多的时间，逐个梳理重写。在Libra（字节跳动内部的ABTest平台，参考谷歌分层实验框架方案）上开实验，平均耗时 -9%，从统计数据上来说，实验组比对照组有了显著的改进，但从最终结果上，整体的p95、p99超时都出现了进一步恶化。

这说明真正的问题不在于正排的计算，优化的思路出现了偏差。

考虑到晚高峰期间的cpu占用率也只是刚超过50%，也就是说有可能性能问题在于锁，但pprof的 block 和 mutex 都是空的，没有线索。

猜测问题有可能在日志，代码里确实用得不少。日志用的是 github.com/ngaut/logging 库，每一次调用都会用到两个全局mutex。但通过调整log level 为error级别，大幅减少了日志量，并没有看到性能的改善。

经过搜索，发现 uber 基于 pprof 开发了一个神器 go-torch，可以生成火焰图。安装好 go-torch 及依赖 FlameGraph 以后执行

引用

$ go-torch -u http://$IP:$PORT -f cpu.svg
INFO[14:52:23] Run pprof command: go tool pprof -raw -seconds 30 http://$IP:$PORT/debug/pprof/profile
INFO[14:52:54] Writing svg to cpu.svg

用 Chrome 打开 cpu.svg，人肉排查：

可以看到，在NewFilter旁边竟然还有一个耗时接近的 runtime.growslice ，结合实际代码（略作简化），可以推测是 slice 的初始化长度不足。

matchAds := make([]*ad, 0, 4096)
adsBitMap.GetList(func(seq int) {
if NewFilter(ctx, ad) {
matchAds = append(matchAds, adlist[seq])
}
})

// 顺便提一下，bitmap是一个uint64数组，GetList(f) 是将每一个等于1的bit索引传给 f
// GetList方法里面用了cpu的BSF指令来提高性能。

实际上最终定向后得到的广告往往在数万甚至数十万的级别，而 go 的 slice 扩容在超过1024个元素以后是1.25倍，可想而知会出现大量的内存分配和拷贝，导致性能随着广告数量的增加逐渐恶化。最近的广告数量也确实有了大幅的上升 —— 逻辑上形成了闭环。

经过优化，使用更大的初始化长度，并且使用 sync.Pool 来进一步减少内存分配，最终上线后p95和p99都下降了超过50%，效果显著。

参考：
golang 使用pprof和go-torch做性能分析 https://www.cnblogs.com/li-peng/p/9391543.html

Jul 2

Golang rand.Rand 并发panic: index out of range

felix021 @ 2019-7-2 19:16 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(4286) | Via 本站原创

线上服务Panic，部分日志如下

引用

err: runtime error: index out of range
Traceback:
goroutine 19209941 [running]:
...
panic(0x191d0e0, 0x2e078d0)
  /usr/local/go/src/runtime/panic.go:502 +0x229
math/rand.(*rngSource).Uint64(...)
  /usr/local/go/src/math/rand/rng.go:246
math/rand.(*rngSource).Int63(0xc438bb2a00, 0x0)
  /usr/local/go/src/math/rand/rng.go:231 +0x8a
math/rand.(*Rand).Int63(0xc4279b3a70, 0x0)
  /usr/local/go/src/math/rand/rand.go:82 +0x33
math/rand.(*Rand).Int(0xc4279b3a70, 0x0)
  /usr/local/go/src/math/rand/rand.go:100 +0x2b
...

放狗搜了一下：math.Rand is not safe for concurrent use

from: https://github.com/golang/go/issues/3611

这个 issue 的 4 楼还提到 "top-level functions like strings.Split or fmt.Printf or rand.Int63 may be called from any goroutine at any time"

翻了一下源码，rand.Int() 用是自带 lock 的 globalRand 对象

func Int() int { return globalRand.Int() }

...

var globalRand = New(&lockedSource{src: NewSource(1).(Source64)})

...

type lockedSource struct {
  lk sync.Mutex
  src Source64
}

...

func (r *lockedSource) Uint64() (n uint64) {
  r.lk.Lock()
  n = r.src.Uint64()
  r.lk.Unlock()
  return
}

看了下调用代码，之前的实现为了避免多个 goroutine 竞争同一个锁，所以 new 了一个 rand.Rand 对象，但没考虑到这个对象不支持并发。

最终的解决方案，是实现了一个 safeRander 。

具体代码不适合贴，核心逻辑是初始化 N 个 rand.Rand 对象和对应的 N 个锁，以及一个 index，每次调用 Int() 时，先 atomic.AddUint32(&index, 1) % N，加上对应的锁，再用对应的 rand.Rand 对象。

这样只要并发使用的goroutine不超过N个，就不会出现竞争；就算超过，竞争出现的频率也大幅减少了，而且也可以通过增加 N 来优化。

Mar 31

LeetCode 69 - x 的平方根

felix021 @ 2019-3-31 15:34 [IT » 程序设计] 评论(1) , 引用(0) , 阅读(3727) | Via 本站原创

这是罗凯同学布置的 Golang 学习作业。

这题之前用 Python 刷过，用的是二分法，在 [1, n / 2] 区间内，找到第一个 x，使得 x ^ 2 <= n < (x + 1) ^ 2 ，用的是 STL 中 lowerbound 的算法。

class Solution(object):
def mySqrt(self, x):
"""
:type x: int
:rtype: int
"""
if x < 0:
raise Exception("invalid input")
if x < 2:
return x

left = 1
length = x / 2
while length > 1:
half = length / 2
middle = left + half
if middle * middle > x:
length = half
else:
left = middle
length = length - half
return left

罗凯同学提到，应该使用牛顿迭代法来完成。这个方法是听说过的，但是早就忘了，于是到 wikipedia 去找了一下：

https://zh.wikipedia.org/wiki/%E7%89%9B%E9%A1%BF%E6%B3%95

求函数 f(x) 的零点，可以通过选取曲线上的任意一个点 x0 开始，然后计算 x1 = x0 - f(x1) / f'(x1) 的方式迭代，通常得到一个比 x0 更接近零点的 x1 。通过不断迭代，最终我们能找到一个零点 xn 。

对于求平方根，我们是要找到一个 x，使得 x ^2 - n = 0，也就是这里的 f(x) = x ^ 2 - n, f'(x) = 2 * x （勉强还记得这个求导公式……）

有了这个，答案就呼之欲出了：

import "math"

func mySqrt(x int) int {
f := func (i float64) float64 {
return i * i - float64(x)
}
g := func (i float64) float64 {
return 2 * i
}
var i float64 = 1.0
for math.Abs(f(i)) > 1e-6 {
i = i - f(i) / g(i)
}
return int(math.Floor(i))
}

做完以后，我想起 Quake III 的作者 John Carmack 的平方根倒数速算法，摘录一段内容：( src: https://blog.csdn.net/zyex1108/article/details/53540824 )

引用

Quake-III Arena (雷神之锤3)是90年代的经典游戏之一。该系列的游戏不但画面和内容不错，而且即使计算机配置低，也能极其流畅地运行。这要归功于它3D引擎的开发者约翰-卡马克（John Carmack）。事实上早在90年代初DOS时代，只要能在PC上搞个小动画都能让人惊叹一番的时候，John Carmack就推出了石破天惊的Castle Wolfstein, 然后再接再励，doom, doomII, Quake...每次都把3-D技术推到极致。他的3D引擎代码资极度高效，几乎是在压榨PC机的每条运算指令。

这个平方根倒数算法正是其中的一个例子。在3D游戏引擎中，求取照明和投影的波动角度与反射效果时，常需计算平方根倒数，而求平方根的常用算法效率较低。

Carmack 通过使用一个惊为天人的魔术常量 0x5f3759df，只需要做 1 次迭代（Quaker III源码中的为了提高精度的第二次迭代被注视掉了），就能得到一个足够精度的平方根，大幅提高了 3D 引擎的运行效率。

关于这个魔术常量，Carmack 表示并不是他自己发明的，至今为止仍未能确切知晓算法中所使用的特殊常数的起源。但 Carmack 凭一己之力，撑起了一个 3D 引擎的时代，以至于在1999年，登上了美国时代杂志评选出来的科技领域50大影响力人物榜单，并且名列第10位。

感兴趣的同学，可以在 Wikipedia 的平方根倒数速算法了解更多细节：

https://zh.wikipedia.org/wiki/%E5%B9%B3%E6%96%B9%E6%A0%B9%E5%80%92%E6%95%B0%E9%80%9F%E7%AE%97%E6%B3%95

Jan 29

跳表 - 简明教程 in Python

felix021 @ 2018-9-19 05:30 [IT » 程序设计] 评论(0) , 引用(0) , 阅读(4711) | Via 本站原创

# 1. 什么是跳表

跳表（Skip List）是基于链表 + 随机化实现的一个有序数据结构，可以达到平均 O(logN) 的查找、插入、删除效率，在实际运行中的效率往往超过 AVL 等平衡二叉树，而且其实现相对更简单、内存消耗更低。

Redis 的 ZSET 底层实现就是用的 Skip List，这里是 [Antirez对此的说明](https://news.ycombinator.com/item?id=1171423)。

这是一个典型的跳表：

[0] -> 0 -> 1 -> 3 -> 4 -> 5 -> 6 -> 7 -> 9 -> nil
[1] -> 0 ------> 3 ------> 5 ------> 7 ------> nil
[2]----------------------> 5-----------------> nil

解释一下：

1. SkipList 是一个多层的链表

2. 第[0]层的链表包含所有节点，其他层的链表包含部分节点，层次越高，节点越少

3. 每层链表之间会共享相同的节点（节省内存，但为了方便展示，每一层都输出了它的值）

4. 对于某个节点，在插入时通过概率判断它最高会出现在哪一层，并且也会出现在之下的每一层

通过这样的设计，当需要查找某个 key 时，可以从最高层的链表开始往前找，在这一层遇到末尾或者大于 key 的节点时往下走一个层，直到找到 key 节点。

例如:

引用

4 的查找路径为 [2] -> [1] -> 0 -> 3 -> 3@[0] -> 4
6 的查找路径为 [2] -> 5 -> 5@[1] -> 5@[0] -> 6
8 的查找路径为 [2] -> 5 -> 5@[1] -> 7 -> 7@[0] -> 9 （找不到）

# 2. 跳表的节点

从上面的描述，我们大概可以知道 (1) 每个节点需要保存一个 key; (2) 每个节点需要有多个next指针 (3) 其 next 指针的数量会在插入时确定

因此我们可以用下面这个 class 来表示节点:

class Node(object)
def __init__(self, height, key):
self.key = key
self.next = [None] * height

def height(self):
return len(self.next)

# 3. 创建跳表

一个新创建的跳表是没有节点的。但为了实现的简单起见，可以添加一个头节点：

class SkipList(object):
def __init__(self):
self.head = Node(0, None) #头节点高度为0，不需要key

到目前为止都特别简单，但是还什么也干不了。

# 4. 创建节点

创建节点时，需要先按一定的概率分布确定其高度。

为了保证高层的节点比低层少，我们可以用这样的概率分布：

引用

Height(n) = p^n

实现其实非常简单：

import random

def randomHeight(self, p = 0.5):
height = 1
while random.uniform(0, 1) < p and self.head.height() >= height:
height += 1
return height

这样可以保证平均的路径长度是 log(n) 。

精确一点的话，实际上是 log(n-1, 1/p) / p，也就是说， p 的选择会影响跳表层数、平均路径长度。

具体的计算比较复杂，有兴趣可以参考跳表的原论文《Skip Lists: A Probabilistic Alternative to Balanced Trees》。（TL;DR）

然后我们就可以这样来创建一个新的节点：

node = Node(self.randomHeight(), key)

# 5. 添加节点

如果只是为空跳表添加一个新的节点，只要更新头结点的每一个next指针：

def insertFirstNode(self, key):
node = Node(self.randomHeight(), key)
while node.height > self.head.height():
self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

for level in range(node.height()):
node.next[level] = self.head.next[level]
self.head.next[level] = node

但很显然这个方法只能用一次。

如果跳表中已经有多个节点，那我们就必须找到每一层中适合插入的位置：

def getUpdateList(self, key):
update = [None] * self.head.height()
for level in range(len(update)):
x = self.head
while x.next[level] is not None and x.next[level].key < key:
x = x.next[level]
update[level] = x
return update

这个函数返回一个 update 节点数组，其中的每个节点都是在这一层中小于 key 的最后一个节点。

也就是说，在 level = i 层，总是可以把新的节点插入 update[i] 之后：

def insert(self, key):
node = Node(self.randomHeight(), key)
while node.height > self.head.height():
self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

update = self.getUpdateList(key)
next0 = update[0].next[0]
if next0 is not None and next0.key == key:
return # 0层总是包含所有元素；如果 update[0] 的下一个节点与key相等，则无需插入。

for level in range(node.height()):
node.next[level] = update[level].next[level]
update[level].next[level] = node

但是由于这一版 getUpdateList 是 O(n) 的，插入效率并没有达到跳表的设计目标。

# 6. 添加节点++

考虑这一点：跳表的每一层都是有序的。

也就是说，我们在找到 update[n] = x 以后，其实可以从节点 x 的 n - 1 层继续查找 update[n-1] 应该是哪个节点。

由于查找路径的平均长度是 log(N) ，所以我们可以实现一个更快的 getUpdateList 方法

注意，需要从最高层开始查

def getUpdateList(self, key):
update = [None] * self.head.height()
x = self.head
for level in reversed(range(len(update))):
while x.next[level] is not None and x.next[level].key < key:
x = x.next[level]
update[level] = x
return update

# 7. 里程碑1

把上面的代码整合起来，我们就可以得到第一版跳表代码：能够插入节点。

为了更好地展示我们的成果，我们可以用这样一个函数，把链表按第1节的例子样式输出：

def dump(self):
for i in range(self.head.height()):
sys.stdout.write('[H]')
x = self.head.next[0]
y = self.head.next[i]
while x is not None:
s = ' -> %s' % x.key
if x is y:
y = y.next[i]
else:
s = '-' * len(s)
x = x.next[0]
sys.stdout.write(s)
print ' -> <nil>'
print

试试看：

sl = SkipList()
for i in range(10):
sl.insert(sl)
s1.dump()

[H] -> 0 -> 1 -> 2 -> 3 -> 4 -> 5 -> 6 -> 7 -> 8 -> 9 -> <nil>
[H]----- -> 1 -> 2 -> 3---------- -> 6 -> 7---------- -> <nil>
[H]---------- -> 2-------------------- -> 7---------- -> <nil>

多尝试几次，以及选择不同的 p 值，可以观察生成跳表的区别。

# 8. 查找节点

实际上查找节点的过程，已经包含在 insert 的实现里了：

def find(self, key):
update = self.getUpdateList(key)
if len(update) == 0:
return None

next0 = update[0].next[0]
if next0 is not None and next0.key == key:
return next0 # 0层总是包含所有元素；如果 update[0] 的下一个节点与key相等，则无需插入。
else:
return None

# 9. 删除节点

既然已经能找出 update 节点数组，在 level = i 层，只要判断 update[i].next[i] 是否等于要删除的 key 就可以了：

def remove(self, key):
update = self.getUpdateList(key)
for i, node in enumerate(update):
if node.next[i] is not None and node.next[i].key == key:
node.next[i] = node.next[i].next[i]

# 10. 里程碑2

整合 find 和 update 数组，就可以实现跳表的基础操作了，试试看：

node = sl.find(3)
print node

for i in range(7, 14):
sl.remove(i)
sl.dump()

# 11. 其他

我们在 Node 中只添加了一个 key 属性，在具体的实现中，我们往往可能需要针对 key 存储一个 value，例如 Python 自带的 dict 实现。改造起来也很简单:

1. node 中添加一个 value 属性，并且添加相应的初始化逻辑(__init__方法)

2. 将 SkipList.insert 修改为 `insert(self, key, value)`，在新建 Node 时指定其 value

3. 再添加一个 `update(self, key, value)` API，方便调用方的使用

4. 可以考虑针对语言适配，例如实现 python 的 __getitem__ 、 __setitem__ 等魔术方法

# 12. 完整代码

#coding:utf-8

import random

class Node(object):
def __init__(self, height, key=None):
self.key = key
self.next = [None] * height

def height(self):
return len(self.next)

class SkipList(object):
def __init__(self):
self.head = Node(0, None) #头节点高度为0，不需要key

def randomHeight(self, p = 0.5):
height = 1
while random.uniform(0, 1) < p and self.head.height() >= height:
height += 1
return height

def insert(self, key):
node = Node(self.randomHeight(), key)
print node.height(), node.key
while node.height() > self.head.height():
self.head.next.append(None) #保证头节点的next数组覆盖所有层次的链表

update = self.getUpdateList(key)
if update[0].next[0] is not None and update[0].next[0].key == key:
return # 0层总是包含所有元素；如果 update[0] 的下一个节点与key相等，则无需插入。

for level in range(node.height()):
node.next[level] = update[level].next[level]
update[level].next[level] = node

def getUpdateList(self, key):
update = [None] * self.head.height()
x = self.head
for level in reversed(range(len(update))):
while x.next[level] is not None and x.next[level].key < key:
x = x.next[level]
update[level] = x
return update

def dump(self):
for i in range(self.head.height()):
sys.stdout.write('[H]')
x = self.head.next[0]
y = self.head.next[i]
while x is not None:
s = ' -> %s' % x.key
if x is y:
y = y.next[i]
else:
s = '-' * len(s)
x = x.next[0]
sys.stdout.write(s)
print ' -> <nil>'
print

def find(self, key):
update = self.getUpdateList(key)
if len(update) == 0:
return None

next0 = update[0].next[0]
if next0 is not None and next0.key == key:
return next0 # 0层总是包含所有元素；如果 update[0] 的下一个节点与key相等，则无需插入。
else:
return None

def remove(self, key):
update = self.getUpdateList(key)
for i, node in enumerate(update):
if node.next[i] is not None and node.next[i].key == key:
node.next[i] = node.next[i].next[i]

完。

分页： 2/23

1 2 3 4 5 6 7 8 9 10

[ 显示模式：摘要 | 列表 ]

Felix021