CancelableExecutorServiceFuturePool.scala

class CancelableExecutorServiceFuturePool(val executor: ExecutorService) extends FuturePool {
  def apply[T](f: => T): Future[T] = {
    val runOk = new AtomicBoolean(true)
    val p = new Promise[T]
    val task = new Runnable {
      val saved = Local.save()

      def run() {
        // Make an effort to skip work in the case the promise
        // has been cancelled or already defined.
        if (!runOk.compareAndSet(true, false))
          return

        val current = Local.save()
        Local.restore(saved)

        try
          p.update(Try(f))
        finally
          Local.restore(current)
      }
    }


    // This is safe: the only thing that can call task.run() is
    // executor, the only thing that can raise an interrupt is the
    // receiver of this value, which will then be fully initialized.
    val javaFuture = try executor.submit(task) catch {
      case e: RejectedExecutionException =>
        runOk.set(false)
        p.setException(e)
        null
    }

    p.setInterruptHandler {
      case cause =>
        if(javaFuture != null) javaFuture.cancel(true)
        if (runOk.compareAndSet(true, false)) {
          val exc = new CancellationException
          exc.initCause(cause)
          p.setException(exc)
        }
    }

    p
  }
}