Merge pull request #2561 from weaveworks/parallel-reduce

execute reducer branches in parallel

render/memoise.go

@@ -3,20 +3,29 @@ package render
 import (
 	"fmt"
 	"math/rand"
+	"sync"
 
 	"github.com/bluele/gcache"
 
 	"github.com/weaveworks/scope/report"
 )
 
+// renderCache is keyed on the combination of Memoiser and report
+// id. It contains promises of report.Nodes, which result from
+// rendering the report with the Memoiser's renderer.
+//
+// The use of promises ensures that in the absence of cache evictions
+// a memoiser will only ever render a report once, even when Render()
+// is invoked concurrently.
 var renderCache = gcache.New(100).LRU().Build()
 
 type memoise struct {
+	sync.Mutex
 	Renderer
 	id string
 }
 
-// Memoise wraps the renderer in a loving embrace of caching
+// Memoise wraps the renderer in a loving embrace of caching.
 func Memoise(r Renderer) Renderer {
 	return &memoise{
 		Renderer: r,
@@ -24,23 +33,55 @@ func Memoise(r Renderer) Renderer {
 	}
 }
 
-// Render produces a set of Nodes given a Report.
-// Ideally, it just retrieves it from the cache, otherwise it calls through to
-// `r` and stores the result.
+// Render produces a set of Nodes given a Report.  Ideally, it just
+// retrieves a promise from the cache and returns its value, otherwise
+// it stores a new promise and fulfils it by calling through to
+// m.Renderer.
+//
+// The cache is bypassed when rendering a report with a decorator.
 func (m *memoise) Render(rpt report.Report, dct Decorator) report.Nodes {
+	if dct != nil {
+		return m.Renderer.Render(rpt, dct)
+	}
+
 	key := fmt.Sprintf("%s-%s", rpt.ID, m.id)
-	if dct == nil {
-		if result, err := renderCache.Get(key); err == nil {
-			return result.(report.Nodes)
-		}
+
+	m.Lock()
+	v, err := renderCache.Get(key)
+	if err == nil {
+		m.Unlock()
+		return v.(*promise).Get()
 	}
+	promise := newPromise()
+	renderCache.Set(key, promise)
+	m.Unlock()
+
 	output := m.Renderer.Render(rpt, dct)
-	if dct == nil {
-		renderCache.Set(key, output)
-	}
+
+	promise.Set(output)
+
 	return output
 }
 
+type promise struct {
+	val  report.Nodes
+	done chan struct{}
+}
+
+func newPromise() *promise {
+	return &promise{done: make(chan struct{})}
+}
+
+func (p *promise) Set(val report.Nodes) {
+	p.val = val
+	close(p.done)
+}
+
+func (p *promise) Get() report.Nodes {
+	<-p.done
+	return p.val
+}
+
 // ResetCache blows away the rendered node cache.
 func ResetCache() {
 	renderCache.Purge()

render/render.go

@@ -39,11 +39,25 @@ func MakeReduce(renderers ...Renderer) Renderer {
 
 // Render produces a set of Nodes given a Report.
 func (r *Reduce) Render(rpt report.Report, dct Decorator) report.Nodes {
-	result := report.Nodes{}
-	for _, renderer := range *r {
-		result = result.Merge(renderer.Render(rpt, dct))
+	l := len(*r)
+	switch l {
+	case 0:
+		return report.Nodes{}
 	}
-	return result
+	c := make(chan report.Nodes, l)
+	for _, renderer := range *r {
+		renderer := renderer // Pike!!
+		go func() {
+			c <- renderer.Render(rpt, dct)
+		}()
+	}
+	for ; l > 1; l-- {
+		left, right := <-c, <-c
+		go func() {
+			c <- left.Merge(right)
+		}()
+	}
+	return <-c
 }
 
 // Stats implements Renderer