Обновление коллекций Scala потокобезопасно - PullRequest
Новая
       7

Обновление коллекций Scala потокобезопасно

19 голосов
/ 29 июля 2011

Мне было интересно, есть ли какой-нибудь простой способ безопасно обновить неизменяемые коллекции scala.Рассмотрим следующий код: 

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

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

Этот код не является потокобезопасным, верно?Под этим я подразумеваю, что если у нас есть два потока, вызывающих метод обновления, и допустим, что x - это карта, содержащая {1 => 2}, а поток A вызывает обновление ((3,4)), и ему удается только выполнить x + (p)часть кода.Затем происходит перепланирование, и поток B вызывает update ((13,37)) и успешно обновляет переменную x.Нить А продолжается и заканчивается.

После того, как все это закончится, значение x будет равно карте, содержащей {1 => 2, 3 => 4}, правильно?Вместо желаемого {1 => 2, 3 => 4, 13 => 37}.Есть ли простой способ это исправить?Я надеюсь, что понятно то, что я спрашиваю:)

Кстати, я знаю, что есть решения, подобные Akka STM, но я бы предпочел не использовать их, если в этом нет необходимости.

Большое спасибо за любыеответ!

edit: Кроме того, я бы предпочел решение без блокировки.Eeeew:)

Ответы [ 4 ]

20 голосов
/ 29 июля 2011

В вашем случае, как писал Маурисио, ваша коллекция уже защищена от потоков, потому что она неизменна. Единственная проблема - переназначение var, которое не может быть атомарной операцией. Для этой конкретной проблемы проще всего использовать классы nice в java.util.concurrent.atomic, а именно 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
    }
  }
}
4 голосов
/ 11 февраля 2015
  1. из коробки
  2. валентность
  3. безблокировочный
  4. O (1)

Проверьте это: http://www.scala -lang.org / api / 2.11.4 / index.html # scala.collection.concurrent.TrieMap

3 голосов
/ 29 июля 2011

Сама коллекция является поточно-ориентированной, так как не имеет общего изменяемого состояния, но ваш код - нет, и нет способа исправить это без блокировки, так как у вас есть общее изменяемое состояние. Лучше всего заблокировать сам метод, помечая его как синхронизированный.

Другим решением будет использование изменяемой параллельной карты, возможно java.util.concurrent.ConcurrentMap .

1 голос
/ 24 января 2014

Re.Ответ Жан-Филиппа Пелле: вы можете сделать это немного более пригодным для повторного использования: 

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
}
...