engine: resources, graph: Change the done channel into a ctx

This is part one of porting Watch to context.

engine/graph/actions.go

@@ -272,7 +272,7 @@ func (obj *Engine) Worker(vertex pgraph.Vertex) error {
 		defer close(obj.state[vertex].eventsChan) // we close this on behalf of res
 
 		// This is a close reverse-multiplexer. If any of the channels
-		// close, then it will cause the doneChan to close. That way,
+		// close, then it will cause the doneCtx to cancel. That way,
 		// multiple different folks can send a close signal, without
 		// every worrying about duplicate channel close panics.
 		obj.state[vertex].wg.Add(1)
@@ -289,7 +289,7 @@ func (obj *Engine) Worker(vertex pgraph.Vertex) error {
 			}
 
 			// the main "done" signal gets activated here!
-			close(obj.state[vertex].doneChan)
+			obj.state[vertex].doneCtxCancel() // cancels doneCtx
 		}()
 
 		var err error
@@ -308,7 +308,7 @@ func (obj *Engine) Worker(vertex pgraph.Vertex) error {
 						case <-timer.C: // the wait is over
 							return errDelayExpired // special
 
-						case <-obj.state[vertex].init.Done:
+						case <-obj.state[vertex].init.DoneCtx.Done():
 							return nil
 						}
 					}
@@ -359,7 +359,7 @@ func (obj *Engine) Worker(vertex pgraph.Vertex) error {
 
 	// If this exits cleanly, we must unblock the reverse-multiplexer.
 	// I think this additional close is unnecessary, but it's not harmful.
-	defer close(obj.state[vertex].eventsDone) // causes doneChan to close
+	defer close(obj.state[vertex].eventsDone) // causes doneCtx to cancel
 	limiter := rate.NewLimiter(res.MetaParams().Limit, res.MetaParams().Burst)
 	var reserv *rate.Reservation
 	var reterr error
@@ -376,7 +376,7 @@ Loop:
 			// we then save so we can return it to the caller of us.
 			if err != nil {
 				failed = true
-				close(obj.state[vertex].watchDone)   // causes doneChan to close
+				close(obj.state[vertex].watchDone)   // causes doneCtx to cancel
 				reterr = errwrap.Append(reterr, err) // permanent failure
 				continue
 			}
@@ -411,7 +411,7 @@ Loop:
 		// pause if one was requested...
 		select {
 		case <-obj.state[vertex].pauseSignal: // channel closes
-			// NOTE: If we allowed a doneChan below to let us out
+			// NOTE: If we allowed a doneCtx below to let us out
 			// of the resumeSignal wait, then we could loop around
 			// and run this again, causing a panic. Instead of this
 			// being made safe with a sync.Once, we instead run a
@@ -457,7 +457,7 @@ Loop:
 					}
 					if e != nil {
 						failed = true
-						close(obj.state[vertex].limitDone) // causes doneChan to close
+						close(obj.state[vertex].limitDone) // causes doneCtx to cancel
 						reterr = errwrap.Append(reterr, e) // permanent failure
 						break LimitWait
 					}
@@ -497,7 +497,7 @@ Loop:
 						}
 						if e != nil {
 							failed = true
-							close(obj.state[vertex].limitDone) // causes doneChan to close
+							close(obj.state[vertex].limitDone) // causes doneCtx to cancel
 							reterr = errwrap.Append(reterr, e) // permanent failure
 							break RetryWait
 						}
@@ -545,7 +545,7 @@ Loop:
 			// this dies. If Process fails permanently, we ask it
 			// to exit right here... (It happens when we loop...)
 			failed = true
-			close(obj.state[vertex].processDone) // causes doneChan to close
+			close(obj.state[vertex].processDone) // causes doneCtx to cancel
 			reterr = errwrap.Append(reterr, err) // permanent failure
 			continue
 

engine/graph/engine.go

