Semaphore

Using locks in parallel programming is a common way to ensure that only one thread accesses a critical section of code at a time. This helps prevent race conditions and ensures data consistency. Scala relies on Java’s concurrency utilities for lock management. Here, we'll use ReentrantLock from the java.util.concurrent.locks package.

Using `ReentrantLock` in Scala

Importing the necessary classes:

import java.util.concurrent.locks.ReentrantLock
import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}
import scala.concurrent.duration._

Creating and using a `ReentrantLock`:

Create the lock:
```
val lock = new ReentrantLock()
```

Acquire and release the lock:

Use lock.lock() to acquire the lock and lock.unlock() to release it. It’s crucial to ensure that the lock is always released, typically using a try-finally block.

def accessResource(threadName: String): Unit = {
  lock.lock()
  try {
    println(s"$threadName is accessing the resource.")
    Thread.sleep(1000) // Simulate some work with the resource
  } finally {
    println(s"$threadName is releasing the resource.")
    lock.unlock()
  }
}

Run tasks in parallel:

Use Scala's Future to run tasks in parallel.

val tasks = for (i <- 1 to 10) yield Future {
  accessResource(s"Thread-$i")
}

val aggregatedFuture = Future.sequence(tasks)

Wait for all tasks to complete:

Use Await.result to wait for all futures to complete.
```
Await.result(aggregatedFuture, 10.seconds)
```

Complete Example

Here is the complete code example putting all the pieces together:

import java.util.concurrent.locks.ReentrantLock
import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}
import scala.concurrent.duration._

object LockExample extends App {
  // Create a reentrant lock
  val lock = new ReentrantLock()

  // Define a function to access the resource
  def accessResource(threadName: String): Unit = {
    lock.lock()
    try {
      println(s"$threadName is accessing the resource.")
      Thread.sleep(1000) // Simulate some work with the resource
    } finally {
      println(s"$threadName is releasing the resource.")
      lock.unlock()
    }
  }

  // Run tasks in parallel using Futures
  val tasks = for (i <- 1 to 10) yield Future {
    accessResource(s"Thread-$i")
  }

  // Aggregate all futures
  val aggregatedFuture = Future.sequence(tasks)

  // Wait for all tasks to complete
  Await.result(aggregatedFuture, 10.seconds)
}

LockExample.main(Array())

Explanation

ReentrantLock Creation: new ReentrantLock() creates a reentrant lock.
Access Resource Function: accessResource function locks the critical section, performs the operation (simulated by Thread.sleep), and then unlocks the critical section. The try-finally block ensures the lock is always released.
Parallel Tasks: A sequence of 10 futures is created, each calling accessResource with a different thread name.
Aggregated Future: Future.sequence combines the futures into a single future that completes when all the tasks are done.
Await Completion: Await.result waits for the aggregated future to complete, ensuring the main program does not exit prematurely.

Alternative: Using `synchronized`

In addition to using ReentrantLock, Scala provides a simpler way to achieve mutual exclusion using the synchronized keyword. This is a more concise way to protect critical sections of code.

Using `synchronized`:

object SynchronizedExample extends App {
  // Define a shared resource
  var sharedCounter = 0

  // Define a function to access the resource
  def incrementCounter(threadName: String): Unit = synchronized {
    println(s"$threadName is accessing the resource.")
    sharedCounter += 1
    println(s"$threadName incremented the counter to $sharedCounter.")
    Thread.sleep(1000) // Simulate some work with the resource
  }

  // Run tasks in parallel using Futures
  val tasks = for (i <- 1 to 10) yield Future {
    incrementCounter(s"Thread-$i")
  }

  // Aggregate all futures
  val aggregatedFuture = Future.sequence(tasks)

  // Wait for all tasks to complete
  Await.result(aggregatedFuture, 10.seconds)
}

SynchronizedExample.main(Array())

Explanation

Synchronized Block: The synchronized keyword is used to ensure that only one thread can execute the incrementCounter function at a time.
Shared Resource: A shared variable sharedCounter is incremented by each thread, ensuring mutual exclusion.
Parallel Tasks: A sequence of 10 futures is created, each calling incrementCounter with a different thread name.
Aggregated Future: Future.sequence combines the futures into a single future that completes when all the tasks are done.
Await Completion: Await.result waits for the aggregated future to complete, ensuring the main program does not exit prematurely.

Keyboard shortcuts

Learning Scala