增强大规模任务和高效人员的通道性能

“纵有疾风来，人生不言弃”，这句话送给正在学习Golang的朋友们，也希望在阅读本文《增强大规模任务和高效人员的通道性能》后，能够真的帮助到大家。我也会在后续的文章中，陆续更新Golang相关的技术文章，有好的建议欢迎大家在评论留言，非常感谢！

我想有效地解决一个概念上类似于查找空间中 p 点的平均成对距离的问题，这是我为了解决这个问题而使用的示例。计算可以很好地并行化，所以我想用 go 来解决它。在顺序程序中，我需要运行两个嵌套循环，外部循环在 i = 0...p-1 上，内部循环在 j = i+1...p-1 上。然后，我会计算点 i 和 j 之间的距离，将它们全部相加，最后除以点对的数量。因此，计算需要覆盖可能的点对组合的“三角形”。

在 go 中，我的第一次尝试是使用相同的逻辑，但通过通道将计算分配给工作函数。我的方法如下：

package main

import "math"
import "sync"
import "fmt"
import "math/rand"
import "github.com/schollz/progressbar"

const nprocs = 32
const npoints = 30000

type pair struct {
    p1 [3]float64
    p2 [3]float64
}

func square(f float64) (float64) {
    return f * f
}

func progress(total int64, ch <-chan int) {
    bar := progressbar.default(total)
    for i := range ch {
        bar.add(i)
    }
}

func worker(idx int, sumbuffer []float64, in <-chan pair, out chan<- int, wg *sync.waitgroup) {
    count := int64(0)
    
    defer wg.done();
    
    for pair := range in {
        dist := math.sqrt(square(pair.p1[0]-pair.p2[0]) + square(pair.p1[1]-pair.p2[1]) + square(pair.p1[2]-pair.p2[2]))
        sumbuffer[idx] += dist
        
        count++
        if count % (2<<15) == 0 {
            out <- (2<<15)
        }
    }
}

func main() {
    var sumbuffer [nprocs]float64
    var points    [npoints][3]float64
    
    var sum float64
    
    for i := 0; i < npoints; i++ {
        for j := 0; j < 3; j++ {
            points[i][j] = rand.float64()
        }
    }
    
    wg := &sync.waitgroup{};
    wg.add(nprocs);
    
    progressch := make(chan int, 4 * nprocs)
    pairch := make(chan pair, 4 * nprocs)
    npairs := int64(npoints - 1) * int64(npoints) / int64(2)
    go progress(npairs, progressch)
    
    for i := 0; i < nprocs; i++ {
        go worker(i, sumbuffer[:], pairch, progressch, wg)
    }
    
    for i := int64(0); i < npoints; i++ {
        for j := int64(i+1); j < npoints; j++ {
            pairch <- pair{points[i], points[j]}
        }
    }
    
    close(pairch)
    wg.wait();
    
    for i := 0; i < nprocs; i++ {
        sum += sumbuffer[i]
    }
    sum /= float64(npairs)
    
    fmt.println("average distance:", sum)
}

但是，该程序的运行速度没有我预期的那么快。在第二次尝试中，我摆脱了分配任务的渠道，而是手动在工作人员之间划分计算。然后，每个worker首先需要计算自己要覆盖“三角形”的哪一部分，相当麻烦。

package main

import "math"
import "sync"
import "fmt"
import "math/rand"
import "github.com/schollz/progressbar"

const nProcs = 32
const nPoints = 30000

func square(f float64) (float64) {
    return f * f
}

func progress(total int64, ch <-chan int) {
    bar := progressbar.Default(total)
    for i := range ch {
        bar.Add(i)
    }
}

func worker(idx int, points [][3]float64, sumBuffer []float64, start int64, stop int64, out chan<- int, wg *sync.WaitGroup) {
    count := start
    length := int64(len(points))
    
    defer wg.Done();
    
    // Calculate start value for loop index i
    iStart := int64(0)
    pointCount := int64(0)
    for k := length - 1; k >= 0; k-- {
        if pointCount + k > start {
            break
        } else {
            pointCount += k
            iStart++
        }
    }
    firstLoop := true
        
    for i := int64(iStart); i < length && count < stop; i++ {
        // Calculate start value for loop index j
        var jStart int64
        if firstLoop {
            jStart = (i + 1) + (start - pointCount)
        } else {
            jStart = i + 1
        }
        
        for j := int64(jStart); j < length && count < stop; j++ {
            dist := math.Sqrt(square(points[i][0]-points[j][0]) + square(points[i][1]-points[j][1]) + square(points[i][2]-points[j][2]))
            sumBuffer[idx] += dist
            
            count++
            if count % (2<<15) == 0 {
                out <- (2<<15)
            }
        }
        
        firstLoop = false
    }
}

