SingleThreadedEventLoop: add read-only iterator of channels (netty#12610)

Motivation:

For servers there is a need for per eventloop "watchdog" timer that ticks for all eventloop channels. For moderately high number of connections, It is cheaper cpu-wise than per channel scheduling.

Now this requires additional data structure to hold active channels - which is wasteful as It duplicates the one from SingleThreadedEventLoop (its practically useful impls - epoll, kqueue, nio).

Modification:

Add read-only iterator of eventloop channels exposed on SingleThreadedEventLoop as Iterator<Channel> registeredChannelsIterator()

Result:

SingleThreadedEventLoop channels may be iterated without additional redundant data structures


Co-authored-by: Norman Maurer <[email protected]>

Author: mostroverkhov

testsuite/src/main/java/io/netty/testsuite/transport/AbstractSingleThreadEventLoopTest.java

@@ -27,18 +27,27 @@
 import io.netty.channel.local.LocalServerChannel;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.Promise;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.function.Executable;
 
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 public abstract class AbstractSingleThreadEventLoopTest {
 
@@ -156,6 +165,134 @@ public void run() {
         assertRejection(loop);
     }
 
+    @Test
+    @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
+    public void testChannelsIteratorEmpty() throws Exception {
+        assumeTrue(supportsChannelIteration());
+        EventLoopGroup group = newEventLoopGroup();
+        final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
+        try {
+            runBlockingOn(loop, new Runnable() {
+                @Override
+                public void run() {
+                    final Iterator<Channel> iterator = loop.registeredChannelsIterator();
+
+                    assertFalse(iterator.hasNext());
+                    assertThrows(NoSuchElementException.class, new Executable() {
+                        @Override
+                        public void execute() {
+                            iterator.next();
+                        }
+                    });
+                }
+            });
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
+    @Test
+    @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
+    public void testChannelsIterator() throws Exception {
+        assumeTrue(supportsChannelIteration());
+        EventLoopGroup group = newEventLoopGroup();
+        final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
+        try {
+            final Channel ch1 = newChannel();
+            final Channel ch2 = newChannel();
+            loop.register(ch1).syncUninterruptibly();
+            loop.register(ch2).syncUninterruptibly();
+            assertEquals(2, registeredChannels(loop));
+
+            runBlockingOn(loop, new Runnable() {
+                @Override
+                public void run() {
+                    final Iterator<Channel> iterator = loop.registeredChannelsIterator();
+
+                    assertTrue(iterator.hasNext());
+                    Channel actualCh1 = iterator.next();
+                    assertNotNull(actualCh1);
+
+                    assertTrue(iterator.hasNext());
+                    Channel actualCh2 = iterator.next();
+                    assertNotNull(actualCh2);
+
+                    Set<Channel> expected = new HashSet<Channel>(4);
+                    expected.add(ch1);
+                    expected.add(ch2);
+                    expected.remove(actualCh1);
+                    expected.remove(actualCh2);
+                    assertTrue(expected.isEmpty());
+
+                    assertFalse(iterator.hasNext());
+                    assertThrows(NoSuchElementException.class, new Executable() {
+                        @Override
+                        public void execute() {
+                            iterator.next();
+                        }
+                    });
+                }
+            });
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
+    @Test
+    @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
+    public void testChannelsIteratorRemoveThrows() throws Exception {
+        assumeTrue(supportsChannelIteration());
+        EventLoopGroup group = newEventLoopGroup();
+        final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
+
+        try {
+            final Channel ch = newChannel();
+            loop.register(ch).syncUninterruptibly();
+            assertEquals(1, registeredChannels(loop));
+
+            runBlockingOn(loop, new Runnable() {
+                @Override
+                public void run() {
+                    assertThrows(UnsupportedOperationException.class, new Executable() {
+                        @Override
+                        public void execute() {
+                            loop.registeredChannelsIterator().remove();
+                        }
+                    });
+                }
+            });
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
+    private static void runBlockingOn(EventLoop eventLoop, final Runnable action) {
+        final Promise<Void> promise = eventLoop.newPromise();
+            eventLoop.execute(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        action.run();
+                        promise.setSuccess(null);
+                    } catch (Throwable t) {
+                        promise.tryFailure(t);
+                    }
+                }
+            });
+        try {
+            promise.await();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+        Throwable cause = promise.cause();
+        if (cause != null) {
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            }
+            throw new RuntimeException(cause);
+        }
+    }
+
     private static final Runnable NOOP = new Runnable() {
         @Override
         public void run() { }
@@ -170,6 +307,9 @@ private static void assertRejection(EventExecutor loop) {
         }
     }
 
+    protected boolean supportsChannelIteration() {
+        return false;
+    }
     protected abstract EventLoopGroup newEventLoopGroup();
     protected abstract Channel newChannel();
     protected abstract Class<? extends ServerChannel> serverChannelClass();

testsuite/src/main/java/io/netty/testsuite/transport/DefaultEventLoopTest.java

@@ -19,11 +19,38 @@
 import io.netty.channel.DefaultEventLoopGroup;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.ServerChannel;
+import io.netty.channel.SingleThreadEventLoop;
 import io.netty.channel.local.LocalChannel;
 import io.netty.channel.local.LocalServerChannel;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.function.Executable;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class DefaultEventLoopTest extends AbstractSingleThreadEventLoopTest {
 
+    @Test
+    @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
+    public void testChannelsIteratorNotSupported() throws Exception {
+        EventLoopGroup group = newEventLoopGroup();
+        final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
+        try {
+            final Channel ch = newChannel();
+            loop.register(ch).syncUninterruptibly();
+
+            assertThrows(UnsupportedOperationException.class, new Executable() {
+                @Override
+                public void execute() throws Throwable {
+                    loop.registeredChannelsIterator();
+                }
+            });
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+
     @Override
     protected EventLoopGroup newEventLoopGroup() {
         return new DefaultEventLoopGroup();

testsuite/src/main/java/io/netty/testsuite/transport/NioEventLoopTest.java

@@ -24,6 +24,11 @@
 
 public class NioEventLoopTest extends AbstractSingleThreadEventLoopTest {
 
+    @Override
+    protected boolean supportsChannelIteration() {
+        return true;
+    }
+
     @Override
     protected EventLoopGroup newEventLoopGroup() {
         return new NioEventLoopGroup();

transport-classes-epoll/src/main/java/io/netty/channel/epoll/EpollEventLoop.java

@@ -15,6 +15,7 @@
  */
 package io.netty.channel.epoll;
 
+import io.netty.channel.Channel;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.EventLoopTaskQueueFactory;
@@ -34,6 +35,7 @@
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicLong;
@@ -279,6 +281,16 @@ public int registeredChannels() {
         return channels.size();
     }
 
+    @Override
+    public Iterator<Channel> registeredChannelsIterator() {
+        assert inEventLoop();
+        IntObjectMap<AbstractEpollChannel> ch = channels;
+        if (ch.isEmpty()) {
+            return ChannelsReadOnlyIterator.empty();
+        }
+        return new ChannelsReadOnlyIterator<AbstractEpollChannel>(ch.values());
+    }
+
     private long epollWait(long deadlineNanos) throws IOException {
         if (deadlineNanos == NONE) {
             return Native.epollWait(epollFd, events, timerFd,

transport-classes-kqueue/src/main/java/io/netty/channel/kqueue/KQueueEventLoop.java

@@ -15,6 +15,7 @@
  */
 package io.netty.channel.kqueue;
 
+import io.netty.channel.Channel;
 import io.netty.channel.EventLoop;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.EventLoopTaskQueueFactory;
@@ -33,6 +34,7 @@
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
@@ -353,6 +355,16 @@ public int registeredChannels() {
         return channels.size();
     }
 
+    @Override
+    public Iterator<Channel> registeredChannelsIterator() {
+        assert inEventLoop();
+        IntObjectMap<AbstractKQueueChannel> ch = channels;
+        if (ch.isEmpty()) {
+            return ChannelsReadOnlyIterator.empty();
+        }
+        return new ChannelsReadOnlyIterator<AbstractKQueueChannel>(ch.values());
+    }
+
     @Override
     protected void cleanup() {
         try {

transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollEventLoopTest.java

@@ -40,6 +40,11 @@
 
 public class EpollEventLoopTest extends AbstractSingleThreadEventLoopTest {
 
+    @Override
+    protected boolean supportsChannelIteration() {
+        return true;
+    }
+
     @Override
     protected EventLoopGroup newEventLoopGroup() {
         return new EpollEventLoopGroup();

transport-native-kqueue/src/test/java/io/netty/channel/kqueue/KQueueEventLoopTest.java

@@ -30,6 +30,11 @@
 
 public class KQueueEventLoopTest extends AbstractSingleThreadEventLoopTest {
 
+    @Override
+    protected boolean supportsChannelIteration() {
+        return true;
+    }
+
     @Override
     protected EventLoopGroup newEventLoopGroup() {
         return new KQueueEventLoopGroup();

transport/src/main/java/io/netty/channel/SingleThreadEventLoop.java

@@ -22,6 +22,8 @@
 import io.netty.util.internal.SystemPropertyUtil;
 import io.netty.util.internal.UnstableApi;
 
+import java.util.Iterator;
+import java.util.NoSuchElementException;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadFactory;
@@ -154,4 +156,62 @@ public int pendingTasks() {
     public int registeredChannels() {
         return -1;
     }
+
+    /**
+     * @return read-only iterator of active {@link Channel}s registered with this {@link EventLoop}.
+     *         The returned value is not guaranteed to be exact accurate and
+     *         should be viewed as a best effort. This method is expected to be called from within
+     *         event loop.
+     * @throws UnsupportedOperationException if operation is not supported by implementation.
+     */
+    @UnstableApi
+    public Iterator<Channel> registeredChannelsIterator() {
+        throw new UnsupportedOperationException("registeredChannelsIterator");
+    }
+
+    protected static final class ChannelsReadOnlyIterator<T extends Channel> implements Iterator<Channel> {
+        private final Iterator<T> channelIterator;
+
+        public ChannelsReadOnlyIterator(Iterable<T> channelIterable) {
+            this.channelIterator =
+                    ObjectUtil.checkNotNull(channelIterable, "channelIterable").iterator();
+        }
+
+        @Override
+        public boolean hasNext() {
+            return channelIterator.hasNext();
+        }
+
+        @Override
+        public Channel next() {
+            return channelIterator.next();
+        }
+
+        @Override
+        public void remove() {
+            throw new UnsupportedOperationException("remove");
+        }
+
+        @SuppressWarnings("unchecked")
+        public static <T> Iterator<T> empty() {
+            return (Iterator<T>) EMPTY;
+        }
+
+        private static final Iterator<Object> EMPTY = new Iterator<Object>() {
+            @Override
+            public boolean hasNext() {
+                return false;
+            }
+
+            @Override
+            public Object next() {
+                throw new NoSuchElementException();
+            }
+
+            @Override
+            public void remove() {
+                throw new UnsupportedOperationException("remove");
+            }
+        };
+    }
 }