1
\$\begingroup\$

This is the same problem as the Python task scheduler, but in Go.

I chose not to use a sync.Cond (I couldn't seem to get that right), but instead use channels.

package main

import (
    "bufio"
    "container/heap"
    "log"
    "os"
    "sync"
    "time"
)

// Task represents a scheduled cancellable function.
type Task struct {
    start     time.Time
    index     int
    fn        func()
    cancelled bool
}

// Cancel prevents a Task from being run, if it hasn't run yet.
func (t *Task) Cancel() {
    t.cancelled = true
}

// A min-heap of tasks, ordered by start time.
type tasks []*Task

func (ts tasks) Len() int {
    return len(ts)
}

func (ts tasks) Less(i, j int) bool {
    return ts[i].start.Before(ts[j].start)
}

func (ts tasks) Swap(i, j int) {
    ts[i], ts[j] = ts[j], ts[i]
    ts[i].index = i
    ts[j].index = j
}

func (ts *tasks) Push(t interface{}) {
    task := t.(*Task)
    n := len(*ts)
    task.index = n
    *ts = append(*ts, task)
}

func (ts *tasks) Pop() interface{} {
    old := *ts
    n := len(*ts)
    t := old[n-1]
    t.index = -1 // for safety
    *ts = old[0 : n-1]
    return t
}



var (
    mu sync.Mutex
    ts tasks
    ch chan bool
)

func init() {
    ch = make(chan bool)
    ts = make([]*Task, 0)
    heap.Init(&ts)
}

func schedule(fn func(), start time.Time) *Task {
    t := &Task{start: start, fn: fn}

    log.Printf("scheduling task %v\n", *t)

    mu.Lock()
    heap.Push(&ts, t)
    go func() {
        ch <- true
    }()
    mu.Unlock()

    log.Printf("scheduled task %v\n", *t)

    return t
}

func dispatcher() {
    for {
        // default timeout of 1 day so the loop isn't busy
        timeout := time.Duration(24 * time.Hour)

        mu.Lock()
        for len(ts) != 0 && ts[0].cancelled {
            heap.Pop(&ts)
        }
        if len(ts) != 0 {
            task := ts[0]
            timeout = task.start.Sub(time.Now())
        }
        mu.Unlock()

        select {
        case <-time.After(timeout):
            mu.Lock()
            task := heap.Pop(&ts).(*Task)
            mu.Unlock()
            go task.fn()
        case <-ch:
        }
    }
}

Testing this via print debugging gives me the right console output. Is there a better way to structure the code? Are there any hidden race conditions?

Here's my test:

func main() {
    var wg sync.WaitGroup

    start := time.Now()

    task := func(j int) func() {
        return func() {
            log.Printf("running %d: %v \n", j, time.Since(start))
        }
    }

    wg.Add(1)
    go func() {
        log.Println("Press ENTER to quit.")
        r := bufio.NewReader(os.Stdin)
        r.ReadRune()
        close(ch)
        log.Print("Goodbye!")
        wg.Done()
    }()
    go dispatcher()
    go func() {
        schedule(task(1), start.Add(1*time.Second))
        t := schedule(task(2), start.Add(2*time.Second))
        t.Cancel()
        schedule(task(3), start.Add(3*time.Second))
        schedule(task(4), start.Add(2500*time.Millisecond))

        time.Sleep(5 * time.Second)
        newStart := time.Now()
        schedule(task(5), newStart.Add(5*time.Second))
        schedule(task(6), newStart.Add(4*time.Second))
        schedule(task(7), newStart.Add(3500*time.Millisecond))
    }()

    wg.Wait()
}
\$\endgroup\$
  • \$\begingroup\$ This code does not work. Task struct is missing, tasks is not a type, heap is not defined... Please update your question to fix this issues \$\endgroup\$ – felix May 23 '18 at 6:24
1
\$\begingroup\$

Are there any hidden race conditions?

No. All racy code is protected with mutexes and there are no send/receive operations that may deadlock.

Is there a better way to structure the code?

Sure.

Actually I don't like the way you've done several parts and I'll point out them as the post continues. So lets focus on writing idiomatic Go code here.

init

init function originally landed in Go to support initializations of various packages.

Some example uses of init from Go standard packages source code:

  • compress/bzip2 initializes crctab
  • net/http/pprof registers /debug/pprof/* routes
  • image/png, image/jpeg and image/gif register image formats
  • sql.Register which is used to register SQL drivers typically reside in init.

You can find out more usages of init with this command:

grep -noRzP 'func init\(\) \{[\s\S]+?\n\}' "$(go env GOROOT)"

So init is mostly used for package level initialization, there is no point in using one for an executable command. We already have main for these things.

Nice thing about main: it is established and common entry point in many programming languages. We should not abuse it adding yet another program entry point.

To be honest I was able to find usages of init with package main within the standard library code. And it just feels super wrong to me. I see init as a nice feature for initializating the library package, but it doesn't make sense when used in executable one.

init is not second main.

Sheduler

Your sheduler on it's own works fine. I can't see any possible errors or racy parts. The code is solid.

But the code structure, initialization and usage are in a big mess. There is no sheduler constructor and there is no way to stop sheduler routine. Exported and unexported identifiers doesn't make sense. That's how I see it.

In your previous question you've implemented same functionality with Python. Sheduler is implemented as a class with required methods. Let's make same thing here with Go.

I like your post Concurrent memo: mutexes vs. channels and the conclusion part you've done. Channels are nice language primitive. Let's focus on implementing sheduler with channels instead of mutexes and see how far we can go with it.

Note: I won't use package main here. All tests will reside in separate file so we can run them with go test.

This is the interface we'll be doing:

func NewSheduler() *Sheduler {}
func NewShedulerContext(context.Context) *Sheduler {}
func (*Sheduler) Stop()
func (*Sheduler) Wait()
func (*Sheduler) Shedule(time.Time, func()) *Task
func (*Task) Cancel(*Task)

NewSheduler will create a sheduler instance and run the background routine which can be stopped with Stop. Wait method will wait for sheduler to stop. Shedule will add new tasks and return them. Cancel will remove the task from sheduling queue. NewShedulerContext is a nice addition to stop sheduler with context.

Here follows a possible rewrite so it will be simpler to show particular parts of implementation.

sheduler.go contains annotated sheduler implementaion.

package sheduler

import (
    "time"
    "context"
    "sort"
)

type (

    Task struct {
        at time.Time
        fn func()
    }

    Sheduler struct {
        ctx context.Context
        cancel func()

        // chans to process tasks
        addC, removeC chan *Task
    }
)

// start is the main sheduler routine that handles the queue.
// All mutations happen here so we are race free by default.
func (s *Sheduler) start() {

    defer s.cancel()

    // Note: `never` should not be closed or used, except for reading
    var next <- chan time.Time
    var timer *time.Timer
    never := make(<- chan time.Time)

    tasks := []*Task{}

    for {

        // Every loop iteration tasks[0] is checked and called
        // if it's time to
        if len(tasks) > 0 {
            if time.Now().After(tasks[0].at) {

                // Time to call a task
                t := tasks[0]
                tasks = tasks[1:]

                // Task may call sheduler.Add,
                // so it's wise to use `go` here
                go t.fn()

                continue

            } else if timer == nil {

                // There is a task,
                // but timer is nil, creating one
                timer = time.NewTimer(time.Until(tasks[0].at))
                next = timer.C
            }
        } else {

            // There is no a single task
            next = never
        }




        select {

        // Stop was called
        case <- s.ctx.Done():
            return

        // Wait for next task to shedule
        case <- next:
            timer = nil

        // Wait for new task
        case task := <- s.addC:
            tasks = append(tasks, task)

            sort.Slice(tasks, func (i, j int) bool {
                return tasks[i].at.Before(tasks[j].at)
            })

            if tasks[0] == task && timer != nil {
                timer.Stop()
                timer = nil
            }

        // Wait for task to remove
        case task := <- s.removeC:
            for i, t := range tasks {
                if t == task {
                    if i == 0 && timer != nil {
                        timer.Stop()
                        timer = nil
                    }
                    tasks = append(tasks[:i],tasks[i+1:]...)
                    break
                }
            }
        }
    }
}

// Wait waits untill sheduler stops.
func (s *Sheduler) Wait() {
    <- s.ctx.Done()
}

// Stop stops sheduler routine.
func (s *Sheduler) Stop() {
    s.cancel()
}

// Shedule adds shedules new task.
// It will return nil if sheduler was stopped.
func (s *Sheduler) Shedule(at time.Time, fn func()) *Task {
    task := &Task{at, fn}
    select {
    case <- s.ctx.Done():
        return nil
    case s.addC <- task:
        return task
    }
}

// Cancel cancels sheduled task.
func (s *Sheduler) Cancel(task *Task) {
    select {
    case <- s.ctx.Done():
    case s.removeC <- task:
    }
}

// NewShedulerContext is like NewSheduler but includes context.
// The provided context is used to stop the sheduler.
func NewShedulerContext(parentCtx context.Context) *Sheduler {

    ctx, cancel := context.WithCancel(parentCtx)
    s := &Sheduler{
        ctx:     ctx,
        cancel:  cancel,
        addC:    make(chan *Task),
        removeC: make(chan *Task),
    }
    go s.start()
    return s
}

// NewSheduler creates new sheduler and starts it.
// Stop must be called to stop sheduler routine free resources.
func NewSheduler() *Sheduler {
    return NewShedulerContext(context.Background())
}

sheduler_test.go contains several test cases. Use it with go test.

package sheduler

import (
    . "testing"
    "time"
)

func TestShedulerStops(t *T) {

    s := NewSheduler()

    go func() {
        s.Stop()
    }()

    s.Wait()
}

func TestShedulerShedulesInOrder(t *T) {

    s := NewSheduler()

    x := 0

    s.Shedule(time.Now().Add(time.Second), func() {
        if x++; x != 1 { t.Error("Expected to run first") }
    })

    s.Shedule(time.Now().Add(time.Second * 2), func() {
        if x++; x != 2 { t.Error("Expected to run second") }
    })

    s.Shedule(time.Now().Add(time.Second * 3), func() {
        if x++; x != 3 { t.Error("Expected to run last") }
        s.Stop()
    })

    s.Wait()
}

It's always important to annotate the source code. Golang has super nice toolchain that will use your comments to produce a package documentation.

I've decided to use channels here. addC and removeC from Sheduler struct are used to send tasks into running routine. The sheduler routine start will select from both of them and mutate tasks slice. The implementation is race free as all mutations happen in a single routine.

Note: functions Shedule and Cancel both write to channel. This may block forever when sheduler routine is stopped. To protect ourselfs from deadlocks we'll use Done from context.Context. Both functions use select with <- s.ctx.Done case to exit, when sheduler routine stops.

To keep the code short implementation uses simple slice of tasks as a queue. This can be changed to heap or similar container.

Afterwords

In this review I've tried to address the ideas of writing idiomatic Go code. Apart from yours I've added several other problems like sheduler cancelation, use of channels, etc. This review may not be one you've actually wanted to see so let me know.

Feel free to ping me in the chat and comments if there is an issue or so.

\$\endgroup\$
  • \$\begingroup\$ You give a grep command with /lib/go/src; much better would be give the command using go env GOROOT in backticks. (It's /usr/local/go for me and many others). \$\endgroup\$ – Dave C Sep 3 '18 at 19:25
  • \$\begingroup\$ @DaveC, nice tip, fixed. Seems like my distro takes it different :) \$\endgroup\$ – sineemore Sep 4 '18 at 11:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.