This stream is already in use ошибка

Корректное отключение

Для корректного завершения сетевого подключения обе стороны должны послать пакеты с сигналом о завершении (FIN), которые указывают что стороны не будут больше отсылать данные, также каждая сторона должна подтвердить (ACK) получение сигнала о завершении сетевого обмена данными. FIN инициируется когда приложение вызывает метод close(), shutdown() или exit(). После завершения работы метода close() ядро переходит в режим ожидания подтверждения от второй стороны приема сигнала о завершении. Это делает возможной ситуацию когда процесс инициировавший отключение будет завершен прежде чем ядро освободит рессурсы связанные с подключением, и снова разрешит использовать порт для связывания с другим процесоом (в этом случае, при попытке использования порта мы получим исключение AddressAlreadyInUse).

На изображении:

1. Имеется установленное соединение, состояние ESTABLISHED
2. Клиент инициирует окончание подключения, посылает серверу сигнал о завершении подключения (FIN), переходит в состояние ожидания ответа сервера (FIN_WAIT_1)
3. Сервер получает сигнал о завершении подключения и отправляет подтверждение (ACK), переходит в состояние ожидания завершения подключения (CLOSE_WAIT) (вызывает close())
4. Сервер отсылает клиенту сигнал о том что успешно закрыл подключение (FIN) и пробует прочитать подтверждение клиента (ACK), после чего не дожидаясь его отключается.
5. Теперь клиенту может придти два сигнала в разной очередности

   ACK — клиент получил подтверждение о том что сервер понял его намерение закрыть подключение

   • Клиент переходит в состояние ожидания сигнала об окончании закрытия подключения (FIN) от сервера (FIN_WAIT_2)
   • Клиент получает сигнал о закрытии подключения сервером (FIN), отправляет подтверждение (ACK), некоторое время ждет(TIME_WAIT) и отключается (ядро освобождает рессурсы) (CLOSED)

   FIN — клиент получает сигнал о закрытии подключения на стороне сервера(FIN), раньше чем подтверждение от сервера (ACK), о получении инициирующего сигнала о закрытии от клиента (FIN)

   1. Клиент отправляет подтверждение приема сигнала о том что сервер закрывает соединение, и переходит в состояние отключения (CLOSING)
   2. После отключения пробует считать сигнал подтверждения от сервера (который был отправлен сервером сразу после получения от клиента сигнала о завершении работы, пункт 2), некоторое время ожидает(TIME_WAIT) и ядро освобождает рессурсы (CLOSING).

На рисунке показаны все воможные состояния, которые могут быть во время корректного завершения, в зависимости от порядка получения пакетов FIN и ACK от удаленной стороны. Обратите внимание, если вы инициировали завершение подключения (левая половина рисунка), то другая сторона не будет ожидать подтверждения получения вами пакета FIN (правая половина рисунка). Сотояние TIME_WAIT требуется на случай если подтверждение (ACK) которое вы отправили не было получено на другой стороне, или на случай появления ложных пакетов по какой-то причине. Я не знаю почему на стороне сервера не сделали состояние TIME_WAIT, хотя если клиент инициирует закрытие, это безусловно и не должно требовать ожидания. Состояние TIME_WAIT может удерживать порт в течение нескольких минут после завершения процесса. Время удержания варьируется в зависимости от операционной системы, в некоторых операционных системах оно является динамическим, стандартные значения лежат в диапазоне от 1 до 4 минут.

Если обе стороны успеют инициировать сигнал завершения, раньше чем получат его от другой стороны, то обе стороны будут вынуждены пройти через ожидание (TIME_WAIT).

Корректное отключение слушающей стороны

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

Обратите внимание, на стороне слушающего сокета невозможно гарантировать чистое закрытие. Пока вы проверяете использование соединения методом select() до закрытия, существует крошечная, но реальная возможность появления входящего подключения после вызова select() и до вызова close().

Непредвиденное отключение удаленной стороны

При внезапном отключении сервера, локальная сторона инициирует закрытие соединения, и в этом случае TIME_WAIT неизбежен. Если удаленная сторона исчезает из-за сбоя сети или перезагрузки машины (редкие случаи), локальный порт будет оставаться привязанным вплоть до истечения таймаута состояния TIME_WAIT. Хуже того, некоторые старые операционные системы, не реализуют таймаут для состояния FIN_WAIT_2, и могут оставаться в нем бесконечно долго, в этом случае спасти может только перезагрузка системы.

