Do not schedule notify task if there are no listeners attached to the promise. (#8797)

Motivation:

If there are no listeners attached to the promise when full-filling it we do not need to schedule a task to notify.

Modifications:

- Don't schedule a task if there is nothing to notify.
- Add unit tests.

Result:

Fixes https://github.com/netty/netty/issues/8795.

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

@@ -91,7 +91,6 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
     @Override
     public Promise<V> setSuccess(V result) {
         if (setSuccess0(result)) {
-            notifyListeners();
             return this;
         }
         throw new IllegalStateException("complete already: " + this);
@@ -99,17 +98,12 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
 
     @Override
     public boolean trySuccess(V result) {
-        if (setSuccess0(result)) {
+        return setSuccess0(result);
-            notifyListeners();
-            return true;
-        }
-        return false;
     }
 
     @Override
     public Promise<V> setFailure(Throwable cause) {
         if (setFailure0(cause)) {
-            notifyListeners();
             return this;
         }
         throw new IllegalStateException("complete already: " + this, cause);
@@ -117,11 +111,7 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
 
     @Override
     public boolean tryFailure(Throwable cause) {
-        if (setFailure0(cause)) {
+        return setFailure0(cause);
-            notifyListeners();
-            return true;
-        }
-        return false;
     }
 
     @Override
@@ -315,8 +305,9 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
         if (RESULT_UPDATER.compareAndSet(this, null, CANCELLATION_CAUSE_HOLDER)) {
-            checkNotifyWaiters();
+            if (checkNotifyWaiters()) {
-            notifyListeners();
+                notifyListeners();
+            }
             return true;
         }
         return false;
@@ -535,16 +526,23 @@ public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
     private boolean setValue0(Object objResult) {
         if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
             RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
-            checkNotifyWaiters();
+            if (checkNotifyWaiters()) {
+                notifyListeners();
+            }
             return true;
         }
         return false;
     }
 
-    private synchronized void checkNotifyWaiters() {
+    /**
+     * Check if there are any waiters and if so notify these.
+     * @return {@code true} if there are any listeners attached to the promise, {@code false} otherwise.
+     */
+    private synchronized boolean checkNotifyWaiters() {
         if (waiters > 0) {
             notifyAll();
         }
+        return listeners != null;
     }
 
     private void incWaiters() {

common/src/test/java/io/netty/util/concurrent/DefaultPromiseTest.java

@@ -21,6 +21,7 @@ import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -63,6 +64,41 @@ public class DefaultPromiseTest {
         return max(stackOverflowDepth << 1, stackOverflowDepth);
     }
 
+    @Test
+    public void testCancelDoesNotScheduleWhenNoListeners() {
+        EventExecutor executor = Mockito.mock(EventExecutor.class);
+        Mockito.when(executor.inEventLoop()).thenReturn(false);
+
+        Promise<Void> promise = new DefaultPromise<Void>(executor);
+        promise.cancel(false);
+        Mockito.verify(executor, Mockito.never()).execute(Mockito.any(Runnable.class));
+        assertTrue(promise.isCancelled());
+    }
+
+    @Test
+    public void testSuccessDoesNotScheduleWhenNoListeners() {
+        EventExecutor executor = Mockito.mock(EventExecutor.class);
+        Mockito.when(executor.inEventLoop()).thenReturn(false);
+
+        Object value = new Object();
+        Promise<Object> promise = new DefaultPromise<Object>(executor);
+        promise.setSuccess(value);
+        Mockito.verify(executor, Mockito.never()).execute(Mockito.any(Runnable.class));
+        assertSame(value, promise.getNow());
+    }
+
+    @Test
+    public void testFailureDoesNotScheduleWhenNoListeners() {
+        EventExecutor executor = Mockito.mock(EventExecutor.class);
+        Mockito.when(executor.inEventLoop()).thenReturn(false);
+
+        Exception cause = new Exception();
+        Promise<Void> promise = new DefaultPromise<Void>(executor);
+        promise.setFailure(cause);
+        Mockito.verify(executor, Mockito.never()).execute(Mockito.any(Runnable.class));
+        assertSame(cause, promise.cause());
+    }
+
     @Test(expected = CancellationException.class)
     public void testCancellationExceptionIsThrownWhenBlockingGet() throws InterruptedException, ExecutionException {
         final Promise<Void> promise = new DefaultPromise<>(ImmediateEventExecutor.INSTANCE);