pgraph: Build the sync group into the graph structure

This hides the sync/wait logic inside the graph itself.
This commit is contained in:
James Shubin
2016-12-20 04:23:51 -05:00
parent 6312b9225f
commit dd8d17232f
3 changed files with 21 additions and 13 deletions

View File

@@ -450,7 +450,7 @@ func (g *Graph) Worker(v *Vertex) error {
// Start is a main kick to start the graph. It goes through in reverse topological
// sort order so that events can't hit un-started vertices.
func (g *Graph) Start(wg *sync.WaitGroup, first bool) { // start or continue
func (g *Graph) Start(first bool) { // start or continue
log.Printf("State: %v -> %v", g.setState(graphStateStarting), g.getState())
defer log.Printf("State: %v -> %v", g.setState(graphStateStarted), g.getState())
t, _ := g.TopologicalSort()
@@ -459,11 +459,11 @@ func (g *Graph) Start(wg *sync.WaitGroup, first bool) { // start or continue
for _, v := range Reverse(t) {
if !v.Res.IsWatching() { // if Watch() is not running...
wg.Add(1)
g.wg.Add(1)
// must pass in value to avoid races...
// see: https://ttboj.wordpress.com/2015/07/27/golang-parallelism-issues-causing-too-many-open-files-error/
go func(vv *Vertex) {
defer wg.Done()
defer g.wg.Done()
// TODO: if a sufficient number of workers error,
// should something be done? Will these restart
// after perma-failure if we have a graph change?
@@ -502,6 +502,11 @@ func (g *Graph) Start(wg *sync.WaitGroup, first bool) { // start or continue
}
}
// Wait waits for all the graph vertex workers to exit.
func (g *Graph) Wait() {
g.wg.Wait()
}
// Pause sends pause events to the graph in a topological sort order.
func (g *Graph) Pause() {
log.Printf("State: %v -> %v", g.setState(graphStatePausing), g.getState())