func (m *taskQueue[R]) peek() (t Task[R]) {
	m.mu.Lock()
	if m.heap.Len() == 0 {
		m.mu.Unlock()
		return nil
	}
	t = (*m.heap)[0].value
	m.mu.Unlock()
	return
}
peek the top priority item without deletion
package sched
import (
	"fmt"
	"sync"
	"time"
	"container/heap"
	"context"
	"runtime"
	"sync/atomic"
	"unsafe"
)
type Future [R any] struct {
	Err error
	value atomic.Value
}
func (f *Future[R]) Get() R {
	var v interface{}
	for ; v == nil && f.Err == nil; v = f.value.Load() {
		runtime.Gosched()
	}
	return v.(R)
}
func (f *Future[R]) put(v R) {
	f.value.Store(v)
}
type Task [R any] interface {
	GetID() (id string)
	GetExecution() (execute time.Time)
	GetRetryTime() (execute time.Time)
	Execute() (result R, retry bool, fail error)
}
type sched [R any, T Task[R]] struct {
	running uint64  
	pausing uint64  
	timer unsafe.Pointer  
	cancel atomic.Value  
	tasks *taskQueue[R]
}
func NewSched[R any, T Task[R]]() *sched[R, T] {
	return &sched[R, T]{
		timer: unsafe.Pointer(time.NewTimer(0)),
		tasks: newTaskQueue[R](),
	}
}
func (sched0 *sched[R, T]) Stop() {
	sched0.Pause()
	for atomic.LoadUint64(&sched0.running) > 0 {
		runtime.Gosched()
	}
	atomic.AddUint64(&sched0.pausing, ^uint64(0))
}
func (sched0 *sched[R, T]) Wait() {
	for sched0.tasks.length() != 0 {
		runtime.Gosched()
	}
}
func (sched0 *sched[R, T]) Submit(t T) *Future[R] {
	return sched0.schedule(t, t.GetExecution())
}
func (sched0 *sched[R, T]) Trigger(t T) *Future[R] {
	return sched0.schedule(t, time.Now())
}
func (sched0 *sched[R, T]) Pause() {
	atomic.AddUint64(&sched0.pausing, 1)
	sched0.pause()
}
func (sched0 *sched[R, T]) Resume() {
	atomic.AddUint64(&sched0.pausing, ^uint64(0))  
	sched0.resume()
}
func (s *sched[R, T]) schedule(t T, when time.Time) *Future[R] {
	s.pause()
	if future, ok := s.tasks.update(t, when); ok {
		s.resume()
		return future
	}
	future := s.tasks.push(newTaskItem[R](t, when))
	s.resume()
	return future
}
func (s *sched[R, T]) reschedule(t *task[R], when time.Time) {
	s.pause()
	t.priority = when
	s.tasks.push(t)
	s.resume()
}
func (s *sched[R, T]) getTimer() (t *time.Timer) {
	for {
		t = (*time.Timer)(atomic.LoadPointer(&s.timer))
		if t != nil {
			return
		}
		runtime.Gosched()
	}
}
func (s *sched[R, T]) setTimer(d time.Duration) {
	for {
		old := atomic.SwapPointer(&s.timer, nil)
		if old != nil {
			if (*time.Timer)(old).Stop() {
				(*time.Timer)(old).Reset(d)
				if atomic.CompareAndSwapPointer(&s.timer, nil, old) {
					return
				}
				runtime.Gosched()
				continue
			}
		}
		if atomic.CompareAndSwapPointer(&s.timer, old,
			unsafe.Pointer(time.NewTimer(d))) {
			if old != nil {
				(*time.Timer)(old).Stop()
			}
			return
		}
		runtime.Gosched()
	}
}
func (s *sched[R, T]) pause() {
	old := atomic.LoadPointer(&s.timer)
	if old != nil {
		(*time.Timer)(old).Stop()
	}
}
func (s *sched[R, T]) resume() {
	ctx, cancel := context.WithCancel(context.Background())
	if x, ok := s.cancel.Load().(context.CancelFunc); ok {
		x()
	}
	s.cancel.Store(cancel)
	t := s.tasks.peek()
	if t == nil {
		return
	}
	s.setTimer(t.GetExecution().Sub(time.Now()))
	go func(ctx context.Context) {
		select {
		case <-s.getTimer().C:
			s.worker()
		case <-ctx.Done():
		}
	}(ctx)
}
func (s *sched[R, T]) worker() {
	if atomic.LoadUint64(&s.pausing) > 0 {
		return
	}
	task := s.tasks.pop()
	if task == nil {
		return
	}
	s.resume()
	s.arrival(task)
}
func (s *sched[R, T]) arrival(t *task[R]) {
	atomic.AddUint64(&s.running, 1)
	s.execute(t)
	atomic.AddUint64(&s.running, ^uint64(0))  
}
func (s *sched[R, T]) execute(t *task[R]) {
	defer func() {
		if r := recover(); r != nil {
			t.future.Err = fmt.Errorf(
				"sched: task %s panic while executing, reason: %v",
				t.value.GetID(), r)
		}
	}()
	if t.value.GetExecution().After(time.Now()) {
		s.reschedule(t, t.value.GetExecution())
		return
	}
	result, retry, err := t.value.Execute()
	if retry || err != nil {
		s.reschedule(t, t.value.GetRetryTime())
		return
	}
	t.future.put(result)
}
type taskQueue [R any] struct {
	heap   *taskHeap[R]
	lookup map[string]*task[R]
	mu     sync.Mutex
}
func newTaskQueue [R any] () *taskQueue[R] {
	return &taskQueue[R]{
		heap:   &taskHeap[R]{},
		lookup: map[string]*task[R]{},
	}
}
func (m *taskQueue[R]) length() (l int) {
	m.mu.Lock()
	l = m.heap.Len()
	m.mu.Unlock()
	return
}
func (m *taskQueue[R]) push(t *task[R]) *Future[R] {
	m.mu.Lock()
	heap.Push(m.heap, t)           
	m.lookup[t.value.GetID()] = t  
	m.mu.Unlock()
	return t.future
}
func (m *taskQueue[R]) pop() *task[R] {
	m.mu.Lock()
	if m.heap.Len() == 0 {
		m.mu.Unlock()
		return nil
	}
	item := heap.Pop(m.heap).(*task[R])      
	delete(m.lookup, item.value.GetID())  
	m.mu.Unlock()
	return item
}