aivika-distributed-0.8: Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs
{-# LANGUAGE TypeFamilies, FlexibleInstances, OverlappingInstances #-}
-- |
-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Event
-- Copyright : Copyright (c) 2015-2017, David Sorokin <david.sorokin@gmail.com>
-- License : BSD3
-- Maintainer : David Sorokin <david.sorokin@gmail.com>
-- Stability : experimental
-- Tested with: GHC 7.10.3
--
-- The module defines an event queue.
--
module Simulation.Aivika.Distributed.Optimistic.Internal.Event
(queueInputMessages,
queueOutputMessages,
queueLog,
expectEvent,
processMonitorSignal) where
import Data.Maybe
import Data.IORef
import Data.Time.Clock
import System.Timeout
import Control.Monad
import Control.Monad.Trans
import Control.Exception
import qualified Control.Distributed.Process as DP
import qualified Simulation.Aivika.PriorityQueue.Pure as PQ
import Simulation.Aivika.Trans
import Simulation.Aivika.Trans.Internal.Types
import Simulation.Aivika.Trans.Internal.Event
import Simulation.Aivika.Trans.Internal.Cont
import Simulation.Aivika.Trans.Internal.Process
import Simulation.Aivika.Distributed.Optimistic.Internal.Priority
import Simulation.Aivika.Distributed.Optimistic.Internal.Channel
import Simulation.Aivika.Distributed.Optimistic.Internal.DIO
import Simulation.Aivika.Distributed.Optimistic.Internal.IO
import Simulation.Aivika.Distributed.Optimistic.Internal.Message
import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer
import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp
import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.SignalHelper
import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue
import {-# SOURCE #-} Simulation.Aivika.Distributed.Optimistic.Internal.OutputMessageQueue
import Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue
import Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog
import {-# SOURCE #-} qualified Simulation.Aivika.Distributed.Optimistic.Internal.Ref.Strict as R
import Simulation.Aivika.Distributed.Optimistic.State
-- | Convert microseconds to seconds.
microsecondsToSeconds :: Int -> Rational
microsecondsToSeconds x = (fromInteger $ toInteger x) / 1000000
-- | An implementation of the 'EventQueueing' type class.
instance EventQueueing DIO where
-- | The event queue type.
data EventQueue DIO =
EventQueue { queueInputMessages :: InputMessageQueue,
-- ^ the input message queue
queueOutputMessages :: OutputMessageQueue,
-- ^ the output message queue
queueTransientMessages :: TransientMessageQueue,
-- ^ the transient message queue
queueLog :: UndoableLog,
-- ^ the undoable log of operations
queuePQ :: R.Ref (PQ.PriorityQueue (Point DIO -> DIO ())),
-- ^ the underlying priority queue
queueBusy :: IORef Bool,
-- ^ whether the queue is currently processing events
queueTime :: IORef Double,
-- ^ the actual time of the event queue
queueGlobalTime :: IORef Double,
-- ^ the global time
queueInFind :: IORef Bool,
-- ^ whether the queue is in find mode
queueProcessMonitorNotificationSource :: SignalSource DIO DP.ProcessMonitorNotification
-- ^ the source of process monitor notifications
}
newEventQueue specs =
do f <- liftIOUnsafe $ newIORef False
t <- liftIOUnsafe $ newIORef $ spcStartTime specs
gt <- liftIOUnsafe $ newIORef $ spcStartTime specs
pq <- R.newRef0 PQ.emptyQueue
log <- newUndoableLog
transient <- newTransientMessageQueue
output <- newOutputMessageQueue $ enqueueTransientMessage transient
input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime
infind <- liftIOUnsafe $ newIORef False
s <- newDIOSignalSource0
return EventQueue { queueInputMessages = input,
queueOutputMessages = output,
queueTransientMessages = transient,
queueLog = log,
queuePQ = pq,
queueBusy = f,
queueTime = t,
queueGlobalTime = gt,
queueInFind = infind,
queueProcessMonitorNotificationSource = s }
enqueueEvent t (Event m) =
Event $ \p ->
let pq = queuePQ $ runEventQueue $ pointRun p
in invokeEvent p $
R.modifyRef pq $ \x -> PQ.enqueue x t m
runEventWith processing (Event e) =
Dynamics $ \p ->
do p0 <- invokeEvent p currentEventPoint
invokeEvent p0 $ enqueueEvent (pointTime p) (return ())
invokeEvent p $ syncEvents processing
e p
eventQueueCount =
Event $ \p ->
let pq = queuePQ $ runEventQueue $ pointRun p
in invokeEvent p $
fmap PQ.queueCount $ R.readRef pq
-- | The first stage of rolling the changes back.
rollbackEventPre :: Bool -> TimeWarp DIO ()
rollbackEventPre including =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
rollbackLog (queueLog q) (pointTime p) including
-- | The post stage of rolling the changes back.
rollbackEventPost :: Bool -> TimeWarp DIO ()
rollbackEventPost including =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
rollbackOutputMessages (queueOutputMessages q) (pointTime p) including
-- | Rollback the event time.
rollbackEventTime :: TimeWarp DIO ()
rollbackEventTime =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
t = pointTime p
---
--- logDIO DEBUG $
--- "Setting the queue time = " ++ show t
---
liftIOUnsafe $ writeIORef (queueTime q) t
t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)
when (t0 > t) $
do ---
--- logDIO DEBUG $
--- "Setting the global time = " ++ show t
---
liftIOUnsafe $ writeIORef (queueGlobalTime q) t
-- | Return the current event time.
currentEventTime :: Event DIO Double
{-# INLINE currentEventTime #-}
currentEventTime =
Event $ \p ->
do let q = runEventQueue $ pointRun p
liftIOUnsafe $ readIORef (queueTime q)
-- | Return the current event point.
currentEventPoint :: Event DIO (Point DIO)
{-# INLINE currentEventPoint #-}
currentEventPoint =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t' <- liftIOUnsafe $ readIORef (queueTime q)
if t' == pointTime p
then return p
else let sc = pointSpecs p
t0 = spcStartTime sc
dt = spcDT sc
n' = fromIntegral $ floor ((t' - t0) / dt)
in return p { pointTime = t',
pointIteration = n',
pointPhase = -1 }
-- | Process the pending events.
processPendingEventsCore :: Bool -> Dynamics DIO ()
processPendingEventsCore includingCurrentEvents = Dynamics r where
r p =
do let q = runEventQueue $ pointRun p
f = queueBusy q
f' <- liftIOUnsafe $ readIORef f
if f'
then error $
"Detected an event loop, which may indicate to " ++
"a logical error in the model: processPendingEventsCore"
else do liftIOUnsafe $ writeIORef f True
call q p p
liftIOUnsafe $ writeIORef f False
call q p p0 =
do let pq = queuePQ q
r = pointRun p
-- process external messages
p1 <- invokeEvent p0 currentEventPoint
ok <- invokeEvent p1 $ runTimeWarp processChannelMessages
if not ok
then call q p p1
else do -- proceed with processing the events
f <- invokeEvent p1 $ fmap PQ.queueNull $ R.readRef pq
unless f $
do (t2, c2) <- invokeEvent p1 $ fmap PQ.queueFront $ R.readRef pq
let t = queueTime q
t' <- liftIOUnsafe $ readIORef t
when (t2 < t') $
-- error "The time value is too small: processPendingEventsCore"
error $
"The time value is too small (" ++ show t2 ++
" < " ++ show t' ++ "): processPendingEventsCore"
when ((t2 < pointTime p) ||
(includingCurrentEvents && (t2 == pointTime p))) $
do let sc = pointSpecs p
t0 = spcStartTime sc
dt = spcDT sc
n2 = fromIntegral $ floor ((t2 - t0) / dt)
p2 = p { pointTime = t2,
pointIteration = n2,
pointPhase = -1 }
---
--- ps <- dioParams
--- when (dioLoggingPriority ps <= DEBUG) $
--- invokeEvent p2 $
--- writeLog (queueLog q) $
--- logDIO DEBUG $
--- "Reverting the queue time " ++ show t2 ++ " --> " ++ show t'
---
liftIOUnsafe $ writeIORef t t2
invokeEvent p2 $ R.modifyRef pq PQ.dequeue
catchComp
(c2 p2)
(\e@(SimulationRetry _) -> invokeEvent p2 $ handleEventRetry e)
call q p p2
-- | Process the pending events synchronously, i.e. without past.
processPendingEvents :: Bool -> Dynamics DIO ()
processPendingEvents includingCurrentEvents = Dynamics r where
r p =
do let q = runEventQueue $ pointRun p
t = queueTime q
t' <- liftIOUnsafe $ readIORef t
if pointTime p < t'
then error $
"The current time is less than " ++
"the time in the queue: processPendingEvents"
else invokeDynamics p m
m = processPendingEventsCore includingCurrentEvents
-- | A memoized value.
processEventsIncludingCurrent :: Dynamics DIO ()
processEventsIncludingCurrent = processPendingEvents True
-- | A memoized value.
processEventsIncludingEarlier :: Dynamics DIO ()
processEventsIncludingEarlier = processPendingEvents False
-- | A memoized value.
processEventsIncludingCurrentCore :: Dynamics DIO ()
processEventsIncludingCurrentCore = processPendingEventsCore True
-- | A memoized value.
processEventsIncludingEarlierCore :: Dynamics DIO ()
processEventsIncludingEarlierCore = processPendingEventsCore True
-- | Process the events.
processEvents :: EventProcessing -> Dynamics DIO ()
processEvents CurrentEvents = processEventsIncludingCurrent
processEvents EarlierEvents = processEventsIncludingEarlier
processEvents CurrentEventsOrFromPast = processEventsIncludingCurrentCore
processEvents EarlierEventsOrFromPast = processEventsIncludingEarlierCore
-- | Whether there is an overflow.
isEventOverflow :: Event DIO Bool
isEventOverflow =
Event $ \p ->
do let q = runEventQueue $ pointRun p
n1 <- liftIOUnsafe $ logSize (queueLog q)
n2 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q)
ps <- dioParams
let th1 = dioUndoableLogSizeThreshold ps
th2 = dioTransientMessageQueueSizeThreshold ps
if (n1 >= th1) || (n2 >= th2)
then do logDIO NOTICE $
"t = " ++ (show $ pointTime p) ++
": detected the event overflow"
return True
else return False
-- | Throttle the message channel.
throttleMessageChannel :: TimeWarp DIO ()
throttleMessageChannel =
TimeWarp $ \p ->
do -- invokeEvent p requestGlobalTime
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
liftIOUnsafe $
timeout dt $ awaitChannel ch
invokeTimeWarp p $ processChannelMessages
-- | Process the channel messages.
processChannelMessages :: TimeWarp DIO ()
processChannelMessages =
TimeWarp $ \p ->
do ch <- messageChannel
f <- liftIOUnsafe $ channelEmpty ch
unless f $
do xs <- liftIOUnsafe $ readChannel ch
forM_ xs $ \x ->
do p' <- invokeEvent p currentEventPoint
invokeTimeWarp p' $ processChannelMessage x
p' <- invokeEvent p currentEventPoint
f2 <- invokeEvent p' isEventOverflow
when f2 $
invokeTimeWarp p' throttleMessageChannel
-- | Process the channel message.
processChannelMessage :: LogicalProcessMessage -> TimeWarp DIO ()
processChannelMessage x@(QueueMessage m) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
infind <- liftIOUnsafe $ readIORef (queueInFind q)
deliverAcknowledgementMessage (acknowledgementMessage infind m)
t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)
p' <- invokeEvent p currentEventPoint
if messageReceiveTime m < t0
then do f <- fmap dioAllowSkippingOutdatedMessage dioParams
if f
then invokeEvent p' logOutdatedMessage
else error "Received the outdated message: processChannelMessage"
else invokeTimeWarp p' $
enqueueMessage (queueInputMessages q) m
processChannelMessage x@(QueueMessageBulk ms) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
infind <- liftIOUnsafe $ readIORef (queueInFind q)
deliverAcknowledgementMessages $ map (acknowledgementMessage infind) ms
t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q)
forM_ ms $ \m ->
do p' <- invokeEvent p currentEventPoint
if messageReceiveTime m < t0
then do f <- fmap dioAllowSkippingOutdatedMessage dioParams
if f
then invokeEvent p' logOutdatedMessage
else error "Received the outdated message: processChannelMessage"
else invokeTimeWarp p' $
enqueueMessage (queueInputMessages q) m
processChannelMessage x@(AcknowledgementQueueMessage m) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
liftIOUnsafe $
processAcknowledgementMessage (queueTransientMessages q) m
processChannelMessage x@(AcknowledgementQueueMessageBulk ms) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
liftIOUnsafe $
forM_ ms $
processAcknowledgementMessage (queueTransientMessages q)
processChannelMessage x@ComputeLocalTimeMessage =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
liftIOUnsafe $
writeIORef (queueInFind q) True
t' <- invokeEvent p getLocalTime
sender <- messageInboxId
receiver <- timeServerId
sendLocalTimeDIO receiver sender t'
processChannelMessage x@(GlobalTimeMessage globalTime) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
liftIOUnsafe $
do writeIORef (queueInFind q) False
resetAcknowledgementMessageTime (queueTransientMessages q)
invokeEvent p $
updateGlobalTime globalTime
processChannelMessage x@(ProcessMonitorNotificationMessage y@(DP.ProcessMonitorNotification _ pid reason)) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
invokeEvent p $
triggerSignal (queueProcessMonitorNotificationSource q) y
processChannelMessage x@(ReconnectProcessMessage pid) =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logMessage x
---
invokeEvent p $
reconnectProcess pid
processChannelMessage x@(ProvideLogicalProcessStateMessage pid) =
TimeWarp $ \p ->
do ---
--- invokeEvent p $
--- logMessage x
---
invokeEvent p $
sendState pid
processChannelMessage x@AbortSimulationMessage =
TimeWarp $ \p ->
do ---
--- invokeEvent p $
--- logMessage x
---
invokeEvent p $
throwEvent $
SimulationAbort "Aborted by the outer process."
-- | Return the local minimum time.
getLocalTime :: Event DIO Double
getLocalTime =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t1 <- liftIOUnsafe $ readIORef (queueTime q)
t2 <- liftIOUnsafe $ transientMessageQueueTime (queueTransientMessages q)
t3 <- liftIOUnsafe $ acknowledgementMessageTime (queueTransientMessages q)
let t' = t1 `min` t2 `min` t3
---
--- n <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q)
--- logDIO ERROR $
--- "t = " ++ show (pointTime p) ++
--- ": queue time = " ++ show t1 ++
--- ", unacknowledged time = " ++ show t2 ++
--- ", marked acknowledged time = " ++ show t3 ++
--- ", transient queue size = " ++ show n ++
--- " -> " ++ show t'
---
return t'
-- | Update the global time.
updateGlobalTime :: Double -> Event DIO ()
updateGlobalTime t =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t' <- invokeEvent p getLocalTime
if t > t'
then logDIO WARNING $
"t = " ++ show t' ++
": Ignored the global time that is greater than the current local time"
else do liftIOUnsafe $
writeIORef (queueGlobalTime q) t
invokeEvent p $
reduceEvents t
-- | Request for the global minimum time.
requestGlobalTime :: Event DIO ()
requestGlobalTime =
Event $ \p ->
do let q = runEventQueue $ pointRun p
---
--- invokeEvent p $
--- logRequestGlobalTime
---
sender <- messageInboxId
receiver <- timeServerId
sendRequestGlobalTimeDIO receiver sender
-- | Show the message.
showMessage :: Message -> ShowS
showMessage m =
showString "{ " .
showString "sendTime = " .
shows (messageSendTime m) .
showString ", receiveTime = " .
shows (messageReceiveTime m) .
(if messageAntiToggle m
then showString ", antiToggle = True"
else showString "") .
showString " }"
-- | Log the message at the specified time.
logMessage :: LogicalProcessMessage -> Event DIO ()
logMessage (QueueMessage m) =
Event $ \p ->
logDIO INFO $
"t = " ++ (show $ pointTime p) ++
": QueueMessage " ++
showMessage m []
logMessage (QueueMessageBulk ms) =
Event $ \p ->
logDIO INFO $
"t = " ++ (show $ pointTime p) ++
": QueueMessageBulk [ " ++
let fs = foldl1 (\a b -> a . showString ", " . b) $ map showMessage ms
in fs [] ++ " ]"
logMessage m =
Event $ \p ->
logDIO DEBUG $
"t = " ++ (show $ pointTime p) ++
": " ++ show m
-- | Log that the local time is to be synchronized.
logSyncLocalTime :: Event DIO ()
logSyncLocalTime =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)
logDIO DEBUG $
"t = " ++ (show $ pointTime p) ++
", global t = " ++ (show t') ++
": synchronizing the local time..."
-- | Log that the local time is to be synchronized in ring 0.
logSyncLocalTime0 :: Event DIO ()
logSyncLocalTime0 =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)
logDIO DEBUG $
"t = " ++ (show $ pointTime p) ++
", global t = " ++ (show t') ++
": synchronizing the local time in ring 0..."
-- | Log that the global time is requested.
logRequestGlobalTime :: Event DIO ()
logRequestGlobalTime =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)
logDIO DEBUG $
"t = " ++ (show $ pointTime p) ++
", global t = " ++ (show t') ++
": requesting for a new global time..."
-- | Log an evidence of the premature IO.
logPrematureIO :: Event DIO ()
logPrematureIO =
Event $ \p ->
logDIO ERROR $
"t = " ++ (show $ pointTime p) ++
": detected a premature IO action"
-- | Log an evidence of receiving the outdated message.
logOutdatedMessage :: Event DIO ()
logOutdatedMessage =
Event $ \p ->
logDIO WARNING $
"t = " ++ (show $ pointTime p) ++
": skipping the outdated message"
-- | Reduce events till the specified time.
reduceEvents :: Double -> Event DIO ()
reduceEvents t =
Event $ \p ->
do let q = runEventQueue $ pointRun p
liftIOUnsafe $
do reduceInputMessages (queueInputMessages q) t
reduceOutputMessages (queueOutputMessages q) t
reduceLog (queueLog q) t
instance {-# OVERLAPPING #-} MonadIO (Event DIO) where
liftIO m =
Event $ \p ->
do ok <- invokeEvent p $
runTimeWarp $
syncLocalTime $
return ()
if ok
then liftIOUnsafe m
else do f <- fmap dioAllowPrematureIO dioParams
if f
then do ---
--- invokeEvent p $ logPrematureIO
---
liftIOUnsafe m
else error $
"Detected a premature IO action at t = " ++
(show $ pointTime p) ++ ": liftIO"
-- | Synchronize the local time executing the specified computation.
syncLocalTime :: Dynamics DIO () -> TimeWarp DIO ()
syncLocalTime m =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
t = pointTime p
invokeDynamics p m
t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)
if t' > t
then error "Inconsistent time: syncLocalTime"
else if (t == spcStartTime (pointSpecs p)) || (t' == pointTime p)
then return ()
else do ---
--- invokeEvent p logSyncLocalTime
---
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
f <- liftIOUnsafe $
timeout dt $ awaitChannel ch
ok <- invokeEvent p $ runTimeWarp processChannelMessages
if ok
then do case f of
Just _ ->
invokeTimeWarp p $ syncLocalTime m
Nothing ->
do -- invokeEvent p requestGlobalTime
invokeTimeWarp p $ syncLocalTime0 m
else return ()
-- | Synchronize the local time executing the specified computation in ring 0.
syncLocalTime0 :: Dynamics DIO () -> TimeWarp DIO ()
syncLocalTime0 m =
TimeWarp $ \p ->
do let q = runEventQueue $ pointRun p
t = pointTime p
invokeDynamics p m
t' <- liftIOUnsafe $ readIORef (queueGlobalTime q)
if t' > t
then error "Inconsistent time: syncLocalTime0"
else if t' == pointTime p
then return ()
else do ---
--- invokeEvent p logSyncLocalTime0
---
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
f <- liftIOUnsafe $
timeout dt $ awaitChannel ch
ok <- invokeEvent p $ runTimeWarp processChannelMessages
if ok
then do case f of
Just _ ->
invokeTimeWarp p $ syncLocalTime m
Nothing ->
error "Detected a deadlock when synchronizing the local time: syncLocalTime0"
else return ()
-- | Run the computation and return a flag indicating whether there was no rollback.
runTimeWarp :: TimeWarp DIO () -> Event DIO Bool
runTimeWarp m =
Event $ \p ->
do let q = runEventQueue $ pointRun p
v0 <- liftIOUnsafe $ inputMessageQueueVersion (queueInputMessages q)
invokeTimeWarp p m
v2 <- liftIOUnsafe $ inputMessageQueueVersion (queueInputMessages q)
return (v0 == v2)
-- | Synchronize the events.
syncEvents :: EventProcessing -> Event DIO ()
syncEvents processing =
Event $ \p ->
do ok <- invokeEvent p $
runTimeWarp $
syncLocalTime $
processEvents processing
unless ok $
invokeEvent p $
syncEvents processing
-- | 'DIO' is an instance of 'EventIOQueueing'.
instance EventIOQueueing DIO where
enqueueEventIO t h =
enqueueEvent t $
Event $ \p ->
do ok <- invokeEvent p $
runTimeWarp $
syncLocalTime $
return ()
when ok $
invokeEvent p h
-- | Handle the 'Event' retry.
handleEventRetry :: SimulationRetry -> Event DIO ()
handleEventRetry e =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t = pointTime p
---
logDIO INFO $
"t = " ++ show t ++
": retrying the computations..."
---
invokeTimeWarp p $
retryInputMessages (queueInputMessages q)
let loop =
do ---
--- logDIO DEBUG $
--- "t = " ++ show t ++
--- ": waiting for arriving a message..."
---
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
f <- liftIOUnsafe $
timeout dt $ awaitChannel ch
ok <- invokeEvent p $ runTimeWarp processChannelMessages
when ok $
case f of
Just _ -> loop
Nothing -> loop0
loop0 =
do ---
--- logDIO DEBUG $
--- "t = " ++ show t ++
--- ": waiting for arriving a message in ring 0..."
---
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
f <- liftIOUnsafe $
timeout dt $ awaitChannel ch
ok <- invokeEvent p $ runTimeWarp processChannelMessages
when ok $
case f of
Just _ -> loop
Nothing ->
error $
"Detected a deadlock when retrying the computations: handleEventRetry\n" ++
"--- the nested exception ---\n" ++ show e
loop
-- | Reconnect to the remote process.
reconnectProcess :: DP.ProcessId -> Event DIO ()
reconnectProcess pid =
Event $ \p ->
do let q = runEventQueue $ pointRun p
---
logDIO NOTICE $
"t = " ++ show (pointTime p) ++
": reconnecting to " ++ show pid ++ "..."
---
infind <- liftIOUnsafe $ readIORef (queueInFind q)
let ys = queueInputMessages q
ys' <- liftIOUnsafe $
fmap (map $ acknowledgementMessage infind) $
filterInputMessages (\x -> messageSenderId x == pid) ys
unless (null ys') $
sendAcknowledgementMessagesDIO pid ys'
xs <- liftIOUnsafe $ transientMessages (queueTransientMessages q)
let xs' = filter (\x -> messageReceiverId x == pid) xs
unless (null xs') $
sendMessagesDIO pid xs'
-- | A signal triggered when coming the process monitor notification from the Cloud Haskell back-end.
processMonitorSignal :: Signal DIO DP.ProcessMonitorNotification
processMonitorSignal =
Signal { handleSignal = \h ->
Event $ \p ->
let q = runEventQueue (pointRun p)
s = publishSignal (queueProcessMonitorNotificationSource q)
in invokeEvent p $
handleSignal s h
}
-- | Suspend the 'Event' computation until the specified computation is determined.
--
-- The tested computation should depend on messages that come from other logical processes.
-- Moreover, the event must be initiated through the event queue.
expectEvent :: Event DIO (Maybe a) -> (a -> Event DIO ()) -> Event DIO ()
expectEvent m cont =
Event $ \p ->
do let q = runEventQueue $ pointRun p
t = pointTime p
---
logDIO INFO $
"t = " ++ show (pointTime p) ++
": expecting the computation result: expectEvent"
---
let loop =
do ---
--- logDIO DEBUG $
--- "t = " ++ show (pointTime p) ++
--- ": testing the predicate: expectEvent"
---
x <- invokeEvent p m
case x of
Just a -> invokeEvent p $ cont a
Nothing -> next loop0
loop0 =
do ---
--- logDIO DEBUG $
--- "t = " ++ show (pointTime p) ++
--- ": testing the predicate in ring 0: expectEvent"
---
x <- invokeEvent p m
case x of
Just a -> invokeEvent p $ cont a
Nothing -> next $ error "Detected a deadlock: expectEvent"
next loop' =
do pq <- invokeEvent p $ R.readRef $ queuePQ q
let f = PQ.queueNull pq
if f
then await loop'
else do let (t2, _) = PQ.queueFront pq
if t < t2
then await loop'
else invokeEvent p $
enqueueEvent t $
Event $ \p -> loop
await loop' =
do ---
--- logDIO DEBUG $
--- "t = " ++ show t ++
--- ": waiting for arriving a message: expectEvent"
---
ch <- messageChannel
dt <- fmap dioSyncTimeout dioParams
f <- liftIOUnsafe $
timeout dt $ awaitChannel ch
ok <- invokeEvent p $ runTimeWarp processChannelMessages
when ok $
case f of
Just _ -> loop
Nothing -> loop'
loop
-- | Send the simulation monitoring message about the current state of the logical process.
sendState :: DP.ProcessId -> Event DIO ()
sendState pid =
Event $ \p ->
do let q = runEventQueue $ pointRun p
pq <- invokeEvent p $ R.readRef (queuePQ q)
n1 <- liftIOUnsafe $ logSize (queueLog q)
n2 <- liftIOUnsafe $ inputMessageQueueSize (queueInputMessages q)
n3 <- liftIOUnsafe $ outputMessageQueueSize (queueOutputMessages q)
n4 <- liftIOUnsafe $ transientMessageQueueSize (queueTransientMessages q)
n5 <- liftIOUnsafe $ inputMessageQueueVersion (queueInputMessages q)
let n6 = PQ.queueCount pq
sc = pointSpecs p
t0 = spcStartTime sc
t2 = spcStopTime sc
tq <- liftIOUnsafe $ readIORef (queueTime q)
t' <- invokeEvent p getLocalTime
ps <- dioParams
let name = dioName ps
inbox <- messageInboxId
liftDistributedUnsafe $
DP.send pid $
LogicalProcessState { lpStateId = inbox,
lpStateName = name,
lpStateStartTime = t0,
lpStateStopTime = t2,
lpStateLocalTime = t',
lpStateEventQueueTime = tq,
lpStateEventQueueSize = n6,
lpStateLogSize = n1,
lpStateInputMessageCount = n2,
lpStateOutputMessageCount = n3,
lpStateTransientMessageCount = n4,
lpStateRollbackCount = n5 }