bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-07-08 14:39:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

up

Browse Source
This commit is contained in:
John
2019-01-12 20:20:30 +08:00
parent 33dd0f9922
commit 241706cbbf
6 changed files with 92 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
g/container/glist/glist_z_bench_test.go
View File

@ -12,33 +12,41 @@ import (
"testing"
)
var l = New()
var (
l = New()
bn = 20000000
)
func Benchmark_PushBack(b *testing.B) {
b.N = bn
for i := 0; i < b.N; i++ {
l.PushBack(i)
}
}
func Benchmark_PushFront(b *testing.B) {
b.N = bn
for i := 0; i < b.N; i++ {
l.PushFront(i)
}
}
func Benchmark_Len(b *testing.B) {
b.N = bn
for i := 0; i < b.N; i++ {
l.Len()
}
}
func Benchmark_PopFront(b *testing.B) {
b.N = bn
for i := 0; i < b.N; i++ {
l.PopFront()
}
}
func Benchmark_PopBack(b *testing.B) {
b.N = bn
for i := 0; i < b.N; i++ {
l.PopBack()
}

4
g/os/gwheel/gwheel.go
View File

@ -22,8 +22,8 @@ const (
gPANIC_EXIT = "exit"
gDEFAULT_TIMES = math.MaxInt32
gDEFAULT_SLOT_NUMBER = 10
gDEFAULT_WHEEL_INTERVAL = 50*time.Millisecond
gDEFAULT_WHEEL_LEVEL = 10
gDEFAULT_WHEEL_INTERVAL = 30*time.Millisecond
gDEFAULT_WHEEL_LEVEL = 5
)
var (

83
g/os/gwheel/gwheel_entry.go
View File

@ -20,25 +20,27 @@ type Entry struct {
status *gtype.Int // 任务状态(0: ready; 1: running; -1: closed)
times *gtype.Int // 还需运行次数
create int64 // 注册时的时间轮ticks
update *gtype.Int64 // 更新时间(上一次检查/运行时间, 毫秒)
interval int64 // 设置的运行间隔(时间轮刻度数量)
createMs int64 // 创建时间(毫秒)
updateMs int64 // 更新时间(上一次检查/运行时间, 毫秒)
intervalMs int64 // 间隔时间(毫秒)
}
// 任务执行方法
type JobFunc func()
type JobFunc = func()
// 创建循环任务(失败返回nil)
func (w *wheel) newEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
ims := interval.Nanoseconds()/1e6
num := ims/w.intervalMs
func (w *wheel) addEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
ms := interval.Nanoseconds()/1e6
num := ms/w.intervalMs
if num == 0 {
return nil
// 如果安装的任务间隔小于时间轮刻度，
// 那么将会在下一刻度被执行
num = 1
}
nowMs := time.Now().UnixNano()/1e6
ticks := w.ticks.Val()
entry := &Entry {
nowMs := time.Now().UnixNano()/1e6
ticks := w.ticks.Val()
entry := &Entry {
id : time.Now().UnixNano(),
wheel : w,
singleton : gtype.NewBool(singleton),
@ -46,26 +48,13 @@ func (w *wheel) newEntry(interval time.Duration, job JobFunc, singleton bool, ti
times : gtype.NewInt(times),
job : job,
create : ticks,
update : gtype.NewInt64(nowMs),
interval : num,
createMs : nowMs,
intervalMs : ims,
updateMs : nowMs,
intervalMs : ms,
}
// 安装的slot(可能多个)
set := make(map[int64]struct{})
index := int64(ticks%int64(w.number))
for i, j := int64(0), int64(0); i < int64(times); {
t := (i + index + num) % w.number
if _, ok := set[t]; ok {
break
}
set[t] = struct{}{}
w.slots[t].PushBack(entry)
//fmt.Println("addEntry:", w.level, t, ticks, entry.create, time.Now(), entry.id)
i += 1
j += num
}
//fmt.Println("addEntry:", w.level, ticks, entry.create, time.Now(), entry.update.Val())
// 安装任务
w.slots[(ticks + num) % w.number].PushBack(entry)
return entry
}
@ -105,12 +94,12 @@ func (entry *Entry) Run() {
}
// 检测当前任务是否可运行, 参数为当前时间的纳秒数, 精度更高
func (entry *Entry) runnableCheck(nowTicks int64, nowMs int64) bool {
func (entry *Entry) check(nowTicks int64, nowMs int64) bool {
//if entry.intervalMs == 1400 {
// fmt.Println(nowTicks, entry.create, entry.interval, time.Now())
//}
// 运行检查
if diff := nowTicks - entry.create; diff > 0 && diff%entry.interval == 0 {
// 是否关闭
if entry.status.Val() == STATUS_CLOSED {
return false
}
// 是否单例
if entry.IsSingleton() {
if entry.status.Set(STATUS_RUNNING) == STATUS_RUNNING {
@ -126,29 +115,31 @@ func (entry *Entry) runnableCheck(nowTicks int64, nowMs int64) bool {
}
}
// 是否不限制运行次数
if times > 2000000000 {
if times < 2000000000 && times > 1000000000 {
times = gDEFAULT_TIMES
entry.times.Set(gDEFAULT_TIMES)
}
// 分层转换
if entry.wheel.level > 0 {
if diff := nowMs - entry.update.Val(); diff < entry.intervalMs {
delay := time.Duration(entry.intervalMs - diff)*time.Millisecond
//fmt.Println("LEVEL:", entry.wheel.level, times, delay,
// time.Duration(entry.intervalMs)*time.Millisecond, time.Now(),
// entry.update.Val(),
//)
// 往底层添加
entry.wheel.wheels.newEntry(delay, entry.job, false, 1, entry.wheel)
// 延迟重新添加
if times > 0 {
entry.wheel.wheels.DelayAddTimes(delay, time.Duration(entry.intervalMs)*time.Millisecond, times, entry.job)
if diffMs := nowMs - entry.updateMs; diffMs < entry.intervalMs {
if leftMs := entry.intervalMs - diffMs; leftMs > entry.wheel.wheels.intervalMs {
entry.status.Set(STATUS_CLOSED)
delay := time.Duration(leftMs)*time.Millisecond
// 往底层添加
entry.wheel.wheels.addEntry(delay, entry.job, false, 1)
// 延迟重新添加
if times > 0 {
entry.wheel.wheels.DelayAddTimes(
delay,
time.Duration(entry.intervalMs)*time.Millisecond,
times,
entry.job,
)
}
return false
}
entry.status.Set(STATUS_CLOSED)
return false
}
}
entry.update.Set(nowMs)
return true
}
return false

69
g/os/gwheel/gwheel_loop.go
View File

@ -7,7 +7,6 @@
package gwheel
import (
"container/list"
"gitee.com/johng/gf/g/container/glist"
"time"
)
@ -29,29 +28,26 @@ func (w *wheel) start() {
}()
}
// 执行时间轮刻度逻辑, 遍历检查可执行循环任务，并异步执行
// 执行时间轮刻度逻辑
func (w *wheel) proceed() {
n := w.ticks.Add(1)
l := w.slots[int(n%w.number)]
if l.Len() > 0 {
go func(l *glist.List, nowTicks int64) {
nowMs := time.Now().UnixNano()/1e6
removeArray := make([]*glist.Element, 0)
l.RLockFunc(func(list *list.List) {
for e := list.Front(); e != nil; e = e.Next() {
entry := e.Value.(*Entry)
// 任务是否已关闭，那么需要删除
if entry.Status() == STATUS_CLOSED {
removeElement := e
removeArray = append(removeArray, removeElement)
continue
}
// 是否满足运行条件
if !entry.runnableCheck(nowTicks, nowMs) {
continue
}
nowMs := time.Now().UnixNano()/1e6
for i := l.Len(); i > 0; i-- {
v := l.PopFront()
if v == nil {
break
}
entry := v.(*Entry)
if entry.Status() == STATUS_CLOSED {
continue
}
// 是否满足运行条件
if entry.check(nowTicks, nowMs) {
// 异步执行运行
go func(entry *Entry, e *glist.Element, l *glist.List) {
go func(entry *Entry) {
defer func() {
if err := recover(); err != nil {
if err != gPANIC_EXIT {
@ -60,33 +56,24 @@ func (w *wheel) proceed() {
entry.Close()
}
}
switch entry.Status() {
case STATUS_CLOSED:
//fmt.Println("remove:", w.level, entry.interval)
l.Remove(e)
case STATUS_RUNNING:
entry.SetStatus(STATUS_READY)
if entry.Status() == STATUS_RUNNING {
entry.SetStatus(STATUS_READY)
}
}()
//if entry.interval == 8 {
// fmt.Println(w.level, ticks, entry.create, entry.interval,
// entry.times.Val(),
// entry.Status(),
// time.Duration(w.interval),
// time.Duration(entry.interval)*time.Duration(w.interval),
// time.Now(),
// entry.id,
// )
//}
entry.job()
}(entry, e, l)
}(entry)
}
// 是否继续添运行
if entry.status.Val() != STATUS_CLOSED {
left := entry.interval - (nowTicks - entry.create)
if left <= 0 {
left = entry.interval
entry.create = nowTicks
entry.createMs = nowMs
entry.updateMs = nowMs
}
w.slots[(nowTicks + left) % w.number].PushBack(entry)
}
})
if len(removeArray) > 0 {
l.BatchRemove(removeArray)
}
}(l, n)
}

26
g/os/gwheel/gwheel_wheels.go
View File

@ -37,7 +37,7 @@ func New(slot int, interval time.Duration, level...int) *Wheels {
n := time.Duration(ws.levels[i - 1].totalMs)*time.Millisecond
w := ws.newWheel(i, slot, n)
ws.levels[i] = w
ws.levels[i - 1].newEntry(n, w.proceed, false, gDEFAULT_TIMES)
ws.levels[i - 1].addEntry(n, w.proceed, false, gDEFAULT_TIMES)
} else {
ws.levels[i] = ws.newWheel(i, slot, interval)
}
@ -67,22 +67,22 @@ func (ws *Wheels) newWheel(level int, slot int, interval time.Duration) *wheel {
// 添加循环任务
func (ws *Wheels) Add(interval time.Duration, job JobFunc) *Entry {
return ws.newEntry(interval, job, false, gDEFAULT_TIMES)
return ws.addEntry(interval, job, false, gDEFAULT_TIMES)
}
// 添加单例运行循环任务
func (ws *Wheels) AddSingleton(interval time.Duration, job JobFunc) *Entry {
return ws.newEntry(interval, job, true, gDEFAULT_TIMES)
return ws.addEntry(interval, job, true, gDEFAULT_TIMES)
}
// 添加只运行一次的循环任务
func (ws *Wheels) AddOnce(interval time.Duration, job JobFunc) *Entry {
return ws.newEntry(interval, job, false, 1)
return ws.addEntry(interval, job, false, 1)
}
// 添加运行指定次数的循环任务
func (ws *Wheels) AddTimes(interval time.Duration, times int, job JobFunc) *Entry {
return ws.newEntry(interval, job, false, times)
return ws.addEntry(interval, job, false, times)
}
// 延迟添加循环任务
@ -121,35 +121,31 @@ func (ws *Wheels) Close() {
}
// 添加循环任务
func (ws *Wheels) newEntry(interval time.Duration, job JobFunc, singleton bool, times int, from...*wheel) *Entry {
func (ws *Wheels) addEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
intervalMs := interval.Nanoseconds()/1e6
pos, cmp := ws.binSearchIndex(intervalMs)
if len(from) > 0 {
pos = from[0].level - 1
cmp = -1
}
switch cmp {
// n比最后匹配值小
case -1:
i := pos
for ; i > 0; i-- {
if intervalMs > ws.levels[i].totalMs {
return ws.levels[i].newEntry(interval, job, singleton, times)
return ws.levels[i].addEntry(interval, job, singleton, times)
}
}
return ws.levels[i].newEntry(interval, job, singleton, times)
return ws.levels[i].addEntry(interval, job, singleton, times)
// n比最后匹配值大
case 1:
i := pos
for ; i < ws.length - 1; i++ {
if intervalMs < ws.levels[i].totalMs {
return ws.levels[i].newEntry(interval, job, singleton, times)
return ws.levels[i].addEntry(interval, job, singleton, times)
}
}
return ws.levels[i].newEntry(interval, job, singleton, times)
return ws.levels[i].addEntry(interval, job, singleton, times)
case 0:
return ws.levels[pos].newEntry(interval, job, singleton, times)
return ws.levels[pos].addEntry(interval, job, singleton, times)
}
return nil
}

8
geg/os/gwheel/gwheel1.go
View File

@ -7,9 +7,11 @@ import (
)
func main() {
fmt.Println("START:", time.Now())
gwheel.Add(1400*time.Millisecond, func() {
fmt.Println(time.Now())
now := time.Now()
interval := 1400*time.Millisecond
gwheel.Add(interval, func() {
fmt.Println(time.Now(), time.Duration(time.Now().UnixNano() - now.UnixNano()))
now = time.Now()
})
select { }