Thread dump и анализ кейса

📸 Дампом потоков называют снимок состояния всех текущих потоков JVM. Он очень полезен для диагностики проблем с производительностью, блокировками, утечками ресурсов. Дамп представляет из себя текстовый файл, где для каждого из существующих потоков java-процесса вы можете найти его состояние (RUNNABLE, BLOCKED, WAITING, TIMED_WAITING, TERMINATED) и полный стек вызовов.

⬆️ Вспоминаем ситуацию, описанную в предыдущем посте:

Программа перестает писать info-логи в файл. В логах нет никаких ошибок. Вы проверяете ваш java-процесс с помощью ps -aux | grep java, - он жив, то есть программа не завершилась.

🔧 В этом случае для диагностики вы можете прибегнуть к thread dump. Примеры утилит, которые помогут вам сделать снимок потоков: jstack, jvisualvm, JMC (Java Mission Control), jcmd. Ниже пример команды для утилиты jstack, на вход необходимо передать pid java-процесса.

jstack <pid> > threaddump.txt

🧨 Ниже приведен фрагмент thread dump, представляющий интересующий нас поток-producer.

"Thread-0" #16 daemon prio=5 os_prio=31 cpu=0.97ms elapsed=3.79s tid=0x00007fd90d0ed800 nid=0x5f03 waiting on condition  [0x000070000a578000]
   java.lang.Thread.State: WAITING (parking)
  at jdk.internal.misc.Unsafe.park(java[email protected]/Native Method)
 . . . . . . . 
  at java.util.concurrent.LinkedBlockingQueue.put(java[email protected]/LinkedBlockingQueue.java:343)
  at ru.quipy.Producer.supply(Example.kt:30)
  at ru.quipy.ExampleKt.main$lambda-4(Example.kt:48)
  at ru.quipy.ExampleKt$$Lambda$32/0x0000000132014658.run(Unknown Source)
  at java.lang.Thread.run(java[email protected]/Thread.java:840)

🧐 Как мы видим, поток находится в состоянии перманентного (не ограниченного временем) ожидания (java.lang.Thread.State: WAITING) в методе LinkedBlockingQueue.put. Если вы находитесь на собеседовании, то можете сказать, что лучше использовать аналогичный метод LinkedBlockingQueue.offer, который не блокируется навечно, а использует ограниченную временем блокировку, что даст нам возможность выйти из ожидания, понять, что операция не удалась и сделать об этом запись в лог. Аналогично и для метода LinkedBlockingQueue.take: вообще все блокирущие методы стоит ограничивать временем, тогда не возникнет ситуация с блокировкой потока при завершении программы, которую вы описали в комментах к предыдущему посту.

🔍 Очень подозрительно, но мы не можем найти дамп потока consumer из чего может следовать вывод, что такового в JVM нет. Вероятно, данный поток уже завершился. Это объясняет, почему он больше не делает записи в логи и почему поток supplier не может поместить ничего в очередь: она достигла максимального размера из-за отсутствия потока-потребителя.

🕵🏽 Поток завершился в следствие появления неожиданного исключения. Это могло произойти в следующей строке: log.info("Consumed $code. Mapped to ${codeMappings[code]}"), если элемента с индексом code в списке не существует.

🤯 А исключения мы могли не увидеть в логах по нескольким причинам. Например, вызывающий нас код мог поймать исключение, завершить поток и не записать об этом в лог. Или же был использован базовый обработчик исключения в потоке (interface UncaughtExceptionHandler), дефолтная версия которого (class ThreadGroup) просто делает запись в System.err, который мог не перенаправляться в лог.

🥇 Какие выводы мы можем сделать?

1. Следует оборачивать происходящее в потоке в глобальный обработчик исключений try-catch, логировать перехваченные исключения и поддерживать работу потока, чтобы приложение могло прогрессировать даже вопреки ошибочным ситуациям.

