diff --git a/TODO.MD b/TODO.MD index a314ae96e..473307f16 100644 --- a/TODO.MD +++ b/TODO.MD @@ -1,8 +1,4 @@ # ON THE WAY -1. orm增加更多数据库支持； -1. 增加对于数据表Model的封装； -1. 更多数据库的ORM功能支持； -1. 考虑gdb对象管理增加二级连接池特性，提高New&Close性能； 1. 增加图形验证码支持，至少支持数字和英文字母； 1. 增加热编译工具，提高开发环境的开发/测试效率（媲美PHP开发效率）； 1. 增加可选择性的orm tag特性，用以数据表记录与struct对象转换的键名属性映射； @@ -51,8 +47,8 @@ 1. glist增加Element类型的并发安全处理； 1. 改进证书打开失败时的WebServer错误提示，前置HOOK校验后关闭后续的HOOK逻辑执行； 1. 目前WebServer的HOOK是按照优先级执行的，需要增加覆盖特性； - - +1. 增加jumplist的数据结构容器； +1. DelayQueue/PriorityQueue； # DONE diff --git a/g/container/garray/garray_z_bench_test.go b/g/container/garray/garray_z_bench_test.go new file mode 100644 index 000000000..26ac1b0ab --- /dev/null +++ b/g/container/garray/garray_z_bench_test.go @@ -0,0 +1,43 @@ +// Copyright 2017 gf Author(https://gitee.com/johng/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://gitee.com/johng/gf. + +// go test *.go -bench=".*" -benchmem + +package garray_test + +import ( + "gitee.com/johng/gf/g/container/garray" + "testing" +) + +var ( + sortedIntArray = garray.NewSortedIntArray(0) +) + +func BenchmarkSortedIntArray_Add(b *testing.B) { + b.N = 1000 + for i := 0; i < b.N; i++ { + sortedIntArray.Add(i) + } +} + +func BenchmarkSortedIntArray_Search(b *testing.B) { + for i := 0; i < b.N; i++ { + sortedIntArray.Search(i) + } +} + +func BenchmarkSortedIntArray_PopLeft(b *testing.B) { + for i := 0; i < b.N; i++ { + sortedIntArray.PopLeft() + } +} + +func BenchmarkSortedIntArray_PopRight(b *testing.B) { + for i := 0; i < b.N; i++ { + sortedIntArray.PopLeft() + } +} diff --git a/g/container/garray/garray_test.go b/g/container/garray/garray_z_unit_test.go similarity index 100% rename from g/container/garray/garray_test.go rename to g/container/garray/garray_z_unit_test.go diff --git a/g/container/glist/glist.go b/g/container/glist/glist.go index 4c1a4fb99..8775603af 100644 --- a/g/container/glist/glist.go +++ b/g/container/glist/glist.go @@ -272,6 +272,16 @@ func (l *List) Remove(e *Element) (value interface{}) { return } +// 批量删除数据项 +func (l *List) BatchRemove(es []*Element) { + l.mu.Lock() + for _, e := range es { + l.list.Remove(e) + } + l.mu.Unlock() + return +} + // 删除所有数据项 func (l *List) RemoveAll() { l.mu.Lock() diff --git a/g/container/gring/gring.go b/g/container/gring/gring.go index 7f126b56d..da71a585c 100644 --- a/g/container/gring/gring.go +++ b/g/container/gring/gring.go @@ -4,8 +4,7 @@ // If a copy of the MIT was not distributed with this file, // You can obtain one at https://gitee.com/johng/gf. -// Package gring provides a concurrent-safe(alternative) ring(circular lists). -// 并发安全的环. +// Package gring provides a concurrent-safe(alternative) ring(circular lists)/并发安全的环. package gring import ( diff --git a/g/os/gwheel/gwheel.go b/g/os/gwheel/gwheel.go index f850e557a..6aecdcba7 100644 --- a/g/os/gwheel/gwheel.go +++ b/g/os/gwheel/gwheel.go @@ -20,7 +20,7 @@ const ( STATUS_RUNNING = 1 STATUS_CLOSED = -1 gPANIC_EXIT = "exit" - gDEFAULT_TIMES = math.MaxInt64 + gDEFAULT_TIMES = math.MaxInt32 gDEFAULT_SLOT_NUMBER = 10 gDEFAULT_WHEEL_INTERVAL = 50*time.Millisecond gDEFAULT_WHEEL_LEVEL = 10 @@ -28,47 +28,47 @@ const ( var ( // 默认的wheel管理对象 - defaultWheel = NewDefault() + defaultWheels = New(gDEFAULT_SLOT_NUMBER, gDEFAULT_WHEEL_INTERVAL, gDEFAULT_WHEEL_LEVEL) ) // 添加执行方法，可以给定名字，以便于后续执行删除 func Add(interval time.Duration, job JobFunc) *Entry { - return defaultWheel.Add(interval, job) + return defaultWheels.Add(interval, job) } // 添加单例运行循环任务 func AddSingleton(interval time.Duration, job JobFunc) *Entry { - return defaultWheel.AddSingleton(interval, job) + return defaultWheels.AddSingleton(interval, job) } // 添加只运行一次的循环任务 func AddOnce(interval time.Duration, job JobFunc) *Entry { - return defaultWheel.AddOnce(interval, job) + return defaultWheels.AddOnce(interval, job) } // 添加运行指定次数的循环任务 func AddTimes(interval time.Duration, times int, job JobFunc) *Entry { - return defaultWheel.AddTimes(interval, times, job) + return defaultWheels.AddTimes(interval, times, job) } // 延迟添加循环任务，delay参数单位为秒 func DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) { - defaultWheel.DelayAdd(delay, interval, job) + defaultWheels.DelayAdd(delay, interval, job) } // 延迟添加单例循环任务，delay参数单位为秒 func DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) { - defaultWheel.DelayAddSingleton(delay, interval, job) + defaultWheels.DelayAddSingleton(delay, interval, job) } // 延迟添加只运行一次的循环任务，delay参数单位为秒 func DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) { - defaultWheel.DelayAddOnce(delay, interval, job) + defaultWheels.DelayAddOnce(delay, interval, job) } // 延迟添加运行指定次数的循环任务，delay参数单位为秒 func DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) { - defaultWheel.DelayAddTimes(delay, interval, times, job) + defaultWheels.DelayAddTimes(delay, interval, times, job) } // 在Job方法中调用，停止当前运行的任务 diff --git a/g/os/gwheel/gwheel_entry.go b/g/os/gwheel/gwheel_entry.go index ca6207293..2625519ae 100644 --- a/g/os/gwheel/gwheel_entry.go +++ b/g/os/gwheel/gwheel_entry.go @@ -13,42 +13,59 @@ import ( // 循环任务项 type Entry struct { - singleton *gtype.Bool // 任务是否单例运行 - status *gtype.Int // 任务状态(0: ready; 1: running; -1: closed) - times *gtype.Int64 // 还需运行次数 - create int64 // 注册时的时间轮ticks - interval int64 // 设置的运行间隔(时间轮刻度数量) - job JobFunc // 注册循环任务方法 + id int64 // ID + job JobFunc // 注册循环任务方法 + wheel *wheel // 所属时间轮 + singleton *gtype.Bool // 任务是否单例运行 + status *gtype.Int // 任务状态(0: ready; 1: running; -1: closed) + times *gtype.Int // 还需运行次数 + create int64 // 注册时的时间轮ticks + update *gtype.Int64 // 更新时间(上一次检查/运行时间, 毫秒) + interval int64 // 设置的运行间隔(时间轮刻度数量) + createMs int64 // 创建时间(毫秒) + intervalMs int64 // 间隔时间(毫秒) } // 任务执行方法 type JobFunc func() -// 创建循环任务 +// 创建循环任务(失败返回nil) func (w *wheel) newEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry { - // 安装任务的间隔时间(纳秒) - n := interval.Nanoseconds() - // 计算出所需的插槽数量 - num := int(n/w.interval) + ims := interval.Nanoseconds()/1e6 + num := ims/w.intervalMs if num == 0 { - // 如果添加的任务间隔时间比时间轮的刻度还小， - // 那么默认为1个刻度 - num = 1 + return nil } + nowMs := time.Now().UnixNano()/1e6 ticks := w.ticks.Val() entry := &Entry { - singleton : gtype.NewBool(singleton), - status : gtype.NewInt(STATUS_READY), - times : gtype.NewInt64(int64(times)), - job : job, - create : ticks, - interval : int64(num), + id : time.Now().UnixNano(), + wheel : w, + singleton : gtype.NewBool(singleton), + status : gtype.NewInt(STATUS_READY), + times : gtype.NewInt(times), + job : job, + create : ticks, + update : gtype.NewInt64(nowMs), + interval : num, + createMs : nowMs, + intervalMs : ims, } - // 计算安装的slot数量(可能多个) - index := int(ticks%int64(w.number)) - for i := 0; i < w.number; i += num { - w.slots[(i + index + num) % w.number].PushBack(entry) + // 安装的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()) return entry } @@ -79,7 +96,7 @@ func (entry *Entry) SetSingleton(enabled bool) { // 设置任务的运行次数 func (entry *Entry) SetTimes(times int) { - entry.times.Set(int64(times)) + entry.times.Set(times) } // 执行任务 @@ -88,9 +105,8 @@ func (entry *Entry) Run() { } // 检测当前任务是否可运行, 参数为当前时间的纳秒数, 精度更高 -func (entry *Entry) runnableCheck(ticks int64) bool { - diff := ticks - entry.create - if diff > 0 && diff%int64(entry.interval) == 0 { +func (entry *Entry) runnableCheck(nowTicks int64, nowMs int64) bool { + if diff := nowTicks - entry.create; diff > 0 && diff%entry.interval == 0 { // 是否关闭 if entry.status.Val() == STATUS_CLOSED { return false @@ -102,11 +118,37 @@ func (entry *Entry) runnableCheck(ticks int64) bool { } } // 次数限制 - if entry.times.Add(-1) < 0 { - if entry.status.Set(STATUS_CLOSED) == STATUS_CLOSED { + times := entry.times.Add(-1) + if times <= 0 { + entry.status.Set(STATUS_CLOSED) + if times < 0 { return false } } + // 是否不限制运行次数 + if times > 2000000000 { + 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) + } + entry.status.Set(STATUS_CLOSED) + return false + } + } + entry.update.Set(nowMs) return true } return false diff --git a/g/os/gwheel/gwheel_jobloop.go b/g/os/gwheel/gwheel_loop.go similarity index 53% rename from g/os/gwheel/gwheel_jobloop.go rename to g/os/gwheel/gwheel_loop.go index 89d7a66a2..9c2204206 100644 --- a/g/os/gwheel/gwheel_jobloop.go +++ b/g/os/gwheel/gwheel_loop.go @@ -9,17 +9,20 @@ package gwheel import ( "container/list" "gitee.com/johng/gf/g/container/glist" + "time" ) // 开始循环 func (w *wheel) start() { go func() { + ticker := time.NewTicker(time.Duration(w.intervalMs)*time.Millisecond) for { select { case <- w.closed: + ticker.Stop() return - case <- w.ticker.C: + case <- ticker.C: w.proceed() } } @@ -29,18 +32,26 @@ func (w *wheel) start() { // 执行时间轮刻度逻辑, 遍历检查可执行循环任务，并异步执行 func (w *wheel) proceed() { n := w.ticks.Add(1) - l := w.slots[int(n%int64(w.number))] + l := w.slots[int(n%w.number)] if l.Len() > 0 { - go func(l *glist.List, ticks int64) { + 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(ticks) { + if !entry.runnableCheck(nowTicks, nowMs) { continue } // 异步执行运行 - go func(e *glist.Element, l *glist.List) { + go func(entry *Entry, e *glist.Element, l *glist.List) { defer func() { if err := recover(); err != nil { if err != gPANIC_EXIT { @@ -51,18 +62,32 @@ func (w *wheel) proceed() { } switch entry.Status() { case STATUS_CLOSED: + //fmt.Println("remove:", w.level, entry.interval) l.Remove(e) case STATUS_RUNNING: entry.SetStatus(STATUS_READY) - } }() - entry.Run() - }(e, l) + //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) } }) + if len(removeArray) > 0 { + l.BatchRemove(removeArray) + } }(l, n) } } diff --git a/g/os/gwheel/gwheel_wheel.go b/g/os/gwheel/gwheel_wheel.go deleted file mode 100644 index 8f063da76..000000000 --- a/g/os/gwheel/gwheel_wheel.go +++ /dev/null @@ -1,107 +0,0 @@ -// Copyright 2019 gf Author(https://gitee.com/johng/gf). All Rights Reserved. -// -// This Source Code Form is subject to the terms of the MIT License. -// If a copy of the MIT was not distributed with this file, -// You can obtain one at https://gitee.com/johng/gf. - -package gwheel - -import ( - "gitee.com/johng/gf/g/container/glist" - "gitee.com/johng/gf/g/container/gtype" - "time" -) - -// 单层时间轮 -type wheel struct { - slots []*glist.List // 所有的循环任务项, 按照Slot Number进行分组 - number int // Slot Number - closed chan struct{} // 停止事件 - ticks *gtype.Int64 // 当前时间轮已转动的刻度数量 - ticker *time.Ticker // 时间轮刻度间隔 - interval int64 // 时间间隔(slot时间长度, 纳秒) -} - - -// 创建使用默认值的时间轮 -func NewDefault() *wheel { - return newWheel(gDEFAULT_SLOT_NUMBER, gDEFAULT_WHEEL_INTERVAL) -} - -// 创建自定义的循环任务管理对象 -func newWheel(slot int, interval time.Duration) *wheel { - w := &wheel { - slots : make([]*glist.List, slot), - number : slot, - closed : make(chan struct{}, 1), - ticks : gtype.NewInt64(), - ticker : time.NewTicker(interval), - interval : interval.Nanoseconds(), - } - for i := 0; i < w.number; i++ { - w.slots[i] = glist.New() - } - return w -} - -// 添加循环任务 -func (w *wheel) Add(interval time.Duration, job JobFunc) *Entry { - return w.newEntry(interval, job, false, gDEFAULT_TIMES) -} - -// 添加单例运行循环任务 -func (w *wheel) AddSingleton(interval time.Duration, job JobFunc) *Entry { - return w.newEntry(interval, job, true, gDEFAULT_TIMES) -} - -// 添加只运行一次的循环任务 -func (w *wheel) AddOnce(interval time.Duration, job JobFunc) *Entry { - return w.newEntry(interval, job, false, 1) -} - -// 添加运行指定次数的循环任务 -func (w *wheel) AddTimes(interval time.Duration, times int, job JobFunc) *Entry { - return w.newEntry(interval, job, false, times) -} - -// 延迟添加循环任务 -func (w *wheel) DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) { - w.AddOnce(delay, func() { - w.Add(interval, job) - }) -} - -// 延迟添加单例循环任务 -func (w *wheel) DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) { - w.AddOnce(delay, func() { - w.AddSingleton(interval, job) - }) -} - -// 延迟添加只运行一次的循环任务 -func (w *wheel) DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) { - w.AddOnce(delay, func() { - w.AddOnce(interval, job) - }) -} - -// 延迟添加只运行一次的循环任务 -func (w *wheel) DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) { - w.AddOnce(delay, func() { - w.AddTimes(interval, times, job) - }) -} - -// 任务数量 -func (w *wheel) Size() (size int) { - for _, l := range w.slots { - size += l.Len() - } - return -} - -// 关闭循环任务 -func (w *wheel) Close() { - w.ticker.Stop() - w.closed <- struct{}{} -} diff --git a/g/os/gwheel/gwheel_wheels.go b/g/os/gwheel/gwheel_wheels.go index d0e6d4c4d..dd5bf3ac7 100644 --- a/g/os/gwheel/gwheel_wheels.go +++ b/g/os/gwheel/gwheel_wheels.go @@ -7,15 +7,17 @@ package gwheel import ( + "gitee.com/johng/gf/g/container/glist" + "gitee.com/johng/gf/g/container/gtype" "time" ) // 分层时间轮 type Wheels struct { - levels []*wheel // 分层 - length int // 层数 - number int // 每一层Slot Number - interval int64 // 最小时间刻度(纳秒) + levels []*wheel // 分层 + length int // 层数 + number int // 每一层Slot Number + intervalMs int64 // 最小时间刻度(毫秒) } // 创建分层时间轮 @@ -24,64 +26,146 @@ func New(slot int, interval time.Duration, level...int) *Wheels { if len(level) > 0 { length = level[0] } - w := &Wheels { - levels : make([]*wheel, length), - length : length, - number : slot, - interval : interval.Nanoseconds(), + ws := &Wheels { + levels : make([]*wheel, length), + length : length, + number : slot, + intervalMs : interval.Nanoseconds()/1e6, } for i := 0; i < length; i++ { if i > 0 { - n := interval*time.Duration(slot*i) - w.levels[i] = newWheel(slot, n) - w.levels[i - 1].Add(n, func() { - w.levels[i].proceed() - }) + 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) } else { - w.levels[i] = newWheel(slot, interval) + ws.levels[i] = ws.newWheel(i, slot, interval) } } - w.levels[0].start() + ws.levels[0].start() + return ws +} + +// 创建自定义的循环任务管理对象 +func (ws *Wheels) newWheel(level int, slot int, interval time.Duration) *wheel { + w := &wheel { + wheels : ws, + level : level, + slots : make([]*glist.List, slot), + number : int64(slot), + closed : make(chan struct{}, 1), + ticks : gtype.NewInt64(), + totalMs : int64(slot)*interval.Nanoseconds()/1e6, + createMs : time.Now().UnixNano()/1e6, + intervalMs : interval.Nanoseconds()/1e6, + } + for i := int64(0); i < w.number; i++ { + w.slots[i] = glist.New() + } return w } // 添加循环任务 -func (w *Wheels) Add(interval time.Duration, job JobFunc) *Entry { - n := interval.Nanoseconds() - pos, cmp := w.binSearchIndex(n) +func (ws *Wheels) Add(interval time.Duration, job JobFunc) *Entry { + return ws.newEntry(interval, job, false, gDEFAULT_TIMES) +} + +// 添加单例运行循环任务 +func (ws *Wheels) AddSingleton(interval time.Duration, job JobFunc) *Entry { + return ws.newEntry(interval, job, true, gDEFAULT_TIMES) +} + +// 添加只运行一次的循环任务 +func (ws *Wheels) AddOnce(interval time.Duration, job JobFunc) *Entry { + return ws.newEntry(interval, job, false, 1) +} + +// 添加运行指定次数的循环任务 +func (ws *Wheels) AddTimes(interval time.Duration, times int, job JobFunc) *Entry { + return ws.newEntry(interval, job, false, times) +} + +// 延迟添加循环任务 +func (ws *Wheels) DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) { + ws.AddOnce(delay, func() { + ws.Add(interval, job) + }) +} + +// 延迟添加单例循环任务 +func (ws *Wheels) DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) { + ws.AddOnce(delay, func() { + ws.AddSingleton(interval, job) + }) +} + +// 延迟添加只运行一次的循环任务 +func (ws *Wheels) DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) { + ws.AddOnce(delay, func() { + ws.AddOnce(interval, job) + }) +} + +// 延迟添加只运行一次的循环任务 +func (ws *Wheels) DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) { + ws.AddOnce(delay, func() { + ws.AddTimes(interval, times, job) + }) +} + +// 关闭分层时间轮 +func (ws *Wheels) Close() { + for _, w := range ws.levels { + w.Close() + } +} + +// 添加循环任务 +func (ws *Wheels) newEntry(interval time.Duration, job JobFunc, singleton bool, times int, from...*wheel) *Entry { + intervalMs := interval.Nanoseconds()/1e6 + pos, cmp := ws.binSearchIndex(intervalMs) + if len(from) > 0 { + pos = from[0].level - 1 + cmp = -1 + } switch cmp { - case -1 : - for i := pos; i >= 0; i-- { - if n > w.levels[i].interval { - return w.levels[i].Add(time.Duration(n), job) + // 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 w.levels[0].Add(time.Duration(n), job) - case 1 : - for i := pos; i < w.length; i++ { - if n > w.levels[i].interval { - return w.levels[i].Add(time.Duration(n), job) + return ws.levels[i].newEntry(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 w.levels[w.length - 1].Add(time.Duration(n), job) - case 0 : - return w.levels[pos].Add(time.Duration(n), job) + return ws.levels[i].newEntry(interval, job, singleton, times) + + case 0: + return ws.levels[pos].newEntry(interval, job, singleton, times) } return nil } -func (w *Wheels) binSearchIndex(n int64)(index int, result int) { +func (ws *Wheels) binSearchIndex(n int64)(index int, result int) { min := 0 - max := w.length - 1 + max := ws.length - 1 mid := 0 cmp := -2 for min <= max { mid = int((min + max) / 2) switch { - case w.levels[mid].interval == n : cmp = 0 - case w.levels[mid].interval > n : cmp = 1 - case w.levels[mid].interval < n : cmp = -1 + case ws.levels[mid].totalMs == n : cmp = 0 + case ws.levels[mid].totalMs > n : cmp = -1 + case ws.levels[mid].totalMs < n : cmp = 1 } switch cmp { case -1 : max = mid - 1 diff --git a/g/os/gwheel/gwheel_wheels_wheel.go b/g/os/gwheel/gwheel_wheels_wheel.go new file mode 100644 index 000000000..e15e1b12d --- /dev/null +++ b/g/os/gwheel/gwheel_wheels_wheel.go @@ -0,0 +1,30 @@ +// Copyright 2019 gf Author(https://gitee.com/johng/gf). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://gitee.com/johng/gf. + +package gwheel + +import ( + "gitee.com/johng/gf/g/container/glist" + "gitee.com/johng/gf/g/container/gtype" +) + +// 单层时间轮 +type wheel struct { + wheels *Wheels // 所属分层时间轮 + level int // 所属分层索引号 + slots []*glist.List // 所有的循环任务项, 按照Slot Number进行分组 + number int64 // Slot Number + closed chan struct{} // 停止事件 + ticks *gtype.Int64 // 当前时间轮已转动的刻度数量 + totalMs int64 // 整个时间轮的时间长度(毫秒)=number*interval + createMs int64 // 创建时间(毫秒) + intervalMs int64 // 时间间隔(slot时间长度, 毫秒) +} + +// 关闭循环任务 +func (w *wheel) Close() { + w.closed <- struct{}{} +} \ No newline at end of file diff --git a/g/os/gwheel/gwheel_z_unit_1_test.go b/g/os/gwheel/gwheel_z_unit_1_test.go index e2cb9f470..1802825dc 100644 --- a/g/os/gwheel/gwheel_z_unit_1_test.go +++ b/g/os/gwheel/gwheel_z_unit_1_test.go @@ -16,9 +16,14 @@ import ( "time" ) + +func New() *gwheel.Wheels { + return gwheel.New(10, 10*time.Millisecond) +} + func TestWheel_Add_Close(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) //fmt.Println("start", time.Now()) wheel.Add(time.Second, func() { @@ -47,7 +52,7 @@ func TestWheel_Add_Close(t *testing.T) { func TestWheel_Singleton(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.AddSingleton(time.Second, func() { array.Append(1) @@ -63,7 +68,7 @@ func TestWheel_Singleton(t *testing.T) { func TestWheel_Once(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.AddOnce(time.Second, func() { array.Append(1) @@ -85,7 +90,7 @@ func TestWheel_Once(t *testing.T) { func TestWheel_DelayAdd(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.DelayAdd(time.Second, time.Second, func() { array.Append(1) @@ -99,7 +104,7 @@ func TestWheel_DelayAdd(t *testing.T) { func TestWheel_DelayAdd_Singleton(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.DelayAddSingleton(time.Second, time.Second, func() { array.Append(1) @@ -115,7 +120,7 @@ func TestWheel_DelayAdd_Singleton(t *testing.T) { func TestWheel_DelayAdd_Once(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.DelayAddOnce(time.Second, time.Second, func() { array.Append(1) @@ -133,7 +138,7 @@ func TestWheel_DelayAdd_Once(t *testing.T) { func TestWheel_ExitJob(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.Add(time.Second, func() { array.Append(1) diff --git a/g/os/gwheel/gwheel_z_unit_2_test.go b/g/os/gwheel/gwheel_z_unit_2_test.go index 997d52159..2595c15af 100644 --- a/g/os/gwheel/gwheel_z_unit_2_test.go +++ b/g/os/gwheel/gwheel_z_unit_2_test.go @@ -10,14 +10,13 @@ package gwheel_test import ( "gitee.com/johng/gf/g/container/garray" - "gitee.com/johng/gf/g/os/gwheel" "gitee.com/johng/gf/g/util/gtest" "testing" "time" ) func TestWheel_Entry_Operation(t *testing.T) { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) entry := wheel.Add(time.Second, func() { array.Append(1) @@ -30,7 +29,7 @@ func TestWheel_Entry_Operation(t *testing.T) { } func TestWheel_Entry_Singleton(t *testing.T) { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) entry := wheel.Add(time.Second, func() { array.Append(1) @@ -45,7 +44,7 @@ func TestWheel_Entry_Singleton(t *testing.T) { } func TestWheel_Entry_Once(t *testing.T) { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) entry := wheel.Add(time.Second, func() { array.Append(1) diff --git a/g/os/gwheel/gwheel_z_unit_3_test.go b/g/os/gwheel/gwheel_z_unit_3_test.go index 5c794fdbf..eba64b243 100644 --- a/g/os/gwheel/gwheel_z_unit_3_test.go +++ b/g/os/gwheel/gwheel_z_unit_3_test.go @@ -10,7 +10,6 @@ package gwheel_test import ( "gitee.com/johng/gf/g/container/garray" - "gitee.com/johng/gf/g/os/gwheel" "gitee.com/johng/gf/g/util/gtest" "testing" "time" @@ -18,7 +17,7 @@ import ( func TestWheel_Times(t *testing.T) { gtest.Case(t, func() { - wheel := gwheel.NewDefault() + wheel := New() array := garray.New(0, 0) wheel.AddTimes(time.Second, 2, func() { array.Append(1) diff --git a/geg/os/gwheel/gwheel1.go b/geg/os/gwheel/gwheel1.go index b9dda2edd..4aebce633 100644 --- a/geg/os/gwheel/gwheel1.go +++ b/geg/os/gwheel/gwheel1.go @@ -7,9 +7,10 @@ import ( ) func main() { - _, err := gwheel.Add(time.Second, func() { + fmt.Println("START:", time.Now()) + gwheel.Add(1400*time.Millisecond, func() { fmt.Println(time.Now()) }) - fmt.Println(err) + select { } } diff --git a/geg/other/test.go b/geg/other/test.go index ee0a92e73..89b37603b 100644 --- a/geg/other/test.go +++ b/geg/other/test.go @@ -1,26 +1,12 @@ package main import ( - "gitee.com/johng/gf/g/container/gtype" - "gitee.com/johng/gf/g/os/glog" - "gitee.com/johng/gf/g/os/gwheel" - "time" + "fmt" ) func main() { - v := gtype.NewInt() - w := gwheel.New(1, 10*time.Millisecond) - glog.Println("start") - for i := 0; i < 100000; i++ { - w.AddOnce(time.Second, func() { - v.Add(1) - }) - } - glog.Println("end") - time.Sleep(1020*time.Millisecond) - glog.Println(v.Val()) - //gwheel.AddSingleton(time.Second, func() { - // fmt.Println(time.Now().String()) - //}) - //select { } + i := int64(5) + n := int64(3) + m := i/n + fmt.Println(m) }