Future cancellation by example

Cancellator.scala

import com.twitter.conversions.time._
import com.twitter.finagle.util.DefaultTimer
import com.twitter.util._
import java.util.concurrent.atomic.AtomicInteger
import scala.util.control.NoStackTrace

object Cancellator {

  class Cancelled(n: Int) extends NoStackTrace

  val pool = FuturePool.unboundedPool

  def run(): Unit = {
    val resultFs = (0 until 10000).map { i =>
      val p = new Promise[String]
      p.setInterruptHandler {
        case e: Throwable =>
          // Cancelation doesn't cause the value to be satisfied, so we try to set it here:
          val _ = p.updateIfEmpty(Throw(e))
      }
      
      // Try to cancel the exception and satisfy it at the same time:
      val ready = new AtomicInteger(0)
      pool {
        ready.incrementAndGet()
        while (ready.get != 2) {}
        p.raise(new Cancelled(i))
      }
      pool {
        ready.incrementAndGet()
        while (ready.get != 2) {}
        val _ = p.updateIfEmpty(Return("ok!"))
      }
      
      p.liftToTry
    }

    val countsF = Future.collect(resultFs).map { results =>
      val errors = results.count(_.isThrow)
      (results.size - errors, errors)
    }

    val (ok, errors) = Await.result(countsF)
    println(s"results: $ok ok, $errors errors")
  }
}

console.out

scala> (0 to 10).foreach(_ => Cancellator.run())
results: 9598 ok, 402 errors
results: 9572 ok, 428 errors
results: 9603 ok, 397 errors
results: 9393 ok, 607 errors
results: 9481 ok, 519 errors
results: 9564 ok, 436 errors
results: 9270 ok, 730 errors
results: 9437 ok, 563 errors
results: 9549 ok, 451 errors
results: 9328 ok, 672 errors
results: 9210 ok, 790 errors