plot: Buffering logic

lib/plot.go

@@ -6,6 +6,9 @@ import (
 	"io"
 	"math"
 	"strconv"
+	"time"
+
+	"github.com/tsenart/vegeta/lib/lttb"
 )
 
 // An HTMLPlot represents an interactive HTML time series
@@ -14,51 +17,89 @@ type HTMLPlot struct {
 	title     string
 	threshold int
 	series    map[string]*attackSeries
+	label     func(*Result) string
 }
 
 // attackSeries groups the two timeSeries an attack results in:
 // OK and Error data points
-type attackSeries struct{ ok, err *timeSeries }
+type attackSeries struct {
+	began  time.Time
+	seq    uint64
+	buf    map[uint64]point
+	series map[string]*timeSeries
+	label  func(*Result) string
+}
+
+type point struct {
+	ts  *timeSeries
+	seq uint64
+	t   time.Time
+	v   float64
+}
+
+// newAttackSeries returns a new attackSeries that partitions added
+// Results with the given labelling function.
+func newAttackSeries(label func(*Result) string) *attackSeries {
+	return &attackSeries{
+		buf:    map[uint64]point{},
+		series: map[string]*timeSeries{},
+		label:  label,
+	}
+}
 
 // add adds the given result to the OK timeSeries if the Result
 // has no error, or to the Error timeSeries otherwise.
 func (as *attackSeries) add(r *Result) {
-	var (
-		s     **timeSeries
-		label string
-	)
+	label := as.label(r)
 
-	if r.Error == "" {
-		s, label = &as.ok, "OK"
-	} else {
-		s, label = &as.err, "Error"
+	s, ok := as.series[label]
+	if !ok {
+		s = newTimeSeries(r.Attack, label)
+		as.series[label] = s
 	}
 
-	if *s == nil {
-		*s = newTimeSeries(r.Attack, label, r.Timestamp)
+	p := point{
+		ts:  s,
+		seq: r.Seq,
+		t:   r.Timestamp,
+		v:   r.Latency.Seconds() * 1000,
 	}
 
-	t := uint64(r.Timestamp.Sub((*s).began)) / 1e6 // ns -> ms
-	v := r.Latency.Seconds() * 1000
+	if as.buf[p.seq] = p; p.seq != as.seq {
+		return // buffer
+	} else if as.seq == 0 {
+		as.began = r.Timestamp // first point in attack
+	}
 
-	(*s).add(t, v)
+	// found successor
+	for {
+		p, ok := as.buf[as.seq]
+		if !ok {
+			return
+		}
+
+		delete(as.buf, as.seq)
+		p.ts.add(p.seq, uint64(p.t.Sub(as.began))/1e6, p.v)
+		as.seq++
+	}
 }
 
 // NewHTMLPlot returns an HTMLPlot with the given title,
-// downsampling threshold.
-func NewHTMLPlot(title string, threshold int) *HTMLPlot {
+// downsampling threshold, and result labeling function.
+func NewHTMLPlot(title string, threshold int, label func(*Result) string) *HTMLPlot {
 	return &HTMLPlot{
 		title:     title,
 		threshold: threshold,
 		series:    map[string]*attackSeries{},
+		label:     label,
 	}
 }
 
 // Add adds the given Result to the HTMLPlot time series.
 func (p *HTMLPlot) Add(r *Result) {
 	s, ok := p.series[r.Attack]
 	if !ok {
-		s = &attackSeries{}
+		s = newAttackSeries(p.label)
 		p.series[r.Attack] = s
 	}
 	s.add(r)
@@ -67,7 +108,7 @@ func (p *HTMLPlot) Add(r *Result) {
 // Close closes the HTML plot for writing.
 func (p *HTMLPlot) Close() {
 	for _, as := range p.series {
-		for _, ts := range []*timeSeries{as.ok, as.err} {
+		for _, ts := range as.series {
 			if ts != nil {
 				ts.data.Finish()
 			}
@@ -149,7 +190,7 @@ func (p *HTMLPlot) data() (dataPoints, []string, error) {
 	)
 
 	for _, as := range p.series {
-		for _, s := range [...]*timeSeries{as.ok, as.err} {
+		for _, s := range as.series {
 			if s != nil {
 				series = append(series, s)
 				count += s.len
@@ -173,12 +214,12 @@ func (p *HTMLPlot) data() (dataPoints, []string, error) {
 		}
 
 		for _, p := range points {
-			point := make([]float64, size)
-			for j := range point {
-				point[j] = nan
+			pt := make([]float64, size)
+			for j := range pt {
+				pt[j] = nan
 			}
-			point[0], point[i+1] = p.x, p.y
-			data = append(data, point)
+			pt[0], pt[i+1] = p.X, p.Y
+			data = append(data, pt)
 		}
 
 		labels[i+1] = s.attack + ": " + s.label

lib/plot_test.go

@@ -22,7 +22,13 @@ func BenchmarkHTMLPlot(b *testing.B) {
 		}
 	}
 
-	plot := NewHTMLPlot("Vegeta Plot", 5000)
+	plot := NewHTMLPlot("Vegeta Plot", 5000, func(r *Result) string {
+		if r.Code >= 200 && r.Code < 300 {
+			return "OK"
+		}
+		return "Error"
+	})
+
 	b.Run("Add", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
 			plot.Add(&rs[i%len(rs)])

lib/timeseries.go

@@ -9,22 +9,20 @@ import (
 
 type timeSeries struct {
 	attack string
-	label  string // OK or ERROR
-	began  time.Time
+	label  string
 	data   *tsz.Series
 	len    int
 }
 
-func newTimeSeries(attack, label string, began time.Time) *timeSeries {
+func newTimeSeries(attack, label string) *timeSeries {
 	return &timeSeries{
 		attack: attack,
 		label:  label,
-		began:  began,
 		data:   tsz.New(0),
 	}
 }
 
-func (ts *timeSeries) add(t uint64, v float64) {
+func (ts *timeSeries) add(seq, t uint64, v float64) {
 	ts.data.Push(t, v)
 	ts.len++
 }
@@ -36,7 +34,7 @@ func (ts *timeSeries) iter() lttb.Iter {
 		for i := 0; i < count && it.Next(); i++ {
 			t, v := it.Values()
 			ps = append(ps, lttb.Point{
-				X: time.Duration(t).Seconds(),
+				X: time.Duration(t * 1e6).Seconds(),
 				Y: v,
 			})
 		}

plot.go

@@ -71,7 +71,14 @@ func plot(files []string, threshold int, title, output string) error {
 	sigch := make(chan os.Signal, 1)
 	signal.Notify(sigch, os.Interrupt)
 
-	plot := vegeta.NewHTMLPlot(title, threshold)
+	plot := vegeta.NewHTMLPlot(title, threshold, func(r *vegeta.Result) string {
+		if r.Error == "" {
+			return "OK"
+		} else {
+			return "Error"
+		}
+	})
+
 decode:
 	for {
 		select {