This code is intended to process commands in a CQRS+Messaging system. It will parallelize the message processing to the degree you specify by the msgToIdFn parameter. The function's intention is to extract the target ID from the command message. This will be used to correlate all messages with that ID to the same agent. Agents are automatically created for each ID as needed and destroyed when their inbox empties.
Right now runFn is expected to execute any message. Likely there will be some match ... with based routing in there to exercise the appropriate module with the appropriate parameters and services.
The comments are verbose because I am learning and also explaining how it works to my team. I can trim them down if too distracting.
I'm looking for possible improvements to code style, missed edge cases, and efficiency. For example, I am returning Some agents in order to continue the main dispatcher loop or None to stop it -- that doesn't seem quite right to me. See other Known Issues at the bottom.
module MessageDispatcher
// the functionality exposed by starting a dispatcher
type MessageDispatcher<'message, 'result> = {
Dispatch:'message -> Async<'result>;
Shutdown: unit -> unit
}
// the things that the agents can be instructed to do
type private AgentMessage<'message> =
| Run of 'message
| Stop
// the things that the dispatcher can be instructed to do
type private DispatcherMessage<'key, 'message when 'key : comparison> =
| Dispatch of 'message
| CompleteRun of 'key
| Shutdown
// it made sense to make MessageCount mutable
// otherwise, code would have to create a new record, remove old from agentsMap, add new to agentMap... every time
type private AgentCounter<'message, 'result> = {
mutable MessageCount: int;
Agent: MailboxProcessor<AgentMessage<'message> * ('result -> unit)>
}
/// Start the dispatcher
/// runFn is how you plug in execution code
/// signature: 'message -> 'result
/// msgToIdFn extracts the id from the message, so it can be linked to a specific agent
/// signature: 'message -> 'key
/// sysFailFn wraps an unhandled exception in the caller's desired result type
/// signature: exn -> 'result
let Start<'key, 'message, 'result when 'key : comparison> (runFn:'message -> 'result) (msgToIdFn:'message -> 'key) (sysFailFn: exn -> 'result) =
// function to create a new agent
let createAgent runCompleted =
// create a start a new mailbox processor
MailboxProcessor.Start
<| fun inbox -> // inbox is the same as the agent itself
let rec loop () = async { // define the async message processing loop
let! (message, reply) = inbox.Receive() // wait on next message
match message with
| Stop -> return () // we are done, exit the loop
| Run msg ->
try // in case there are any unhandled exceptions
let result = runFn msg // run message
reply(result) // no crash returns normal result
with
| ex -> reply(sysFailFn ex) // system failure of some kind
runCompleted() // notify completion of this message
return! loop () // continue to run
}
loop () // start the message processing loop
// this is the dispatcher through which all messages flow
let dispatchAgent =
// create and start a new mailbox
MailboxProcessor.Start
<| fun inbox -> // inbox is the same as dispatchAgent
// this function stops and removes an agent from the map
// returns a copy of agentMap with the agent removed
let decommission staleId (staleAgent:MailboxProcessor<_>) agentMap =
// stop agent
staleAgent.Post(Stop, ignore)
// remove from map
Map.remove staleId agentMap
// this function queues a message to the agent
// returns a copy of the (possibly) modified agentMap
let queueMessage msg reply agentMap =
// run the msgToKey function to get the id out of the message
let id = msgToIdFn msg
// return the agent, and the agent map in case it was changed
let (agent, agents) =
// see if the agent already exists
match Map.tryFind id agentMap with
| Some agentCounter -> // existing agent
agentCounter.MessageCount <- agentCounter.MessageCount + 1 // increase count
(agentCounter.Agent, agentMap) // return the agent and original map
| None -> // new agent
// create agent
let runCompletedFn () = inbox.Post(CompleteRun id, ignore) // create func to run when messages complete
let agent = createAgent runCompletedFn // create the agent
let agentCounter = { MessageCount = 1; Agent = agent } // create the agentCounter record
// return agent // return agentmap with agent added
(agentCounter.Agent, Map.add id agentCounter agentMap)
agent.Post(Run msg, reply) // send the message to the agent
Some agents // Some is roughly equivalent to not null with this value
// this function updates counters when an agent completes a message
// it will also decommission an agent that has no messages
// returns agentMap, it will be changed if an agent was decommissioned
let completeMessage id agentMap =
let agentCounter = Map.find id agentMap // get the agentCounter
agentCounter.MessageCount <- agentCounter.MessageCount - 1 // update the counter, <- is mutable style
let agents =
match agentCounter.MessageCount with
| 0 -> decommission id agentCounter.Agent agentMap // immediately decommission
| _ -> agentMap // return the same agent map -- nothing changed
Some agents
// this function will stop all current agents and then exit
// return None to tell the dispatcher to stop running
let shutdown agentMap =
agentMap
|> Map.iter (fun id agentCounter -> agentCounter.Agent.Post(Stop, ignore))
None
// this is the core loop to dispatch messages
let rec loop agentMap = async { // run async
// let! and return! here is kinda like C# await
let! (message, reply) = inbox.Receive() // receive the next message and reply channel
let optionalAgents =
match message with // run the appropriate function
| Dispatch msg -> queueMessage msg reply agentMap
| CompleteRun id -> completeMessage id agentMap
| Shutdown -> shutdown agentMap
match optionalAgents with
| Some agents -> return! loop agents // agents provided means continue to run
| None -> return () // no agents provided means exit the loop
}
loop Map.empty<'key, AgentCounter<'message, 'result>> // start the loop with an empty map
// this sends a dispatch message to the dispatcher agent
let dispatchFn (msg:'message) =
let message = Dispatch msg // create a dispatch message from the given message
// send the message and reply channel
dispatchAgent.PostAndAsyncReply(fun replyChannel -> (message, replyChannel.Reply))
// the reply channel allows callers to get a response back
// this sends the shutdown message to the dispatcher agent
let shutdownFn () = dispatchAgent.Post(Shutdown, ignore)
// this exposes the public functions that this dispatcher supports
let dispatcher = {Dispatch = dispatchFn; Shutdown = shutdownFn}
// return the dispatcher
dispatcher
Known Issues
- Immutable collection performance
- Already tested vs mutable with various usage patterns.
- Worst pattern was around 90k msgs/s vs mutable's 140k
- Best pattern was around 280k msgs/s vs mutable's 290k
- Considering .NET Web API will be doing good to get around 10k HTTP requests/sec, and our likely system throughput will be far less, I'm not worried about it being a bottleneck. If it becomes one, I can always switch to a mutable collection later.
- Caching agents for reuse instead of immediately decommissioning
- I tested this. Surprisingly, it didn't make much difference and created several more edge cases.
- In my tests it's only faster when using mutable collections, and then only marginally so (vs mutable non-cached) except for limited (and possibly unlikely) usage patterns.
Addendum
I have a more updated version of this. I experimented with returning an interface as well, but I didn't care for it. So I converted it into a more f-sharp-y representation using module functions for all operations. It is using mutable collections for maximum performance, however.
