修改可能的数据竞争的代码,使用version管理node的迁移

如果node的version和表头的version不一样，就使用惰性删除

Author: guonaihong

time_wheel.go

@@ -59,13 +59,13 @@ func newTimeWheel() *timeWheel {
 func (t *timeWheel) init() {
 
 	for i := 0; i < nearSize; i++ {
-		t.t1[i] = newTimeHead()
+		t.t1[i] = newTimeHead(1, uint64(i))
 
 	}
 
 	for i := 0; i < 4; i++ {
 		for j := 0; j < levelSize; j++ {
-			t.t2Tot5[i][j] = newTimeHead()
+			t.t2Tot5[i][j] = newTimeHead(uint64(i+2), uint64(j))
 		}
 	}
 
@@ -90,9 +90,11 @@ func (t *timeWheel) add(node *timeNode, jiffies uint64) *timeNode {
 	expire := node.expire
 	idx := expire - jiffies
 
+	level, index := uint64(1), uint64(0)
+
 	if idx < nearSize {
 
-		index := uint64(expire) & nearMask
+		index = uint64(expire) & nearMask
 		head = t.t1[index]
 
 	} else {
@@ -106,8 +108,9 @@ func (t *timeWheel) add(node *timeNode, jiffies uint64) *timeNode {
 			}
 
 			if uint64(idx) < levelMax(i+1) {
-				index := int64(expire) >> (nearShift + i*levelShift) & levelMask
+				index = uint64(expire >> (nearShift + i*levelShift) & levelMask)
 				head = t.t2Tot5[i][index]
+				level = uint64(i) + 2
 				break
 			}
 		}
@@ -118,7 +121,7 @@ func (t *timeWheel) add(node *timeNode, jiffies uint64) *timeNode {
 		panic("not found head")
 	}
 
-	head.lockPushBack(node)
+	head.lockPushBack(node, level, index)
 
 	return node
 }
@@ -164,7 +167,7 @@ func (t *timeWheel) Stop() {
 // 移动链表
 func (t *timeWheel) cascade(levelIndex int, index int) {
 
-	tmp := newTimeHead()
+	tmp := newTimeHead(0, 0)
 
 	l := t.t2Tot5[levelIndex][index]
 	l.Lock()
@@ -175,6 +178,8 @@ func (t *timeWheel) cascade(levelIndex int, index int) {
 
 	l.ReplaceInit(&tmp.Head)
 
+	// 每次链表的元素被移动走，都修改version
+	atomic.AddUint64(&l.version, 1)
 	l.Unlock()
 
 	offset := unsafe.Offsetof(tmp.Head)
@@ -218,10 +223,10 @@ func (t *timeWheel) moveAndExec() {
 		return
 	}
 
-	head := newTimeHead()
+	head := newTimeHead(0, 0)
 	t1 := t.t1[index]
 	t1.ReplaceInit(&head.Head)
-
+	atomic.AddUint64(&t1.version, 1)
 	t.t1[index].Unlock()
 
 	// 执行
@@ -231,6 +236,11 @@ func (t *timeWheel) moveAndExec() {
 	head.ForEachSafe(func(pos *list.Head) {
 		val := (*timeNode)(pos.Entry(offset))
 		head.Del(pos)
+
+		if atomic.LoadUint32(&val.stop) == haveStop {
+			return
+		}
+
 		go val.callback()
 
 		if val.isSchedule {

time_wheel_node.go

@@ -4,64 +4,86 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
 	"github.com/antlabs/stl/list"
 )
 
 const (
 	haveStop = uint32(1)
-	//stopGrab = 1 << (iota + 1)
-	//pushGrab
 )
 
 // 先使用sync.Mutex实现功能
 // 后面使用cas优化
 type Time struct {
 	timeNode
 	sync.Mutex
+
+	// |---16bit---|---16bit---|------32bit-----|
+	// |---level---|---index---|-------seq------|
+	// level 在near盘子里就是1, 在T2ToTt[0]盘子里就是2起步
+	// index 就是各自盘子的索引值
+	// seq   自增id
+	version uint64
 }
 
-func newTimeHead() *Time {
+func newTimeHead(level uint64, index uint64) *Time {
 	head := &Time{}
+	head.version = genVersionHeight(level, index)
 	head.Init()
 	return head
 }
 
-func (t *Time) lockPushBack(node *timeNode) {
+func genVersionHeight(level uint64, index uint64) uint64 {
+	return level<<(32+16) | index<<32
+}
+
+func (t *Time) lockPushBack(node *timeNode, level uint64, index uint64) {
 	t.Lock()
 	defer t.Unlock()
-	if atomic.LoadUint32(&node.lock) == haveStop {
+	if atomic.LoadUint32(&node.stop) == haveStop {
 		return
 	}
 
 	t.AddTail(&node.Head)
-	node.list = t
+	atomic.StorePointer(&node.list, unsafe.Pointer(t))
+	//更新节点的version信息
+	atomic.StoreUint64(&node.version, atomic.LoadUint64(&t.version))
 }
 
 type timeNode struct {
 	expire     uint64
 	userExpire time.Duration
 	callback   func()
+	stop       uint32
+	list       unsafe.Pointer //存放表头信息
+	version    uint64         //保存节点版本信息
 	isSchedule bool
-	close      uint32
-	lock       uint32
-
-	list *Time
 
 	list.Head
 }
 
+// 一个timeNode节点有4个状态
+// 1.存在于初始化链表中
+// 2.被移动到tmp链表
+// 3.1 和 3.2是if else的状态
+// 	3.1被移动到new链表
+// 	3.2直接执行
+// 1和3.1状态是没有问题的
+// 2和3.2状态会是没有锁保护下的操作,会有数据竞争
 func (t *timeNode) Stop() {
-	//这里为什么修改成cpyList := t.list
-	//如果直接使用t.list.Lock()和t.list.Unlock()就会和lockPushBack函数里的node.list = t 是竞争关系
-	//lockPushBack拿到锁，修改国t.list的值。这时候Stop函数里面的t.list.Lock()持有旧链表里的锁。t.list.unlock新链表里的锁，发触发unlock unlock情况。
-	
-	//TODO：思考有没有新的竞争关系。。。
-	cpyList := t.list
+
+	atomic.StoreUint32(&t.stop, haveStop)
+
+	// 使用版本号算法让timeNode知道自己是否被移动了
+	// timeNode的version和表头的version一样表示没有被移动可以直接删除
+	// 如果不一样，可能在第2或者3.2状态，使用惰性删除
+	cpyList := (*Time)(atomic.LoadPointer(&t.list))
 	cpyList.Lock()
 	defer cpyList.Unlock()
+	if atomic.LoadUint64(&t.version) != atomic.LoadUint64(&cpyList.version) {
+		return
+	}
 
-	atomic.StoreUint32(&t.close, haveStop)
-
-	t.list.Del(&t.Head)
+	cpyList.Del(&t.Head)
 }

time_wheel_test.go

@@ -19,6 +19,11 @@ func Test_LevelMax(t *testing.T) {
 	assert.Equal(t, levelMax(4), uint64(1<<(nearShift+4*levelShift)))
 }
 
+func Test_GenVersion(t *testing.T) {
+	assert.Equal(t, genVersionHeight(1, 0xf), uint64(0x0001000f00000000))
+	assert.Equal(t, genVersionHeight(1, 64), uint64(0x0001004000000000))
+}
+
 func Test_hour(t *testing.T) {
 	tw := newTimeWheel()
 

timer_test.go

@@ -28,7 +28,7 @@ func Test_ScheduleFunc(t *testing.T) {
 	}()
 
 	tm.Run()
-	assert.Equal(t, count, uint32(5))
+	assert.Equal(t, atomic.LoadUint32(&count), uint32(5))
 }
 
 func Test_AfterFunc(t *testing.T) {
@@ -64,7 +64,23 @@ func Test_AfterFunc(t *testing.T) {
 		tm.Stop()
 	}()
 	tm.Run()
-	assert.Equal(t, count, uint32(2))
+	assert.Equal(t, atomic.LoadUint32(&count), uint32(2))
+}
+
+func Test_Node_Stop_1(t *testing.T) {
+	tm := NewTimer()
+	count := uint32(0)
+	node := tm.AfterFunc(time.Millisecond*10, func() {
+		atomic.AddUint32(&count, 1)
+	})
+	go func() {
+		time.Sleep(time.Millisecond * 30)
+		node.Stop()
+		tm.Stop()
+	}()
+
+	tm.Run()
+	assert.NotEqual(t, count, 1)
 }
 
 func Test_Node_Stop(t *testing.T) {