gtsocial-umbx

scheduler.go (6813B)

---

 package sched
 
 import (
 	"context"
 	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"codeberg.org/gruf/go-runners"
 )
 
 // precision is the maximum time we can offer scheduler run-time precision down to.
 const precision = time.Millisecond
 
 var (
 	// neverticks is a timer channel that never ticks (it's starved).
 	neverticks = make(chan time.Time)
 
 	// alwaysticks is a timer channel that always ticks (it's closed).
 	alwaysticks = func() chan time.Time {
 		ch := make(chan time.Time)
 		close(ch)
 		return ch
 	}()
 )
 
 // Scheduler provides a means of running jobs at specific times and
 // regular intervals, all while sharing a single underlying timer.
 type Scheduler struct {
 	jobs []*Job           // jobs is a list of tracked Jobs to be executed
 	jch  chan interface{} // jch accepts either Jobs or job IDs to notify new/removed jobs
 	svc  runners.Service  // svc manages the main scheduler routine
 	jid  atomic.Uint64    // jid is used to iteratively generate unique IDs for jobs
 	rgo  func(func())     // goroutine runner, allows using goroutine pool to launch jobs
 }
 
 // Start will attempt to start the Scheduler. Immediately returns false if the Service is already running, and true after completed run.
 func (sch *Scheduler) Start(gorun func(func())) bool {
 	var block sync.Mutex
 
 	// Use mutex to synchronize between started
 	// goroutine and ourselves, to ensure that
 	// we don't return before Scheduler init'd.
 	block.Lock()
 	defer block.Unlock()
 
 	ok := sch.svc.GoRun(func(ctx context.Context) {
 		// Create Scheduler job channel
 		sch.jch = make(chan interface{})
 
 		// Set goroutine runner function
 		if sch.rgo = gorun; sch.rgo == nil {
 			sch.rgo = func(f func()) { go f() }
 		}
 
 		// Unlock start routine
 		block.Unlock()
 
 		// Enter main loop
 		sch.run(ctx)
 	})
 
 	if ok {
 		// Wait on goroutine
 		block.Lock()
 	}
 
 	return ok
 }
 
 // Stop will attempt to stop the Scheduler. Immediately returns false if not running, and true only after Scheduler is fully stopped.
 func (sch *Scheduler) Stop() bool {
 	return sch.svc.Stop()
 }
 
 // Running will return whether Scheduler is running (i.e. NOT stopped / stopping).
 func (sch *Scheduler) Running() bool {
 	return sch.svc.Running()
 }
 
 // Done returns a channel that's closed when Scheduler.Stop() is called.
 func (sch *Scheduler) Done() <-chan struct{} {
 	return sch.svc.Done()
 }
 
 // Schedule will add provided Job to the Scheduler, returning a cancel function.
 func (sch *Scheduler) Schedule(job *Job) (cancel func()) {
 	switch {
 	// Check a job was passed
 	case job == nil:
 		panic("nil job")
 
 	// Check we are running
 	case !sch.Running():
 		panic("scheduler not running")
 	}
 
 	// Calculate next job ID
 	last := sch.jid.Load()
 	next := sch.jid.Add(1)
 	if next < last {
 		panic("job id overflow")
 	}
 
 	// Pass job to scheduler
 	job.id = next
 	sch.jch <- job
 
 	// Take ptrs to current state chs
 	ctx := sch.svc.Done()
 	jch := sch.jch
 
 	// Return cancel function for job ID
 	return func() {
 		select {
 		// Sched stopped
 		case <-ctx:
 
 		// Cancel this job
 		case jch <- next:
 		}
 	}
 }
 
 // run is the main scheduler run routine, which runs for as long as ctx is valid.
 func (sch *Scheduler) run(ctx context.Context) {
 	var (
 		// now stores the current time, and will only be
 		// set when the timer channel is set to be the
 		// 'alwaysticks' channel. this allows minimizing
 		// the number of calls required to time.Now().
 		now time.Time
 
 		// timerset represents whether timer was running
 		// for a particular run of the loop. false means
 		// that tch == neverticks || tch == alwaysticks.
 		timerset bool
 
 		// timer tick channel (or always / never ticks).
 		tch <-chan time.Time
 
 		// timer notifies this main routine to wake when
 		// the job queued needs to be checked for executions.
 		timer *time.Timer
 
 		// stopdrain will stop and drain the timer
 		// if it has been running (i.e. timerset == true).
 		stopdrain = func() {
 			if timerset && !timer.Stop() {
 				<-timer.C
 			}
 		}
 	)
 
 	// Create a stopped timer.
 	timer = time.NewTimer(1)
 	<-timer.C
 
 	for {
 		// Reset timer state.
 		timerset = false
 
 		if len(sch.jobs) > 0 {
 			// Get now time.
 			now = time.Now()
 
 			// Sort jobs by next occurring.
 			sort.Sort(byNext(sch.jobs))
 
 			// Get next job time.
 			next := sch.jobs[0].Next()
 
 			// If this job is _just_ about to be ready, we don't bother
 			// sleeping. It's wasted cycles only sleeping for some obscenely
 			// tiny amount of time we can't guarantee precision for.
 			if until := next.Sub(now); until <= precision/1e3 {
 				// This job is behind,
 				// set to always tick.
 				tch = alwaysticks
 			} else {
 				// Reset timer to period.
 				timer.Reset(until)
 				tch = timer.C
 				timerset = true
 			}
 		} else {
 			// Unset timer
 			tch = neverticks
 		}
 
 		select {
 		// Scheduler stopped
 		case <-ctx.Done():
 			stopdrain()
 			return
 
 		// Timer ticked, run scheduled
 		case t := <-tch:
 			if !timerset {
 				// 'alwaysticks' returns zero
 				// times, BUT 'now' will have
 				// been set during above sort.
 				t = now
 			}
 			sch.schedule(t)
 
 		// Received update, handle job/id
 		case v := <-sch.jch:
 			sch.handle(v)
 			stopdrain()
 		}
 	}
 }
 
 // handle takes an interfaces received from Scheduler.jch and handles either:
 // - Job --> new job to add.
 // - uint64 --> job ID to remove.
 func (sch *Scheduler) handle(v interface{}) {
 	switch v := v.(type) {
 	// New job added
 	case *Job:
 		// Get current time
 		now := time.Now()
 
 		// Update the next call time
 		next := v.timing.Next(now)
 		v.next.Store(next)
 
 		// Append this job to queued
 		sch.jobs = append(sch.jobs, v)
 
 	// Job removed
 	case uint64:
 		for i := 0; i < len(sch.jobs); i++ {
 			if sch.jobs[i].id == v {
 				// This is the job we're looking for! Drop this
 				sch.jobs = append(sch.jobs[:i], sch.jobs[i+1:]...)
 				return
 			}
 		}
 	}
 }
 
 // schedule will iterate through the scheduler jobs and execute those necessary, updating their next call time.
 func (sch *Scheduler) schedule(now time.Time) {
 	for i := 0; i < len(sch.jobs); {
 		// Scope our own var
 		job := sch.jobs[i]
 
 		// We know these jobs are ordered by .Next(), so as soon
 		// as we reach one with .Next() after now, we can return
 		if job.Next().After(now) {
 			return
 		}
 
 		// Pass to runner
 		sch.rgo(func() {
 			job.Run(now)
 		})
 
 		// Update the next call time
 		next := job.timing.Next(now)
 		job.next.Store(next)
 
 		if next.IsZero() {
 			// Zero time, this job is done and can be dropped
 			sch.jobs = append(sch.jobs[:i], sch.jobs[i+1:]...)
 			continue
 		}
 
 		// Iter
 		i++
 	}
 }
 
 // byNext is an implementation of sort.Interface to sort Jobs by their .Next() time.
 type byNext []*Job
 
 func (by byNext) Len() int {
 	return len(by)
 }
 
 func (by byNext) Less(i int, j int) bool {
 	return by[i].Next().Before(by[j].Next())
 }
 
 func (by byNext) Swap(i int, j int) {
 	by[i], by[j] = by[j], by[i]
 }