Skip to content

Sequenced Collections

Jump to bottom
Sebastian Fischer edited this page Dec 19, 2023 · 10 revisions

Sequenced Collections provide new interfaces for the collections framework which unify and extend the functionality of existing collections with a defined encounter order of elements. The new interface SequencedCollection includes methods to access (get, add, and remove) elements at both ends of a sequenced collection. The interface SequencedMap defines similar operations for maps with a defined encounter order of entries.

Affected Collection Types

Various existing collection types now implement the new interfaces in JDK 21. They can be grouped into different categories based on characteristics of the underlying encounter order.

Some collections are ordered based on an element-based ordering defined by a Comparator or the natural ordering provided by Comparable implementations. In other collections, the encounter order is based on the so-called insertion order that is determined by the order in which methods to add and remove elements have been called.

Some collections can contain equal elements multiple times. Other collections cannot contain such duplicates. Maps are similar to the latter kind of collections because they cannot contain multiple entries with the same key.

The following table groups implementations or extensions of the new interfaces based on the mentioned properties. We do not show implementations or extensions of interfaces listed in this table.

insertion order element-based order
with duplicates List, Deque
without duplicates LinkedHashSet SortedSet
maps LinkedHashMap SortedMap

We can see that there are no sequenced collections supporting duplicates with an element-based encounter order. For collections without duplicates, however, there are implementations with both insertion order and element-based ordering.

The following methods in the SequencedCollection interface have been promoted from the existing Deque interface and provide access to both ends:

  • getFirst and getLast query elements
  • addFirst and addLast insert elements
  • removeFirst and removeLast delete elements

Corresponding methods for maps are called differently:

  • firstEntry and lastEntry query entries
  • putFirst and putLast insert entries
  • pollFirstEntry and pollLastEntry delete entries

The methods to insert elements (or entries in maps) are interesting for collections without duplicates or with an element-based encounter order. The following table summarizes their behavior.

insertion order element-based order
with duplicates addFirst/addLast always insert new element
without duplicates addFirst/addLast reposition existing element addFirst/addLast not supported
maps putFirst/putLast reposition existing entry putFirst/putLast not supported

The decision to unify collections with different underlying characteristics under a single interface likely involved weighing simplicity against specificity. Unsupported operations from the table above are implemented using UnsupportedOperationException.

Sequenced collections also define a reversed method that returns a view of the collection they are called on with reversed encounter order.

Using Sequenced Collections

As an example of using sequenced collections, we will extend the echo server presented in the section on Virtual Threads so it can print the most recently echoed messages to the terminal. Based on an existing implementation of SequencedCollection, we will define a container with a fixed capacity that only stores the most recently added elements.

The echo server is multi-threaded so we need to use a thread-safe sequenced collection to store the messages. We could define a synchronized wrapper (or use a predefined one) around any sequenced collection to make it thread safe. Alternatively, we can use concurrent collections. Unlike synchronized wrappers, concurrent collections are not governed by a single exclusion lock. As a consequence, they might be beneficial for maintaining the throughput of our server.

To help pick an appropriate concurrent sequenced collection, we group them using the same criteria as before.

insertion order element-based order
with duplicates ConcurrentLinkedDeque
without duplicates ConcurrentSkipListSet
maps ConcurrentSkipListMap

Ideally, we would like to use an insertion-ordered concurrent sequenced collection without duplicates (a hypothetical ConcurrentLinkedHashSet) to benefit from the repositioning of existing elements on insertion. However, such an implementation is not available in the standard library. All provided sequenced collections without duplicates are based on element-based ordering, but our use case requires insertion-based ordering. If we want to use a predefined concurrent collection to store the most recent messages sent to our echo server then ConcurrentLinkedDeque is the only option.

We can implement a container of a given number of most recently added elements as follows.

public record MostRecentlyAdded<T>(int capacity, SequencedCollection<T> elements) {
    public MostRecentlyAdded(int capacity) {
        this(capacity, new ConcurrentLinkedDeque<>());
    }
    public void add(T elem) {
        elements.addLast(elem);
        while (capacity < elements.size()) {
            elements.removeFirst();
        }
    }
}

Note that the calls to addLast, size, and removeFirst are not synchronized. As a consequence, the capacity of our container might deviate from the intended capacity if multiple threads call the add method concurrently. Moreover, the container can hold the same element multiple times, which might be relevant when the echo server receives the same message repeatedly.

We can extend the echo server as follows to collect the most recently received messages.

-public record Server(ServerSocket socket, ExecutorService executor) implements Closeable {
+public record Server(ServerSocket socket, ExecutorService executor, MostRecentlyAdded<String> messages) implements Closeable {
     public static void main(String[] args) throws IOException {
         try (Server server = new Server()) {
+            handleInput();
             server.start();
         }
     }
     private Server() throws IOException {
-        this(new ServerSocket(0), Executors.newVirtualThreadPerTaskExecutor());
+        this(new ServerSocket(0), Executors.newVirtualThreadPerTaskExecutor(), new MostRecentlyAdded<>(20));
     }
     public Server {
         System.out.println("listening on port %s".formatted(socket.getLocalPort()));
     }
     @Override
     public void close() throws IOException {
         socket.close();
         executor.shutdown();
     }
+    private void handleInput() {
+        executor.submit(() -> {
+            try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
+                 PrintWriter writer = new PrintWriter(System.out, true)) {
+                reader.lines().forEach(unused -> messages.elements().forEach(writer::println));
+            }
+            return null;
+        });
+    }
     private void start() throws IOException {
         while (!socket.isClosed()) serve(socket.accept());
     }
     private void serve(Socket client) {
         executor.submit(() -> {
             try (BufferedReader reader = new BufferedReader(new InputStreamReader(client.getInputStream()));
                  PrintWriter writer = new PrintWriter(client.getOutputStream(), true)) {
-                reader.lines().forEach(writer::println);
+                reader.lines().forEach(msg -> {
+                    messages.add(msg);
+                    writer.println(msg);
+                });
             } finally {
                 client.close();
             }
             return null;
         });
     }
 }

The new method handleInput starts a new virtual thread that reads lines from standard input and reacts to each line by printing the twenty most recent messages received by the server. Messages are collected before returning them to their sender in the serve method.

Task: Synchronized Wrapper for LinkedHashSet

Change the implementation of MostRecentlyAdded to use a LinkedHashSet instead of a ConcurrentLinkedDeque. LinkedHashSet is not thread-safe, so you should synchronize access to the stored elements in your implementation. Try using both implicit locks (via the synchronized keyword) as well as explicit locks (via an appropriate implementation of the Lock interface.) Which implementation would you expect to perform better when used in the echo server? Compare the throughput of the different variants using your implementation of the stress test from a previous task or the one provided on the solutions branch.

Solutions

Synchronized Wrapper for LinkedHashSet

Here is the same stress test used before when using an echo server that collects the most recent messages in a ConcurrentLinkedDeque.

me@java-dev:~/java21-demo$ time ./java.sh sebfisch.echo.Clients localhost 39485 10000 10
real	0m19.957s
user	0m22.845s
sys	0m12.338s

It is a bit slower than the server not collecting messages but not by much. We can use implicit locks to synchronize access to a LinkedHashSet as follows.

 public record MostRecentlyAdded<T>(int capacity, SequencedCollection<T> elements) {
     public MostRecentlyAdded(int capacity) {
-        this(capacity, new ConcurrentLinkedDeque<>());
+        this(capacity, new LinkedHashSet<>());
     }
-    public void add(T elem) {
+    public synchronized void add(T elem) {
         elements.addLast(elem);
         while (capacity < elements.size()) {
             elements.removeFirst();
         }
     }
+    public synchronized void forEach(Consumer<T> action) {
+        elements.forEach(action);
+    }
 }

We add a forEach method for synchronized read access and use it in the server instead of accessing the elements directly from a different thread.

     private void handleInput() {
         executor.submit(() -> {
             try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
                  PrintWriter writer = new PrintWriter(System.out, true)) {
-                reader.lines().forEach(unused -> messages.elements().forEach(writer::println));
+                reader.lines().forEach(unused -> messages.forEach(writer::println));
             }
             return null;
         });
     }

The same stress test used before takes around five seconds longer now.

me@java-dev:~/java21-demo$ time ./java.sh sebfisch.echo.Clients localhost 41763 10000 10
real	0m24.874s
user	0m22.949s
sys	0m10.934s

Here is an alternative implementation of MostRecentlyAdded using an explicit lock.

public record MostRecentlyAdded<T>(int capacity, SequencedCollection<T> elements, Lock lock) {
    public MostRecentlyAdded(int capacity) {
        this(capacity, new LinkedHashSet<>(), new ReentrantLock());
    }
    public SequencedCollection<T> add(T elem) {
        lock.lock();
        try {
            elements.addLast(elem);
            SequencedCollection<T> removed = new LinkedList<>();
            while (capacity < elements.size()) {
                removed.addLast(elements.removeFirst());
            }
            return removed;
        } finally {
            lock.unlock();
        }
    }
    public void forEach(Consumer<T> action) {
        lock.lock();
        try {
            elements.forEach(action);
        } finally {
            lock.unlock();
        }
    }
}

The test runs a bit faster now but not as fast as with the concurrent collection.

me@java-dev:~/java21-demo$ time ./java.sh sebfisch.echo.Clients localhost 37995 10000 10
real	0m21.721s
user	0m21.208s
sys	0m10.469s

Virtual threads blocked on an implicit lock are not unmounted from their carrier thread. This might explain why the variant using explicit locks has slightly higher throughput. However, for a definitive evaluation more rigorous benchmarks would be necessary.

Clone this wiki locally