Add default methods to EventExecutor / EventExecutorGroup / EventLoop / EventLoopGroup (#11649)

Motivation:

We can remove some classes and duplication if we add default methods

Modifications:

- Add default methods to EventExecutor / EventExecutorGroup / EventLoop / EventLoopGroup
- Remove code duplication
- Remove AbstractEventExecutorGroup as it is not needed anymore

Result:

Cleanup and removal of code-duplication. Also makes it easier for people to implement their custom executors / groups

common/src/main/java/io/netty/util/concurrent/AbstractEventExecutor.java

@@ -18,14 +18,9 @@ package io.netty.util.concurrent;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
 import java.util.concurrent.AbstractExecutorService;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Abstract base class for {@link EventExecutor} implementations.
@@ -35,46 +30,8 @@ public abstract class AbstractEventExecutor extends AbstractExecutorService impl
     static final long DEFAULT_SHUTDOWN_QUIET_PERIOD = 2;
     static final long DEFAULT_SHUTDOWN_TIMEOUT = 15;
 
-    private final Collection<EventExecutor> selfCollection = Collections.singleton(this);
     private final Future<?> successfulVoidFuture = DefaultPromise.newSuccessfulPromise(this, null).asFuture();
 
-    @Override
-    public EventExecutor next() {
-        return this;
-    }
-
-    @Override
-    public final boolean inEventLoop() {
-        return inEventLoop(Thread.currentThread());
-    }
-
-    @Override
-    public final Iterator<EventExecutor> iterator() {
-        return selfCollection.iterator();
-    }
-
-    @Override
-    public final Future<?> shutdownGracefully() {
-        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
-    }
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public abstract void shutdown();
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public List<Runnable> shutdownNow() {
-        shutdown();
-        return Collections.emptyList();
-    }
-
     @Override
     public <V> Future<V> newSucceededFuture(V result) {
         if (result == null) {

common/src/main/java/io/netty/util/concurrent/AbstractEventExecutorGroup.java

@@ -1,117 +0,0 @@
-/*
- * Copyright 2013 The Netty Project
- *
- * The Netty Project licenses this file to you under the Apache License,
- * version 2.0 (the "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at:
- *
- *   https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
- * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
- * License for the specific language governing permissions and limitations
- * under the License.
- */
-package io.netty.util.concurrent;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import static io.netty.util.concurrent.AbstractEventExecutor.*;
-
-
-/**
- * Abstract base class for {@link EventExecutorGroup} implementations.
- */
-public abstract class AbstractEventExecutorGroup implements EventExecutorGroup {
-    @Override
-    public Future<?> submit(Runnable task) {
-        return next().submit(task);
-    }
-
-    @Override
-    public <T> Future<T> submit(Runnable task, T result) {
-        return next().submit(task, result);
-    }
-
-    @Override
-    public <T> Future<T> submit(Callable<T> task) {
-        return next().submit(task);
-    }
-
-    @Override
-    public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
-        return next().schedule(command, delay, unit);
-    }
-
-    @Override
-    public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
-        return next().schedule(callable, delay, unit);
-    }
-
-    @Override
-    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
-        return next().scheduleAtFixedRate(command, initialDelay, period, unit);
-    }
-
-    @Override
-    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
-        return next().scheduleWithFixedDelay(command, initialDelay, delay, unit);
-    }
-
-    @Override
-    public Future<?> shutdownGracefully() {
-        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
-    }
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public abstract void shutdown();
-
-    /**
-     * @deprecated {@link #shutdownGracefully(long, long, TimeUnit)} or {@link #shutdownGracefully()} instead.
-     */
-    @Override
-    @Deprecated
-    public List<Runnable> shutdownNow() {
-        shutdown();
-        return Collections.emptyList();
-    }
-
-    @Override
-    public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
-            throws InterruptedException {
-        return next().invokeAll(tasks);
-    }
-
-    @Override
-    public <T> List<java.util.concurrent.Future<T>> invokeAll(
-            Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
-        return next().invokeAll(tasks, timeout, unit);
-    }
-
-    @Override
-    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
-        return next().invokeAny(tasks);
-    }
-
-    @Override
-    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
-            throws InterruptedException, ExecutionException, TimeoutException {
-        return next().invokeAny(tasks, timeout, unit);
-    }
-
-    @Override
-    public void execute(Runnable command) {
-        next().execute(command);
-    }
-}

