go - Go 中 t-digest 数据结构的线程安全实现？

Question

我正在尝试在 Go 中运行一个函数的基准测试，如果更多的 goroutine 同时调用它，我怀疑它会减慢速度。我想在基准测试期间捕获函数调用的 P50 和 P95 延迟，例如使用 github.com/influxdata/tdigest ，就像下面这样:

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "testing"
    "time"

    "github.com/influxdata/tdigest"
)

func BenchmarkSomething(b *testing.B) {
    concurrencies := []int{1, 10, 30}
    total := 1000

    for _, concurrency := range concurrencies {
        b.Run(fmt.Sprintf("concurrency_%d", concurrency), func(b *testing.B) {
            td := tdigest.New()
            for i := 0; i < b.N; i++ {
                send := make(chan struct{}, total)
                var wg sync.WaitGroup
                for i := 0; i < concurrency; i++ {
                    wg.Add(1)
                    go func() {
                        for range send {
                            start := time.Now()
                            duration := time.Duration(int64(rand.NormFloat64()*1e6)) + time.Millisecond
                            time.Sleep(time.Duration(duration))
                            td.Add(float64(time.Since(start)), 1)
                        }
                    }()
                }

                for i := 0; i < total; i++ {
                    send <- struct{}{}
                }
                close(send)
            }
            b.Log("P50 latency:", td.Quantile(0.50))
        })
    }
}

但是，如果我尝试运行它，我会得到 index out of range错误:

> go test -bench .
goos: darwin
goarch: amd64
pkg: github.com/kurtpeek/send-closed-channel
BenchmarkSomething/concurrency_1-8          panic: runtime error: index out of range

goroutine 422 [running]:
github.com/influxdata/tdigest.(*TDigest).updateCumulative(0xc0001b8100)
    /Users/kurt/go/pkg/mod/github.com/influxdata/tdigest@v0.0.1/tdigest.go:154 +0x115
github.com/influxdata/tdigest.(*TDigest).process(0xc0001b8100)
    /Users/kurt/go/pkg/mod/github.com/influxdata/tdigest@v0.0.1/tdigest.go:119 +0x2b0
github.com/influxdata/tdigest.(*TDigest).AddCentroid(0xc0001b8100, 0x413385b200000000, 0x3ff0000000000000)
    /Users/kurt/go/pkg/mod/github.com/influxdata/tdigest@v0.0.1/tdigest.go:85 +0x80
github.com/influxdata/tdigest.(*TDigest).Add(...)
    /Users/kurt/go/pkg/mod/github.com/influxdata/tdigest@v0.0.1/tdigest.go:58
github.com/kurtpeek/send-closed-channel.BenchmarkSomething.func1.1(0xc000271620, 0xc0001b8100)
    /Users/kurt/go/src/github.com/kurtpeek/send-closed-channel/main_test.go:30 +0x100
created by github.com/kurtpeek/send-closed-channel.BenchmarkSomething.func1
    /Users/kurt/go/src/github.com/kurtpeek/send-closed-channel/main_test.go:25 +0xa8
exit status 2
FAIL    github.com/kurtpeek/send-closed-channel 0.266s

我怀疑从多个 goroutine 调用这个函数是不安全的？有没有办法获得线程安全的估计器？

更新

我试图通过将所有延迟放入 channel 中，然后调用 td.Add() 来使代码线程安全。基准运行后该 channel 中的所有延迟。这是我的尝试:

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "testing"
    "time"

    "github.com/influxdata/tdigest"
)

func BenchmarkSomething(b *testing.B) {
    concurrencies := []int{1, 10, 30}
    total := 1000

    for _, concurrency := range concurrencies {
        b.Run(fmt.Sprintf("concurrency_%d", concurrency), func(b *testing.B) {
            latencies := make(chan time.Duration, total)
            for i := 0; i < b.N; i++ {
                send := make(chan struct{}, total)
                var wg sync.WaitGroup
                for i := 0; i < concurrency; i++ {
                    wg.Add(1)
                    go func() {
                        for range send {
                            start := time.Now()
                            duration := time.Duration(int64(rand.NormFloat64()*1e6)) + time.Millisecond
                            time.Sleep(time.Duration(duration))
                            latencies <- time.Since(start)
                        }
                    }()
                }

                for i := 0; i < total; i++ {
                    send <- struct{}{}
                }
                close(send)
            }
            td := tdigest.New()
            for latency := range latencies {
                td.Add(float64(latency), 1)
            }
            close(latencies)

            b.Log("P50 latency:", td.Quantile(0.50))
        })
    }
}

但是，当我运行它时，出现死锁错误:

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
testing.(*B).run1(0xc0000ba000, 0xc0000ba000)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:251 +0x9e
testing.(*B).Run(0xc0000ba340, 0x11404ff, 0x12, 0x1146e70, 0x10b0400)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:553 +0x2a3
testing.runBenchmarks.func1(0xc0000ba340)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:455 +0x78
testing.(*B).runN(0xc0000ba340, 0x1)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.runBenchmarks(0x1144b57, 0x27, 0xc0000b4060, 0x1237cd0, 0x1, 0x1, 0x60)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:461 +0x39f
testing.(*M).Run(0xc0000dc000, 0x0)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/testing.go:1078 +0x413
main.main()
    _testmain.go:42 +0x13e

goroutine 5 [chan receive]:
testing.(*B).run1(0xc0000ba4e0, 0xc0000ba4e0)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:251 +0x9e
testing.(*B).Run(0xc0000ba000, 0xc00001a100, 0xd, 0xc00000c020, 0x0)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:553 +0x2a3
github.com/kurtpeek/send-closed-channel.BenchmarkSomething(0xc0000ba000)
    /Users/kurt/go/src/github.com/kurtpeek/send-closed-channel/main_test.go:18 +0x8f
testing.(*B).runN(0xc0000ba000, 0x1)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.(*B).run1.func1(0xc0000ba000)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:249 +0x5a
created by testing.(*B).run1
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:242 +0x7d

goroutine 19 [chan receive]:
github.com/kurtpeek/send-closed-channel.BenchmarkSomething.func1(0xc0000ba4e0)
    /Users/kurt/go/src/github.com/kurtpeek/send-closed-channel/main_test.go:41 +0x1a6
testing.(*B).runN(0xc0000ba4e0, 0x1)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.(*B).run1.func1(0xc0000ba4e0)
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:249 +0x5a
created by testing.(*B).run1
    /usr/local/Cellar/go@1.12/1.12.12/libexec/src/testing/benchmark.go:242 +0x7d
exit status 2
FAIL    github.com/kurtpeek/send-closed-channel 1.503s

知道如何制作，以便我可以调用td.Add的测试后顺序？

Answer 1

在高层次上，我通过将问题分为两个步骤来解决这个问题:将延迟发送到缓冲 channel (在这种情况下，所需的缓冲区大小是预先知道的)，然后调用 td.Add()在结果上。这样可以避免调用 td.Add()同时仍然实现在并发设置中测量延迟的预期结果。