Implement a timing wheel for millions of concurrent tasks.

timingWheel, err := NewTimingWheel(time.Second, slots, func(k, v interface{}) {  key, ok := k.(string)  if !ok {    return  }  cache.Del(key)})if err ! = nil {  return nil, err}cache.timingWheel = timingWheel

// The initialization of TimingWheelfunc newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (  *TimingWheel, error) {  tw := &TimingWheel{    interval: interval, // time frame interval    ticker: ticker, // the ticker to trigger the execution    slots: make([]*list.List, numSlots), // the slots to put tasks    timers: NewSafeMap(), // map to store task with {key, value}    tickedPos: numSlots - 1, // the previous ticked position    execute: execute, // execute function    numSlots: numSlots, // the number of slots    setChannel: make(chan timingEntry), // the channel to set tasks    moveChannel: make(chan baseEntry), // the channel to move tasks    removeChannel: make(chan interface{}), // the channel to remove tasks    drainChannel: make(chan func(key, value interface{})), // the channel to drain tasks    stopChannel: make(chan lang.PlaceholderType), // the channel to stop TimingWheel  }  // Prepare all the lists stored in the slot  tw.initSlots()  // start asynchronous concurrent process, use channel for task communication and passing  go tw.run()  return tw, nil}

func (tw *TimingWheel) run() {  for {    select {    // Timer does time push -> scanAndRunTasks()    case <-tw.ticker.Chan():      tw.onTick()    // add task will input task to setChannel    case task := <-tw.setChannel:      tw.setTask(&task)      ...    }  }}

func (c *Cache) Set(key string, value interface{}) {  c.lock.  _, ok := c.data[key]  c.data[key] = value  c.lruCache.add(key)  c.lock.Unlock()  expiry := c.unstableExpiry.AroundDuration(c.expiry)  if ok {    c.timingWheel.MoveTimer(key, expiry)  } else {    c.timingWheel.SetTimer(key, value, expiry)  }}

func (tw *TimingWheel) moveTask(task baseEntry) {  // timers: Map => get [positionEntry "pos, task"] by key  val, ok := tw.timers.Get(task.key)  if !ok {    return  }  timer := val.(*positionEntry)  // {delay < interval} => delay is less than a time frame interval,  // the task should be executed immediately  if task.delay < tw.interval {    threading.GoSafe(func() {      tw.execute(timer.item.key, timer.item.value)    })    return  }  // If > interval, calculate the new pos, circle out of the time wheel by delaying the time  pos, circle := tw.getPositionAndCircle(task.delay)  if pos >= timer.pos {    timer.item.circle = circle    // Record the offset of the move    timer.item.diff = pos - timer.pos  } else if circle > 0 {    // move to the level of (circle-1)    circle --    timer.item.circle = circle    // because it's an array, add numSlots [which is the equivalent of going to the next level]    timer.item.diff = tw.numSlots + pos - timer.pos  } else {    // If offset is ahead of schedule, task is still on the first level    // mark the old task for deletion and requeue it for execution    timer.item.removed = true    newItem := &timingEntry{      baseEntry: task,      value: timer.item.value,    }    tw.slots[pos].PushBack(newItem)    tw.setTimerPosition(pos, newItem)  }}

// Timer 'execute every internal'func (tw *TimingWheel) onTick() {  // update the current tick position on every execution  tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots  // Get the chain of tasks stored in the tick position at this time  l := tw.slots[tw.tickedPos]  tw.scanAndRunTasks(l)}

func (tw *TimingWheel) scanAndRunTasks(l *list.List) {  // store the tasks {key, value} that currently need to be executed [the arguments needed by execute, passed in turn to execute]  var tasks []timingTask  for e := l.Front(); e ! = nil; {    task := e.Value.(*timingEntry)    // mark for deletion, do the real deletion in scan, delete data from map    if task.removed {      next := e.Next()      l.Remove(e)      tw.timers.Del(task.key)      e = next      continue    } else if task.circle > 0 {      // the current execution point has expired, but at the same time it's not in the first level,      // so even though the current level is done, it drops to the next level      // but it doesn't modify pos      task.circle--      e = e.Next()      continue    } else if task.diff > 0 {      // because the diff has already been marked, it needs to go into the queue again      next := e.Next()      l.Remove(e)      pos := (tw.tickedPos + task.diff) % tw.numSlots      tw.slots[pos].PushBack(task)      tw.setTimerPosition(pos, task)      task.diff = 0      e = next      continue    }    // the above cases are all non-executable cases, those that can be executed will be added to tasks    tasks = append(tasks, timingTask{      key: task.key,      value: task.value,    })    next := e.Next()    l.Remove(e)    tw.timers.Del(task.key)    e = next  }  // for range tasks, then just execute each task->execute  tw.runTasks(tasks)}

// interval: 4min, d: 60min, numSlots: 16, tickedPos = 15// step = 15, pos = 14, circle = 0func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) {  steps := int(d / tw.interval)  pos = (tw.tickedPos + steps) % tw.numSlots  circle = (steps - 1) / tw.numSlots  return}