Back to Questions
1.8k
views6
comments
Exception: "multiple waiting reader continuations for mailbox" - reprise
f#async

I posted a problem getting an exception (above) when attempting to use a MailboxProcessor for a PostAndReply message. I've now reduced the problem to a (hopefully simple) NUnit test that exhibits the behavior most of the time. When run under NUnit, I generally get the exception after the "Prepare to Call" message before issuing the reply. I also get the error when the Console writes are removed.

Can someone see what I'm doing wrong?

Thanks much,

--Don

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

 
module messageReplyTest
open System
open System.Threading
open System.Diagnostics
open Microsoft.FSharp.Control
open NUnit.Framework
open NUnit.Framework.Constraints

type PipelineTestMsg = 
     | CallMsg of   string AsyncReplyChannel
     | Empty 

type TestPipelineHandler (runPipeline : IDisposable -> string ) =
    let rec runPipe =
        MailboxProcessor.Start(fun inbox ->
            let rec waitForMsg () =
                    async {let! msg = inbox.Receive()                           
                           let! _ = 
                                    async{
                                       match msg with
                                        | CallMsg (replyChannel) ->
                                               replyChannel.Reply("strResult")
                                        | _ -> failwith "Invalid Message"
                                       return! waitForMsg()
                                         } |> Async.StartChild
                           return! waitForMsg() 
                           }
            waitForMsg ())
    member x.ProcessPageWithReply () =
//            Console.WriteLine "Prepare to Call"
            let res = runPipe.PostAndReply
                        (fun replyChannel -> CallMsg(replyChannel))
//            Console.WriteLine "Returned from invocation"
            ()

[<TestFixture>]
type Tests() = 

  [<Test>]
  member x.TestMessageReply () =
    let foo (x : IDisposable) =
        "result"
    let handler = new TestPipelineHandler(foo)
    let bar = handler.ProcessPageWithReply ()
    Console.WriteLine "Wait for exception"
    Thread.Sleep 3000
    Console.WriteLine "Return from Sleep"
.