common/src/main/java/io/netty/util/concurrent/EventExecutor.java

@@ -15,6 +15,9 @@
  */
 package io.netty.util.concurrent;
 
+import java.util.Collections;
+import java.util.Iterator;
+
 /**
  * The {@link EventExecutor} is a special {@link EventExecutorGroup} which comes
  * with some handy methods to see if a {@link Thread} is executed in a event loop.
@@ -28,12 +31,21 @@ public interface EventExecutor extends EventExecutorGroup {
      * Returns a reference to itself.
      */
     @Override
-    EventExecutor next();
+    default EventExecutor next() {
+        return this;
+    }
+
+    @Override
+    default Iterator<EventExecutor> iterator() {
+        return Collections.singleton(this).iterator();
+    }
 
     /**
      * Calls {@link #inEventLoop(Thread)} with {@link Thread#currentThread()} as argument
      */
-    boolean inEventLoop();
+    default boolean inEventLoop() {
+        return inEventLoop(Thread.currentThread());
+    }
 
     /**
      * Return {@code true} if the given {@link Thread} is executed in the event loop,

common/src/main/java/io/netty/util/concurrent/EventExecutorGroup.java

@@ -15,11 +15,18 @@
  */
 package io.netty.util.concurrent;
 
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static io.netty.util.concurrent.AbstractEventExecutor.DEFAULT_SHUTDOWN_QUIET_PERIOD;
+import static io.netty.util.concurrent.AbstractEventExecutor.DEFAULT_SHUTDOWN_TIMEOUT;
 
 /**
  * The {@link EventExecutorGroup} is responsible for providing the {@link EventExecutor}'s to use
@@ -40,7 +47,9 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
      *
      * @return the {@link #terminationFuture()}
      */
-    Future<?> shutdownGracefully();
+    default Future<?> shutdownGracefully() {
+        return shutdownGracefully(DEFAULT_SHUTDOWN_QUIET_PERIOD, DEFAULT_SHUTDOWN_TIMEOUT, TimeUnit.SECONDS);
+    }
 
     /**
      * Signals this executor that the caller wants the executor to be shut down.  Once this method is called,
@@ -76,7 +85,10 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
      */
     @Override
     @Deprecated
-    List<Runnable> shutdownNow();
+    default List<Runnable> shutdownNow() {
+        shutdown();
+        return Collections.emptyList();
+    }
 
     /**
      * Returns one of the {@link EventExecutor}s managed by this {@link EventExecutorGroup}.
@@ -87,23 +99,65 @@ public interface EventExecutorGroup extends ScheduledExecutorService, Iterable<E
     Iterator<EventExecutor> iterator();
 
     @Override
-    Future<?> submit(Runnable task);
+    default Future<?> submit(Runnable task) {
+        return next().submit(task);
+    }
 
     @Override
-    <T> Future<T> submit(Runnable task, T result);
+    default <T> Future<T> submit(Runnable task, T result) {
+        return next().submit(task, result);
+    }
 
     @Override
-    <T> Future<T> submit(Callable<T> task);
+    default <T> Future<T> submit(Callable<T> task) {
+        return next().submit(task);
+    }
 
     @Override
-    ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
+    default ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        return next().schedule(command, delay, unit);
+    }
 
     @Override
-    <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
+    default <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+        return next().schedule(callable, delay, unit);
+    }
 
     @Override
-    ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
+    default ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
+        return next().scheduleAtFixedRate(command, initialDelay, period, unit);
+    }
 
     @Override
-    ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
+    default ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
+        return next().scheduleWithFixedDelay(command, initialDelay, delay, unit);
+    }
+
+    @Override
+    default <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
+            throws InterruptedException {
+        return next().invokeAll(tasks);
+    }
+
+    @Override
+    default <T> List<java.util.concurrent.Future<T>> invokeAll(
+            Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
+        return next().invokeAll(tasks, timeout, unit);
+    }
+
+    @Override
+    default <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+        return next().invokeAny(tasks);
+    }
+
+    @Override
+    default <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        return next().invokeAny(tasks, timeout, unit);
+    }
+
+    @Override
+    default void execute(Runnable command) {
+        next().execute(command);
+    }
 }

common/src/main/java/io/netty/util/concurrent/MultithreadEventExecutorGroup.java

@@ -33,12 +33,12 @@ import static io.netty.util.internal.ObjectUtil.checkPositive;
  * {@link EventExecutorGroup} implementation that handles their tasks with multiple threads at
  * the same time.
  */
