diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,9 @@ +Version 0.7.3 +----- + +* Updated so that external software tools could monitor the distributed simulation. + Version 0.7.2 ----- diff --git a/Simulation/Aivika/Distributed/Optimistic.hs b/Simulation/Aivika/Distributed/Optimistic.hs --- a/Simulation/Aivika/Distributed/Optimistic.hs +++ b/Simulation/Aivika/Distributed/Optimistic.hs @@ -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 diff --git a/Simulation/Aivika/Distributed/Optimistic/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/DIO.hs --- a/Simulation/Aivika/Distributed/Optimistic/DIO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/DIO.hs @@ -14,9 +14,12 @@ module Simulation.Aivika.Distributed.Optimistic.DIO (DIO, DIOParams(..), + DIOEnv(..), DIOStrategy(..), runDIO, + runDIOWithEnv, defaultDIOParams, + defaultDIOEnv, dioParams, messageInboxId, timeServerId, diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/DIO.hs @@ -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, diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Event.hs @@ -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 } diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/KeepAliveManager.hs @@ -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 () diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Message.hs @@ -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) diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs --- a/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/TimeServer.hs @@ -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. diff --git a/Simulation/Aivika/Distributed/Optimistic/State.hs b/Simulation/Aivika/Distributed/Optimistic/State.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/State.hs @@ -0,0 +1,68 @@ + +{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.State +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- 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 diff --git a/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs b/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs --- a/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs +++ b/Simulation/Aivika/Distributed/Optimistic/TimeServer.hs @@ -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 diff --git a/aivika-distributed.cabal b/aivika-distributed.cabal --- a/aivika-distributed.cabal +++ b/aivika-distributed.cabal @@ -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] . \[2] . \[3] . + \[4] + . 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 diff --git a/tests/MachRep1SimpleWithMonitoring.hs b/tests/MachRep1SimpleWithMonitoring.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep1SimpleWithMonitoring.hs @@ -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 diff --git a/tests/MachRep2DistributedReproducible.hs b/tests/MachRep2DistributedReproducible.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2DistributedReproducible.hs @@ -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 diff --git a/tests/MachRep2DistributedReproducibleFaultTolerant.hs b/tests/MachRep2DistributedReproducibleFaultTolerant.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2DistributedReproducibleFaultTolerant.hs @@ -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 diff --git a/tests/MachRep2DistributedWithMonitoring.hs b/tests/MachRep2DistributedWithMonitoring.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2DistributedWithMonitoring.hs @@ -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 diff --git a/tests/MachRep2WithMonitoring.hs b/tests/MachRep2WithMonitoring.hs new file mode 100644 --- /dev/null +++ b/tests/MachRep2WithMonitoring.hs @@ -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