runner: fix possible shutdown deadlocks

worker/runner.go

 // Copyright 2012, 2013 Canonical Ltd.
 // Licensed under the AGPLv3, see LICENCE file for details.
 
 package worker
 
 import (
         "errors"
         "launchpad.net/juju-core/log"
         "launchpad.net/tomb"
+        "local/runtime/debug"
         "time"
 )
 
 // RestartDelay holds the length of time that a worker
 // will wait between exiting and restarting.
 var RestartDelay = 3 * time.Second
 
 // Worker is implemented by a running worker.
 type Worker interface {
         // Kill asks the worker to stop without necessarily
@@ skipping 26 matching lines @@
         worker Worker
 }
 
 type doneInfo struct {
         id  string
         err error
 }
 
 // NewRunner creates a new Runner.  When a worker finishes, if its error
 // is deemed fatal (determined by calling isFatal), all the other workers
-// will be stopped and the runner itself will finish.  Of all the errors
-// returned by the stopped workers, only the most important one
-// (determined by calling moreImportant) will be returned from
-// Runner.Wait.
+// will be stopped and the runner itself will finish.  Of all the fatal errors
+// returned by the stopped workers, only the most important one,
+// determined by calling moreImportant, will be returned from
+// Runner.Wait (non-fatal errors will not be returned from Wait).

[fwereade, 2013/05/27 20:35:53]

This deserves a sentence, not parens, imo

 //
 // The function isFatal(err) returns whether err is a fatal error.  The
 // function moreImportant(err0, err1) returns whether err0 is considered
-// more important than err1..
+// more important than err1.
 func NewRunner(isFatal func(error) bool, moreImportant func(err0, err1 error) bool) *Runner {
         runner := &Runner{
                 startc:        make(chan startReq),
                 stopc:         make(chan string),
                 donec:         make(chan doneInfo),
                 startedc:      make(chan startInfo),
                 isFatal:       isFatal,
                 moreImportant: moreImportant,
         }
         go func() {
@@ skipping 33 matching lines @@
         case <-runner.tomb.Dead():
         }
         return ErrDead
 }
 
 func (runner *Runner) Wait() error {
         return runner.tomb.Wait()
 }
 
 func (runner *Runner) Kill() {
+        log.Debugf("worker: killing runner %p %s", runner, debug.Callers(0, 20))
         runner.tomb.Kill(nil)
 }
 
 // Stop kills the given worker and waits for it to exit.
 func Stop(worker Worker) error {
         worker.Kill()
         return worker.Wait()
 }
 
