Cancellable futures?

[hbf]

Dear all,

I am using Scala futures (which, since Scala 2.10 are Akka futures) and have run into a need for cancellation, which Scala futures do not support out-of-the-box.

I have read in a past blog post to this group that Twitter has a futures implementation supporting cancellation through a onCancellation { block } construct and a cancel() method that propagates through map/flatMap.

Is there any extension/source code/snippets out there that "extend" Scala's futures to provide a similar functionality like onCancellation mentioned above?

It can really be "simple". I have a service that takes requests and returns futures holding responses. As I am doing a lot of requests and all of them may get canceled, I need a way to stop the outstanding requests to not completely congest the system.

(I am aware that this is slightly off-topic, but given that Akka developers made the Scala futures, I am hoping that you might have answers, and possibly even code.)

Best,

Kaspar

[√iktor Ҡlang]

IMHO cancellation is a bad practice since it assumes that any reader can cancel something for all other readers.

What I'd do if I were you is to create the Promise, hand it over to a thread that produces the value, and that thread intermittently inspects the Promises' isCompleted to determine whether it should continue producing the value, if not it just bails out. Then anyone who holds the Promise may complete it with a Failed(new CancellationException("Cancelled")), but the guys holding it's Future cannot cancel it.

Cheers,

√

 

Best,

Kaspar

--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ: http://akka.io/faq/
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To post to this group, send email to akka...@googlegroups.com.
To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user?hl=en.
 
 

--
Viktor Klang

Akka Tech Lead

Typesafe - The software stack for applications that scale

Twitter: @viktorklang

[David Ross]

So a Promise can be broken but you can't change the Future. Love the metaphors.

[Jonas Bonér]

LOL.

--
Jonas Bonér
CTO Typesafe - The software stack for applications that scale
Phone: +46 733 777 123
Twitter: @jboner

--

[√iktor Ҡlang]

On Oct 25, 2012 7:14 PM, "David Ross" <dyr...@klout.com> wrote:
>

> So a Promise can be broken but you can't change the Future. Love the metaphors.

Yes!

[hbf]

Hi Viktor,

Thanks for your quick response and for sharing your experience, including the side effects in form of pearls of wisdom ;-)

IMHO cancellation is a bad practice since it assumes that any reader can cancel something for all other readers.

Just to make sure I understand what you mean: With "reader" you are referring to the Future being passed around. So the ability to cancel the computation it encapsulates and the ability to view its result should be considered different things. Correct?

 

What I'd do if I were you is to create the Promise, hand it over to a thread that produces the value, and that thread intermittently inspects the Promises' isCompleted to determine whether it should continue producing the value, if not it just bails out. Then anyone who holds the Promise may complete it with a Failed(new CancellationException("Cancelled")), but the guys holding it's Future cannot cancel it.

The problem I am having with this approach is that – unless I do not understand what you are saying – it does not easily compose (with map/flatMap/etc. constructs I mean).

Here is a concrete example: I have a throttler that delays requests in order to not exceed a given rate. The throttler has a next() method that provides a future that gets completed when the caller may start the task. The request scheduler uses it like this:

  val f = throttler.next() flatMap { _ => request() }

where request() returns a future that performs the actual request and completes with the response. Now schedule 1000 requests. This will fill up the timeline for an hour, say (in the sense that if you scheduled another request, it would have to wait for an hour at least). At this point, you realize that none of the 1000 responses are actually needed and you cancel them all. Observe that in this example, just doing a no-op as the producer does not remove the tasks from the timeline. So still, a new request will wait for an hour. (More technically: merely failing the promise that flatMap uses internally does not propagate the cancellation to the future that the promise is waiting for.)

I tried to come up with a (admittedly very rough) attempt for a CancellableFuture, see https://gist.github.com/2e186d84230dd7bc80d1

It supports the map/flatMap/etc constructs and, in order to address the first point you brought up, it does not extend but wraps an ordinary Scala future; so whenever you want to pass the future to somebody who shall not have the right to abort the computation, you pass the Scala future. I'd be interested in any feedback you may have as I am not an expert with concurrency stuff.

Best,

Kaspar

[Alec Zorab]

What you're doing is a bad fit for futures and a good fit for an actor...

Best,

Kaspar

--

[hbf]

I was asking myself the question, too, to be honest...

So how would you achieve some sort of cancellation with actors then? (I do not need any hard guarantees: if I cancel something, it should preferably cancel.)

[Alec Zorab]

val throttler = actorOf(Props[Throttler])

throttler ! ScheduleTask(taskId1, task)

throttler ! ScheduleTask(taskId2, task)

throttler ! ScheduleTask(taskId3, task)

throttler ! ScheduleTask(taskId4, task)

