Back to Questions
5.6k
views6
comments
Constrain the async computation by timeout
f#workflowsasyncasynchronous

Hi,

I'm writing a function which is supposed to constrain the async computation by a timeout:

1
2
3

    let withTimeout (timeout : TimeSpan) (computation : 'a Async) : 'a Async =
        let timeoutExpired = throw timeout (new TimeoutException ("Timeout expired"))
        [| computation; timeoutExpired |] |> choice

where the throw is:

1
2
3

    let throw (timeout : TimeSpan) (ex : Exception) : 'a Async = async {
        do! Async.Sleep (int timeout.TotalMilliseconds)
        return raise ex }

and the choice is:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

    let choice (computations : 'a Async array) : 'a Async =
        let invokeOnce funcs =
            let counter = ref 0
            let invokeOnce' f x =
                if (Interlocked.CompareExchange (counter, 1, 0) = 0) then
                    f x
            let (a, b, c) = funcs
            (invokeOnce' a, invokeOnce' b, invokeOnce' c)

        let callback (success, error, cancellation) =
            let (success, error, cancellation) = invokeOnce (success, error, cancellation)
            let cts = new CancellationTokenSource()
            let fetchResult (computation : 'a Async) : unit Async = async {
                try
                    let! result = computation
                    cts.Cancel ()
                    success result
                with
                    | :? OperationCanceledException as ex -> cancellation ex
                    | ex -> error ex }
        
            let startComputation x = Async.StartImmediate ((fetchResult x), cts.Token)
            Array.map startComputation computations |> ignore
    
        Async.FromContinuations callback

The way I expect it to work is if the wrapped async completes within the specified timeout the exception would not propagate further, but it seems to be not the case.
Here is how I test it:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

    let defer (timeout : TimeSpan) (value : 'a) : 'a Async = async {
        do! Async.Sleep (int timeout.TotalMilliseconds)
        return value }

    let test () =
        let x = "x" |> defer (TimeSpan.FromSeconds 4.) |> withTimeout (TimeSpan.FromSeconds 3.)
        let y = "y" |> defer (TimeSpan.FromSeconds 1.)
        let z = "z" |> defer (TimeSpan.FromSeconds 2.)
        let all = choice [| x; y; z |]
        async {
            try     
                let! r = all
                printfn "got %s" r
            with
                | ex -> printfn "exception %s" (ex.ToString()) }
        |> Async.RunSynchronously

If I run the test in FSI it works fine, but when run as a console application I get a complaint on an unhandled exception (the timeout one).
It seems like the choice function propagates not only the first completed computation or something else is going wrong.

Could someone suggest how to improve it so it gets the job done?
Also I'm not sure if I need to catch exceptions in fetchResult.

.

This kind of code sure does make my brain hurt. I would try tracing more to figure out what's the problem. One idea that comes to mind is `Async.StartImmediate`, does it not confine your code to run on a single thread, whereas you expect it to run in parallel?

If I had to do this I would try to be lazy and use existing API, for ex:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

    open System.Threading.Tasks

    let withTimeout (timeout : TimeSpan) (computation : Async<'T>) =
        async {
            let! t = Async.StartChild(computation, int timeout.TotalMilliseconds)
            return! t
        }

    let choice (cs: seq<Async<'T>>) =
        async {
            let tasks : Task<Task<'T>> = Task.WhenAny (Seq.map Async.StartAsTask cs)
            let! t = Async.AwaitTask(tasks)
            return t.Result
        }
By on 9/19/2012 5:43 AM ()

What exactly in this code makes your brain hurt?
This is a genuine question from someone who started to use the language recently.

Yes, I do start the computations on a single thread, but it doesn't imply they are going to execute on it. Using Async.StartChild seems to be suboptimal to me - why would you bother the thread pool just for starting async computations? Take Async.Sleep or WebRequest.GetResponseAsync as examples of non blocking computations.

I'm aware of the Task.WhenAny method, but my intent was to reimplement it using the async infrastructure. I'm a bit surprised it doesn't exist in the standard library.

By on 9/21/2012 6:22 AM ()

> What exactly in this code makes your brain hurt?

Attempting to verify it. Multi-threaded code is difficult to reason about. Is it not? If it was easy, I bet you would not be asking this question.

> Yes, I do start the computations on a single thread, but it doesn't imply they are going to execute on it.

Have you tested this? You are assuming that Thread.Sleep will switch threads, which is not unreasonable to infer from its MSDN documentation stating that it will use a Timer object, not block any threads, and therefore execute the continuation on a ThreadPool thread. But then with this example I get all computations execute on the same thread:

1
2
3
4
5
6
7
8
9

let test () =
    for i in 1 .. 3 do
        async {
            do stdout.WriteLine("{1} running on {0}", Thread.CurrentThread.ManagedThreadId, i)
            do! Async.Sleep(1000)
            do stdout.WriteLine("{1} running on {0}", Thread.CurrentThread.ManagedThreadId, i)
            return ()
        }
        |> Async.StartImmediate

Note that the threads are actually different when Async.Start is used. Perhaps Async.Sleep behaves differently in conjunction with Async.StartImmediate as an optimization? For the code above it does not create a problem, but maybe it does for yours.

And for what it's worth, your original code runs OK for me as a console application.

By on 9/24/2012 6:01 AM ()

I'll retest it once I get to the VS.
Maybe my repl session didn't reset when I tested it.
Thanks for you time!

By on 9/24/2012 7:32 AM ()

This one

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

    let withTimeout (timeout : TimeSpan) (computation : 'a Async) : 'a Async =
        let callback (success, error, cancellation) =
            let (success, error, cancellation) = invokeOnce (success, error, cancellation)

            let rec timer = new Timer(new TimerCallback(timeoutExpired))
            and timeoutExpired _ =
                let ex = new TimeoutException ("Timeout expired") :> Exception
                timer.Dispose ()
                error ex
            and fetchResult (computation : 'a Async) : unit Async = async {
                let! result = computation
                timer.Dispose ()
                success result }

            let _ = timer.Change (timeout, TimeSpan.FromMilliseconds (float Timeout.Infinite))
            Async.StartImmediate (fetchResult computation)
            ()

        Async.FromContinuations callback

seems to work as expected.

By on 8/30/2012 7:12 PM ()

Actually there is a more elegant version than the one with explicit timer management:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

    let withTimeout (timeout : TimeSpan) (computation : 'a Async) : 'a Async =
        let callback (success, error, cancellation) =
            let (success, error, cancellation) = invokeOnce (success, error, cancellation)
            let fetchResult = async {
                let! result = computation
                success result }
            let timeoutExpired = async {
                do! Async.Sleep (int timeout.TotalMilliseconds)
                let ex = new TimeoutException ("Timeout expired") :> Exception
                error ex }

            Async.StartImmediate fetchResult
            Async.StartImmediate timeoutExpired

        Async.FromContinuations callback

Still can't figure why the first one doesn't work.

By on 8/30/2012 7:47 PM ()

    Topic tags

    IntelliFactory Offices Copyright (c) 2011-2012 IntelliFactory. All rights reserved.
    Home | Products | Consulting | Trainings | Blogs | Jobs | Contact Us | Terms of Use | Privacy Policy | Cookie Policy     		Built with WebSharper