Collections Scala Performance Parallel Multi-core Parallel Operations Par

August 31, 2025

Parallel Collections: Data Parallelism

Harness the power of multi-core processors with Scala's parallel collections to automatically parallelize operations and dramatically improve performance for computational workloads.

Lesson 29 🚀 Practice

Parallel Collections: Data Parallelism

Introduction

Parallel collections in Scala provide an elegant way to leverage multi-core processors for data-parallel computations. By simply adding .par to most collection operations, you can automatically distribute work across multiple cores, significantly improving performance for CPU-intensive tasks.

This lesson will teach you how to use parallel collections effectively, understand when parallelization helps, and avoid common pitfalls while building high-performance applications.

Understanding Parallel Collections

Basic Parallel Operations

import scala.collection.parallel.CollectionConverters._

// Converting collections to parallel
val numbers = (1 to 1000000).toList
val parallelNumbers = numbers.par

println(s"Original collection type: ${numbers.getClass.getSimpleName}")
println(s"Parallel collection type: ${parallelNumbers.getClass.getSimpleName}")

// Timing helper function
def timeOperation[T](name: String)(operation: => T): T = {
  val start = System.nanoTime()
  val result = operation
  val end = System.nanoTime()
  println(f"$name: ${(end - start) / 1e6}%.2f ms")
  result
}

// CPU-intensive operation for demonstration
def expensiveFunction(n: Int): Int = {
  var result = n
  for (_ <- 1 to 1000) {
    result = (result * 1.1).toInt
  }
  result
}

// Compare sequential vs parallel performance
val data = (1 to 100000).toList

timeOperation("Sequential map") {
  val result = data.map(expensiveFunction)
  result.length
}

timeOperation("Parallel map") {
  val result = data.par.map(expensiveFunction)
  result.length
}

// Parallel filter operations
timeOperation("Sequential filter") {
  val result = numbers.filter(_ % 17 == 0)
  result.length
}

timeOperation("Parallel filter") {
  val result = numbers.par.filter(_ % 17 == 0)
  result.length
}

// Parallel reduce operations
timeOperation("Sequential sum") {
  numbers.sum
}

timeOperation("Parallel sum") {
  numbers.par.sum
}

// Custom parallel reduce
timeOperation("Sequential reduce") {
  numbers.reduce(_ + _)
}

timeOperation("Parallel reduce") {
  numbers.par.reduce(_ + _)
}

// Parallel foreach (for side effects)
import java.util.concurrent.atomic.AtomicInteger

val counter = new AtomicInteger(0)

timeOperation("Sequential foreach") {
  data.foreach(_ => counter.incrementAndGet())
}

counter.set(0)

timeOperation("Parallel foreach") {
  data.par.foreach(_ => counter.incrementAndGet())
}

println(s"Final counter value: ${counter.get()}")

// Parallel find operations
val target = 567890

timeOperation("Sequential find") {
  numbers.find(_ == target)
}

timeOperation("Parallel find") {
  numbers.par.find(_ == target)
}

// Parallel exists and forall
timeOperation("Sequential exists") {
  numbers.exists(_ > 999000)
}

timeOperation("Parallel exists") {
  numbers.par.exists(_ > 999000)
}

timeOperation("Sequential forall") {
  numbers.forall(_ > 0)
}

timeOperation("Parallel forall") {
  numbers.par.forall(_ > 0)
}

Parallel Collection Types

// Different parallel collection types
val list = List(1, 2, 3, 4, 5)
val vector = Vector(1, 2, 3, 4, 5)
val array = Array(1, 2, 3, 4, 5)
val range = (1 to 5)
val set = Set(1, 2, 3, 4, 5)
val map = Map("a" -> 1, "b" -> 2, "c" -> 3)

// Converting to parallel collections
val parList = list.par
val parVector = vector.par
val parArray = array.par
val parRange = range.par
val parSet = set.par
val parMap = map.par

println("Parallel collection types:")
println(s"Parallel List: ${parList.getClass.getSimpleName}")
println(s"Parallel Vector: ${parVector.getClass.getSimpleName}")
println(s"Parallel Array: ${parArray.getClass.getSimpleName}")
println(s"Parallel Range: ${parRange.getClass.getSimpleName}")
println(s"Parallel Set: ${parSet.getClass.getSimpleName}")
println(s"Parallel Map: ${parMap.getClass.getSimpleName}")

// Converting back to sequential
val backToSeq = parList.seq
println(s"Back to sequential: ${backToSeq.getClass.getSimpleName}")

// Parallel string operations
val text = "The quick brown fox jumps over the lazy dog " * 10000
val words = text.split("\\s+").toVector

timeOperation("Sequential word processing") {
  words
    .filter(_.length > 3)
    .map(_.toUpperCase)
    .groupBy(_.head)
    .view.mapValues(_.length)
    .toMap
    .size
}

timeOperation("Parallel word processing") {
  words.par
    .filter(_.length > 3)
    .map(_.toUpperCase)
    .groupBy(_.head)
    .view.mapValues(_.length)
    .toMap
    .size
}

// Parallel matrix operations
case class Matrix(data: Vector[Vector[Double]]) {
  def +(other: Matrix): Matrix = {
    Matrix(
      data.zip(other.data).par.map { case (row1, row2) =>
        row1.zip(row2).map { case (a, b) => a + b }
      }.seq
    )
  }

  def *(scalar: Double): Matrix = {
    Matrix(
      data.par.map(row => row.map(_ * scalar)).seq
    )
  }

  def transpose: Matrix = {
    Matrix(
      data.head.indices.par.map { col =>
        data.map(_(col))
      }.seq.toVector
    )
  }
}

// Create test matrices
val size = 500
val matrix1 = Matrix(
  Vector.fill(size)(Vector.fill(size)(scala.util.Random.nextDouble()))
)
val matrix2 = Matrix(
  Vector.fill(size)(Vector.fill(size)(scala.util.Random.nextDouble()))
)

timeOperation("Matrix addition") {
  val result = matrix1 + matrix2
  result.data.length
}

timeOperation("Matrix scalar multiplication") {
  val result = matrix1 * 2.5
  result.data.length
}

timeOperation("Matrix transpose") {
  val result = matrix1.transpose
  result.data.length
}

// Parallel groupBy operations
case class Student(name: String, grade: Int, subject: String, score: Double)

val students = (1 to 10000).map { i =>
  Student(
    s"Student$i",
    (i % 4) + 9,  // Grades 9-12
    List("Math", "Science", "English", "History")(i % 4),
    50 + scala.util.Random.nextDouble() * 50  // Scores 50-100
  )
}.toVector

timeOperation("Sequential groupBy grade") {
  students.groupBy(_.grade).view.mapValues(_.length).toMap
}

timeOperation("Parallel groupBy grade") {
  students.par.groupBy(_.grade).view.mapValues(_.length).toMap
}

// Advanced parallel aggregations
timeOperation("Sequential grade analysis") {
  students
    .groupBy(_.subject)
    .view.mapValues { subjectStudents =>
      val scores = subjectStudents.map(_.score)
      Map(
        "average" -> scores.sum / scores.length,
        "max" -> scores.max,
        "min" -> scores.min,
        "count" -> scores.length
      )
    }.toMap
}

timeOperation("Parallel grade analysis") {
  students.par
    .groupBy(_.subject)
    .view.mapValues { subjectStudents =>
      val scores = subjectStudents.map(_.score)
      Map(
        "average" -> scores.sum / scores.length,
        "max" -> scores.max,
        "min" -> scores.min,
        "count" -> scores.length
      )
    }.toMap
}

Configuring Parallelism

Task Support and Thread Pools

import scala.collection.parallel.{ForkJoinTaskSupport, ExecutorServiceTaskSupport}
import java.util.concurrent.{ForkJoinPool, Executors}

// Default parallelism level
val defaultParCollection = (1 to 1000).par
println(s"Default parallelism level: ${defaultParCollection.tasksupport.parallelismLevel}")

// Custom ForkJoinPool
val customPool = new ForkJoinPool(8)  // 8 threads
val customParCollection = (1 to 1000).par
customParCollection.tasksupport = new ForkJoinTaskSupport(customPool)

println(s"Custom parallelism level: ${customParCollection.tasksupport.parallelismLevel}")

// Timing with different parallelism levels
def testParallelismLevel(level: Int, data: Vector[Int]): Unit = {
  val pool = new ForkJoinPool(level)
  val parCollection = data.par
  parCollection.tasksupport = new ForkJoinTaskSupport(pool)

  val time = timeOperation(s"Parallelism level $level") {
    parCollection.map(expensiveFunction).length
  }

  pool.shutdown()
}

val testData = (1 to 10000).toVector

// Test different parallelism levels
List(1, 2, 4, 8, 16).foreach(level => testParallelismLevel(level, testData))

// Using ExecutorService instead of ForkJoinPool
val executorService = Executors.newFixedThreadPool(4)
val executorParCollection = testData.par
executorParCollection.tasksupport = new ExecutorServiceTaskSupport(executorService)

timeOperation("ExecutorService TaskSupport") {
  executorParCollection.map(x => x * x).length
}

executorService.shutdown()

// Configuring threshold for parallel operations
class ConfigurableParallelCollection[T](data: Vector[T]) {
  def mapWithThreshold[R](f: T => R, threshold: Int): Vector[R] = {
    if (data.length < threshold) {
      data.map(f)
    } else {
      data.par.map(f).seq
    }
  }

  def filterWithThreshold(p: T => Boolean, threshold: Int): Vector[T] = {
    if (data.length < threshold) {
      data.filter(p)
    } else {
      data.par.filter(p).seq
    }
  }
}

val configurableCollection = new ConfigurableParallelCollection((1 to 1000).toVector)

timeOperation("Small threshold (sequential)") {
  configurableCollection.mapWithThreshold(_ * 2, threshold = 10000).length
}

timeOperation("Large threshold (parallel)") {
  configurableCollection.mapWithThreshold(_ * 2, threshold = 100).length
}

// Dynamic parallelism based on system resources
object AdaptiveParallelism {
  val availableProcessors = Runtime.getRuntime.availableProcessors()

  def createOptimalParallelCollection[T](data: Vector[T]): Vector[T] = {
    val optimalThreads = math.max(2, availableProcessors - 1)  // Leave one core for OS
    val pool = new ForkJoinPool(optimalThreads)

    val parData = data.par
    parData.tasksupport = new ForkJoinTaskSupport(pool)

    parData.seq  // Return the configured parallel collection
  }

  def processWithAdaptiveParallelism[T, R](
    data: Vector[T], 
    operation: T => R,
    minSizeForParallel: Int = 1000
  ): Vector[R] = {
    if (data.length < minSizeForParallel) {
      data.map(operation)
    } else {
      val parData = data.par
      parData.tasksupport = new ForkJoinTaskSupport(
        new ForkJoinPool(availableProcessors)
      )
      parData.map(operation).seq
    }
  }
}

println(s"Available processors: ${AdaptiveParallelism.availableProcessors}")

val largeDataset = (1 to 50000).toVector

timeOperation("Adaptive parallel processing") {
  AdaptiveParallelism.processWithAdaptiveParallelism(
    largeDataset, 
    (x: Int) => math.sqrt(x.toDouble).toInt
  ).length
}

// Nested parallelism considerations
def nestedParallelOperation(data: Vector[Vector[Int]]): Vector[Vector[Int]] = {
  // Outer parallelism
  data.par.map { innerVector =>
    // Inner parallelism - be careful with nested parallelism
    if (innerVector.length > 1000) {
      innerVector.par.map(_ * 2).seq
    } else {
      innerVector.map(_ * 2)
    }
  }.seq
}

val nestedData = Vector.fill(100)(Vector.fill(1000)(scala.util.Random.nextInt(1000)))

timeOperation("Nested parallel operation") {
  nestedParallelOperation(nestedData).length
}

// Memory considerations with parallel collections
def memoryEfficientParallelProcessing[T, R](
  data: Vector[T],
  operation: T => R,
  chunkSize: Int = 10000
): Vector[R] = {

  data.grouped(chunkSize).map { chunk =>
    chunk.par.map(operation).seq
  }.flatten.toVector
}

val hugeDataset = (1 to 100000).toVector

timeOperation("Memory-efficient parallel processing") {
  memoryEfficientParallelProcessing(
    hugeDataset,
    (x: Int) => expensiveFunction(x),
    chunkSize = 5000
  ).length
}

When to Use Parallel Collections

Performance Analysis and Guidelines

// Measuring scalability
def measureScalability[T](
  data: Vector[Int],
  operation: Int => T,
  maxThreads: Int = 16
): Unit = {

  println(s"Scalability analysis for ${data.length} elements:")
  println("Threads\tTime (ms)\tSpeedup\tEfficiency")

  // Sequential baseline
  val sequentialTime = timeOperation("Sequential baseline") {
    data.map(operation)
  }

  (1 to maxThreads by 2).foreach { threads =>
    val pool = new ForkJoinPool(threads)
    val parData = data.par
    parData.tasksupport = new ForkJoinTaskSupport(pool)

    val parallelTime = {
      val start = System.nanoTime()
      parData.map(operation)
      val end = System.nanoTime()
      (end - start) / 1e6
    }

    val speedup = sequentialTime / parallelTime
    val efficiency = speedup / threads

    println(f"$threads\t$parallelTime%.2f\t\t$speedup%.2f\t$efficiency%.2f")

    pool.shutdown()
  }
}

