По какой причине java.util.concurrent.ExecutorService.awaitTermination() может не дожидаться завершения всех запущенных задач?

Question

[ElenaKovalinskaya]

У меня возникла проблема при использовании java.util.concurrent.ExecutorService.awaitTermination(). Код, который следует после awaitTermination(), не исполняется, и я не понимаю, почему это происходит.

ExecutorService pool = Executors.newFixedThreadPool(6);

        for (HierarchyItem rootItem : rootObjectItems) {
            pool.submit(() -> {
                final Thread currentThread = Thread.currentThread();
                currentThread.setName("Processing " + rootItem.getName());
                try {
                    addressObjectProcessor.handle(rootItem);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }

        pool.shutdown();
        try {
            boolean poolTermination = pool.awaitTermination(5, TimeUnit.HOURS);
            System.out.println(poolTermination);
            if (! poolTermination) {
                pool.shutdownNow();
                if (!pool.awaitTermination(5, TimeUnit.MINUTES))
                    System.err.println("Executor service did not terminate. " + Instant.now());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        /// код после pool.awaitTermination. Он уже не исполняется
         . . . .

По логам я вижу, что все задачи абсолютно точно завершаются (внутри addressObjectProcessor.handle метода в самом конце в логи пишется результат исполнения задачи), но код после awaitTermination не исполняется. Такая проблема возникает при количестве задач (размер rootObjectItems) = 90. Когда их 5-10-15-20, механизм отрабатывает хорошо. Все выглядит так, как будто основной поток засыпает или умирает уже до того момента, когда все задачи завершаются, и даже после таймаута ничего не происходит. Время обработки каждой задачи - от 3 до 60 минут, но в целом время обработки всех 90 задач в 6 потоках занимает не более 4 часов.

Результат исполнения awaitTermination в логи не выводится, и InterruptedException исключение не выбрасывается. Строка System.out.println(poolTermination); уже не исполняется.

Сервис, в котором используется этот механизм, запущен в wildfly 16.

Буду очень благодарна любой зацепке.

Comments

[mayton2019]

Я не понимаю твоих сомнений.

У тебя здесь - бизнес логика
addressObjectProcessor.handle(rootItem);
и главная задача и главный вопрос - отработала ли она? Все процессы или джобы завершились?
А техническая сторона - судьба java threads нам безразлична. Умирают ли они в течение 5 часов или за 1 секунду - это не главное.

Кстати почему ты не используешь Future/Callable. Было-бы логично запускать джобы с получением ФАКТА результата. А у тебя они запускаются в режиме запустил-и-забыл. Или запустил и тут-же скомандовал shutdown.

[ElenaKovalinskaya]

mayton2019, спасибо тебе за ответ!

У тебя здесь - бизнес логика
addressObjectProcessor.handle(rootItem);

- да, все верно. На данный момент можно считать, что мне безразлично, отработали ли все задачи, отправленные в ExecutorService . В контексте задачи, которую выполняет мой сервис, я могу запустить эти задачи в режиме -запустил-и-забыл, поэтому мне использование Future/Callable показалось немножко избыточным.

Или запустил и тут-же скомандовал shutdown

- Метод shutdown сами потоки не завершает, он сигнализирует о том, что прием заявок на исполнение закрывается. Проблема заключается в том, что pool.awaitTermination должен ожидать завершения всех задач, чтобы потом после этого выполнить код, следующий после него. Но что-то происходит, и основной поток, который должен дождаться выполнение всех задач, вдруг прекращает свое действие. Я сейчас пытаюсь с помощью jstack понять, что с ним происходит. Но мне все же интересно, какие могут быть причины этому.

[mayton2019]

ElenaKovalinskaya, замени

addressObjectProcessor.handle(rootItem);
на

Thead.sleep(...случайное время порядка минуты...)
и протестируй. Будет такое-же поведение?

[ElenaKovalinskaya]

mayton2019, пришлось сделать перерыв в решении этой проблемы, меня переключили на другую задачу :)
Я протестировала, подставив Thead.sleep вместо addressObjectProcessor.handle. В итоге, поведение такое же

[mayton2019]

Добавь сюда максимальное число логгирования. Буквально через каждую строчку добавь какой-то месседж. Уникальный.

pool.shutdown();
        try {
            boolean poolTermination = pool.awaitTermination(5, TimeUnit.HOURS);
            System.out.println(poolTermination);
            if (! poolTermination) {
                pool.shutdownNow();
                if (!pool.awaitTermination(5, TimeUnit.MINUTES))
                    System.err.println("Executor service did not terminate. " + Instant.now());
            }

И запускай снова

[ElenaKovalinskaya]

mayton2019, Спасибо тебе за ответ. Я заменила этот участок кода на метод:

private void shutdownAndAwaitTermination(final ExecutorService executorService, final Duration waitForTermination) {
        logger.info("Завершаю прием задач в пул с помощью shutdown");
        executorService.shutdown();
        try {
            logger.info("Начинаю ожидать завершение всех задач (awaitTermination)");
            boolean terminated = executorService.awaitTermination(waitForTermination.toHours(), TimeUnit.HOURS);
            logger.info("Ожидание завершено, все задачи должны быть завершены, результат завершения : " + terminated);
            if (!terminated) {
                logger.info("Все же все задачи не завершились. Пытаюсь принудительно их завершить с помощью shutdownNow");
                executorService.shutdownNow();
                logger.info("Ожидаю завершения задач после shutdownNow");
                if (!executorService.awaitTermination(5, TimeUnit.MINUTES))
                    logger.info("Executor service did not terminate. " + Instant.now());
            }
        } catch (InterruptedException ex) {
            logger.info("Выброшено исключение InterruptedException, снова пробую shutdownNow");
            executorService.shutdownNow();
            logger.info("Прерываю текущий поток");
            Thread.currentThread().interrupt();
        }

        logger.info("Метод shutdownAndAwaitTermination завершен");
    }

и, судя по логам, по какой-то непонятной причине даже executorService.shutdown(); не был выполнен. Ни одна строка логов из этого метода не была выведена.

[mayton2019]

ElenaKovalinskaya, нужно посмотреть в каком статусе стоят приладные джобы.

сделай как тут

jstack [-F] [-l] [-m] <pid>
На выходе будет длинный текстовый файл. В нем будет дерево работающих threads и где-то
будут их состояние. Так вот. Я думаю что они стоят в таком хитром состоянии что executorService.shutdown()
либо их не может остановить либо очень делает это крайне медленно.

Вот посмотри что за состояние.

[ElenaKovalinskaya]

mayton2019, смотри какая штука. У меня в общей сложности 90 задач должно выполниться. В ExecutorService задаю пул из 6 потоков. Сначала запускатеся 6 потоков сразу, потом когда один из этих шести завершается, вместо него запускается новый, и так далее, пока все задачи не будут выполнены. Каждому новому потоку я задаю имя.
Я сформировала файл дампа с помощью jstack, в нем я по именам вижу 6 потоков, которые выполнялись последними. У них состояние WAITING (parking). Там еще 83 потока, имена которых я не узнаю, у которых состояние WAITING (parking), RUNNABLE или WAITING (on object monitor)
Потоку, который запускает всю эту тучу потоков, я тоже дала имя. Но в файле дампа я его не вижу почему-то.

Меня очень удивляет то, что executorService.shutdown() так и не был выполнен. Это очень странно, поскольку запуск выполнения джобов должен запускаться только после executorService.shutdown(). Он не останавливает сами потоки. Он останавливает прием новых заявок и запускает выполнение тех, что уже были отправлены в executorService. Так должно быть, по идее. Но в моем случае все как-то по-другому работает.

[mayton2019]

Копируй исходники ThreadPoolExec в свой пакет. И делай логгирование здесь. Где-то он висит. И надо понять где.

public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

Возможно он слишком аккуратно работает а тебе надо рубануть рубильник.