Как ограничить время работы потока?

Question

[R I]

Создаю сервер на Java, каждому клиенту - свой поток, для создания и управления потоками использую класс ExecutorService.
Но беда в том, что некоторые запросы от клиента могут занять слишком много времени, а то и вовсе зависнуть в ожидании ответа от сторонних серверов (такое редко, но случается). Чтобы избавится от этой проблемы я решил по таймауту удалять поток, чтобы не забивать пул потоков, но как это реализовать?

Answer 1

[FanKiLL]

В ExecutorService треды переиспользуются в этом и вся суть этих пулов.
Каждую задача что вы кидаете например Runnable попадает в очередь и потом выполняется ексекютором когда какой нибудь тред освободится.

Если посмотреть на названия тредов в екзекуторе - они будут примерно такими pool-2-thread-4
Достучатся до треда в котором выполняется ваш код, вы можете прям из Runnable или Callable смотря что вы там скармливаете executor'y

Thread.currentThread().getName() //например получить имя треда в котором выполняется Runnable

Если хотите прервать тред по каким то причинам, то примерно так.
Thread.currentThread().interrupt();

Comments

[bimeg]

Но в случае с InputStream это не освободит поток. Чтобы поток освободился, чтото должно заметить, что его интерраптнули.

[R I]

Ну это я знал, а как по времени ограничить его работу (потока)?

[FanKiLL]

@bimeg я не совсем понимаю откуда тут inputStrem, то есть я понимаю что вы имеете ввиду - но человек же не будет писать доступ к ресурсу с нуля и прям в каждом runnable. Наверно какая то библиотека будет использоваться для доступа по http.

[bimeg]

Судя по всему там чистые сокеты.

[FanKiLL]

@IzeBerg У вас же код в методе run выполняется последовательно, засекайте время ответа от сервера - если оно больше того что вам нужно - останавливайте тред.

Или можете смотреть сколько вообще тред занимает cpu времени

long nanos = ManagementFactory.getThreadMXBean().getThreadCpuTime(Thread.currentThread().getId());

и так же останавливайте.

Или сделайте задачу, которая будет проверять ваши треды раз в N времени на затраченное время как я описал выше getThreadCpuTime и прибивать их.

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

[FanKiLL]

@IzeBerg Если вам нужно именно по таймауту всё делать, то используйте Callable и Future как вам предложили выше.

Там есть два метода

V get() throws . . .;
V get(long timeout, TimeUnit unit)

Вызов первого метода get() устанавливает блокировку до тех пор, пока не завершится вычисление. Второй метод генерирует исключение TimeoutException, если истекает таймаут до завершения вычислений. Если прерывается поток, выполняющий вычисление, оба метода генерируют исключение InterruptedException. Если вычисление уже завершено, get() немедленно возвращает управление.

[FanKiLL]

@IzeBerg То бишь, каждый вызов стороннего ресурса выносите в отдельный Callable и в своём Runnable где уже вы их вызываете, пробуйте получить ответ вызвав get.(1, TimeUnit.SECONDS);

[R I]

Мне нужно именно сам поток обработки запроса останавливать. Т.е. тот, который получается при подключении клиента и отправки его обработки в ExecutorService. Если он (не зависимо какие идут действия занимает слижком много времени - то просто забить на клиента и освободить поток)

[FanKiLL]

@bimeg Извиняюсь, вы имели ввиду InputStream подключённого по соккету клиента... Я думал InputStream от обращения к стороннему сервису.

[FanKiLL]

@IzeBerg я не думаю что вы найдёте такой Executor где у тредов есть TTL ( Time To Live ) по истечению которого он убьёт тред - ибо это влечёт за собой кучу ошибок и не освобожденные ресурсов.

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

Как узнать сколько времени работает треад - я написал выше.
Если этот метод вам не подходит:
Можно банально записывать в Runnable в переменную long stratTime = время когда началось выполнение и потом сравнивать с вашим значением.
Или записывать время когда клиент подключился - тут смотря какая у вас логика.

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

Answer 2

[bimeg]

Использовать таймаут соединения - если долго то забивать на клиента. Или NIO - там можно обрабатывать соединение без блокировок (и как бонус можно заводить один поток на несколько клиентов).

Comments

[R I]

Таймаут соединения не катит, поток-то не очистится. Мне нужно именно под каждого клиента свой поток и каждому потоку таймаут выполнения, после которого поток останавливается и очищается (будет супер, если еще и соединение с клиентом закроется).

[bimeg]

Если у вас Executor то он сам следит за потоками. Зачем их убивать? Пусть себе крутятся, ведь Executor умеет их переиспользовать. Вам важно прервать выполнение вашей задачи которую вы засабмитили в Executor. Вот об том я и написал.

[R I]

Т.е. просто поставить тайм-аут на сокет (который после Listener.accept();)? Или как?

