Scala Параллелизм замедляется

Question

Я делаю предисловие к тому факту, что я относительный новичок в Java / Scala, поэтому я не исключаю, что есть нечто очевидное, чего я не делаю.

У меня есть приложение Scala, которое через Hibernate подключается к базе данных MySQL. Приложение предназначено для обработки большого объема данных, около 2 750 000 записей, поэтому я постарался максимально оптимизировать его.

Он работает на моей рабочей станции, представляющей собой QuadCore Intel Xeon с 6 ГБ ОЗУ (1033 МГц), и работает хорошо и быстро для первых 70 тыс. Записей, завершая их примерно за 15 минут. К тому времени, он достиг 90 КБ, это заняло около 25 минут, поэтому что-то замедляет его до ползания.

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

Код выглядит примерно так:

val recordCount = repo.recordCount
val batchSize = 100
val batches = (0 to recordCount by batchSize).toList
val batchJobs = {
    for (batchStart <- batches) yield {
        future(new RecordFormatter().formatRecords(new Repo(sessionFactory.openSession),batchStart,batchSize)
    }
awaitAll(100000,batchJobs: *_)

Внутри RecordFormatter (который на самом деле не назван так, если вы удивляетесь моей безумной схеме именования), он выполняет запрос для следующих 100 записей, а затем другой запрос, чтобы получить обратно фактические записи (используя между конечные значения) затем записывает их в текстовый файл как CSV. Если посмотреть на вывод таймера, каждая операция в форматере записи занимает около 5 секунд, чтобы извлечь записи, а затем 0,1 секунды, чтобы вывести их в файл.

Несмотря на это, после замедления он обрабатывает только около 12 пакетов по 100 записей в минуту, а не 40 пакетов по 100 записей в минуту при первом запуске процесса.

Он регулярно очищает сессию и закрывает ее в конце каждого запуска RecordFormatter (каждый RecordFormatter имеет свой собственный сеанс).

В основном я ищу какие-то известные ошибки со Scala и Futures. Я заметил, что когда он замедляется, он, кажется, не использует все восемь возможных потоков, которые, безусловно, могли бы объяснить падение скорости, но для меня загадка, почему он внезапно останавливается и всегда находится около отметки 75k.

Спасибо!

РЕДАКТИРОВАТЬ: обновленный код, чтобы показать, что он использует yield и awaitAll в случае, если это имеет значение.

Answer 1

Попробуйте ограничить максимальное количество потоков, которые создаст библиотека актеров (фьючерсы поддерживаются актерами). Потоки актеров ЧРЕЗВЫЧАЙНО тяжеловесны, и при определенных условиях планировщик создаст их, как будто завтра не наступит. Это занимает тонну кучи и может заставить вашу программу тратить огромное количество времени на сборку мусора.

Это можно сделать, установив параметр arguments.maxPoolSize в командной строке ... который будет выглядеть примерно так: -Dactors.maxPoolSize = 32 или любое другое максимальное число потоков, которое вы хотите.

Я также настоятельно рекомендую запустить вашу программу -Xprof, чтобы узнать, сколько времени занимает GC.

Answer 2

Приложение jconsole, которое поставляется в комплекте с JDK (в $JAVA_HOME/bin/jconsole), можно использовать для подключения к приложению во время его работы. Это очень хорошо, чтобы рассказать вам несколько вещей:

1. Приложение тратит все свое время на сборку мусора?
2. Что делают потоки приложения?

Не могли бы вы опубликовать результаты здесь?

Answer 3

похоже на проблему с памятью. Я бы получил дамп на использование памяти, посмотреть, как она себя ведет. Если времена gc слишком велики, у вас есть виновник. Тогда вы можете просто увеличить объем памяти, доступной для JVM, чтобы он снова заработал.

В любом случае, не конвертируйте batches в Список. Это ненужно. Это было бы необходимо, если бы вы использовали for / yield (в Scala 2.7), но, поскольку вы ничего не приносите, тогда Range - лучший выбор.