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.base@17.0.9/Native Method)
 . . . . . . . 
  at java.util.concurrent.LinkedBlockingQueue.put(java.base@17.0.9/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.base@17.0.9/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

Траблшутинг кейс

💣 У вас есть простой сервис, который принимает входящие http-запросы, в среднем около 1к rps (requests per second). Он добавляет к запросу небольшое количество метаданных и пересылает по HTTP далее, в следующий сервис.

👀 Вы заметили, что сервис стал хуже справляться с пиковыми нагрузками: 6k rps. Симптомы следующие:

1. При увеличении входящего трафика начинают скапливаться внутренние очереди

2. Время выполнения http запроса в следующий сервис возрастает в разы или даже на порядки. (В норме среднее 30ms, при нагрузке начинает нарастать и растет вплоть до 20-30s)

3. Возрастает значение метрики количества используемых JVM файловых дескрипторов (рост примерно со 100 до 170).

4. Возрастают также и другие показатели: RAM, CPU.

🔎 Давайте исследовать:

1. Подозрение падает на следующий в цепочке сервис. Мы предполагаем, что под нагрузкой время его ответа увеличивается, и это отражается на нашем приложении.

🏓Команда сервиса опровергает эту догадку, предъявляя нам метрики, где видно: среднее время обработки запроса на их стороне оставалось стабильным в течение всего времени.

🧑🏻‍🔬 Мы проводим и свой (не самый репрезентативный) эксперимент - во время пиковой нагрузки, используя какой-нибудь http-client, например, curl, делаем несколько запросов к их сервису с нашей машины и видим, что время ответа около 30ms.

2. Рост файловых дескрипторов. Вспоминаем, что в Linux TCP коннекция является файлом, то есть с ростом числа подключений растет и количество захваченных дескрипторов.

📡 Кто может раздувать количество соединений? - Http-client. Зачем? Из-за особенности протокола HTTP1. Запросы в нем блокируют подключение: пока один запрос не будет выполнен, следующий запрос не отправляется.

3. Зная эту особенность HTTP, давайте считать пропускную способность:

📲 Длительность запроса 30ms, то есть одно подключение может за секунду обрабатывать 1000ms/30ms≈33.3 запроса. Чтобы удовлетворить обычную нагрузку на сервер в 1к, нужно около 1000rps/33.3req ≈ 30 подключений к следующему сервису. Мы знаем, что у нас было 100 дескрипторов, но не знаем, какие из них были подключениями.

4. Мы можем посчитать, сколько нужно подключений, чтобы сервис выдержал 6к запросов без накапливания очереди.

⚖️ Итак 6000 / 33.33 будет ≈ 180 подключений. У нас же есть около 170 (и мы не знаем, сколько из них действительно наши). В любом случае у нас будет накапливаться очередь внутри http-client, в которой запросы могут проводить долгое время, особенно в период повышенной нагрузки.

⚙️ Что можно сделать

1. Увеличить количество подключений per host. Почти у всех http-клиентов есть возможность ограничить количество подключений к одному хосту. Часто это совсем небольшое значение. Например, в OkHttp client или AsyncHttpClient оно по умолчанию равно 5. И все клиенты (почти) дают возможность выставлять кастомное значение этого параметра, например:

AsyncHttpClientConfig.Builder.setMaxConnectionsPerHost(maxConnectionsPerHost)

2. Перейти на протокол HTTP версии 2. Он лишен такого недостатка, как блокирующиеся коннекции. С помощью процесса мультиплексирования HTTP 2 дает возможность одновременно передавать и получать данные большого количества запросов, используя даже единственное соединение. Но переход на него в данном случае возможен только, если вы можете перевести на него и следующий в цепочке сервис, что не всегда в вашей власти.

Ставьте 🔥, если такой формат постов кажется вам интересным :)

#http #case #interview

⬆️ В прошлом посте мы с помощью закона Амдала выясняли, сколько нужно потоков, чтобы ускорить наш код в 10 раз. Полученный ответ, 20 потоков, занял второе место в нашем опросе, но все еще может казаться довольно удивительным.

🧑‍💻 Как и обещал, я провел небольшой эксперимент для проверки теоретических расчетов и постарался найти объяснение полученным на практике результатам.

Подробнее об этом, а также о выводах и практических рекомендациях читайте по ссылке