Download frame­work here.

All posts are here:

• Part I - Workers and ParallelWorkers
• Part II - Agents and con­trol mes­sages
• Part III - Default er­ror man­age­ment
• Part IV - Custom er­ror man­age­ment
• Part V - Timeout man­age­ment
• Part VI - Hot swap­ping of code
• Part VII - An auc­tion frame­work
• Part VIII — Implementing MapReduce (user model)
• Part IX — Counting words …

Timeout man­age­ment

Timeouts are very im­por­tant in mes­sage based sys­tems. In essence, if you are not get­ting mes­sages for a cer­tain pe­riod of time, that usu­ally means some­thing. It might be that some­thing crashed, that other agents think that you are not on­line, or any other num­ber of things. Hence the need to set time­outs and re­act when they are trig­gered.

You do that by writ­ing the fol­low­ing:

counter1 <--SetTimeoutHandler 1000 
            (fun state -> printfn "I'm still waiting for a message in state %A, come on ..." 
                                                            state; ContinueProcessing(state))              

Which gen­er­ates the fol­low­ing mes­sage every sec­ond:

**I’m still wait­ing for a mes­sage in state 2, come on

I’m still wait­ing for a mes­sage in state 2, come on .…**

The first pa­ra­me­ter to SetTimeoutHandler is how long to wait be­fore trig­ger­ing the han­dler. The sec­ond pa­ra­me­ter is the han­dler that gets called when­ever no mes­sage is re­ceived for that amount of time. Notice that the han­dler takes the cur­rent state of the agent and re­turns ContinueProcessing(state).  This tells the agent to con­tinue pro­cess­ing mes­sages and sets the cur­rent state to state.

The fol­low­ing code:

counter1 <-- 2

Then gen­er­ates:

**I’m still wait­ing for a mes­sage in state 4, come on

I’m still wait­ing for a mes­sage in state 4, come on**

ContinueProcessing is just one of the three pos­si­ble val­ues of the (terribly named) AfterError:

type AfterError =
| ContinueProcessing of obj
| StopProcessing
| RestartProcessing

Let’s see what RestartProcessing does.

counter1 <-- SetTimeoutHandler 1000  (fun state -> printfn "Restart from state %A" state
                                                                        ; RestartProcessing)

Which, as ex­pected, gen­er­ates a nice stream of:

**Restart from state 0

Restart from state 0**

To bring things back to nor­mal (aka no time­out) you can just pass –1 as in:

counter1 <-- SetTimeoutHandler -1  (fun state -> ContinueProcessing(state))

Also, you can stop the agent when a time­out oc­curs by re­turn­ing the aptly named StopProcessing:

counter1 <-- SetTimeoutHandler 1000  (fun state -> printfn "Restart from state %A" state; 
                                                                             StopProcessing)

Another in­ter­est­ing thing you might want to do is hot swap­ping of code. More on that in the next part …