Как установить количество потоков по умолчанию для Scala 2.10 параллельных коллекций?

В Scala до 2.10 я могу установить parallelism в defaultForkJoinPool (как в этом ответе scala степень параллельных коллекций parallelism), В Scala 2.10 этот API больше не существует. Хорошо документировано, что мы можем установить parallelism в одну коллекцию (http://docs.scala-lang.org/overviews/parallel-collections/configuration.html), назначив свой объект taskSupport.

Тем не менее, я использую параллельные коллекции по всей своей кодовой базе и не хотел бы добавлять дополнительные две строки для каждого экземпляра коллекции. Есть ли способ настроить общий размер пула потоков по умолчанию, чтобы someCollection.par.map(f(_)) автоматически использовал количество потоков по умолчанию?

Ответы

Ответ 1

Я знаю, что вопрос уже не один месяц, но у меня точно такой же вопрос. Googling не помог, и я не смог найти ничего похожего на новый API.

Настройка -Dscala.concurrent.context.maxThreads = n, как предлагается здесь: Установить уровень parallelism для всех коллекций в Scala 2.10?, казалось бы, не было эффект, но я не уверен, правильно ли я использовал его (я запускаю свое приложение с "java" в среде без "scala", установленное явно, это может быть причиной).

Я не знаю, почему scala -people удалил этот основной сеттер из соответствующего объекта пакета.

Однако часто можно использовать отражение для работы с неполным/странным интерфейсом:

def setParallelismGlobally(numThreads: Int): Unit = {
  val parPkgObj = scala.collection.parallel.`package`
  val defaultTaskSupportField = parPkgObj.getClass.getDeclaredFields.find{
    _.getName == "defaultTaskSupport"
  }.get

  defaultTaskSupportField.setAccessible(true)
  defaultTaskSupportField.set(
    parPkgObj, 
    new scala.collection.parallel.ForkJoinTaskSupport(
      new scala.concurrent.forkjoin.ForkJoinPool(numThreads)
    ) 
  )
}

Для тех, кто не знаком с более неясными особенностями Scala, вот короткое объяснение:

scala.collection.parallel.`package`

обращается к объекту пакета с переменной defaultTaskSupport (он похож на статическую переменную Java, но на самом деле является переменной-членом объекта пакета). Для идентификатора требуются обратные элементы, поскольку package - зарезервированное ключевое слово. Затем мы получаем частное конечное поле, которое мы хотим (getField ( "defaultTaskSupport" ) по какой-то причине не работает?...), скажите, чтобы он был доступен, чтобы иметь возможность его модифицировать, а затем заменить его значением на наш собственный ForkJoinTaskSupport.

Я пока не понимаю точный механизм создания параллельных коллекций, но исходный код признака Combiner предполагает, что значение defaultTaskSupport должно как-то просачиваться в параллельные коллекции.

Обратите внимание, что вопрос качественно такой же, как и более старый вопрос: "У меня есть Math.random() по всей моей кодовой базе, как я могу установить семя на фиксированное число для целей отладки?" (См. Например: Установить семя на Math.random()). В обоих случаях у нас есть какая-то глобальная "статическая" переменная, которую мы неявно используем в миллионах разных мест, мы хотим ее изменить, но для этой переменной = > мы не используем отображение.

Уродливый, черт возьми, но, похоже, все отлично. Если вам нужно ограничить общее количество потоков, не забывайте, что сборщик мусора работает в отдельном потоке.