// Test with CPU-intensive operation
println("CPU-intensive operation scalability:")
measureScalability((1 to 20000).toVector, expensiveFunction)

// Test with less CPU-intensive operation
println("\nSimple operation scalability:")
measureScalability((1 to 100000).toVector, (x: Int) => x * x + 1)

// Guidelines for when to use parallel collections
object ParallelCollectionGuidelines {

  // 1. CPU-intensive operations benefit most
  def cpuIntensiveExample(): Unit = {
    val data = (1 to 50000).toVector

    // Good candidate: complex mathematical operations
    timeOperation("Sequential: Complex math") {
      data.map(n => math.pow(math.sin(n), 2) + math.cos(n)).length
    }

    timeOperation("Parallel: Complex math") {
      data.par.map(n => math.pow(math.sin(n), 2) + math.cos(n)).length
    }
  }

  // 2. Simple operations may not benefit
  def simpleOperationExample(): Unit = {
    val data = (1 to 1000000).toVector

    // Poor candidate: simple operations with overhead
    timeOperation("Sequential: Simple increment") {
      data.map(_ + 1).length
    }

    timeOperation("Parallel: Simple increment") {
      data.par.map(_ + 1).length
    }
  }

  // 3. I/O operations are usually not suitable
  def ioOperationExample(): Unit = {
    val urls = (1 to 100).map(i => s"http://example.com/api/$i").toVector

    // Not suitable: I/O bound operations
    def simulateHttpCall(url: String): String = {
      Thread.sleep(10)  // Simulate network latency
      s"Response from $url"
    }

    timeOperation("Sequential: I/O simulation") {
      urls.map(simulateHttpCall).length
    }

    timeOperation("Parallel: I/O simulation") {
      urls.par.map(simulateHttpCall).length
    }
  }

  // 4. Data size matters
  def dataSizeExample(): Unit = {
    val sizes = List(100, 1000, 10000, 100000)

    sizes.foreach { size =>
      val data = (1 to size).toVector

      println(s"\nData size: $size")

      timeOperation("Sequential") {
        data.map(expensiveFunction).length
      }

      timeOperation("Parallel") {
        data.par.map(expensiveFunction).length
      }
    }
  }

  // 5. Shared mutable state is problematic
  def sharedStateExample(): Unit = {
    val data = (1 to 100000).toVector

    // Problematic: shared mutable state
    var sum = 0

    timeOperation("Sequential with shared state") {
      data.foreach(sum += _)
    }

    println(s"Sequential sum: $sum")

    // This will produce incorrect results!
    sum = 0
    timeOperation("Parallel with shared state (WRONG!)") {
      data.par.foreach(sum += _)
    }

    println(s"Parallel sum (incorrect): $sum")

    // Correct approach: use reduce or atomic operations
    timeOperation("Parallel with reduce (CORRECT)") {
      val correctSum = data.par.reduce(_ + _)
      println(s"Parallel sum (correct): $correctSum")
    }
  }
}

println("\n=== Parallel Collection Guidelines ===")

println("\n1. CPU-intensive operations:")
ParallelCollectionGuidelines.cpuIntensiveExample()

println("\n2. Simple operations:")
ParallelCollectionGuidelines.simpleOperationExample()

println("\n3. I/O operations:")
ParallelCollectionGuidelines.ioOperationExample()

println("\n4. Data size impact:")
ParallelCollectionGuidelines.dataSizeExample()

println("\n5. Shared state problems:")
ParallelCollectionGuidelines.sharedStateExample()

// Performance profiling helper
class ParallelPerformanceProfiler {
  def profile[T, R](
    data: Vector[T],
    operation: T => R,
    name: String = "Operation"
  ): Unit = {

    println(s"\n=== Profiling: $name ===")
    println(s"Data size: ${data.length}")

    // Memory usage before
    val runtime = Runtime.getRuntime
    runtime.gc()
    val memoryBefore = runtime.totalMemory() - runtime.freeMemory()

    // Sequential execution
    val seqTime = {
      val start = System.nanoTime()
      data.map(operation)
      val end = System.nanoTime()
      (end - start) / 1e6
    }

    runtime.gc()
    val memoryAfterSeq = runtime.totalMemory() - runtime.freeMemory()

    // Parallel execution
    val parTime = {
      val start = System.nanoTime()
      data.par.map(operation)
      val end = System.nanoTime()
      (end - start) / 1e6
    }

    runtime.gc()
    val memoryAfterPar = runtime.totalMemory() - runtime.freeMemory()

    // Results
    val speedup = seqTime / parTime
    val efficiency = speedup / Runtime.getRuntime.availableProcessors()

    println(f"Sequential time: $seqTime%.2f ms")
    println(f"Parallel time: $parTime%.2f ms")
    println(f"Speedup: $speedup%.2fx")
    println(f"Efficiency: ${efficiency * 100}%.1f%%")
    println(f"Memory (sequential): ${(memoryAfterSeq - memoryBefore) / 1024 / 1024}%.1f MB")
    println(f"Memory (parallel): ${(memoryAfterPar - memoryAfterSeq) / 1024 / 1024}%.1f MB")

    if (speedup > 1.2) {
      println("✓ Good parallelization candidate")
    } else if (speedup > 0.8) {
      println("⚠ Marginal benefit from parallelization")
    } else {
      println("✗ Poor parallelization candidate")
    }
  }
}

val profiler = new ParallelPerformanceProfiler()

// Profile different operations
profiler.profile(
  (1 to 10000).toVector,
  (x: Int) => math.sqrt(x * x + 1),
  "Square root calculation"
)

profiler.profile(
  (1 to 100000).toVector,
  (x: Int) => x.toString.length,
  "String conversion"
)

profiler.profile(
  (1 to 5000).toVector,
  expensiveFunction,
  "Expensive computation"
)

Advanced Parallel Patterns

Custom Parallel Algorithms

// Parallel merge sort implementation
object ParallelMergeSort {
  def mergeSort[T](arr: Array[T])(implicit ord: Ordering[T]): Array[T] = {
    if (arr.length <= 1) arr
    else {
      val mid = arr.length / 2
      val (left, right) = arr.splitAt(mid)

      // Parallel execution of recursive calls
      val sortedLeft = if (left.length > 1000) {
        // Use parallel for large arrays
        Future { mergeSort(left) }(scala.concurrent.ExecutionContext.global)
      } else {
        scala.concurrent.Future.successful(mergeSort(left))
      }

      val sortedRight = if (right.length > 1000) {
        Future { mergeSort(right) }(scala.concurrent.ExecutionContext.global)
      } else {
        scala.concurrent.Future.successful(mergeSort(right))
      }

      // Merge results
      import scala.concurrent.Await
      import scala.concurrent.duration._

      merge(Await.result(sortedLeft, 10.seconds), Await.result(sortedRight, 10.seconds))
    }
  }

  private def merge[T](left: Array[T], right: Array[T])(implicit ord: Ordering[T]): Array[T] = {
    val result = new Array[T](left.length + right.length)(scala.reflect.ClassTag(left.getClass.getComponentType))
    var i, j, k = 0

    while (i < left.length && j < right.length) {
      if (ord.lteq(left(i), right(j))) {
        result(k) = left(i)
        i += 1
      } else {
        result(k) = right(j)
        j += 1
      }
      k += 1
    }

    while (i < left.length) {
      result(k) = left(i)
      i += 1
      k += 1
    }

    while (j < right.length) {
      result(k) = right(j)
      j += 1
      k += 1
    }

    result
  }
}

// Test parallel merge sort
val unsortedArray = Array.fill(50000)(scala.util.Random.nextInt(100000))

timeOperation("Parallel merge sort") {
  val sorted = ParallelMergeSort.mergeSort(unsortedArray.clone())
  sorted.length
}

timeOperation("Built-in sort") {
  val sorted = unsortedArray.clone()
  scala.util.Sorting.quickSort(sorted)
  sorted.length
}

// Parallel matrix multiplication
class ParallelMatrix(val data: Array[Array[Double]]) {
  val rows: Int = data.length
  val cols: Int = if (rows > 0) data(0).length else 0

  def *(other: ParallelMatrix): ParallelMatrix = {
    require(this.cols == other.rows, "Matrix dimensions don't match")

    val result = Array.ofDim[Double](this.rows, other.cols)

    // Parallel computation of result matrix
    (0 until this.rows).par.foreach { i =>
      (0 until other.cols).par.foreach { j =>
        result(i)(j) = (0 until this.cols).map { k =>
          this.data(i)(k) * other.data(k)(j)
        }.sum
      }
    }

    new ParallelMatrix(result)
  }

  def +(other: ParallelMatrix): ParallelMatrix = {
    require(this.rows == other.rows && this.cols == other.cols, "Matrix dimensions don't match")

    val result = Array.ofDim[Double](rows, cols)

    (0 until rows).par.foreach { i =>
      (0 until cols).foreach { j =>
        result(i)(j) = this.data(i)(j) + other.data(i)(j)
      }
    }

    new ParallelMatrix(result)
  }
}

// Test parallel matrix operations
val matrix1 = new ParallelMatrix(Array.fill(200, 200)(scala.util.Random.nextDouble()))
val matrix2 = new ParallelMatrix(Array.fill(200, 200)(scala.util.Random.nextDouble()))

timeOperation("Parallel matrix addition") {
  val result = matrix1 + matrix2
  result.rows
}

timeOperation("Parallel matrix multiplication") {
  val result = matrix1 * matrix2
  result.rows
}

// Parallel search algorithms
object ParallelSearch {
  def parallelBinarySearch[T](
    arr: Array[T], 
    target: T,
    threshold: Int = 10000
  )(implicit ord: Ordering[T]): Option[Int] = {

    def search(start: Int, end: Int): Option[Int] = {
      if (start > end) None
      else if (end - start < threshold) {
        // Use sequential search for small ranges
        sequentialBinarySearch(arr, target, start, end)
      } else {
        val mid = start + (end - start) / 2
        val midValue = arr(mid)

        if (ord.equiv(midValue, target)) Some(mid)
        else if (ord.lt(midValue, target)) {
          // Search right half in parallel
          import scala.concurrent.{Future, ExecutionContext}
          implicit val ec: ExecutionContext = ExecutionContext.global

          val leftFuture = Future { search(start, mid - 1) }
          val rightResult = search(mid + 1, end)

          rightResult.orElse {
            import scala.concurrent.Await
            import scala.concurrent.duration._
            Await.result(leftFuture, 5.seconds)
          }
        } else {
          search(start, mid - 1)
        }
      }
    }

    search(0, arr.length - 1)
  }

  private def sequentialBinarySearch[T](
    arr: Array[T], 
    target: T, 
    start: Int, 
    end: Int
  )(implicit ord: Ordering[T]): Option[Int] = {
    var left = start
    var right = end

    while (left <= right) {
      val mid = left + (right - left) / 2
      val midValue = arr(mid)

      if (ord.equiv(midValue, target)) return Some(mid)
      else if (ord.lt(midValue, target)) left = mid + 1
      else right = mid - 1
    }

    None
  }
}

// Test parallel search
val sortedArray = (1 to 1000000).toArray
val searchTarget = 567890

timeOperation("Sequential binary search") {
  sortedArray.indexOf(searchTarget)
}

timeOperation("Parallel binary search") {
  ParallelSearch.parallelBinarySearch(sortedArray, searchTarget)
}

// Parallel aggregation patterns
object ParallelAggregation {

  case class Statistics(count: Long, sum: Double, min: Double, max: Double) {
    def mean: Double = if (count > 0) sum / count else 0.0

    def combine(other: Statistics): Statistics = {
      Statistics(
        count + other.count,
        sum + other.sum,
        math.min(min, other.min),
        math.max(max, other.max)
      )
    }
  }

  def computeStatistics(data: Vector[Double]): Statistics = {
    if (data.isEmpty) {
      Statistics(0, 0.0, Double.MaxValue, Double.MinValue)
    } else {
      data.par.aggregate(Statistics(0, 0.0, Double.MaxValue, Double.MinValue))(
        // Sequential operation: combine element with accumulator
        (stats, value) => Statistics(
          stats.count + 1,
          stats.sum + value,
          math.min(stats.min, value),
          math.max(stats.max, value)
        ),
        // Parallel operation: combine two accumulators
        (stats1, stats2) => stats1.combine(stats2)
      )
    }
  }

  def parallelGroupStatistics[K](
    data: Vector[(K, Double)]
  ): Map[K, Statistics] = {
    data.par
      .groupBy(_._1)
      .view
      .mapValues(pairs => computeStatistics(pairs.map(_._2).toVector))
      .toMap
  }
}