-public class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
+public class MultithreadEventExecutorGroup implements EventExecutorGroup {
 
     private final EventExecutor[] children;
     private final List<EventExecutor> readonlyChildren;
     private final AtomicInteger terminatedChildren = new AtomicInteger();
-    private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
+    private final Promise<?> terminationFuture = GlobalEventExecutor.INSTANCE.newPromise();
     private final boolean powerOfTwo;
 
     /**

common/src/main/java/io/netty/util/concurrent/UnorderedThreadPoolEventExecutor.java

@@ -18,10 +18,7 @@ package io.netty.util.concurrent;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
-import java.util.Collections;
-import java.util.Iterator;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Delayed;
 import java.util.concurrent.RejectedExecutionHandler;
@@ -44,7 +41,6 @@ public final class UnorderedThreadPoolEventExecutor extends ScheduledThreadPoolE
             UnorderedThreadPoolEventExecutor.class);
 
     private final Promise<?> terminationFuture = GlobalEventExecutor.INSTANCE.newPromise();
-    private final Set<EventExecutor> executorSet = Collections.singleton(this);
 
     /**
      * Calls {@link UnorderedThreadPoolEventExecutor#UnorderedThreadPoolEventExecutor(int, ThreadFactory)}
@@ -77,16 +73,6 @@ public final class UnorderedThreadPoolEventExecutor extends ScheduledThreadPoolE
         super(corePoolSize, threadFactory, handler);
     }
 
-    @Override
-    public EventExecutor next() {
-        return this;
-    }
-
-    @Override
-    public boolean inEventLoop() {
-        return false;
-    }
-
     @Override
     public boolean inEventLoop(Thread thread) {
         return false;
@@ -110,11 +96,6 @@ public final class UnorderedThreadPoolEventExecutor extends ScheduledThreadPoolE
         terminationFuture.trySuccess(null);
     }
 
-    @Override
-    public Future<?> shutdownGracefully() {
-        return shutdownGracefully(2, 15, TimeUnit.SECONDS);
-    }
-
     @Override
     public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
         // TODO: At the moment this just calls shutdown but we may be able to do something more smart here which
@@ -128,11 +109,6 @@ public final class UnorderedThreadPoolEventExecutor extends ScheduledThreadPoolE
         return terminationFuture.asFuture();
     }
 
-    @Override
-    public Iterator<EventExecutor> iterator() {
-        return executorSet.iterator();
-    }
-
     @Override
     protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
         return runnable instanceof NonNotifyRunnable ?

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

@@ -26,6 +26,11 @@ import io.netty.util.concurrent.OrderedEventExecutor;
  */
 public interface EventLoop extends OrderedEventExecutor, EventLoopGroup {
 
+    @Override
+    default EventLoop next() {
+        return this;
+    }
+
     /**
      * Returns an <em>internal-use-only</em> object that provides unsafe operations.
      */

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

@@ -167,11 +167,6 @@ public class SingleThreadEventLoop extends SingleThreadEventExecutor implements
                 : PlatformDependent.newMpscQueue(maxPendingTasks);
     }
 
-    @Override
-    public final EventLoop next() {
-        return this;
-    }
-
     @Override
     protected final boolean wakesUpForTask(Runnable task) {
         return !(task instanceof NonWakeupRunnable);