Go语言中如何实现线程安全的 t-digest 数据结构？

珍惜时间，勤奋学习！今天给大家带来《Go语言中如何实现线程安全的 t-digest 数据结构？》，正文内容主要涉及到等等，如果你正在学习Golang，或者是对Golang有疑问，欢迎大家关注我！后面我会持续更新相关内容的，希望都能帮到正在学习的大家！

我正在尝试在 go 中运行一个函数的基准测试，我怀疑如果更多的 goroutine 同时调用它，它的速度会变慢。我想使用 github.com/influxdata/tdigest 等在基准测试期间捕获函数调用的 p50 和 p95 延迟，如下所示：

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/[email protected]/tdigest.go:154 +0x115
github.com/influxdata/tdigest.(*tdigest).process(0xc0001b8100)
    /users/kurt/go/pkg/mod/github.com/influxdata/[email protected]/tdigest.go:119 +0x2b0
github.com/influxdata/tdigest.(*tdigest).addcentroid(0xc0001b8100, 0x413385b200000000, 0x3ff0000000000000)
    /users/kurt/go/pkg/mod/github.com/influxdata/[email protected]/tdigest.go:85 +0x80
github.com/influxdata/tdigest.(*tdigest).add(...)
    /users/kurt/go/pkg/mod/github.com/influxdata/[email protected]/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 调用这个函数是不安全的？有没有办法获得线程安全的估计器？

更新

我试图通过将所有延迟放入一个通道中，然后在基准测试运行后为该通道中的所有延迟调用 td.add() 来使代码线程安全。这是我的尝试：

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/[email protected]/1.12.12/libexec/src/testing/benchmark.go:251 +0x9e
testing.(*B).Run(0xc0000ba340, 0x11404ff, 0x12, 0x1146e70, 0x10b0400)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:553 +0x2a3
testing.runBenchmarks.func1(0xc0000ba340)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:455 +0x78
testing.(*B).runN(0xc0000ba340, 0x1)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.runBenchmarks(0x1144b57, 0x27, 0xc0000b4060, 0x1237cd0, 0x1, 0x1, 0x60)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:461 +0x39f
testing.(*M).Run(0xc0000dc000, 0x0)
    /usr/local/Cellar/[email protected]/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/[email protected]/1.12.12/libexec/src/testing/benchmark.go:251 +0x9e
testing.(*B).Run(0xc0000ba000, 0xc00001a100, 0xd, 0xc00000c020, 0x0)
    /usr/local/Cellar/[email protected]/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/[email protected]/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.(*B).run1.func1(0xc0000ba000)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:249 +0x5a
created by testing.(*B).run1
    /usr/local/Cellar/[email protected]/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/[email protected]/1.12.12/libexec/src/testing/benchmark.go:176 +0xb3
testing.(*B).run1.func1(0xc0000ba4e0)
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:249 +0x5a
created by testing.(*B).run1
    /usr/local/Cellar/[email protected]/1.12.12/libexec/src/testing/benchmark.go:242 +0x7d
exit status 2
FAIL    github.com/kurtpeek/send-closed-channel 1.503s

知道如何制作以便我可以在测试后按顺序调用 td.add 吗？

解决方案

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

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~