所以我正在尝试实现一个 saxpy 的实现，它既被阻塞又可以使用我机器上可用的 8 核并行计算。我首先假设适合我机器的 L1 缓存(拆分 256kB - 128kB 数据，128kB 代码)的小尺寸向量 x 和 y 可以串行计算。为了测试这个假设，我编写了两个 saxpy 实现，一个是 saxpy (BSS) 的阻塞串行版本，另一个是 saxpy (BPS) 的阻塞并行版本。仅当向量的大小大于 4096 个元素时才使用分块算法。以下是实现:

const cachecap = 32*1024/8 // 4096
func blocked_serial_saxpy(a float64, x []float64, incx int, b float64, y []float64, incy int, z []float64, incz int) {
    zn := len(z)
    //fmt.Println("zn: ", zn)
    if zn <= cachecap {
        serial_saxpy(a, x, incx, b, y, incy, z, incz)
        return
    }

    nblocks := zn/cachecap + 1
    //fmt.Println("nblocks: ", nblocks)
    for i := 0; i < nblocks; i++ {
        beg := i * cachecap
        end := (i + 1) * cachecap
        if end >= zn {
            end = zn
        }
        //fmt.Println("beg, end: ", beg, end)
        xb := x[beg:end]
        yb := y[beg:end]
        zb := z[beg:end]
        serial_saxpy(a, xb, incx, b, yb, incy, zb, incz)
    }
}
func blocked_parallel_saxpy(a float64, x []float64, incx int, b float64, y []float64, incy int, z []float64, incz int) {
    zn := len(z)
    if zn <= cachecap {
        //fmt.Println("zn <= cachecap: using serial_saxpy")
        serial_saxpy(a, x, incx, b, y, incy, z, incz)
        return
    }

    nblocks := zn/cachecap + 1
    //fmt.Println("nblocks: ", nblocks)
    nworkers := runtime.GOMAXPROCS(0)
    if nblocks < nworkers {
        nworkers = nblocks
    }
    //fmt.Println("nworkers: ", nworkers)

    //buf := blockSize*nworkers
    //if buf > nblocks {
    //  buf = nblocks
    //}
    //sendchan := make(chan block, buf)
    sendchan := make(chan block, nblocks)

    var wg sync.WaitGroup
    for i := 0; i < nworkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            a, b := a, b
            incx, incy, incz := incx, incy, incz
            for blk := range sendchan {
                beg, end := blk.beg, blk.end
                serial_saxpy(a, x[beg:end], incx, b, y[beg:end], incy, z[beg:end], incz)
            }
        }()
    }

    for i := 0; i < nblocks; i++ {
        beg := i * cachecap
        end := (i + 1) * cachecap
        if end >= zn {
            end = zn
        }
        //fmt.Println("beg:end", beg, end)
        sendchan <- block{beg, end}
    }
    close(sendchan)
    wg.Wait()
}

func serial_saxpy(a float64, x []float64, incx int, b float64, y []float64, incy int, z []float64, incz int) {
    if incx <= 0 || incy <= 0 || incz <= 0 {
        panic("AxpBy: zero or negative increments not supported")
    }

    n := len(z) / incz
    if incx == 1 && incy == 1 && incz == 1 {
        if a == 1 && b == 1 {
            for i := 0; i < n; i++ {
                z[i] = x[i] + y[i]
            }
            return
        }

        if a == 0 && b == 1 {
            copy(z, y)
            //for i := 0; i < n; i++ {
            //  z[i] = y[i]
            //}
            return
        }

        if a == 1 && b == 0 {
            copy(z, x)
            //for i := 0; i < n; i++ {
            //  z[i] = x[i]
            //}
            return
        }

        if a == 0 && b == 0 {
            return
        }

        for i := 0; i < n; i++ {
            z[i] = a*x[i] + b*y[i]
        }
        return
    }

    // unequal increments or equal increments != 1
    ix, iy, iz := 0, 0, 0
    if a == 1 && b == 1 {
        for i := 0; i < n; i, ix, iy, iz = i+1, ix+incx, iy+incy, iz+incz {
            z[iz] = x[ix] + y[iy]
        }
        return
    }

    if a == 0 && b == 1 {
        for i := 0; i < n; i, ix, iy, iz = i+1, ix+incx, iy+incy, iz+incz {
            z[iz] = y[iy]
        }
        return
    }

    if a == 1 && b == 0 {
        for i := 0; i < n; i, ix, iy, iz = i+1, ix+incx, iy+incy, iz+incz {
            z[iz] = x[ix]
        }
        return
    }

    if a == 0 && b == 0 {
        return
    }

    for i := 0; i < n; i, ix, iy, iz = i+1, ix+incx, iy+incy, iz+incz {
        z[iz] = a*x[ix] + b*y[iy]
    }
}