Если же локальное приложение (клиент) падает в процессе активного соединения, порт будет занят пока не завершится состояние TIME_WAIT, то же верно для приложений, закрытых в процессе подключения к удаленной стороне(pending).

Способы избежания проблем

Опция SO_REUSEADDR

Можно использовать метод setsockopt(), для установки опции SO_REUSEADDR, что позволит создавать привязку к порту даже если он еще находится в состоянии TIME_WAIT (привязка к порту будет разрешена только для одного процесса). Это самый простой и эффективный метод избежать сообщения «address already in use».

Но, как ни странно, использование опции SO_REUSEADDR, может привести к более трудноотлавливаемым ошибкам чем «address already in use». SO_REUSEADDR позволяет использовать порт застрявший в TIME_WAIT, но вы все еще сможете использовать этот порт в том процессе, в котором он привязан изначально.

WHAT?

Предположим я использую локальный порт 1010 и подключаюсь на порт 300 сервера foobar.com, затем клиент отключается и порт переходит в состояние TIME_WAIT, и я могу использовать этот порт (1010) в любом подключении за исключением подключения к foobar.com на порт 300.
Ситуация в которой это может вызвать проблему может быть такой: моя программа пытается найти зарезервированный локальный порт(<1024) для привязки, чтобы подключиться к службе которая требует зарезервированный порт, и если я буду использовать опцию SO_REUSEADDR, то при каждом запуске программы на моей машине я буду получать тот же зарезервированный порт, даже если он висит в TIME_WAIT, и могу получить «Address already in use», в том месте где порт был использован в последний раз. В этом случае нужно отказаться от использования опции SO_REUSEADDR.

Некоторые не любят использовать SO_REUSEADDR, т.к. эта опция имеет проблемы с безопасностью. В некоторых операционных системах эта опция может позволить разным процессам использовать один и тот же порт одновременно. И это проблема, потому что большинство серверов привязываются к порту не используя конкретный адрес, вместо этого они используют INADDR_ANY (команда netstat отобразит их как *.8080). Таким образом, если сервер связывается с адресом *.8080, то другой процесс, от другого пользователя локальной машины, может подключиться к адресу local_machine.8080 (и намерения его могут быть совсем не хорошими), и перехватывать все ваши подключения, т.к. он указал более конкретный адрес. Эта проблема проявляется только на многопользовательских системах, не имеющих ограничений для учетных записей, и это не является уязвимостью которая доступна снаружи локальной машины, ее можно легко избежать используя привязку к конкретному адресу машины (не используя INADDR_ANY).

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

Клиент отключается первым

Глядя на рисунок выше, мы видим, что состояния TIME_WAIT можно избежать когда закрытие инициируется на удаленной стороне, а значит проблем можно избежать, если сервер позволяет клиенту инициировать отключение первым. Для этого можно построить архитектуру пользовательского протокола таким образом, что клиент знает когда ему нужно инициировать закрытие. Сервер может произвести безопасное отключение получив команду EOF от клиента, однако нам все равно придется установить таймаут ожидания отключения клиента, чтобы тот смог корректно завершить работу. Почти всегда достаточно подождать несколько секунд, пока соединение с сервером не будет корректно завершено.

Эту концепцию, вероятно, имеет смысл назвать «удаленная сторона отключается первой», иначе мы будем зависить от того что мы называем клиентом и что сервером. Если вы разрабатываете некую систему, состоящую из нескольких клиентских программ, которые находятся на одной машине и обращаются к разным серверам, то вы захотите перенести ответственность за отключение на сервера, для сберегания рессурсов клиентской машины.

К примеру, я написал скрипт который использует remote shell(rsh), для общения со всеми машинами моей сети, и он выполняет работу параллельно, постоянно используя несколько открытых соединения. Для rsh доступно меньше 1024 портов. Сначала я использовал команду «rsh -n», которая вызывает отключение локальной стороны в первую очередь. После нескольких тестов все свободные порты меньше 1024, оказались в состоянии TIME_WAIT, и процесс остановился. Удаление опции -n приводит к инициированию отключения на удаленной стороне, и проблема TIME_WAIT устраняется, однако это может привести rsh к зависанию в ожидании входаящего подключения. И если вы закрываете входящее подключение локально, порт снова окажется в состояние TIME_WAIT. В конечном счете я просто отказался от использования rsh и написал свою реализацию на perl (текущую версию можно скачать тут)

Уменьшение таймаута

Если, по какой-то причине, ни один из изложенных вариантов вам не подходит, есть возможность сократить таймаут состояния TIME_WAIT. Возможность и реализация такой операции зависит от операционной системы которую вы используете. Стоить помнить, что слишком короткий таймаут может иметь негативные последствия, в частности при потере пакетов или в перегруженных сетях.

Java 8 Stream API is used to process data collections such as arrays, lists, and maps. If we often work with streams, we may have encountered the following error: java.lang.IllegalStateException: stream has already been operated upon or closed.

Let us examine the root cause of the problem and how to solve it.

1. Reason

The Java Streams process the collection items with 2 types of operations:

• Intermediate operations: that help in method chaining and return a new Stream of processed items such as filter, sort, or map.
• Aggregate or terminal operations: terminate the stream by performing common operations or reduction on them. After calling the terminal operation, the Stream is considered closed. For example, max, min or collect items into a list or set.

We can use the terminal operation on a Stream only once. Calling a terminal operation in a already closed Stream causes the IllegalStateException.

Let us understand with an example. In the following example, we are using the same Stream twice. First for getting the even numbers, and then for getting the odd numbers.

Stream<Integer> numberStream = Stream.of(123, 234, 11, 57, 60, -4);

List<Integer> evenNumbers = numberStream
	.filter(integer -> integer % 2 == 0)
	.collect(Collectors.toList());

List<Integer> oddNumbers = numberStream
	.filter(integer -> integer % 2 == 1)    
	.collect(Collectors.toList());

The program output.

Even numbers:
234
60
-4
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
	at java.base/java.util.stream.AbstractPipeline.<init>(AbstractPipeline.java:203)
	at java.base/java.util.stream.ReferencePipeline.<init>(ReferencePipeline.java:94)
	at java.base/java.util.stream.ReferencePipeline$StatelessOp.<init>(ReferencePipeline.java:696)
	at java.base/java.util.stream.ReferencePipeline$2.<init>(ReferencePipeline.java:165)
	at java.base/java.util.stream.ReferencePipeline.filter(ReferencePipeline.java:164)
	at org.example.Main.main(Main.java:18)

Process finished with exit code 1

We can see that the first filtering operation on the stream was executed successfully, but the second operation returned an Exception. This is caused by the fact that the stream is consumed and closed by the first operation, more precisely by the collect() function.

When we started processing the Stream a second time, we get the exception with the message “stream has already been operated upon or closed”.

2. Solution

One solution for this problem would be creating a new stream every time before every stream processing, but there exists a more elegant method to handle this.

The Supplier interface is a functional interface that can be assigned using a lambda function. After initializing the supplier, we can use the get() function. The Supplier.get() returns a newly created Stream Object each time on which we can perform stream operations safely.

Supplier<Stream<Integer>> streamSupplier = () -> Stream.of(123, 234, 11, 57, 60, -4);

List<Integer> evenNumbers = streamSupplier.get()
        .filter(integer -> integer % 2 == 0)
        .collect(Collectors.toList());

List<Integer> oddNumbers = streamSupplier.get()
        .filter(integer -> integer % 2 == 1)
        .collect(Collectors.toList());

After running the above code, we will not have any errors and the stream will be processed both times successfully.

3. Conclusion

In this short tutorial, we learned the root cause and solution of the Stream exception “IllegalStateException: stream has already been operated upon or closed“. We learned to fix the error using the Supplier as well as creating a new Stream every time.

Happy Learning !!

#1

#2

4atty

#3

ranis17

#4

MonsterParamon

#5

☆★☆БАТЯ☆★☆™

#6

MonsterParamon

#7

☆★☆БАТЯ☆★☆™

#8

MonsterParamon

#9

ranis17

#10

☆★☆БАТЯ☆★☆™

#11

MonsterParamon

#12

ranis17

#13

MonsterParamon

#14

Kakoin

#15

MonsterParamon

#16

Kakoin

#17

WINS

#18

☆★☆БАТЯ☆★☆™

#19

WINS

#20

wait-angel

Java Stream reuse – traverse stream multiple times?

Exception

java.lang.IllegalStateException: stream has already been operated upon or closed

2. Look At Problem

2.1 From Java 8 API:

A stream should be operated on (invoking an intermediate or terminal stream operation) only once.

2.2 Explanation:

