8

I've a ConcurrentLinkedQueue that's accessed by multiple threads; the objects in it are immutable. In one thread, I need a snapshot of the data, which I'm doing by calling stream on it. Is it safe? I'm aware of the non-interference requirement, but it seems to be talking about modification from one of the stream operations ("stream pipelines whose source might not be concurrent should never modify the stream's data source"), not necessarily externally. Besides, the ConcurrentLinkedQueue is designed for concurrent access, so there's that.

Abhijit Sarkar
  • 21,927
  • 20
  • 110
  • 219
  • 1
    that "snapshot" is going to be whatever state the terminal operation will observe. So your terminal operation might "see" that this is the last element and take it via `peek` for example; but immediately after that (let's say before the terminal operation finishes) , someone might add more elements. That is the what you would expect from a weakly consistent iterator... – Eugene Jun 08 '18 at 10:11

1 Answers1

10

From the documentation in link you have provided

For most data sources, preventing interference means ensuring that the data source is not modified at all during the execution of the stream pipeline. The notable exception to this are streams whose sources are concurrent collections, which are specifically designed to handle concurrent modification. Concurrent stream sources are those whose Spliterator reports the CONCURRENT characteristic

From documentation of the SplitIterator (its CONCURRENT characteristic)

static final int CONCURRENT

Characteristic value signifying that the element source may be safely concurrently modified (allowing additions, replacements, and/or removals) by multiple threads without external synchronization. If so, the Spliterator is expected to have a documented policy concerning the impact of modifications during traversal.

This is implementation from Collection interface stream method (which is not overridden in ConcurrentLinkedQueue)

default Stream<E> stream() {
        return StreamSupport.stream(spliterator(), false);
}

So as long as ConcurrentLinkedQueue uses CONCURRENT SplitIterator (which it does) I would assume you can safely iterate your ConcurrentLinkedQueue using stream().

Michal
  • 970
  • 7
  • 11
  • I find it interesting that even through concurrent, the stream is not parallel. Also, are the external changes to the collection visible to the stream? – Abhijit Sarkar Jun 08 '18 at 08:36
  • 2
    @AbhijitSarkar the stream will be parallel if you use `parallelStream()`. Or `stream().parallel()`. That’s the mode of the stream pipeline, which has nothing to do with the source. The visibility of external changes to the stream is the same as specified for the iterators, i.e. [*weakly consistent*](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/package-summary.html#Weakly) for most concurrent collections. – Holger Jun 08 '18 at 08:42
  • 1
    … since the `stream()` implementation delegates to `spliterator()`, you have to look at [the specific `spliterator()` implementation](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/ConcurrentLinkedQueue.html#spliterator--) – Holger Jun 08 '18 at 08:47
  • @Holger I don't understand your comment regarding parallel. The 2nd argument to `StreamSupport.stream` is `parallel`, which is `false` above. – Abhijit Sarkar Jun 08 '18 at 10:04
  • 2
    @AbhijitSarkar that’s the implementation of `stream()`. If you want a parallel stream, use `parallelStream()`. The implementation of `parallelStream()` looks exactly the same, except for using `true` instead of `false`. Besides that, you can change the stream’s mode by calling `parallel()` or `sequential()` on the stream. The type of the collection, i.e. whether concurrent or not, does not matter for the stream’s execution mode. – Holger Jun 08 '18 at 10:07