// Test parallel aggregation
val statisticsData = Vector.fill(100000)(scala.util.Random.nextDouble() * 100)

timeOperation("Parallel statistics computation") {
  val stats = ParallelAggregation.computeStatistics(statisticsData)
  println(f"Count: ${stats.count}, Mean: ${stats.mean}%.2f, Min: ${stats.min}%.2f, Max: ${stats.max}%.2f")
}

val groupedData = (1 to 50000).map(i => 
  (s"Group${i % 10}", scala.util.Random.nextDouble() * 100)
).toVector

timeOperation("Parallel group statistics") {
  val groupStats = ParallelAggregation.parallelGroupStatistics(groupedData)
  println(s"Computed statistics for ${groupStats.size} groups")
}

Summary

In this lesson, you've mastered parallel collections and data parallelism:

✅ Parallel Basics: Converting collections and basic operations
✅ Configuration: Task support and thread pool management
✅ Performance Analysis: When and how to use parallelization
✅ Guidelines: Identifying good candidates for parallelization
✅ Advanced Patterns: Custom parallel algorithms and aggregations
✅ Best Practices: Avoiding pitfalls and optimizing performance
✅ Real-world Applications: Matrix operations, search, and statistics

Parallel collections provide an easy way to leverage multi-core processors, but understanding when and how to use them effectively is crucial for building high-performance applications.

What's Next

In the next lesson, we'll explore the Actor model and Akka, learning how to build concurrent, distributed systems using message-passing concurrency that scales from single machines to clusters.

🚀 Practice Exercise

Put your knowledge to practice with hands-on coding exercises

Advanced ⏱️ 3h

Instructions

Parallel Collections

Build a comprehensive parallel computing framework that demonstrates advanced parallel collection patterns, custom parallel algorithms, performance optimization techniques, and sophisticated concurrent data processing systems.

Your program should:
1. Implement custom parallel collection operations and algorithms
2. Build a parallel data processing pipeline with load balancing
3. Create advanced parallel algorithms (sorting, searching, graph processing)
4. Implement parallel stream processing with backpressure handling
5. Build performance monitoring and optimization tools for parallel operations
6. Create fault-tolerant parallel processing systems with error recovery

Requirements:
- Implement custom parallel spliterators and combiners
- Build advanced parallel algorithms from scratch (merge sort, parallel search)
- Create load-balanced parallel processing with work stealing
- Implement parallel stream processing with proper resource management
- Build performance profiling and optimization tools
- Create fault-tolerant parallel systems with graceful degradation

Example usage:
```scala
val processor = AdvancedParallelProcessor(cores = 8)
val pipeline = processor
.createPipeline(dataset)
.withLoadBalancing()
.withFaultTolerance()
.process()
```

Interactive Playground

⚠️ Spoiler Alert! Try to solve the exercise yourself first. Learning happens through practice!

