Use OneTimeTask where possible to reduce object creation

Motivation:

We should use OneTimeTask where possible to reduce object creation.

Modifications:

Replace Runnable with OneTimeTask

Result:

Less object creation

codec/src/main/java/io/netty/handler/codec/compression/JZlibEncoder.java

@@ -26,6 +26,7 @@ import io.netty.channel.ChannelPromise;
 import io.netty.channel.ChannelPromiseNotifier;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.internal.EmptyArrays;
+import io.netty.util.internal.OneTimeTask;
 
 import java.util.concurrent.TimeUnit;
 
@@ -252,7 +253,7 @@ public class JZlibEncoder extends ZlibEncoder {
             return finishEncode(ctx, promise);
         } else {
             final ChannelPromise p = ctx.newPromise();
-            executor.execute(new Runnable() {
+            executor.execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     ChannelFuture f = finishEncode(ctx(), p);
@@ -351,7 +352,7 @@ public class JZlibEncoder extends ZlibEncoder {
 
         if (!f.isDone()) {
             // Ensure the channel is closed even if the write operation completes in time.
-            ctx.executor().schedule(new Runnable() {
+            ctx.executor().schedule(new OneTimeTask() {
                 @Override
                 public void run() {
                     ctx.close(promise);

codec/src/main/java/io/netty/handler/codec/compression/JdkZlibEncoder.java

@@ -22,6 +22,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.ChannelPromiseNotifier;
 import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.internal.OneTimeTask;
 
 import java.util.concurrent.TimeUnit;
 import java.util.zip.CRC32;
@@ -163,7 +164,7 @@ public class JdkZlibEncoder extends ZlibEncoder {
             return finishEncode(ctx, promise);
         } else {
             final ChannelPromise p = ctx.newPromise();
-            executor.execute(new Runnable() {
+            executor.execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     ChannelFuture f = finishEncode(ctx(), p);
@@ -259,7 +260,7 @@ public class JdkZlibEncoder extends ZlibEncoder {
 
         if (!f.isDone()) {
             // Ensure the channel is closed even if the write operation completes in time.
-            ctx.executor().schedule(new Runnable() {
+            ctx.executor().schedule(new OneTimeTask() {
                 @Override
                 public void run() {
                     ctx.close(promise);

handler/src/main/java/io/netty/handler/ssl/SslHandler.java

@@ -393,10 +393,10 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
      */
     public ChannelFuture close(final ChannelPromise future) {
         final ChannelHandlerContext ctx = this.ctx;
-        ctx.executor().execute(new Runnable() {
+        ctx.executor().execute(new OneTimeTask() {
             @Override
             public void run() {
-                SslHandler.this.outboundClosed = true;
+                outboundClosed = true;
                 engine.closeOutbound();
                 try {
                     write(ctx, Unpooled.EMPTY_BUFFER, future);
@@ -1179,7 +1179,7 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
             }
 
             final CountDownLatch latch = new CountDownLatch(1);
-            delegatedTaskExecutor.execute(new Runnable() {
+            delegatedTaskExecutor.execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     try {
@@ -1414,7 +1414,7 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
             return;
         }
 
-        final ScheduledFuture<?> timeoutFuture = ctx.executor().schedule(new Runnable() {
+        final ScheduledFuture<?> timeoutFuture = ctx.executor().schedule(new OneTimeTask() {
             @Override
             public void run() {
                 if (p.isDone()) {
@@ -1456,7 +1456,7 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
         final ScheduledFuture<?> timeoutFuture;
         if (closeNotifyTimeoutMillis > 0) {
             // Force-close the connection if close_notify is not fully sent in time.
-            timeoutFuture = ctx.executor().schedule(new Runnable() {
+            timeoutFuture = ctx.executor().schedule(new OneTimeTask() {
                 @Override
                 public void run() {
                     logger.warn("{} Last write attempt timed out; force-closing the connection.", ctx.channel());

handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java

@@ -28,6 +28,7 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelProgressivePromise;
 import io.netty.channel.ChannelPromise;
 import io.netty.util.ReferenceCountUtil;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -113,7 +114,7 @@ public class ChunkedWriteHandler
             }
         } else {
             // let the transfer resume on the next event loop round
-            ctx.executor().execute(new Runnable() {
+            ctx.executor().execute(new OneTimeTask() {
 
                 @Override
                 public void run() {

handler/src/main/java/io/netty/handler/timeout/WriteTimeoutHandler.java

@@ -24,6 +24,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOutboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
+import io.netty.util.internal.OneTimeTask;
 
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -110,7 +111,7 @@ public class WriteTimeoutHandler extends ChannelOutboundHandlerAdapter {
     private void scheduleTimeout(final ChannelHandlerContext ctx, final ChannelPromise future) {
         if (timeoutNanos > 0) {
             // Schedule a timeout.
-            final ScheduledFuture<?> sf = ctx.executor().schedule(new Runnable() {
+            final ScheduledFuture<?> sf = ctx.executor().schedule(new OneTimeTask() {
                 @Override
                 public void run() {
                     // Was not written yet so issue a write timeout

handler/src/main/java/io/netty/handler/traffic/ChannelTrafficShapingHandler.java

@@ -18,6 +18,7 @@ package io.netty.handler.traffic;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
+import io.netty.util.internal.OneTimeTask;
 
 import java.util.ArrayDeque;
 import java.util.concurrent.TimeUnit;
@@ -192,7 +193,7 @@ public class ChannelTrafficShapingHandler extends AbstractTrafficShapingHandler
             checkWriteSuspend(ctx, delay, queueSize);
         }
         final long futureNow = newToSend.relativeTimeAction;
-        ctx.executor().schedule(new Runnable() {
+        ctx.executor().schedule(new OneTimeTask() {
             @Override
             public void run() {
                 sendAllValid(ctx, futureNow);

handler/src/main/java/io/netty/handler/traffic/GlobalTrafficShapingHandler.java

@@ -21,6 +21,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.PlatformDependent;
 
 import java.util.ArrayDeque;
@@ -360,7 +361,7 @@ public class GlobalTrafficShapingHandler extends AbstractTrafficShapingHandler {
         }
         final long futureNow = newToSend.relativeTimeAction;
         final PerChannel forSchedule = perChannel;
-        ctx.executor().schedule(new Runnable() {
+        ctx.executor().schedule(new OneTimeTask() {
             @Override
             public void run() {
                 sendAllValid(ctx, forSchedule, futureNow);

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java

@@ -650,7 +650,7 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
                     // Schedule connect timeout.
                     int connectTimeoutMillis = config().getConnectTimeoutMillis();
                     if (connectTimeoutMillis > 0) {
-                        connectTimeoutFuture = eventLoop().schedule(new Runnable() {
+                        connectTimeoutFuture = eventLoop().schedule(new OneTimeTask() {
                             @Override
                             public void run() {
                                 ChannelPromise connectPromise = AbstractEpollStreamChannel.this.connectPromise;
@@ -859,7 +859,7 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel {
                 if (!closed) {
                     // trigger a read again as there may be something left to read and because of epoll ET we
                     // will not get notified again until we read everything from the socket
-                    eventLoop().execute(new Runnable() {
+                    eventLoop().execute(new OneTimeTask() {
                         @Override
                         public void run() {
                             epollInReady();

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannel.java

@@ -23,6 +23,7 @@ import io.netty.channel.unix.DomainSocketAddress;
 import io.netty.channel.unix.DomainSocketChannel;
 import io.netty.channel.unix.FileDescriptor;
 import io.netty.channel.unix.Socket;
+import io.netty.util.internal.OneTimeTask;
 
 import java.net.SocketAddress;
 
@@ -200,7 +201,7 @@ public final class EpollDomainSocketChannel extends AbstractEpollStreamChannel i
                 pipeline.fireExceptionCaught(t);
                 // trigger a read again as there may be something left to read and because of epoll ET we
                 // will not get notified again until we read everything from the socket
-                eventLoop().execute(new Runnable() {
+                eventLoop().execute(new OneTimeTask() {
                     @Override
                     public void run() {
                         epollInReady();

transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java

@@ -20,6 +20,7 @@ import gnu.io.CommPortIdentifier;
 import gnu.io.SerialPort;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.oio.OioByteStreamChannel;
+import io.netty.util.internal.OneTimeTask;
 
 import java.net.SocketAddress;
 import java.util.concurrent.TimeUnit;
@@ -143,7 +144,7 @@ public class RxtxChannel extends OioByteStreamChannel {
 
                 int waitTime = config().getOption(WAIT_TIME);
                 if (waitTime > 0) {
-                    eventLoop().schedule(new Runnable() {
+                    eventLoop().schedule(new OneTimeTask() {
                         @Override
                         public void run() {
                             try {

transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java

@@ -34,6 +34,7 @@ import io.netty.channel.sctp.SctpChannelConfig;
 import io.netty.channel.sctp.SctpMessage;
 import io.netty.channel.sctp.SctpNotificationHandler;
 import io.netty.channel.sctp.SctpServerChannel;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.logging.InternalLogger;
@@ -359,7 +360,7 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     bindAddress(localAddress, promise);
@@ -384,7 +385,7 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     unbindAddress(localAddress, promise);

transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpServerChannel.java

@@ -25,6 +25,7 @@ import io.netty.channel.ChannelPromise;
 import io.netty.channel.nio.AbstractNioMessageChannel;
 import io.netty.channel.sctp.DefaultSctpServerChannelConfig;
 import io.netty.channel.sctp.SctpServerChannelConfig;
+import io.netty.util.internal.OneTimeTask;
 
 import java.io.IOException;
 import java.net.InetAddress;
@@ -159,7 +160,7 @@ public class NioSctpServerChannel extends AbstractNioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     bindAddress(localAddress, promise);
@@ -184,7 +185,7 @@ public class NioSctpServerChannel extends AbstractNioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     unbindAddress(localAddress, promise);

transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java

@@ -33,6 +33,7 @@ import io.netty.channel.sctp.SctpChannelConfig;
 import io.netty.channel.sctp.SctpMessage;
 import io.netty.channel.sctp.SctpNotificationHandler;
 import io.netty.channel.sctp.SctpServerChannel;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.logging.InternalLogger;
@@ -420,7 +421,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     bindAddress(localAddress, promise);
@@ -445,7 +446,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     unbindAddress(localAddress, promise);

transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpServerChannel.java

@@ -25,6 +25,7 @@ import io.netty.channel.ChannelPromise;
 import io.netty.channel.oio.AbstractOioMessageChannel;
 import io.netty.channel.sctp.DefaultSctpServerChannelConfig;
 import io.netty.channel.sctp.SctpServerChannelConfig;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -234,7 +235,7 @@ public class OioSctpServerChannel extends AbstractOioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     bindAddress(localAddress, promise);
@@ -259,7 +260,7 @@ public class OioSctpServerChannel extends AbstractOioMessageChannel
                 promise.setFailure(t);
             }
         } else {
-            eventLoop().execute(new Runnable() {
+            eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     unbindAddress(localAddress, promise);

transport/src/main/java/io/netty/bootstrap/AbstractBootstrap.java

@@ -29,6 +29,7 @@ import io.netty.channel.EventLoopGroup;
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.GlobalEventExecutor;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.StringUtil;
 
 import java.net.InetAddress;
@@ -341,7 +342,7 @@ public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C ext
 
         // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
         // the pipeline in its channelRegistered() implementation.
-        channel.eventLoop().execute(new Runnable() {
+        channel.eventLoop().execute(new OneTimeTask() {
             @Override
             public void run() {
                 if (regFuture.isSuccess()) {

transport/src/main/java/io/netty/bootstrap/Bootstrap.java

@@ -23,6 +23,7 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.AttributeKey;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -158,7 +159,7 @@ public class Bootstrap extends AbstractBootstrap<Bootstrap, Channel> {
 
         // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
         // the pipeline in its channelRegistered() implementation.
-        channel.eventLoop().execute(new Runnable() {
+        channel.eventLoop().execute(new OneTimeTask() {
             @Override
             public void run() {
                 if (regFuture.isSuccess()) {

transport/src/main/java/io/netty/bootstrap/ServerBootstrap.java

@@ -28,6 +28,7 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.ServerChannel;
 import io.netty.util.AttributeKey;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
@@ -274,7 +275,7 @@ public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerCh
                 // stop accept new connections for 1 second to allow the channel to recover
                 // See https://github.com/netty/netty/issues/1328
                 config.setAutoRead(false);
-                ctx.channel().eventLoop().schedule(new Runnable() {
+                ctx.channel().eventLoop().schedule(new OneTimeTask() {
                     @Override
                     public void run() {
                        config.setAutoRead(true);

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

@@ -24,6 +24,7 @@ import io.netty.util.Recycler.Handle;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.internal.InternalThreadLocalMap;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
@@ -629,7 +630,7 @@ public final class ChannelOutboundBuffer {
 
     void close(final ClosedChannelException cause) {
         if (inFail) {
-            channel.eventLoop().execute(new Runnable() {
+            channel.eventLoop().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     close(cause);

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

@@ -324,7 +324,7 @@ final class DefaultChannelPipeline implements ChannelPipeline {
                 remove0(ctx);
                 return ctx;
             } else {
-               future = ctx.executor().submit(new Runnable() {
+               future = ctx.executor().submit(new OneTimeTask() {
                    @Override
                    public void run() {
                        synchronized (DefaultChannelPipeline.this) {
@@ -405,7 +405,7 @@ final class DefaultChannelPipeline implements ChannelPipeline {
                 replace0(ctx, newCtx);
                 return ctx.handler();
             } else {
-                future = newCtx.executor().submit(new Runnable() {
+                future = newCtx.executor().submit(new OneTimeTask() {
                     @Override
                     public void run() {
                         synchronized (DefaultChannelPipeline.this) {
@@ -465,7 +465,7 @@ final class DefaultChannelPipeline implements ChannelPipeline {
 
     private void callHandlerAdded(final ChannelHandlerContext ctx) {
         if (ctx.channel().isRegistered() && !ctx.executor().inEventLoop()) {
-            ctx.executor().execute(new Runnable() {
+            ctx.executor().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     callHandlerAdded0(ctx);
@@ -504,7 +504,7 @@ final class DefaultChannelPipeline implements ChannelPipeline {
 
     private void callHandlerRemoved(final AbstractChannelHandlerContext ctx) {
         if (ctx.channel().isRegistered() && !ctx.executor().inEventLoop()) {
-            ctx.executor().execute(new Runnable() {
+            ctx.executor().execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     callHandlerRemoved0(ctx);

transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannel.java

@@ -25,6 +25,7 @@ import io.netty.channel.EventLoop;
 import io.netty.channel.oio.OioByteStreamChannel;
 import io.netty.channel.socket.ServerSocketChannel;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.util.internal.OneTimeTask;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
@@ -152,7 +153,7 @@ public class OioSocketChannel extends OioByteStreamChannel
                 future.setFailure(t);
             }
         } else {
-            loop.execute(new Runnable() {
+            loop.execute(new OneTimeTask() {
                 @Override
                 public void run() {
                     shutdownOutput(future);