throttler ! ScheduleTask(taskId5, task)

throttler ! CancelTask(taskId3, task)

[√iktor Ҡlang]

On Fri, Oct 26, 2012 at 11:21 AM, hbf <kaspar....@gmail.com> wrote:

Hi Viktor,

Thanks for your quick response and for sharing your experience, including the side effects in form of pearls of wisdom ;-)

IMHO cancellation is a bad practice since it assumes that any reader can cancel something for all other readers.

Just to make sure I understand what you mean: With "reader" you are referring to the Future being passed around. So the ability to cancel the computation it encapsulates and the ability to view its result should be considered different things. Correct?

Exactly. This is incidentally also what C# tasks do.

 

 

What I'd do if I were you is to create the Promise, hand it over to a thread that produces the value, and that thread intermittently inspects the Promises' isCompleted to determine whether it should continue producing the value, if not it just bails out. Then anyone who holds the Promise may complete it with a Failed(new CancellationException("Cancelled")), but the guys holding it's Future cannot cancel it.

The problem I am having with this approach is that – unless I do not understand what you are saying – it does not easily compose (with map/flatMap/etc. constructs I mean).

(f map identity).cancel() <-- this should also cancel f right?

 

Here is a concrete example: I have a throttler that delays requests in order to not exceed a given rate. The throttler has a next() method that provides a future that gets completed when the caller may start the task. The request scheduler uses it like this:

  val f = throttler.next() flatMap { _ => request() }

where request() returns a future that performs the actual request and completes with the response. Now schedule 1000 requests. 

This will fill up the timeline for an hour, say (in the sense that if you scheduled another request, it would have to wait for an hour at least). At this point, you realize that none of the 1000 responses are actually needed and you cancel them all.

Who is "you"? Why do you schedule things that you might not need in the first place? If you already have a 1000 of them, have a 1000 Promises instead, and complete them all with CancellationException?

 

Observe that in this example, just doing a no-op as the producer does not remove the tasks from the timeline. So still, a new request will wait for an hour. (More technically: merely failing the promise that flatMap uses internally does not propagate the cancellation to the future that the promise is waiting for.)

I tried to come up with a (admittedly very rough) attempt for a CancellableFuture, see https://gist.github.com/2e186d84230dd7bc80d1

It supports the map/flatMap/etc constructs and, in order to address the first point you brought up, it does not extend but wraps an ordinary Scala future; so whenever you want to pass the future to somebody who shall not have the right to abort the computation, you pass the Scala future. I'd be interested in any feedback you may have as I am not an expert with concurrency stuff.

I'd probably just have a Promise instead and have it completeWith the other Future, and then do tryFailure(new CancellationException) to cancel it.

Cheers,

√

 

Best,

Kaspar

--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ: http://akka.io/faq/
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To post to this group, send email to akka...@googlegroups.com.
To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user?hl=en.
 
 

[Alec Zorab]

I've used a similar scheme for managing subscriptions in the past

val token = subscriptionManager.ask(subscribe).mapTo[Cancellable]

...

token.cancel()

Obviously this way requires an appropriate implementation of Cancellable.

On 7 November 2012 15:12, hbf <kaspar....@dreizak.com> wrote:

I have been playing with Viktor's recommendation to use a Promise and intermittently inspect its isCompleted to determine whether the work is completed already or cancelled. So far I have worked with Futures but in order to not have to deal with locking/synchronization issues, I'd like to try this with actors. I am wondering how to do this.

Here are some thoughts:

1. The approach 

val throttler = actorOf(Props[Throttler])

throttler ! ScheduleTask(taskId1, task)

…

throttler ! CancelTask(taskId3, task)

seems okay but it requires keeping track of tasks ids, which gets difficult if you want to compose tasks.

2. What about marking "cancellable" messages with a special trait, `CancellableMessage`, say, as follows: You pass in a Promise, which only ever, if at all, is completed with a failure. It is failed if and only if the request is to be cancelled. Cancellation is on a best-effort basis (there is no guarantee that the request is not processed). Only the creator of the message can pass in the Promise; in this way, only the creator has the power to cancel, or pass the promise to others who may do so as well. 

3. Those among my actors that support cancellation will call 'isCancelled' (which under the hood checks the Promise of the `CancellableMessage`) to learn whether to abort. `CancellableMessage` exposes the Promise's Future so that actors can listen to cancellation by registering an onFailure callback. (Note: I have run into the need to be notified of cancellation several times; for example, a throttler may need to "unschedule" the message, allowing other messages to be scheduled earlier.)

Has anybody had any experiences with an approach along these lines?

One question I have is how this would be made to work in a scenario where some actors are on remote machines.

Cheers,

Kaspar