Отмена с будущим и обещанием в Scala

Это продолжение моего предыдущего вопроса.

Предположим, у меня есть задача, которая выполняет прерываемый вызов блокировки. Я хотел бы запустить его как Future и отмените это с failure метод Promise,

Я хотел бы, чтобы отмена работала следующим образом:

  • Если отменить задачу до ее завершения, я бы хотел, чтобы она завершилась "немедленно", прерывая блокирующий вызов, если он уже начался, и мне хотелось бы, чтобы Future вызывать onFailure,

  • Если кто-то отменит задачу после ее завершения, я хотел бы получить статус, в котором говорится, что отмена не удалась, поскольку задача уже завершена.

Имеет ли это смысл? Возможно ли реализовать в Scala? Есть ли примеры таких реализаций?

Источник

4 ответа

Решение

scala.concurrent.Future доступно только для чтения, поэтому один читатель не может все испортить для других.

Кажется, что вы должны быть в состоянии реализовать то, что вы хотите, следующим образом:

def cancellableFuture[T](fun: Future[T] => T)(implicit ex: ExecutionContext): (Future[T], () => Boolean) = {
  val p = Promise[T]()
  val f = p.future
  p tryCompleteWith Future(fun(f))
  (f, () => p.tryFailure(new CancellationException))
}

val (f, cancel) = cancellableFuture( future => {
  while(!future.isCompleted) continueCalculation // isCompleted acts as our interrupted-flag

  result  // when we're done, return some result
})

val wasCancelled = cancel() // cancels the Future (sets its result to be a CancellationException conditionally)
Источник

Вот прерываемая версия кода Виктора согласно его комментариям (Виктор, пожалуйста, поправьте меня, если я неверно истолковал).

object CancellableFuture extends App {

  def interruptableFuture[T](fun: () => T)(implicit ex: ExecutionContext): (Future[T], () => Boolean) = {
    val p = Promise[T]()
    val f = p.future
    val aref = new AtomicReference[Thread](null)
    p tryCompleteWith Future {
      val thread = Thread.currentThread
      aref.synchronized { aref.set(thread) }
      try fun() finally {
        val wasInterrupted = (aref.synchronized { aref getAndSet null }) ne thread
        //Deal with interrupted flag of this thread in desired
      }
    }

    (f, () => {
      aref.synchronized { Option(aref getAndSet null) foreach { _.interrupt() } }
      p.tryFailure(new CancellationException)
    })
  }

  val (f, cancel) = interruptableFuture[Int] { () =>
    val latch = new CountDownLatch(1)

    latch.await(5, TimeUnit.SECONDS)    // Blocks for 5 sec, is interruptable
    println("latch timed out")

    42  // Completed
  }

  f.onFailure { case ex => println(ex.getClass) }
  f.onSuccess { case i => println(i) }

  Thread.sleep(6000)   // Set to less than 5000 to cancel

  val wasCancelled = cancel()

  println("wasCancelled: " + wasCancelled)
}

С Thread.sleep(6000) выход:

latch timed out
42
wasCancelled: false

С Thread.sleep(1000) выход:

wasCancelled: true
class java.util.concurrent.CancellationException
Источник

Фьючерсы Твиттера реализуют отмену. Посмотрите здесь:

https://github.com/twitter/util/blob/master/util-core/src/main/scala/com/twitter/util/Future.scala

Строка 563 показывает абстрактный метод, ответственный за это. Фьючерсы Scala в настоящее время не поддерживают отмену.

Источник

Вы можете использовать библиотеку Monix вместо Future

https://monix.io/

Источник
Другие вопросы по тегам