@@ -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,16 +517,24 @@ 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+ let stmRead = atomically . readTChan
533+ Mailbox -> [ matchAny return
534+ , matchSTM (readTChan chan) return]
535+ InputChan -> [ matchSTM (readTChan chan) return
537+ in getNextInput' matches 100
530 -- when reading inputs, we generally want to maintain a degree of
539 -- when reading inputs, we generally want to maintain a degree of
531 -- fairness in choosing between the TChan and our mailbox, but to
540 -- fairness in choosing between the TChan and our mailbox, but to
@@ -533,6 +542,8 @@ mxAgentWithFinalize mxId initState handlers dtor = do
533 -- between the two (using the ChannelSelector) each time we call
542 -- between the two (using the ChannelSelector) each time we call
534 -- getNextInput - it returns the opposite ChannelSelector to the
543 -- getNextInput - it returns the opposite ChannelSelector to the
535 -- one which succeeded last time it was called.
544 -- one which succeeded last time it was called.
546+ -----------------------------------------------------------------------
537 -- This strategy works well, yet we wish to avoid blocking on one
548 -- 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
549 -- input if the other is empty/busy, so we begin by reading both
@@ -543,6 +554,8 @@ mxAgentWithFinalize mxId initState handlers dtor = do
543 -- system resoures. Instead, we switch ten times, after which (if no
554 -- 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
555 -- data were obtained) we block on the event bus, since that is our
545 -- main priority.
556 -- main priority.
557+ -----------------------------------------------------------------------
547 -- An agent can of course, choose to override which source should be
560 -- An agent can of course, choose to override which source should be
548 -- checked first. We consider this a /hint/ rather than a dictat.
561 -- checked first. We consider this a /hint/ rather than a dictat.
@@ -553,19 +566,32 @@ mxAgentWithFinalize mxId initState handlers dtor = do
553 -- in which case we needn't worry too much about the overheads
566 -- in which case we needn't worry too much about the overheads
554 -- described thus far.
567 -- described thus far.
569+ -- TODO: use mersenne-random / (System.Random.MWC) to get a
570+ -- uniform distribution of values between 10..2000
572+ getNextInput' ms n = do
573+ mIn <- receiveTimeout n ms
575+ Nothing -> tryNextInput ms n
576+ Just msg -> return msg
579+ | n < 2000 = getNextInput' ms $ n * 2
556− getNextInput' InputChan c' 0 = do
580+ | otherwise = getNextInput' ms 100
582+ -- getNextInput' c' 0 = do
557− m <- liftIO $ atomically $ readTChan c'
583+ -- m <- liftIO $ atomically $ readTChan c'
558− return (m, Mailbox)
584+ -- return (m, Mailbox)
559− getNextInput' InputChan c' n = do
585+ -- getNextInput' InputChan c' n = do
560− inputs <- liftIO $ atomically $ tryReadTChan c'
586+ -- inputs <- liftIO $ atomically $ tryReadTChan c'
561− case inputs of
587+ -- case inputs of
562− Nothing -> getNextInput' Mailbox c' (n - 1)
588+ -- Nothing -> getNextInput' Mailbox c' (n - 1)
563− Just m -> return (m, Mailbox)
589+ -- Just m -> return (m, Mailbox)
564− getNextInput' Mailbox c' n = do
590+ -- getNextInput' Mailbox c' n = do
565− m <- receiveTimeout 0 [ matchAny return ]
591+ -- m <- receiveTimeout 0 [ matchAny return ]
566− case m of
592+ -- case m of
567− Nothing -> getNextInput' InputChan c' (n - 1)
593+ -- Nothing -> getNextInput' InputChan c' (n - 1)
568− Just msg -> return (msg, InputChan)
594+ -- Just msg -> return (msg, InputChan)
570 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
596 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
571 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
597 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst