aivika-distributed 0.2 → 0.3
raw patch · 18 files changed
+538/−239 lines, 18 filesPVP ok
version bump matches the API change (PVP)
API changes (from Hackage documentation)
+ Simulation.Aivika.Distributed.Optimistic.DIO: registerDIO :: DIO ()
+ Simulation.Aivika.Distributed.Optimistic.TimeServer: curryTimeServer :: (Int, TimeServerParams) -> Process ()
- Simulation.Aivika.Distributed.Optimistic.TimeServer: timeServer :: TimeServerParams -> Process ()
+ Simulation.Aivika.Distributed.Optimistic.TimeServer: timeServer :: Int -> TimeServerParams -> Process ()
Files
- CHANGELOG.md +11/−0
- Simulation/Aivika/Distributed/Optimistic/DIO.hs +1/−0
- Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +13/−6
- Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +85/−86
- Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +44/−6
- Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs +13/−6
- Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot +1/−1
- Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +166/−110
- Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs +163/−0
- Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +2/−1
- aivika-distributed.cabal +5/−2
- tests/MachRep1.hs +5/−3
- tests/MachRep1Simple.hs +3/−2
- tests/MachRep2.hs +5/−3
- tests/MachRep2Distributed.hs +6/−4
- tests/MachRep2Reproducible.hs +5/−3
- tests/MachRep2Sync.hs +5/−3
- tests/MachRep2SyncIO.hs +5/−3
+ CHANGELOG.md view
@@ -0,0 +1,11 @@++Version 0.3+-----++* Started using Samadi's algorithm to synchronize the global virtual time.++* The local processes must call registerDIO to connect to the time server.++* Increased the default synchronization time-out and delay.++* Increased the default log size threshold.
Simulation/Aivika/Distributed/Optimistic/DIO.hs view
@@ -21,6 +21,7 @@ timeServerId, logDIO, terminateDIO,+ registerDIO, unregisterDIO) where import Control.Monad
Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs view
@@ -18,6 +18,7 @@ runDIO, defaultDIOParams, terminateDIO,+ registerDIO, unregisterDIO, dioParams, messageChannel,@@ -128,9 +129,9 @@ defaultDIOParams :: DIOParams defaultDIOParams = DIOParams { dioLoggingPriority = WARNING,- dioUndoableLogSizeThreshold = 500000,+ dioUndoableLogSizeThreshold = 1000000, dioOutputMessageQueueSizeThreshold = 10000,- dioSyncTimeout = 5000000,+ dioSyncTimeout = 60000000, dioAllowPrematureIO = False, dioAllowProcessingOutdatedMessage = False }@@ -161,6 +162,16 @@ liftDistributedUnsafe $ DP.send receiver (TerminateTimeServerMessage sender) +-- | Register the simulation process in the time server, which+-- requires some initial quorum to start synchronizing the global time.+registerDIO :: DIO ()+registerDIO =+ do logDIO INFO "Registering the simulation process..."+ sender <- messageInboxId+ receiver <- timeServerId+ liftDistributedUnsafe $+ DP.send receiver (RegisterLocalProcessMessage sender)+ -- | Unregister the simulation process from the time server -- without affecting the processes in other nodes connected to -- the corresponding time server.@@ -188,10 +199,6 @@ logProcess ps INFO "Terminating the inbox process..." --- DP.terminate- ---- logProcess ps INFO "Registering the simulation process..."- ---- DP.send serverId (RegisterLocalProcessMessage inboxId) return (inboxId, unDIO m DIOContext { dioChannel = ch, dioInboxId = inboxId, dioTimeServerId = serverId,
Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs view
@@ -42,6 +42,7 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp 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 as R @@ -58,6 +59,8 @@ -- ^ 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 ())),@@ -68,14 +71,8 @@ -- ^ the actual time of the event queue queueGlobalTime :: IORef Double, -- ^ the global time- queueLocalTime :: IORef Double,- -- ^ the long-term queue local time- queueLocalTime0 :: IORef Double,- -- ^ the short-term queue local time- queueLocalTimeTimestamp :: IORef UTCTime,- -- ^ the queue local time timestamp- queueLocalTimeInterval :: NominalDiffTime- -- ^ the queue local time interval for updating+ queueInFind :: IORef Bool+ -- ^ whether the queue is in find mode } newEventQueue specs =@@ -84,23 +81,19 @@ gt <- liftIOUnsafe $ newIORef $ spcStartTime specs pq <- R.newRef0 PQ.emptyQueue log <- newUndoableLog- output <- newOutputMessageQueue+ transient <- newTransientMessageQueue+ output <- newOutputMessageQueue $ enqueueTransientMessage transient input <- newInputMessageQueue log rollbackEventPre rollbackEventPost rollbackEventTime- loct <- liftIOUnsafe $ newIORef $ spcStartTime specs- loct0 <- liftIOUnsafe $ newIORef $ spcStartTime specs- loctstamp <- liftIOUnsafe $ getCurrentTime >>= newIORef- locdt <- fmap (fromRational . microsecondsToSeconds . dioSyncTimeout) dioParams + infind <- liftIOUnsafe $ newIORef False return EventQueue { queueInputMessages = input, queueOutputMessages = output,+ queueTransientMessages = transient, queueLog = log, queuePQ = pq, queueBusy = f, queueTime = t, queueGlobalTime = gt,- queueLocalTime = loct,- queueLocalTime0 = loct0,- queueLocalTimeTimestamp = loctstamp,- queueLocalTimeInterval = locdt }+ queueInFind = infind } enqueueEvent t (Event m) = Event $ \p ->@@ -145,10 +138,7 @@ --- logDIO DEBUG $ --- "Setting the queue time = " ++ show t ---- liftIOUnsafe $- do writeIORef (queueTime q) t- modifyIORef' (queueLocalTime q) (min t)- modifyIORef' (queueLocalTime0 q) (min t)+ liftIOUnsafe $ writeIORef (queueTime q) t t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q) when (t0 > t) $ do ---@@ -156,7 +146,6 @@ --- "Setting the global time = " ++ show t --- liftIOUnsafe $ writeIORef (queueGlobalTime q) t- invokeEvent p sendLocalTime -- | Return the current event time. currentEventTime :: Event DIO Double@@ -298,8 +287,7 @@ throttleMessageChannel :: TimeWarp DIO () throttleMessageChannel = TimeWarp $ \p ->- do invokeEvent p updateLocalTime- invokeEvent p sendLocalTime+ do invokeEvent p requestGlobalTime ch <- messageChannel dt <- fmap dioSyncTimeout dioParams liftIOUnsafe $@@ -331,6 +319,8 @@ --- invokeEvent p $ --- logMessage x ---+ infind <- liftIOUnsafe $ readIORef (queueInFind q)+ deliverAcknowledgmentMessage (acknowledgmentMessage infind m) t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q) when (messageReceiveTime m < t0) $ do f <- fmap dioAllowProcessingOutdatedMessage dioParams@@ -346,6 +336,8 @@ --- invokeEvent p $ --- logMessage x ---+ infind <- liftIOUnsafe $ readIORef (queueInFind q)+ deliverAcknowledgmentMessages $ map (acknowledgmentMessage infind) ms t0 <- liftIOUnsafe $ readIORef (queueGlobalTime q) forM_ ms $ \m -> do when (messageReceiveTime m < t0) $@@ -355,31 +347,49 @@ else error "Received the outdated message: processChannelMessage" invokeTimeWarp p $ enqueueMessage (queueInputMessages q) m-processChannelMessage x@(GlobalTimeMessage globalTime) =+processChannelMessage x@(AcknowledgmentQueueMessage m) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p --- --- invokeEvent p $ --- logMessage x ---- case globalTime of- Nothing -> return ()- Just t0 ->- invokeEvent p $- updateGlobalTime t0- invokeEvent p updateLocalTime- t <- invokeEvent p getLocalTime+ liftIOUnsafe $+ processAcknowledgmentMessage (queueTransientMessages q) m+processChannelMessage x@(AcknowledgmentQueueMessageBulk ms) =+ TimeWarp $ \p ->+ do let q = runEventQueue $ pointRun p+ ---+ --- invokeEvent p $+ --- logMessage x+ ---+ liftIOUnsafe $+ forM_ ms $+ processAcknowledgmentMessage (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 liftDistributedUnsafe $- DP.send receiver (GlobalTimeMessageResp sender t)-processChannelMessage x@(LocalTimeMessageResp globalTime) =+ DP.send receiver (LocalTimeMessage sender t')+processChannelMessage x@(GlobalTimeMessage globalTime) = TimeWarp $ \p -> do let q = runEventQueue $ pointRun p --- --- invokeEvent p $ --- logMessage x ---+ liftIOUnsafe $+ do writeIORef (queueInFind q) False+ resetAcknowledgmentMessageTime (queueTransientMessages q) invokeEvent p $ updateGlobalTime globalTime processChannelMessage x@TerminateLocalProcessMessage =@@ -391,12 +401,32 @@ liftDistributedUnsafe $ DP.terminate +-- | 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 $ acknowledgmentMessageTime (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- invokeEvent p updateLocalTime t' <- invokeEvent p getLocalTime if t > t' then logDIO WARNING $@@ -407,6 +437,20 @@ 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+ liftDistributedUnsafe $+ DP.send receiver (RequestGlobalTimeMessage sender)+ -- | Show the message. showMessage :: Message -> ShowS showMessage m =@@ -463,16 +507,16 @@ ", global t = " ++ (show t') ++ ": synchronizing the local time in ring 0..." --- | Log that the local time is sent to the time server.-logSendLocalTime :: Event DIO ()-logSendLocalTime =+-- | 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') ++- ": sending the local time to the time server after delay..."+ ": requesting for a new global time..." -- | Log an evidence of the premature IO. logPrematureIO :: Event DIO ()@@ -520,48 +564,6 @@ "Detected a premature IO action at t = " ++ (show $ pointTime p) ++ ": liftIO" --- | Update the local time.-updateLocalTime :: Event DIO Bool-updateLocalTime =- Event $ \p ->- do let q = runEventQueue $ pointRun p- timestamp0 <- liftIOUnsafe $ readIORef (queueLocalTimeTimestamp q)- timestamp <- liftIOUnsafe getCurrentTime- let dt = queueLocalTimeInterval q- if timestamp >= addUTCTime dt timestamp0- then do ---- --- logDIO DEBUG $ "t = " ++ (show $ pointTime p) ++ ": updating the local time"- ---- liftIOUnsafe $- do t <- readIORef (queueTime q)- loct0 <- readIORef (queueLocalTime0 q)- writeIORef (queueLocalTime q) (min t loct0)- writeIORef (queueLocalTime0 q) t- writeIORef (queueLocalTimeTimestamp q) timestamp- return (t /= loct0)- else return False---- | Get the local time.-getLocalTime :: Event DIO Double-getLocalTime =- Event $ \p ->- let q = runEventQueue $ pointRun p- in liftIOUnsafe $ readIORef (queueLocalTime q)---- | Send the local time to the time server.-sendLocalTime :: Event DIO ()-sendLocalTime =- Event $ \p ->- do ---- --- invokeEvent p logSendLocalTime- ---- invokeEvent p updateLocalTime- t <- invokeEvent p getLocalTime- sender <- messageInboxId- receiver <- timeServerId- liftDistributedUnsafe $- DP.send receiver (LocalTimeMessage sender t)- -- | Synchronize the local time executing the specified computation. syncLocalTime :: Dynamics DIO () -> TimeWarp DIO () syncLocalTime m =@@ -587,11 +589,8 @@ Just _ -> invokeTimeWarp p $ syncLocalTime m Nothing ->- do f <- invokeEvent p updateLocalTime- invokeEvent p sendLocalTime- if f- then invokeTimeWarp p $ syncLocalTime m- else invokeTimeWarp p $ syncLocalTime0 m+ do invokeEvent p requestGlobalTime+ invokeTimeWarp p $ syncLocalTime0 m else return () -- | Synchronize the local time executing the specified computation in ring 0.
Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs view
@@ -15,6 +15,8 @@ (Message(..), antiMessage, antiMessages,+ AcknowledgmentMessage(..),+ acknowledgmentMessage, LocalProcessMessage(..), TimeServerMessage(..)) where @@ -70,15 +72,51 @@ (messageReceiverId x == messageReceiverId y) && (messageAntiToggle x == messageAntiToggle y) +-- | Represents an acknowledgement message.+data AcknowledgmentMessage =+ AcknowledgmentMessage { acknowledgmentSequenceNo :: Int,+ -- ^ The sequence number.+ acknowledgmentSendTime :: Double,+ -- ^ The send time.+ acknowledgmentReceiveTime :: Double,+ -- ^ The receive time.+ acknowledgmentSenderId :: DP.ProcessId,+ -- ^ The sender of the source message.+ acknowledgmentReceiverId :: DP.ProcessId,+ -- ^ The receiver of the source message.+ acknowledgmentAntiToggle :: Bool,+ -- ^ Whether this is an anti-message acknowledgment.+ acknowledgmentMarked :: Bool+ -- ^ Whether the acknowledgment is marked.+ } deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary AcknowledgmentMessage++-- | Create an acknowledgment message specifying whether it will be marked.+acknowledgmentMessage :: Bool -> Message -> AcknowledgmentMessage+acknowledgmentMessage marked x =+ AcknowledgmentMessage { acknowledgmentSequenceNo = messageSequenceNo x,+ acknowledgmentSendTime = messageSendTime x,+ acknowledgmentReceiveTime = messageReceiveTime x,+ acknowledgmentSenderId = messageSenderId x,+ acknowledgmentReceiverId = messageReceiverId x,+ acknowledgmentAntiToggle = messageAntiToggle x,+ acknowledgmentMarked = marked+ }+ -- | The message sent to the local process. data LocalProcessMessage = QueueMessage Message -- ^ the message has come from the remote process | QueueMessageBulk [Message] -- ^ a bulk of messages that have come from the remote process- | GlobalTimeMessage (Maybe Double)+ | AcknowledgmentQueueMessage AcknowledgmentMessage+ -- ^ the acknowledgment message has come from the remote process+ | AcknowledgmentQueueMessageBulk [AcknowledgmentMessage]+ -- ^ a bulk of acknowledgment messages that have come from the remote process+ | ComputeLocalTimeMessage+ -- ^ the time server requests for a local minimum time+ | GlobalTimeMessage Double -- ^ the time server sent a global time- | LocalTimeMessageResp Double- -- ^ the time server replied to 'LocalTimeMessage' sending its global time in response | TerminateLocalProcessMessage -- ^ the time server asked to terminate the process deriving (Eq, Show, Typeable, Generic)@@ -90,10 +128,10 @@ -- ^ register the local process in the time server | UnregisterLocalProcessMessage DP.ProcessId -- ^ unregister the local process from the time server- | GlobalTimeMessageResp DP.ProcessId Double- -- ^ the local process replied to 'GlobalTimeMessage' sending its local time in response+ | RequestGlobalTimeMessage DP.ProcessId+ -- ^ the local process requested for the global minimum time | LocalTimeMessage DP.ProcessId Double- -- ^ the local process sent its local time+ -- ^ the local process sent its local minimum time | TerminateTimeServerMessage DP.ProcessId -- ^ the local process asked to terminate the time server deriving (Eq, Show, Typeable, Generic)
Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs view
@@ -38,18 +38,21 @@ -- | Specifies the output message queue. data OutputMessageQueue =- OutputMessageQueue { outputMessages :: DLL.DoubleLinkedList Message,+ OutputMessageQueue { outputEnqueueTransientMessage :: Message -> IO (),+ -- ^ Enqueue the transient message.+ outputMessages :: DLL.DoubleLinkedList Message, -- ^ The output messages. outputMessageSequenceNo :: IORef Int -- ^ The next sequence number. } -- | Create a new output message queue.-newOutputMessageQueue :: DIO OutputMessageQueue-newOutputMessageQueue =+newOutputMessageQueue :: (Message -> IO ()) -> DIO OutputMessageQueue+newOutputMessageQueue transient = do ms <- liftIOUnsafe DLL.newList rn <- liftIOUnsafe $ newIORef 0- return OutputMessageQueue { outputMessages = ms,+ return OutputMessageQueue { outputEnqueueTransientMessage = transient,+ outputMessages = ms, outputMessageSequenceNo = rn } -- | Return the output message queue size.@@ -68,6 +71,7 @@ do m' <- liftIOUnsafe $ DLL.listLast (outputMessages q) when (messageSendTime m' > messageSendTime m) $ error "A new output message comes from the past: sendMessage."+ liftIOUnsafe $ outputEnqueueTransientMessage q m deliverMessage m liftIOUnsafe $ DLL.listAddLast (outputMessages q) m @@ -75,8 +79,11 @@ rollbackOutputMessages :: OutputMessageQueue -> Double -> Bool -> DIO () rollbackOutputMessages q t including = do ms <- liftIOUnsafe $ extractMessagesToRollback q t including- deliverAntiMessages $ map antiMessage ms- -- forM_ ms $ deliverAntiMessage . antiMessage+ let ms' = map antiMessage ms+ liftIOUnsafe $+ forM_ ms' $ outputEnqueueTransientMessage q+ deliverAntiMessages ms'+ -- forM_ ms' deliverAntiMessage -- | Return the messages to roolback by the specified time. extractMessagesToRollback :: OutputMessageQueue -> Double -> Bool -> IO [Message]
Simulation/Aivika/Distributed/Optimistic/Internal/OutputMessageQueue.hs-boot view
@@ -26,7 +26,7 @@ data OutputMessageQueue -newOutputMessageQueue :: DIO OutputMessageQueue+newOutputMessageQueue :: (Message -> IO ()) -> DIO OutputMessageQueue outputMessageQueueSize :: OutputMessageQueue -> IO Int
Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs view
@@ -14,13 +14,16 @@ module Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer (TimeServerParams(..), defaultTimeServerParams,- timeServer) where+ timeServer,+ curryTimeServer) where import qualified Data.Map as M+import qualified Data.Set as S import Data.Maybe import Data.IORef import Data.Typeable import Data.Binary+import Data.Time.Clock import GHC.Generics @@ -28,7 +31,6 @@ import Control.Monad.Trans import Control.Concurrent import qualified Control.Distributed.Process as DP-import Control.Distributed.Process.Closure (remotable, mkClosure) import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.Message@@ -40,7 +42,7 @@ tsExpectTimeout :: Int, -- ^ the timeout in microseconds within which a new message is expected tsTimeSyncDelay :: Int- -- ^ the further delay in microseconds before the time synchronization+ -- ^ the delay in microseconds between the time synchronization sessions } deriving (Eq, Ord, Show, Typeable, Generic) instance Binary TimeServerParams@@ -49,51 +51,63 @@ data TimeServer = TimeServer { tsParams :: TimeServerParams, -- ^ the time server parameters+ tsInitQuorum :: Int,+ -- ^ the initial quorum of registered local processes to start the simulation+ tsInInit :: IORef Bool,+ -- ^ whether the time server is in the initial mode tsProcesses :: IORef (M.Map DP.ProcessId LocalProcessInfo), -- ^ the information about local processes- tsGlobalTime :: IORef (Maybe Double),+ tsProcessesInFind :: IORef (S.Set DP.ProcessId),+ -- ^ the processed used in the current finding of the global time+ tsGlobalTime :: IORef (Maybe Double) -- ^ the global time of the model- tsGlobalTimeInvalid :: IORef Bool- -- ^ whether the global time is invalid } -- | The information about the local process. data LocalProcessInfo =- LocalProcessInfo { lpLocalTime :: IORef (Maybe Double),+ LocalProcessInfo { lpLocalTime :: IORef (Maybe Double) -- ^ the local time of the process- lpSentGlobalTime :: IORef (Maybe Double)- -- ^ the global time sent to the process for the last time } -- | The default time server parameters. defaultTimeServerParams :: TimeServerParams defaultTimeServerParams = TimeServerParams { tsLoggingPriority = WARNING,- tsExpectTimeout = 1000,- tsTimeSyncDelay = 1000+ tsExpectTimeout = 100000,+ tsTimeSyncDelay = 1000000 } --- | Create a new time server.-newTimeServer :: TimeServerParams -> IO TimeServer-newTimeServer ps =- do m <- newIORef M.empty+-- | Create a new time server by the specified initial quorum and parameters.+newTimeServer :: Int -> TimeServerParams -> IO TimeServer+newTimeServer n ps =+ do f <- newIORef True+ m <- newIORef M.empty+ s <- newIORef S.empty t0 <- newIORef Nothing- f <- newIORef False return TimeServer { tsParams = ps,+ tsInitQuorum = n,+ tsInInit = f, tsProcesses = m,- tsGlobalTime = t0,- tsGlobalTimeInvalid = f+ tsProcessesInFind = s,+ tsGlobalTime = t0 } -- | Process the time server message. processTimeServerMessage :: TimeServer -> TimeServerMessage -> DP.Process () processTimeServerMessage server (RegisterLocalProcessMessage pid) =- liftIO $- do t1 <- newIORef Nothing- t2 <- newIORef Nothing- writeIORef (tsGlobalTimeInvalid server) True- modifyIORef (tsProcesses server) $- M.insert pid LocalProcessInfo { lpLocalTime = t1, lpSentGlobalTime = t2 }+ join $ liftIO $+ do m <- readIORef (tsProcesses server)+ case M.lookup pid m of+ Just x ->+ return $+ logTimeServer server WARNING $+ "Time Server: already registered process identifier " ++ show pid+ Nothing ->+ do t <- newIORef Nothing+ modifyIORef (tsProcesses server) $+ M.insert pid LocalProcessInfo { lpLocalTime = t }+ return $+ tryStartTimeServer server processTimeServerMessage server (UnregisterLocalProcessMessage pid) = join $ liftIO $ do m <- readIORef (tsProcesses server)@@ -103,43 +117,26 @@ logTimeServer server WARNING $ "Time Server: unknown process identifier " ++ show pid Just x ->- do t0 <- readIORef (tsGlobalTime server) - t <- readIORef (lpLocalTime x)- when (t0 .>=. t) $- writeIORef (tsGlobalTimeInvalid server) True- writeIORef (tsProcesses server) $- M.delete pid m- return $ return ()+ do modifyIORef (tsProcesses server) $+ M.delete pid+ modifyIORef (tsProcessesInFind server) $+ S.delete pid+ return $+ tryProvideTimeServerGlobalTime server processTimeServerMessage server (TerminateTimeServerMessage pid) = do pids <- liftIO $ do m <- readIORef (tsProcesses server) writeIORef (tsProcesses server) M.empty+ writeIORef (tsProcessesInFind server) S.empty writeIORef (tsGlobalTime server) Nothing- writeIORef (tsGlobalTimeInvalid server) True return $ filter (/= pid) (M.keys m) forM_ pids $ \pid -> DP.send pid TerminateLocalProcessMessage logTimeServer server INFO "Time Server: terminating..." DP.terminate-processTimeServerMessage server (GlobalTimeMessageResp pid t') =- join $ liftIO $- do m <- readIORef (tsProcesses server)- case M.lookup pid m of- Nothing ->- return $- do logTimeServer server WARNING $- "Time Server: unknown process identifier " ++ show pid- processTimeServerMessage server (RegisterLocalProcessMessage pid)- processTimeServerMessage server (GlobalTimeMessageResp pid t')- Just x ->- do t0 <- readIORef (tsGlobalTime server)- t <- readIORef (lpLocalTime x)- when (t /= Just t') $- do writeIORef (lpLocalTime x) (Just t')- when ((t0 .>=. t) || (t0 .>=. Just t')) $- writeIORef (tsGlobalTimeInvalid server) True- return $ return ()+processTimeServerMessage server (RequestGlobalTimeMessage pid) =+ tryComputeTimeServerGlobalTime server processTimeServerMessage server (LocalTimeMessage pid t') = join $ liftIO $ do m <- readIORef (tsProcesses server)@@ -151,19 +148,11 @@ processTimeServerMessage server (RegisterLocalProcessMessage pid) processTimeServerMessage server (LocalTimeMessage pid t') Just x ->- do t0 <- readIORef (tsGlobalTime server)- t <- readIORef (lpLocalTime x)- if t == Just t'- then return $ return ()- else do writeIORef (lpLocalTime x) (Just t')- when ((t0 .>=. t) || (t0 .>=. Just t')) $- writeIORef (tsGlobalTimeInvalid server) True- t0' <- readIORef (lpSentGlobalTime x)- if (t0 == t0') || (isNothing t0)- then return $ return ()- else do writeIORef (lpSentGlobalTime x) t0- return $- DP.send pid (LocalTimeMessageResp $ fromJust t0)+ do writeIORef (lpLocalTime x) (Just t')+ modifyIORef (tsProcessesInFind server) $+ S.delete pid+ return $+ tryProvideTimeServerGlobalTime server -- | Whether the both values are defined and the first is greater than or equaled to the second. (.>=.) :: Maybe Double -> Maybe Double -> Bool@@ -175,68 +164,136 @@ (.>.) (Just x) (Just y) = x > y (.>.) _ _ = False --- | Validate the time server.-validateTimeServer :: TimeServer -> DP.Process ()-validateTimeServer server =- do f <- liftIO $ readIORef (tsGlobalTimeInvalid server)- when f $- do t0 <- timeServerGlobalTime server- case t0 of- Nothing -> return ()- Just t0 ->- do t' <- liftIO $ readIORef (tsGlobalTime server)- when (t' .>. Just t0) $- logTimeServer server NOTICE- "Time Server: the global time has decreased"- liftIO $- do writeIORef (tsGlobalTime server) (Just t0)- writeIORef (tsGlobalTimeInvalid server) False- m <- liftIO $ readIORef (tsProcesses server)- forM_ (M.assocs m) $ \(pid, x) ->- do t0' <- liftIO $ readIORef (lpSentGlobalTime x)- when (t0' /= Just t0) $- do liftIO $ writeIORef (lpSentGlobalTime x) (Just t0)- DP.send pid (GlobalTimeMessage $ Just t0) +-- | Try to start synchronizing the global time.+tryStartTimeServer :: TimeServer -> DP.Process ()+tryStartTimeServer server =+ join $ liftIO $+ do f <- readIORef (tsInInit server)+ if not f+ then return $+ return ()+ else do m <- readIORef (tsProcesses server)+ if M.size m < tsInitQuorum server+ then return $+ return ()+ else do writeIORef (tsInInit server) False+ return $+ do logTimeServer server INFO $+ "Time Server: starting"+ tryComputeTimeServerGlobalTime server+ +-- | Try to compute the global time and provide the local processes with it.+tryComputeTimeServerGlobalTime :: TimeServer -> DP.Process ()+tryComputeTimeServerGlobalTime server =+ join $ liftIO $+ do f <- readIORef (tsInInit server)+ if f+ then return $+ return ()+ else do s <- readIORef (tsProcessesInFind server)+ if S.size s > 0+ then return $+ return ()+ else return $+ computeTimeServerGlobalTime server +-- | Try to provide the local processes wth the global time. +tryProvideTimeServerGlobalTime :: TimeServer -> DP.Process ()+tryProvideTimeServerGlobalTime server =+ join $ liftIO $+ do f <- readIORef (tsInInit server)+ if f+ then return $+ return ()+ else do s <- readIORef (tsProcessesInFind server)+ if S.size s > 0+ then return $+ return ()+ else return $+ provideTimeServerGlobalTime server++-- | Initiate computing the global time.+computeTimeServerGlobalTime :: TimeServer -> DP.Process ()+computeTimeServerGlobalTime server =+ do logTimeServer server DEBUG $+ "Time Server: computing the global time..."+ zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server)+ forM_ zs $ \(pid, x) ->+ liftIO $+ modifyIORef (tsProcessesInFind server) $+ S.insert pid+ forM_ zs $ \(pid, x) ->+ DP.send pid ComputeLocalTimeMessage++-- | Provide the local processes with the global time.+provideTimeServerGlobalTime :: TimeServer -> DP.Process ()+provideTimeServerGlobalTime server =+ do t0 <- liftIO $ timeServerGlobalTime server+ logTimeServer server DEBUG $+ "Time Server: providing the global time = " ++ show t0+ case t0 of+ Nothing -> return ()+ Just t0 ->+ do t' <- liftIO $ readIORef (tsGlobalTime server)+ when (t' .>. Just t0) $+ logTimeServer server NOTICE+ "Time Server: the global time has decreased"+ liftIO $ writeIORef (tsGlobalTime server) (Just t0)+ zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server)+ forM_ zs $ \(pid, x) ->+ DP.send pid (GlobalTimeMessage t0)+ -- | Return the time server global time.-timeServerGlobalTime :: TimeServer -> DP.Process (Maybe Double)+timeServerGlobalTime :: TimeServer -> IO (Maybe Double) timeServerGlobalTime server =- do t0 <- liftIO $ readIORef (tsGlobalTime server)- zs <- liftIO $ fmap M.assocs $ readIORef (tsProcesses server)+ do zs <- fmap M.assocs $ readIORef (tsProcesses server) case zs of [] -> return Nothing ((pid, x) : zs') ->- do t <- liftIO $ readIORef (lpLocalTime x)+ do t <- readIORef (lpLocalTime x) loop zs t where loop [] acc = return acc loop ((pid, x) : zs') acc =- do t <- liftIO $ readIORef (lpLocalTime x)+ do t <- readIORef (lpLocalTime x) case t of Nothing ->- do DP.send pid (GlobalTimeMessage Nothing)- loop zs' Nothing+ loop zs' Nothing Just _ -> loop zs' (liftM2 min t acc) --- | Start the time server.-timeServer :: TimeServerParams -> DP.Process ()-timeServer ps =- do server <- liftIO $ newTimeServer ps+-- | Convert seconds to microseconds.+secondsToMicroseconds :: Double -> Int+secondsToMicroseconds x = fromInteger $ toInteger $ round (1000000 * x)++-- | Start the time server by the specified initial quorum and parameters.+-- The quorum defines the number of local processes that must be registered in+-- the time server before the global time synchronization is started.+timeServer :: Int -> TimeServerParams -> DP.Process ()+timeServer n ps =+ do server <- liftIO $ newTimeServer n ps logTimeServer server INFO "Time Server: starting..."- forever $- do m <- DP.expectTimeout (tsExpectTimeout ps) :: DP.Process (Maybe TimeServerMessage)- case m of- Nothing -> return ()- Just m ->- do ---- logTimeServer server DEBUG $- "Time Server: " ++ show m- ---- processTimeServerMessage server m- liftIO $- threadDelay (tsTimeSyncDelay ps)- validateTimeServer server+ let loop utc0 =+ do m <- DP.expectTimeout (tsExpectTimeout ps) :: DP.Process (Maybe TimeServerMessage)+ case m of+ Nothing -> return ()+ Just m ->+ do ---+ logTimeServer server DEBUG $+ "Time Server: " ++ show m+ ---+ processTimeServerMessage server m+ utc <- liftIO getCurrentTime+ let dt = fromRational $ toRational (diffUTCTime utc utc0)+ if secondsToMicroseconds dt > tsTimeSyncDelay ps+ then do tryComputeTimeServerGlobalTime server+ loop utc+ else loop utc0+ liftIO getCurrentTime >>= loop +-- | A curried version of 'timeServer' for starting the time server on remote node.+curryTimeServer :: (Int, TimeServerParams) -> DP.Process ()+curryTimeServer (n, ps) = timeServer n ps+ -- | Log the message with the specified priority. logTimeServer :: TimeServer -> Priority -> String -> DP.Process () {-# INLINE logTimeServer #-}@@ -244,4 +301,3 @@ when (tsLoggingPriority (tsParams server) <= p) $ DP.say $ embracePriority p ++ " " ++ message-
+ Simulation/Aivika/Distributed/Optimistic/Internal/TransientMessageQueue.hs view
@@ -0,0 +1,163 @@++-- |+-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue+-- Copyright : Copyright (c) 2015-2016, David Sorokin <david.sorokin@gmail.com>+-- License : BSD3+-- Maintainer : David Sorokin <david.sorokin@gmail.com>+-- Stability : experimental+-- Tested with: GHC 8.0.1+--+-- This module defines a transient message queue.+--+module Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue+ (TransientMessageQueue,+ newTransientMessageQueue,+ transientMessageQueueSize,+ transientMessageQueueTime,+ enqueueTransientMessage,+ processAcknowledgmentMessage,+ acknowledgmentMessageTime,+ resetAcknowledgmentMessageTime,+ deliverAcknowledgmentMessage,+ deliverAcknowledgmentMessages) where++import qualified Data.Set as S+import Data.List+import Data.IORef++import Control.Monad+import Control.Monad.Trans+import qualified Control.Distributed.Process as DP++import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.Internal.DIO+import Simulation.Aivika.Distributed.Optimistic.Internal.IO++-- | Specifies the transient message queue.+data TransientMessageQueue =+ TransientMessageQueue { queuePrototypeMessages :: IORef (S.Set TransientMessageQueueItem),+ -- ^ The prototype messages.+ queueMarkedMessageTime :: IORef Double+ -- ^ The marked acknowledgment message time.+ }++-- | Represents an acknowledgement message representative.+data TransientMessageQueueItem =+ TransientMessageQueueItem { itemSequenceNo :: Int,+ -- ^ The sequence number.+ itemSendTime :: Double,+ -- ^ The send time.+ itemReceiveTime :: Double,+ -- ^ The receive time.+ itemSenderId :: DP.ProcessId,+ -- ^ The sender of the source message.+ itemReceiverId :: DP.ProcessId,+ -- ^ The receiver of the source message.+ itemAntiToggle :: Bool+ -- ^ Whether it was the anti-message acknowledgment.+ } deriving (Eq, Show)++instance Ord TransientMessageQueueItem where++ x <= y+ | (itemReceiveTime x < itemReceiveTime y) = True+ | (itemReceiveTime x > itemReceiveTime y) = False+ | (itemSendTime x < itemSendTime y) = True+ | (itemSendTime x > itemSendTime y) = False+ | (itemSequenceNo x < itemSequenceNo y) = True+ | (itemSequenceNo x > itemSequenceNo y) = False+ | (itemReceiverId x < itemReceiverId y) = True+ | (itemReceiverId x > itemReceiverId y) = False+ | (itemSenderId x < itemSenderId y) = True+ | (itemSenderId x > itemSenderId y) = False+ | (itemAntiToggle x < itemAntiToggle y) = True+ | (itemAntiToggle x > itemAntiToggle y) = False+ | otherwise = True++-- | Return a queue item by the specified transient message.+transientMessageQueueItem :: Message -> TransientMessageQueueItem+transientMessageQueueItem m =+ TransientMessageQueueItem { itemSequenceNo = messageSequenceNo m,+ itemSendTime = messageSendTime m,+ itemReceiveTime = messageReceiveTime m,+ itemSenderId = messageSenderId m,+ itemReceiverId = messageReceiverId m,+ itemAntiToggle = messageAntiToggle m }++-- | Return a queue item by the specified acknowledgment message.+acknowledgmentMessageQueueItem :: AcknowledgmentMessage -> TransientMessageQueueItem+acknowledgmentMessageQueueItem m =+ TransientMessageQueueItem { itemSequenceNo = acknowledgmentSequenceNo m,+ itemSendTime = acknowledgmentSendTime m,+ itemReceiveTime = acknowledgmentReceiveTime m,+ itemSenderId = acknowledgmentSenderId m,+ itemReceiverId = acknowledgmentReceiverId m,+ itemAntiToggle = acknowledgmentAntiToggle m }++-- | Create a new transient message queue.+newTransientMessageQueue :: DIO TransientMessageQueue+newTransientMessageQueue =+ do ms <- liftIOUnsafe $ newIORef S.empty+ r <- liftIOUnsafe $ newIORef (1 / 0)+ return TransientMessageQueue { queuePrototypeMessages = ms,+ queueMarkedMessageTime = r }++-- | Get the transient message queue size.+transientMessageQueueSize :: TransientMessageQueue -> IO Int+transientMessageQueueSize q =+ fmap S.size $ readIORef (queuePrototypeMessages q)++-- | Get the virtual time of the transient message queue.+transientMessageQueueTime :: TransientMessageQueue -> IO Double+transientMessageQueueTime q =+ do s <- readIORef (queuePrototypeMessages q)+ if S.null s+ then return (1 / 0)+ else let m = S.findMin s+ in return (itemReceiveTime m)++-- | Enqueue the transient message.+enqueueTransientMessage :: TransientMessageQueue -> Message -> IO ()+enqueueTransientMessage q m =+ modifyIORef (queuePrototypeMessages q) $+ S.insert (transientMessageQueueItem m)++-- | Enqueue the acknowledgment message.+enqueueAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO ()+enqueueAcknowledgmentMessage q m =+ modifyIORef' (queueMarkedMessageTime q) $+ min (acknowledgmentReceiveTime m)++-- | Process the acknowledgment message.+processAcknowledgmentMessage :: TransientMessageQueue -> AcknowledgmentMessage -> IO () +processAcknowledgmentMessage q m =+ do modifyIORef (queuePrototypeMessages q) $+ S.delete (acknowledgmentMessageQueueItem m)+ when (acknowledgmentMarked m) $+ enqueueAcknowledgmentMessage q m++-- | Get the minimal marked acknowledgment message time.+acknowledgmentMessageTime :: TransientMessageQueue -> IO Double+acknowledgmentMessageTime q =+ readIORef (queueMarkedMessageTime q)++-- | Reset the marked acknowledgment message time.+resetAcknowledgmentMessageTime :: TransientMessageQueue -> IO ()+resetAcknowledgmentMessageTime q =+ writeIORef (queueMarkedMessageTime q) (1 / 0)++-- | Deliver the acknowledgment message on low level.+deliverAcknowledgmentMessage :: AcknowledgmentMessage -> DIO ()+deliverAcknowledgmentMessage x =+ liftDistributedUnsafe $+ DP.send (acknowledgmentSenderId x) (AcknowledgmentQueueMessage x)++-- | Deliver the acknowledgment messages on low level.+deliverAcknowledgmentMessages :: [AcknowledgmentMessage] -> DIO ()+deliverAcknowledgmentMessages xs =+ let ys = groupBy (\a b -> acknowledgmentSenderId a == acknowledgmentSenderId b) xs+ dlv [] = return ()+ dlv zs@(z : _) =+ liftDistributedUnsafe $+ DP.send (acknowledgmentSenderId z) (AcknowledgmentQueueMessageBulk zs)+ in forM_ ys dlv
Simulation/Aivika/Distributed/Optimistic/TimeServer.hs view
@@ -12,6 +12,7 @@ module Simulation.Aivika.Distributed.Optimistic.TimeServer (TimeServerParams(..), defaultTimeServerParams,- timeServer) where+ timeServer,+ curryTimeServer) where import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer
aivika-distributed.cabal view
@@ -1,9 +1,10 @@ name: aivika-distributed-version: 0.2+version: 0.3 synopsis: Parallel distributed discrete event simulation module for the Aivika library description: This package extends the aivika-transformers [1] package with facilities for running parallel distributed simulations.- It uses an optimistic strategy known as the Time Warp method.+ It uses an optimistic strategy known as the Time Warp method. To synchronize the global virtual time, since version 0.3+ it uses Samadi's algorithm. . \[1] <http://hackage.haskell.org/package/aivika-transformers> .@@ -25,6 +26,7 @@ tests/MachRep2Reproducible.hs tests/MachRep2Sync.hs tests/MachRep2SyncIO.hs+ CHANGELOG.md library @@ -49,6 +51,7 @@ Simulation.Aivika.Distributed.Optimistic.Internal.Ref Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer Simulation.Aivika.Distributed.Optimistic.Internal.TimeWarp+ Simulation.Aivika.Distributed.Optimistic.Internal.TransientMessageQueue Simulation.Aivika.Distributed.Optimistic.Internal.UndoableLog build-depends: base >= 3 && < 6,
tests/MachRep1.hs view
@@ -104,7 +104,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = NOTICE }- m = do runSimulation (slaveModel masterId) specs+ m = do registerDIO+ runSimulation (slaveModel masterId) specs unregisterDIO runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process Double)@@ -112,14 +113,15 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = NOTICE }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a master = \backend nodes -> do liftIO . putStrLn $ "Slaves: " ++ show nodes let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams (masterId, masterProcess) <- runMasterModel timeServerId 2 forM_ [1..2] $ \i -> do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)
tests/MachRep1Simple.hs view
@@ -71,7 +71,8 @@ do DP.say "Started simulating..." let ps = defaultDIOParams { dioLoggingPriority = NOTICE } m =- do a <- runSimulation model specs+ do registerDIO+ a <- runSimulation model specs terminateDIO return a (modelId, modelProcess) <- runDIO m ps timeServerId@@ -81,7 +82,7 @@ master = \backend nodes -> do liftIO . putStrLn $ "Slaves: " ++ show nodes let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 1 timeServerParams runModel timeServerId liftIO $ threadDelay 1000000
tests/MachRep2.hs view
@@ -197,7 +197,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do runSimulation (slaveModel masterId) specs+ m = do registerDIO+ runSimulation (slaveModel masterId) specs unregisterDIO runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))@@ -205,7 +206,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a @@ -213,7 +215,7 @@ do liftIO . putStrLn $ "Slaves: " ++ show nodes let n = 2 timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams (masterId, masterProcess) <- runMasterModel timeServerId n forM_ [1..n] $ \i -> do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)
tests/MachRep2Distributed.hs view
@@ -216,7 +216,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do runSimulation (slaveModel masterId) specs+ m = do registerDIO+ runSimulation (slaveModel masterId) specs unregisterDIO startSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process ()@@ -224,14 +225,15 @@ do (slaveId, slaveProcess) <- runSlaveModel x slaveProcess -remotable ['startSlaveModel, 'timeServer]+remotable ['startSlaveModel, 'curryTimeServer] runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double)) runMasterModel timeServerId n = runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a @@ -240,7 +242,7 @@ let [n0, n1, n2] = nodes timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG } -- timeServerId <- DP.spawnLocal $ timeServer timeServerParams- timeServerId <- DP.spawn n0 ($(mkClosure 'timeServer) timeServerParams)+ timeServerId <- DP.spawn n0 ($(mkClosure 'curryTimeServer) (3 :: Int, timeServerParams)) -- (masterId, masterProcess) <- runMasterModel timeServerId 2 -- forM_ [1..2] $ \i -> -- do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)
tests/MachRep2Reproducible.hs view
@@ -213,7 +213,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do runSimulation (slaveModel masterId i) specs+ m = do registerDIO+ runSimulation (slaveModel masterId i) specs unregisterDIO -- remotable ['runSlaveModel, 'timeServer]@@ -223,7 +224,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a @@ -231,7 +233,7 @@ do liftIO . putStrLn $ "Slaves: " ++ show nodes let n = length seeds timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams (masterId, masterProcess) <- runMasterModel timeServerId n forM_ [1..n] $ \i -> do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId, i)
tests/MachRep2Sync.hs view
@@ -202,7 +202,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do runSimulation (slaveModel masterId) specs+ m = do registerDIO+ runSimulation (slaveModel masterId) specs unregisterDIO runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))@@ -210,7 +211,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a @@ -218,7 +220,7 @@ do liftIO . putStrLn $ "Slaves: " ++ show nodes let n = 2 timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams (masterId, masterProcess) <- runMasterModel timeServerId n forM_ [1..n] $ \i -> do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)
tests/MachRep2SyncIO.hs view
@@ -204,7 +204,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do runSimulation (slaveModel masterId) specs+ m = do registerDIO+ runSimulation (slaveModel masterId) specs unregisterDIO runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))@@ -212,7 +213,8 @@ runDIO m ps timeServerId where ps = defaultDIOParams { dioLoggingPriority = WARNING }- m = do a <- runSimulation (masterModel n) specs+ m = do registerDIO+ a <- runSimulation (masterModel n) specs terminateDIO return a @@ -220,7 +222,7 @@ do liftIO . putStrLn $ "Slaves: " ++ show nodes let n = 2 timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+ timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams (masterId, masterProcess) <- runMasterModel timeServerId n forM_ [1..n] $ \i -> do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)