func Example_mergeMultipleStreams() {
	// Scenario:
	// We have multiple database shards. On each shard, there is a process
	// collecting query response times from the database logs and inserting
	// them into a Stream (created via NewTargeted(0.90)), much like the
	// Simple example. These processes expose a network interface for us to
	// ask them to serialize and send us the results of their
	// Stream.Samples so we may Merge and Query them.
	//
	// NOTES:
	// * These sample sets are small, allowing us to get them
	// across the network much faster than sending the entire list of data
	// points.
	//
	// * For this to work correctly, we must supply the same quantiles
	// a priori the process collecting the samples supplied to NewTargeted,
	// even if we do not plan to query them all here.
	ch := make(chan quantile.Samples)
	getDBQuerySamples(ch)
	q := quantile.NewTargeted(0.90)
	for samples := range ch {
		q.Merge(samples)
	}
	fmt.Println("perc90:", q.Query(0.90))
}

func updateSampleInMap(m map[string]*quantile.Stream, name string, value float64) {
	var sample *quantile.Stream

	sample, ok := m[name]
	if !ok {
		sample = quantile.NewTargeted(0.50, 0.95, 0.99)
	}

	sample.Insert(value)
	m[name] = sample
}

// NewMetrics computes and returns a Metrics struct out of a slice of Results.
func NewMetrics(r Results) *Metrics {
	m := &Metrics{StatusCodes: map[string]int{}}

	if len(r) == 0 {
		return m
	}

	var (
		errorSet       = map[string]struct{}{}
		quants         = quantile.NewTargeted(0.50, 0.95, 0.99)
		totalSuccess   int
		totalLatencies time.Duration
		latest         time.Time
	)

	for _, result := range r {
		quants.Insert(float64(result.Latency))
		m.StatusCodes[strconv.Itoa(int(result.Code))]++
		totalLatencies += result.Latency
		m.BytesOut.Total += result.BytesOut
		m.BytesIn.Total += result.BytesIn
		if result.Latency > m.Latencies.Max {
			m.Latencies.Max = result.Latency
		}
		if end := result.Timestamp.Add(result.Latency); end.After(latest) {
			latest = end
		}
		if result.Code >= 200 && result.Code < 400 {
			totalSuccess++
		}
		if result.Error != "" {
			errorSet[result.Error] = struct{}{}
		}
	}

	m.Requests = uint64(len(r))
	m.Duration = r[len(r)-1].Timestamp.Sub(r[0].Timestamp)
	m.Rate = float64(m.Requests) / m.Duration.Seconds()
	m.Wait = latest.Sub(r[len(r)-1].Timestamp)
	m.Latencies.Mean = time.Duration(float64(totalLatencies) / float64(m.Requests))
	m.Latencies.P50 = time.Duration(quants.Query(0.50))
	m.Latencies.P95 = time.Duration(quants.Query(0.95))
	m.Latencies.P99 = time.Duration(quants.Query(0.99))
	m.BytesIn.Mean = float64(m.BytesIn.Total) / float64(m.Requests)
	m.BytesOut.Mean = float64(m.BytesOut.Total) / float64(m.Requests)
	m.Success = float64(totalSuccess) / float64(m.Requests)

	m.Errors = make([]string, 0, len(errorSet))
	for err := range errorSet {
		m.Errors = append(m.Errors, err)
	}

	return m
}

