Scala параллельные коллекции и изменяемое состояние

Question

У меня есть функция, которая выполняет один шаг вычислений путем обновления карты (в основе лежит изменяемый HashMap). Я хочу сделать несколько таких вычислений параллельно (каждая «цепочка» работает со своим изменяемым HashMap).

Я сделал это, поместив HashMaps в параллельную коллекцию и применив функцию к каждому HashMap, используя map.

Теперь я испытал пропущенные записи на картах. При отладке карта содержит запись, как только точка останова исключения останавливает программу (и тогда перезапуск вычисления немного раньше, отбрасывая некоторые слои стековых фреймов).

Это поведение исчезает, когда я использую последовательные коллекции. Так возможно ли, что есть какое-то неправильное поведение (или ошибка), вызванное тем, что один и тот же HashMap работает в разных потоках?

Я не опубликовал пример кода, так как я не думаю, что поведение воспроизводимо. Насколько мне известно, единственные изменяемые данные содержатся внутри тех HashMaps, которые содержат состояние вычислений.

По запросу образец моего кода, в котором создается параллельная карта (reset) и изменяется (step).

class ParallelInferer[V <: DiscreteVariable, TInf <: InferenceAlgorithm[V]](val numChains: Int,
                                              val inferer: InferenceAlgorithm[V],
                                              val random: Random) {
  //tuples of inferer state and the according random number generator
  protected var states: ParSeq[(inferer.Repr,Random)] = _
  protected var steps = 0

  reset()

  def reset() {
    steps = 0

    val seed = random.nextInt()

    //todo why does parallelizing fail here (key not found on a map)
    states = (1 to numChains).par
      .map(n => new Random(seed + n))    //create the rngs
      .map(rng => (inferer.createInitialState(rng),rng))
  }

  def step() {
    //advance every chain by one
    states = states.map{case (repr, rng) => (inferer.computeStep(repr, rng),rng)}
    steps = steps + 1
  }
}

Пояснение к коду

Класс ParallelInferer предназначен (также) для неизменного вывода. Таким образом, изменчивость не видна непосредственно в опубликованном коде, но я думаю, что это важная часть, которая показана.

У каждого алгоритма логического вывода есть понятие состояния, это состояние имеет тип InferenceAlgorithm#Repr - как видно из использования inferer.Repr как части переменной states. Логические устройства работают путем отображения Repr (и объекта Random) на новый Repr с их функцией computeStep. Это можно увидеть в def step(). Теперь некоторые умельцы используют изменяемый HashMap как часть своего состояния. Их метод computeStep возвращает ту же карту, которую он получил в качестве аргумента, после ее изменения.

Вопрос

1. Можно ли как-то исправить это поведение?
2. Неправильно ли я использую параллельные коллекции и нужно ли распараллеливать мою задачу иначе?

Редактировать

Я только что снова запустил распараллеленную версию, и я думаю, что это также приводит к тому, что алгоритм не завершает работу, хотя это происходит при последовательной работе. Ну, не , что удивительно, не правда ли?

Может кто-нибудь порассуждать о том, почему это происходит?

Answer 1

Да, абсолютно.Изменяемые HashMaps по умолчанию не являются потокобезопасными, и их использование таким образом может привести к неопределенному поведению.Пропущенные записи на самом деле довольно мягкое проявление.В зависимости от базовой реализации, также возможно повредить структуру данных HashMap до такой степени, что ваша программа заходит в бесконечный цикл.

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

import scala.collection.mutable._

val map = new HashMap[Key, Value] with SynchronizedMap[Key, Value] 

Я бы не сказал, что основная проблема заключается в том, что вы используете параллельные коллекции неправильно, а вместо этого любая *Параллельная программа 1007 *, использующая изменяемые структуры данных, будет иметь такие проблемы.Гораздо более простой способ сделать это - использовать неизменяемые карты, и ваши параллельные шаги обработки возвращают новые неизменяемые карты.Это звучит вычислительно дорого, но не обязательно, из-за базовой реализации неизменяемых хэш-карт.