+type workerInfo struct {
+        start        func() (Worker, error)
+        worker       Worker
+        restartDelay time.Duration
+        stopping     bool
+}
+
 func (runner *Runner) run() error {
-        type workerInfo struct {
-                start        func() (Worker, error)
-                worker       Worker
-                restartDelay time.Duration
-                stopping     bool
-        }
         // workers holds the current set of workers.  All workers with a
         // running goroutine have an entry here.
         workers := make(map[string]*workerInfo)
         var finalError error
-loop:
+
+        // isDying holds whether the runner is currently dying.  When it
+        // is dying (whether as a result of being killed or due to a
+        // fatal error), all existing workers are killed, no new workers
+        // will be started, and the loop will exit when all existing
+        // workers have stopped.
+        isDying := false
+        tombDying := runner.tomb.Dying()
         for {
+                if isDying && len(workers) == 0 {
+                        return finalError
+                }
                 select {
-                case <-runner.tomb.Dying():
-                        break loop
+                case <-tombDying:
+                        log.Infof("worker: runner is dying")

[thumper, 2013/06/06 04:37:44]

FYI, you could create a package level logger now using:
  var logger = loggo.GetLogger("juju.worker")

Then use
  logger.Info("runner is dying")

As the module is included in the output.

[fwereade, 2013/06/06 16:36:15]

+100 :)

[rog, 2013/06/19 08:19:43]

i think that probably deserves another CL. we could change loads of stuff in
that way.

+                        isDying = true
+                        killAll(workers)
+                        tombDying = nil
                 case req := <-runner.startc:
+                        if isDying {
+                                log.Infof("worker: ignoring start request for %q when dying", req.id)
+                                break
+                        }
                         info := workers[req.id]
                         if info == nil {
                                 workers[req.id] = &workerInfo{
                                         start:        req.start,
                                         restartDelay: RestartDelay,
                                 }
                                 go runner.runWorker(0, req.id, req.start)
                                 break
                         }
                         if !info.stopping {
                                 // The worker is already running, so leave it alone
                                 break
                         }
-                        // The worker previously existed and is currently
-                        // being stopped.  When it eventually does stop,
-                        // we'll restart it immediately with the new
-                        // start function.
+                        // The worker previously existed and is
+                        // currently being stopped.  When it eventually
+                        // does stop, we'll restart it immediately with
+                        // the new start function.
                         info.start = req.start
                         info.restartDelay = 0
                 case id := <-runner.stopc:
-                        info := workers[id]
-                        if info == nil {
-                                // The worker doesn't exist so nothing to do.
-                                break
+                        if info := workers[id]; info != nil {
+                                killWorker(id, info)
                         }
-                        if info.worker != nil {
-                                info.worker.Kill()
-                                info.worker = nil
+                case info := <-runner.startedc:
+                        workerInfo := workers[info.id]
+                        workerInfo.worker = info.worker
+                        if isDying {
+                                killWorker(info.id, workerInfo)
                         }
-                        info.stopping = true
-                        info.start = nil
-                case info := <-runner.startedc:
-                        workers[info.id].worker = info.worker
                 case info := <-runner.donec:
                         workerInfo := workers[info.id]
                         if !workerInfo.stopping && info.err == nil {
                                 info.err = errors.New("unexpected quit")
                         }
                         if info.err != nil {
-                                log.Errorf("worker: worker %q: %v", info.id, info.err)
                                 if runner.isFatal(info.err) {
-                                        finalError = info.err
+                                        log.Errorf("worker: fatal %q: %v", info.id, info.err)
+                                        if finalError == nil || runner.moreImportant(info.err, finalError) {
+                                                finalError = info.err
+                                        }
                                         delete(workers, info.id)
-                                        break loop
+                                        if !isDying {
+                                                isDying = true
+                                                killAll(workers)
+                                        }
+                                        break
+                                } else {
+                                        log.Errorf("worker: exited %q: %v", info.id, info.err)
                                 }
                         }
                         if workerInfo.start == nil {
                                 // The worker has been deliberately stopped;
                                 // we can now remove it from the list of workers.
                                 delete(workers, info.id)
                                 break
                         }
                         go runner.runWorker(workerInfo.restartDelay, info.id, workerInfo.start)
                         workerInfo.restartDelay = RestartDelay
                 }
         }
-        for _, info := range workers {
-                if info.worker != nil {
-                        info.worker.Kill()
-                        info.worker = nil
-                }
+        panic("unreachable")
+}
+
+func killAll(workers map[string]*workerInfo) {
+        for id, info := range workers {
+                killWorker(id, info)
         }
-        for len(workers) > 0 {
-                info := <-runner.donec
-                if runner.moreImportant(info.err, finalError) {
-                        finalError = info.err
-                }
-                if info.err != nil {
-                        log.Errorf("worker: worker %q: %v", info.id, info.err)
-                }
-                delete(workers, info.id)
+}
+
+func killWorker(id string, info *workerInfo) {
+        if info.worker != nil {
+                log.Debugf("worker: killing %q", id)
+                info.worker.Kill()
+                info.worker = nil
         }
-        return finalError
+        info.stopping = true
+        info.start = nil
 }
 
 // runWorker starts the given worker after waiting for the given delay.
 func (runner *Runner) runWorker(delay time.Duration, id string, start func() (Worker, error)) {
         if delay > 0 {
+                log.Infof("worker: restarting %q in %v", id, delay)
                 select {
                 case <-runner.tomb.Dying():
                         runner.donec <- doneInfo{id, nil}
                         return
                 case <-time.After(delay):
                 }
         }
+        log.Infof("worker: start %q", id)
         worker, err := start()
         if err == nil {
                 runner.startedc <- startInfo{id, worker}
                 err = worker.Wait()
         }
         runner.donec <- doneInfo{id, err}
 }