@@ -289,6 +289,7 @@ import Control.Distributed.Process.Internal.Primitives
289289 , receiveTimeout
290290 , matchChan
291291 , matchAny
292+ , matchSTM
292293 , unwrapMessage
293294 , onException
294295 , register
@@ -505,8 +506,8 @@ mxAgentWithFinalize mxId initState handlers dtor = do
505506 -> TChan Message
506507 -> MxAgentState s
507508 -> Process ()
508− runAgent eh hs r c s =
509− runAgentWithFinalizer eh hs r c s
509+ runAgent eh hs cs c s =
510+ runAgentWithFinalizer eh hs cs c s
510511 `onException` runAgentFinalizer eh s
511512
512513 runAgentWithFinalizer :: MxAgent s ()
@@ -516,56 +517,34 @@ mxAgentWithFinalize mxId initState handlers dtor = do
516517 -> MxAgentState s
517518 -> Process ()
518519 runAgentWithFinalizer eh' hs' cs' c' s' = do
519− (msg, selector') <- getNextInput cs' c'
520+ msg <- getNextInput cs' c'
520521 (action, state) <- runPipeline msg s' $ pipeline hs'
521522 case action of
522− MxAgentReady -> runAgent eh' hs' selector' c' state
523+ MxAgentReady -> runAgent eh' hs' InputChan c' state
523524 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
524525 MxAgentDeactivate _ -> runAgentFinalizer eh' state
525526 MxAgentSkip -> error "IllegalState"
526527-- MxAgentBecome h' -> runAgent h' c state
527528
528− getNextInput sel chan = getNextInput' sel chan (10 :: Int)
529−
530− -- when reading inputs, we generally want to maintain a degree of
531− -- fairness in choosing between the TChan and our mailbox, but to
532− -- ultimately favour the TChan overall. We do this by flipping
533− -- between the two (using the ChannelSelector) each time we call
534− -- getNextInput - it returns the opposite ChannelSelector to the
535− -- one which succeeded last time it was called.
536− --
537− -- This strategy works well, yet we wish to avoid blocking on one
538− -- input if the other is empty/busy, so we begin by reading both
539− -- sources conditionally - tryReadTChan for the event bus and
540− -- receiveTimeout for the mailbox. If polling (either source) does
541− -- not yield an input, we swap to the other, but we cannot continue
542− -- in this fashion ad infinitum, since that would waste considerable
543− -- system resoures. Instead, we switch ten times, after which (if no
544− -- data were obtained) we block on the event bus, since that is our
545− -- main priority.
546− --
547− -- An agent can of course, choose to override which source should be
548− -- checked first. We consider this a /hint/ rather than a dictat.
549− -- When reading from the mailbox, we perform a non-blocking read.
550− -- We assume that most agents will prefer using mxReceiveChan to its
551− -- mailbox reading counterpart, this we expect the event bus to be
552− -- non-empty or the entire subsystem is likely doing very little work,
553− -- in which case we needn't worry too much about the overheads
554− -- described thus far.
555−
556− getNextInput' InputChan c' 0 = do
557− m <- liftIO $ atomically $ readTChan c'
558− return (m, Mailbox)
559− getNextInput' InputChan c' n = do
560− inputs <- liftIO $ atomically $ tryReadTChan c'
561− case inputs of
562− Nothing -> getNextInput' Mailbox c' (n - 1)
563− Just m -> return (m, Mailbox)
564− getNextInput' Mailbox c' n = do
565− m <- receiveTimeout 0 [ matchAny return ]
566− case m of
567− Nothing -> getNextInput' InputChan c' (n - 1)
568− Just msg -> return (msg, InputChan)
529+ getNextInput sel chan =
530+ let stmRead = atomically . readTChan
531+ matches =
532+ case sel of
533+ Mailbox -> [ matchAny return
534+ , matchSTM (readTChan chan) return]
535+ InputChan -> [ matchSTM (readTChan chan) return
536+ , matchAny return]
537+ in getNextInput' matches 0
538+
539+ getNextInput' ms n = do
540+ mIn <- receiveTimeout n ms
541+ case mIn of
542+ Nothing -> tryNextInput ms n
543+ Just msg -> return msg
544+
545+ tryNextInput ms n
546+ | n > 0 && n < 2000 = getNextInput' ms $ n * 2
547+ | otherwise = getNextInput' ms 100
569548
570549 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
571550 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst