Paralelismo de tareas Scan

La operación de Scan con los resultados parciales de la operación de reducción es una operación donde el valor de la posición i depende del valor de la posición i-1.

scala> Array(1,2,3).scan(100)((acc,x) => acc + x)
val res0: Array[Int] = Array(100, 101, 103, 106)

Operaciones de scan

Existen dos tipos principales de operaciones scan:

1. scanLeft que asocia por la izquierda
2. scanRight que asocia por la derecha

scala> Array(1,2,3).scanRight(100)((acc,x) => acc + x)
val res1: Array[Int] = Array(106, 105, 103, 100)

En este caso ambas operaciones dan resultados distintos porque los valores intermedios se calculan en orden distinto, aunque el resultado final sea el mismo.

Concepto teórico: Paralelización de Scan

¿Se puede paralelizar Scan suponiendo que \(f\) es asociativa?

A priori pareciera que no, ya que dependemos de valores previos. Parece que es necesario que un valor dado espere que los anteriores hayan sido calculados.

Idea: Utilizar resultados intermedios de tal forma que rompamos la dependencia, sin embargo, esto requiere hacer más cálculos.

El algoritmo de scan paralelo funciona en dos fases: 1. Fase ascendente: Calcula resultados parciales en el árbol 2. Fase descendente: Propaga los acumuladores hacia abajo

Implementación

/*
 * This Scala source file was generated by the Gradle 'init' task.
 */
package taller
import common._

object App {
  // Fase ascendente: calcula resultados parciales del árbol
  // Recorre el árbol de abajo hacia arriba, calculando la reducción de cada subárbol
  def llenarSubiendo[A](t: Tree[A], f:(A,A) => A): TreeRes[A] = {
    t match{
      case Leaf(v) => LeafRes(v)  // Caso base: hoja con valor v
      case Node(left, right) => {
        // Procesa recursivamente los subárboles izquierdo y derecho en paralelo
        val (lr, rr) = parallel(
          llenarSubiendo(left, f),
          llenarSubiendo(right, f)
        )
        // Crea nodo con resultado parcial = f(resultado_izquierdo, resultado_derecho)
        NodeRes(lr, f(lr.res, rr.res), rr)
      }
    }
  }

  // Fase descendente: propaga el acumulador hacia abajo
  // Distribuye los valores acumulados a través del árbol
  def llenarBajando[A](t: TreeRes[A], acc: A, f:(A,A) => A): Tree[A] = {
    t match{
      case LeafRes(v) => Leaf(f(acc, v))  // Caso base: aplica función al acumulador y valor de la hoja
      case NodeRes(leaf, v, right) => {
        // Procesa recursivamente en paralelo:
        // - Subárbol izquierdo con el acumulador actual
        // - Subárbol derecho con acumulador actual + resultado del subárbol izquierdo
        val (l, r) = parallel(
          llenarBajando(leaf, acc, f),
          llenarBajando(right, f(acc, leaf.res), f)
        )
        Node(l, r)
      }
    }
  }

  def main(args: Array[String]): Unit = {
    // Árbol de ejemplo para probar la implementación
    val tree: Tree[Int] = Node(
      Node(Leaf(10), Leaf(12)),
      Node(Leaf(3), Leaf(4))
    )

    // Fase 1: Calcular resultados parciales (fase ascendente)
    val treeRes: TreeRes[Int] = llenarSubiendo[Int](tree, (acc: Int, x: Int) => acc + x)
    println(treeRes)

    // Fase 2: Propagar acumuladores (fase descendente) con valor inicial 100
    val res:Tree[Int] = llenarBajando[Int](treeRes, 100, (acc: Int, x: Int) => acc + x) 
    println(res)
  }

}

Explicación del algoritmo

El método llenarSubiendo genera el árbol de resultados parciales, que consiste en calcular la función en los subárboles izquierdo y derecho de forma recursiva.

El método llenarBajando toma el árbol de resultados parciales y propaga el acumulador hacia abajo. En los hijos derechos acumula también el resultado del hermano izquierdo, lo que permite romper la dependencia secuencial.

Condición importante: Para que este algoritmo funcione correctamente, la operación \(f\) debe ser asociativa.