// NewMetrics computes and returns a Metrics struct out of a slice of Results
func NewMetrics(results []Result, statheader string) *Metrics {
	m := &Metrics{
		Requests:    uint64(len(results)),
		StatusCodes: map[string]int{},
		Headers:     map[string]int{},
	}
	errorSet := map[string]struct{}{}
	quants := quantile.NewTargeted(0.50, 0.95, 0.99)
	totalSuccess, totalLatencies := 0, time.Duration(0)
	headers := strings.Split(statheader, ",")

	for _, result := range results {
		quants.Insert(float64(result.Latency))
		m.StatusCodes[strconv.Itoa(int(result.Code))]++
		totalLatencies += result.Latency
		m.BytesOut.Total += result.BytesOut
		m.BytesIn.Total += result.BytesIn
		if result.Latency > m.Latencies.Max {
			m.Latencies.Max = result.Latency
		}

		if result.Code >= 200 && result.Code < 300 {
			totalSuccess++
		}

		if result.Error != "" {
			errorSet[result.Error] = struct{}{}
		}

		for _, header := range headers {
			h := result.Header.Get(header)
			if h != "" {
				m.Headers[h]++
			}
		}
	}

	m.Duration = results[len(results)-1].Timestamp.Sub(results[0].Timestamp)
	m.Latencies.Mean = time.Duration(float64(totalLatencies) / float64(m.Requests))
	m.Latencies.P50 = time.Duration(quants.Query(0.50))
	m.Latencies.P95 = time.Duration(quants.Query(0.95))
	m.Latencies.P99 = time.Duration(quants.Query(0.99))
	m.BytesIn.Mean = float64(m.BytesIn.Total) / float64(m.Requests)
	m.BytesOut.Mean = float64(m.BytesOut.Total) / float64(m.Requests)
	m.Success = float64(totalSuccess) / float64(m.Requests)

	m.Errors = make([]string, 0, len(errorSet))
	for err := range errorSet {
		m.Errors = append(m.Errors, err)
	}

	return m
}

func (q *Quantile) QueryHandler() *quantile.Stream {
	q.Lock()
	now := time.Now()
	for q.IsDataStale(now) {
		q.moveWindow()
	}

	merged := quantile.NewTargeted(q.Percentiles...)
	merged.Merge(q.streams[0].Samples())
	merged.Merge(q.streams[1].Samples())
	q.Unlock()
	return merged
}

func New(WindowTime time.Duration, Percentiles []float64) *Quantile {
	q := Quantile{
		currentIndex:   0,
		lastMoveWindow: time.Now(),
		MoveWindowTime: WindowTime / 2,
		Percentiles:    Percentiles,
	}
	for i := 0; i < 2; i++ {
		q.streams[i] = *quantile.NewTargeted(Percentiles...)
	}
	q.currentStream = &q.streams[0]
	return &q
}

func NewStatsAnalyzer(statsChan chan OpStat) *StatsAnalyzer {
	stream := make(map[OpType]*quantile.Stream)
	intervalStream := make(map[OpType]*quantile.Stream)
	for _, opType := range AllOpTypes {
		stream[opType] = quantile.NewTargeted(0.5, 0.7, 0.9, 0.95, 0.99)
		intervalStream[opType] = quantile.NewTargeted(0.5, 0.7, 0.9, 0.95, 0.99)
	}
	statsAnalyzer := &StatsAnalyzer{
		statsChan:           statsChan,
		startTime:           time.Now(),
		stream:              stream,
		maxLatency:          make(map[OpType]float64),
		opsExecuted:         0,
		opsErrors:           0,
		counts:              make(map[OpType]int64),
		intervalStartTime:   time.Now(),
		intervalStream:      intervalStream,
		intervalMaxLatency:  make(map[OpType]float64),
		intervalOpsExecuted: 0,
		intervalOpsErrors:   0,
		intervalCounts:      make(map[OpType]int64),
		mutex:               &sync.Mutex{},
	}

	go func() {
		for {
			op, ok := <-statsAnalyzer.statsChan
			if !ok {
				break
			}
			statsAnalyzer.process(op)
		}
	}()

	return statsAnalyzer
}

// Aggregate the values by name as them come in via the input channel
func (stats *ProgramStats) aggregateValues() {
	for namedValue := range stats.Input {
		stats.Mutex.Lock()

		sample, ok := stats.stats[namedValue.name]
		// Zero value not good enough, so initialize
		if !ok {
			sample = quantile.NewTargeted(0.50, 0.95, 0.99)
		}

		sample.Insert(namedValue.value)
		stats.stats[namedValue.name] = sample
		stats.Mutex.Unlock()
	}
}

// NewMetrics computes and returns a Metrics struct out of a slice of Results
func NewMetrics(results []Result) *Metrics {
	m := &Metrics{StatusCodes: map[string]int{}}
	if len(results) == 0 {
		return m
	}
	errorSet := map[string]struct{}{}
	quants := quantile.NewTargeted(0.50, 0.95, 0.99)
	totalSuccess, totalLatencies := 0, time.Duration(0)

	for _, result := range results {
		quants.Insert(float64(result.Latency))
		m.StatusCodes[strconv.Itoa(int(result.Code))]++
		totalLatencies += result.Latency
		m.BytesOut.Total += result.BytesOut
		m.BytesIn.Total += result.BytesIn
		if result.Latency > m.Latencies.Max {
			m.Latencies.Max = result.Latency
		}
		if result.Code >= 200 && result.Code < 250 {
			totalSuccess++
		}
		if result.Error != "" {
			errorSet[result.Error] = struct{}{}
		}
	}

	m.Requests = uint64(len(results))
	m.Duration = results[len(results)-1].Timestamp.Sub(results[0].Timestamp)
	m.QPS = float64(m.Requests) / m.Duration.Seconds()
	m.Latencies.Mean = time.Duration(float64(totalLatencies) / float64(m.Requests))
	m.Latencies.P50 = time.Duration(quants.Query(0.50))
	m.Latencies.P95 = time.Duration(quants.Query(0.95))
	m.Latencies.P99 = time.Duration(quants.Query(0.99))
	m.BytesIn.Mean = float64(m.BytesIn.Total) / float64(m.Requests)
	m.BytesOut.Mean = float64(m.BytesOut.Total) / float64(m.Requests)
	m.Success = float64(totalSuccess) / float64(m.Requests)

	m.Errors = make([]string, 0, len(errorSet))
	for err := range errorSet {
		m.Errors = append(m.Errors, err)
	}

	return m
}

func drawOpenIssueAge(ctx *context, per *period, filename string) {
	start, end := ctx.StartTime(), ctx.EndTime()

	l := end.Sub(start)/DayDuration + 1
	qs := make([]*quantile.Stream, l)
	for i := range qs {
		qs[i] = quantile.NewTargeted(0.25, 0.50, 0.75)
	}
	ctx.WalkIssues(func(i github.Issue, isPullRequest bool) {
		if isPullRequest {
			return
		}
		created := i.CreatedAt
		closed := end
		if i.ClosedAt != nil {
			closed = *i.ClosedAt
		}
		firsti := created.Sub(start) / DayDuration
		for k := firsti; k <= closed.Sub(start)/DayDuration; k++ {
			qs[k].Insert(float64(k - firsti))
		}
	})

	p, err := plot.New()
	if err != nil {
		panic(err)
	}

	p.Title.Text = "Age of Open Issues"
	p.X.Label.Text = fmt.Sprintf("Date from %s to %s", per.start.Format(DateFormat), per.end.Format(DateFormat))
	p.Y.Label.Text = "Age (days)"
	err = plotutil.AddLines(p, "25th percentile", per.seqFloats(quantileAt(qs, 0.25), DayDuration),
		"Median", per.seqFloats(quantileAt(qs, 0.50), DayDuration),
		"75th percentile", per.seqFloats(quantileAt(qs, 0.75), DayDuration))
	if err != nil {
		panic(err)
	}
	p.X.Tick.Marker = newDayTicker(p.X.Tick.Marker, per.start)

	// Save the plot to a PNG file.
	if err := p.Save(defaultWidth, defaultHeight, filename); err != nil {
		panic(err)
	}
}

func Example_window() {
	// Scenario: We want the 90th, 95th, and 99th percentiles for each
	// minute.

	ch := make(chan float64)
	go sendStreamValues(ch)

	tick := time.NewTicker(1 * time.Minute)
	q := quantile.NewTargeted(0.90, 0.95, 0.99)
	for {
		select {
		case t := <-tick.C:
			flushToDB(t, q.Samples())
			q.Reset()
		case v := <-ch:
			q.Insert(v)
		}
	}
}

func Example_simple() {
	ch := make(chan float64)
	go sendFloats(ch)

	// Compute the 50th, 90th, and 99th percentile.
	q := quantile.NewTargeted(0.50, 0.90, 0.99)
	for v := range ch {
		q.Insert(v)
	}

	fmt.Println("perc50:", q.Query(0.50))
	fmt.Println("perc90:", q.Query(0.90))
	fmt.Println("perc99:", q.Query(0.99))
	fmt.Println("count:", q.Count())
	// Output:
	// perc50: 5
	// perc90: 14
	// perc99: 40
	// count: 2388
}

