Merge pull request #3236 from weaveworks/fast-merger

Faster report merging through mutating objects

app/benchmark_internal_test.go

@@ -65,7 +65,7 @@ func BenchmarkReportUpgrade(b *testing.B) {
 
 func BenchmarkReportMerge(b *testing.B) {
 	reports := upgradeReports(readReportFiles(b, *benchReportPath))
-	merger := NewSmartMerger()
+	merger := NewFastMerger()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		merger.Merge(reports)
@@ -75,7 +75,7 @@ func BenchmarkReportMerge(b *testing.B) {
 func getReport(b *testing.B) report.Report {
 	r := fixture.Report
 	if *benchReportPath != "" {
-		r = NewSmartMerger().Merge(upgradeReports(readReportFiles(b, *benchReportPath)))
+		r = NewFastMerger().Merge(upgradeReports(readReportFiles(b, *benchReportPath)))
 	}
 	return r
 }

app/collector.go

@@ -107,7 +107,7 @@ func NewCollector(window time.Duration) Collector {
 		waitableCondition: waitableCondition{
 			waiters: map[chan struct{}]struct{}{},
 		},
-		merger: NewSmartMerger(),
+		merger: NewFastMerger(),
 	}
 }
 
@@ -292,7 +292,7 @@ func NewFileCollector(path string, window time.Duration) (Collector, error) {
 		go replay(collector, timestamps, reports)
 		return collector, nil
 	}
-	return StaticCollector(NewSmartMerger().Merge(reports).Upgrade()), nil
+	return StaticCollector(NewFastMerger().Merge(reports).Upgrade()), nil
 }
 
 func timestampFromFilepath(path string) (time.Time, error) {

app/merger.go

@@ -13,50 +13,20 @@ type Merger interface {
 	Merge([]report.Report) report.Report
 }
 
-type dumbMerger struct{}
+type fastMerger struct{}
 
-// MakeDumbMerger makes a Merger which merges together reports in the simplest possible way.
-func MakeDumbMerger() Merger {
-	return dumbMerger{}
+// NewFastMerger makes a Merger which merges together reports, mutating the one we are building up
+func NewFastMerger() Merger {
+	return fastMerger{}
 }
 
-func (dumbMerger) Merge(reports []report.Report) report.Report {
+func (fastMerger) Merge(reports []report.Report) report.Report {
 	rpt := report.MakeReport()
 	id := murmur3.New64()
 	for _, r := range reports {
-		rpt = rpt.Merge(r)
+		rpt.UnsafeMerge(r)
 		id.Write([]byte(r.ID))
 	}
 	rpt.ID = fmt.Sprintf("%x", id.Sum64())
 	return rpt
 }
-
-type smartMerger struct{}
-
-// NewSmartMerger makes a Merger which merges reports in
-// parallel. Speed up comes from the fact that a) most merges are
-// between small reports, and b) we take advantage of available cores.
-func NewSmartMerger() Merger {
-	return smartMerger{}
-}
-
-func (smartMerger) Merge(reports []report.Report) report.Report {
-	l := len(reports)
-	switch l {
-	case 0:
-		return report.MakeReport()
-	case 1:
-		return reports[0]
-	}
-	c := make(chan report.Report, l)
-	for _, r := range reports {
-		c <- r
-	}
-	for ; l > 1; l-- {
-		left, right := <-c, <-c
-		go func() {
-			c <- left.Merge(right)
-		}()
-	}
-	return <-c
-}

app/merger_test.go

@@ -27,7 +27,7 @@ func TestMerger(t *testing.T) {
 	want.Endpoint.AddNode(report.MakeNode("bar"))
 	want.Endpoint.AddNode(report.MakeNode("baz"))
 
-	for _, merger := range []app.Merger{app.MakeDumbMerger(), app.NewSmartMerger()} {
+	for _, merger := range []app.Merger{app.NewFastMerger()} {
 		// Test the empty list case
 		if have := merger.Merge([]report.Report{}); !reflect.DeepEqual(have, report.MakeReport()) {
 			t.Errorf("Bad merge: %s", test.Diff(have, want))
@@ -44,12 +44,8 @@ func TestMerger(t *testing.T) {
 	}
 }
 
-func BenchmarkSmartMerger(b *testing.B) {
-	benchmarkMerger(b, app.NewSmartMerger())
-}
-
-func BenchmarkDumbMerger(b *testing.B) {
-	benchmarkMerger(b, app.MakeDumbMerger())
+func BenchmarkFastMerger(b *testing.B) {
+	benchmarkMerger(b, app.NewFastMerger())
 }
 
 const numHosts = 15

app/multitenant/aws_collector.go

@@ -154,7 +154,7 @@ func NewAWSCollector(config AWSCollectorConfig) (AWSCollector, error) {
 		s3:        config.S3Store,
 		userIDer:  config.UserIDer,
 		tableName: config.DynamoTable,
-		merger:    app.NewSmartMerger(),
+		merger:    app.NewFastMerger(),
 		inProcess: newInProcessStore(reportCacheSize, config.Window),
 		memcache:  config.MemcacheClient,
 		window:    config.Window,

report/report.go

@@ -305,16 +305,21 @@ func (r Report) Copy() Report {
 // original is not modified.
 func (r Report) Merge(other Report) Report {
 	newReport := r.Copy()
-	newReport.DNS = newReport.DNS.Merge(other.DNS)
-	newReport.Sampling = newReport.Sampling.Merge(other.Sampling)
-	newReport.Window = newReport.Window + other.Window
-	newReport.Plugins = newReport.Plugins.Merge(other.Plugins)
-	newReport.WalkPairedTopologies(&other, func(ourTopology, theirTopology *Topology) {
-		*ourTopology = ourTopology.Merge(*theirTopology)
-	})
+	newReport.UnsafeMerge(other)
 	return newReport
 }
 
+// UnsafeMerge merges another Report into the receiver. The original is modified.
+func (r *Report) UnsafeMerge(other Report) {
+	r.DNS = r.DNS.Merge(other.DNS)
+	r.Sampling = r.Sampling.Merge(other.Sampling)
+	r.Window = r.Window + other.Window
+	r.Plugins = r.Plugins.Merge(other.Plugins)
+	r.WalkPairedTopologies(&other, func(ourTopology, theirTopology *Topology) {
+		ourTopology.UnsafeMerge(*theirTopology)
+	})
+}
+
 // WalkTopologies iterates through the Topologies of the report,
 // potentially modifying them
 func (r *Report) WalkTopologies(f func(*Topology)) {

report/topology.go

@@ -163,6 +163,21 @@ func (t Topology) Merge(other Topology) Topology {
 	}
 }
 
+// UnsafeMerge merges the other object into this one, modifying the original.
+func (t *Topology) UnsafeMerge(other Topology) {
+	if t.Shape == "" {
+		t.Shape = other.Shape
+	}
+	if t.Label == "" {
+		t.Label, t.LabelPlural = other.Label, other.LabelPlural
+	}
+	t.Nodes.UnsafeMerge(other.Nodes)
+	t.Controls = t.Controls.Merge(other.Controls)
+	t.MetadataTemplates = t.MetadataTemplates.Merge(other.MetadataTemplates)
+	t.MetricTemplates = t.MetricTemplates.Merge(other.MetricTemplates)
+	t.TableTemplates = t.TableTemplates.Merge(other.TableTemplates)
+}
+
 // Nodes is a collection of nodes in a topology. Keys are node IDs.
 // TODO(pb): type Topology map[string]Node
 type Nodes map[string]Node
@@ -186,14 +201,19 @@ func (n Nodes) Merge(other Nodes) Nodes {
 		return n
 	}
 	cp := n.Copy()
+	cp.UnsafeMerge(other)
+	return cp
+}
+
+// UnsafeMerge merges the other object into this one, modifying the original.
+func (n *Nodes) UnsafeMerge(other Nodes) {
 	for k, v := range other {
-		if n, ok := cp[k]; ok { // don't overwrite
-			cp[k] = v.Merge(n)
+		if existing, ok := (*n)[k]; ok { // don't overwrite
+			(*n)[k] = v.Merge(existing)
 		} else {
-			cp[k] = v
+			(*n)[k] = v
 		}
 	}
-	return cp
 }
 
 // Validate checks the topology for various inconsistencies.