2. Следует явно ограничивать временем методы, которые блокируют поток исполнения или использовать их варианты/альтернативы с таймаутом!

#java #kotlin #threads #interview #case

Давайте потокам осмысленные имена

🚨 Это архиважная вещь в промышленной разработке на java. Вы могли заметить, что необходимый нам поток из поста выше 👆🏼носит название Thread-0 . Я без труда нашел его в thread dump только потому, что программа была мала. Но даже ее дамп включал 21 поток, подавляющее большинство которых было служебными потоками JVM (в основном потоками сборщика мусора).

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

🧵 Давайте начнем с одиночных потоков. Конечно, мы не часто создаем потоки через конструктор класса Thread, но полезно знать, что этот конструктор позволяет вам дать потоку имя.

val producerThread = Thread(null, { producer.supply() }, "producer-thread")

🏭Теперь давайте перейдем к пулам потоков. Статический метод Executors.newFixedThreadPool, через который удобно создавать пулы потоков, принимает на вход необходимый размер пула и еще один интересный параметр с типом ThreadFactory. У этого интерфейса всего один метод Thread newThread(Runnable task). Пул потоков делегирует переданному объекту ThreadFactory задачу создания потоков. Давайте реализуем свой вариант ThreadFactory, который именует потоки необходимым образом и передадим такой объект нашему пулу:

class NamedThreadFactory(private val prefix: String) : ThreadFactory {
    private val sequence = AtomicInteger(1)

    override fun newThread(r: Runnable): Thread {
        val thread = Thread(r)
        val seq = sequence.getAndIncrement()
        thread.name = prefix + (if (seq > 1) "-$seq" else "")
        return thread
    }
}

🚲 На вход классу NamedThreadFactory передаем строку, которая будет являться префиксом имени всех потоков, созданных этой фабрикой. С помощью атомарного каунтера sequence мы сможем создавать уникальную часть имени для наших потоков🥇🥈🥉. Теперь давайте передадим эту фабрику пулу:

val httpCallsExecutor = Executors.newFixedThreadPool(16, NamedThreadFactory(“http-calls-pool”))

🎯 Вуаля, теперь в thread dump ваши потоки будут иметь осмысленные имена и вам не придется тратить уйму времени на изучение стека ненужных потоков!

#java #kotlin #threads

Использование Thread.sleep для тестирования асинхронных программ

📲 Тестируем сервис отправки push-уведомлений. Внешние процессы вызвают его, чтобы отправить push-уведомление c заданным текстом на мобильное устройство. Сервис возвращает идентификатор уведомления, по которому можно проверить его статус. Само уведомление может быть отправлено позже, для этого наш сервис совершает http-вызов к внешнему провайдеру.

interface NotificationService {
    fun submitNotification(deviceId: UUID, text: String): UUID
}

Как будем тестировать?

👺 Если тестирование модульное, то, вероятно, мы хотим сделать mock для http-client-a, с помощью которого будет посылаться уведомление и проверить, что один из его методов был вызван. Аналогично можно сделать mock классов доступа к базе данных и проверить, что статус оповещения был обновлен. Однако, если написать такой код, то может оказаться, что тест в большом количестве случаев не проходит:

notificationService.submitNotification(deviceId, “Hello”)
verify(dbService, times(1)).updateStatus(any());

⏳ Между вызовом сервиса и проверкой прошло очень мало времени, запрос мог провести его в буферах-очередях, ждать выполнения http-вызова, или выполнения запроса в БД. Вероятность этого еще выше, если тест интеграционный - сразу после самбита уведомления мы пытаемся получить его статус из базы данных, но он не успевает измениться.

⏰ Чтобы сделать тест "зеленым", частенько выбирают некоторую “взятую с потолка” константу, например, 5 секунд и помещают инструкцию, которая заставляет поток "уснуть", между вызовом сервиса и проверкой:

notificationService.submitNotification(deviceId, “Hello”)
Thread.sleep(5000L) // 5000 milliseconds
verify(dbService, times(1)).updateStatus(any());

Как правило, это решает проблему прохождения теста, но генерирует несколько новых:

1. Пяти секунд может быть недостаточно в определенных обстоятельствах и тест будет падать, но не всегда, а, например, в 1% случаев. Не так много, чтобы переписать тест, но достаточно, чтобы с завидной регулярностью фейлить весь билд и раздражать 😤. То есть вы своими руками делаете свои тесты недетерминированными, flucky тестами.

2. Вторая причина часто более заметна для команды. Инструкция Thread.sleep(5000L) выставляет нижнюю границу выполнения теста равной пяти секундам, тогда как в удачном сценарии тест мог занять 10, 50, 100, 200 миллисекунд, секунду. Что, если у вас 200 тестов? Время выполнения 1000 секунд или около 16 минут, тогда как в удачном сценарии ваш билд мог занять во много раз меньше - 1, 2 … 4 минуты. Часто проблема нарастает очень плавно с увеличением количества тестов и неопытные команды не могут понять, в чем причина увеличения времени сборки 🤔.

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

#kotlin #java #async

Как “не спать” при тестировании асинхронных программ.

🛌 В предыдущем посте я описал мотивацию использования Thread.sleep при тестировании асинхронного кода и связанные с этим проблемы. Однако, пассивное ожидание потока очень помогает нам при тестировании асинхронного кода, ведь тесту нужно дождаться завершения асинхронной операции, чтобы начать проводить различные проверки.

⏳ Проблему для нас создает тот факт, что мы не знаем, сколько времени в действительности займет тестируемая операция (от выполнения к выполнению ее длительность может варьироваться). Поэтому приходится подбирать некую константу, которая в большинстве случаев превышает максимальную длительность операции. Пример: отправка оповещения в среднем занимает около 150ms, но в 3% случаев превышает 5 секунд. Thread.sleep(5_000) даст нам зеленый тест в c 97% вероятностью.

⏰ Нам бы хотелось, чтобы поток ожидал примерно столько же, сколько в действительности выполняется тестируемая операция. Если ее фактическая длительность 150ms, то поток должен находиться в ожидании (150ms + некоторая фиксированная константа, например, еще 50ms). Если операция выполнялась 5s, то и время ожидания должно быть 5s + 50ms.

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

🔔Пример: клиент отправляет уведомление. Мы хотим протестировать, что после отправки его статус в базе данных будет обновлен на processed. Будем с периодичностью в 50ms делать запрос в базу данных, получать статус и, если он еще не изменен на processed, то ждать следующие 50ms. Нам станет известно, что статус обновился, через кратчайший промежуток времени.

Давайте посмотрим на пример кода, где в качестве такого механизма используется библиотека Awaitility.

id = service.submitNotification(deviceId, “Hello”)

with()
    .pollInterval(50_MILLISECONDS)
    .await()
    .atMost(8, SECONDS)
    .until(notificationStatusIsUpdated());

⚙️ У библиотеки есть большое количество конфигурируемых параметров. Например:
1. Период времени между проверкой условия (poll interval)
2. Начальная задержка перед первой проверкой
3. Минимальное время ожидания (например, вы хотите, чтобы оповещение отправлялось не раньше, чем через 5 секунд)
4. Максимальное время ожидания, за которое ваше условие должно выполниться
5. Игнорируемые исключения
6. Пул потоков, на котором будет выполняться тестирование

🐈‍⬛ В библиотеке также присутствует модуль для Kotlin, который помогает использовать API библиотеки в конструкциях, похожих на естественный язык, используя инфиксные функции.

🏋️‍♀️ Вы можете для тренировки написать собственный небольшой инструмент для ожидания (на java, kotlin, kotlin-coroutines) и поделиться кодом в комментариях 😊

#java #kotlin #async