Generic reply from agent/mailboxprocessor?

Question

I currently have an agent that does heavy data processing by constantly posting "work" messages to itself.

Sometimes clients to this agent wants to interrupt this processing to access the data in a safe manner.

For this I thought that posting an async to the agent that the agent can run whenever it's in a safe state would be nice. This works fine and the message looks like this:

type Message = |Sync of Async<unit>*AsyncReplyChannel<unit> 

And the agent processing simply becomes:

match mailbox.Receive () with
| Sync (async, reply) -> async |> Async.RunSynchronously |> reply.Reply

This works great as long as clients don't need to return some value from the async as I've constrained the async/reply to be of type unit and I cannot use a generic type in the discriminated union.

My best attempts to solve this has involved wrapper asyncs and waithandles, but this seems messy and not as elegant as I've come to expect from F#. I'm also new to async workflows in F# so it's very possible that I've missed/misunderstood some concepts here.

So the question is; how can I return generic types in a agent response?

Answer 1

The thing that makes this difficult is that, in your current version, the agent would somehow have to calculate the value and then pass it to the channel, without knowing what is the type of the value. Doing that in a statically typed way in F# is tricky.

If you make the message generic, then it will work, but the agent will only be able to handle messages of one type (the type T in Message<T>).

An alternative is to simply pass Async<unit> to the agent and let the caller do the value passing for each specific type. So, you can write message & agent just like this:

type Message = | Sync of Async<unit>

let agent = MailboxProcessor.Start(fun inbox -> async {
  while true do
    let! msg = inbox.Receive ()
    match msg with
    | Sync (work) -> do! work })

When you use PostAndReply, you get access to the reply channel - rather than passing the channel to the agent, you can just use it in the local async block:

let num = agent.PostAndReply(fun chan -> Sync(async { 
  let ret = 42 
  chan.Reply(ret) }))

let str = agent.PostAndReply(fun chan -> Sync(async { 
  let ret = "hi"
  chan.Reply(ret) }))