Updating Scala collections thread-safely

Question

i was wondering if there is any 'easy' way to update immutable scala collections safely. Consider following code:

class a {
   private var x = Map[Int,Int]()

   def update(p:(Int,Int)) { x = x + (p) }
}

This code is not thread safe, correct? By that i mean that if we have two threads invoking update method and lets say that x is map containing { 1=>2 } and thread A invokes update((3,4)) and only manages to execute the x + (p) part of the code. Then rescheduling occurs and thread B invokes update((13,37)) and successfully updates the variable x. The thread A continues and finishes.

After all this finishes, value x would equal map containing { 1=>2, 3=>4 }, correct? Instead of desired { 1=>2, 3=>4, 13=>37 }. Is there a simple way to fix that? I hope it's undestandable what I'm asking :)

Btw, i know there are solutions like Akka STM but i would prefer not to use those, unless necessary.

Thanks a lot for any answer!

edit: Also, i would prefer solution without locking. Eeeew :)

Answer 1

In your case, as Maurício wrote, your collection is already thread safe because it is immutable. The only problem is reassigning the var, which may not be an atomic operation. For this particular problem, the easiest option is to use of the nice classes in java.util.concurrent.atomic, namely AtomicReference.

import java.util.concurrent.atomic.AtomicReference

class a {
  private val x = new AtomicReference(Map[Int,Int]())

  def update(p:(Int,Int)) {
    while (true) {
      val oldMap = x.get // get old value
      val newMap = oldMap + p // update
      if (x.compareAndSet(oldMap, newMap))
        return // exit if update was successful, else repeat
    }
  }
}

Answer 2

The collection itself is thread safe as it has no shared mutable state, but your code is not and there is no way to fix this without locking, as you do have shared mutable state. Your best option is to lock the method itself marking it as synchronized.

The other solution would be use a mutable concurrent map, possibly java.util.concurrent.ConcurrentMap.

Answer 3

1. out of the box
2. atomicity
3. lock-free
4. O(1)

Check this out: http://www.scala-lang.org/api/2.11.4/index.html#scala.collection.concurrent.TrieMap

Answer 4

Re. Jean-Philippe Pellet's answer: you can make this a little bit more re-usable:

def compareAndSetSync[T](ref: AtomicReference[T])(logic: (T => T)) {
  while(true) {
    val snapshot = ref.get
    val update = logic(snapshot)
    if (ref.compareAndSet(snapshot, update)) return
  }
}

def compareSync[T,V](ref: AtomicReference[T])(logic: (T => V)): V = {
  var continue = true
  var snapshot = ref.get
  var result = logic(snapshot)
  while (snapshot != ref.get) {
    snapshot = ref.get
    result = logic(snapshot)
  }
  result
}