func main() {
    var sumBuffer [nProcs]float64
    var points    [nPoints][3]float64
    
    var sum float64
    
    for i := 0; i < nPoints; i++ {
        for j := 0; j < 3; j++ {
            points[i][j] = rand.Float64()
        }
    }
    
    wg := &sync.WaitGroup{};
    wg.Add(nProcs);
    
    progressCh := make(chan int, 4 * nProcs)
    nPairs := int64(nPoints - 1) * int64(nPoints) / int64(2)
    go progress(nPairs, progressCh)
    
    step := int64(math.Ceil(float64(nPairs) / float64(nProcs)))
    for i := 0; i < nProcs; i++ {
        go worker(i, points[:], sumBuffer[:], int64(i) * step, int64(i+1) * step, progressCh, wg)
    }

    wg.Wait();
    
    for i := 0; i < nProcs; i++ {
        sum += sumBuffer[i]
    }
    sum /= float64(nPairs)
    
    fmt.Println("Average distance:", sum)
}

现在第二个程序快了大约 100 倍！然而，我什至没有利用该版本中 go 的优势，而且我可以用 c++ 编写相同的程序。如何改进第一个程序，使得使用通道带来的开销不那么严重？或者这只是渠道效率的限制，而对于我的用例来说，渠道根本不是正确的选择？

除此之外，我对 go 还很陌生。我确信我的程序对于 go 来说不太理想。欢迎任何关于如何改进我的风格的评论。

解决方案

通道版本可能会更慢，因为相对于 goroutine 之间的调度和通信的开销来说，工作（计算距离）非常小。

最好将点对分成更大的块，就像您在第二个示例中所做的那样。

这里有一个简单的方法来了解发生了什么：

var sum float64
var count int64

for i, p := range points {
    for _, p2 := range points[i+1:] {
        sum += p.distanceto(p2)
        count++
    }
}

fmt.printf("%d points, %d pairs, avg %f\n", len(points), count, sum/float64(count))

当我尝试这个时，我的笔记本电脑变热，然后输出：

100000 points, 4999950000 pairs, avg 0.660843
duration 22.623835855s

每次距离计算大约需要 5 纳秒，速度非常快，而且分工太小。

只有一个 cpu 核心正在执行这项工作：

由于我的笔记本电脑有 8 个内核，因此我尝试创建一个示例，其中 8 个工作人员处理相同的点对块。

8 workers, 100000 points, 4999950000 pairs, avg 0.660843
duration 7.481476421s

速度并没有提高 8 倍，但它是一种改进，并且可以让这些闲置核心投入工作：

示例：

package main

import (
    "fmt"
    "math"
    "math/rand"
    "runtime"
    "time"
)

const nPoints = 100000

var points [nPoints]*Point

type Point struct {
    x, y, z float64
}

func (p *Point) distanceTo(p2 *Point) float64 {
    a := p.x - p2.x
    b := p.y - p2.y
    c := p.z - p2.z
    return math.Sqrt(a*a + b*b + c*c)
}

func main() {
    initPoints()

    start := time.Now()
    run()
    fmt.Printf("duration %v\n", time.Since(start))
}

func initPoints() {
    for i := 0; i < nPoints; i++ {
        points[i] = &Point{rand.Float64(), rand.Float64(), rand.Float64()}
    }
}

func run() {
    requests := make(chan Request)
    results := make(chan Result)

    nWorkers := runtime.NumCPU() //change, eg: 1 or 2

    for n := 0; n < nWorkers; n++ {
        go worker(requests, results)
    }

    go func() {
        nPairs := nPoints * (nPoints - 1) / 2
        batchSize := nPairs / nWorkers

        req := Request{iStart: 0, jStart: 1, count: 0}

        for i := 0; i < nPoints; i++ {
            for j := i + 1; j < nPoints; j++ {
                req.count++
                if req.count == batchSize {
                    requests <- req
                    req = Request{iStart: i, jStart: j, count: 0}
                }
            }
        }

        if req.count > 0 {
            requests <- req
        }

        close(requests)
    }()

    var sum, avg float64
    var count int64

    for n := 0; n < nWorkers; n++ {
        r := <-results
        fmt.Printf("worker %d: %v\n", n+1, r)
        sum += r.sum
        count += r.count
    }
    close(results)

    if count > 0 {
        avg = sum / float64(count)
    }

    fmt.Printf("%d workers, %d points, %d pairs, avg %f\n", nWorkers, len(points), count, avg)
}

func worker(requests chan Request, results chan Result) {
    r := Result{}
    for req := range requests {
        req.work(&r)
    }

    results <- r
}

func (req *Request) work(result *Result) {
    count := 0
    jStart := req.jStart

    for i := req.iStart; i < nPoints; i++ {
        p := points[i]

        if i > req.iStart {
            jStart = i + 1
        }

        for j := jStart; j < nPoints; j++ {
            p2 := points[j]

            result.sum += p.distanceTo(p2)
            result.count++

            count++
            if count == req.count {
                return
            }
        }
    }
}

type Request struct {
    iStart int
    jStart int
    count  int
}

type Result struct {
    sum   float64
    count int64
}

func (r Result) String() string {
    avg := 0.0
    if r.count > 0 {
        avg = r.sum / float64(r.count)
    }
    return fmt.Sprintf("%d comparisons, avg distance %f", r.count, avg)
}

pprof profiling blog post 很好地演示了查看程序将时间花在哪里的技术。

以上就是《增强大规模任务和高效人员的通道性能》的详细内容，更多关于的资料请关注golang学习网公众号！