func drawIssueSolvedDuration(ctx *context, per *period, filename string) {
	start, end := ctx.StartTime(), ctx.EndTime()

	l := end.Sub(start)/MonthDuration + 1
	qs := make([]*quantile.Stream, l)
	for i := range qs {
		qs[i] = quantile.NewTargeted(0.50)
	}
	ctx.WalkIssues(func(i github.Issue, isPullRequest bool) {
		if isPullRequest {
			return
		}
		// count unresolved as the longest period
		d := end.Sub(start)
		if i.ClosedAt != nil {
			d = i.ClosedAt.Sub(*i.CreatedAt)
		}
		for k := i.CreatedAt.Sub(start) / MonthDuration; k <= end.Sub(start)/MonthDuration; k++ {
			qs[k].Insert(float64(d) / float64(DayDuration))
		}
	})

	p, err := plot.New()
	if err != nil {
		panic(err)
	}

	p.Title.Text = "Solved Duration of Issues"
	p.X.Label.Text = fmt.Sprintf("Month from %s to %s", per.start.Format(DateFormat), per.end.Format(DateFormat))
	p.Y.Label.Text = "Duration (days)"
	err = plotutil.AddLines(p, "Median", per.seqFloats(quantileAt(qs, 0.50), MonthDuration))
	if err != nil {
		panic(err)
	}
	p.X.Tick.Marker = newMonthTicker(p.X.Tick.Marker, per.start)

	// Save the plot to a PNG file.
	if err := p.Save(defaultWidth, defaultHeight, filename); err != nil {
		panic(err)
	}
}

func init() {
	for _, area := range areas {
		quants[area] = quantile.NewTargeted(targets...)
	}
}

func benchmarkRelay(b *testing.B, p benchmarkParams) {
	b.SetBytes(int64(p.requestSize))
	b.ReportAllocs()

	services := make(map[string][]string)

	servers := make([]benchmark.Server, p.servers)
	for i := range servers {
		servers[i] = benchmark.NewServer(
			benchmark.WithServiceName("svc"),
			benchmark.WithRequestSize(p.requestSize),
			benchmark.WithExternalProcess(),
		)
		defer servers[i].Close()
		services["svc"] = append(services["svc]"], servers[i].HostPort())
	}

	relay, err := benchmark.NewRealRelay(services)
	require.NoError(b, err, "Failed to create relay")
	defer relay.Close()

	clients := make([]benchmark.Client, p.clients)
	for i := range clients {
		clients[i] = benchmark.NewClient([]string{relay.HostPort()},
			benchmark.WithServiceName("svc"),
			benchmark.WithRequestSize(p.requestSize),
			benchmark.WithExternalProcess(),
			benchmark.WithTimeout(10*time.Second),
		)
		defer clients[i].Close()
		require.NoError(b, clients[i].Warmup(), "Warmup failed")
	}

	quantileVals := []float64{0.50, 0.95, 0.99, 1.0}
	quantiles := make([]*quantile.Stream, p.clients)
	for i := range quantiles {
		quantiles[i] = quantile.NewTargeted(quantileVals...)
	}

	wc := newWorkerControl(p.clients)
	dec := testutils.Decrementor(b.N)

	for i, c := range clients {
		go func(i int, c benchmark.Client) {
			// Do a warm up call.
			c.RawCall(1)

			wc.WorkerStart()
			defer wc.WorkerDone()

			for {
				tokens := dec.Multiple(200)
				if tokens == 0 {
					break
				}

				durations, err := c.RawCall(tokens)
				if err != nil {
					b.Fatalf("Call failed: %v", err)
				}

				for _, d := range durations {
					quantiles[i].Insert(float64(d))
				}
			}
		}(i, c)
	}

	var started time.Time
	wc.WaitForStart(func() {
		b.ResetTimer()
		started = time.Now()
	})
	wc.WaitForEnd()
	duration := time.Since(started)

	fmt.Printf("\nb.N: %v Duration: %v RPS = %0.0f\n", b.N, duration, float64(b.N)/duration.Seconds())

	// Merge all the quantiles into 1
	for _, q := range quantiles[1:] {
		quantiles[0].Merge(q.Samples())
	}

	for _, q := range quantileVals {
		fmt.Printf("  %0.4f = %v\n", q, time.Duration(quantiles[0].Query(q)))
	}
	fmt.Println()
}