util/util-core: skip scheduler on empty waitqueue

Problem:
Empty waitqueues are submitted to the scheduler. This
will happen when a promise with no continuations is
satisfied

Solution:
Skip empty waitqueues

Result:
Lets also make WaitQueue.Empty an object to make
debugging references easier

JIRA Issues: CSL-11517

Differential Revision: https://phabricator.twitter.biz/D796346

Author: hamdiallam

util-core/src/main/scala/com/twitter/util/Promise.scala

@@ -92,8 +92,11 @@ object Promise {
       loop(this, WaitQueue.empty)
     }
 
-    final def runInScheduler(t: Try[A]): Unit =
-      Scheduler.submit(new Runnable() { def run(): Unit = WaitQueue.this.run(t) })
+    final def runInScheduler(t: Try[A]): Unit = {
+      if (this ne WaitQueue.Empty) {
+        Scheduler.submit(() => WaitQueue.this.run(t))
+      }
+    }
 
     @tailrec
     private def run(t: Try[A]): Unit =
@@ -107,7 +110,7 @@ object Promise {
 
   private[util] object WaitQueue {
 
-    val Empty: WaitQueue[Nothing] = new WaitQueue[Nothing] {
+    object Empty extends WaitQueue[Nothing] {
       final def first: K[Nothing] =
         throw new IllegalStateException("WaitQueue.Empty")
 

util-core/src/test/scala/com/twitter/concurrent/SchedulerTest.scala

@@ -150,7 +150,6 @@ abstract class LocalSchedulerTest(lifo: Boolean) extends AnyFunSuite with Matche
     val record = sampleBlockingFraction(0.0)
     assert(record == null)
   }
-
 }
 
 class LocalSchedulerFifoTest extends LocalSchedulerTest(false)

util-core/src/test/scala/com/twitter/util/PromiseTest.scala

@@ -1,10 +1,20 @@
 package com.twitter.util
 
+import com.twitter.concurrent.Scheduler
+import com.twitter.concurrent.LocalScheduler
 import com.twitter.conversions.DurationOps._
 import org.scalatest.funsuite.AnyFunSuite
 
 class PromiseTest extends AnyFunSuite {
 
+  class TrackingScheduler extends LocalScheduler(false) {
+    @volatile var submitted: Int = 0
+    override def submit(r: Runnable): Unit = {
+      submitted = submitted + 1
+      super.submit(r)
+    }
+  }
+
   test("Promise.detach should not detach other attached promises") {
     val p = new Promise[Unit]
     val attached1 = Promise.attached(p)
@@ -305,4 +315,32 @@ class PromiseTest extends AnyFunSuite {
     assert("main thread" == Await.result(p2, 3.seconds))
     assert("main thread" == Await.result(p, 3.seconds))
   }
+
+  test("promise with no continuations is not submitted to the scheduler") {
+    val old = Scheduler()
+    try {
+      val sched = new TrackingScheduler()
+      Scheduler.setUnsafe(sched)
+      val p = new Promise[Unit]()
+      p.setDone()
+      assert(sched.submitted == 0)
+    } finally {
+      Scheduler.setUnsafe(old)
+    }
+  }
+
+  test("registered continuation on satisfied promise is submitted to the scheduler") {
+    val old = Scheduler()
+    try {
+      val sched = new TrackingScheduler()
+      Scheduler.setUnsafe(sched)
+      val p = new Promise[Unit]()
+      p.setDone()
+
+      p.ensure(())
+      assert(sched.submitted == 1)
+    } finally {
+      Scheduler.setUnsafe(old)
+    }
+  }
 }