然后我为三个函数 blocked_serial_saxpy、blocked_pa​​rallel_saxpy 和 serial_saxpy 编写了基准测试。下图显示了向量大小分别为 1e3、1e4、1e5、2e5、3e5、4e5、6e5、8e5 和 1e6 的基准测试结果:

为了帮助我可视化 blocked_pa​​rallel_saxpy 实现的性能，我绘制了结果，这就是我获得的结果: 看着情节，让我想知道，为什么我没有看到并行加速，当所有 CPU 都被使用并且在 blocked_pa​​rallel_saxpy 基准测试期间达到 100%。任务管理器的图像如下:

有人可以帮助我了解这里发生了什么吗？我所看到的是问题的症状还是应该的方式？如果是前者，有办法解决吗？

编辑:我已将 blocked_pa​​rallel_saxpy 代码修改为以下内容。我将总 block 数 (nblocks) 划分为 nworker goroutines 计算 nworker no。 block ，并行。另外，我已经删除了 channel 。我已经对代码进行了基准测试，它的性能与 channel 的并行实现相同，因此我没有附加基准测试。

func blocked_parallel_saxpy(a float64, x []float64, incx int, b float64, y []float64, incy int, z []float64, incz int) {
    zn := len(z)
    if zn <= cachecap {
        serial_saxpy(a, x, incx, b, y, incy, z, incz)
        return
    }

    nblocks := zn/cachecap + 1
    nworkers := runtime.GOMAXPROCS(0)
    if nblocks < nworkers {
        nworkers = nblocks
    }

    var wg sync.WaitGroup
    for i := 0; i < nworkers; i++ {
        for j := 0; j < nworkers && (i+j) < nblocks; j++ {
            wg.Add(1)
            go func(i, j int) {
                defer wg.Done()
                a, b := a, b
                incx, incy, incz := incx, incy, incz
                k := i + j
                beg := k * cachecap
                end := (k + 1) * cachecap
                if end >= zn {
                    end = zn
                }
                serial_saxpy(a, x[beg:end], incx, b, y[beg:end], incy, z[beg:end], incz)
            }(i, j)
        }
    wg.Wait()
}

Edit.2:我写了另一个版本的 blocked_pa​​rallel_saxpy，同样没有 channel 。这一次，我生成了 NumCPU 个 goroutine，每个处理 nblocks/nworkers + 1 个 block ，其中每个 block 都是 cachecap 个。元素的长度。即使在这里，代码的执行也与前两个实现相同。

func blocked_parallel_saxpy(a float64, x []float64, incx int, b float64, y []float64, incy int, z []float64, incz int) {
    zn := len(z)
    if zn <= cachecap {
        serial_saxpy(a, x, incx, b, y, incy, z, incz)
        return
    }

    nblocks := zn/cachecap + 1
    nworkers := runtime.GOMAXPROCS(runtime.NumCPU())
    if nblocks < nworkers {
        nworkers = nblocks
    }

    k := nblocks/nworkers + 1
    var wg sync.WaitGroup
    wg.Add(nworkers)
    for i := 0; i < nworkers; i++ {
        go func(i int) {
            defer wg.Done()
            for j := 0; j < k && (j+i*k) < nblocks; j++ {
                beg := (j + i*k) * cachecap
                end := beg + cachecap
                if end > zn {
                    end = zn
                }
                //fmt.Printf("i:%d, j:%d, k:%d, [beg:end]=[%d:%d]\n", i, j, k, beg, end)
                serial_saxpy(a, x[beg:end], incx, b, y[beg:end], incy, z[beg:end], incz)
            }
        }(i)
    }

    wg.Wait()
}

最佳答案

我会尝试没有 channel 的并行版本，其中每个工作人员每 8 个 block 计算一次，无需协调。

关于go - Go 中的并行 saxpy 实现不能很好地跨内核扩展，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46140741/