redhat/timewheel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add timing mode

Browse Source
This commit is contained in:
redhat 2025-06-10 09:08:14 +08:00
parent 30cec6eb5a
commit 2f36f2338b
2 changed files with 37 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
timewheel.go
View File

@ -7,10 +7,12 @@ import (
) )
type taskSlice struct { type taskSlice struct {
id string id string
pos int pos int
cycle int cycle int
task func() rawCycle int
mode TimeMode
task func()
} }
type TimeWheel struct { type TimeWheel struct {
@ -25,6 +27,13 @@ type TimeWheel struct {
removeTaskC chan string removeTaskC chan string
} }
type TimeMode int
const (
TimerTypeOnce TimeMode = iota
TimeTypeLoop
)
func New(slots int, interval time.Duration) *TimeWheel { func New(slots int, interval time.Duration) *TimeWheel {
if slots <= 0 { if slots <= 0 {
slots = 10 slots = 10
@ -38,7 +47,7 @@ func New(slots int, interval time.Duration) *TimeWheel {
interval: interval, interval: interval,
ticker: time.NewTicker(interval), ticker: time.NewTicker(interval),
slots: make([]*list.List, 0, slots), slots: make([]*list.List, 0, slots),
taskMap: map[string]*list.Element{}, taskMap: make(map[string]*list.Element),
stopC: make(chan struct{}), stopC: make(chan struct{}),
addTaskC: make(chan *taskSlice), addTaskC: make(chan *taskSlice),
removeTaskC: make(chan string), removeTaskC: make(chan string),
@ -53,14 +62,16 @@ func New(slots int, interval time.Duration) *TimeWheel {
return t return t
} }
func (t *TimeWheel) AddTask(id string, task func(), flower time.Time) { func (t *TimeWheel) AddTask(id string, mode TimeMode, task func(), flower time.Time) {
cycle, pos := t.getSlotPosAndCycle(flower) cycle, pos := t.getSlotPosAndCycle(flower)
t.addTaskC <- &taskSlice{ t.addTaskC <- &taskSlice{
id: id, id: id,
task: task, task: task,
cycle: cycle, cycle: cycle,
pos: pos, mode: mode,
rawCycle: cycle,
pos: pos,
} }
} }
@ -110,7 +121,7 @@ func (t *TimeWheel) updateCurtSlot() {
} }
func (t *TimeWheel) execute(l *list.List) { func (t *TimeWheel) execute(l *list.List) {
for e := l.Back(); e != nil; { for e := l.Front(); e != nil; {
event := e.Value.(*taskSlice) event := e.Value.(*taskSlice)
if event.cycle > 0 { if event.cycle > 0 {
event.cycle-- event.cycle--
@ -128,10 +139,15 @@ func (t *TimeWheel) execute(l *list.List) {
event.task() event.task()
}() }()
next := e.Next() if event.mode == TimerTypeOnce {
l.Remove(e) next := e.Next()
delete(t.taskMap, event.id) l.Remove(e)
e = next delete(t.taskMap, event.id)
e = next
} else {
e = e.Next()
event.cycle = event.rawCycle
}
} }
} }

10
timewheel_test.go
View File

@ -6,17 +6,19 @@ import (
) )
func Test_timeWheel(t *testing.T) { func Test_timeWheel(t *testing.T) {
timeWheel := New(10, 500*time.Millisecond) timeWheel := New(10, 100*time.Millisecond)
defer timeWheel.Stop() defer timeWheel.Stop()
<-time.After(10 * time.Millisecond)
t.Errorf("test2, %v", time.Now()) t.Errorf("test2, %v", time.Now())
timeWheel.AddTask("test1", func() { timeWheel.AddTask("test1", TimeTypeLoop, func() {
t.Errorf("test1, %v", time.Now()) t.Errorf("test1, %v", time.Now())
}, time.Now().Add(time.Second)) }, time.Now().Add(time.Second))
timeWheel.AddTask("test2", func() { timeWheel.AddTask("test2", TimeTypeLoop, func() {
t.Errorf("test2, %v", time.Now()) t.Errorf("test2, %v", time.Now())
}, time.Now().Add(5*time.Second)) }, time.Now().Add(5*time.Second))
timeWheel.AddTask("test2", func() { timeWheel.AddTask("test2", TimerTypeOnce, func() {
t.Errorf("test2, %v", time.Now()) t.Errorf("test2, %v", time.Now())
}, time.Now().Add(3*time.Second)) }, time.Now().Add(3*time.Second))