[bimeg]

Не вполне знаю что за Listener, но обычно так:
Socket.setSoTimeout(milliseconds);

Тогда InputStream будет кидаться IOException'ами если клиент не отвечает дольше чем таймаут.

[bimeg]

И OutputStream тоже, да.

[R I]

ServerSocket Listener = new ServerSocket(CONNECTION_PORT);

Сейчас потестил, поставил на сокет (который клиента) таймаут в 1 мс (Socket.setSoTimeout(1);) - никаких ошибок, хотя среднее время на выполнение тестовых запросов (все одинаковые) ~305 мс..
Что не так?

[bimeg]

Хым. Либо я гдето ошибся либо сам сокет у вас не тупит, а тупит чтото другое. Проверю свои слова как до дому доберусь.

[bimeg]

В следующем примере получил
java.net.SocketTimeoutException: Read timed out
и поток завершился.

public static void main(String[] args)
	{
		new Thread(new Runnable()
		{

			@Override
			public void run()
			{
				try (ServerSocket socket = new ServerSocket(1234))
				{
					Socket accept = socket.accept();

					accept.setSoTimeout(100);

					int read = accept.getInputStream().read();

					System.out.println(read);
				}
				catch (IOException e)
				{
					e.printStackTrace();
				}
			}
		}).start();

		new Thread(new Runnable()
		{

			@Override
			public void run()
			{
				try (Socket socket = new Socket("localhost", 1234))
				{
					Thread.sleep(1000);
				}
				catch (UnknownHostException | InterruptedException e)
				{
					e.printStackTrace();
				}
				catch (IOException e)
				{
					e.printStackTrace();
				}
			}
		}).start();
	}

[R I]

Эм, вы пытаетесь достучатся до сокета-сервера, у которого тайм-аут на прием соединений - зачем? Мне нужно, чтобы сервер, получив клиента создал ему поток, а этот поток по истечению N-ого кол-ва миллисекунд/секунд завершил свое выполнение по таймауту. У вас-же таймаут тоит не там, сервер после таймаута просто не может принять клиентов.

[bimeg]

1) Тайм аут не на прием соединения а на ИО с клиентского сокета
2) Поток не услышав ответа от клиента в тетчение 100мс закончился с исключением.
3) Видимо я не понял что вы хотите на самом деле.

Answer 3

[bromzh]

Можно запускать эти задачи асинхронно. В loop'е или экзекуторе должна быть функция запуска задачи с таймером.

Comments

[R I]

А пример можно?

[bromzh]

stackoverflow.com/questions/2758612/executorservic...
Вот вроде то. Сам я в яве с асинхронностью не работал, поэтому конкретных классов не скажу. Просто общий смысл такой, что нужно создать Future, и запускать его нужном месте. В питоне, например, это IOLoop: бесконечный цикл, в котором можно исполнять задачи не блокируя IO. И в этом IOLoop есть внутренний таймер, по которому можно ограничивать Future.
В яве асинхронность сделана на потоках, так что тут будут детали различаться, но смысл такой же.

[bimeg]

Не. В джаве так не выйдет. И Future тут ваще никоим образом не поможет. Для похожего нужно юзать NIO.

[bromzh]

@bimeg Почему не подходят? Future вроде бы как раз и позволяет запускать асинхронно куски кода.
V get(long timeout, TimeUnit unit) позволяет ставить таймер на исполнение кода.

[bimeg]

Ну если так, то можно. Но это уже будет какойто велосипед, учитывая что давно есть нужные средства для non-blocking IO.

[bromzh]

Так вроде это вещи из разных классов. Future нужен, чтобы не ждать результата, т.е. он обеспечивает возможность вызывать функции асинхронно. А NIO - это неблокирующее IO. В nio-сокетах можно ведь запустить Future на выполнение? Если во future не использовать блокирующие операции, то всё будет асинхронно.
Или я ошибаюсь?

[bimeg]

Асинхронность дает не Future а Executor (thread pool). Future лишь позволяет другому потоку получить некие знания о состоянии задачи которая вполняется на worker thread'е.

Далее если я правильно понял вашу мысль, то есть Executor и некий поток который имеет массив Future'ов и в цикле обходит их и дергает get с таймаутом. И если какойто затупил то прибивает его. В контексте вопроса это не вполне хорошее решенее, помоему. Ибо get не дает узнать изза чего произошел затуп (клиент долгий или какието вычилсения) т.е. есть шанс прибить поток безосновательно.

Nio же позволяет одному потоку держать несколько коннешконов и безблокировки (не в смысле в отдельном потоке, а в смысле "если нечего читать еще то зачем ждать") в цикле их обходить и он же узнает, что один их них затупил по таймауту и тогда его можно убрать из очереди. Тогда не нужен лишний поток для обслуживания Future'ов. Да и сами Future'ы тоже не нужны вероятно.