@@ -251,7 +251,7 @@ func (obj *Engine) Commit() error {
 	free := []func() error{} // functions to run after graphsync to reset...
 	vertexRemoveFn := func(vertex pgraph.Vertex) error {
 		// wait for exit before starting new graph!
-		close(obj.state[vertex].removeDone) // causes doneChan to close
+		close(obj.state[vertex].removeDone) // causes doneCtx to cancel
 		obj.state[vertex].Resume()          // unblock from resume
 		obj.waits[vertex].Wait()            // sync
 

engine/graph/state.go

@@ -18,6 +18,7 @@
 package graph
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -60,9 +61,12 @@ type State struct {
 	isStateOK bool  // is state OK or do we need to run CheckApply ?
 	workerErr error // did the Worker error?
 
-	// doneChan closes when Watch should shut down. When any of the
+	// doneCtx is cancelled when Watch should shut down. When any of the
 	// following channels close, it causes this to close.
-	doneChan chan struct{}
+	doneCtx context.Context
+
+	// doneCtxCancel is the cancel function for doneCtx.
+	doneCtxCancel func()
 
 	// processDone is closed when the Process/CheckApply function fails
 	// permanently, and wants to cause Watch to exit.
@@ -131,7 +135,7 @@ func (obj *State) Init() error {
 		return fmt.Errorf("the Logf function is missing")
 	}
 
-	obj.doneChan = make(chan struct{})
+	obj.doneCtx, obj.doneCtxCancel = context.WithCancel(context.Background())
 
 	obj.processDone = make(chan struct{})
 	obj.watchDone = make(chan struct{})
@@ -161,7 +165,7 @@ func (obj *State) Init() error {
 		// Watch:
 		Running: obj.event,
 		Event:   obj.event,
-		Done:    obj.doneChan,
+		DoneCtx: obj.doneCtx,
 
 		// CheckApply:
 		Refresh: func() bool {
@@ -338,7 +342,7 @@ func (obj *State) Pause() error {
 	select {
 	case <-obj.pausedAck.Wait(): // we got it!
 		// we're paused
-	case <-obj.doneChan:
+	case <-obj.doneCtx.Done():
 		return engine.ErrClosed
 	}
 	obj.paused = true
@@ -401,7 +405,7 @@ func (obj *State) poll(interval uint32) error {
 		case <-ticker.C: // received the timer event
 			obj.init.Logf("polling...")
 
-		case <-obj.init.Done: // signal for shutdown request
+		case <-obj.init.DoneCtx.Done(): // signal for shutdown request
 			return nil
 		}
 

engine/resources.go

@@ -18,6 +18,7 @@
 package engine
 
 import (
+	"context"
 	"encoding/gob"
 	"fmt"
 
@@ -101,9 +102,10 @@ type Init struct {
 	// Event sends an event notifying the engine of a possible state change.
 	Event func()
 
-	// Done returns a channel that will close to signal to us that it's time
+	// DoneCtx returns a context that will cancel to signal to us that it's
-	// for us to shutdown.
+	// time for us to shutdown.
-	Done chan struct{}
+	// TODO: this is temporary until Watch supports context directly.
+	DoneCtx context.Context
 
 	// Called from within CheckApply:
 

engine/resources/augeas.go

@@ -154,7 +154,7 @@ func (obj *AugeasRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/aws_ec2.go

@@ -502,7 +502,7 @@ func (obj *AwsEc2Res) longpollWatch() error {
 				send = true
 			}
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 
@@ -596,7 +596,7 @@ func (obj *AwsEc2Res) snsWatch() error {
 			obj.init.Logf("State: %v", msg.event)
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/config_etcd.go

@@ -99,8 +99,7 @@ func (obj *ConfigEtcdRes) Watch() error {
 	obj.wg.Add(1)
 	defer obj.wg.Done()
 	// FIXME: add timeout to context
-	// The obj.init.Done channel is closed by the engine to signal shutdown.
+	ctx, cancel := context.WithCancel(obj.init.DoneCtx)
-	ctx, cancel := util.ContextWithCloser(context.Background(), obj.init.Done)
 	defer cancel()
 	ch, err := obj.init.World.IdealClusterSizeWatch(util.CtxWithWg(ctx, obj.wg))
 	if err != nil {
@@ -121,7 +120,7 @@ Loop:
 			}
 			// pass through and send an event
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 		}
 
 		obj.init.Event() // notify engine of an event (this can block)

engine/resources/consul_kv.go

@@ -162,10 +162,10 @@ func (obj *ConsulKVRes) Watch() error {
 				// Unexpected situation, bug in consul API...
 				select {
 				case ch <- fmt.Errorf("unexpected behaviour in Consul API"):
-				case <-obj.init.Done: // signal for shutdown request
+				case <-obj.init.DoneCtx.Done(): // signal for shutdown request
 				}
 
-			case <-obj.init.Done: // signal for shutdown request
+			case <-obj.init.DoneCtx.Done(): // signal for shutdown request
 			}
 			return
 		}
@@ -186,7 +186,7 @@ func (obj *ConsulKVRes) Watch() error {
 			}
 			obj.init.Event()
 
-		case <-obj.init.Done: // signal for shutdown request
+		case <-obj.init.DoneCtx.Done(): // signal for shutdown request
 			return nil
 		}
 	}

engine/resources/cron.go

@@ -296,7 +296,7 @@ func (obj *CronRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 		// do all our event sending all together to avoid duplicate msgs

engine/resources/dhcp.go

@@ -461,7 +461,7 @@ func (obj *DHCPServerRes) Watch() error {
 		case <-closeSignal: // something shut us down early
 			return closeError
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 
@@ -518,7 +518,7 @@ func (obj *DHCPServerRes) CheckApply(apply bool) (bool, error) {
 	//select {
 	//case <-ch:
 	////case <-obj.interruptChan: // TODO: if we ever support InterruptableRes
-	//case <-obj.init.Done: // closed by the engine to signal shutdown
+	//case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	//}
 
 	// Cheap runtime validation!
@@ -1056,7 +1056,7 @@ func (obj *DHCPHostRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/docker_container.go

@@ -196,7 +196,7 @@ func (obj *DockerContainerRes) Watch() error {
 			}
 			return err
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/docker_image.go

@@ -158,7 +158,7 @@ func (obj *DockerImageRes) Watch() error {
 			}
 			return err
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/exec.go

@@ -252,7 +252,7 @@ func (obj *ExecRes) Watch() error {
 				send = true
 			}
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/file.go

@@ -497,7 +497,7 @@ func (obj *FileRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/group.go

@@ -105,7 +105,7 @@ func (obj *GroupRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/hostname.go

@@ -135,7 +135,7 @@ func (obj *HostnameRes) Watch() error {
 		case <-signals:
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/http.go

@@ -335,7 +335,7 @@ func (obj *HTTPServerRes) Watch() error {
 		case <-closeSignal: // something shut us down early
 			return closeError
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 
@@ -725,7 +725,7 @@ func (obj *HTTPFileRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/kv.go

@@ -26,7 +26,6 @@ import (
 
 	"github.com/purpleidea/mgmt/engine"
 	"github.com/purpleidea/mgmt/engine/traits"
-	"github.com/purpleidea/mgmt/util"
 	"github.com/purpleidea/mgmt/util/errwrap"
 )
 
@@ -132,8 +131,8 @@ func (obj *KVRes) Close() error {
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *KVRes) Watch() error {
 	// FIXME: add timeout to context
-	// The obj.init.Done channel is closed by the engine to signal shutdown.
+	// The obj.init.DoneCtx context is closed by the engine to signal shutdown.
-	ctx, cancel := util.ContextWithCloser(context.Background(), obj.init.Done)
+	ctx, cancel := context.WithCancel(obj.init.DoneCtx)
 	defer cancel()
 
 	ch, err := obj.init.World.StrMapWatch(ctx, obj.getKey()) // get possible events!
@@ -159,7 +158,7 @@ func (obj *KVRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/mount.go

@@ -266,7 +266,7 @@ func (obj *MountRes) Watch() error {
 
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/msg.go

@@ -99,7 +99,7 @@ func (obj *MsgRes) Watch() error {
 	//var send = false // send event?
 	for {
 		select {
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/net.go

@@ -299,7 +299,7 @@ func (obj *NetRes) Watch() error {
 
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/noop.go

@@ -66,7 +66,7 @@ func (obj *NoopRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/nspawn.go

@@ -184,7 +184,7 @@ func (obj *NspawnRes) Watch() error {
 				send = true
 			}
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/password.go

@@ -196,7 +196,7 @@ func (obj *PasswordRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/pippet.go

@@ -96,7 +96,7 @@ func (obj *PippetRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/pkg.go

@@ -143,7 +143,7 @@ func (obj *PkgRes) Watch() error {
 
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/print.go

@@ -73,7 +73,7 @@ func (obj *PrintRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/resources_test.go

@@ -20,6 +20,7 @@
 package resources
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -485,7 +486,9 @@ func TestResources1(t *testing.T) {
 			changedChan := make(chan bool, 1) // buffered!
 			readyChan := make(chan struct{})
 			eventChan := make(chan struct{})
-			doneChan := make(chan struct{})
+			doneCtx, doneCtxCancel := context.WithCancel(context.Background())
+			defer doneCtxCancel()
+
 			debug := testing.Verbose() // set via the -test.v flag to `go test`
 			logf := func(format string, v ...interface{}) {
 				t.Logf(fmt.Sprintf("test #%d: ", index)+format, v...)
@@ -507,7 +510,7 @@ func TestResources1(t *testing.T) {
 				},
 
 				// Watch listens on this for close/pause events.
-				Done:  doneChan,
+				DoneCtx: doneCtx,
 				Debug:   debug,
 				Logf:    logf,
 
@@ -629,7 +632,7 @@ func TestResources1(t *testing.T) {
 					}
 				}
 				t.Logf("test #%d: shutting down Watch", index)
-				close(doneChan) // send Watch shutdown command
+				doneCtxCancel() // send Watch shutdown command
 			}()
 		Loop:
 			for {

engine/resources/svc.go

@@ -172,7 +172,7 @@ func (obj *SvcRes) Watch() error {
 				// loop so that we can see the changed invalid signal
 				obj.init.Logf("daemon reload")
 
-			case <-obj.init.Done: // closed by the engine to signal shutdown
+			case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 				return nil
 			}
 		} else {
@@ -215,7 +215,7 @@ func (obj *SvcRes) Watch() error {
 			case err := <-subErrors:
 				return errwrap.Wrapf(err, "unknown %s error", obj)
 
-			case <-obj.init.Done: // closed by the engine to signal shutdown
+			case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 				return nil
 			}
 		}

engine/resources/test.go

@@ -128,7 +128,7 @@ func (obj *TestRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/tftp.go

@@ -200,7 +200,7 @@ func (obj *TFTPServerRes) Watch() error {
 		case <-closeSignal: // something shut us down early
 			return closeError
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 
@@ -551,7 +551,7 @@ func (obj *TFTPFileRes) Watch() error {
 	obj.init.Running() // when started, notify engine that we're running
 
 	select {
-	case <-obj.init.Done: // closed by the engine to signal shutdown
+	case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 	}
 
 	//obj.init.Event() // notify engine of an event (this can block)

engine/resources/timer.go

@@ -84,7 +84,7 @@ func (obj *TimerRes) Watch() error {
 			send = true
 			obj.init.Logf("received tick")
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/user.go

@@ -139,7 +139,7 @@ func (obj *UserRes) Watch() error {
 			}
 			send = true
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}
 

engine/resources/virt.go

@@ -441,7 +441,7 @@ func (obj *VirtRes) Watch() error {
 		case err := <-errorChan:
 			return errwrap.Wrapf(err, "unknown libvirt error")
 
-		case <-obj.init.Done: // closed by the engine to signal shutdown
+		case <-obj.init.DoneCtx.Done(): // closed by the engine to signal shutdown
 			return nil
 		}