From 241706cbbf8f7eb944c198ef23861b9fed8dd6c7 Mon Sep 17 00:00:00 2001 From: John Date: Sat, 12 Jan 2019 20:20:30 +0800 Subject: [PATCH] up --- g/container/glist/glist_z_bench_test.go | 10 ++- g/os/gwheel/gwheel.go | 4 +- g/os/gwheel/gwheel_entry.go | 83 +++++++++++-------------- g/os/gwheel/gwheel_loop.go | 69 +++++++++----------- g/os/gwheel/gwheel_wheels.go | 26 ++++---- geg/os/gwheel/gwheel1.go | 8 ++- 6 files changed, 92 insertions(+), 108 deletions(-) diff --git a/g/container/glist/glist_z_bench_test.go b/g/container/glist/glist_z_bench_test.go index d2d941402..9228f3992 100644 --- a/g/container/glist/glist_z_bench_test.go +++ b/g/container/glist/glist_z_bench_test.go @@ -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() } diff --git a/g/os/gwheel/gwheel.go b/g/os/gwheel/gwheel.go index 6aecdcba7..0290de409 100644 --- a/g/os/gwheel/gwheel.go +++ b/g/os/gwheel/gwheel.go @@ -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 ( diff --git a/g/os/gwheel/gwheel_entry.go b/g/os/gwheel/gwheel_entry.go index 2625519ae..0fb11318f 100644 --- a/g/os/gwheel/gwheel_entry.go +++ b/g/os/gwheel/gwheel_entry.go @@ -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 diff --git a/g/os/gwheel/gwheel_loop.go b/g/os/gwheel/gwheel_loop.go index 9c2204206..74f9d90b9 100644 --- a/g/os/gwheel/gwheel_loop.go +++ b/g/os/gwheel/gwheel_loop.go @@ -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) } diff --git a/g/os/gwheel/gwheel_wheels.go b/g/os/gwheel/gwheel_wheels.go index dd5bf3ac7..f6e7972e2 100644 --- a/g/os/gwheel/gwheel_wheels.go +++ b/g/os/gwheel/gwheel_wheels.go @@ -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 } diff --git a/geg/os/gwheel/gwheel1.go b/geg/os/gwheel/gwheel1.go index 4aebce633..43fc75e2b 100644 --- a/geg/os/gwheel/gwheel1.go +++ b/geg/os/gwheel/gwheel1.go @@ -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 { }