kotlin - 使用 groupBy/groupingBy/aggregate 并行求和到较小的桶中？

Question

我收集了一些“东西”，我想将它们汇总到更小的桶中。 (在我的特定情况下，我将图像的亮度 channel 下采样 8 倍。)

我希望它在普通的多核 Android 设备上尽可能快，我认为这意味着 Coroutine-per-bucket。 (因为如果不需要的话，我没有任何理由使用 IntAdders)

朴素的线性解决方案:

val SCALE = 8
image.planes[0].buffer.toByteArray().forEachIndexed { index, byte ->
    val x1 = index % image.width
    val y1 = index / image.width
    val x2 = x1 / SCALE
    val y2 = y1 / SCALE
    val quadIdx = y2 * (image.width / SCALE) + x2
    summedQuadLum[quadIdx] += (byte.toInt() and 0xFF)
}

这不太好 - 需要预先声明 summedQuadLum 集合，并且没有任何并行工作的机会。

我想使用groupBy，还是groupingBy？或聚合？)但这些似乎都使用值来确定新键，而我需要使用键来确定新键。我认为最小的开销是 withIndex ，可以这样做

val thumbSums = bufferArray.withIndex().groupingBy { (idx, _) ->
    val x1 = idx % previewImageDimension.width
    val y1 = idx / previewImageDimension.width
    val x2 = x1 / SCALE
    val y2 = y1 / SCALE
    y2 * (previewImageDimension.width / SCALE) + x2
}.aggregate { _, acc: Int?, (_, lum), _ ->
    (acc ?: 0) + (lum.toInt() and 0xFF)
}.values.toIntArray()

好多了，它真的很接近 - 如果我能弄清楚如何对协程中的每个存储桶求和，我认为它会像预期的那样好。

Answer 1

因此，在 groupingBy 之后，我们有一个 Grouping 对象，我们可以用它来聚合值。重要的是要注意分组本身还没有完成，我们基本上已经描述了如何对值进行分组以及原始数组的迭代器。从这里我们有几个选择:

1. 创建 Channel从迭代器启动一些工作协程来并行使用它。 channel 支持扇出，因此源中的每一项仅由一个工作线程处理。这里的问题是所有工作人员都需要更新结果数组中的不同项目，因此需要同步，这就是它变得棘手且可能效率低下的地方。
2. 为了避免多个工作人员写入同一项目，我们需要告诉每个工作人员要处理哪些项目。这意味着要么每个 worker 都应该处理所有元素，只挑选合适的元素，要么我们应该提前预先计算组并为 worker 提供这些组。这两种方法的性能与串行算法几乎相同，因此没有任何意义。

因此，为了有效地并行化，我们希望避免共享可变状态，因为它需要同步。显然我们也不想预先计算组。

我的建议是从另一方面出发 - 我们不要将原始数组映射到采样数组，而是将采样数组映射到原始数组。所以我们说

这种方法使得每个值都由一个工作人员独立计算，因此不需要同步。现在我们可以这样实现:

suspend fun sample() {
   val asyncFactor = 8
   val src = Image(bufferArray, width)
   val dst = Image(src.width / SCALE, src.height / SCALE)

   val chunkSize = dst.sizeBytes / asyncFactor 
   val jobs = Array(asyncFactor) { idx ->
       async(Dispatchers.Default) {
           val chunkStartIdx = chunkSize * idx
           val chunkEndIdxExclusive = min(chunkStartIdx + chunkSize, dst.sizeBytes)
           calculateSampledImageForIndexes(src, dst, chunkStartIdx, chunkEndIdxExclusive, SCALE)
       }
   }
   awaitAll(*jobs)
}

private fun calculateSampledImageForIndexes(src: Image, dst: Image, startIdx: Int, exclusiveEndIdx: Int, scaleFactor: Int) {
    for (i in startIdx until exclusiveEndIdx) {
        val destX = i % dst.width
        val destY = i / dst.width

        val srcX = destX * scaleFactor
        val srcY = destY * scaleFactor

        var sum = 0
        for (xi in 0 until scaleFactor) {
            for (yi in 0 until scaleFactor) {
                sum += src[srcX + xi, srcY + yi]
            }
        }
        dst[destX, destY] = sum / (scaleFactor * scaleFactor)
    }
}

其中 Image 是图像数据缓冲区的方便包装器:

class Image(val buffer: ByteArray, val width: Int) {
    val height = buffer.size / width

    val sizeBytes get() = buffer.size

    constructor(w: Int, h: Int) : this(ByteArray(w * h), w)

    operator fun get(x: Int, y: Int): Byte = buffer[clampX(x) * width + clampY(y)]

    operator fun set(x: Int, y: Int, value: Int) {
        buffer[x * width + y] = (value and 0xFF).toByte()
    }

    private fun clampX(x: Int) = max(min(x, width), 0)
    private fun clampY(y: Int) = max(min(y, height), 0)
}

此外，通过这种方法，您可以轻松实现许多 image processing基于卷积运算的函数，例如模糊和边缘检测。