Java Future has cancel method, which can interrupt the thread, which runs the Future task. For example, if I wrap an interruptible blocking call in a Java Future I can interrupt it later.
Scala Future provides no cancel method. Suppose I wrap an interruptible blocking call in a Scala Future. How can I interrupt it?
This is not yet a part of the Futures API, but may be added as an extension in the future.
As a workaround, you could use the firstCompletedOf to wrap 2 futures - the future you want to cancel and a future that comes from a custom Promise. You could then cancel the thus created future by failing the promise:
def cancellable[T](f: Future[T])(customCode: => Unit): (() => Unit, Future[T]) = {
val p = Promise[T]
val first = Future firstCompletedOf Seq(p.future, f)
val cancellation: () => Unit = {
() =>
first onFailure { case e => customCode}
p failure new Exception
}
(cancellation, first)
}
Now you can call this on any future to obtain a "cancellable wrapper". Example use-case:
val f = callReturningAFuture()
val (cancel, f1) = cancellable(f) {
cancelTheCallReturningAFuture()
}
// somewhere else in code
if (condition) cancel() else println(Await.result(f1))
EDIT:
For a detailed discussion on cancellation, see Chapter 4 in the Learning concurrent programming in Scala book.
customCode above. –
Lineament cancel ? –
Customer cancel and the time when customCode actually stops the body of the future (say by example that customCode sets a boolean flag that is checked by the future's body to know whether to abort), anything might happen. In particular the future's body might start to execute. The end result: while the future returned by cancellable says it was cancelled, the futures's body actually executed. That's a real problem as soon as the future's body performs any side effect. This makes using your code actually subtly dangerous. –
Elevated condition evaluated to true) and the moment you actually interrupt the thread, it might be too late and the original future's body might already have executed. –
Elevated Future.cancel is not a good idea due to the reasons discussed; another problem is that a Future is a shared read-only handle and as such it should not provide methods which interfere with other readers. What you can do is to pass a Future into the code which you want to run and have that code check the Future periodically, then you have a principled way of interrupting the computation by completing the corresponding Promise. –
Coahuila I haven't tested this, but this expands on the answer of Pablo Francisco Pérez Hidalgo. Instead of blocking waiting for the java Future, we use an intermediate Promise instead.
import java.util.concurrent.{Callable, FutureTask}
import scala.concurrent.{ExecutionContext, Promise}
import scala.util.Try
class Cancellable[T](executionContext: ExecutionContext, todo: => T) {
private val promise = Promise[T]()
def future = promise.future
private val jf: FutureTask[T] = new FutureTask[T](
new Callable[T] {
override def call(): T = todo
}
) {
override def done() = promise.complete(Try(get()))
}
def cancel(): Unit = jf.cancel(true)
executionContext.execute(jf)
}
object Cancellable {
def apply[T](todo: => T)(implicit executionContext: ExecutionContext): Cancellable[T] =
new Cancellable[T](executionContext, todo)
}
By cancelling I guess you would like to violently interrupt the future.
Found this segment of code: https://gist.github.com/viktorklang/5409467
Did a few tests and seems to work fine!
Enjoy :)
I think it is possible to reduce the complexity of the implementations provided by making use of the Java 7 Future interface and its implementations.
Cancellable can build a Java future which is the one to be cancelled by its cancel method. Another future can wait for its completion thus becoming the observable interface which is itself immutable in state:
class Cancellable[T](executionContext: ExecutionContext, todo: => T) {
private val jf: FutureTask[T] = new FutureTask[T](
new Callable[T] {
override def call(): T = todo
}
)
executionContext.execute(jf)
implicit val _: ExecutionContext = executionContext
val future: Future[T] = Future {
jf.get
}
def cancel(): Unit = jf.cancel(true)
}
object Cancellable {
def apply[T](todo: => T)(implicit executionContext: ExecutionContext): Cancellable[T] =
new Cancellable[T](executionContext, todo)
}
jf.get call should be wrapped into scala.concurrent.blocking block for good measure. –
Carley jf.get is being called within a block passed to a Future constructor, IMHO there is no danger from potential blocking indeed, is it? –
Rodl blocking) seems it might apply. I'll check it out, thanks for your suggestion :) –
Rodl blocking hint so that underlying thread pool temporarily increases the number of workers. You're welcome! –
Carley ExecutionContext which provides a whole new thread for each execution: class ProlificExecutor extends Executor { override def execute(command: Runnable): Unit = new Thread(command) start } –
Rodl Promise and complete it from FutureTask.done()... –
Carley © 2022 - 2024 — McMap. All rights reserved.