Some further clarification.
The exception you probably see right now (in latest released F#, be it a latest Oct 2009 CTP or VS2010 Beta2) is a manifestation of a bug in Async.StartChild, since fixed.
On our current internal bits you will see an exception which gives you much more clue as to what’s going on:

1
2

> System.Exception: multiple waiting reader continuations for mailbox
   at <StartupCode$FSharp-Core>.$Control.-ctor@1856-4.Invoke(AsyncParams`1 _arg10)


And here is the reason for that: as others noted on this thread,

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

let rec waitForMsg () =
        async {let! msg = inbox.Receive()                          
               let! _ =
                        async{
                           match msg with
                            | CallMsg (replyChannel) ->
                                   replyChannel.Reply("strResult")
                            | _ -> failwith "Invalid Message"
                           return! waitForMsg()
                             } |> Async.StartChild
               return! waitForMsg()
               }


you call waitForMsg() (i.e. inbox.Receive()) twice simultaneously in two different threads. This is not allowed – access to mailbox should be serialized in our implementation.

If you want to parallelize your replies, by all means do so but just keep waiting for messages on a single thread:

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

let rec waitForMsg () =
        async {let! msg = inbox.Receive()                           
               let! _ =
                        async{
                           match msg with
                            | CallMsg (replyChannel) ->
                                   replyChannel.Reply("strResult")
                            | _ -> failwith "Invalid Message"
                           return()
                        } |> Async.StartChild
               return! waitForMsg()
               }


Incidentally, using Async.StartChild only make sense if you actually want to wait on child completion, because Async.StartChild returns the wait async.
Since you are ignoring the return value, you will be better off just using an Async.Start:

1
2
3
4
5
6
7
8
9
10
11

let rec waitForMsg () =
        async { let! msg = inbox.Receive()                          
                async{
                   match msg with
                    | CallMsg (replyChannel) ->
                           replyChannel.Reply("strResult")
                    | _ -> failwith "Invalid Message"
                   return()
                } |> Async.Start
                return! waitForMsg()
              }



This also has the advantage of working in released F# versions J
Feel free to follow-up offline at dmitry(dot)lomov(at)microsoft(dot)com

By on 1/22/2010 5:54 PM ()

Oops I jumped the gun a bit - you are not hitting the bug in Async.StartChild, you are hitting the genuine diagnostics - and you say it right there in the title :)

The rest of my comment stands.

By on 1/23/2010 1:59 AM ()

The spawn child looked fishy to me, and I think it was causing the problem. The ``let!'' should rotate to an available thread.

The code below will always get the exception on the second loop for me. But, removing the spawn child doesn't crash.

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

module messageReplyTest
open System
open System.Threading
open System.Diagnostics
open Microsoft.FSharp.Control

type PipelineTestMsg = 
     | CallMsg of   string AsyncReplyChannel
     | Empty 

type TestPipelineHandler1 (runPipeline : IDisposable -> string ) =
    let rec runPipe =
        MailboxProcessor.Start(fun inbox ->
            let rec waitForMsg () =
                    async {let! msg = inbox.Receive()                           
                           let! _ = 
                                    async{
                                       match msg with
                                        | CallMsg (replyChannel) ->
                                               replyChannel.Reply("strResult")
                                        | _ -> failwith "Invalid Message"
                                       return! waitForMsg()
                                         } |> Async.StartChild
                           return! waitForMsg() 
                           }
            waitForMsg ())
    member x.ProcessPageWithReply () =
//            Console.WriteLine "Prepare to Call"
            let res = runPipe.PostAndReply
                        (fun replyChannel -> CallMsg(replyChannel))
//            Console.WriteLine "Returned from invocation"
            ()

type TestPipelineHandler2 (runPipeline : IDisposable -> string ) =
    let rec runPipe =
        MailboxProcessor.Start(fun inbox ->
            let rec waitForMsg () =
                    async {let! msg = inbox.Receive()                           
                           let! _ = 
                                    async{
                                       match msg with
                                        | CallMsg (replyChannel) ->
                                               replyChannel.Reply("strResult")
                                        | _ -> failwith "Invalid Message"
                                       return! waitForMsg()
                                         }
                           return! waitForMsg() 
                           }
            waitForMsg ())
    member x.ProcessPageWithReply () =
//            Console.WriteLine "Prepare to Call"
            let res = runPipe.PostAndReply
                        (fun replyChannel -> CallMsg(replyChannel))
//            Console.WriteLine "Returned from invocation"
            ()


while true do
    let foo (x : IDisposable) =
        "result"
    let handler = new TestPipelineHandler2(foo)
    let bar = handler.ProcessPageWithReply ()
    Console.WriteLine "Wait for exception"
    Thread.Sleep 3000
    Console.WriteLine "Return from Sleep"

I don't know if it was from removing the irrevant parts. But, you could reduce runpipe to something like this:

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

 

    let rec runPipe () =
        MailboxProcessor.Start(fun inbox ->
            let rec waitForMsg () =
                async {
                    let! msg = inbox.Receive()                           
                    match msg with
                    | CallMsg (replyChannel) -> replyChannel.Reply("strResult")
                    return! waitForMsg()
                }
            waitForMsg ())
By on 1/14/2010 12:24 PM ()

Thanks, good observation.

The extra async you saw wrapped a semaphore access which I removed in building the test case. I'll take another look and see if it is really necessary. At the time, I thought it was, but maybe I was having a brain freeze.

I'm still not clear on why my code doesn't work, but right now, getting something working is a higher priority.

Thx,

--Don

By on 1/14/2010 12:44 PM ()

The extra async is fine, if it is necessary. It looks like the Async.StartChild is forking the execution. (So, if you put a break point at let! _ before you have 1 worker thread running, after you have 2.

2 threads is not bad in itself. But, both execute waitForMsg(). Then you have two threads simultaneously accessing let! msg = inbox.Receive() and you get the exception.

So removing the Async.StartChild, you get only one path through the let! _ code. I don't think the trailing return! in the outer async is ever hit without the startchild.

By on 1/14/2010 1:51 PM ()

Very good - thanks.

--Don

By on 1/14/2010 4:46 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