Taking a deeper look at what happens after stream creation.

A) Stream has intermediate operations, short-circuiting terminal operation and terminal operations.
B) Once a Stream is created, A pipe lines associated to this stream will be created.
C) On the stream, We can perform operations like filter(), map(), findAny(), findFirst(), count(), sum() methods on it.
D) Once we perform any operation on the stream, then that operation is placed into the current stream.
E) All operations will be executed and closes stream.

filter(), map() methods are intermediate operations.
findAny(), findFirst() are short-circuiting terminal operation
count() and sum() are terminal operations.

2.3 Example 1:

Next, think if we create a substreams from main stream as below.

String conent = "line 1 n line 2 \n line 3 \n line 4 \n line 5";

Here a stream is created with pipeline.

Stream mainStream = Stream.of(conent.split("n"));

Now, calling filter method on mainStream. Now filter operation is placed in the current pipeline.

mainStream.filter(line -> line.indexOf("line") > -1);

Now again calling filter method on mainStream. That will try to put the filter operation in the pipe line. But the pipe line is currently in operated mode with above filter. So, it it not possible to create another pipe line on the same stream.

mainStream.filter(line -> line.indexOf("1") > -1);

So it will throw now Runtime exception.

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
 at java.base/java.util.stream.AbstractPipeline.(AbstractPipeline.java:203)
 at java.base/java.util.stream.ReferencePipeline.(ReferencePipeline.java:94)
 at java.base/java.util.stream.ReferencePipeline$StatelessOp.(ReferencePipeline.java:696)
 at java.base/java.util.stream.ReferencePipeline$2.(ReferencePipeline.java:165)
 at java.base/java.util.stream.ReferencePipeline.filter(ReferencePipeline.java:164)
 at com.java.w3schools.blog.exceptions.streams.StreamIntermediateException.main(StreamIntermediateException.java:17)

2.4 Example 2:

Another example program to produce the same exception.

// Creating Stream from String.
String conent = "line 1 n line 2 \n line 3 \n line 4 \n line 5";
Stream mainStream = Stream.of(conent.split("n"));

// Operation on mainStream 
Optional firstValue = mainStream.findFirst();
System.out.println("firstValue : " + firstValue.get());

// Operation on mainStream

Optional anyValue = mainStream.findAny();
System.out.println("anyValue : " + anyValue.get());

See in this example, calling two methods on mainStream.

First when calling findFirst() then it returns value. But next invoking findAny(), at this time stream or pipe line is already closed. So it can not perform operation on the same stream.

Output:

firstValue : line 1 
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
 at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)
 at java.base/java.util.stream.ReferencePipeline.findAny(ReferencePipeline.java:548)
 at com.java.w3schools.blog.exceptions.streams.StreamCloseException.main(StreamCloseException.java:25)

Here firstValue is printed but while getting value from findAny() operation it has failed to get the stream because stream is already closed.

Note:

A Stream should be operated on (invoking an intermediate or terminal stream operation) only once. A Stream implementation may throw IllegalStateException if it detects that the Stream is being reused.

3. Solutions

We have many scenarios where the same stream is required to perform many operations. See the above example 2 where two values are needed.

To solve this issue, We need to create two-stream instances for these two separate operations.

Note:

Instead of creating two stream objects separately, We can use built in API Functional Interface Supplier<T> which is super handy in usage.

Supplier> supplier = () -> Stream.of(conent.split("n"));

// findFirst Operation
Optional firstValue = supplier.get().findFirst();
System.out.println("firstValue : " + firstValue.get());

// findAny Operation
Optional anyValue =supplier.get().findAny();
System.out.println("anyValue : " + anyValue.get());

In this program, created a Supplier object with type of Stream<String>. Supplier has a abstract method that is get() which returns Stream<String>. Now, We have to use Lambda Expression for Supplier Functional Interface with () zero arguments.

So, Whenever we call supplier.get() method than a new fresh Stream<String> object will be created and can perform any stream operation on it.

Output:

Output is returned properly without any errors.

firstValue : line 1 
anyValue : line 5

4. Conclusion

In this article, We’ve seen how to fix «java.lang.IllegalStateException: stream has already been operated upon or closed» problem when working with Java 8 Stream objects.

And also seen how to use terminal operations on the same stream using Supplier interface which avoids famous exception «stream has already been operated upon or closed».

All the examples shown in this article are available over GitHub with complete examples.