```scala
import scala.collection.parallel.CollectionConverters._
import scala.collection.parallel.{ParIterable, ParSeq}
import scala.collection.parallel.ForkJoinTaskSupport
import scala.collection.parallel.mutable.ParArray
import java.util.concurrent.{ForkJoinPool, RecursiveTask, ForkJoinTask}
import java.util.concurrent.atomic.{AtomicInteger, AtomicLong, AtomicReference}
import scala.util.{Random, Try, Success, Failure}
import scala.collection.mutable
import java.util.concurrent.ConcurrentHashMap
import scala.jdk.CollectionConverters._

object AdvancedParallelFramework extends App {
println("=== Advanced Parallel Collections Framework ===")

// 1. Custom Parallel Algorithms - Parallel Merge Sort
println("\n1. Custom Parallel Merge Sort Implementation")

class ParallelMergeSort[T: Ordering] {
private val threshold = 1000 // Switch to sequential below this size

def sort(array: Array[T]): Array[T] = {
val forkJoinPool = new ForkJoinPool()
try {
val task = new MergeSortTask(array, 0, array.length)
forkJoinPool.invoke(task)
} finally {
forkJoinPool.shutdown()
}
}

private class MergeSortTask(array: Array[T], start: Int, end: Int)
extends RecursiveTask[Array[T]] {

def compute(): Array[T] = {
val length = end - start

if (length <= threshold) {
// Sequential sort for small arrays
val subArray = array.slice(start, end)
subArray.sorted
} else {
// Parallel divide and conquer
val mid = start + length / 2

val leftTask = new MergeSortTask(array, start, mid)
val rightTask = new MergeSortTask(array, mid, end)

leftTask.fork()
val rightResult = rightTask.compute()
val leftResult = leftTask.join()

merge(leftResult, rightResult)
}
}

private def merge(left: Array[T], right: Array[T]): Array[T] = {
val result = new Array[T](left.length + right.length)
var i = 0
var j = 0
var k = 0

val ordering = implicitly[Ordering[T]]

while (i < left.length && j < right.length) {
if (ordering.lteq(left(i), right(j))) {
result(k) = left(i)
i += 1
} else {
result(k) = right(j)
j += 1
}
k += 1
}

while (i < left.length) {
result(k) = left(i)
i += 1
k += 1
}

while (j < right.length) {
result(k) = right(j)
j += 1
k += 1
}

result
}
}
}

// Test parallel merge sort
val testSize = 100000
val testArray = Array.fill(testSize)(Random.nextInt(1000000))
val testArrayCopy = testArray.clone()

def timeOperation[T](name: String)(operation: => T): T = {
val start = System.currentTimeMillis()
val result = operation
val end = System.currentTimeMillis()
println(f"$name took ${end - start}%,d ms")
result
}

val parallelSorter = new ParallelMergeSort[Int]()

val sequentialSorted = timeOperation("Sequential sort") {
testArray.sorted
}

val parallelSorted = timeOperation("Parallel merge sort") {
parallelSorter.sort(testArrayCopy)
}

println(s"Arrays match: ${sequentialSorted.sameElements(parallelSorted)}")
println(s"Array size: ${testSize:,} elements")

// 2. Advanced Parallel Data Pipeline
println("\n2. Advanced Parallel Data Pipeline")

case class DataRecord(id: Long, value: Double, category: String, timestamp: Long)
case class ProcessedRecord(id: Long, processedValue: Double, category: String, processingTime: Long)
case class AggregateResult(category: String, count: Int, sum: Double, avg: Double, min: Double, max: Double)

class AdvancedParallelPipeline {
private val processingStats = new AtomicLong(0)
private val errorCount = new AtomicInteger(0)

def createLargeDataset(size: Int): Vector[DataRecord] = {
val categories = Vector("A", "B", "C", "D", "E")
(1L to size.toLong).map { i =>
DataRecord(
id = i,
value = Random.nextDouble() * 1000,
category = categories(Random.nextInt(categories.length)),
timestamp = System.currentTimeMillis() + i * 1000
)
}.toVector
}

def processRecord(record: DataRecord): ProcessedRecord = {
val start = System.currentTimeMillis()

// Simulate complex processing
var processedValue = record.value
for (_ <- 1 to 100) {
processedValue = math.sqrt(processedValue * processedValue + 1.0)
}

// Simulate occasional processing errors
if (Random.nextDouble() < 0.01) {
errorCount.incrementAndGet()
throw new RuntimeException(s"Processing failed for record ${record.id}")
}

val processingTime = System.currentTimeMillis() - start
processingStats.incrementAndGet()

ProcessedRecord(record.id, processedValue, record.category, processingTime)
}

def processDatasetParallel(dataset: Vector[DataRecord],
parallelismLevel: Int = Runtime.getRuntime.availableProcessors()):
(Vector[ProcessedRecord], Map[String, AggregateResult]) = {

// Configure custom task support
val customTaskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelismLevel))
val parallelDataset = dataset.par
parallelDataset.tasksupport = customTaskSupport

try {
// Parallel processing with error handling
val processed = parallelDataset.map { record =>
Try(processRecord(record)) match {
case Success(processedRecord) => Some(processedRecord)
case Failure(_) => None
}
}.seq.flatten

// Parallel aggregation
val parallelProcessed = processed.par
parallelProcessed.tasksupport = customTaskSupport

val aggregates = parallelProcessed
.groupBy(_.category)
.mapValues { records =>
val values = records.map(_.processedValue)
AggregateResult(
category = records.head.category,
count = records.size,
sum = values.sum,
avg = values.sum / values.size,
min = values.min,
max = values.max
)
}.seq

(processed, aggregates)
} finally {
customTaskSupport.environment.shutdown()
}
}

def getStats: (Long, Int) = (processingStats.get(), errorCount.get())
}

val pipeline = new AdvancedParallelPipeline()
val largeDataset = pipeline.createLargeDataset(50000)

println(s"Processing dataset of ${largeDataset.size:,} records")

val (processedRecords, aggregates) = timeOperation("Parallel pipeline processing") {
pipeline.processDatasetParallel(largeDataset, parallelismLevel = 6)
}

val (processed, errors) = pipeline.getStats
println(s"Pipeline stats - Processed: $processed, Errors: $errors")
println(s"Output records: ${processedRecords.size:,}")

println("Category aggregates:")
aggregates.toSeq.sortBy(_._1).foreach { case (category, result) =>
println(f"  $category: count=${result.count}%,d, avg=${result.avg}%.2f, range=${result.min}%.2f-${result.max}%.2f")
}

// 3. Work-Stealing Parallel Processing
println("\n3. Work-Stealing Parallel Task System")

trait ParallelTask[T] {
def compute(): T
def canSplit: Boolean
def split(): (ParallelTask[T], ParallelTask[T])
def threshold: Int = 1000
}

class ParallelSum(data: Vector[Int], start: Int, end: Int) extends ParallelTask[Long] {
def compute(): Long = data.slice(start, end).map(_.toLong).sum

def canSplit: Boolean = (end - start) > threshold

def split(): (ParallelTask[Long], ParallelTask[Long]) = {
val mid = start + (end - start) / 2
(new ParallelSum(data, start, mid), new ParallelSum(data, mid, end))
}
}

class WorkStealingExecutor(parallelism: Int = Runtime.getRuntime.availableProcessors()) {
private val forkJoinPool = new ForkJoinPool(parallelism)

def execute[T](task: ParallelTask[T]): T = {
val forkJoinTask = new ForkJoinTaskWrapper(task)
forkJoinPool.invoke(forkJoinTask)
}

def shutdown(): Unit = forkJoinPool.shutdown()

private class ForkJoinTaskWrapper[T](task: ParallelTask[T]) extends RecursiveTask[T] {
def compute(): T = {
if (task.canSplit) {
val (leftTask, rightTask) = task.split()
val leftWrapper = new ForkJoinTaskWrapper(leftTask)
val rightWrapper = new ForkJoinTaskWrapper(rightTask)

leftWrapper.fork()
val rightResult = rightWrapper.compute()
val leftResult = leftWrapper.join()

// Combine results (this is simplified - in real implementation,
// the task would define how to combine)
rightResult
} else {
task.compute()
}
}
}
}

val workStealingData = (1 to 1000000).toVector
val workStealingSum = new ParallelSum(workStealingData, 0, workStealingData.length)
val workStealingExecutor = new WorkStealingExecutor(4)

val workStealingResult = timeOperation("Work-stealing parallel sum") {
workStealingExecutor.execute(workStealingSum)
}

val expectedSum = workStealingData.map(_.toLong).sum
println(s"Work-stealing result: ${workStealingResult:,}")
println(s"Expected result: ${expectedSum:,}")
println(s"Results match: ${workStealingResult == expectedSum}")

workStealingExecutor.shutdown()

// 4. Parallel Graph Processing
println("\n4. Parallel Graph Processing")

case class Graph[T](adjacencyList: Map[T, Set[T]]) {
def nodes: Set[T] = adjacencyList.keySet ++ adjacencyList.values.flatten
def neighbors(node: T): Set[T] = adjacencyList.getOrElse(node, Set.empty)
}

class ParallelGraphProcessor[T] {

def parallelBFS(graph: Graph[T], start: T, parallelism: Int = 4): List[T] = {
val visited = ConcurrentHashMap.newKeySet[T]()
val result = mutable.ListBuffer[T]()
val currentLevel = ConcurrentHashMap.newKeySet[T]()

visited.add(start)
currentLevel.add(start)
result += start

val taskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelism))

while (!currentLevel.isEmpty) {
val currentNodes = currentLevel.asScala.toVector
currentLevel.clear()

// Process current level in parallel
val parallelNodes = currentNodes.par
parallelNodes.tasksupport = taskSupport

val nextLevelNodes = parallelNodes.flatMap { node =>
graph.neighbors(node).filter(neighbor => !visited.contains(neighbor))
}.seq.distinct

nextLevelNodes.foreach { node =>
if (visited.add(node)) { // add returns true if not already present
currentLevel.add(node)
result += node
}
}
}

taskSupport.environment.shutdown()
result.toList
}

def parallelConnectedComponents(graph: Graph[T], parallelism: Int = 4): Map[T, Int] = {
val visited = ConcurrentHashMap.newKeySet[T]()
val components = new ConcurrentHashMap[T, Int]()
val componentId = new AtomicInteger(0)

val taskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelism))

try {
val nodes = graph.nodes.toVector.par
nodes.tasksupport = taskSupport

nodes.foreach { node =>
if (!visited.contains(node)) {
val currentComponentId = componentId.getAndIncrement()
val component = parallelBFS(graph, node, 2) // Use smaller parallelism for subproblems
component.foreach { componentNode =>
visited.add(componentNode)
components.put(componentNode, currentComponentId)
}
}
}
} finally {
taskSupport.environment.shutdown()
}

components.asScala.toMap
}
}

// Create a test graph
val testGraph = Graph(Map(
"A" -> Set("B", "C"),
"B" -> Set("A", "D"),
"C" -> Set("A", "E"),
"D" -> Set("B", "F"),
"E" -> Set("C", "G"),
"F" -> Set("D"),
"G" -> Set("E"),
"H" -> Set("I"), // Separate component
"I" -> Set("H", "J"),
"J" -> Set("I"),
"K" -> Set(), // Isolated node
"L" -> Set("M", "N"), // Another component
"M" -> Set("L", "N"),
"N" -> Set("L", "M")
))

val graphProcessor = new ParallelGraphProcessor[String]()

val bfsResult = timeOperation("Parallel BFS from A") {
graphProcessor.parallelBFS(testGraph, "A")
}
println(s"BFS traversal from A: ${bfsResult.mkString(" -> ")}")

val components = timeOperation("Parallel connected components") {
graphProcessor.parallelConnectedComponents(testGraph)
}
println("Connected components:")
components.groupBy(_._2).foreach { case (componentId, nodes) =>
println(s"  Component $componentId: ${nodes.keys.toSeq.sorted.mkString(", ")}")
}

// 5. Parallel Stream Processing with Backpressure
println("\n5. Parallel Stream Processing")

class ParallelStreamProcessor[T, R](
bufferSize: Int = 1000,
parallelism: Int = Runtime.getRuntime.availableProcessors()
) {

private val taskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelism))

def process[U](stream: Iterator[T])(
transform: T => R
)(
reduce: (R, R) => R
): R = {

val batches = stream.grouped(bufferSize).toVector
println(s"Processing ${batches.length} batches with buffer size $bufferSize")

val parallelBatches = batches.par
parallelBatches.tasksupport = taskSupport

// Process each batch in parallel
val batchResults = parallelBatches.map { batch =>
batch.map(transform).reduce(reduce)
}

// Combine batch results
if (batchResults.nonEmpty) {
batchResults.reduce(reduce)
} else {
throw new IllegalArgumentException("Empty stream")
}
}

def processWithMonitoring[U](stream: Iterator[T])(
transform: T => R
)(
reduce: (R, R) => R
): (R, ProcessingStats) = {

val stats = new ProcessingStats()
val startTime = System.currentTimeMillis()

val result = process(stream) { item =>
stats.incrementProcessed()
val itemStart = System.nanoTime()
try {
val transformed = transform(item)
val itemEnd = System.nanoTime()
stats.addProcessingTime(itemEnd - itemStart)
transformed
} catch {
case ex: Exception =>
stats.incrementErrors()
throw ex
}
}(reduce)

stats.setTotalTime(System.currentTimeMillis() - startTime)
(result, stats)
}

def shutdown(): Unit = {
taskSupport.environment.shutdown()
}
}

class ProcessingStats {
private val processedCount = new AtomicLong(0)
private val errorCount = new AtomicInteger(0)
private val totalProcessingTime = new AtomicLong(0)
private var totalTime: Long = 0

def incrementProcessed(): Unit = processedCount.incrementAndGet()
def incrementErrors(): Unit = errorCount.incrementAndGet()
def addProcessingTime(time: Long): Unit = totalProcessingTime.addAndGet(time)
def setTotalTime(time: Long): Unit = totalTime = time

def getProcessedCount: Long = processedCount.get()
def getErrorCount: Int = errorCount.get()
def getAverageProcessingTime: Double = {
val count = processedCount.get()
if (count > 0) totalProcessingTime.get().toDouble / count / 1_000_000.0 // Convert to milliseconds
else 0.0
}
def getTotalTime: Long = totalTime
def getThroughput: Double = {
if (totalTime > 0) processedCount.get().toDouble / (totalTime / 1000.0) // items per second
else 0.0
}
}

val streamProcessor = new ParallelStreamProcessor[Int, Long](bufferSize = 5000, parallelism = 4)
val streamData = (1 to 100000).iterator

val (streamResult, streamStats) = timeOperation("Parallel stream processing") {
streamProcessor.processWithMonitoring(streamData) { x =>
// Simulate some processing
var result = x.toLong
for (_ <- 1 to 10) {
result = math.sqrt(result * result + 1.0).toLong
}
result
}(_ + _)
}

println(s"Stream processing result: ${streamResult:,}")
println(f"Processing stats:")
println(f"  Processed: ${streamStats.getProcessedCount:,} items")
println(f"  Errors: ${streamStats.getErrorCount}")
println(f"  Average processing time: ${streamStats.getAverageProcessingTime}%.3f ms/item")
println(f"  Throughput: ${streamStats.getThroughput:,.0f} items/second")

streamProcessor.shutdown()

// 6. Performance Monitoring and Optimization
println("\n6. Performance Monitoring Framework")

class ParallelPerformanceMonitor {
private val metrics = new ConcurrentHashMap[String, PerformanceMetric]()

def measureParallelOperation[T](operationName: String, parallelLevels: List[Int])
(operation: Int => T): ParallelPerformanceResult[T] = {

val results = parallelLevels.map { level =>
val metric = new PerformanceMetric()
val startTime = System.currentTimeMillis()
val startMemory = getUsedMemory()

val result = operation(level)

val endTime = System.currentTimeMillis()
val endMemory = getUsedMemory()

metric.setExecutionTime(endTime - startTime)
metric.setMemoryUsed(endMemory - startMemory)
metric.setParallelismLevel(level)

metrics.put(s"${operationName}_$level", metric)

(level, endTime - startTime, result)
}

ParallelPerformanceResult(operationName, results)
}

private def getUsedMemory(): Long = {
val runtime = Runtime.getRuntime
runtime.totalMemory() - runtime.freeMemory()
}

def getMetrics: Map[String, PerformanceMetric] = metrics.asScala.toMap

def printReport(): Unit = {
println("=== Performance Report ===")
metrics.asScala.toSeq.sortBy(_._1).foreach { case (name, metric) =>
println(f"$name:")
println(f"  Parallelism: ${metric.getParallelismLevel}")
println(f"  Time: ${metric.getExecutionTime}%,d ms")
println(f"  Memory: ${metric.getMemoryUsed / (1024 * 1024)}%,d MB")
}
}
}

case class ParallelPerformanceResult[T](
operationName: String,
results: List[(Int, Long, T)]
) {
def getBestParallelismLevel: Int = {
results.minBy(_._2)._1
}

def getSpeedupFactors: List[(Int, Double)] = {
val baselineTime = results.find(_._1 == 1).map(_._2.toDouble).getOrElse(results.head._2.toDouble)
results.map { case (level, time, _) =>
(level, baselineTime / time)
}
}

def printAnalysis(): Unit = {
println(s"Performance Analysis for $operationName:")
println("Parallelism Level -> Time (ms) -> Speedup Factor")
getSpeedupFactors.foreach { case (level, speedup) =>
val time = results.find(_._1 == level).get._2
println(f"  $level%2d cores -> $time%,6d ms -> ${speedup}%.2fx speedup")
}
println(f"Best parallelism level: ${getBestParallelismLevel} cores")
}
}

class PerformanceMetric {
private var executionTime: Long = 0
private var memoryUsed: Long = 0
private var parallelismLevel: Int = 0

def setExecutionTime(time: Long): Unit = executionTime = time
def setMemoryUsed(memory: Long): Unit = memoryUsed = memory
def setParallelismLevel(level: Int): Unit = parallelismLevel = level

def getExecutionTime: Long = executionTime
def getMemoryUsed: Long = memoryUsed
def getParallelismLevel: Int = parallelismLevel
}

val monitor = new ParallelPerformanceMonitor()
val testDataForMonitoring = (1 to 200000).toVector

val performanceResult = monitor.measureParallelOperation(
"matrix_multiplication_simulation",
List(1, 2, 4, 6, 8)
) { parallelismLevel =>
val taskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelismLevel))
val parallelData = testDataForMonitoring.par
parallelData.tasksupport = taskSupport

try {
// Simulate matrix multiplication workload
parallelData.map { x =>
var result = x.toDouble
for (_ <- 1 to 200) {
result = math.sqrt(result * result + 1.0)
}
result.toInt
}.sum
} finally {
taskSupport.environment.shutdown()
}
}

performanceResult.printAnalysis()
monitor.printReport()

// 7. Fault-Tolerant Parallel Processing
println("\n7. Fault-Tolerant Parallel Processing")

class FaultTolerantParallelProcessor[T, R](
maxRetries: Int = 3,
parallelism: Int = Runtime.getRuntime.availableProcessors()
) {

private val taskSupport = new ForkJoinTaskSupport(new ForkJoinPool(parallelism))
private val failureCount = new AtomicInteger(0)
private val retryCount = new AtomicInteger(0)

def processWithFaultTolerance(data: Vector[T])(
processor: T => R
): (Vector[R], FaultToleranceStats) = {

val parallelData = data.par
parallelData.tasksupport = taskSupport

val results = parallelData.map { item =>
processItemWithRetry(item, processor, maxRetries)
}.seq.flatten

val stats = FaultToleranceStats(
totalItems = data.size,
successfulItems = results.size,
failedItems = data.size - results.size,
totalRetries = retryCount.get(),
totalFailures = failureCount.get()
)

(results, stats)
}

private def processItemWithRetry(item: T, processor: T => R, retriesLeft: Int): Option[R] = {
Try(processor(item)) match {
case Success(result) => Some(result)
case Failure(ex) =>
failureCount.incrementAndGet()
if (retriesLeft > 0) {
retryCount.incrementAndGet()
Thread.sleep(10) // Brief delay before retry
processItemWithRetry(item, processor, retriesLeft - 1)
} else {
None
}
}
}

def shutdown(): Unit = taskSupport.environment.shutdown()
}

case class FaultToleranceStats(
totalItems: Int,
successfulItems: Int,
failedItems: Int,
totalRetries: Int,
totalFailures: Int
) {
def successRate: Double = successfulItems.toDouble / totalItems
def retryRate: Double = totalRetries.toDouble / totalItems

def printStats(): Unit = {
println("Fault Tolerance Stats:")
println(f"  Total items: $totalItems%,d")
println(f"  Successful: $successfulItems%,d")
println(f"  Failed: $failedItems%,d")
println(f"  Success rate: ${successRate * 100}%.1f%%")
println(f"  Total retries: $totalRetries%,d")
println(f"  Retry rate: ${retryRate}%.2f retries/item")
println(f"  Total failures: $totalFailures%,d")
}
}

val faultTolerantProcessor = new FaultTolerantParallelProcessor[Int, String](maxRetries = 2, parallelism = 4)
val faultTestData = (1 to 10000).toVector

// Processor that randomly fails
def unreliableProcessor(x: Int): String = {
if (Random.nextDouble() < 0.1) { // 10% failure rate
throw new RuntimeException(s"Processing failed for $x")
}
Thread.sleep(1) // Simulate some work
s"Processed-$x"
}

val (faultResults, faultStats) = timeOperation("Fault-tolerant parallel processing") {
faultTolerantProcessor.processWithFaultTolerance(faultTestData)(unreliableProcessor)
}

faultStats.printStats()
println(s"Results sample: ${faultResults.take(5).mkString(", ")}")

faultTolerantProcessor.shutdown()

println("\n=== Advanced Parallel Framework Complete! ===")
println("Advanced concepts demonstrated:")
println("- Custom parallel merge sort with Fork/Join framework")
println("- Advanced parallel data processing pipeline")
println("- Work-stealing parallel execution model")
println("- Parallel graph processing algorithms")
println("- Parallel stream processing with backpressure")
println("- Performance monitoring and optimization tools")
println("- Fault-tolerant parallel processing with retry logic")
println("- Resource management and proper cleanup")
}
```

Comments

Be the first to comment on this lesson!

Parallel Collections: Data Parallelism

Introduction

Understanding Parallel Collections

Basic Parallel Operations

Parallel Collection Types

Configuring Parallelism

Task Support and Thread Pools

When to Use Parallel Collections

Performance Analysis and Guidelines

Advanced Parallel Patterns

Custom Parallel Algorithms

Summary

What's Next

🚀 Practice Exercise

Instructions

Interactive Playground

Comments

Sign In

Sign Up

Reset Password

Sign Out

Reset Password