diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,16 @@ +Version 0.6 +----- + +* Using the mwc-random package for generating random numbers by default. + +Version 0.5.1 +----- + +* Added functions expectEvent and expectProcess. + +* Added the Guard module. + Version 0.5 ----- 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 @@ -13,6 +13,7 @@ module Simulation.Aivika.Distributed.Optimistic (-- * Modules module Simulation.Aivika.Distributed.Optimistic.DIO, + module Simulation.Aivika.Distributed.Optimistic.Guard, module Simulation.Aivika.Distributed.Optimistic.Generator, module Simulation.Aivika.Distributed.Optimistic.Message, module Simulation.Aivika.Distributed.Optimistic.QueueStrategy, @@ -21,6 +22,7 @@ module Simulation.Aivika.Distributed.Optimistic.TimeServer) where import Simulation.Aivika.Distributed.Optimistic.DIO +import Simulation.Aivika.Distributed.Optimistic.Guard import Simulation.Aivika.Distributed.Optimistic.Generator import Simulation.Aivika.Distributed.Optimistic.Message import Simulation.Aivika.Distributed.Optimistic.QueueStrategy 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 @@ -39,6 +39,7 @@ import Simulation.Aivika.Trans.Process import Simulation.Aivika.Trans.Ref.Base import Simulation.Aivika.Trans.QueueStrategy +import Simulation.Aivika.Trans.Internal.Types import Simulation.Aivika.Distributed.Optimistic.Internal.Message import Simulation.Aivika.Distributed.Optimistic.Internal.DIO diff --git a/Simulation/Aivika/Distributed/Optimistic/Generator.hs b/Simulation/Aivika/Distributed/Optimistic/Generator.hs --- a/Simulation/Aivika/Distributed/Optimistic/Generator.hs +++ b/Simulation/Aivika/Distributed/Optimistic/Generator.hs @@ -18,6 +18,7 @@ import Control.Monad.Trans import System.Random +import qualified System.Random.MWC as MWC import Data.IORef @@ -69,7 +70,10 @@ newGenerator tp = case tp of SimpleGenerator -> - liftIOUnsafe newStdGen >>= newRandomGenerator + do let g = MWC.uniform <$> + MWC.withSystemRandom (return :: MWC.GenIO -> IO MWC.GenIO) + g' <- liftIOUnsafe g + newRandomGenerator01 (liftIOUnsafe g') SimpleGeneratorWithSeed x -> error "Unsupported generator type SimpleGeneratorWithSeed: newGenerator" CustomGenerator g -> diff --git a/Simulation/Aivika/Distributed/Optimistic/Guard.hs b/Simulation/Aivika/Distributed/Optimistic/Guard.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Guard.hs @@ -0,0 +1,166 @@ + +{-# LANGUAGE DeriveGeneric, DeriveDataTypeable #-} + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Guard +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 8.0.2 +-- +-- This module defines guards that allow correct finishing the distributed simulation. +-- +module Simulation.Aivika.Distributed.Optimistic.Guard + (-- * Guards With Message Passing + runMasterGuard, + runSlaveGuard, + -- * Guards Without Message Passing + runMasterGuard_, + runSlaveGuard_) where + +import GHC.Generics + +import Data.Typeable +import Data.Binary +import qualified Data.Map as M + +import Control.Monad + +import qualified Control.Distributed.Process as DP +import Control.Distributed.Process.Serializable + +import Simulation.Aivika.Trans +import Simulation.Aivika.Trans.Internal.Types +import Simulation.Aivika.Distributed.Optimistic.Internal.Expect +import Simulation.Aivika.Distributed.Optimistic.DIO +import Simulation.Aivika.Distributed.Optimistic.Message + +-- | Represent the master message. +data MasterMessage a = MasterMessage DP.ProcessId (Maybe a) + deriving (Show, Typeable, Generic) + +-- | Represent the slave message. +data SlaveMessage a = SlaveMessage DP.ProcessId a + deriving (Show, Typeable, Generic) + +instance Binary a => Binary (MasterMessage a) +instance Binary a => Binary (SlaveMessage a) + +-- | Represents the master guard that waits for all slaves to finish. +data MasterGuard a = MasterGuard { masterGuardSlaveMessages :: Ref DIO (M.Map DP.ProcessId a) + -- ^ the messages of slaves connected to the master + } + +-- | Represents the slave guard that waits for the master's acknowledgement. +data SlaveGuard a = SlaveGuard { slaveGuardAcknowledgedMessage :: Ref DIO (Maybe (Maybe a)) + -- ^ whether the slave process was acknowledged by the master + } + +-- | Create a new master guard. +newMasterGuard :: Serializable a => Event DIO (MasterGuard a) +newMasterGuard = + do r <- liftSimulation $ newRef M.empty + handleSignal messageReceived $ \(SlaveMessage slaveId a) -> + modifyRef r $ M.insert slaveId a + return MasterGuard { masterGuardSlaveMessages = r } + +-- | Create a new slave guard. +newSlaveGuard :: Serializable a => Event DIO (SlaveGuard a) +newSlaveGuard = + do r <- liftSimulation $ newRef Nothing + handleSignal messageReceived $ \(MasterMessage masterId a) -> + writeRef r (Just a) + return SlaveGuard { slaveGuardAcknowledgedMessage = r } + +-- | Await until the specified number of slaves are connected to the master. +awaitMasterGuard :: Serializable b + => MasterGuard a + -- ^ the master guard + -> Int + -- ^ the number of slaves to wait + -> (M.Map DP.ProcessId a -> Event DIO (M.Map DP.ProcessId b)) + -- ^ process the messages sent by slaves + -> Process DIO (M.Map DP.ProcessId b) +awaitMasterGuard guard n transform = + expectProcess $ + do m <- readRef $ masterGuardSlaveMessages guard + if M.size m < n + then return Nothing + else do m' <- transform m + inboxId <- liftComp messageInboxId + forM_ (M.keys m) $ \slaveId -> + sendMessage slaveId (MasterMessage inboxId $ M.lookup slaveId m') + return $ Just m' + +-- | Await until the specified master receives notifications from the slave processes. +awaitSlaveGuard :: (Serializable a, + Serializable b) + => SlaveGuard a + -- ^ the slave guard + -> DP.ProcessId + -- ^ the master process identifier + -> Event DIO b + -- ^ the message generator + -> Process DIO (Maybe a) + -- ^ the master's reply +awaitSlaveGuard guard masterId generator = + do liftEvent $ + do b <- generator + inboxId <- liftComp messageInboxId + sendMessage masterId (SlaveMessage inboxId b) + expectProcess $ + readRef $ slaveGuardAcknowledgedMessage guard + +-- | Run the master guard by the specified number of slaves and transform function. +runMasterGuard :: (Serializable a, + Serializable b) + => Int + -- ^ the number of slaves to wait + -> (M.Map DP.ProcessId a -> Event DIO (M.Map DP.ProcessId b)) + -- ^ how to transform the messages from the slave processes in the stop time + -> Process DIO (M.Map DP.ProcessId b) +runMasterGuard n transform = + do source <- liftSimulation newSignalSource + liftEvent $ + do guard <- newMasterGuard + enqueueEventWithStopTime $ + runProcess $ + do b <- awaitMasterGuard guard n transform + liftEvent $ + triggerSignal source b + processAwait $ publishSignal source + +-- | Run the slave guard by the specified master process identifier and message generator. +runSlaveGuard :: (Serializable a, + Serializable b) + => DP.ProcessId + -- ^ the master process identifier + -> Event DIO a + -- ^ in the stop time generate a message to pass to the master process + -> Process DIO (Maybe b) + -- ^ the message returned by the master process +runSlaveGuard masterId generator = + do source <- liftSimulation newSignalSource + liftEvent $ + do guard <- newSlaveGuard + enqueueEventWithStopTime $ + runProcess $ + do b <- awaitSlaveGuard guard masterId generator + liftEvent $ + triggerSignal source b + processAwait $ publishSignal source + +-- | Run the master guard by the specified number of slaves when there is no message passing. +runMasterGuard_ :: Int -> Process DIO () +runMasterGuard_ n = + do _ <- runMasterGuard n transform :: Process DIO (M.Map DP.ProcessId ()) + return () + where transform m = return m + +-- | Run the slave guard by the specified master process identifier when there is no message passing. +runSlaveGuard_ :: DP.ProcessId -> Process DIO () +runSlaveGuard_ masterId = + do _ <- runSlaveGuard masterId generator :: Process DIO (Maybe ()) + return () + where generator = return () 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 @@ -15,6 +15,7 @@ (queueInputMessages, queueOutputMessages, queueLog, + expectEvent, processMonitorSignal) where import Data.Maybe @@ -32,6 +33,9 @@ import Simulation.Aivika.Trans import Simulation.Aivika.Trans.Internal.Types +import Simulation.Aivika.Trans.Internal.Event +import Simulation.Aivika.Trans.Internal.Cont +import Simulation.Aivika.Trans.Internal.Process import Simulation.Aivika.Distributed.Optimistic.Internal.Priority import Simulation.Aivika.Distributed.Optimistic.Internal.Channel @@ -761,3 +765,66 @@ in invokeEvent p $ handleSignal s h } + + +-- | Suspend the 'Event' computation until the specified computation is determined. +-- +-- The tested computation should depend on messages that come from other local processes. +-- Moreover, the event must be initiated through the event queue. +expectEvent :: Event DIO (Maybe a) -> (a -> Event DIO ()) -> Event DIO () +expectEvent m cont = + Event $ \p -> + do let q = runEventQueue $ pointRun p + t = pointTime p + --- + logDIO INFO $ + "t = " ++ show (pointTime p) ++ + ": expecting the computation result: expectEvent" + --- + let loop = + do --- + --- logDIO DEBUG $ + --- "t = " ++ show (pointTime p) ++ + --- ": testing the predicate: expectEvent" + --- + x <- invokeEvent p m + case x of + Just a -> invokeEvent p $ cont a + Nothing -> next loop0 + loop0 = + do --- + --- logDIO DEBUG $ + --- "t = " ++ show (pointTime p) ++ + --- ": testing the predicate in ring 0: expectEvent" + --- + x <- invokeEvent p m + case x of + Just a -> invokeEvent p $ cont a + Nothing -> next $ error "Detected a deadlock: expectEvent" + next loop' = + do pq <- invokeEvent p $ R.readRef $ queuePQ q + let f = PQ.queueNull pq + if f + then await loop' + else do let (t2, _) = PQ.queueFront pq + if t < t2 + then await loop' + else invokeEvent p $ + enqueueEvent t $ + Event $ \p -> loop + await loop' = + do --- + --- logDIO DEBUG $ + --- "t = " ++ show t ++ + --- ": waiting for arriving a message: expectEvent" + --- + ch <- messageChannel + dt <- fmap dioSyncTimeout dioParams + f <- liftIOUnsafe $ + timeout dt $ awaitChannel ch + ok <- invokeEvent p $ runTimeWarp processChannelMessages + when ok $ + case f of + Just _ -> loop + Nothing -> loop' + loop diff --git a/Simulation/Aivika/Distributed/Optimistic/Internal/Expect.hs b/Simulation/Aivika/Distributed/Optimistic/Internal/Expect.hs new file mode 100644 --- /dev/null +++ b/Simulation/Aivika/Distributed/Optimistic/Internal/Expect.hs @@ -0,0 +1,43 @@ + +-- | +-- Module : Simulation.Aivika.Distributed.Optimistic.Internal.Expect +-- Copyright : Copyright (c) 2015-2017, David Sorokin +-- License : BSD3 +-- Maintainer : David Sorokin +-- Stability : experimental +-- Tested with: GHC 7.10.3 +-- +-- Implements the 'expectProcess' function that allows waiting for some computation to have a determined result. +-- +module Simulation.Aivika.Distributed.Optimistic.Internal.Expect + (expectProcess) where + +import Control.Monad +import Control.Monad.Trans + +import Simulation.Aivika.Trans +import Simulation.Aivika.Trans.Internal.Types +import Simulation.Aivika.Trans.Internal.Simulation +import Simulation.Aivika.Trans.Internal.Event +import Simulation.Aivika.Trans.Internal.Cont +import Simulation.Aivika.Trans.Internal.Process + +import Simulation.Aivika.Distributed.Optimistic.DIO +import Simulation.Aivika.Distributed.Optimistic.Internal.Event + +-- | Suspend the 'Process' until the specified computation is determined. +-- +-- The tested computation should depend on messages that come from other local processes. +-- Moreover, the process must be initiated through the event queue. +-- +-- In the current implementation there is a limitation that this function can be used only +-- once for the entire local process simulation; otherwise, a race condition may arise. +expectProcess :: Event DIO (Maybe a) -> Process DIO a +expectProcess m = + Process $ \pid -> + Cont $ \c -> + Event $ \p -> + do let t = pointTime p + invokeEvent p $ + expectEvent m $ + resumeCont c 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.5 +version: 0.6 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. @@ -17,7 +17,7 @@ is that this is the optimistic distributed simulation with possible rollbacks. It is assumed that optimistic methods tend to better support the parallelism inherited in the models. . - You may test the distributed simulation using your own notebook only, although the package is still destined to be + You may test the distributed simulation using your own laptop only, although the package is still destined to be used with a multi-core computer, or computers connected in the distributed cluster. . There are additional packages that allow you to run the distributed simulation experiments by @@ -25,7 +25,7 @@ or a set of reports consisting of HTML pages with charts, histograms, links to CSV tables, summary statistics etc. Please consult the AivikaSoft [3] website for more details. . - Regarding the speed of simulation, the rough estimations are as follows. The simulation is slower up to + Regarding the speed of simulation, the rough estimations are as follows. The distributed simulation module is slower up to 6-9 times in comparison with the sequential aivika [2] simulation library using the equivalent sequential models. Note that you can run up to 7 parallel local processes on a single 8-core processor computer and run the Time Server process too. On a 36-core processor, you can launch up to 35 local processes simultaneously. @@ -51,7 +51,8 @@ build-type: Simple tested-with: GHC == 7.10.1 -extra-source-files: tests/MachRep1.hs +extra-source-files: tests/Guard1.hs + tests/MachRep1.hs tests/MachRep1Simple.hs tests/MachRep2.hs tests/MachRep2Distributed.hs @@ -67,6 +68,7 @@ exposed-modules: Simulation.Aivika.Distributed Simulation.Aivika.Distributed.Optimistic Simulation.Aivika.Distributed.Optimistic.DIO + Simulation.Aivika.Distributed.Optimistic.Guard Simulation.Aivika.Distributed.Optimistic.Generator Simulation.Aivika.Distributed.Optimistic.QueueStrategy Simulation.Aivika.Distributed.Optimistic.Message @@ -79,6 +81,7 @@ other-modules: Simulation.Aivika.Distributed.Optimistic.Internal.Channel Simulation.Aivika.Distributed.Optimistic.Internal.DIO Simulation.Aivika.Distributed.Optimistic.Internal.Event + Simulation.Aivika.Distributed.Optimistic.Internal.Expect Simulation.Aivika.Distributed.Optimistic.Internal.InputMessageQueue Simulation.Aivika.Distributed.Optimistic.Internal.IO Simulation.Aivika.Distributed.Optimistic.Internal.KeepAliveManager @@ -98,13 +101,14 @@ mtl >= 2.1.1, stm >= 2.4.2, random >= 1.0.0.3, + mwc-random >= 0.13.0.0, binary >= 0.6.4.0, time >= 1.5.0.1, containers >= 0.4.0.0, exceptions >= 0.8.0.2, distributed-process >= 0.6.1, - aivika >= 5.1, - aivika-transformers >= 5.1 + aivika >= 5.2, + aivika-transformers >= 5.2 extensions: MultiParamTypeClasses, FlexibleInstances, diff --git a/tests/Guard1.hs b/tests/Guard1.hs new file mode 100644 --- /dev/null +++ b/tests/Guard1.hs @@ -0,0 +1,150 @@ + +{--# LANGUAGE TemplateHaskell #--} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE DeriveDataTypeable #-} + +-- 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 System.Environment (getArgs) + +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 = 10000.0, + spcDT = 1.0, + spcMethod = RungeKutta4, + spcGeneratorType = SimpleGenerator } + +newtype TotalUpTimeChange = TotalUpTimeChange { runTotalUpTimeChange :: Double } + deriving (Eq, Ord, Show, Typeable, Generic) + +instance Binary TotalUpTimeChange + +-- | A sub-model. +slaveModel :: DP.ProcessId -> Simulation DIO () +slaveModel masterId = + do let machine = + do upTime <- + liftParameter $ + randomExponential meanUpTime + holdProcess upTime + --- + liftEvent $ + sendMessage masterId (TotalUpTimeChange upTime) + --- + repairTime <- + liftParameter $ + randomExponential meanRepairTime + holdProcess repairTime + machine + + runProcessInStartTime machine + + runProcessInStartTime $ + runSlaveGuard_ masterId + + runEventInStartTime $ + enqueueEventIOWithStopTime $ + do liftIO $ + putStrLn "The sub-model finished" + + runEventInStopTime $ + return () + +-- | The main model. +masterModel :: Int -> Simulation DIO Double +masterModel n = + do totalUpTime <- newRef 0.0 + + --- + let totalUpTimeChanged :: Signal DIO TotalUpTimeChange + totalUpTimeChanged = messageReceived + + runEventInStartTime $ + handleSignal totalUpTimeChanged $ \x -> + modifyRef totalUpTime (+ runTotalUpTimeChange x) + --- + + runProcessInStartTime $ + runMasterGuard_ n + + let upTimeProp = + do x <- readRef totalUpTime + y <- liftDynamics time + return $ x / (fromIntegral n * y) + + runEventInStopTime $ + do liftIO $ + putStrLn "The main model finished" + upTimeProp + +runSlaveModel :: (DP.ProcessId, DP.ProcessId) -> DP.Process (DP.ProcessId, DP.Process ()) +runSlaveModel (timeServerId, masterId) = + runDIO m ps timeServerId + where + ps = defaultDIOParams { dioLoggingPriority = NOTICE } + m = do registerDIO + runSimulation (slaveModel masterId) specs + unregisterDIO + +runMasterModel :: DP.ProcessId -> Int -> DP.Process (DP.ProcessId, DP.Process Double) +runMasterModel timeServerId n = + runDIO m ps timeServerId + where + ps = defaultDIOParams { dioLoggingPriority = NOTICE } + m = do registerDIO + a <- runSimulation (masterModel n) specs + terminateDIO + return a + +master = \backend nodes -> + do liftIO . putStrLn $ "Slaves: " ++ show nodes + let timeServerParams = defaultTimeServerParams { tsLoggingPriority = DEBUG } + timeServerId <- DP.spawnLocal $ timeServer 3 timeServerParams + (masterId, masterProcess) <- runMasterModel timeServerId 2 + forM_ [1..2] $ \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