Only run one SSL task per delegation (#11462)

Motivation:

We should only run one SSL task per delegation to allow more SSLEngines to make progress in a timely manner

Modifications:

- Only run one task per delegation to the executor
- Only create new SSL task if really needed
- Only schedule if not on the EventExecutor thread

Result:

More fair usage of resources and less allocations

common/src/main/java/io/netty/util/internal/Hidden.java

@@ -84,7 +84,10 @@ class Hidden {
                     "io.netty.handler.ssl.SslHandler",
                     "runAllDelegatedTasks"
             );
-
+            builder.allowBlockingCallsInside(
+                    "io.netty.handler.ssl.SslHandler",
+                    "runDelegatedTasks"
+            );
             builder.allowBlockingCallsInside(
                     "io.netty.util.concurrent.GlobalEventExecutor",
                     "takeTask");

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

@@ -394,6 +394,9 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
 
     private final boolean startTls;
 
+    private final SslTasksRunner sslTaskRunnerForUnwrap = new SslTasksRunner(true);
+    private final SslTasksRunner sslTaskRunner = new SslTasksRunner(false);
+
     private SslHandlerCoalescingBufferQueue pendingUnencryptedWrites;
     private Promise<Channel> handshakePromise = new LazyChannelPromise();
     private final LazyChannelPromise sslClosePromise = new LazyChannelPromise();
@@ -1525,19 +1528,24 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
      */
     private boolean runDelegatedTasks(boolean inUnwrap) {
         if (delegatedTaskExecutor == ImmediateExecutor.INSTANCE || inEventLoop(delegatedTaskExecutor)) {
-            // We should run the task directly in the EventExecutor thread and not offload at all.
+            // We should run the task directly in the EventExecutor thread and not offload at all. As we are on the
+            // EventLoop we can just run all tasks at once.
             runAllDelegatedTasks(engine);
             return true;
         } else {
-            executeDelegatedTasks(inUnwrap);
+            executeDelegatedTask(inUnwrap);
             return false;
         }
     }
 
-    private void executeDelegatedTasks(boolean inUnwrap) {
+    private void executeDelegatedTask(boolean inUnwrap) {
+        executeDelegatedTask(inUnwrap ? sslTaskRunnerForUnwrap : sslTaskRunner);
+    }
+
+    private void executeDelegatedTask(SslTasksRunner task) {
         setState(STATE_PROCESS_TASK);
         try {
-            delegatedTaskExecutor.execute(new SslTasksRunner(inUnwrap));
+            delegatedTaskExecutor.execute(task);
         } catch (RejectedExecutionException e) {
             clearState(STATE_PROCESS_TASK);
             throw e;
@@ -1615,9 +1623,10 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
             try {
                 HandshakeStatus status = engine.getHandshakeStatus();
                 switch (status) {
-                    // There is another task that needs to be executed and offloaded to the delegatingTaskExecutor.
+                    // There is another task that needs to be executed and offloaded to the delegatingTaskExecutor as
+                    // a result of this. Let's reschedule....
                     case NEED_TASK:
-                        executeDelegatedTasks(inUnwrap);
+                        executeDelegatedTask(this);
 
                         break;
 
@@ -1693,18 +1702,25 @@ public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundH
         @Override
         public void run() {
             try {
-                runAllDelegatedTasks(engine);
+                Runnable task = engine.getDelegatedTask();
+                if (task == null) {
+                    // The task was processed in the meantime. Let's just return.
+                    return;
+                }
+                task.run();
 
-                // All tasks were processed.
+                EventExecutor executor = ctx.executor();
-                assert engine.getHandshakeStatus() != HandshakeStatus.NEED_TASK;
+                if (executor.inEventLoop()) {
-
+                    resumeOnEventExecutor();
-                // Jump back on the EventExecutor.
+                } else {
-                ctx.executor().execute(new Runnable() {
+                    // Jump back on the EventExecutor.
-                    @Override
+                    executor.execute(new Runnable() {
-                    public void run() {
+                        @Override
-                        resumeOnEventExecutor();
+                        public void run() {
-                    }
+                            resumeOnEventExecutor();
-                });
+                        }
+                    });
+                }
             } catch (final Throwable cause) {
                 handleException(cause);
             }