@@ -289,6 +289,7 @@ import Control.Distributed.Process.Internal.Primitives
289 , receiveTimeout
289 , receiveTimeout
290 , matchChan
290 , matchChan
291 , matchAny
291 , matchAny
292 , unwrapMessage
293 , unwrapMessage
293 , onException
294 , onException
294 , register
295 , register
@@ -505,8 +506,8 @@ mxAgentWithFinalize mxId initState handlers dtor = do
505 -> TChan Message
506 -> TChan Message
506 -> MxAgentState s
507 -> MxAgentState s
507 -> Process ()
508 -> Process ()
508− runAgent eh hs r c s =
509+ runAgent eh hs cs c s =
509− runAgentWithFinalizer eh hs r c s
510+ runAgentWithFinalizer eh hs cs c s
510 `onException` runAgentFinalizer eh s
511 `onException` runAgentFinalizer eh s
512 runAgentWithFinalizer :: MxAgent s ()
513 runAgentWithFinalizer :: MxAgent s ()
@@ -516,56 +517,34 @@ mxAgentWithFinalize mxId initState handlers dtor = do
516 -> MxAgentState s
517 -> MxAgentState s
517 -> Process ()
518 -> Process ()
518 runAgentWithFinalizer eh' hs' cs' c' s' = do
519 runAgentWithFinalizer eh' hs' cs' c' s' = do
519− (msg, selector') <- getNextInput cs' c'
520+ msg <- getNextInput cs' c'
520 (action, state) <- runPipeline msg s' $ pipeline hs'
521 (action, state) <- runPipeline msg s' $ pipeline hs'
521 case action of
522 case action of
522− MxAgentReady -> runAgent eh' hs' selector' c' state
523+ MxAgentReady -> runAgent eh' hs' InputChan c' state
523 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
524 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
524 MxAgentDeactivate _ -> runAgentFinalizer eh' state
525 MxAgentDeactivate _ -> runAgentFinalizer eh' state
525 MxAgentSkip -> error "IllegalState"
526 MxAgentSkip -> error "IllegalState"
526-- MxAgentBecome h' -> runAgent h' c state
527-- MxAgentBecome h' -> runAgent h' c state
528− getNextInput sel chan = getNextInput' sel chan (10 :: Int)
529+ getNextInput sel chan =
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.
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
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.
556− getNextInput' InputChan c' 0 = do
557− m <- liftIO $ atomically $ readTChan c'
530+ let stmRead = atomically . readTChan
558− return (m, Mailbox)
531+ matches =
559− getNextInput' InputChan c' n = do
560− inputs <- liftIO $ atomically $ tryReadTChan c'
561− case inputs of
532+ case sel of
562− Nothing -> getNextInput' Mailbox c' (n - 1)
533+ Mailbox -> [ matchAny return
563− Just m -> return (m, Mailbox)
534+ , matchSTM (readTChan chan) return]
564− getNextInput' Mailbox c' n = do
535+ InputChan -> [ matchSTM (readTChan chan) return
537+ in getNextInput' matches 0
539+ getNextInput' ms n = do
565− m <- receiveTimeout 0 [ matchAny return ]
540+ mIn <- receiveTimeout n ms
566− case m of
541+ case mIn of
567− Nothing -> getNextInput' InputChan c' (n - 1)
542+ Nothing -> tryNextInput ms n
568− Just msg -> return (msg, InputChan)
543+ Just msg -> return msg
546+ | n > 0 && n < 2000 = getNextInput' ms $ n * 2
547+ | otherwise = getNextInput' ms 100
570 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
549 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
571 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
550 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst