aivika-distributed 0.7.2 → 0.7.3

raw patch · 16 files changed

+1526/−12 lines, 16 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• Simulation/Aivika/Distributed/Optimistic.hs +2/−0
• Simulation/Aivika/Distributed/Optimistic/DIO.hs +3/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +65/−6
• Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +52/−4
• Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +7/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +6/−0
• Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +65/−1
• Simulation/Aivika/Distributed/Optimistic/State.hs +68/−0
• Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +3/−0
• aivika-distributed.cabal +11/−1
• tests/MachRep1SimpleWithMonitoring.hs +100/−0
• tests/MachRep2DistributedReproducible.hs +286/−0
• tests/MachRep2DistributedReproducibleFaultTolerant.hs +301/−0
• tests/MachRep2DistributedWithMonitoring.hs +311/−0
• tests/MachRep2WithMonitoring.hs +241/−0

@@ -1,4 +1,9 @@ +Version 0.7.3+-----++* Updated so that external software tools could monitor the distributed simulation.+ Version 0.7.2 ----- 

@@ -19,6 +19,7 @@         module Simulation.Aivika.Distributed.Optimistic.QueueStrategy,         module Simulation.Aivika.Distributed.Optimistic.Priority,         module Simulation.Aivika.Distributed.Optimistic.Ref.Base,+        module Simulation.Aivika.Distributed.Optimistic.State,         module Simulation.Aivika.Distributed.Optimistic.TimeServer) where  import Simulation.Aivika.Distributed.Optimistic.DIO@@ -28,4 +29,5 @@ import Simulation.Aivika.Distributed.Optimistic.QueueStrategy import Simulation.Aivika.Distributed.Optimistic.Priority import Simulation.Aivika.Distributed.Optimistic.Ref.Base+import Simulation.Aivika.Distributed.Optimistic.State import Simulation.Aivika.Distributed.Optimistic.TimeServer

@@ -14,9 +14,12 @@ module Simulation.Aivika.Distributed.Optimistic.DIO        (DIO,         DIOParams(..),+        DIOEnv(..),         DIOStrategy(..),         runDIO,+        runDIOWithEnv,         defaultDIOParams,+        defaultDIOEnv,         dioParams,         messageInboxId,         timeServerId,

@@ -14,10 +14,13 @@ module Simulation.Aivika.Distributed.Optimistic.Internal.DIO        (DIO(..),         DIOParams(..),+        DIOEnv(..),         DIOStrategy(..),         invokeDIO,         runDIO,+        runDIOWithEnv,         defaultDIOParams,+        defaultDIOEnv,         terminateDIO,         registerDIO,         unregisterDIO,@@ -61,11 +64,14 @@ import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager+import Simulation.Aivika.Distributed.Optimistic.State  -- | The parameters for the 'DIO' computation. data DIOParams =   DIOParams { dioLoggingPriority :: Priority,               -- ^ The logging priority+              dioName :: String,+              -- ^ The name of the logical process.               dioUndoableLogSizeThreshold :: Int,               -- ^ The undoable log size threshold used for detecting an overflow               dioOutputMessageQueueSizeThreshold :: Int,@@ -91,12 +97,22 @@               -- ^ The interval in microseconds for sending keep-alive messages               dioTimeServerAcknowledgementTimeout :: Int,               -- ^ The timeout in microseconds for receiving an acknowledgement message from the time server+              dioSimulationMonitoringInterval :: Int,+              -- ^ The interval in microseconds between sending the simulation monitoring messages+              dioSimulationMonitoringTimeout :: Int,+              -- ^ The timeout in microseconds when processing the monitoring messages               dioStrategy :: DIOStrategy               -- ^ The logical process strategy             } deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary DIOParams +-- | Those 'DIO' environment parameters that cannot be serialized and passed to another process via the net.+data DIOEnv =+  DIOEnv { dioSimulationMonitoringAction :: Maybe (LogicalProcessState -> DP.Process ())+           -- ^ The simulation monitoring action+         }+ -- | The logical process strategy. data DIOStrategy = WaitIndefinitelyForTimeServer                    -- ^ Wait for the time server forever@@ -176,6 +192,7 @@ defaultDIOParams :: DIOParams defaultDIOParams =   DIOParams { dioLoggingPriority = WARNING,+              dioName = "LP",               dioUndoableLogSizeThreshold = 1000000,               dioOutputMessageQueueSizeThreshold = 10000,               dioTransientMessageQueueSizeThreshold = 10000,@@ -188,9 +205,16 @@               dioProcessReconnectingDelay = 5000000,               dioKeepAliveInterval = 5000000,               dioTimeServerAcknowledgementTimeout = 5000000,+              dioSimulationMonitoringInterval = 30000000,+              dioSimulationMonitoringTimeout = 100000,               dioStrategy = TerminateDueToTimeServerTimeout 300000000             } +-- | The default environment parameters for the 'DIO' computation+defaultDIOEnv :: DIOEnv+defaultDIOEnv =+  DIOEnv { dioSimulationMonitoringAction = Nothing }+ -- | Return the computation context. dioContext :: DIO DIOContext dioContext = DIO return@@ -295,7 +319,11 @@ -- | Run the computation using the specified parameters along with time server process -- identifier and return the inbox process identifier and a new simulation process. runDIO :: DIO a -> DIOParams -> DP.ProcessId -> DP.Process (DP.ProcessId, DP.Process a)-runDIO m ps serverId =+runDIO m ps serverId = runDIOWithEnv m ps defaultDIOEnv serverId++-- | A full version of 'runDIO' that also allows specifying the environment parameters.+runDIOWithEnv :: DIO a -> DIOParams -> DIOEnv -> DP.ProcessId -> DP.Process (DP.ProcessId, DP.Process a)+runDIOWithEnv m ps env serverId =   do ch <- liftIO newChannel      let keepAliveParams =            KeepAliveParams { keepAliveLoggingPriority = dioLoggingPriority ps,@@ -306,7 +334,7 @@      unregisteredFromTimeServer <- liftIO $ newTVarIO False      timeServerTerminating <- liftIO $ newTVarIO False      timeServerTimestamp <- liftIO $ getCurrentTime >>= newIORef-     let loop =+     let loop0 =            forever $            do let f1 :: LogicalProcessMessage -> DP.Process InternalLogicalProcessMessage                   f1 x = return (InternalLogicalProcessMessage x)@@ -351,10 +379,12 @@                       do ---                          logProcess ps INFO $ "Monitoring process " ++ show pid                          ----                         DP.monitor pid-                         liftIO $-                           addKeepAliveReceiver keepAliveManager pid-                         return ()+                         f <- liftIO $+                              existsKeepAliveReceiver keepAliveManager pid+                         unless f $+                           do DP.monitor pid+                              liftIO $+                                addKeepAliveReceiver keepAliveManager pid                     ReMonitorProcessMessage pids ->                       handleReMonitorProcessMessage pids ps ch                     TrySendKeepAliveMessage ->@@ -393,6 +423,11 @@                      logProcess ps DEBUG $ "Received " ++ show m                      ---                      return ()+         loop =+           C.finally loop0+           (liftIO $+            do atomicWriteIORef terminated True+               writeChannel ch AbortSimulationMessage)      inboxId <-        DP.spawnLocal $        C.catch loop (handleException ps)@@ -419,6 +454,30 @@                        do validateTimeServer ps inboxId timeServerTimestamp                           loop        in C.catch loop (handleException ps)+     case dioSimulationMonitoringAction env of+       Nothing  -> return ()+       Just act ->+         do monitorId <-+              DP.spawnLocal $+              let loop =+                    do f <- liftIO $ readIORef terminated+                       unless f $+                         do x <- DP.expectTimeout (dioSimulationMonitoringTimeout ps)+                            case x of+                              Nothing -> return ()+                              Just st -> act st+                            loop+              in C.catch loop (handleException ps)+            DP.spawnLocal $+              let loop =+                    do f <- liftIO $ readIORef terminated+                       unless f $+                         do liftIO $+                              do threadDelay (dioSimulationMonitoringInterval ps)+                                 writeChannel ch (ProvideLogicalProcessStateMessage monitorId)+                            loop+              in C.catch loop (handleException ps)+            return ()      let simulation =            unDIO m DIOContext { dioChannel = ch,                                 dioInboxId = inboxId,

@@ -50,6 +50,7 @@ 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@@ -422,11 +423,23 @@      ---      invokeEvent p $        reconnectProcess pid-processChannelMessage AbortSimulationMessage =+processChannelMessage x@(ProvideLogicalProcessStateMessage pid) =   TimeWarp $ \p ->-  invokeEvent p $-  throwEvent $-  SimulationAbort "Aborted by the outer process."+  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@@ -825,3 +838,38 @@                   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 }

@@ -14,6 +14,7 @@         KeepAliveParams(..),         newKeepAliveManager,         addKeepAliveReceiver,+        existsKeepAliveReceiver,         trySendKeepAlive,         trySendKeepAliveUTC) where @@ -61,6 +62,12 @@ addKeepAliveReceiver manager pid =   modifyIORef (keepAliveReceivers manager) $   S.insert pid++-- | Whether the keep-alive message receiver exists.+existsKeepAliveReceiver :: KeepAliveManager -> DP.ProcessId -> IO Bool+existsKeepAliveReceiver manager pid =+  readIORef (keepAliveReceivers manager) >>=+  return . S.member pid  -- | Try to send a keep-alive message. trySendKeepAlive :: KeepAliveManager -> DP.Process ()

@@ -30,6 +30,8 @@ import qualified Control.Distributed.Process as DP import Control.Distributed.Process.Serializable +import Simulation.Aivika.Distributed.Optimistic.State+ -- | Represents a message. data Message =   Message { messageSequenceNo :: Int,@@ -123,6 +125,8 @@                              -- ^ the process monitor notification                            | ReconnectProcessMessage DP.ProcessId                              -- ^ finish reconnecting to the specified process+                           | ProvideLogicalProcessStateMessage DP.ProcessId+                             -- ^ provide the logical process state                            | AbortSimulationMessage                              -- ^ abort the simulation                            deriving (Show, Typeable, Generic)@@ -142,6 +146,8 @@                          -- ^ the logical process requested for the global minimum time                        | LocalTimeMessage DP.ProcessId Double                          -- ^ the logical process sent its local minimum time+                       | ProvideTimeServerStateMessage DP.ProcessId+                         -- ^ send the time server monitoring state message                        | ReMonitorTimeServerMessage [DP.ProcessId]                          -- ^ re-monitor the logical processes by their identifiers                        deriving (Eq, Show, Typeable, Generic)

@@ -13,9 +13,12 @@ -- module Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer        (TimeServerParams(..),+        TimeServerEnv(..),         TimeServerStrategy(..),         defaultTimeServerParams,+        defaultTimeServerEnv,         timeServer,+        timeServerWithEnv,         curryTimeServer) where  import qualified Data.Map as M@@ -37,11 +40,14 @@  import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.Message+import Simulation.Aivika.Distributed.Optimistic.State  -- | The time server parameters. data TimeServerParams =   TimeServerParams { tsLoggingPriority :: Priority,                      -- ^ the logging priority+                     tsName :: String,+                     -- ^ the monitoring name of the time server                      tsReceiveTimeout :: Int,                      -- ^ the timeout in microseconds used when receiving messages                      tsTimeSyncTimeout :: Int,@@ -56,12 +62,22 @@                      -- ^ whether the automatic reconnecting to processes is enabled when enabled monitoring                      tsProcessReconnectingDelay :: Int,                      -- ^ the delay in microseconds before reconnecting+                     tsSimulationMonitoringInterval :: Int,+                     -- ^ the interval in microseconds between sending the simulation monitoring messages+                     tsSimulationMonitoringTimeout :: Int,+                     -- ^ the timeout in microseconds when processing the simulation monitoring messages                      tsStrategy :: TimeServerStrategy                      -- ^ the time server strategy                    } deriving (Eq, Ord, Show, Typeable, Generic)  instance Binary TimeServerParams +-- | Those time server environment parameters that cannot be serialized and passed to another process via the net.+data TimeServerEnv =+  TimeServerEnv { tsSimulationMonitoringAction :: Maybe (TimeServerState -> DP.Process ())+                  -- ^ the simulation monitoring action+                }+ -- | The time server strategy. data TimeServerStrategy = WaitIndefinitelyForLogicalProcess                           -- ^ wait for the logical process forever@@ -117,6 +133,7 @@ defaultTimeServerParams :: TimeServerParams defaultTimeServerParams =   TimeServerParams { tsLoggingPriority = WARNING,+                     tsName = "Time Server",                      tsReceiveTimeout = 100000,                      tsTimeSyncTimeout = 60000000,                      tsTimeSyncDelay = 1000000,@@ -124,9 +141,16 @@                      tsProcessMonitoringDelay = 3000000,                      tsProcessReconnectingEnabled = False,                      tsProcessReconnectingDelay = 5000000,+                     tsSimulationMonitoringInterval = 30000000,+                     tsSimulationMonitoringTimeout = 100000,                      tsStrategy = TerminateDueToLogicalProcessTimeout 300000000                    } +-- | The default time server environment parameters.+defaultTimeServerEnv :: TimeServerEnv+defaultTimeServerEnv =+  TimeServerEnv { tsSimulationMonitoringAction = Nothing }+ -- | Create a new time server by the specified initial quorum and parameters. newTimeServer :: Int -> TimeServerParams -> IO TimeServer newTimeServer n ps =@@ -258,6 +282,17 @@             writeIORef (lpTimestamp x) utc             return $               return ()+processTimeServerMessage server (ProvideTimeServerStateMessage pid) =+  do let ps   = tsParams server+         name = tsName ps+     serverId <- DP.getSelfPid+     t <- liftIO $ readIORef (tsGlobalTime server)+     m <- liftIO $ readIORef (tsProcesses server)+     let msg = TimeServerState { tsStateId = serverId,+                                 tsStateName = name,+                                 tsStateGlobalVirtualTime = t,+                                 tsStateLogicalProcesses = M.keys m }+     DP.send pid msg processTimeServerMessage server (ReMonitorTimeServerMessage pids) =   do forM_ pids $ \pid ->        do ---@@ -451,7 +486,11 @@ -- The quorum defines the number of logical processes that must be registered in -- the time server before the global time synchronization is started. timeServer :: Int -> TimeServerParams -> DP.Process ()-timeServer n ps =+timeServer n ps = timeServerWithEnv n ps defaultTimeServerEnv++-- | A full version of 'timeServer' that allows specifying the environment parameters.+timeServerWithEnv :: Int -> TimeServerParams -> TimeServerEnv -> DP.Process ()+timeServerWithEnv n ps env =   do server <- liftIO $ newTimeServer n ps      logTimeServer server INFO "Time Server: starting..."      let loop utc0 =@@ -491,6 +530,31 @@                 then do tryComputeTimeServerGlobalTime server                         loop utc                 else loop utc0+     case tsSimulationMonitoringAction env of+       Nothing  -> return ()+       Just act ->+         do serverId  <- DP.getSelfPid+            monitorId <-+              DP.spawnLocal $+              let loop =+                    do f <- liftIO $ readIORef (tsTerminating server)+                       unless f $+                         do x <- DP.expectTimeout (tsSimulationMonitoringTimeout ps)+                            case x of+                              Nothing -> return ()+                              Just st -> act st+                            loop+              in C.catch loop (handleTimeServerException server)+            DP.spawnLocal $+              let loop =+                    do f <- liftIO $ readIORef (tsTerminating server)+                       unless f $+                         do liftIO $+                              threadDelay (tsSimulationMonitoringInterval ps)+                            DP.send serverId (ProvideTimeServerStateMessage monitorId)+                            loop+              in C.catch loop (handleTimeServerException server)+            return ()      C.catch (liftIO getCurrentTime >>= loop) (handleTimeServerException server)   -- | Handle the process monitor notification.

@@ -0,0 +1,68 @@++{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-}++-- |+-- Module     : Simulation.Aivika.Distributed.Optimistic.State+-- 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+--+-- This module defines the monitoring states.+--+module Simulation.Aivika.Distributed.Optimistic.State+       (LogicalProcessState(..),+        TimeServerState(..)) where++import GHC.Generics++import Data.Typeable+import Data.Binary++import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Serializable++-- | Represents the state of the logical process.+data LogicalProcessState =+  LogicalProcessState { lpStateId :: DP.ProcessId,+                        -- ^ the process identifier+                        lpStateName :: String,+                        -- ^ the process name+                        lpStateStartTime :: Double,+                        -- ^ the start time+                        lpStateStopTime :: Double,+                        -- ^ the stop time+                        lpStateLocalTime :: Double,+                        -- ^ the local time of the process+                        lpStateEventQueueTime :: Double,+                        -- ^ the event queue time of the process+                        lpStateEventQueueSize :: Int,+                        -- ^ the event queue size+                        lpStateLogSize :: Int,+                        -- ^ the log size of the process+                        lpStateInputMessageCount :: Int,+                        -- ^ the count of the input messages+                        lpStateOutputMessageCount :: Int,+                        -- ^ the count of the output messages+                        lpStateTransientMessageCount :: Int,+                        -- ^ the count of the transient messages that did not receive an acknowledgement+                        lpStateRollbackCount :: Int+                        -- ^ the count of rollbacks+                      } deriving (Eq, Show, Typeable, Generic)++instance Binary LogicalProcessState++-- | Represents the state of the time server.+data TimeServerState =+  TimeServerState { tsStateId :: DP.ProcessId,+                    -- ^ the process identifier+                    tsStateName :: String,+                    -- ^ the time server name+                    tsStateGlobalVirtualTime :: Maybe Double,+                    -- ^ the global virtual time+                    tsStateLogicalProcesses :: [DP.ProcessId]+                    -- ^ the registered logical process identifiers+                  } deriving (Eq, Show, Typeable, Generic)++instance Binary TimeServerState

@@ -11,9 +11,12 @@ -- module Simulation.Aivika.Distributed.Optimistic.TimeServer        (TimeServerParams(..),+        TimeServerEnv(..),         TimeServerStrategy(..),         defaultTimeServerParams,+        defaultTimeServerEnv,         timeServer,+        timeServerWithEnv,         curryTimeServer) where  import Simulation.Aivika.Distributed.Optimistic.Internal.TimeServer

@@ -1,5 +1,5 @@ name:            aivika-distributed-version:         0.7.2+version:         0.7.3 synopsis:        Parallel distributed discrete event simulation module for the Aivika library description:     This package extends the aivika-transformers [1] package and allows running parallel distributed simulations.@@ -34,12 +34,16 @@     dramatically decrease the speed of distributed simulation, especially if they cause rollbacks. Thus, much depends on      the distributed model itself.     .+    Finally, you can use the following test model [4] as an example.+    .     \[1] <http://hackage.haskell.org/package/aivika-transformers>     .     \[2] <http://hackage.haskell.org/package/aivika>     .     \[3] <http://www.aivikasoft.com>     .+    \[4] <https://github.com/dsorokin/aivika-distributed-test>+    . category:        Simulation license:         BSD3 license-file:    LICENSE@@ -54,11 +58,16 @@ extra-source-files:  tests/Guard1.hs                      tests/MachRep1.hs                      tests/MachRep1Simple.hs+                     tests/MachRep1SimpleWithMonitoring.hs                      tests/MachRep2.hs                      tests/MachRep2Distributed.hs+                     tests/MachRep2DistributedReproducible.hs+                     tests/MachRep2DistributedReproducibleFaultTolerant.hs+                     tests/MachRep2DistributedWithMonitoring.hs                      tests/MachRep2Reproducible.hs                      tests/MachRep2Sync.hs                      tests/MachRep2SyncIO.hs+                     tests/MachRep2WithMonitoring.hs                      tests/SimpleLocalnetHelper.hs                      tests/cluster.conf                      CHANGELOG.md@@ -76,6 +85,7 @@                      Simulation.Aivika.Distributed.Optimistic.Ref.Base                      Simulation.Aivika.Distributed.Optimistic.Ref.Base.Lazy                      Simulation.Aivika.Distributed.Optimistic.Ref.Base.Strict+                     Simulation.Aivika.Distributed.Optimistic.State                      Simulation.Aivika.Distributed.Optimistic.TimeServer      other-modules:   Simulation.Aivika.Distributed.Optimistic.Internal.Channel

@@ -0,0 +1,100 @@++-- It corresponds to model MachRep1 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--+-- Two machines, which sometimes break down.+-- Up time is exponentially distributed with mean 1.0, and repair time is+-- exponentially distributed with mean 0.5. There are two repairpersons,+-- so the two machines can be repaired simultaneously if they are down+-- at the same time.+--+-- Output is long-run proportion of up time. Should get value of about+-- 0.66.++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 10000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }+        +model :: Simulation DIO Double+model =+  do totalUpTime <- newRef 0.0+     +     let machine =+           do upTime <-+                liftParameter $+                randomExponential meanUpTime+              holdProcess upTime+              liftEvent $ +                modifyRef totalUpTime (+ upTime)+              repairTime <-+                liftParameter $+                randomExponential meanRepairTime+              holdProcess repairTime+              machine++     runProcessInStartTime machine+     runProcessInStartTime machine++     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (2 * y)++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       liftIO $+       putStrLn "Test IO"+     +     runEventInStopTime upTimeProp++runModel :: DP.ProcessId -> DP.Process ()+runModel timeServerId =+  do DP.say "Started simulating..."+     let ps = defaultDIOParams { dioSimulationMonitoringInterval = 5000000 }+         env = defaultDIOEnv { dioSimulationMonitoringAction = Just monitorSimulation }+         m =+           do registerDIO+              a <- runSimulation model specs+              terminateDIO+              return a+     (modelId, modelProcess) <- runDIOWithEnv m ps env timeServerId+     a <- modelProcess+     DP.say $ "The result is " ++ show a++monitorSimulation :: Show a => a -> DP.Process ()+monitorSimulation = DP.say . show++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let timeServerParams = defaultTimeServerParams { tsSimulationMonitoringInterval = 5000000 }+         env = defaultTimeServerEnv { tsSimulationMonitoringAction = Just monitorSimulation }+     timeServerId  <- DP.spawnLocal $ timeServerWithEnv 1 timeServerParams env+     runModel timeServerId+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" rtable+  startMaster backend (master backend)+    where+      rtable :: DP.RemoteTable+      -- rtable = __remoteTable initRemoteTable+      rtable = initRemoteTable

@@ -0,0 +1,286 @@++{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++-- Compiling and Running+---+-- Compile using+--+-- ghc -threaded MachRep2Distributed.hs+--+-- Fire up some slave nodes (for the example, we run them on a single machine):+--+-- ./MachRep2DistributedReproducible slave localhost 8080 &+-- ./MachRep2DistributedReproducible slave localhost 8081 &+-- ./MachRep2DistributedReproducible slave localhost 8082 &+--+-- And start the master node:+--+-- ./MachRep2DistributedReproducible master localhost 8088+--++import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import System.Random++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying to the messages.+delta = 1e-6++-- | The initial seeds.+seeds = [456, 789]++-- | Create a new random generator by the specified seed.+newRandomRef :: Int -> Simulation DIO (Ref DIO StdGen)+newRandomRef = newRef . mkStdGen++-- | Generate a random number with the specified mean+randomRefExponential :: RandomGen g => Ref DIO g -> Double -> Event DIO Double+randomRefExponential r mu =+  do g <- readRef r+     let (x, g') = random g+     writeRef r g'+     return (- log x * mu)++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Int -> Simulation DIO ()+slaveModel masterId i =+  do inboxId <- liftComp messageInboxId+     g <- newRandomRef $ seeds !! (i - 1)++     let machine =+           do t <- liftDynamics time+              upTime <- randomRefExponential g meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <- randomRefExponential g meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       liftIO $+       putStrLn "The sub-model finished"++     runEventInStopTime $+       return ()++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId, i) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = WARNING }+    m  = do registerDIO+            runSimulation (slaveModel masterId i) specs+            unregisterDIO++startSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process ()+startSlaveModel x@(timeServerId, masterId, i) =+  do (slaveId, slaveProcess) <- runSlaveModel x+     slaveProcess++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 registerDIO+            a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let [n0, n1, n2] = nodes+         timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG }+     -- timeServerId <- DP.spawnLocal $ 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)+     --      DP.spawnLocal slaveProcess+     (masterId, masterProcess) <- runMasterModel timeServerId 2+     forM_ (zip [n1, n2] [(1 :: Int)..]) $ \(node, i) ->+        DP.spawn node ($(mkClosure 'startSlaveModel) (timeServerId, masterId, i))+     a <- masterProcess+     DP.say $+       "The result is " ++ show a++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["master", host, port] -> do+      backend <- initializeBackend host port rtable+      startMaster backend (master backend)+    ["slave", host, port] -> do+      backend <- initializeBackend host port rtable+      startSlave backend+  where+    rtable :: DP.RemoteTable+    rtable = __remoteTable initRemoteTable

@@ -0,0 +1,301 @@++{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++-- Compiling and Running+---+-- Compile using+--+-- ghc -threaded MachRep2Distributed.hs+--+-- Fire up some slave nodes (for the example, we run them on a single machine):+--+-- ./MachRep2DistributedReproducibleFaultTolerant slave 1 &+-- ./MachRep2DistributedReproducibleFaultTolerant slave 2 &+-- ./MachRep2DistributedReproducibleFaultTolerant slave 3 &+--+-- And start the master node:+--+-- ./MachRep2DistributedReproducibleFaultTolerant master 0+--++import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import System.Random++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++import SimpleLocalnetHelper++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying to the messages.+delta = 1e-6++-- | The initial seeds.+seeds = [456, 789]++-- | Create a new random generator by the specified seed.+newRandomRef :: Int -> Simulation DIO (Ref DIO StdGen)+newRandomRef = newRef . mkStdGen++-- | Generate a random number with the specified mean+randomRefExponential :: RandomGen g => Ref DIO g -> Double -> Event DIO Double+randomRefExponential r mu =+  do g <- readRef r+     let (x, g') = random g+     writeRef r g'+     return (- log x * mu)++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Int -> Simulation DIO ()+slaveModel masterId i =+  do inboxId <- liftComp messageInboxId+     g <- newRandomRef $ seeds !! (i - 1)++     let machine =+           do t <- liftDynamics time+              upTime <- randomRefExponential g meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <- randomRefExponential g meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       liftIO $+       putStrLn "The sub-model finished"++     runEventInStopTime $+       return ()++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId, i) =+  runDIO m ps timeServerId+  where+    ps = defaultDIOParams { dioLoggingPriority = NOTICE,+                            dioProcessMonitoringEnabled = True,+                            dioProcessReconnectingEnabled = True }+    m  = do registerDIO+            runSimulation (slaveModel masterId i) specs+            unregisterDIO++startSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process ()+startSlaveModel x@(timeServerId, masterId, i) =+  do (slaveId, slaveProcess) <- runSlaveModel x+     DP.send slaveId (MonitorProcessMessage timeServerId)+     DP.send masterId (MonitorProcessMessage slaveId)+     DP.send slaveId (MonitorProcessMessage masterId)+     slaveProcess++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 = NOTICE,+                            dioProcessMonitoringEnabled = True,+                            dioProcessReconnectingEnabled = True }+    m  = do registerDIO+            a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let [n0, n1, n2] = nodes+         timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG,+                                                      tsProcessMonitoringEnabled = True,+                                                      tsProcessReconnectingEnabled = True }+     -- timeServerId <- DP.spawnLocal $ 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)+     --      DP.spawnLocal slaveProcess+     (masterId, masterProcess) <- runMasterModel timeServerId 2+     DP.send masterId (MonitorProcessMessage timeServerId)+     forM_ (zip [n1, n2] [(1 :: Int)..]) $ \(node, i) ->+        DP.spawn node ($(mkClosure 'startSlaveModel) (timeServerId, masterId, i))+     a <- masterProcess+     liftIO $+       putStrLn $+       "The result is " ++ show a+     -- DP.say $+     --   "The result is " ++ show a++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["master", i] -> do+      (backend, nodes) <- getMasterConfBackend (read i) rtable+      startMasterProcess backend nodes master+    ["slave", i] -> do+      backend <- getSlaveConfBackend (read i) rtable+      startSlave backend+  where+    rtable :: DP.RemoteTable+    rtable = __remoteTable initRemoteTable

@@ -0,0 +1,311 @@++{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++-- Compiling and Running+---+-- Compile using+--+-- ghc -threaded MachRep2Distributed.hs+--+-- Fire up some slave nodes (for the example, we run them on a single machine):+--+-- ./MachRep2DistributedReproducibleFaultTolerant slave 1 &+-- ./MachRep2DistributedReproducibleFaultTolerant slave 2 &+-- ./MachRep2DistributedReproducibleFaultTolerant slave 3 &+--+-- And start the master node:+--+-- ./MachRep2DistributedReproducibleFaultTolerant master 0+--++import System.Environment (getArgs)++import Data.Typeable+import Data.Binary++import System.Random++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++import SimpleLocalnetHelper++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying to the messages.+delta = 1e-6++-- | The initial seeds.+seeds = [456, 789]++-- | Create a new random generator by the specified seed.+newRandomRef :: Int -> Simulation DIO (Ref DIO StdGen)+newRandomRef = newRef . mkStdGen++-- | Generate a random number with the specified mean+randomRefExponential :: RandomGen g => Ref DIO g -> Double -> Event DIO Double+randomRefExponential r mu =+  do g <- readRef r+     let (x, g') = random g+     writeRef r g'+     return (- log x * mu)++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Int -> Simulation DIO ()+slaveModel masterId i =+  do inboxId <- liftComp messageInboxId+     g <- newRandomRef $ seeds !! (i - 1)++     let machine =+           do t <- liftDynamics time+              upTime <- randomRefExponential g meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <- randomRefExponential g meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       liftIO $+       putStrLn "The sub-model finished"++     runEventInStopTime $+       return ()++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId, i) =+  runDIOWithEnv m ps env timeServerId+  where+    ps = defaultDIOParams { dioProcessMonitoringEnabled = True,+                            dioProcessReconnectingEnabled = True,+                            dioSimulationMonitoringInterval = 5000000 }+    env = defaultDIOEnv { dioSimulationMonitoringAction = Just monitorSimulation }+    m  = do registerDIO+            runSimulation (slaveModel masterId i) specs+            unregisterDIO++startSlaveModel :: (DP.ProcessId, DP.ProcessId, Int) -> DP.Process ()+startSlaveModel x@(timeServerId, masterId, i) =+  do (slaveId, slaveProcess) <- runSlaveModel x+     DP.send slaveId (MonitorProcessMessage timeServerId)+     DP.send masterId (MonitorProcessMessage slaveId)+     DP.send slaveId (MonitorProcessMessage masterId)+     slaveProcess++startTimeServer :: Int -> DP.Process ()+startTimeServer n =+  let timeServerParams = defaultTimeServerParams { tsProcessMonitoringEnabled = True,+                                                   tsProcessReconnectingEnabled = True,+                                                   tsSimulationMonitoringInterval = 5000000 }+      timeServerEnv = defaultTimeServerEnv { tsSimulationMonitoringAction = Just monitorSimulation }+  in timeServerWithEnv n timeServerParams timeServerEnv++monitorSimulation :: Show a => a -> DP.Process ()+monitorSimulation = DP.say . show++remotable ['startSlaveModel, 'startTimeServer]++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))+runMasterModel timeServerId n =+  runDIOWithEnv m ps env timeServerId+  where+    ps = defaultDIOParams { dioProcessMonitoringEnabled = True,+                            dioProcessReconnectingEnabled = True,+                            dioSimulationMonitoringInterval = 5000000 }+    env = defaultDIOEnv { dioSimulationMonitoringAction = Just monitorSimulation }+    m  = do registerDIO+            a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let [n0, n1, n2] = nodes+     -- timeServerId <- DP.spawnLocal $ timeServer timeServerParams+     timeServerId <- DP.spawn n0 ($(mkClosure 'startTimeServer) (3 :: Int))+     -- (masterId, masterProcess) <- runMasterModel timeServerId 2+     -- forM_ [1..2] $ \i ->+     --   do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)+     --      DP.spawnLocal slaveProcess+     (masterId, masterProcess) <- runMasterModel timeServerId 2+     DP.send masterId (MonitorProcessMessage timeServerId)+     forM_ (zip [n1, n2] [(1 :: Int)..]) $ \(node, i) ->+        DP.spawn node ($(mkClosure 'startSlaveModel) (timeServerId, masterId, i))+     a <- masterProcess+     liftIO $+       putStrLn $+       "The result is " ++ show a+     -- DP.say $+     --   "The result is " ++ show a++main :: IO ()+main = do+  args <- getArgs+  case args of+    ["master", i] -> do+      (backend, nodes) <- getMasterConfBackend (read i) rtable+      startMasterProcess backend nodes master+    ["slave", i] -> do+      backend <- getSlaveConfBackend (read i) rtable+      startSlave backend+  where+    rtable :: DP.RemoteTable+    rtable = __remoteTable initRemoteTable

@@ -0,0 +1,241 @@++{--# LANGUAGE TemplateHaskell #--}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveDataTypeable #-}++-- It corresponds to model MachRep2 described in document +-- Introduction to Discrete-Event Simulation and the SimPy Language+-- [http://heather.cs.ucdavis.edu/~matloff/156/PLN/DESimIntro.pdf]. +-- SimPy is available on [http://simpy.sourceforge.net/].+--   +-- The model description is as follows.+--   +-- Two machines, but sometimes break down. Up time is exponentially +-- distributed with mean 1.0, and repair time is exponentially distributed +-- with mean 0.5. In this example, there is only one repairperson, so +-- the two machines cannot be repaired simultaneously if they are down +-- at the same time.+--+-- In addition to finding the long-run proportion of up time as in+-- model MachRep1, let’s also find the long-run proportion of the time +-- that a given machine does not have immediate access to the repairperson +-- when the machine breaks down. Output values should be about 0.6 and 0.67. ++import Data.Typeable+import Data.Binary++import GHC.Generics++import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import qualified Control.Distributed.Process as DP+import Control.Distributed.Process.Closure+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet++import Simulation.Aivika.Trans+import Simulation.Aivika.Distributed++meanUpTime = 1.0+meanRepairTime = 0.5++specs = Specs { spcStartTime = 0.0,+                spcStopTime = 1000.0,+                spcDT = 1.0,+                spcMethod = RungeKutta4,+                spcGeneratorType = SimpleGenerator }++-- | The time shift when replying to the messages.+delta = 1e-6++data TotalUpTimeChange = TotalUpTimeChange (DP.ProcessId, Double) deriving (Eq, Ord, Show, Typeable, Generic)+data TotalUpTimeChangeResp = TotalUpTimeChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NRepChange = NRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NRepChangeResp = NRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data NImmedRepChange = NImmedRepChange (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)+data NImmedRepChangeResp = NImmedRepChangeResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data RepairPersonCount = RepairPersonCount DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RepairPersonCountResp = RepairPersonCountResp (DP.ProcessId, Int) deriving (Eq, Ord, Show, Typeable, Generic)++data RequestRepairPerson = RequestRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data RequestRepairPersonResp = RequestRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++data ReleaseRepairPerson = ReleaseRepairPerson DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)+data ReleaseRepairPersonResp = ReleaseRepairPersonResp DP.ProcessId deriving (Eq, Ord, Show, Typeable, Generic)++instance Binary TotalUpTimeChange+instance Binary TotalUpTimeChangeResp++instance Binary NRepChange+instance Binary NRepChangeResp++instance Binary NImmedRepChange+instance Binary NImmedRepChangeResp++instance Binary RepairPersonCount+instance Binary RepairPersonCountResp++instance Binary RequestRepairPerson+instance Binary RequestRepairPersonResp++instance Binary ReleaseRepairPerson+instance Binary ReleaseRepairPersonResp++-- | A sub-model.+slaveModel :: DP.ProcessId -> Simulation DIO ()+slaveModel masterId =+  do inboxId <- liftComp messageInboxId++     let machine =+           do t <- liftDynamics time+              upTime <-+                liftParameter $ randomExponential meanUpTime+              enqueueMessage masterId (t + delta + upTime) (TotalUpTimeChange (inboxId, upTime))+              enqueueMessage masterId (t + delta + upTime) (NRepChange (inboxId, 1))+              enqueueMessage masterId (t + delta + upTime) (RepairPersonCount inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCountResp (senderId, n)) ->+       do t <- liftDynamics time+          when (n == 1) $+            enqueueMessage masterId (t + delta) (NImmedRepChange (inboxId, 1))+          enqueueMessage masterId (t + delta) (RequestRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPersonResp senderId) ->+       do t <- liftDynamics time+          repairTime <-+            liftParameter $ randomExponential meanRepairTime+          enqueueMessage masterId (t + delta + repairTime) (ReleaseRepairPerson inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPersonResp senderId) ->+       machine+  +     runEventInStartTime machine++     runEventInStartTime $+       enqueueEventIOWithStopTime $+       liftComp $+       logDIO INFO "The sub-model has finished"++     runEventInStopTime $+       return ()++-- | The main model.       +masterModel :: Int -> Simulation DIO (Double, Double)+masterModel count =+  do totalUpTime <- newRef 0.0+     nRep <- newRef 0+     nImmedRep <- newRef 0++     let maxRepairPersonCount = 1+     repairPerson <- newFCFSResource maxRepairPersonCount++     inboxId <- liftComp messageInboxId++     runEventInStartTime $+       handleSignal messageReceived $ \(TotalUpTimeChange (senderId, x)) ->+       do modifyRef totalUpTime (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (TotalUpTimeChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NRepChange (senderId, x)) ->+       do modifyRef nRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(NImmedRepChange (senderId, x)) ->+       do modifyRef nImmedRep (+ x)+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (NImmedRepChangeResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(RepairPersonCount senderId) ->+       do n <- resourceCount repairPerson+          t <- liftDynamics time+          enqueueMessage senderId (t + delta) (RepairPersonCountResp (inboxId, n))++     runEventInStartTime $+       handleSignal messageReceived $ \(RequestRepairPerson senderId) ->+       runProcess $+       do requestResource repairPerson+          t <- liftDynamics time+          liftEvent $+            enqueueMessage senderId (t + delta) (RequestRepairPersonResp inboxId)++     runEventInStartTime $+       handleSignal messageReceived $ \(ReleaseRepairPerson senderId) ->+       do t <- liftDynamics time+          releaseResourceWithinEvent repairPerson+          enqueueMessage senderId (t + delta) (ReleaseRepairPersonResp inboxId)+          +     let upTimeProp =+           do x <- readRef totalUpTime+              y <- liftDynamics time+              return $ x / (fromIntegral count * y)++         immedProp =+           do n <- readRef nRep+              nImmed <- readRef nImmedRep+              return $+                fromIntegral nImmed /+                fromIntegral n++     runEventInStopTime $+       do x <- upTimeProp+          y <- immedProp+          return (x, y)++runSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process (DP.ProcessId, DP.Process ())+runSlaveModel (timeServerId, masterId) =+  runDIOWithEnv m ps env timeServerId+  where+    ps = defaultDIOParams { dioSimulationMonitoringInterval = 5000000 }+    env = defaultDIOEnv { dioSimulationMonitoringAction = Just monitorSimulation }+    m  = do registerDIO+            runSimulation (slaveModel masterId) specs+            unregisterDIO++runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process (Double, Double))+runMasterModel timeServerId n =+  runDIOWithEnv m ps env timeServerId+  where+    ps = defaultDIOParams { dioSimulationMonitoringInterval = 5000000 }+    env = defaultDIOEnv { dioSimulationMonitoringAction = Just monitorSimulation }+    m  = do registerDIO+            a <- runSimulation (masterModel n) specs+            terminateDIO+            return a++monitorSimulation :: Show a => a -> DP.Process ()+monitorSimulation = DP.say . show++master = \backend nodes ->+  do liftIO . putStrLn $ "Slaves: " ++ show nodes+     let n = 2+         timeServerParams = defaultTimeServerParams { tsSimulationMonitoringInterval = 5000000 }+         timeServerEnv = defaultTimeServerEnv { tsSimulationMonitoringAction = Just monitorSimulation }+     timeServerId <- DP.spawnLocal $ timeServerWithEnv 3 timeServerParams timeServerEnv+     (masterId, masterProcess) <- runMasterModel timeServerId n+     forM_ [1..n] $ \i ->+       do (slaveId, slaveProcess) <- runSlaveModel (timeServerId, masterId)+          DP.spawnLocal slaveProcess+     a <- masterProcess+     DP.say $+       "The result is " ++ show a+  +main :: IO ()+main = do+  backend <- initializeBackend "localhost" "8080" rtable+  startMaster backend (master backend)+    where+      rtable :: DP.RemoteTable+      -- rtable = __remoteTable initRemoteTable+      rtable = initRemoteTable